fix: keep systemd's OSC 3008 shell context out of command output (#41)

Backport of the 0.15.1 fix onto 0ba5081, the commit netOrk pins, which
cannot take 0.15.x before it moves to napalm-device-types 4.22.

systemd 258 (Ubuntu 26.04) ships /etc/profile.d/80-systemd-osc-context.sh:
an interactive bash wraps every command in OSC 3008 sequences from PS0 and
PROMPT_COMMAND, each with a fresh UUID. They ended up in every value read
through netmiko's shell, and because the prompt never looked the same twice,
the prompt line stayed in the output too. os_version grew past 255
characters and netOrk could not store the poll.

The session now unsets PS0 and PROMPT_COMMAND right after switching the
history off and reads its prompt anew; _send() strips terminal codes from
what it returns.

CHANGELOG.md

@@ -7,6 +7,98 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- `get_host_firewall()` (napalm-device-types' `HostFirewallMixin`, NetOrk/netork#1344): the host's own netfilter rules — `nft -j list ruleset`, with the tables iptables manages taken from `iptables-save` — as chains and rules in one shape. Read-only, as root, on an exec channel (the nft JSON is one long line). Requires napalm-device-types 4.21.0; version 0.13.0.
+- `get_service_packages()` (napalm-device-types' `ServicePackagesMixin`, NetOrk/netork#1236): which installed packages each systemd service is made of — the one shipping its unit file and the one shipping its `ExecStart` binary. Read-only, on an exec channel under `sh`, without sudo. Requires napalm-device-types 4.20.0; version 0.12.0.
+
+### Security
+- The SSH host key is verified (NetOrk/netork#1167). `open()` hands netmiko the key the caller expects for the port in `optional_args["ssh_host_keys"]` (as a `known_hosts` file that exists only while it connects, with `ssh_strict`), so any other key is refused before the login (`SshHostKeyMismatch`); without one netmiko accepts the first key. Every key seen goes to `optional_args["ssh_host_key_observer"]`. OpenMediaVault and QNAP get this through `LinuxDriver.open()`. Requires napalm-device-types 4.19.0; version 0.11.0.
+
+### Changed
+- **Breaking:** `get_device_warnings` is gone, and with it the `apt_proxy_url` argument.
+  The driver reports facts and the application names warnings (napalm-device-types 4.14):
+  `get_package_proxy()` says which proxy apt fetches through; pending updates were always
+  in `get_available_updates()`. Requires napalm-device-types 4.14.0; version 0.9.0.
+
+### Changed
+
+- **Breaking:** `run_device_action` is gone, with `fix_snmp`, `fix_apt_proxy`,
+  `fix_docker_permissions` and the unused `apt_update_upgrade`. The driver implements
+  napalm-device-types 4.13's fix templates instead, and the caller supplies every value:
+  `configure_snmp_agent(settings)` (one `rocommunity` line per allowed source, udp/161
+  opened in ufw or iptables for those sources only, each step judged by its exit
+  status), `configure_package_proxy(url)`, `grant_container_engine_access(user)` and
+  `ensure_package(name)`. No community, contact, location or address of an application
+  is built in any more. Requires napalm-device-types 4.13.0 or later; version 0.8.0.
+- The sudo hint names no application's settings.
+
+### Fixed
+
+- The type annotations resolve: `Dict`, `List` and `Optional` were used without an
+  import (#29).
+
+### Added
+
+- `get_system_identity()` (napalm-device-types 4.9 `SystemIdentityMixin`): the host's
+  SMBIOS system UUID, product serial and board serial, read as root on an exec channel,
+  vendor placeholders empty. netOrk matches a server to its BMC on these. Requires
+  napalm-device-types 4.9.0 or later; version 0.7.0.
+
+### Changed
+
+- Requires napalm-device-types 4.7.1 or later; version 0.6.0.
+- **Breaking:** `get_users()` is now `get_os_users()`, following napalm-device-types
+  4.7, which renamed `OSDriver.get_users` because NAPALM's own `get_users` has
+  another shape. The return value is unchanged (a list of `UserDict`).
+  `get_users()` is NAPALM's again and raises `NotImplementedError`.
+
+### Fixed
+
+- Listing updates on yum hosts (RHEL 7, CentOS 7) no longer fails with
+  napalm-device-types 4.7: its `parse_dnf_security` needs the `__DNF_RC=<status>`
+  line `DNF_SECURITY_COMMAND` prints, and the yum security query had none, so it
+  raised `ValueError`. The yum query now prints that marker the same way
+  (`{ ...; echo __DNF_RC=$?; } | cat`). A yum or dnf security query that failed,
+  or whose output was cut short before the marker, still leaves `security` unknown
+  (`None`) instead of reporting no advisories.
+
+### Security
+
+- `run_command(privileged=True, stdin=...)` raises `ValueError` while a sudo
+  password is set. `sudo -S` reads the password from stdin, and a sudo that does
+  not prompt (NOPASSWD, a cached timestamp) left it for the command, which read
+  it as the first line of the data: netOrk wrote it into a file that way
+  (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
 
 ### Added

README.md

@@ -60,7 +60,7 @@ with Driver(
     packages = dev.get_packages()
     updates  = dev.get_pending_updates()
     services = dev.get_services()
-    users    = dev.get_users()
+    users    = dev.get_os_users()
     procs    = dev.get_processes()
     jobs     = dev.get_cron_jobs()
 
@@ -106,7 +106,7 @@ with Driver(
 | `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman |
 | `get_services()` | ✅ | systemd, one round trip (fallback: SysV `service`) |
 | `manage_service(name, action)` | ✅ | systemd: start, stop, restart, enable, disable |
-| `get_users()` | ✅ | `/etc/passwd` + `/etc/group` |
+| `get_os_users()` | ✅ | `/etc/passwd` + `/etc/group` |
 | `get_processes()` | ✅ | `ps axo` |
 | `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` |
 
@@ -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

@@ -18,11 +18,12 @@ A specific package manager can be forced with
 
 from __future__ import annotations
 
+import ipaddress
 import logging
 import re
 import socket
 from shlex import quote as _shlex_quote
-from typing import Any
+from typing import Any, Dict, List, Optional
 
 from netmiko import ConnectHandler
 from netmiko.exceptions import (
@@ -37,29 +38,40 @@ from napalm_device_types import (
     ByteStream,
     CommandResult,
     ContainerEngineMixin,
+    EnsurePackageMixin,
     FingerprintRule,
+    HostFirewallMixin,
     HostStatusMixin,
     KernelFactsMixin,
     ListeningSocketsMixin,
     OpenSshServerMixin,
     OSDriver,
+    PackageProxyMixin,
+    ServiceClustersMixin,
+    ServicePackagesMixin,
+    SnmpAgentMixin,
+    SystemIdentityMixin,
     SystemdServicesMixin,
     SystemdUnavailable,
     VirtualIpsMixin,
     parse_apt_upgradable,
     open_stream_on_transport,
     parse_dnf_security,
+    run_fix_steps,
     run_on_transport,
     strip_terminal_codes,
 )
+from napalm_device_types.ssh_host_keys import HostKeyCheck
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
-    DeviceActionResultDict,
+    FixResultDict,
+    FixStepOutcomeDict,
     PackageDict,
     ProcessDict,
     ServiceDict,
     SNMPConfigDict,
+    SnmpAgentSettingsDict,
     UpdateDict,
     UserDict,
 )
@@ -69,8 +81,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)
@@ -125,18 +137,59 @@ _REFRESH = {
     "yum": "yum makecache -q 2>&1",
     "apk": "apk update -q 2>&1",
 }
-_YUM_SECURITY_COMMAND = "LC_ALL=C yum updateinfo list security -q 2>/dev/null"
+#: yum's counterpart of DNF_SECURITY_COMMAND, in the same shape: parse_dnf_security
+#: (napalm-device-types 4.7) needs the ``__DNF_RC=`` line, and behind ``| cat`` it is
+#: the only place yum's own exit status survives.
+_YUM_SECURITY_COMMAND = (
+    "{ LC_ALL=C yum updateinfo list security -q 2>/dev/null; echo __DNF_RC=$?; } | cat"
+)
+#: The exit status ``DNF_SECURITY_COMMAND`` and ``_YUM_SECURITY_COMMAND`` report.
+_DNF_RC_RE = re.compile(r"^__DNF_RC=(\d+)\s*$", re.MULTILINE)
 
 #: Restart the host two seconds later, detached from this session: the launcher's
 #: exit status comes back before the host goes down, and closing the session
 #: cannot take the restart with it.
 _REBOOT_DETACHED = "sh -c '(trap \"\" HUP; sleep 2; /sbin/reboot) </dev/null >/dev/null 2>&1 &'"
 
+#: Why ``run_command`` refuses stdin to a privileged command while a sudo
+#: password is set (NetOrk/netork#1029).
+_PRIVILEGED_STDIN = (
+    "run_command(privileged=True) takes no stdin while a sudo password is set: "
+    "sudo -S reads the password from stdin, and a sudo that does not prompt "
+    "(NOPASSWD, a cached timestamp) leaves it for the command, which would read it "
+    "as the first line of the data. Stage the data without privilege (umask 077, "
+    "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"
+
+#: Sent right after ``_HISTORY_OFF`` (#41): systemd 258's
+#: ``/etc/profile.d/80-systemd-osc-context.sh`` wraps every command in OSC 3008
+#: sequences from ``PS0`` and ``PROMPT_COMMAND``, each with a fresh UUID. It appends
+#: to ``PROMPT_COMMAND`` as an array, so only ``unset`` removes its hook.
+_PROMPT_HOOKS_OFF = " unset PS0 PROMPT_COMMAND; 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. "
+    "sudo requires a password on this host and no sudo password was given: pass one, "
-    "Please add the sudo password to a Credential Profile assigned to this device, "
+    "or allow this user passwordless sudo (NOPASSWD)."
-    "or configure passwordless sudo (NOPASSWD) for this user."
 )
 
 # DMI field values that carry no useful information (OEM defaults, blanks)
@@ -205,10 +258,56 @@ def _arm_vendor_from_model(model: str) -> str:
     return " ".join(brand)
 
 
+#: Where net-snmp reads its configuration, and how each package manager installs it.
+_SNMPD_CONF = "/etc/snmp/snmpd.conf"
+_SNMP_INSTALL = {
+    "apt": "apt-get update -y >/dev/null 2>&1; DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
+    "dnf": "dnf install -y net-snmp net-snmp-utils 2>&1",
+    "yum": "yum install -y net-snmp net-snmp-utils 2>&1",
+    "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1",
+    "pacman": "pacman -Sy --noconfirm net-snmp 2>&1",
+}
+_APT_PROXY_CONF = "/etc/apt/apt.conf.d/00proxy"
+_PROXY_URL = re.compile(r"^https?://[A-Za-z0-9.\-\[\]:@%_~/]+$")
+_USER_NAME = re.compile(r"^[a-z_][a-z0-9_.\-]{0,31}$")
+#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
+_SNMP_WORD = re.compile(r"^[^\s\"#]+$")
+
+
+def _source(value: str) -> str:
+    """An allowed source as an address or network, or ValueError."""
+    return str(ipaddress.ip_network(value.strip(), strict=False))
+
+
+def render_snmpd_conf(settings: SnmpAgentSettingsDict) -> str:
+    """net-snmp's configuration for *settings*: read-only v2c, per allowed source."""
+    community = settings["community"]
+    if not _SNMP_WORD.match(community):
+        raise ValueError("the community must be one word without quotes or #")
+    listen = settings.get("listen")
+    lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"]
+    sources = [_source(s) for s in settings.get("allowed_sources", [])]
+    lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"]
+    for key, directive in (("location", "sysLocation"), ("contact", "sysContact")):
+        value = settings.get(key, "")
+        if value:
+            if "\n" in value:
+                raise ValueError(f"{key} must be one line")
+            lines.append(f"{directive}    {value}")
+    return "\n".join(lines) + "\n"
+
+
 class LinuxDriver(
+    SnmpAgentMixin,
+    EnsurePackageMixin,
+    PackageProxyMixin,
     KernelFactsMixin,
+    ServicePackagesMixin,
     ListeningSocketsMixin,
+    HostFirewallMixin,
     VirtualIpsMixin,
+    ServiceClustersMixin,
+    SystemIdentityMixin,
     OpenSshServerMixin,
     SystemdServicesMixin,
     HostStatusMixin,
@@ -260,8 +359,6 @@ class LinuxDriver(
         # Optional sudo password for privilege escalation (e.g. apt-get update)
         self._sudo_password: str | None = optional_args.get("sudo_password")
         # Expected apt proxy URL — checked as a device warning on apt systems.
-        # Only set when apt_proxy_enabled=true in NetOrk settings; empty string disables the check.
-        self._apt_proxy_url: str = optional_args.get("apt_proxy_url", "")
 
         if optional_args.get("debugging"):
             logger.setLevel(logging.DEBUG)
@@ -270,6 +367,8 @@ class LinuxDriver(
         # port is passed explicitly in open() — remove it from netmiko_optional_args
         # to avoid "multiple values for keyword argument 'port'"
         self.netmiko_optional_args.pop("port", None)
+        # The host key to expect, and where the one seen is reported (netork#1167).
+        self._host_key = HostKeyCheck(optional_args, hostname, self.port)
 
         # Runtime state
         self._device: ConnectHandler | None = None
@@ -282,6 +381,7 @@ class LinuxDriver(
     def open(self) -> None:
         """Open the SSH connection and detect the package manager."""
         try:
+            with self._host_key.netmiko(self.netmiko_optional_args) as netmiko_args_:
                 self._device = ConnectHandler(
                     device_type=self.NETMIKO_DEVICE_TYPE,
                     host=self.hostname,
@@ -290,12 +390,28 @@ class LinuxDriver(
                     port=self.port,
                     secret=self._secret,
                     timeout=self.timeout,
-                **self.netmiko_optional_args,
+                    **netmiko_args_,
                 )
         except NetmikoAuthenticationException as exc:
             raise ConnectionException(str(exc)) from exc
         except NetmikoTimeoutException as exc:
             raise ConnectionException(str(exc)) from exc
+        self._host_key.observe(self._device.remote_conn_pre.get_transport())
+
+        # 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)
+
+        # No systemd OSC 3008 context around every command (#41). netmiko learned
+        # the prompt with those sequences in front, and a prompt carrying a fresh
+        # UUID never matches again, so it reads the prompt anew once they are gone.
+        try:
+            self._device.send_command(_PROMPT_HOOKS_OFF, expect_string=_PROMPT_RE)
+            self._device.set_base_prompt()
+        except Exception:
+            logger.debug("Could not switch the prompt hooks off on %s", self.hostname, exc_info=True)
 
         # Prevent PTY from wrapping long output lines (e.g. docker JSON).
         try:
@@ -329,40 +445,73 @@ class LinuxDriver(
     # ------------------------------------------------------------------
 
     def _send(self, command: str, read_timeout: float = 100) -> str:
-        """Send a command and return stripped output."""
+        """Send a command and return its output without terminal codes, stripped."""
         if not self._device:
             raise ConnectionClosedException("Not connected")
-        return self._device.send_command(
+        output = self._device.send_command(
             command,
             read_timeout=read_timeout,
             cmd_verify=False,
             expect_string=_expect_for(command),
-        ).strip()
+        )
+        return strip_terminal_codes(output).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.
@@ -391,7 +540,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)
 
     # ------------------------------------------------------------------
@@ -415,10 +564,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,
@@ -428,13 +587,25 @@ class LinuxDriver(
         timeout: float = 60,
         stdin: bytes | None = None,
     ) -> CommandResult:
-        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code."""
+        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code.
+
+        :raises ValueError: for *stdin* to a privileged command while a sudo
+            password is set (see ``_PRIVILEGED_STDIN``).
+        """
         line, prefix = self._privileged(command, privileged)
-        data = (prefix or b"") + (stdin or b"") if (prefix or stdin) else None
+        if prefix and stdin is not None:
+            raise ValueError(_PRIVILEGED_STDIN)
+        data = prefix or stdin or None
         return run_on_transport(self._transport(), line, stdin=data, timeout=timeout)
 
     def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
-        """Start *command* on an exec channel and return a stream to it."""
+        """Start *command* on an exec channel and return a stream to it.
+
+        A privileged stream with a sudo password starts with that password, for
+        ``sudo -S``; a sudo that does not prompt leaves it for *command*. Write
+        nothing to such a stream that must not follow the password, as
+        ``_PRIVILEGED_STDIN`` says for ``run_command``.
+        """
         line, prefix = self._privileged(command, privileged)
         return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix)
 
@@ -1013,6 +1184,15 @@ class LinuxDriver(
         """The transport for ``KernelFactsMixin.get_kernel_facts``: read-only, no sudo."""
         return self._send(command, read_timeout=60)
 
+    def _run_service_packages_command(self, command: str) -> str:
+        """The transport for ``ServicePackagesMixin.get_service_packages``.
+
+        An exec channel rather than the PTY -- the command is about 1 kB -- under
+        ``sh``, which it is written for, whatever the login shell. Read-only, so
+        never through sudo.
+        """
+        return self.run_command(f"sh -c {_shlex_quote(command)}", timeout=120).stdout
+
     # ------------------------------------------------------------------
     # ListeningSocketsMixin – the transport for get_listening_sockets
     # ------------------------------------------------------------------
@@ -1026,6 +1206,15 @@ class LinuxDriver(
             return self._run_privileged(command, 60)
         return self._send(command, read_timeout=60)
 
+    def _run_host_firewall_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``HostFirewallMixin.get_host_firewall``: an exec channel.
+
+        ``nft -j`` prints the whole ruleset as one line of JSON, longer than the
+        PTY netmiko holds reliably takes. Read-only; root because both nft and
+        iptables-save need ``CAP_NET_ADMIN`` (NetOrk/netork#1344).
+        """
+        return self.run_command(command, privileged=privileged, timeout=60).stdout
+
     def _run_virtual_ips_command(self, command: str, *, privileged: bool) -> str:
         """The transport for ``VirtualIpsMixin.get_virtual_ips``: an exec channel.
 
@@ -1035,6 +1224,18 @@ class LinuxDriver(
         """
         return self.run_command(command, privileged=privileged, timeout=60).stdout
 
+    def _run_service_clusters_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``ServiceClustersMixin.get_service_clusters``: an exec
+        channel, like the virtual IPs' command, which it resembles in size.
+        Read-only either way; root only because Patroni's configuration is the
+        postgres user's alone."""
+        return self.run_command(command, privileged=privileged, timeout=60).stdout
+
+    def _run_system_identity_command(self, command: str, *, privileged: bool) -> str:
+        """The transport for ``SystemIdentityMixin.get_system_identity``: an exec channel,
+        as root, since ``product_uuid`` and ``product_serial`` are root's."""
+        return self.run_command(command, privileged=privileged, timeout=30).stdout
+
     # ------------------------------------------------------------------
     # OSDriver – package management
     # ------------------------------------------------------------------
@@ -1359,38 +1560,13 @@ class LinuxDriver(
             f"Package manager '{self._pkg_manager}' is not supported"
         )
 
-    def get_device_warnings(self) -> List[dict[str, Any]]:
+    def get_package_proxy(self) -> str | None:
-        """Return warning dicts for issues detected on this device.
+        """The proxy apt fetches through (``00proxy``); None for none or a non-apt host."""
-
+        if self._pkg_manager != "apt":
-        Currently detects:
+            return None
-        - package updates available (uses local package cache)
+        conf = self._send(f"cat {_APT_PROXY_CONF} 2>/dev/null || true")
-        - apt proxy not configured (apt systems only)
+        found = re.search(r'Acquire::http::Proxy\s+"([^"]+)"', conf)
-        """
+        return found.group(1) if found else None
-        warnings: List[dict[str, Any]] = []
-        try:
-            updates = self.get_available_updates()
-        except Exception as exc:
-            logger.warning("get_device_warnings: get_available_updates() failed: %s", exc)
-            updates = []
-        if updates:
-            warnings.append({
-                "code": "updates_available",
-                "meta": {
-                    "count": len(updates),
-                    "packages": [u.get("name", "") for u in updates],
-                },
-            })
-        if self._pkg_manager == "apt" and self._apt_proxy_url:
-            try:
-                current = self._send("cat /etc/apt/apt.conf.d/00proxy 2>/dev/null || true").strip()
-                if self._apt_proxy_url not in current:
-                    warnings.append({
-                        "code": "apt_proxy_missing",
-                        "meta": {"expected_url": self._apt_proxy_url},
-                    })
-            except Exception as exc:
-                logger.warning("get_device_warnings: apt proxy check failed: %s", exc)
-        return warnings
 
     def _get_updates_apt(self) -> list[UpdateDict]:
         # apt list --upgradable does not need root; avoid sudo so it works even
@@ -1423,7 +1599,16 @@ class LinuxDriver(
         """Packages a pending security advisory covers; None when dnf/yum cannot say."""
         cmd = DNF_SECURITY_COMMAND if self._pkg_manager == "dnf" else _YUM_SECURITY_COMMAND
         output, status = _split_status(self._run_privileged(f"{cmd}; echo {_RC_MARKER}$?", 120))
-        return parse_dnf_security(output) if status == 0 else None
+        # Both commands pipe through cat, so the outer status is cat's: dnf/yum's own
+        # is in __DNF_RC. A failed run prints nothing on stdout and must not read as
+        # "no advisories"; neither may output cut short before the marker.
+        rc = _DNF_RC_RE.findall(output)
+        if status != 0 or not rc or rc[-1] != "0":
+            return None
+        try:
+            return parse_dnf_security(output)
+        except ValueError:
+            return None
 
     def refresh_available_updates(self) -> dict[str, Any]:
         """Refresh the package index (apt-get update, dnf makecache, apk update)."""
@@ -1606,7 +1791,7 @@ class LinuxDriver(
     # OSDriver – users
     # ------------------------------------------------------------------
 
-    def get_users(self) -> list[UserDict]:
+    def get_os_users(self) -> list[UserDict]:
         """Return local user accounts from /etc/passwd plus supplementary groups."""
         passwd_out = self._send("getent passwd 2>/dev/null || cat /etc/passwd")
         groups_out = self._send("getent group 2>/dev/null || cat /etc/group")
@@ -1828,216 +2013,89 @@ class LinuxDriver(
             logger.debug("get_snmp_config() failed: %s", exc)
             return None
 
-    def run_device_action(self, action: str) -> DeviceActionResultDict:
+    # ------------------------------------------------------------------
-        """Execute a named action on the device."""
+    # Fix templates (napalm-device-types ``fixes``): the caller supplies every value
-        if action == "fix_docker_permissions":
+    # ------------------------------------------------------------------
-            return self._action_fix_docker_permissions()
-        if action == "fix_snmp":
-            return self._action_fix_snmp()
-        if action == "fix_apt_proxy":
-            return self._action_fix_apt_proxy()
-        if action == "apt_update_upgrade":
-            return self._action_apt_update_upgrade()
-        raise NotImplementedError(f"Unknown action: {action!r}")
 
-    def _action_apt_update_upgrade(self) -> DeviceActionResultDict:
+    def _root_outcome(self, command: str, timeout: float, done: str) -> FixStepOutcomeDict:
-        """Refresh the apt cache and fully upgrade all packages (apt-based systems only).
+        """Run *command* as root and report it as a step: its exit status decides."""
+        output, status = self._sudo_status(command, timeout)
+        if status != 0 and "password is required" in output:
+            output = f"{output}\n{_SUDO_PASSWORD_HINT}"
+        return {"ok": status == 0, "output": (output.strip()[-300:] if status != 0 else done)}
 
-        Uses full-upgrade (not plain upgrade) — plain "apt-get upgrade" refuses
+    def get_management_peer(self) -> str | None:
-        to install/remove packages even when required to satisfy a newer
+        """The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
-        version's dependencies, silently leaving those updates pending.
+        out = self.run_command('printf "%s\\n" "$SSH_CONNECTION"', timeout=15).stdout.split()
-        """
-        if self._pkg_manager != "apt":
-            return {
-                "success": True,
-                "output": f"Skipped — package manager is {self._pkg_manager!r}, not apt.",
-            }
-
-        sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__")
-        if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower():
-            if not self._sudo_password:
-                return {
-                    "success": False,
-                    "output": _SUDO_PASSWORD_HINT,
-                }
-
-        lines: list[str] = []
         try:
-            out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
+            return str(ipaddress.ip_address(out[0])) if out else None
-            lines.append(f"[update] {out.strip()[-300:]}")
+        except ValueError:
-            out = self._sudo(
+            return None
-                "DEBIAN_FRONTEND=noninteractive apt-get full-upgrade -y 2>&1", read_timeout=240
-            )
-            lines.append(f"[upgrade] {out.strip()[-300:]}")
-            return {"success": True, "output": "\n".join(lines)}
-        except Exception as exc:
-            lines.append(f"[error] {exc}")
-            return {"success": False, "output": "\n".join(lines)}
 
-    def _action_fix_snmp(self) -> DeviceActionResultDict:
+    def _install_snmp_agent(self) -> FixStepOutcomeDict:
-        """Install, configure and start snmpd with community 'public'."""
+        pm = self._pkg_manager or self._detect_pkg_manager()
-        lines: list[str] = []
+        command = _SNMP_INSTALL.get(pm or "")
+        if not command:
+            return {"ok": False, "output": f"no supported package manager ({pm or 'none found'})"}
+        return self._root_outcome(command, 300, f"net-snmp installed ({pm})")
 
-        # 0. Verify sudo access before attempting anything
+    def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
-        sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__")
+        written = self._install_file_as_root(_SNMPD_CONF, render_snmpd_conf(settings))
-        if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower():
+        if written.exit_code:
-            if not self._sudo_password:
+            return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
-                return {
+        return {"ok": True, "output": f"wrote {_SNMPD_CONF}"}
-                    "success": False,
-                    "output": _SUDO_PASSWORD_HINT,
-                }
 
-        # 1. Install snmpd if missing
+    def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
-        pkg_mgr = self._detect_pkg_manager()
+        sources = [_source(s) for s in settings.get("allowed_sources", [])]
-        if not pkg_mgr:
+        if not sources:
-            return {"success": False, "output": "Package manager not detected — cannot install snmpd."}
+            return {"ok": True, "output": "no sources given; firewall unchanged"}
-
+        if self._send("command -v ufw 2>/dev/null"):
-        # Refresh the package index first — a freshly provisioned (or simply
+            rules = [f"ufw allow from {s} to any port 161 proto udp" for s in sources]
-        # long-untouched) system's cache can be stale/empty, which makes the
+        elif self._send("command -v iptables 2>/dev/null"):
-        # install below fail outright rather than just being slow.
+            rules = [
-        if pkg_mgr == "apt":
+                f"{{ iptables -C INPUT -s {s} -p udp --dport 161 -j ACCEPT 2>/dev/null"
-            try:
+                f" || iptables -I INPUT -s {s} -p udp --dport 161 -j ACCEPT; }}"
-                update_out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
+                for s in sources
-                lines.append(f"[update] {update_out.strip()[-200:]}")
+            ]
-            except Exception as exc:
-                lines.append(f"[warn] apt-get update failed: {exc}")
-
-        # Install both snmpd (daemon) and snmp (client tools incl. snmpget for probing)
-        install_cmd: dict[str, str] = {
-            "apt":    "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
-            "dnf":    "dnf install -y net-snmp net-snmp-utils 2>&1",
-            "yum":    "yum install -y net-snmp net-snmp-utils 2>&1",
-            "apk":    "apk add --no-cache net-snmp net-snmp-tools 2>&1",
-            "pacman": "pacman -Sy --noconfirm net-snmp 2>&1",
-        }
-        cmd = install_cmd.get(pkg_mgr)
-        if cmd:
-            try:
-                out = self._sudo(cmd, read_timeout=120)
-                lines.append(f"[install] {out.strip()[-200:]}")
-            except Exception as exc:
-                lines.append(f"[error] install failed: {exc}")
-                return {"success": False, "output": "\n".join(lines)}
-
-        # 2. Determine the IP netOrk is connecting from by checking the established SSH connection
-        netork_ip = ""
-        try:
-            # ss shows the remote peer of the current SSH connection
-            raw = self._send(
-                "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1"
-            ).strip()
-            if raw and raw not in ("", "0.0.0.0", "::", "127.0.0.1"):
-                netork_ip = raw
-        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
-        # 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")
-        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]}")
+            return {"ok": True, "output": "no host firewall (ufw, iptables); nothing to open"}
+        return self._root_outcome(" && ".join(rules), 30, f"udp/161 open for {', '.join(sources)}")
 
-        # 3. Open firewall for SNMP (UDP 161) — restrict to netOrk's source IP
+    def _start_snmp_agent(self) -> FixStepOutcomeDict:
-        if netork_ip:
+        return self._root_outcome(
-            try:
+            "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd",
-                ufw = self._send("command -v ufw 2>/dev/null").strip()
+            60,
-                ipt = self._send("command -v iptables 2>/dev/null").strip()
+            "snmpd enabled and running",
-                if ufw:
-                    # Expand to /24 so all containers in the same Docker network can probe
-                    parts = netork_ip.rsplit(".", 1)
-                    subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip
-                    fw_out = self._sudo(
-                        f"ufw allow from {subnet} to any port 161 proto udp 2>&1", read_timeout=10
         )
-                    lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}")
-                elif ipt:
-                    fw_out = self._sudo(
-                        f"iptables -C INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT 2>/dev/null"
-                        f" || iptables -I INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT",
-                        read_timeout=10,
-                    )
-                    lines.append(f"[firewall/iptables] rule added for {netork_ip}:161/udp")
-            except Exception as exc:
-                lines.append(f"[firewall] skipped — {exc}")
 
-        # 4. Restart snmpd.
+    def configure_package_proxy(self, url: str) -> FixResultDict:
-        #    - Redirect all output to /dev/null so netmiko's prompt detection is
+        """apt fetches through *url* (``/etc/apt/apt.conf.d/00proxy``); ``""`` removes it."""
-        #      never confused by service status messages.
-        #    - Append "; echo __OK__" so there is always a known token to wait for.
-        import time as _time
-        # Stop any running snmpd (systemctl-managed or apt-started orphan)
-        self._sudo("systemctl stop snmpd >/dev/null 2>&1; echo s1", read_timeout=15)
-        self._sudo("pkill -9 snmpd >/dev/null 2>&1; echo s2", read_timeout=10)
-        _time.sleep(2)
-        # Enable and start fresh
-        self._sudo("systemctl enable snmpd >/dev/null 2>&1; echo s3", read_timeout=15)
-        self._sudo("systemctl start snmpd >/dev/null 2>&1; echo s4", read_timeout=20)
-        _time.sleep(2)
-        lines.append("[service] snmpd restarted.")
 
-        # 5. Verify snmpd responds via local SNMP probe (sysDescr.0).
+        def write() -> FixStepOutcomeDict:
-        #    Success requires actual SNMP data types in the output, not just
-        #    the absence of error keywords.
-        _time.sleep(2)
-        probe_out = self._send(
-            "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1 || true"
-        ).strip()
-        # snmpget returns lines like "STRING: Linux ..." or "Timeticks: (n) ..."
-        _snmp_types = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
-        success = any(t in probe_out for t in _snmp_types)
-        if success:
-            lines.append(f"[ok] SNMP probe successful — community 'public' is working.")
-        else:
-            lines.append(f"[warn] SNMP probe failed — output: {probe_out[:200]}")
-
-        return {"success": success, "output": "\n".join(lines)}
-
-    def _action_fix_docker_permissions(self) -> dict[str, Any]:
-        """Add the SSH user to the 'docker' group via sudo usermod."""
-        user = self._send("whoami 2>/dev/null || id -un").strip().splitlines()[-1].strip()
-        out = self._sudo(f"usermod -aG docker {user}")
-        low = out.lower()
-        success = not any(kw in low for kw in ("error", "invalid", "no such", "command not found"))
-        if not out.strip():
-            out = f"Added {user!r} to the docker group. Reconnect or run a new poll to verify."
-        return {"success": success, "output": out}
-
-    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."}
             if self._pkg_manager != "apt":
-            return {"success": False, "output": f"Package manager is {self._pkg_manager!r}, not apt — skipping."}
+                return {"ok": False, "output": f"package manager is {self._pkg_manager!r}, not apt"}
+            if not url:
+                return self._root_outcome(f"rm -f {_APT_PROXY_CONF}", 30, "proxy removed")
+            if not _PROXY_URL.match(url):
+                raise ValueError(f"not a proxy URL: {url!r}")
+            written = self._install_file_as_root(_APT_PROXY_CONF, f'Acquire::http::Proxy "{url}";\n')
+            if written.exit_code:
+                return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
+            return {"ok": True, "output": f"apt fetches through {url}"}
 
-        content = f'Acquire::http::Proxy "{proxy_url}";\n'
+        return run_fix_steps([("configure", write)])
-        content_b64 = _b64.b64encode(content.encode()).decode()
+
-        self._send(f"echo {content_b64} | base64 -d > /tmp/netork_00proxy")
+    def grant_container_engine_access(self, user: str) -> FixResultDict:
-        self._sudo(
+        """Add *user* to the ``docker`` group, so the engine answers without sudo."""
-            "mv /tmp/netork_00proxy /etc/apt/apt.conf.d/00proxy && "
+
-            "chown root:root /etc/apt/apt.conf.d/00proxy && "
+        def grant() -> FixStepOutcomeDict:
-            "chmod 644 /etc/apt/apt.conf.d/00proxy"
+            if not _USER_NAME.match(user):
+                raise ValueError(f"not a user name: {user!r}")
+            return self._root_outcome(
+                f"usermod -aG docker {user}", 30, f"{user} is in the docker group from the next login"
             )
-        verify = self._send("cat /etc/apt/apt.conf.d/00proxy 2>/dev/null").strip()
+
-        success = proxy_url in verify
+        return run_fix_steps([("grant", grant)])
-        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]}"}
 
     def get_vpn_tunnels(self) -> dict[str, Any]:
         """Return WireGuard status via ``wg show all dump`` (requires root/sudo).
@@ -2052,16 +2110,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",

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.5.0"
+version = "0.13.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
 requires-python = ">=3.9"
@@ -37,7 +37,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.5.0",
+    "napalm-device-types>=4.21.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_command_channel.py

@@ -42,9 +42,15 @@ class FakeChannel:
     def recv_ready(self):
         return False
 
+    def recv(self, size):
+        return b""
+
     def recv_stderr_ready(self):
         return False
 
+    def recv_stderr(self, size):
+        return b""
+
     def exit_status_ready(self):
         return True
 
@@ -107,21 +113,54 @@ def test_a_root_login_needs_no_sudo():
     assert transport.channels[-1].sent == b""
 
 
-def test_stdin_follows_the_sudo_password():
+def test_a_privileged_command_with_a_sudo_password_takes_no_stdin():
+    """sudo -S reads the password from stdin. A sudo that does not prompt --
+    NOPASSWD, a cached timestamp -- leaves it there, and the command reads it as
+    its first line: it wrote the password into an sshd drop-in (NetOrk/netork#1029)."""
     driver, transport = _driver(sudo_password="pw")
 
+    with pytest.raises(ValueError, match="stdin"):
+        driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)
+
+    assert transport.channels == []  # nothing reached the host
+
+
+def test_privileged_stdin_without_a_sudo_password_goes_to_the_command():
+    """sudo -n reads nothing from stdin: the data is the command's alone."""
+    driver, transport = _driver()
+
     driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)
 
-    assert transport.channels[-1].sent == b"pw\ndata"
+    channel = transport.channels[-1]
+    assert channel.command == "sudo -n sh -c 'tee /etc/x'"
+    assert channel.sent == b"data"
+
+
+def test_privileged_stdin_for_a_root_login_goes_to_the_command():
+    driver, transport = _driver(root=True, sudo_password="pw")
+
+    driver.run_command("tee /etc/x", privileged=True, stdin=b"data", timeout=5)
+
+    assert transport.channels[-1].command == "tee /etc/x"
+    assert transport.channels[-1].sent == b"data"
+
+
+def test_unprivileged_stdin_with_a_sudo_password_goes_to_the_command():
+    driver, transport = _driver(sudo_password="pw")
+
+    driver.run_command("cat > /tmp/x", stdin=b"data", timeout=5)
+
+    assert transport.channels[-1].command == "cat > /tmp/x"
+    assert transport.channels[-1].sent == b"data"
 
 
 def test_a_privileged_stream_gets_the_password_first():
     driver, transport = _driver(sudo_password="pw")
 
-    driver.open_stream("cat > /tmp/x", privileged=True)
+    driver.open_stream("journalctl -f", privileged=True)
 
     channel = transport.channels[-1]
-    assert channel.command == "sudo -S -p '' sh -c 'cat > /tmp/x'"
+    assert channel.command == "sudo -S -p '' sh -c 'journalctl -f'"
     assert channel.sent == b"pw\n"
 
 

tests/test_fix_templates.py

@@ -0,0 +1,236 @@
+"""The fix templates (napalm-device-types ``fixes``): the caller supplies every value.
+
+They replace ``run_device_action``. The driver knows how each step is done on a
+Linux host; the community, the allowed sources, the proxy and the user are the
+caller's (napalm-device-types docs/STRUCTURE.md).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from napalm_linux import LinuxDriver
+from napalm_linux.linux import render_snmpd_conf
+
+
+class TestSnmpdConf:
+    def test_one_read_only_line_per_allowed_source(self):
+        conf = render_snmpd_conf(
+            {
+                "community": "c0mm",
+                "allowed_sources": ["10.0.0.5", "10.1.0.0/24"],
+                "contact": "ops",
+                "location": "rack 1",
+            }
+        )
+
+        assert conf == (
+            "agentAddress udp:161\n"
+            "rocommunity c0mm 10.0.0.5/32\n"
+            "rocommunity c0mm 10.1.0.0/24\n"
+            "sysLocation    rack 1\n"
+            "sysContact    ops\n"
+        )
+
+    def test_without_sources_every_address_may_ask(self):
+        assert "rocommunity c0mm\n" in render_snmpd_conf(
+            {"community": "c0mm", "allowed_sources": []}
+        )
+
+    def test_a_listen_address_binds_the_agent(self):
+        conf = render_snmpd_conf({"community": "c", "allowed_sources": [], "listen": "10.0.0.1"})
+        assert conf.startswith("agentAddress udp:10.0.0.1:161\n")
+
+    def test_no_application_name_is_written(self):
+        conf = render_snmpd_conf({"community": "c", "allowed_sources": []})
+        assert "netork" not in conf.lower()
+
+    @pytest.mark.parametrize("community", ["two words", 'quo"te', "has#hash", ""])
+    def test_a_community_that_would_break_the_file_is_refused(self, community):
+        with pytest.raises(ValueError):
+            render_snmpd_conf({"community": community, "allowed_sources": []})
+
+    def test_a_source_that_is_no_address_is_refused(self):
+        with pytest.raises(ValueError):
+            render_snmpd_conf({"community": "c", "allowed_sources": ["10.0.0.5; rm -rf /"]})
+
+
+@pytest.fixture
+def driver():
+    d = LinuxDriver("h", "admin", "p")
+    d._device = MagicMock()
+    d._pkg_manager = "apt"
+    d._root = True
+    return d
+
+
+class TestSnmpAgentSteps:
+    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.0.0.5"]}
+
+    def _record(self, driver, status=0, firewall="ufw"):
+        ran: list[str] = []
+
+        def sudo_status(command, read_timeout=100):
+            ran.append(command)
+            return ("", status)
+
+        driver._sudo_status = sudo_status
+        driver._send = lambda command, read_timeout=100: (
+            f"/usr/sbin/{firewall}" if command == f"command -v {firewall} 2>/dev/null" else ""
+        )
+        driver._install_file_as_root = MagicMock(
+            return_value=MagicMock(exit_code=0, stdout="", stderr="")
+        )
+        return ran
+
+    def test_installs_writes_opens_and_starts(self, driver):
+        ran = self._record(driver)
+
+        result = driver.configure_snmp_agent(self.SETTINGS)
+
+        assert result["success"] is True
+        assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
+        assert "apt-get install -y snmpd snmp" in ran[0]
+        path, content = driver._install_file_as_root.call_args.args
+        assert path == "/etc/snmp/snmpd.conf" and "rocommunity c0mm 10.0.0.5/32" in content
+        assert ran[1] == "ufw allow from 10.0.0.5/32 to any port 161 proto udp"
+        assert "systemctl restart snmpd" in ran[2]
+
+    def test_the_firewall_opens_only_for_the_given_sources_with_iptables(self, driver):
+        ran = self._record(driver, firewall="iptables")
+
+        driver.configure_snmp_agent(self.SETTINGS)
+
+        assert "iptables -I INPUT -s 10.0.0.5/32 -p udp --dport 161 -j ACCEPT" in ran[1]
+
+    def test_no_sources_change_no_firewall(self, driver):
+        ran = self._record(driver)
+
+        result = driver.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
+
+        assert result["success"] is True
+        assert not any("ufw" in c or "iptables" in c for c in ran)
+
+    def test_a_failed_install_stops_the_fix(self, driver):
+        ran = self._record(driver, status=100)
+
+        result = driver.configure_snmp_agent(self.SETTINGS)
+
+        assert result["success"] is False
+        assert len(ran) == 1
+        driver._install_file_as_root.assert_not_called()
+
+
+class TestPackageProxyAndEngineAccess:
+    def test_the_proxy_is_the_caller_s(self, driver):
+        driver._install_file_as_root = MagicMock(
+            return_value=MagicMock(exit_code=0, stdout="", stderr="")
+        )
+
+        result = driver.configure_package_proxy("http://cache.lan:3142")
+
+        assert result["success"] is True
+        path, content = driver._install_file_as_root.call_args.args
+        assert (path, content) == (
+            "/etc/apt/apt.conf.d/00proxy",
+            'Acquire::http::Proxy "http://cache.lan:3142";\n',
+        )
+
+    def test_a_proxy_url_that_would_break_the_file_is_refused(self, driver):
+        driver._install_file_as_root = MagicMock()
+
+        result = driver.configure_package_proxy('http://x"; Acquire::x "y')
+
+        assert result["success"] is False
+        driver._install_file_as_root.assert_not_called()
+
+    def test_only_apt_takes_a_proxy(self, driver):
+        driver._pkg_manager = "dnf"
+
+        assert driver.configure_package_proxy("http://cache.lan:3142")["success"] is False
+
+    def test_engine_access_for_the_user_the_caller_names(self, driver):
+        ran = []
+        driver._sudo_status = lambda command, read_timeout=100: ran.append(command) or ("", 0)
+
+        result = driver.grant_container_engine_access("deploy")
+
+        assert result["success"] is True
+        assert ran == ["usermod -aG docker deploy"]
+
+    def test_a_user_name_that_is_no_user_name_is_refused(self, driver):
+        driver._sudo_status = MagicMock()
+
+        assert driver.grant_container_engine_access("x; reboot")["success"] is False
+        driver._sudo_status.assert_not_called()
+
+
+def test_run_device_action_is_gone():
+    assert (
+        not hasattr(LinuxDriver, "run_device_action")
+        or LinuxDriver.run_device_action is None
+        or ("run_device_action" not in vars(LinuxDriver))
+    )
+    for name in (
+        "_action_fix_snmp",
+        "_action_fix_apt_proxy",
+        "_action_fix_docker_permissions",
+        "_action_apt_update_upgrade",
+    ):
+        assert not hasattr(LinuxDriver, name), name
+
+
+def test_every_annotation_resolves():
+    """``Dict``, ``List`` and ``Optional`` were used without being imported (#29);
+    ``from __future__ import annotations`` hid it until something resolved them."""
+    import inspect
+    import typing
+
+    for name, member in inspect.getmembers(LinuxDriver, inspect.isfunction):
+        if member.__module__ == "napalm_linux.linux":
+            typing.get_type_hints(member)
+
+
+class TestManagementPeer:
+    """The address the host sees this connection come from (``SSH_CONNECTION``)."""
+
+    def _driver(self, answer):
+        d = LinuxDriver("h", "admin", "p")
+        d._device = MagicMock()
+        d.run_command = MagicMock(return_value=MagicMock(stdout=answer, exit_code=0))
+        return d
+
+    def test_the_client_side_of_ssh_connection(self):
+        assert self._driver("10.7.224.20 51234 10.0.0.5 22\n").get_management_peer() == "10.7.224.20"
+
+    def test_an_ipv6_client(self):
+        assert self._driver("fd00::20 51234 fd00::5 22\n").get_management_peer() == "fd00::20"
+
+    @pytest.mark.parametrize("answer", ["", "\n", "garbage", "not-an-ip 1 2 3"])
+    def test_nothing_usable_is_none(self, answer):
+        assert self._driver(answer).get_management_peer() is None
+
+
+class TestFactsNotWarnings:
+    """The driver reports facts; the application names warnings (#28, netOrk #1111)."""
+
+    def test_no_warnings_of_its_own(self):
+        assert "get_device_warnings" not in vars(LinuxDriver)
+
+    def test_the_package_proxy_apt_fetches_through(self, driver):
+        driver._send = lambda command, read_timeout=100: 'Acquire::http::Proxy "http://c:3142";'
+        assert driver.get_package_proxy() == "http://c:3142"
+
+    def test_no_proxy_set(self, driver):
+        driver._send = lambda command, read_timeout=100: ""
+        assert driver.get_package_proxy() is None
+
+    def test_not_an_apt_host(self, driver):
+        driver._pkg_manager = "dnf"
+        assert driver.get_package_proxy() is None
+
+    def test_the_applications_proxy_url_is_no_argument_any_more(self):
+        d = LinuxDriver("h", "admin", "p", optional_args={"apt_proxy_url": "http://c:3142"})
+        assert not hasattr(d, "_apt_proxy_url")

tests/test_linux.py

@@ -4,7 +4,12 @@ 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 (
+    _YUM_SECURITY_COMMAND,
+    LinuxDriver,
+    _arm_vendor_from_model,
+    _split_status,
+)
 
 
 # ---------------------------------------------------------------------------
@@ -31,9 +36,78 @@ 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 recv_stderr(self, size):
+        return b""
+
+    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
@@ -243,7 +317,7 @@ def test_get_available_updates_apt(driver):
 
 
 # ---------------------------------------------------------------------------
-# get_users
+# get_os_users
 # ---------------------------------------------------------------------------
 
 
@@ -260,9 +334,9 @@ GROUP_OUT = (
 )
 
 
-def test_get_users(driver):
+def test_get_os_users(driver):
     with patch.object(driver, "_send", side_effect=[PASSWD_OUT, GROUP_OUT]):
-        users = driver.get_users()
+        users = driver.get_os_users()
 
     admin = next(u for u in users if u["username"] == "admin")
     assert admin["uid"] == 1000
@@ -272,6 +346,14 @@ def test_get_users(driver):
     assert set(admin["groups"]) == {"sudo", "docker", "adm"}
 
 
+def test_get_users_is_left_to_napalm(driver):
+    """napalm-device-types 4.7 renamed OSDriver.get_users to get_os_users:
+    NAPALM's own get_users has another shape, and this driver does not implement it."""
+    assert LinuxDriver.get_users is not LinuxDriver.get_os_users
+    with pytest.raises(NotImplementedError):
+        driver.get_users()
+
+
 # ---------------------------------------------------------------------------
 # ping parsing
 # ---------------------------------------------------------------------------
@@ -725,151 +807,10 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver):
     assert facts["vendor"] == "Linux"  # fallback to VENDOR class attribute
 
 
-# ---------------------------------------------------------------------------
-# _action_fix_snmp / _action_apt_update_upgrade
-# ---------------------------------------------------------------------------
-
-
-def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str:
-    """Canned responses covering every _send() call _action_fix_snmp makes."""
-    if command.startswith("sudo -n true"):
-        return ""  # passwordless sudo works
-    if command.startswith("command -v apt"):
-        return "/usr/bin/apt"
-    if command.startswith("command -v"):
-        return ""  # ufw/iptables not found, other pkg managers not found
-    if command.startswith("ss -tnp"):
-        return ""  # no netork_ip detected — skips firewall step
-    if command.startswith("cat /etc/snmp/snmpd.conf"):
-        return "agentAddress udp:161\nrocommunity public\n"
-    if command.startswith("snmpget"):
-        return "STRING: Linux test"
-    return ""
-
-
-class TestActionFixSnmp:
-    def test_runs_apt_get_update_before_install(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-        sudo_calls: list[str] = []
-
-        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
-            sudo_calls.append(command)
-            return ""
-
-        with (
-            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
-            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
-        ):
-            driver._action_fix_snmp()
-
-        update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c)
-        install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c)
-        assert update_idx < install_idx
-
-    def test_install_exception_returns_failure_instead_of_raising(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-
-        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
-            if "apt-get install" in command:
-                raise TimeoutError("connection timed out")
-            return ""
-
-        with (
-            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
-            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
-        ):
-            result = driver._action_fix_snmp()
-
-        assert result["success"] is False
-        assert "install failed" in result["output"]
-
-    def test_apt_get_update_failure_is_non_fatal(self, driver):
-        """apt-get update failing (e.g. transient network issue) must not
-        abort the whole action — install is still attempted."""
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-
-        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
-            if "apt-get update" in command:
-                raise TimeoutError("network unreachable")
-            return ""
-
-        with (
-            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
-            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
-        ):
-            result = driver._action_fix_snmp()
-
-        assert "apt-get update failed" in result["output"]
-
-
-class TestActionAptUpdateUpgrade:
-    def test_skips_when_not_apt(self, driver):
-        driver._pkg_manager = "dnf"
-        result = driver._action_apt_update_upgrade()
-        assert result["success"] is True
-        assert "Skipped" in result["output"]
-
-    def test_fails_when_sudo_needs_password_and_none_configured(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-        with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"):
-            result = driver._action_apt_update_upgrade()
-        assert result["success"] is False
-        assert "sudo requires a password" in result["output"]
-
-    def test_runs_update_then_upgrade_successfully(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = "secret"  # noqa: S105
-        with (
-            patch.object(driver, "_send", return_value=""),
-            patch.object(
-                driver,
-                "_sudo",
-                side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"],
-            ) as mock_sudo,
-        ):
-            result = driver._action_apt_update_upgrade()
-
-        assert result["success"] is True
-        assert "[update]" in result["output"]
-        assert "[upgrade]" in result["output"]
-        update_call, upgrade_call = mock_sudo.call_args_list
-        assert "apt-get update" in update_call.args[0]
-        # full-upgrade (not plain upgrade) — plain upgrade silently holds back
-        # packages whose newer version needs to install/remove dependencies.
-        assert "apt-get full-upgrade" in upgrade_call.args[0]
-
-    def test_exception_during_upgrade_returns_failure(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = "secret"  # noqa: S105
-        with (
-            patch.object(driver, "_send", return_value=""),
-            patch.object(
-                driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")]
-            ),
-        ):
-            result = driver._action_apt_update_upgrade()
-
-        assert result["success"] is False
-        assert "[error]" in result["output"]
-
-
-class TestRunDeviceActionDispatch:
-    def test_apt_update_upgrade_action_dispatches(self, driver):
-        with patch.object(
-            driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""}
-        ) as mock_action:
-            driver.run_device_action("apt_update_upgrade")
-        mock_action.assert_called_once()
-
-
 class TestNoDockerOfItsOwn:
     """Containers are read and handled by netOrk over the Engine API, through
     ``open_container_engine`` (NetOrk/netork#765). The old CLI methods are gone;
-    only the OS action that lets the login user use Docker stays."""
+    only the fix that lets the login user use Docker stays."""
 
     def test_the_cli_methods_are_gone(self):
         for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"):
@@ -877,7 +818,7 @@ class TestNoDockerOfItsOwn:
 
     def test_the_engine_access_and_the_permission_fix_stay(self):
         assert hasattr(LinuxDriver, "open_container_engine")
-        assert hasattr(LinuxDriver, "_action_fix_docker_permissions")
+        assert callable(LinuxDriver.grant_container_engine_access)
 
 
 class TestUninstallPackage:
@@ -915,9 +856,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 +871,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 +916,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 +987,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 +1012,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 +1022,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 +1030,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 +1106,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 +1124,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 +1139,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 +1148,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")
 
@@ -1243,6 +1193,40 @@ def test_get_kernel_facts_raises_on_output_without_a_report(driver):
             driver.get_kernel_facts()
 
 
+# ---------------------------------------------------------------------------
+# get_service_packages -- the command and its parse live in napalm-device-types
+# ---------------------------------------------------------------------------
+
+_SVCPKG = (
+    "SVCPKG_BEGIN\n[units]\nId=nginx.service\n"
+    "FragmentPath=/lib/systemd/system/nginx.service\n"
+    "ExecStart={ path=/usr/sbin/nginx ; argv[]=/usr/sbin/nginx }\n\n"
+    "[dpkg]\nnginx-common: /lib/systemd/system/nginx.service\n"
+    "nginx-core: /usr/sbin/nginx\nSVCPKG_END\n"
+)
+
+
+def test_get_service_packages_runs_the_shared_command_under_sh_on_an_exec_channel(driver):
+    """Read-only, so never through sudo; ``sh`` because the login shell may be another."""
+    from shlex import quote
+
+    from napalm_device_types import CommandResult, ServicePackagesMixin
+    from napalm_device_types.service_packages import SERVICE_PACKAGES_COMMAND
+
+    assert isinstance(driver, ServicePackagesMixin)
+    calls = []
+
+    def run_command(command, *, privileged=False, timeout=60, stdin=None):
+        calls.append((command, privileged))
+        return CommandResult(_SVCPKG, "", 0)
+
+    driver.run_command = run_command
+
+    assert driver.get_service_packages() == {"nginx": ["nginx-common", "nginx-core"]}
+    assert calls == [(f"sh -c {quote(SERVICE_PACKAGES_COMMAND)}", False)]
+    driver._device.send_command.assert_not_called()
+
+
 # ---------------------------------------------------------------------------
 # Services: listed in one round trip, controlled through systemctl (#7)
 # ---------------------------------------------------------------------------
@@ -1295,7 +1279,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):
@@ -1385,9 +1369,12 @@ class TestAvailableUpdates:
         driver._pkg_manager = "dnf"
         driver._device.send_command.side_effect = [
             "0",  # id -u
-            "openssl-libs.x86_64   1:3.1.4-2.fc40   updates\n"
+            ("openssl-libs.x86_64   1:3.1.4-2.fc40   updates\n"
-            "vim-enhanced.x86_64   2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
+            "vim-enhanced.x86_64   2:9.1.083-1.fc40 updates\n__NETORK_RC=100"),
-            "FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n__NETORK_RC=0",
+            # DNF_SECURITY_COMMAND pipes through cat: the outer status is cat's, dnf's own
+            # is in the __DNF_RC marker.
+            ("FEDORA-2024-1a2b3c4d5e Important/Sec. openssl-libs-1:3.1.4-2.fc40.x86_64\n"
+            "__DNF_RC=0\n__NETORK_RC=0"),
         ]
 
         updates = {u["name"]: u for u in driver.get_available_updates()}
@@ -1400,7 +1387,62 @@ class TestAvailableUpdates:
         driver._device.send_command.side_effect = [
             "0",
             "vim-enhanced.x86_64   2:9.1.083-1.fc40 updates\n__NETORK_RC=100",
-            "Error: updateinfo metadata missing\n__NETORK_RC=1",
+            "Error: updateinfo metadata missing\n__DNF_RC=1\n__NETORK_RC=0",
+        ]
+
+        assert driver.get_available_updates()[0]["security"] is None
+
+    def test_yum_security_command_reports_its_own_exit_status(self):
+        # parse_dnf_security (napalm-device-types 4.7) needs the __DNF_RC marker, and
+        # behind "| cat" the outer status no longer says whether yum failed.
+        assert "echo __DNF_RC=$?" in _YUM_SECURITY_COMMAND
+        assert "yum updateinfo list security" in _YUM_SECURITY_COMMAND
+
+    def test_yum_marks_what_a_security_advisory_covers(self, driver):
+        driver._pkg_manager = "yum"
+        driver._device.send_command.side_effect = [
+            "0",  # id -u
+            ("openssl-libs.x86_64   1:1.0.2k-26.el7_9   updates\n"
+            "vim-enhanced.x86_64   2:7.4.629-8.el7_9   updates\n__NETORK_RC=100"),
+            ("RHSA-2023:3556 Important/Sec. openssl-libs-1:1.0.2k-26.el7_9.x86_64\n"
+            "__DNF_RC=0\n__NETORK_RC=0"),
+        ]
+
+        updates = {u["name"]: u for u in driver.get_available_updates()}
+
+        assert driver._device.send_command.call_count == 3
+        assert "echo __DNF_RC=$?" in driver._device.send_command.call_args_list[2][0][0]
+        assert updates["openssl-libs"]["security"] is True
+        assert updates["vim-enhanced"]["security"] is False
+
+    def test_yum_with_no_advisories_marks_nothing_as_security(self, driver):
+        driver._pkg_manager = "yum"
+        driver._device.send_command.side_effect = [
+            "0",
+            "vim-enhanced.x86_64   2:7.4.629-8.el7_9   updates\n__NETORK_RC=100",
+            "__DNF_RC=0\n__NETORK_RC=0",
+        ]
+
+        assert driver.get_available_updates()[0]["security"] is False
+
+    def test_yum_that_failed_leaves_security_unknown(self, driver):
+        # A failed yum prints nothing on stdout (2>/dev/null): read as "no advisories",
+        # every update would be marked as not a security fix.
+        driver._pkg_manager = "yum"
+        driver._device.send_command.side_effect = [
+            "0",
+            "vim-enhanced.x86_64   2:7.4.629-8.el7_9   updates\n__NETORK_RC=100",
+            "__DNF_RC=1\n__NETORK_RC=0",
+        ]
+
+        assert driver.get_available_updates()[0]["security"] is None
+
+    def test_yum_without_its_exit_status_leaves_security_unknown(self, driver):
+        driver._pkg_manager = "yum"
+        driver._device.send_command.side_effect = [
+            "0",
+            "vim-enhanced.x86_64   2:7.4.629-8.el7_9   updates\n__NETORK_RC=100",
+            "RHSA-2023:3556 Important/Sec. vim-enhanced-2:7.4.629-8.el7_9.x86_64\n__NETORK_RC=0",
         ]
 
         assert driver.get_available_updates()[0]["security"] is None
@@ -1562,8 +1604,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
@@ -1577,6 +1618,84 @@ class TestGetListeningSockets:
         assert plain.startswith("sh -c '")
 
 
+# ---------------------------------------------------------------------------
+# get_system_identity (napalm-device-types SystemIdentityMixin)
+# ---------------------------------------------------------------------------
+
+
+class TestGetSystemIdentity:
+    """What ties a host to its BMC: SMBIOS UUID and serials, read as root on an exec
+    channel (product_uuid and product_serial are root's)."""
+
+    def test_it_reads_the_shared_command_as_root_on_an_exec_channel(self, driver):
+        from napalm_device_types import SYSTEM_IDENTITY_COMMAND, CommandResult, SystemIdentityMixin
+
+        assert isinstance(driver, SystemIdentityMixin)
+        calls = []
+
+        def run_command(command, *, privileged=False, timeout=60, stdin=None):
+            calls.append((command, privileged))
+            out = "IDBEGIN\n31303043-534D-C40C-7A72-6F1300000000\n0123456789\nZM155S026022\nIDEND\n"
+            return CommandResult(out, "", 0)
+
+        driver.run_command = run_command
+
+        identity = driver.get_system_identity()
+
+        assert identity == {
+            "system_uuid": "31303043-534d-c40c-7a72-6f1300000000",
+            "product_serial": "",  # Supermicro's placeholder
+            "board_serial": "ZM155S026022",
+        }
+        assert calls == [(SYSTEM_IDENTITY_COMMAND, True)]
+        driver._device.send_command.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# get_host_firewall (napalm-device-types HostFirewallMixin, NetOrk/netork#1344)
+# ---------------------------------------------------------------------------
+
+_FIREWALL = (
+    "FW_BEGIN\n[iptables-save]\n"
+    "# Generated by iptables-save v1.8.10 (nf_tables) on Wed Oct 21 14:52:11 2026\n"
+    "*filter\n:INPUT ACCEPT [0:0]\n-A INPUT -i ens6 -j DROP\nCOMMIT\n__FW_RC=0\nFW_END\n"
+)
+
+
+class TestGetHostFirewall:
+    """nft and iptables-save need root; the nft JSON is one long line, so the
+    command goes on an exec channel, never typed into the PTY netmiko holds."""
+
+    def test_it_reads_the_shared_command_as_root_on_an_exec_channel(self, driver):
+        from napalm_device_types import CommandResult, HostFirewallMixin
+
+        assert isinstance(driver, HostFirewallMixin)
+        calls = []
+
+        def run_command(command, *, privileged=False, timeout=60, stdin=None):
+            calls.append((command, privileged))
+            return CommandResult(_FIREWALL, "", 0)
+
+        driver.run_command = run_command
+
+        reading = driver.get_host_firewall()
+
+        assert reading["sources"] == ["iptables-nft"]
+        assert reading["chains"][0]["rules"][0]["verdict"] == "drop"
+        [(command, privileged)] = calls
+        assert privileged is True
+        assert command.startswith("sh -c '")
+        driver._device.send_command.assert_not_called()
+
+    def test_without_root_there_is_no_reading(self, driver):
+        from napalm_device_types import CommandResult, HostFirewallUnavailable
+
+        driver.run_command = lambda *a, **k: CommandResult("", "sudo: a password is required\n", 1)
+
+        with pytest.raises((HostFirewallUnavailable, ValueError)):
+            driver.get_host_firewall()
+
+
 # ---------------------------------------------------------------------------
 # get_virtual_ips (napalm-device-types VirtualIpsMixin)
 # ---------------------------------------------------------------------------
@@ -1651,9 +1770,15 @@ class TestGetVirtualIps:
             def recv_ready(self):
                 return False
 
+            def recv(self, size):
+                return b""
+
             def recv_stderr_ready(self):
                 return False
 
+            def recv_stderr(self, size):
+                return b""
+
             def exit_status_ready(self):
                 return True
 
@@ -1675,6 +1800,44 @@ class TestGetVirtualIps:
         assert plain.startswith("sh -c '")
 
 
+# ---------------------------------------------------------------------------
+# get_service_clusters (napalm-device-types ServiceClustersMixin)
+# ---------------------------------------------------------------------------
+
+_CLUSTERS = (
+    "SVCC_BEGIN\n[systemd]\n[patroni patroni.service]\nactive active\nscope netork-db\n"
+    "name db-01\nrestapi 10.7.234.12:8008\nrest-unreachable\n[no-docker]\nSVCC_END\n"
+)
+
+
+class TestGetServiceClusters:
+    """The same template as get_virtual_ips: an exec channel, root first."""
+
+    _exec = staticmethod(TestGetVirtualIps._exec)
+
+    def test_it_reads_with_the_shared_command_on_an_exec_channel(self, driver):
+        from napalm_device_types import ServiceClustersMixin
+
+        assert isinstance(driver, ServiceClustersMixin)
+        calls = self._exec(driver, (_CLUSTERS, "", 0))
+
+        reading = driver.get_service_clusters()
+
+        assert reading["patroni"][0]["cluster"] == "netork-db"
+        [(command, privileged)] = calls
+        assert privileged is True
+        assert command.startswith("sh -c '")
+        driver._device.send_command.assert_not_called()
+
+    def test_when_sudo_refuses_it_reads_what_the_user_may_see(self, driver):
+        calls = self._exec(driver, ("", "sudo: a password is required\n", 1), (_CLUSTERS, "", 0))
+
+        reading = driver.get_service_clusters()
+
+        assert reading["bunkerweb"] == []
+        assert [privileged for _, privileged in calls] == [True, False]
+
+
 class TestHardensItsSshServer:
     """A stock Linux host runs a stock OpenSSH: the hardening netOrk offers for
     ssh-audit findings comes from napalm-device-types, carried over run_command

tests/test_ssh_host_key.py

@@ -0,0 +1,86 @@
+"""open() hands netmiko the host key to expect, and reports the one it saw (NetOrk/netork#1167).
+
+OpenMediaVault and QNAP open through ``LinuxDriver.open()`` and get this too.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import paramiko
+import pytest
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+from netmiko.exceptions import NetmikoTimeoutException
+
+from napalm_linux.linux import LinuxDriver
+
+SERVER_KEY = paramiko.ECDSAKey.generate()
+OTHER_KEY = paramiko.ECDSAKey.generate()
+
+
+def _device() -> MagicMock:
+    device = MagicMock()
+    device.send_command.return_value = "/usr/bin/apt"
+    device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY
+    return device
+
+
+def _driver(seen: list, known: SshHostKey | None = None) -> LinuxDriver:
+    args: dict = {"port": 2222, OBSERVER_ARG: seen.append}
+    if known is not None:
+        args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()}
+    return LinuxDriver("10.0.0.8", "admin", "s3cret", optional_args=args)
+
+
+def test_first_use_lets_netmiko_accept_and_reports_the_key():
+    seen: list = []
+    with patch("napalm_linux.linux.ConnectHandler", return_value=_device()) as handler:
+        _driver(seen).open()
+    kwargs = handler.call_args.kwargs
+    assert kwargs["ssh_strict"] is False
+    assert kwargs["alt_host_keys"] is False
+    assert kwargs["system_host_keys"] is False
+    assert seen[0]["accepted"] is True
+    assert seen[0]["seen"]["fingerprint"] == SshHostKey.of(SERVER_KEY).fingerprint
+    assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.8", 2222)
+
+
+def test_a_known_key_is_the_only_one_netmiko_accepts():
+    seen: list = []
+    known = SshHostKey.of(SERVER_KEY)
+    files: list[str] = []
+
+    def connect(**kwargs):
+        files.append(kwargs["alt_key_file"])
+        with open(kwargs["alt_key_file"]) as f:
+            assert f.read().strip() == known.known_hosts_line("10.0.0.8", 2222)
+        assert kwargs["ssh_strict"] is True and kwargs["alt_host_keys"] is True
+        return _device()
+
+    with patch("napalm_linux.linux.ConnectHandler", side_effect=connect):
+        _driver(seen, known).open()
+    assert not os.path.exists(files[0])
+    assert seen[0]["accepted"] is True
+
+
+def test_a_changed_key_is_refused():
+    seen: list = []
+
+    def refuse(**kwargs):
+        try:
+            raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
+        except paramiko.SSHException as exc:  # what netmiko does with it
+            raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}")
+
+    driver = _driver(seen, SshHostKey.of(OTHER_KEY))
+    with patch("napalm_linux.linux.ConnectHandler", side_effect=refuse):
+        with pytest.raises(SshHostKeyMismatch):
+            driver.open()
+    assert driver._device is None
+    assert seen[0]["accepted"] is False

tests/test_staged_root_writes.py

@@ -0,0 +1,255 @@
+"""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 recv_stderr(self, size):
+        return b""
+
+    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},
+    )
+    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]
+
+
+_SETTINGS = {"community": "c0mm", "allowed_sources": []}
+
+ACTIONS = [
+    (
+        lambda driver: driver.configure_snmp_agent(_SETTINGS),
+        "/etc/snmp/snmpd.conf",
+        b"agentAddress udp:161",
+    ),
+    (
+        lambda driver: driver.configure_package_proxy("http://proxy:3142"),
+        "/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)
+
+        action(driver)
+
+        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)
+
+        action(driver)
+
+        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)
+
+        action(driver)
+
+        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)
+
+        action(driver)
+
+        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)
+
+        action(driver)
+
+        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)
+
+        action(driver)
+
+        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.configure_package_proxy("http://proxy:3142")
+
+    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,262 @@
+"""The sudo password never goes into a command line or a shell history (NetOrk/netork#1039).
+
+``_sudo`` typed ``echo '<password>' | sudo -S -p "" <command>`` into netmiko's
+interactive shell. That shell keeps a history, so every poll appended the
+password to the login user's ``~/.bash_history`` -- 121 lines of it on one host
+-- and while the command ran it was in the process list for every local user.
+
+Privileged commands now run on an exec channel next to netmiko's PTY, the way
+``run_command(privileged=True)`` does: the password reaches ``sudo -S`` on
+stdin, and an exec channel has no history and no PTY to echo it back. The
+interactive session switches its history off before anything else, in case
+anything ever slips through again.
+"""
+
+from __future__ import annotations
+
+import shlex
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from napalm_linux import LinuxDriver
+from napalm_linux.linux import _HISTORY_OFF
+
+_PW = "s3cr3t 'q' $x"
+
+
+class Channel:
+    """An exec channel answering with *output* and *status*; it refuses a line
+    that holds the password, which is what ``ps`` would show."""
+
+    def __init__(self, host: "Transport"):
+        self.host = host
+        self.command = ""
+        self.stdin = b""
+        self.timeout = None
+        self._out = b""
+
+    def exec_command(self, command):
+        assert _PW not in command, f"sudo password in a command line: {command!r}"
+        self.command = command
+        output, status = self.host.answer
+        self._out = output.encode()
+        self.status = status
+
+    def settimeout(self, timeout):
+        self.timeout = timeout
+
+    def sendall(self, data):
+        self.stdin += data
+
+    def shutdown_write(self):
+        pass
+
+    def close(self):
+        pass
+
+    def recv_ready(self):
+        return bool(self._out)
+
+    def recv(self, size):
+        chunk, self._out = self._out[:size], self._out[size:]
+        return chunk
+
+    def recv_stderr_ready(self):
+        return False
+
+    def recv_stderr(self, size):
+        return b""
+
+    def exit_status_ready(self):
+        return not self._out
+
+    def recv_exit_status(self):
+        return self.status
+
+
+class Transport:
+    def __init__(self, output="", status=0):
+        self.answer = (output, status)
+        self.channels: list[Channel] = []
+
+    def open_session(self):
+        self.channels.append(Channel(self))
+        return self.channels[-1]
+
+
+def _driver(*, sudo_password=_PW, root=False, output="", status=0):
+    driver = LinuxDriver("h", "admin", "p", optional_args={"sudo_password": sudo_password})
+    transport = Transport(output, status)
+    driver._device = MagicMock()
+    driver._device.remote_conn_pre = SimpleNamespace(get_transport=lambda: transport)
+    driver._root = root
+    return driver, transport
+
+
+def _script(command_line: str) -> str:
+    """What ``sh -c`` runs in *command_line*, unquoted."""
+    parts = shlex.split(command_line)
+    return parts[parts.index("-c") + 1]
+
+
+class TestSudo:
+    def test_the_password_goes_to_sudo_on_stdin(self):
+        driver, transport = _driver(output="ok\n")
+
+        assert driver._sudo("apt-get update 2>&1") == "ok"
+
+        channel = transport.channels[-1]
+        assert channel.command.startswith("sudo -S -p '' sh -c ")
+        assert channel.stdin == f"{_PW}\n".encode()
+
+    def test_nothing_goes_through_the_interactive_shell(self):
+        """Its history is where the password ended up."""
+        driver, _ = _driver()
+
+        driver._sudo("apt-get update")
+
+        driver._device.send_command.assert_not_called()
+
+    def test_the_whole_command_runs_as_root(self):
+        """``echo pw | sudo -S mv a b && chown …`` ran only ``mv`` as root."""
+        driver, transport = _driver()
+
+        driver._sudo("mv /tmp/x /etc/x && chown root:root /etc/x")
+
+        assert _script(transport.channels[-1].command).endswith(
+            "mv /tmp/x /etc/x && chown root:root /etc/x"
+        )
+
+    def test_a_sudo_that_does_not_prompt_leaves_the_password_to_nobody(self):
+        """NOPASSWD or a cached timestamp leave the password on stdin; the
+        command's stdin is /dev/null, so nothing it runs reads it."""
+        driver, transport = _driver()
+
+        driver._sudo("dpkg --configure -a")
+
+        assert _script(transport.channels[-1].command).startswith("exec </dev/null; ")
+
+    def test_sudos_own_complaints_are_in_the_output(self):
+        """``sudo: a password is required`` goes to stderr; callers read it from the output."""
+        driver, transport = _driver()
+
+        driver._sudo("true")
+
+        assert transport.channels[-1].command.endswith(" 2>&1")
+
+    def test_without_a_password_sudo_never_waits_for_one(self):
+        driver, transport = _driver(sudo_password=None)
+
+        driver._sudo("apt-get update")
+
+        assert transport.channels[-1].command.startswith("sudo -n sh -c ")
+        assert transport.channels[-1].stdin == b""
+
+    def test_a_root_login_needs_no_sudo(self):
+        driver, transport = _driver(root=True)
+
+        driver._sudo("apt-get update")
+
+        assert transport.channels[-1].command.startswith("sh -c ")
+        assert transport.channels[-1].stdin == b""
+
+    def test_the_read_timeout_is_the_channels(self):
+        driver, transport = _driver()
+
+        driver._sudo("apt-get upgrade -y", read_timeout=600)
+
+        assert transport.channels[-1].timeout == 600
+
+
+class TestSudoStatus:
+    def test_the_exit_status_is_the_commands_own(self):
+        driver, _ = _driver(output="E: Unable to locate package nope\n", status=100)
+
+        assert driver._sudo_status("apt-get remove -y nope 2>&1") == (
+            "E: Unable to locate package nope",
+            100,
+        )
+
+    def test_a_refused_sudo_is_a_failure(self):
+        driver, _ = _driver(output="Sorry, try again.\n", status=1)
+
+        assert driver._sudo_status("apt-get remove -y x")[1] == 1
+
+
+class TestRunPrivileged:
+    def test_with_a_password_it_takes_the_exec_channel(self):
+        driver, transport = _driver(output="__NETORK_RC=0\n")
+
+        driver._run_privileged("apt-get update -q 2>&1; echo __NETORK_RC=$?", 180)
+
+        driver._device.send_command.assert_not_called()
+        assert transport.channels[-1].stdin == f"{_PW}\n".encode()
+
+    def test_a_refresh_reads_its_status_from_the_exec_channel(self):
+        driver, _ = _driver(output="Hit:1 http://deb.debian.org\n__NETORK_RC=0\n")
+        driver._pkg_manager = "apt"
+
+        assert driver.refresh_available_updates()["success"] is True
+
+
+class TestWireGuardWithSudo:
+    def test_the_dump_comes_straight_from_the_exec_channel(self):
+        """No PTY to turn its tabs into spaces, so no file in /tmp holding the
+        interfaces' private keys."""
+        dump = "wg0\tPRIV\tPUB\t51820\toff\nwg0\tPEER\t(none)\t1.2.3.4:51820\t10.0.0.2/32\t0\t1\t2\toff\n"
+        driver, transport = _driver(output=dump)
+
+        tunnels = driver.get_vpn_tunnels()
+
+        assert list(tunnels) == ["wireguard-wg0-PEER"]
+        assert "/tmp" not in transport.channels[-1].command
+        driver._device.send_command.assert_not_called()
+
+
+class TestHistoryOff:
+    """Defense in depth: the interactive session keeps no history at all."""
+
+    def test_it_is_the_first_thing_the_session_sends(self):
+        driver = LinuxDriver("h", "admin", "p", optional_args={"sudo_password": _PW})
+        device = MagicMock()
+        device.send_command.return_value = "/usr/bin/apt"
+        with patch("napalm_linux.linux.ConnectHandler", return_value=device):
+            driver.open()
+
+        first = device.send_command.call_args_list[0].args[0]
+        assert first == _HISTORY_OFF
+
+    def test_it_leaves_no_history_file_and_records_nothing_more(self):
+        assert _HISTORY_OFF.startswith(" ")  # HISTCONTROL=ignorespace skips the line itself
+        assert "unset HISTFILE" in _HISTORY_OFF
+        assert "set +o history" in _HISTORY_OFF
+
+    def test_a_shell_that_refuses_it_does_not_stop_the_connection(self):
+        driver = LinuxDriver("h", "admin", "p")
+        device = MagicMock()
+        # history off (refused), prompt hooks off (#41), stty, package manager
+        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)

tests/test_systemd_osc_context.py

@@ -0,0 +1,78 @@
+"""systemd's shell context sequences stay out of every command's output (#41).
+
+systemd 258 (Ubuntu 26.04) ships ``/etc/profile.d/80-systemd-osc-context.sh``.
+An interactive bash then wraps every command in OSC 3008 sequences, from ``PS0``
+before it runs and from ``PROMPT_COMMAND`` before the next prompt, each with a
+fresh UUID. netmiko's interactive shell is such a bash: the sequences ended up in
+every parsed value, and since the prompt never looked the same twice, netmiko
+left the prompt line in the output as well. ``os_version`` grew past 255
+characters and netOrk could not store the poll.
+"""
+
+from unittest.mock import MagicMock, patch
+
+from napalm_linux.linux import _HISTORY_OFF, _PROMPT_HOOKS_OFF, LinuxDriver
+
+#: What ``cat /etc/os-release | grep PRETTY_NAME ...`` returned through netmiko on
+#: an Ubuntu 26.04.1 host (2026-10-23), the prompt line already gone.
+_UBUNTU_2604_OS_RELEASE = (
+    "\x1b]3008;end=87c2e1c5-f850-4429-9f06-749e9be783e6;exit=success\x1b\\"
+    "\x1b]3008;start=bc3bc23b-2652-4826-8192-be447f97dd6f;type=command;"
+    "machineid=35cf42f3beb446bbb9aa95b937a0cf48;user=netork;hostname=ubuntu;"
+    "bootid=d32bd88c-dfce-40fe-9869-257bf50ef0e0;pid=20059;cwd=/home/netork\x1b\\"
+    "Ubuntu 26.04.1 LTS\n"
+)
+
+
+def _opened(device: MagicMock) -> LinuxDriver:
+    driver = LinuxDriver("h", "admin", "p")
+    with patch("napalm_linux.linux.ConnectHandler", return_value=device):
+        driver.open()
+    return driver
+
+
+class TestPromptHooksOff:
+    def test_the_session_drops_the_hooks_right_after_the_history(self):
+        device = MagicMock()
+        device.send_command.return_value = "/usr/bin/apt"
+
+        _opened(device)
+
+        sent = [c.args[0] for c in device.send_command.call_args_list]
+        assert sent[:2] == [_HISTORY_OFF, _PROMPT_HOOKS_OFF]
+
+    def test_it_unsets_both_hooks_including_an_array_prompt_command(self):
+        # The script appends to PROMPT_COMMAND as an array; assigning the empty
+        # string would only clear element 0 and leave systemd's hook running.
+        assert _PROMPT_HOOKS_OFF.startswith(" ")
+        assert "unset PS0 PROMPT_COMMAND" in _PROMPT_HOOKS_OFF
+
+    def test_the_prompt_is_read_again_once_the_hooks_are_gone(self):
+        # netmiko learned the prompt with the sequences in front of it; a prompt
+        # carrying a UUID never matches again, so strip_prompt would keep failing.
+        device = MagicMock()
+        device.send_command.return_value = "/usr/bin/apt"
+
+        _opened(device)
+
+        calls = [(c[0], c.args[0] if c.args else None) for c in device.mock_calls]
+        hooks_off = calls.index(("send_command", _PROMPT_HOOKS_OFF))
+        assert "set_base_prompt" in [name for name, _ in calls[hooks_off + 1 :]]
+
+    def test_a_shell_that_refuses_it_does_not_stop_the_connection(self):
+        device = MagicMock()
+        device.send_command.side_effect = ["", Exception("no prompt"), "", "/usr/bin/apt"]
+
+        driver = _opened(device)
+
+        assert driver._device is device
+
+
+class TestSendStripsTerminalCodes:
+    def test_os_version_comes_back_without_the_context_sequences(self):
+        device = MagicMock()
+        device.send_command.return_value = _UBUNTU_2604_OS_RELEASE
+        driver = LinuxDriver("h", "admin", "p")
+        driver._device = device
+
+        assert driver._send("cat /etc/os-release") == "Ubuntu 26.04.1 LTS"