diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d28d49..63d569a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (NetOrk/netork#1029). Stage the data without privilege and have root install that file. Without a sudo password (`sudo -n`) or for a root login, stdin still goes to the command. +- The sudo password is no longer typed into the interactive shell + (NetOrk/netork#1039). Every privileged command ran there as + `echo '' | sudo -S -p "" `: bash kept it in its history, + so each poll wrote the password into the login user's `~/.bash_history`, and + while the command ran it was in the process list for every local user. + Privileged commands (`_sudo`, `_sudo_status`, and `_run_privileged` with a + sudo password) now run on an exec channel like `run_command(privileged=True)`: + the password reaches `sudo -S` on stdin, there is no PTY to echo it, and the + whole command runs as root in one `sh -c` (before, only the part up to the + first `;`, `&&` or `|` did, so `mv … && chown root:root …` failed its chown). + The interactive session also switches its history off before its first + command (` unset HISTFILE; set +o history`). **History files written before + this version may still hold the sudo password**: remove those lines from the + login user's `~/.bash_history` on your hosts and consider changing the + password. `_sudo_status` takes the exit status from the channel; it is `None` + only when the channel closed without one. ## [0.4.0] – 2026-10-08 diff --git a/README.md b/README.md index 0ceb4eb..658e325 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,11 @@ The SSH user needs read access to: For full functionality it is recommended to run as `root` or grant passwordless `sudo` for the above commands. +With a `sudo_password`, privileged commands run on an exec channel next to the interactive +shell: the password reaches `sudo -S` on stdin, never a command line, so it shows neither in +the process list nor in a shell history. The interactive session switches its history off +before its first command. + `get_services()` and `manage_service()` come from napalm-device-types' `SystemdServicesMixin`; this driver supplies only the transport. An action runs as `timeout 45 systemctl --no-ask-password -- .service`, so a unit that hangs diff --git a/napalm_linux/linux.py b/napalm_linux/linux.py index 31d17f0..099f6f6 100644 --- a/napalm_linux/linux.py +++ b/napalm_linux/linux.py @@ -69,8 +69,8 @@ logger = logging.getLogger("napalm_linux") # Package managers in detection order _PKG_MANAGERS = ["apt", "dnf", "yum", "apk", "pacman"] -#: Printed after a command by ``_sudo_status`` so its exit status survives the -#: trip through an interactive shell. Matched only on a line of its own with a +#: Printed after a command run through the interactive shell (``_run_privileged`` +#: callers) so its exit status survives the trip. Matched only on a line of its own with a #: number after it — an echoed command line carries the literal ``$?`` instead. _RC_MARKER = "__NETORK_RC=" _RC_MARKER_RE = re.compile(rf"^{_RC_MARKER}(\d+)\s*$", re.MULTILINE) @@ -142,6 +142,16 @@ _PRIVILEGED_STDIN = ( "mktemp, cat) and have root install that file." ) +#: Sent before anything else in the interactive session (NetOrk/netork#1039): +#: the shell writes no history file and records nothing more. The leading space +#: keeps this line itself out of a history under ``HISTCONTROL=ignorespace``; +#: a shell without ``set -o history`` (dash) only complains, to /dev/null. +_HISTORY_OFF = " unset HISTFILE; set +o history 2>/dev/null; true" + +#: First in every script ``_sudo`` runs: a sudo that does not prompt (NOPASSWD, +#: a cached timestamp) leaves the password on stdin, and nothing may read it. +_NO_STDIN = "exec /dev/null || true", expect_string=r"[#$>]\s*$") @@ -350,29 +366,38 @@ class LinuxDriver( ).strip() def _sudo(self, command: str, read_timeout: float = 100) -> str: - """Run *command* via sudo, feeding the password via stdin (-S). + """Run *command* as root and return its output, stdout and stderr together. - Falls back to plain execution when no sudo password is configured. + It runs on an exec channel, not in netmiko's interactive shell + (NetOrk/netork#1039): that shell keeps a history, and ``echo | + sudo -S`` typed into it wrote the password into ``~/.bash_history`` on + every poll, and into the process list while it ran. On the exec channel + the password reaches ``sudo -S`` on stdin (``_as_root``), and there is no + PTY to echo it back into the output. + + The whole of *command* runs as root in one ``sh -c``, ``;`` and ``&&`` + included. Without a sudo password ``sudo -n`` fails at once; a root + login runs it directly. """ - if self._sudo_password: - wrapped = f'echo {_shlex_quote(self._sudo_password)} | sudo -S -p "" {command}' - return self._send(wrapped, read_timeout=read_timeout) - return self._send(f'sudo {command}', read_timeout=read_timeout) + return self._sudo_result(command, read_timeout).stdout.strip() + + def _sudo_result(self, command: str, timeout: float) -> CommandResult: + line, prefix = self._as_root(f"{_NO_STDIN}{command}") + return run_on_transport(self._transport(), f"{line} 2>&1", stdin=prefix, timeout=timeout) def _sudo_status(self, command: str, read_timeout: float = 100) -> tuple[str, int | None]: - """Run *command* via sudo and return ``(output, exit_status)``. + """Run *command* as root and return ``(output, exit_status)``. ``_sudo`` callers append ``|| true`` so a failing command yields output - instead of an error, which throws the exit status away. This variant - echoes ``$?`` straight after the sudo pipeline instead — sudo passes - the command's status through, and a failed password is non-zero too. + instead of an error, which throws the exit status away. The exec channel + reports the status itself: the command's, or sudo's when it refused. - The status is ``None`` when the marker never arrived (output cut short), - so a caller can tell "unknown" from "succeeded". + The status is ``None`` when the channel closed without one (paramiko's + -1), so a caller can tell "unknown" from "succeeded". """ - return _split_status( - self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=read_timeout) - ) + result = self._sudo_result(command, read_timeout) + status = result.exit_code if result.exit_code >= 0 else None + return result.stdout.strip(), status def _is_root(self) -> bool: """Whether the SSH user is root, asked once per session. @@ -401,7 +426,7 @@ class LinuxDriver( if self._is_root(): return self._send(command, read_timeout=timeout) if self._sudo_password: - return self._sudo(command, read_timeout=timeout) + return self._sudo(command, read_timeout=timeout) # never typed into the shell return self._send(f"sudo -n {command}", read_timeout=timeout) # ------------------------------------------------------------------ @@ -425,10 +450,20 @@ class LinuxDriver( """ if not privileged or self._is_root(): return command, None + return self._as_root(command) + + def _as_root(self, command: str) -> tuple[str, bytes | None]: + """The ``sh -c`` line that runs *command* as root, and its stdin prefix. + + Root runs it directly; with a sudo password, ``sudo -S`` reads it from + stdin; without one, ``sudo -n`` fails at once instead of prompting. + """ + script = _shlex_quote(command) + if self._is_root(): + return f"sh -c {script}", None if self._sudo_password: - line = f"sudo -S -p '' sh -c {_shlex_quote(command)}" - return line, f"{self._sudo_password}\n".encode() - return f"sudo -n sh -c {_shlex_quote(command)}", None + return f"sudo -S -p '' sh -c {script}", f"{self._sudo_password}\n".encode() + return f"sudo -n sh -c {script}", None def run_command( self, @@ -2075,15 +2110,16 @@ class LinuxDriver( tunnels: dict[str, Any] = {} # ── Attempt 1: wg show all dump via sudo ───────────────────────────── - # Write output to a fixed temp file to preserve literal tab characters. - # PTY output processing expands tabs to spaces, breaking split("\t"). - import hashlib as _hashlib - _tmp = "/tmp/.netork_wg_" + _hashlib.md5(self.hostname.encode()).hexdigest()[:8] + # With a sudo password it runs on the exec channel, which keeps the + # tabs. The PTY expands them to spaces, breaking split("\t"), so + # without one the output goes through a temp file. if self._sudo_password: - self._sudo(f"wg show all dump > {_tmp} 2>&1 || true") + raw = self._sudo("wg show all dump 2>&1 || true") else: + import hashlib as _hashlib + _tmp = "/tmp/.netork_wg_" + _hashlib.md5(self.hostname.encode()).hexdigest()[:8] self._send(f"sudo -n wg show all dump > {_tmp} 2>&1 || true") - raw = self._send(f"cat {_tmp} 2>/dev/null || true; rm -f {_tmp}").strip() + raw = self._send(f"cat {_tmp} 2>/dev/null || true; rm -f {_tmp}").strip() _perm_errors = ( "operation not permitted", "permission denied", diff --git a/tests/test_linux.py b/tests/test_linux.py index cbb6053..0dad30b 100644 --- a/tests/test_linux.py +++ b/tests/test_linux.py @@ -4,7 +4,7 @@ import re import pytest from unittest.mock import MagicMock, patch -from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model +from napalm_linux.linux import LinuxDriver, _arm_vendor_from_model, _split_status # --------------------------------------------------------------------------- @@ -31,9 +31,75 @@ def driver(): d._sudo_password = None d.netmiko_optional_args = {} d._device = MagicMock() + d._device.remote_conn_pre.get_transport.return_value = _ShellAnswers(d) return d +_SCRIPTED_RC = re.compile(r"\n?^__NETORK_RC=(-?\d+)\s*$", re.MULTILINE) + + +class _ShellAnswers: + """The SSH transport of the fixture: an exec channel answers what ``_send`` + answers for the same command line, so a test that scripts the shell's + output scripts ``_sudo``'s too (it runs on an exec channel since + NetOrk/netork#1039). It refuses a line holding the sudo password. + + The exit status is 0, or what a scripted ``__NETORK_RC=`` line says when + the command did not echo that marker itself (``_with_rc``); that line is + then no output. ``test_sudo_off_the_command_line`` covers the channel itself.""" + + def __init__(self, driver): + self._driver = driver + + def open_session(self): + return _AnsweringChannel(self._driver) + + +class _AnsweringChannel: + def __init__(self, driver): + self._driver = driver + self._out = b"" + self._status = 0 + + def exec_command(self, command): + password = self._driver._sudo_password + assert not password or password not in command, f"sudo password in {command!r}" + output = self._driver._send(command) + marker = _SCRIPTED_RC.search(output) + if marker and "echo __NETORK_RC" not in command: + self._status = int(marker.group(1)) + output = _SCRIPTED_RC.sub("", output) + self._out = output.encode() + + def settimeout(self, timeout): + pass + + def sendall(self, data): + pass + + 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 exit_status_ready(self): + return not self._out + + def recv_exit_status(self): + return self._status + + def _mock_send(driver_fixture, output: str): """Patch _send to return *output* for any command.""" driver_fixture._device.send_command.return_value = output @@ -915,9 +981,10 @@ class TestUninstallPackage: def test_a_half_configured_package_falls_back_to_dpkg(self, driver): """`install ok unpacked` is the state apt cannot get out of. On one host only `dpkg --purge --force-all` removed it.""" + driver._root = False driver._device.send_command.side_effect = [ - "E: Sub-process /usr/bin/dpkg returned an error code (1)", - "Removing wazuh-agent (4.14.7-1) ...", + _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100), + _with_rc("Removing wazuh-agent (4.14.7-1) ...", 0), ] result = driver.uninstall_package("wazuh-agent", purge=True) @@ -929,6 +996,7 @@ class TestUninstallPackage: def test_the_fallback_is_not_tried_when_the_first_pass_worked(self, driver): """A forced dpkg purge is a bigger hammer than apt and must stay a last resort, not a routine second step.""" + driver._root = False _mock_send(driver, "Removing wazuh-agent ...") driver.uninstall_package("wazuh-agent", purge=True) @@ -973,23 +1041,25 @@ class TestSudoStatus: assert driver._sudo_status("apt-get remove -y nope")[1] == 100 - def test_the_status_is_read_right_after_sudo_returns(self, driver): - """``$?`` must be read straight after the sudo pipeline — with an - ``|| true`` in between, every command would report 0.""" - driver._sudo_password = "pw" # noqa: S105 + def test_the_command_keeps_its_own_exit_status(self, driver): + """The exec channel reports the status (#1039): no ``|| true`` and no + marker in between, and no password in the line.""" + driver._root = False + driver._sudo_password = "s3cret" # noqa: S105 _mock_send(driver, _with_rc("", 0)) driver._sudo_status("apt-get remove -y x 2>&1") sent = driver._device.send_command.call_args[0][0] - assert sent.startswith("echo pw | sudo -S") - assert sent.endswith("apt-get remove -y x 2>&1; echo __NETORK_RC=$?") + assert sent.startswith("sudo -S -p '' sh -c ") + assert "apt-get remove -y x 2>&1" in sent assert "|| true" not in sent + assert "__NETORK_RC" not in sent - def test_a_missing_marker_means_unknown_not_success(self, driver): - """Output cut short before the marker arrived says nothing about the - exit status; ``None`` says so instead of guessing 0.""" - _mock_send(driver, "Removing wazuh-agent ...") + def test_a_channel_closed_without_a_status_means_unknown_not_success(self, driver): + """paramiko's -1 says nothing about the exit status; ``None`` says so + instead of guessing 0.""" + _mock_send(driver, _with_rc("Removing wazuh-agent ...", -1)) assert driver._sudo_status("apt-get remove -y wazuh-agent") == ( "Removing wazuh-agent ...", @@ -1042,6 +1112,9 @@ _BADSIG_CHUNKS = [ ] +_TRUE_WITH_STATUS = "true; echo __NETORK_RC=$?" + + class TestReadToTheEnd: """A line the command prints can end in ``>``, ``#`` or ``$`` -- apt's ```` after a bad signature. Taken for the prompt, it @@ -1064,7 +1137,7 @@ class TestReadToTheEnd: driver._device = _Channel(_BADSIG_CHUNKS + ["true\n", "__NETORK_RC=0\nchris@vault-01:~$ "]) driver.refresh_available_updates() - output, status = driver._sudo_status("true") + output, status = _split_status(driver._send(_TRUE_WITH_STATUS)) assert status == 0 assert "BADSIG" not in output @@ -1074,7 +1147,7 @@ class TestReadToTheEnd: channel = _Channel(["sudo true; echo __NETORK_RC=$?\n", "__NETORK_RC=0\nchris@vault-01:~$ "]) driver._device = channel - assert driver._sudo_status("true")[1] == 0 + assert _split_status(driver._send(_TRUE_WITH_STATUS))[1] == 0 assert channel.chunks == [] def test_the_marker_alone_is_not_the_end(self, driver): @@ -1082,7 +1155,7 @@ class TestReadToTheEnd: channel = _Channel(["out\n__NETORK_RC=0\n", "chris@vault-01:~$ "]) driver._device = channel - driver._sudo_status("true") + driver._send(_TRUE_WITH_STATUS) assert channel.chunks == [] @@ -1158,10 +1231,10 @@ class TestUninstallExitStatus: driver.uninstall_package("wazuh-agent") sent = driver._device.send_command.call_args[0][0] - assert "|| true" not in sent - assert sent.endswith("; echo __NETORK_RC=$?") + assert "|| true" not in sent # the exec channel reports the status (#1039) def test_apt_failing_by_exit_status_falls_back_to_dpkg(self, driver): + driver._root = False driver._device.send_command.side_effect = [ _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100), _with_rc("Removing wazuh-agent (4.14.7-1) ...", 0), @@ -1176,6 +1249,7 @@ class TestUninstallExitStatus: assert "__NETORK_RC" not in result["output"] def test_the_dpkg_fallback_failing_is_a_failure(self, driver): + driver._root = False driver._device.send_command.side_effect = [ _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100), _with_rc("dpkg: error processing package wazuh-agent (--purge):", 1), @@ -1190,6 +1264,7 @@ class TestUninstallExitStatus: """Even when the output contains words that used to mean failure: apt exits 0 for a package that is already gone, which is the state the caller asked for.""" + driver._root = False _mock_send(driver, _with_rc("Package 'x' is not installed, so not removed", 0)) result = driver.uninstall_package("x", purge=True) @@ -1198,10 +1273,10 @@ class TestUninstallExitStatus: assert driver._device.send_command.call_count == 1 def test_without_an_exit_status_the_output_is_read_as_before(self, driver): - """If the marker never arrived, the keyword check is still the best - answer available — and it errs towards failure on apt's ``E:``.""" + """If the channel closed without a status, the keyword check is still the + best answer available — and it errs towards failure on apt's ``E:``.""" driver._pkg_manager = "dnf" - _mock_send(driver, "E: Sub-process /usr/bin/dpkg returned an error code (1)") + _mock_send(driver, _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", -1)) result = driver.uninstall_package("wazuh-agent") @@ -1295,7 +1370,7 @@ class TestManageService: assert driver.manage_service("cron", "stop")["success"] is True action = self._sent(driver)[1] - assert action.startswith("echo pw | sudo -S") + assert action.startswith("sudo -S -p '' sh -c ") # the password on stdin (#1039) assert "timeout 45 systemctl --no-ask-password stop -- cron.service" in action def test_without_one_sudo_never_waits_for_a_password(self, driver): @@ -1562,8 +1637,7 @@ class TestGetListeningSockets: driver.get_listening_sockets() sent = self._sent(driver)[0] - assert sent.startswith("echo pw | sudo -S") - assert "sh -c '" in sent + assert sent.startswith("sudo -S -p '' sh -c ") # the password on stdin (#1039) def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver): driver._root = False diff --git a/tests/test_sudo_off_the_command_line.py b/tests/test_sudo_off_the_command_line.py new file mode 100644 index 0000000..5e49213 --- /dev/null +++ b/tests/test_sudo_off_the_command_line.py @@ -0,0 +1,258 @@ +"""The sudo password never goes into a command line or a shell history (NetOrk/netork#1039). + +``_sudo`` typed ``echo '' | sudo -S -p "" `` 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 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 &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)