fix: run privileged commands on the exec channel, never typed into the shell #25

Merged
christianmanivong merged 2 commits from fix/sudo-not-in-command-1039 into master 2026-10-12 21:00:59 +00:00
5 changed files with 439 additions and 50 deletions
Showing only changes of commit cf8ee38a90 - Show all commits

No files matched your search

+16
View File
@@ -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 '<password>' | sudo -S -p "" <command>`: 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
+5
View File
@@ -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 <action> -- <unit>.service`, so a unit that hangs
+62 -26
View File
@@ -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; "
#: What to do when sudo wants a password netOrk does not have.
_SUDO_PASSWORD_HINT = (
"sudo requires a password on this device but none is configured in netOrk. "
@@ -307,6 +317,12 @@ class LinuxDriver(
except NetmikoTimeoutException as exc:
raise ConnectionException(str(exc)) from exc
# No history before anything else is typed (NetOrk/netork#1039).
try:
self._device.send_command(_HISTORY_OFF, expect_string=_PROMPT_RE)
except Exception:
logger.debug("Could not switch the shell history off on %s", self.hostname, exc_info=True)
# Prevent PTY from wrapping long output lines (e.g. docker JSON).
try:
self._device.send_command("stty cols 10000 2>/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 <password> |
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,13 +2110,14 @@ 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").
# 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:
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]
if self._sudo_password:
self._sudo(f"wg show all dump > {_tmp} 2>&1 || true")
else:
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()
+98 -24
View File
@@ -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=<n>`` 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
``<ftpmaster@ubuntu.com>`` 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
+258
View File
@@ -0,0 +1,258 @@
"""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 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)