Merge pull request 'chore: adopt the NAPALM driver standard' (#38 ) from chore/driver-standard into main

chore(release): 0.15.0
ci: say which napalm-device-types the git install waits for
2026-10-23 12:55:55 +00:00 · 2026-10-23 12:02:11 +02:00 · 2026-10-23 08:29:33 +02:00 · 2026-10-23 08:29:05 +02:00 · 2026-10-22 22:08:24 +00:00 · 2026-10-22 19:16:40 +02:00
14 changed files with 3225 additions and 1160 deletions
@@ -0,0 +1,17 @@
+name: CI
+
+on:
+  push:
+    branches: [main, master]
+    tags: ["v*"]
+  pull_request:
+
+jobs:
+  driver:
+    uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
+    with:
+      standard-ref: main
+      # Until napalm-device-types 4.22.0 (AptProxyMixin, #53) is on PyPI.
+      pre-install: "git+https://git.netork.io/NAPALM/napalm-device-types.git"
+    secrets:
+      PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
@@ -7,6 +7,152 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0

 ## [Unreleased]

+## [0.15.0] - 2026-10-23
+
+The first release on PyPI. It collects everything since 0.4.0, which was the last version recorded here.
+
+### Changed
+- The driver follows the NAPALM driver standard
+  (https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
+  tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
+  a build check, and a release tag publishes to PyPI. The package now ships `py.typed`
+  and a LICENSE file. **Python 3.9 is no longer supported**; napalm-device-types never
+  ran on it.
+- Exceptions the driver deliberately ignores (the PTY width at login, closing a dead
+  session, unreadable route lines, the snmpd.conf community and port) are now logged
+  at debug level.
+
+### Added
+- apt's proxy through napalm-device-types' `AptProxyMixin` (NetOrk/netork#1397). `get_package_proxy()` is what `apt-config dump` says, not `00proxy` alone: a proxy in `90curtin-aptproxy` or `/etc/apt/apt.conf` wins over it, and was missed. New: `get_package_proxy_report()` (every line that sets it, per-host overrides, the https sources) and `check_package_proxy(url, expect)` (a fetch with `apt-helper`, past any proxy). `configure_package_proxy(url)` now also comments out other proxy lines, reads the result back and fetches one `InRelease` through the proxy; it fails when apt still fetches elsewhere. Reads run on the exec channel. Requires napalm-device-types 4.22.0; version 0.14.0.
+- `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
@@ -0,0 +1,191 @@
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship made available under
+      the License, as indicated by a copyright notice that is included in
+      or attached to the work (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other transformations
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean, as submitted to the Licensor for inclusion
+      in the Work by the copyright owner or by an individual or Legal Entity
+      authorized to submit on behalf of the copyright owner. For the purposes
+      of this definition, "submitted" means any form of electronic, verbal,
+      or written communication sent to the Licensor or its representatives,
+      including but not limited to communication on electronic mailing lists,
+      source code control systems, and issue tracking systems that are managed
+      by, or on behalf of, the Licensor for the purpose of discussing and
+      improving the Work, but excluding communication that is conspicuously
+      marked or designated in writing by the copyright owner as "Not a
+      Contribution."
+
+      "Contributor" shall mean Licensor and any Legal Entity on behalf of
+      whom a Contribution has been received by the Licensor and incorporated
+      within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by the combined Contribution(s) and the Work
+      to which such Contribution(s) was submitted. If You institute patent
+      litigation against any entity (including a cross-claim or counterclaim
+      in a lawsuit) alleging that the Work or any Contribution embodied
+      within the Work constitutes direct or contributory patent infringement,
+      then any patent licenses granted to You under this License for that
+      Work shall terminate as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or Derivative
+          Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, You must include a readable copy of the
+          attribution notices contained within such NOTICE file, in
+          at least one of the following places: within a NOTICE text file
+          distributed as part of the Derivative Works; within the Source
+          form or documentation, if provided along with the Derivative
+          Works; or, within a display generated by the Derivative Works,
+          if and wherever such third-party notices normally appear. The
+          contents of the NOTICE file are for informational purposes only
+          and do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own license statement for Your modifications and
+      may provide additional grant of rights to use, copy, modify, merge,
+      publish, distribute, sublicense, and/or sell copies of the Work,
+      and to permit persons to whom the Work is furnished to do so,
+      subject to the following conditions:
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any conditions of TITLE,
+      MERCHANTIBILITY, or FITNESS FOR A PARTICULAR PURPOSE.
+      See the License for the specific language governing permissions and
+      limitations under the License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or exemplary damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (even if such Contributor has been advised of the possibility
+      of such damages).
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may offer only
+      conditions consistent with this License, subject to the following
+      additional conditions:
+
+      (a) You may offer additional warranty, indemnity, or other liability
+          terms and conditions; and
+
+      (b) You may offer support for the Work in return for a fee.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format in question. It may also be
+      possible to make it available with a creative commons license.
+
+   Copyright 2026 Christian Manivong
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
@@ -24,7 +24,7 @@ pip install napalm napalm-linux
 Or from source:

 ```bash
-git clone https://github.com/chrismanivong/napalm-linux
+git clone https://git.netork.io/NAPALM/napalm-linux
 pip install -e napalm-linux/
 ```

@@ -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
@@ -1,14 +1,14 @@
 [build-system]
-requires = ["setuptools>=68", "wheel"]
+requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"

 [project]
 name = "napalm-linux"
-version = "0.1.0"
+version = "0.15.0"
 description = "NAPALM driver for generic Linux systems via SSH"
 readme = "README.md"
-requires-python = ">=3.9"
-license = { text = "Apache-2.0" }
+requires-python = ">=3.10"
+license = "Apache-2.0"
 authors = [
    { name = "Christian Manivong", email = "christian@manivong.de" },
 ]
@@ -24,43 +24,77 @@ classifiers = [
    "Development Status :: 3 - Alpha",
    "Intended Audience :: Developers",
    "Intended Audience :: System Administrators",
-    "License :: OSI Approved :: Apache Software License",
    "Operating System :: OS Independent",
    "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.9",
    "Programming Language :: Python :: 3.10",
    "Programming Language :: Python :: 3.11",
    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
    "Topic :: System :: Networking",
    "Topic :: System :: Systems Administration",
    "Typing :: Typed",
 ]
 dependencies = [
    "napalm>=4.0",
-    "napalm-device-types>=2.2.0",
+    "napalm-device-types>=4.22.0",
    "netmiko>=4.0.0",
    "paramiko>=5.0.0",  # CVE-2026-44405
 ]

 [project.optional-dependencies]
 dev = [
-    "pytest",
-    "pytest-cov",
-    "black",
-    "ruff",
-    "mypy",
+    "pytest>=8",
+    "pytest-cov>=5",
+    "ruff>=0.6",
+    "mypy>=1.11",
 ]

 [project.entry-points."napalm.drivers"]
 linux = "napalm_linux:LinuxDriver"

 [project.urls]
-Repository = "https://github.com/chrismanivong/napalm-linux"
+Repository = "https://git.netork.io/NAPALM/napalm-linux"
+Issues = "https://git.netork.io/NAPALM/napalm-linux/issues"
+Changelog = "https://git.netork.io/NAPALM/napalm-linux/src/branch/main/CHANGELOG.md"

 [tool.setuptools.packages.find]
 where = ["."]
 include = ["napalm_linux*"]

+[tool.setuptools.package-data]
+napalm_linux = ["py.typed"]
+
 [tool.ruff]
 line-length = 100
-target-version = "py39"
+target-version = "py310"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B"]
+
+[tool.ruff.lint.per-file-ignores]
+# Recorded device output keeps its lines.
+"tests/**" = ["E501"]
+
+[tool.mypy]
+python_version = "3.10"
+strict = true
+# napalm itself is untyped, so every driver subclasses Any.
+disallow_subclassing_any = false
+
+[[tool.mypy.overrides]]
+module = ["napalm.*", "netmiko.*", "paramiko.*"]
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+# Not yet under strict mypy (quality baseline); may only shrink.
+module = ["napalm_linux.linux"]
+ignore_errors = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.coverage.run]
+source = ["napalm_linux"]
+
+[tool.coverage.report]
+fail_under = 57
@@ -0,0 +1,28 @@
+{
+ "coverage_floor": 57.0,
+ "function_complexity": {
+  "napalm_linux/linux.py::LinuxDriver._collect_platform_info": 24,
+  "napalm_linux/linux.py::LinuxDriver._parse_cron_line": 12,
+  "napalm_linux/linux.py::LinuxDriver.get_networks": 13,
+  "napalm_linux/linux.py::LinuxDriver.get_route_to": 23,
+  "napalm_linux/linux.py::LinuxDriver.get_snmp_config": 13,
+  "napalm_linux/linux.py::LinuxDriver.get_vpn_tunnels": 28,
+  "napalm_linux/linux.py::LinuxDriver.search_packages": 26
+ },
+ "function_lines": {
+  "napalm_linux/linux.py::LinuxDriver._collect_platform_info": 106,
+  "napalm_linux/linux.py::LinuxDriver.get_networks": 52,
+  "napalm_linux/linux.py::LinuxDriver.get_route_to": 138,
+  "napalm_linux/linux.py::LinuxDriver.get_vpn_tunnels": 152,
+  "napalm_linux/linux.py::LinuxDriver.ping": 52,
+  "napalm_linux/linux.py::LinuxDriver.search_packages": 106,
+  "napalm_linux/linux.py::LinuxDriver.uninstall_package": 51
+ },
+ "mypy_relaxed": [
+  "napalm_linux.linux"
+ ],
+ "python_file_lines": {
+  "napalm_linux/linux.py": 2307
+ },
+ "untyped_public": []
+}
@@ -0,0 +1,181 @@
+"""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")
@@ -0,0 +1,308 @@
+"""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()
+
+
+_REPORT = (
+    "APROXY_BEGIN\n[dump]\n{dump}[parts]\n{parts}[main]\n[sources]\n"
+    "http://deb.debian.org/debian/dists/trixie/\nAPROXY_END\n"
+)
+
+
+def _host(driver, proxy="", parts="", fetched=0):
+    """What the host answers on the exec channel: apt's report, then a fetch."""
+    ran: list[str] = []
+
+    def run_command(command, *, privileged=False, timeout=60, stdin=None):
+        ran.append(command)
+        if "apt-config dump" in command:
+            dump = f'Acquire::http::Proxy "{proxy}";\n' if proxy else ""
+            return MagicMock(exit_code=0, stdout=_REPORT.format(dump=dump, parts=parts), stderr="")
+        out = f"APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n{fetched}\nAPROXY_END\n"
+        return MagicMock(exit_code=0, stdout=out, stderr="")
+
+    driver.run_command = run_command
+    return ran
+
+
+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="")
+        )
+        _host(driver, proxy="http://cache.lan:3142")
+
+        result = driver.configure_package_proxy("http://cache.lan:3142")
+
+        assert result["success"] is True, result["output"]
+        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_it_reads_back_and_fetches_through_it(self, driver):
+        driver._install_file_as_root = MagicMock(
+            return_value=MagicMock(exit_code=0, stdout="", stderr="")
+        )
+        ran = _host(driver, proxy="http://cache.lan:3142")
+
+        result = driver.configure_package_proxy("http://cache.lan:3142")
+
+        assert [s["step"] for s in result["steps"]] == ["configure", "conflicts", "verify", "fetch"]
+        assert any("dists/trixie/InRelease" in c for c in ran)
+
+    def test_another_file_s_proxy_is_commented_out_as_root(self, driver):
+        driver._install_file_as_root = MagicMock(
+            return_value=MagicMock(exit_code=0, stdout="", stderr="")
+        )
+        _host(
+            driver,
+            proxy="http://cache.lan:3142",
+            parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
+        )
+        ran = []
+        driver._sudo_status = lambda command, read_timeout=100: ran.append(command) or ("", 0)
+
+        driver.configure_package_proxy("http://cache.lan:3142")
+
+        assert ran and ran[0].startswith("sed -i -e '1{")
+        assert ran[0].endswith("/etc/apt/apt.conf.d/90curtin-aptproxy")
+
+    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_the_proxy_answers_past_apt_s_proxy(self, driver):
+        out = "APROXY_BEGIN\n[fetch]\nGet:1 x\n[rc]\n0\n[match]\n3\nAPROXY_END\n"
+        driver.run_command = MagicMock(return_value=MagicMock(exit_code=0, stdout=out, stderr=""))
+
+        result = driver.check_package_proxy("http://c:3142/acng-report.html", expect="Cache")
+
+        assert result == {"reachable": True, "expected": True, "error": ""}
+        assert "Proxy::c=DIRECT" in driver.run_command.call_args.args[0]
+
+    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):
+        _host(driver, proxy="http://c:3142")
+        assert driver.get_package_proxy() == "http://c:3142"
+
+    def test_no_proxy_set(self, driver):
+        _host(driver)
+        assert driver.get_package_proxy() is None
+
+    def test_where_apt_s_proxy_is_set(self, driver):
+        _host(
+            driver,
+            proxy="http://old:3128",
+            parts='/etc/apt/apt.conf.d/90curtin-aptproxy:1:Acquire::http::Proxy "http://old:3128";\n',
+        )
+        report = driver.get_package_proxy_report()
+        assert report["settings"][0]["file"] == "/etc/apt/apt.conf.d/90curtin-aptproxy"
+
+    def test_not_an_apt_host(self, driver):
+        driver._pkg_manager = "dnf"
+        assert driver.get_package_proxy() is None
+        assert driver.get_package_proxy_report() 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")
@@ -0,0 +1,88 @@
+"""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}") from 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
@@ -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 ")
@@ -0,0 +1,261 @@
+"""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()
+        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)