napalm-linux/tests/test_sudo_off_the_command_line.py

"""The sudo password never goes into a command line or a shell history (NetOrk/netork#1039).

``_sudo`` typed ``echo '<password>' | sudo -S -p "" <command>`` into netmiko's
interactive shell. That shell keeps a history, so every poll appended the
password to the login user's ``~/.bash_history`` -- 121 lines of it on one host
-- and while the command ran it was in the process list for every local user.

Privileged commands now run on an exec channel next to netmiko's PTY, the way
``run_command(privileged=True)`` does: the password reaches ``sudo -S`` on
stdin, and an exec channel has no history and no PTY to echo it back. The
interactive session switches its history off before anything else, in case
anything ever slips through again.
"""

from __future__ import annotations

import shlex
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest

from napalm_linux import LinuxDriver
from napalm_linux.linux import _HISTORY_OFF

_PW = "s3cr3t 'q' $x"


class Channel:
    """An exec channel answering with *output* and *status*; it refuses a line
    that holds the password, which is what ``ps`` would show."""

    def __init__(self, host: "Transport"):
        self.host = host
        self.command = ""
        self.stdin = b""
        self.timeout = None
        self._out = b""

    def exec_command(self, command):
        assert _PW not in command, f"sudo password in a command line: {command!r}"
        self.command = command
        output, status = self.host.answer
        self._out = output.encode()
        self.status = status

    def settimeout(self, timeout):
        self.timeout = timeout

    def sendall(self, data):
        self.stdin += data

    def shutdown_write(self):
        pass

    def close(self):
        pass

    def recv_ready(self):
        return bool(self._out)

    def recv(self, size):
        chunk, self._out = self._out[:size], self._out[size:]
        return chunk

    def recv_stderr_ready(self):
        return False

    def recv_stderr(self, size):
        return b""

    def exit_status_ready(self):
        return not self._out

    def recv_exit_status(self):
        return self.status


class Transport:
    def __init__(self, output="", status=0):
        self.answer = (output, status)
        self.channels: list[Channel] = []

    def open_session(self):
        self.channels.append(Channel(self))
        return self.channels[-1]


def _driver(*, sudo_password=_PW, root=False, output="", status=0):
    driver = LinuxDriver("h", "admin", "p", optional_args={"sudo_password": sudo_password})
    transport = Transport(output, status)
    driver._device = MagicMock()
    driver._device.remote_conn_pre = SimpleNamespace(get_transport=lambda: transport)
    driver._root = root
    return driver, transport


def _script(command_line: str) -> str:
    """What ``sh -c`` runs in *command_line*, unquoted."""
    parts = shlex.split(command_line)
    return parts[parts.index("-c") + 1]


class TestSudo:
    def test_the_password_goes_to_sudo_on_stdin(self):
        driver, transport = _driver(output="ok\n")

        assert driver._sudo("apt-get update 2>&1") == "ok"

        channel = transport.channels[-1]
        assert channel.command.startswith("sudo -S -p '' sh -c ")
        assert channel.stdin == f"{_PW}\n".encode()

    def test_nothing_goes_through_the_interactive_shell(self):
        """Its history is where the password ended up."""
        driver, _ = _driver()

        driver._sudo("apt-get update")

        driver._device.send_command.assert_not_called()

    def test_the_whole_command_runs_as_root(self):
        """``echo pw | sudo -S mv a b && chown …`` ran only ``mv`` as root."""
        driver, transport = _driver()

        driver._sudo("mv /tmp/x /etc/x && chown root:root /etc/x")

        assert _script(transport.channels[-1].command).endswith(
            "mv /tmp/x /etc/x && chown root:root /etc/x"
        )

    def test_a_sudo_that_does_not_prompt_leaves_the_password_to_nobody(self):
        """NOPASSWD or a cached timestamp leave the password on stdin; the
        command's stdin is /dev/null, so nothing it runs reads it."""
        driver, transport = _driver()

        driver._sudo("dpkg --configure -a")

        assert _script(transport.channels[-1].command).startswith("exec </dev/null; ")

    def test_sudos_own_complaints_are_in_the_output(self):
        """``sudo: a password is required`` goes to stderr; callers read it from the output."""
        driver, transport = _driver()

        driver._sudo("true")

        assert transport.channels[-1].command.endswith(" 2>&1")

    def test_without_a_password_sudo_never_waits_for_one(self):
        driver, transport = _driver(sudo_password=None)

        driver._sudo("apt-get update")

        assert transport.channels[-1].command.startswith("sudo -n sh -c ")
        assert transport.channels[-1].stdin == b""

    def test_a_root_login_needs_no_sudo(self):
        driver, transport = _driver(root=True)

        driver._sudo("apt-get update")

        assert transport.channels[-1].command.startswith("sh -c ")
        assert transport.channels[-1].stdin == b""

    def test_the_read_timeout_is_the_channels(self):
        driver, transport = _driver()

        driver._sudo("apt-get upgrade -y", read_timeout=600)

        assert transport.channels[-1].timeout == 600


class TestSudoStatus:
    def test_the_exit_status_is_the_commands_own(self):
        driver, _ = _driver(output="E: Unable to locate package nope\n", status=100)

        assert driver._sudo_status("apt-get remove -y nope 2>&1") == (
            "E: Unable to locate package nope",
            100,
        )

    def test_a_refused_sudo_is_a_failure(self):
        driver, _ = _driver(output="Sorry, try again.\n", status=1)

        assert driver._sudo_status("apt-get remove -y x")[1] == 1


class TestRunPrivileged:
    def test_with_a_password_it_takes_the_exec_channel(self):
        driver, transport = _driver(output="__NETORK_RC=0\n")

        driver._run_privileged("apt-get update -q 2>&1; echo __NETORK_RC=$?", 180)

        driver._device.send_command.assert_not_called()
        assert transport.channels[-1].stdin == f"{_PW}\n".encode()

    def test_a_refresh_reads_its_status_from_the_exec_channel(self):
        driver, _ = _driver(output="Hit:1 http://deb.debian.org\n__NETORK_RC=0\n")
        driver._pkg_manager = "apt"

        assert driver.refresh_available_updates()["success"] is True


class TestWireGuardWithSudo:
    def test_the_dump_comes_straight_from_the_exec_channel(self):
        """No PTY to turn its tabs into spaces, so no file in /tmp holding the
        interfaces' private keys."""
        dump = "wg0\tPRIV\tPUB\t51820\toff\nwg0\tPEER\t(none)\t1.2.3.4:51820\t10.0.0.2/32\t0\t1\t2\toff\n"
        driver, transport = _driver(output=dump)

        tunnels = driver.get_vpn_tunnels()

        assert list(tunnels) == ["wireguard-wg0-PEER"]
        assert "/tmp" not in transport.channels[-1].command
        driver._device.send_command.assert_not_called()


class TestHistoryOff:
    """Defense in depth: the interactive session keeps no history at all."""

    def test_it_is_the_first_thing_the_session_sends(self):
        driver = LinuxDriver("h", "admin", "p", optional_args={"sudo_password": _PW})
        device = MagicMock()
        device.send_command.return_value = "/usr/bin/apt"
        with patch("napalm_linux.linux.ConnectHandler", return_value=device):
            driver.open()

        first = device.send_command.call_args_list[0].args[0]
        assert first == _HISTORY_OFF

    def test_it_leaves_no_history_file_and_records_nothing_more(self):
        assert _HISTORY_OFF.startswith(" ")  # HISTCONTROL=ignorespace skips the line itself
        assert "unset HISTFILE" in _HISTORY_OFF
        assert "set +o history" in _HISTORY_OFF

    def test_a_shell_that_refuses_it_does_not_stop_the_connection(self):
        driver = LinuxDriver("h", "admin", "p")
        device = MagicMock()
        device.send_command.side_effect = [Exception("no prompt"), "", "/usr/bin/apt"]
        with patch("napalm_linux.linux.ConnectHandler", return_value=device):
            driver.open()

        assert driver._device is device


@pytest.mark.parametrize(
    "method, args",
    [
        ("install_package", ("curl",)),
        ("uninstall_package", ("curl",)),
        ("apply_updates", ([],)),
    ],
)
def test_package_work_never_types_the_password(method, args):
    driver, transport = _driver(output="done\n")
    driver._pkg_manager = "apt"

    getattr(driver, method)(*args)

    driver._device.send_command.assert_not_called()
    assert transport.channels and all(c.stdin == f"{_PW}\n".encode() for c in transport.channels)