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 '<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.
+- 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_<hash>`; the dump comes straight from the exec channel.
 
 ## [0.4.0] – 2026-10-08
 

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 <action> -- <unit>.service`, so a unit that hangs

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
+#: Printed after a command run through the interactive shell (``_run_privileged``
-#: trip through an interactive shell. Matched only on a line of its own with a
+#: 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 </dev/null; "
+
+#: Where ``_install_file_as_root`` stages content: a fresh file from ``mktemp``
+#: under ``umask 077``, never a fixed name another user could create first.
+_STAGE_TEMPLATE = "/tmp/netork-stage.XXXXXX"
+_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
+_STAGE = (
+    f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
+    '{ cat > "$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 <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:
+        return self._sudo_result(command, read_timeout).stdout.strip()
-            wrapped = f'echo {_shlex_quote(self._sudo_password)} | sudo -S -p "" {command}'
+
-            return self._send(wrapped, read_timeout=read_timeout)
+    def _sudo_result(self, command: str, timeout: float) -> CommandResult:
-        return self._send(f'sudo {command}', read_timeout=read_timeout)
+        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
+        instead of an error, which throws the exit status away. The exec channel
-        echoes ``$?`` straight after the sudo pipeline instead — sudo passes
+        reports the status itself: the command's, or sudo's when it refused.
-        the command's status through, and a failed password is non-zero too.
 
-        The status is ``None`` when the marker never arrived (output cut short),
+        The status is ``None`` when the channel closed without one (paramiko's
-        so a caller can tell "unknown" from "succeeded".
+        -1), so a caller can tell "unknown" from "succeeded".
         """
-        return _split_status(
+        result = self._sudo_result(command, read_timeout)
-            self._sudo(f"{command}; echo {_RC_MARKER}$?", read_timeout=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 f"sudo -S -p '' sh -c {script}", f"{self._sudo_password}\n".encode()
-            return line, f"{self._sudo_password}\n".encode()
+        return f"sudo -n sh -c {script}", None
-        return f"sudo -n sh -c {_shlex_quote(command)}", None
 
     def run_command(
         self,
@@ -1953,25 +2020,24 @@ class LinuxDriver(
         except Exception:
             pass
 
-        # Write snmpd.conf:
+        # Write snmpd.conf: staged unprivileged, installed by root (#1039).
-        # 1. Write to /tmp (no sudo needed, avoids stdin conflict with sudo -S)
-        # 2. sudo mv to /etc/snmp/snmpd.conf
         # 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()
+        try:
-        self._send(f"echo {conf_b64} | base64 -d > /tmp/netork_snmpd.conf")
+            written = self._install_file_as_root("/etc/snmp/snmpd.conf", conf_str)
-        self._sudo("mv /tmp/netork_snmpd.conf /etc/snmp/snmpd.conf && chown root:root /etc/snmp/snmpd.conf && chmod 644 /etc/snmp/snmpd.conf")
+            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()
+        # Staged unprivileged, installed by root (#1039).
-        self._send(f"echo {content_b64} | base64 -d > /tmp/netork_00proxy")
+        written = self._install_file_as_root("/etc/apt/apt.conf.d/00proxy", content)
-        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"
-        )
         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 ─────────────────────────────
+        # ── Attempt 1: wg show all dump as root ──────────────────────────────
-        # Write output to a fixed temp file to preserve literal tab characters.
+        # On the exec channel, which keeps the tabs the PTY expands to spaces.
-        # PTY output processing expands tabs to spaces, breaking split("\t").
+        # It went through /tmp/.netork_wg_<hash> before: a predictable name,
-        import hashlib as _hashlib
+        # holding the interfaces' private keys (#1039).
-        _tmp = "/tmp/.netork_wg_" + _hashlib.md5(self.hostname.encode()).hexdigest()[:8]
+        raw = self._sudo("wg show all dump 2>&1 || true")
-        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()
 
         _perm_errors = (
             "operation not permitted", "permission denied",

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=<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)",
+            _with_rc("E: Sub-process /usr/bin/dpkg returned an error code (1)", 100),
-            "Removing wazuh-agent (4.14.7-1) ...",
+            _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):
+    def test_the_command_keeps_its_own_exit_status(self, driver):
-        """``$?`` must be read straight after the sudo pipeline — with an
+        """The exec channel reports the status (#1039): no ``|| true`` and no
-        ``|| true`` in between, every command would report 0."""
+        marker in between, and no password in the line."""
-        driver._sudo_password = "pw"  # noqa: S105
+        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.startswith("sudo -S -p '' sh -c ")
-        assert sent.endswith("apt-get remove -y x 2>&1; echo __NETORK_RC=$?")
+        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):
+    def test_a_channel_closed_without_a_status_means_unknown_not_success(self, driver):
-        """Output cut short before the marker arrived says nothing about the
+        """paramiko's -1 says nothing about the exit status; ``None`` says so
-        exit status; ``None`` says so instead of guessing 0."""
+        instead of guessing 0."""
-        _mock_send(driver, "Removing wazuh-agent ...")
+        _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 "|| true" not in sent  # the exec channel reports the status (#1039)
-        assert sent.endswith("; echo __NETORK_RC=$?")
 
     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
+        """If the channel closed without a status, the keyword check is still the
-        answer available — and it errs towards failure on apt's ``E:``."""
+        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 sent.startswith("sudo -S -p '' sh -c ")  # the password on stdin (#1039)
-        assert "sh -c '" in sent
 
     def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
         driver._root = False

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_<hash>``: 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 ")

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 '<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)