diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d28d49..8fd41df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,33 @@ 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. +- The snmpd and apt proxy fixes no longer stage their files at a fixed path in + /tmp (`/tmp/netork_snmpd.conf`, `/tmp/netork_00proxy`) and then `mv`, + `chown` and `chmod` them as root. Another local user could create that path + first as a symlink, for example to `/etc/shadow`, and root's `chown`/`chmod` + followed it. The content now goes to a fresh `mktemp` file under `umask 077`, + written without privilege, and root `install`s it as a new file owned by root + (mode 644) next to the target and renames it into place + (`_install_file_as_root`). Nothing runs as root when staging failed. +- `get_vpn_tunnels()` without a sudo password no longer writes `wg show all dump` + (which includes the interfaces' private keys) to the predictable + `/tmp/.netork_wg_`; the dump comes straight from the exec channel. ## [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..c35f79b 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,25 @@ _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 "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }' +) + #: 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 +326,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 +375,61 @@ 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 _install_file_as_root(self, path: str, content: str) -> CommandResult: + """Replace *path* with *content*, as a new file owned by root, mode 644. + + The content is staged without privilege in a fresh ``mktemp`` file only + the login user may read; its stdin is the content's, not a privileged + command's (``_PRIVILEGED_STDIN``). Root then ``install``s that file as a + new regular file next to *path* and renames it over *path*: nothing root + runs follows a symlink or changes the owner or mode of an existing path + (NetOrk/netork#1039). Nothing runs as root when staging failed. + """ + staged = self.run_command(_STAGE, timeout=30, stdin=content.encode()) + lines = staged.stdout.strip().splitlines() + tmp = lines[-1] if lines else "" + if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]: + reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}" + return CommandResult(reason, "", staged.exit_code or 1) + tmp_q, part_q, path_q = (_shlex_quote(p) for p in (tmp, f"{path}.netork-tmp", path)) + return self._sudo_result( + f"install -m 644 -o root -g root -- {tmp_q} {part_q} && mv -f -- {part_q} {path_q};" + f" s=$?; rm -f -- {tmp_q} {part_q}; exit $s", + 30, + ) 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 +458,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 +482,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, @@ -1953,25 +2020,24 @@ class LinuxDriver( except Exception: pass - # Write snmpd.conf: - # 1. Write to /tmp (no sudo needed, avoids stdin conflict with sudo -S) - # 2. sudo mv to /etc/snmp/snmpd.conf + # Write snmpd.conf: staged unprivileged, installed by root (#1039). # agentAddress udp:161 overrides Debian's localhost-only default. - import base64 as _b64 conf_str = ( "agentAddress udp:161\n" "rocommunity public\n" "sysLocation Managed by netOrk\n" "sysContact netork@localhost\n" ) - conf_b64 = _b64.b64encode(conf_str.encode()).decode() - self._send(f"echo {conf_b64} | base64 -d > /tmp/netork_snmpd.conf") - self._sudo("mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && chown root:root /etc/snmp/snmpd.conf && chmod 644 /etc/snmp/snmpd.conf") + try: + written = self._install_file_as_root("/etc/snmp/snmpd.conf", conf_str) + install_error = written.stdout.strip() if written.exit_code else "" + except Exception as exc: + install_error = str(exc) verify = self._send("cat /etc/snmp/snmpd.conf 2>/dev/null").strip() if "agentAddress" in verify and "rocommunity" in verify: lines.append("[config] Wrote /etc/snmp/snmpd.conf — agentAddress udp:161, rocommunity public.") else: - lines.append(f"[warn] snmpd.conf write may have failed: {verify[:100]}") + lines.append(f"[warn] snmpd.conf write may have failed: {(install_error or verify)[:100]}") # 3. Open firewall for SNMP (UDP 161) — restrict to netOrk's source IP if netork_ip: @@ -2040,7 +2106,6 @@ class LinuxDriver( def _action_fix_apt_proxy(self) -> DeviceActionResultDict: """Write /etc/apt/apt.conf.d/00proxy with the configured proxy URL.""" - import base64 as _b64 proxy_url = self._apt_proxy_url if not proxy_url: return {"success": False, "output": "No apt_proxy_url configured."} @@ -2048,18 +2113,16 @@ class LinuxDriver( return {"success": False, "output": f"Package manager is {self._pkg_manager!r}, not apt — skipping."} content = f'Acquire::http::Proxy "{proxy_url}";\n' - content_b64 = _b64.b64encode(content.encode()).decode() - self._send(f"echo {content_b64} | base64 -d > /tmp/netork_00proxy") - self._sudo( - "mv /tmp/netork_00proxy /etc/apt/apt.conf.d/00proxy && " - "chown root:root /etc/apt/apt.conf.d/00proxy && " - "chmod 644 /etc/apt/apt.conf.d/00proxy" - ) + # Staged unprivileged, installed by root (#1039). + written = self._install_file_as_root("/etc/apt/apt.conf.d/00proxy", content) verify = self._send("cat /etc/apt/apt.conf.d/00proxy 2>/dev/null").strip() success = proxy_url in verify if success: return {"success": True, "output": f"Wrote /etc/apt/apt.conf.d/00proxy — proxy: {proxy_url}"} - return {"success": False, "output": f"Write may have failed. File content: {verify[:200]}"} + output = f"Write may have failed. File content: {verify[:200]}" + if written.exit_code: + output = f"{output}\n{written.stdout.strip()}" + return {"success": False, "output": output} def get_vpn_tunnels(self) -> dict[str, Any]: """Return WireGuard status via ``wg show all dump`` (requires root/sudo). @@ -2074,16 +2137,11 @@ 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] - 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() + # ── Attempt 1: wg show all dump as root ────────────────────────────── + # On the exec channel, which keeps the tabs the PTY expands to spaces. + # It went through /tmp/.netork_wg_ before: a predictable name, + # holding the interfaces' private keys (#1039). + raw = self._sudo("wg show all dump 2>&1 || true") _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_staged_root_writes.py b/tests/test_staged_root_writes.py new file mode 100644 index 0000000..9e4f395 --- /dev/null +++ b/tests/test_staged_root_writes.py @@ -0,0 +1,246 @@ +"""Files root writes are staged in a fresh file, never at a fixed path in /tmp (NetOrk/netork#1039). + +The snmpd and apt proxy fixes wrote their content to ``/tmp/netork_snmpd.conf`` +or ``/tmp/netork_00proxy`` as the login user, then ran ``mv … && chown root:root +… && chmod 644 …``. Once the whole of a ``_sudo`` command ran as root, that was +a local privilege escalation: another user creates the fixed path first as a +symlink to ``/etc/shadow``, the login user's write fails (protected_symlinks), +root moves the symlink into place, and root's ``chown``/``chmod`` follow it -- +``/etc/shadow`` mode 644. + +Now the content goes to a ``mktemp`` file under ``umask 077``, unprivileged, and +root ``install``s that file as a new regular file next to the target, then +renames it over the target. Nothing root runs changes the owner or mode of a +path that could be a symlink. +""" + +from __future__ import annotations + +import re +import shlex +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest + +from napalm_linux import LinuxDriver + +_PW = "pw-1039" +_STAGED = "/tmp/netork-stage.Ab12Cd" +#: What another local user planted where the fixes used to stage their content. +_PLANTED = { + "/tmp/netork_snmpd.conf": "/etc/shadow", + "/tmp/netork_00proxy": "/etc/shadow", +} + + +class Host: + """The host's exec channels: every line, its stdin, and whether it ran as root.""" + + def __init__(self, *, stage_output=_STAGED, stage_status=0): + self.runs: list[SimpleNamespace] = [] + self._stage = (stage_output, stage_status) + + def open_session(self): + return _Channel(self) + + def answer(self, command: str) -> tuple[str, int]: + if "mktemp" in command: + return self._stage + return "", 0 + + @property + def root_scripts(self) -> list[str]: + return [run.script for run in self.runs if run.root] + + @property + def unprivileged(self) -> list[SimpleNamespace]: + return [run for run in self.runs if not run.root] + + +class _Channel: + def __init__(self, host: Host): + self._host = host + self._run: SimpleNamespace | None = None + self._out = b"" + + def exec_command(self, command): + assert _PW not in command + root = command.startswith(("sudo ", "sh -c ")) + script = shlex.split(command)[shlex.split(command).index("-c") + 1] if root else command + self._run = SimpleNamespace(line=command, script=script, root=root, stdin=b"") + self._host.runs.append(self._run) + output, self._status = self._host.answer(command) + self._out = output.encode() + + def settimeout(self, timeout): + pass + + def sendall(self, data): + self._run.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 + + +def _shell(command: str, **_kwargs) -> str: + """The interactive shell, for what the fixes still ask it.""" + if command.startswith("command -v apt"): + return "/usr/bin/apt" + if command.startswith("cat /etc/snmp/snmpd.conf"): + return "agentAddress udp:161\nrocommunity public\n" + if command.startswith("cat /etc/apt/apt.conf.d/00proxy"): + return 'Acquire::http::Proxy "http://proxy:3142";' + if command.startswith("snmpget"): + return "STRING: Linux" + return "" + + +def _driver(host: Host) -> LinuxDriver: + driver = LinuxDriver( + "h", + "admin", + "p", + optional_args={"sudo_password": _PW, "apt_proxy_url": "http://proxy:3142"}, + ) + driver._device = MagicMock() + driver._device.send_command.side_effect = _shell + driver._device.remote_conn_pre = SimpleNamespace(get_transport=lambda: host) + driver._root = False + driver._pkg_manager = "apt" + return driver + + +def _every_line(driver: LinuxDriver, host: Host) -> list[str]: + typed = [c.args[0] for c in driver._device.send_command.call_args_list] + return typed + [run.line for run in host.runs] + + +ACTIONS = [ + ("_action_fix_snmp", "/etc/snmp/snmpd.conf", b"agentAddress udp:161"), + ( + "_action_fix_apt_proxy", + "/etc/apt/apt.conf.d/00proxy", + b'Acquire::http::Proxy "http://proxy:3142";', + ), +] + + +@pytest.fixture(autouse=True) +def _no_waiting(): + with patch("time.sleep"): + yield + + +@pytest.mark.parametrize("action, target, content", ACTIONS) +class TestStagedWrite: + def test_the_old_fixed_path_is_never_touched(self, action, target, content): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + for planted in _PLANTED: + assert not any(planted in line for line in _every_line(driver, host)) + + def test_the_content_is_staged_unprivileged_in_a_fresh_private_file( + self, action, target, content + ): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + stage = next(run for run in host.unprivileged if "mktemp" in run.line) + assert "umask 077" in stage.line + assert "mktemp /tmp/netork-stage.XXXXXX" in stage.line + assert content in stage.stdin + + def test_root_installs_a_new_file_and_renames_it_over_the_target(self, action, target, content): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + install = next(script for script in host.root_scripts if "install -m" in script) + part = f"{target}.netork-tmp" + assert f"install -m 644 -o root -g root -- {_STAGED} {part}" in install + assert f"mv -f -- {part} {target}" in install + assert f"rm -f -- {_STAGED} {part}" in install + + def test_nothing_root_runs_changes_owner_or_mode_of_a_path(self, action, target, content): + host = Host() + driver = _driver(host) + + getattr(driver, action)() + + assert not any(re.search(r"\bch(own|mod)\b", script) for script in host.root_scripts) + + def test_nothing_runs_as_root_when_staging_failed(self, action, target, content): + host = Host(stage_output="mktemp: No space left on device\n", stage_status=1) + driver = _driver(host) + + getattr(driver, action)() + + assert not any("install -m" in script for script in host.root_scripts) + + def test_a_staged_path_it_did_not_ask_for_is_refused(self, action, target, content): + host = Host(stage_output="/etc/shadow") + driver = _driver(host) + + getattr(driver, action)() + + assert not any("/etc/shadow" in script for script in host.root_scripts) + + +def test_a_failed_install_is_reported(): + host = Host() + host.answer = lambda command: ( # type: ignore[method-assign] + (_STAGED, 0) if "mktemp" in command else ("install: cannot create\n", 1) + ) + driver = _driver(host) + driver._device.send_command.side_effect = lambda command, **_k: "" + + result = driver._action_fix_apt_proxy() + + assert result["success"] is False + assert "install: cannot create" in result["output"] + + +class TestWireGuardWithoutASudoPassword: + """It went through ``/tmp/.netork_wg_``: a name anyone can predict, and + a file holding the interfaces' private keys.""" + + def test_no_temp_file_and_the_dump_comes_from_the_exec_channel(self): + host = Host() + dump = ( + "wg0\tPRIV\tPUB\t51820\toff\nwg0\tPEER\t(none)\t1.2.3.4:1\t10.0.0.2/32\t0\t1\t2\toff\n" + ) + host.answer = lambda command: (dump, 0) # type: ignore[method-assign] + driver = _driver(host) + driver._sudo_password = None + + tunnels = driver.get_vpn_tunnels() + + assert list(tunnels) == ["wireguard-wg0-PEER"] + assert not any("/tmp" in line for line in _every_line(driver, host)) + assert host.runs[0].line.startswith("sudo -n sh -c ") 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)