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.

.github/workflows/ci.yml

@@ -0,0 +1,48 @@
+name: CI
+
+on:
+  push:
+    branches: ["**"]
+  pull_request:
+    branches: ["**"]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: pip
+
+      - name: Install package with dev extras
+        run: |
+          python -m pip install --upgrade pip
+          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
+          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
+          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
+          python -m pip install -e ".[dev]"
+
+      - name: Run unit tests
+        run: |
+          python -m pytest -q --tb=short
+
+      - name: Build wheel and sdist
+        run: |
+          python -m pip install build
+          python -m build
+
+      - name: Upload dist artifacts
+        # v4 refuses to run on Gitea ("not currently supported on GHES").
+        uses: actions/upload-artifact@v3
+        with:
+          name: dist-${{ matrix.python-version }}
+          path: dist/*

CHANGELOG.md

@@ -7,6 +7,136 @@ 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
+
+- `get_virtual_ips()` through napalm-device-types' `VirtualIpsMixin` (4.2.0): the
+  addresses vip-manager and keepalived configuration lets float onto this host,
+  whether or not it holds them right now (NetOrk/netork#829). The command goes on
+  the exec channel (`run_command`): with its awk filter it is about 4 kB, past the
+  line the PTY reliably takes. Root first, then without it.
+
+## [0.3.0] – 2026-10-08
+
+### Removed
+
+- `get_docker_info()`, `get_docker_outdated()`, `reconstruct_docker_run()` and
+  the `_docker_bin()` hook, with the image-ID helpers they used. netOrk reads
+  and handles containers itself over the Engine API, through
+  `open_container_engine()` (NetOrk/netork#765), and no longer calls them.
+  `run_device_action("fix_docker_permissions")` stays: letting the login user
+  use Docker is an OS action.
+
+## [0.2.0] – 2026-10-07
+
+### Added
+
+- `run_command()` and `open_stream()`, the public command channel from
+  napalm-device-types 2.6.0: an exec channel on the existing SSH transport, so
+  no PTY, separate stderr and a real exit code. `privileged=True` runs as root
+  directly, through `sudo -S` with the password on stdin (never on the command
+  line), or through `sudo -n`, which fails instead of prompting.
+- `ContainerEngineMixin`: `container_engines()` and `open_container_engine()`,
+  whose `open_api()` streams the Docker Engine API over `docker system
+  dial-stdio` (NetOrk/netork#765). The existing Docker methods are unchanged.
+
+### Changed
+
+- Requires napalm-device-types >= 2.6.0.
+
 ## [0.1.0] – 2026-05-29
 
 ### Added

README.md

@@ -48,7 +48,8 @@ with Driver(
     optional_args={
         # "port": 22,
         # "pkg_manager": "apt",   # force package manager; auto-detected by default
-        # "secret": "sudo-pass",  # password for sudo / enable (defaults to login password)
+        # "sudo_password": "sudo-pass",  # for commands that need root; without it,
+        #                                # `sudo -n` (passwordless sudo) is tried
         # "debugging": True,      # enable verbose logging
     },
 ) as dev:
@@ -59,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()
 
@@ -69,6 +70,10 @@ with Driver(
 
     # Upgrade everything with pending updates
     result = dev.apply_updates([])
+
+    # Restart a service (start, stop, restart, enable, disable)
+    result = dev.manage_service("cron", "restart")
+    print(result)  # {"success": True, "output": ""}
 ```
 
 ## Supported NAPALM methods
@@ -99,8 +104,9 @@ with Driver(
 | `get_packages()` | ✅ | apt, dnf, yum, apk, pacman |
 | `get_pending_updates()` | ✅ | apt, dnf, yum, apk, pacman |
 | `apply_updates(packages)` | ✅ | apt, dnf, yum, apk, pacman |
-| `get_services()` | ✅ | systemd (fallback: SysV `service`) |
+| `get_services()` | ✅ | systemd, one round trip (fallback: SysV `service`) |
-| `get_users()` | ✅ | `/etc/passwd` + `/etc/group` |
+| `manage_service(name, action)` | ✅ | systemd: start, stop, restart, enable, disable |
+| `get_os_users()` | ✅ | `/etc/passwd` + `/etc/group` |
 | `get_processes()` | ✅ | `ps axo` |
 | `get_cron_jobs()` | ✅ | user crontabs + `/etc/cron.d/` |
 
@@ -127,13 +133,30 @@ The SSH user needs read access to:
 | `/etc/passwd`, `/etc/group` | world-readable (default) |
 | `/proc/uptime`, `/sys/class/dmi/…` | world-readable (default) |
 | User crontabs (`/var/spool/cron/…`) | `root` or `sudo` required |
-| `systemctl is-enabled <unit>` | unprivileged on most distros |
+| `systemctl list-unit-files`, `systemctl show` | unprivileged |
+| `systemctl start/stop/restart/enable/disable` | `root`, or `sudo` (with `sudo_password`, or passwordless) |
 | `apt list --upgradable` | may require `apt-get update` (root) |
 | `dnf check-update` / `yum check-update` | unprivileged, but slower without cache |
 
 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
+on its way up or down cannot hold the session, and only the exit status decides whether it
+succeeded. Without a sudo password it uses `sudo -n`, which fails at once instead of
+waiting for a password prompt.
+
+On OpenMediaVault (napalm-openmediavault inherits this driver), enabling or disabling a
+unit that OMV manages itself — Samba, NFS, SSH — may be reverted the next time OMV applies
+its configuration.
+
 ## Tested distributions
 
 | Distribution | Version | Package manager | Tested |

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-linux"
-version = "0.1.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>=0.3.0",
+    "napalm-device-types>=4.21.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]

tests/test_command_channel.py

@@ -0,0 +1,179 @@
+"""The public command channel and container engine access (napalm-device-types 2.6.0).
+
+netOrk used to reach a Linux host's shell through the private ``_send``: an
+interactive PTY, stdout and stderr merged, no exit code. ``run_command`` and
+``open_stream`` go through an exec channel on the same SSH transport instead,
+and ``open_container_engine`` builds on them (NetOrk/netork#765). Privileges
+work as they do everywhere else in this driver: root runs directly, a sudo
+password goes to ``sudo -S`` on stdin and never onto a command line, and
+without one ``sudo -n`` fails at once instead of hanging.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from napalm.base.exceptions import ConnectionClosedException
+
+from napalm_linux import LinuxDriver
+
+
+class FakeChannel:
+    def __init__(self):
+        self.command = None
+        self.sent = b""
+
+    def exec_command(self, command):
+        self.command = command
+
+    def settimeout(self, timeout):
+        pass
+
+    def sendall(self, data):
+        self.sent += data
+
+    def shutdown_write(self):
+        pass
+
+    def close(self):
+        pass
+
+    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
+
+    def recv_exit_status(self):
+        return 0
+
+
+class FakeTransport:
+    def __init__(self):
+        self.channels = []
+
+    def open_session(self):
+        self.channels.append(FakeChannel())
+        return self.channels[-1]
+
+
+def _driver(*, root=False, sudo_password=None):
+    driver = LinuxDriver("h", "u", "p", optional_args={"sudo_password": sudo_password})
+    transport = FakeTransport()
+    driver._device = SimpleNamespace(remote_conn_pre=SimpleNamespace(get_transport=lambda: transport))
+    driver._root = root
+    return driver, transport
+
+
+def test_an_unprivileged_command_runs_as_given():
+    driver, transport = _driver()
+
+    result = driver.run_command("docker version", timeout=5)
+
+    assert transport.channels[-1].command == "docker version"
+    assert transport.channels[-1].sent == b""
+    assert result.exit_code == 0
+
+
+def test_a_privileged_command_with_a_sudo_password_reads_it_from_stdin():
+    driver, transport = _driver(sudo_password="s3cr3t")
+
+    driver.run_command("usermod -aG docker u", privileged=True, timeout=5)
+
+    channel = transport.channels[-1]
+    assert channel.command == "sudo -S -p '' sh -c 'usermod -aG docker u'"
+    assert channel.sent == b"s3cr3t\n"
+    assert "s3cr3t" not in channel.command
+
+
+def test_a_privileged_command_without_a_password_fails_fast_instead_of_prompting():
+    driver, transport = _driver()
+
+    driver.run_command("id", privileged=True, timeout=5)
+
+    assert transport.channels[-1].command == "sudo -n sh -c id"
+
+
+def test_a_root_login_needs_no_sudo():
+    driver, transport = _driver(root=True, sudo_password="s3cr3t")
+
+    driver.run_command("id", privileged=True, timeout=5)
+
+    assert transport.channels[-1].command == "id"
+    assert transport.channels[-1].sent == b""
+
+
+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)
+
+    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("journalctl -f", privileged=True)
+
+    channel = transport.channels[-1]
+    assert channel.command == "sudo -S -p '' sh -c 'journalctl -f'"
+    assert channel.sent == b"pw\n"
+
+
+def test_the_container_engine_api_is_a_stream_over_dial_stdio():
+    driver, transport = _driver()
+
+    driver.open_container_engine("docker").open_api()
+
+    assert transport.channels[-1].command == "docker system dial-stdio"
+
+
+def test_the_channel_needs_an_open_connection():
+    driver = LinuxDriver("h", "u", "p")
+
+    with pytest.raises(ConnectionClosedException):
+        driver.run_command("true")

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_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"