CHANGELOG.md

@@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### 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

napalm_linux/linux.py

@@ -18,11 +18,12 @@ A specific package manager can be forced with
 
 from __future__ import annotations
 
+import ipaddress
 import logging
 import re
 import socket
 from shlex import quote as _shlex_quote
-from typing import Any
+from typing import Any, Dict, List, Optional
 
 from netmiko import ConnectHandler
 from netmiko.exceptions import (
@@ -37,12 +38,15 @@ from napalm_device_types import (
     ByteStream,
     CommandResult,
     ContainerEngineMixin,
+    EnsurePackageMixin,
     FingerprintRule,
     HostStatusMixin,
     KernelFactsMixin,
     ListeningSocketsMixin,
     OpenSshServerMixin,
     OSDriver,
+    PackageProxyMixin,
+    SnmpAgentMixin,
     SystemIdentityMixin,
     SystemdServicesMixin,
     SystemdUnavailable,
@@ -50,17 +54,20 @@ from napalm_device_types import (
     parse_apt_upgradable,
     open_stream_on_transport,
     parse_dnf_security,
+    run_fix_steps,
     run_on_transport,
     strip_terminal_codes,
 )
 from napalm_device_types.models import (
     ApplyUpdatesResultDict,
     CronJobDict,
-    DeviceActionResultDict,
+    FixResultDict,
+    FixStepOutcomeDict,
     PackageDict,
     ProcessDict,
     ServiceDict,
     SNMPConfigDict,
+    SnmpAgentSettingsDict,
     UpdateDict,
     UserDict,
 )
@@ -171,9 +178,8 @@ _STAGE = (
 
 #: What to do when sudo wants a password netOrk does not have.
 _SUDO_PASSWORD_HINT = (
-    "sudo requires a password on this device but none is configured in netOrk. "
+    "sudo requires a password on this host and no sudo password was given: pass one, "
-    "Please add the sudo password to a Credential Profile assigned to this device, "
+    "or allow this user passwordless sudo (NOPASSWD)."
-    "or configure passwordless sudo (NOPASSWD) for this user."
 )
 
 # DMI field values that carry no useful information (OEM defaults, blanks)
@@ -242,7 +248,49 @@ def _arm_vendor_from_model(model: str) -> str:
     return " ".join(brand)
 
 
+#: Where net-snmp reads its configuration, and how each package manager installs it.
+_SNMPD_CONF = "/etc/snmp/snmpd.conf"
+_SNMP_INSTALL = {
+    "apt": "apt-get update -y >/dev/null 2>&1; DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
+    "dnf": "dnf install -y net-snmp net-snmp-utils 2>&1",
+    "yum": "yum install -y net-snmp net-snmp-utils 2>&1",
+    "apk": "apk add --no-cache net-snmp net-snmp-tools 2>&1",
+    "pacman": "pacman -Sy --noconfirm net-snmp 2>&1",
+}
+_APT_PROXY_CONF = "/etc/apt/apt.conf.d/00proxy"
+_PROXY_URL = re.compile(r"^https?://[A-Za-z0-9.\-\[\]:@%_~/]+$")
+_USER_NAME = re.compile(r"^[a-z_][a-z0-9_.\-]{0,31}$")
+#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
+_SNMP_WORD = re.compile(r"^[^\s\"#]+$")
+
+
+def _source(value: str) -> str:
+    """An allowed source as an address or network, or ValueError."""
+    return str(ipaddress.ip_network(value.strip(), strict=False))
+
+
+def render_snmpd_conf(settings: SnmpAgentSettingsDict) -> str:
+    """net-snmp's configuration for *settings*: read-only v2c, per allowed source."""
+    community = settings["community"]
+    if not _SNMP_WORD.match(community):
+        raise ValueError("the community must be one word without quotes or #")
+    listen = settings.get("listen")
+    lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"]
+    sources = [_source(s) for s in settings.get("allowed_sources", [])]
+    lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"]
+    for key, directive in (("location", "sysLocation"), ("contact", "sysContact")):
+        value = settings.get(key, "")
+        if value:
+            if "\n" in value:
+                raise ValueError(f"{key} must be one line")
+            lines.append(f"{directive}    {value}")
+    return "\n".join(lines) + "\n"
+
+
 class LinuxDriver(
+    SnmpAgentMixin,
+    EnsurePackageMixin,
+    PackageProxyMixin,
     KernelFactsMixin,
     ListeningSocketsMixin,
     VirtualIpsMixin,
@@ -1940,212 +1988,89 @@ class LinuxDriver(
             logger.debug("get_snmp_config() failed: %s", exc)
             return None
 
-    def run_device_action(self, action: str) -> DeviceActionResultDict:
+    # ------------------------------------------------------------------
-        """Execute a named action on the device."""
+    # Fix templates (napalm-device-types ``fixes``): the caller supplies every value
-        if action == "fix_docker_permissions":
+    # ------------------------------------------------------------------
-            return self._action_fix_docker_permissions()
-        if action == "fix_snmp":
-            return self._action_fix_snmp()
-        if action == "fix_apt_proxy":
-            return self._action_fix_apt_proxy()
-        if action == "apt_update_upgrade":
-            return self._action_apt_update_upgrade()
-        raise NotImplementedError(f"Unknown action: {action!r}")
 
-    def _action_apt_update_upgrade(self) -> DeviceActionResultDict:
+    def _root_outcome(self, command: str, timeout: float, done: str) -> FixStepOutcomeDict:
-        """Refresh the apt cache and fully upgrade all packages (apt-based systems only).
+        """Run *command* as root and report it as a step: its exit status decides."""
+        output, status = self._sudo_status(command, timeout)
+        if status != 0 and "password is required" in output:
+            output = f"{output}\n{_SUDO_PASSWORD_HINT}"
+        return {"ok": status == 0, "output": (output.strip()[-300:] if status != 0 else done)}
 
-        Uses full-upgrade (not plain upgrade) — plain "apt-get upgrade" refuses
+    def get_management_peer(self) -> str | None:
-        to install/remove packages even when required to satisfy a newer
+        """The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
-        version's dependencies, silently leaving those updates pending.
+        out = self.run_command('printf "%s\\n" "$SSH_CONNECTION"', timeout=15).stdout.split()
-        """
-        if self._pkg_manager != "apt":
-            return {
-                "success": True,
-                "output": f"Skipped — package manager is {self._pkg_manager!r}, not apt.",
-            }
-
-        sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__")
-        if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower():
-            if not self._sudo_password:
-                return {
-                    "success": False,
-                    "output": _SUDO_PASSWORD_HINT,
-                }
-
-        lines: list[str] = []
         try:
-            out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
+            return str(ipaddress.ip_address(out[0])) if out else None
-            lines.append(f"[update] {out.strip()[-300:]}")
+        except ValueError:
-            out = self._sudo(
+            return None
-                "DEBIAN_FRONTEND=noninteractive apt-get full-upgrade -y 2>&1", read_timeout=240
-            )
-            lines.append(f"[upgrade] {out.strip()[-300:]}")
-            return {"success": True, "output": "\n".join(lines)}
-        except Exception as exc:
-            lines.append(f"[error] {exc}")
-            return {"success": False, "output": "\n".join(lines)}
 
-    def _action_fix_snmp(self) -> DeviceActionResultDict:
+    def _install_snmp_agent(self) -> FixStepOutcomeDict:
-        """Install, configure and start snmpd with community 'public'."""
+        pm = self._pkg_manager or self._detect_pkg_manager()
-        lines: list[str] = []
+        command = _SNMP_INSTALL.get(pm or "")
+        if not command:
+            return {"ok": False, "output": f"no supported package manager ({pm or 'none found'})"}
+        return self._root_outcome(command, 300, f"net-snmp installed ({pm})")
 
-        # 0. Verify sudo access before attempting anything
+    def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
-        sudo_check = self._send("sudo -n true 2>&1 || echo __SUDO_NEEDS_PW__")
+        written = self._install_file_as_root(_SNMPD_CONF, render_snmpd_conf(settings))
-        if "__SUDO_NEEDS_PW__" in sudo_check or "password is required" in sudo_check.lower():
-            if not self._sudo_password:
-                return {
-                    "success": False,
-                    "output": _SUDO_PASSWORD_HINT,
-                }
-
-        # 1. Install snmpd if missing
-        pkg_mgr = self._detect_pkg_manager()
-        if not pkg_mgr:
-            return {"success": False, "output": "Package manager not detected — cannot install snmpd."}
-
-        # Refresh the package index first — a freshly provisioned (or simply
-        # long-untouched) system's cache can be stale/empty, which makes the
-        # install below fail outright rather than just being slow.
-        if pkg_mgr == "apt":
-            try:
-                update_out = self._sudo("apt-get update -y 2>&1", read_timeout=90)
-                lines.append(f"[update] {update_out.strip()[-200:]}")
-            except Exception as exc:
-                lines.append(f"[warn] apt-get update failed: {exc}")
-
-        # Install both snmpd (daemon) and snmp (client tools incl. snmpget for probing)
-        install_cmd: dict[str, str] = {
-            "apt":    "DEBIAN_FRONTEND=noninteractive apt-get install -y snmpd snmp 2>&1",
-            "dnf":    "dnf install -y net-snmp net-snmp-utils 2>&1",
-            "yum":    "yum install -y net-snmp net-snmp-utils 2>&1",
-            "apk":    "apk add --no-cache net-snmp net-snmp-tools 2>&1",
-            "pacman": "pacman -Sy --noconfirm net-snmp 2>&1",
-        }
-        cmd = install_cmd.get(pkg_mgr)
-        if cmd:
-            try:
-                out = self._sudo(cmd, read_timeout=120)
-                lines.append(f"[install] {out.strip()[-200:]}")
-            except Exception as exc:
-                lines.append(f"[error] install failed: {exc}")
-                return {"success": False, "output": "\n".join(lines)}
-
-        # 2. Determine the IP netOrk is connecting from by checking the established SSH connection
-        netork_ip = ""
-        try:
-            # ss shows the remote peer of the current SSH connection
-            raw = self._send(
-                "ss -tnp 2>/dev/null | awk '/sshd/{print $5}' | head -1 | cut -d: -f1"
-            ).strip()
-            if raw and raw not in ("", "0.0.0.0", "::", "127.0.0.1"):
-                netork_ip = raw
-        except Exception:
-            pass
-
-        # Write snmpd.conf: staged unprivileged, installed by root (#1039).
-        # agentAddress udp:161 overrides Debian's localhost-only default.
-        conf_str = (
-            "agentAddress udp:161\n"
-            "rocommunity public\n"
-            "sysLocation    Managed by netOrk\n"
-            "sysContact     netork@localhost\n"
-        )
-        try:
-            written = self._install_file_as_root("/etc/snmp/snmpd.conf", conf_str)
-            install_error = written.stdout.strip() if written.exit_code else ""
-        except Exception as exc:
-            install_error = str(exc)
-        verify = self._send("cat /etc/snmp/snmpd.conf 2>/dev/null").strip()
-        if "agentAddress" in verify and "rocommunity" in verify:
-            lines.append("[config] Wrote /etc/snmp/snmpd.conf — agentAddress udp:161, rocommunity public.")
-        else:
-            lines.append(f"[warn] snmpd.conf write may have failed: {(install_error or verify)[:100]}")
-
-        # 3. Open firewall for SNMP (UDP 161) — restrict to netOrk's source IP
-        if netork_ip:
-            try:
-                ufw = self._send("command -v ufw 2>/dev/null").strip()
-                ipt = self._send("command -v iptables 2>/dev/null").strip()
-                if ufw:
-                    # Expand to /24 so all containers in the same Docker network can probe
-                    parts = netork_ip.rsplit(".", 1)
-                    subnet = f"{parts[0]}.0/24" if len(parts) == 2 else netork_ip
-                    fw_out = self._sudo(
-                        f"ufw allow from {subnet} to any port 161 proto udp 2>&1", read_timeout=10
-                    )
-                    lines.append(f"[firewall/ufw] {fw_out.strip()[:200]}")
-                elif ipt:
-                    fw_out = self._sudo(
-                        f"iptables -C INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT 2>/dev/null"
-                        f" || iptables -I INPUT -s {netork_ip} -p udp --dport 161 -j ACCEPT",
-                        read_timeout=10,
-                    )
-                    lines.append(f"[firewall/iptables] rule added for {netork_ip}:161/udp")
-            except Exception as exc:
-                lines.append(f"[firewall] skipped — {exc}")
-
-        # 4. Restart snmpd.
-        #    - Redirect all output to /dev/null so netmiko's prompt detection is
-        #      never confused by service status messages.
-        #    - Append "; echo __OK__" so there is always a known token to wait for.
-        import time as _time
-        # Stop any running snmpd (systemctl-managed or apt-started orphan)
-        self._sudo("systemctl stop snmpd >/dev/null 2>&1; echo s1", read_timeout=15)
-        self._sudo("pkill -9 snmpd >/dev/null 2>&1; echo s2", read_timeout=10)
-        _time.sleep(2)
-        # Enable and start fresh
-        self._sudo("systemctl enable snmpd >/dev/null 2>&1; echo s3", read_timeout=15)
-        self._sudo("systemctl start snmpd >/dev/null 2>&1; echo s4", read_timeout=20)
-        _time.sleep(2)
-        lines.append("[service] snmpd restarted.")
-
-        # 5. Verify snmpd responds via local SNMP probe (sysDescr.0).
-        #    Success requires actual SNMP data types in the output, not just
-        #    the absence of error keywords.
-        _time.sleep(2)
-        probe_out = self._send(
-            "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0 2>&1 || true"
-        ).strip()
-        # snmpget returns lines like "STRING: Linux ..." or "Timeticks: (n) ..."
-        _snmp_types = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
-        success = any(t in probe_out for t in _snmp_types)
-        if success:
-            lines.append(f"[ok] SNMP probe successful — community 'public' is working.")
-        else:
-            lines.append(f"[warn] SNMP probe failed — output: {probe_out[:200]}")
-
-        return {"success": success, "output": "\n".join(lines)}
-
-    def _action_fix_docker_permissions(self) -> dict[str, Any]:
-        """Add the SSH user to the 'docker' group via sudo usermod."""
-        user = self._send("whoami 2>/dev/null || id -un").strip().splitlines()[-1].strip()
-        out = self._sudo(f"usermod -aG docker {user}")
-        low = out.lower()
-        success = not any(kw in low for kw in ("error", "invalid", "no such", "command not found"))
-        if not out.strip():
-            out = f"Added {user!r} to the docker group. Reconnect or run a new poll to verify."
-        return {"success": success, "output": out}
-
-    def _action_fix_apt_proxy(self) -> DeviceActionResultDict:
-        """Write /etc/apt/apt.conf.d/00proxy with the configured proxy URL."""
-        proxy_url = self._apt_proxy_url
-        if not proxy_url:
-            return {"success": False, "output": "No apt_proxy_url configured."}
-        if self._pkg_manager != "apt":
-            return {"success": False, "output": f"Package manager is {self._pkg_manager!r}, not apt — skipping."}
-
-        content = f'Acquire::http::Proxy "{proxy_url}";\n'
-        # Staged unprivileged, installed by root (#1039).
-        written = self._install_file_as_root("/etc/apt/apt.conf.d/00proxy", content)
-        verify = self._send("cat /etc/apt/apt.conf.d/00proxy 2>/dev/null").strip()
-        success = proxy_url in verify
-        if success:
-            return {"success": True, "output": f"Wrote /etc/apt/apt.conf.d/00proxy — proxy: {proxy_url}"}
-        output = f"Write may have failed. File content: {verify[:200]}"
         if written.exit_code:
-            output = f"{output}\n{written.stdout.strip()}"
+            return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
-        return {"success": False, "output": output}
+        return {"ok": True, "output": f"wrote {_SNMPD_CONF}"}
+
+    def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
+        sources = [_source(s) for s in settings.get("allowed_sources", [])]
+        if not sources:
+            return {"ok": True, "output": "no sources given; firewall unchanged"}
+        if self._send("command -v ufw 2>/dev/null"):
+            rules = [f"ufw allow from {s} to any port 161 proto udp" for s in sources]
+        elif self._send("command -v iptables 2>/dev/null"):
+            rules = [
+                f"{{ iptables -C INPUT -s {s} -p udp --dport 161 -j ACCEPT 2>/dev/null"
+                f" || iptables -I INPUT -s {s} -p udp --dport 161 -j ACCEPT; }}"
+                for s in sources
+            ]
+        else:
+            return {"ok": True, "output": "no host firewall (ufw, iptables); nothing to open"}
+        return self._root_outcome(" && ".join(rules), 30, f"udp/161 open for {', '.join(sources)}")
+
+    def _start_snmp_agent(self) -> FixStepOutcomeDict:
+        return self._root_outcome(
+            "systemctl enable snmpd && systemctl restart snmpd && systemctl is-active snmpd",
+            60,
+            "snmpd enabled and running",
+        )
+
+    def configure_package_proxy(self, url: str) -> FixResultDict:
+        """apt fetches through *url* (``/etc/apt/apt.conf.d/00proxy``); ``""`` removes it."""
+
+        def write() -> FixStepOutcomeDict:
+            if self._pkg_manager != "apt":
+                return {"ok": False, "output": f"package manager is {self._pkg_manager!r}, not apt"}
+            if not url:
+                return self._root_outcome(f"rm -f {_APT_PROXY_CONF}", 30, "proxy removed")
+            if not _PROXY_URL.match(url):
+                raise ValueError(f"not a proxy URL: {url!r}")
+            written = self._install_file_as_root(_APT_PROXY_CONF, f'Acquire::http::Proxy "{url}";\n')
+            if written.exit_code:
+                return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
+            return {"ok": True, "output": f"apt fetches through {url}"}
+
+        return run_fix_steps([("configure", write)])
+
+    def grant_container_engine_access(self, user: str) -> FixResultDict:
+        """Add *user* to the ``docker`` group, so the engine answers without sudo."""
+
+        def grant() -> FixStepOutcomeDict:
+            if not _USER_NAME.match(user):
+                raise ValueError(f"not a user name: {user!r}")
+            return self._root_outcome(
+                f"usermod -aG docker {user}", 30, f"{user} is in the docker group from the next login"
+            )
+
+        return run_fix_steps([("grant", grant)])
 
     def get_vpn_tunnels(self) -> dict[str, Any]:
         """Return WireGuard status via ``wg show all dump`` (requires root/sudo).

pyproject.toml

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

tests/test_fix_templates.py

@@ -0,0 +1,213 @@
+"""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

tests/test_linux.py

@@ -807,151 +807,10 @@ def test_get_facts_fallback_vendor_when_dmi_empty(driver):
     assert facts["vendor"] == "Linux"  # fallback to VENDOR class attribute
 
 
-# ---------------------------------------------------------------------------
-# _action_fix_snmp / _action_apt_update_upgrade
-# ---------------------------------------------------------------------------
-
-
-def _fix_snmp_send_side_effect(command: str, read_timeout: float = 100) -> str:
-    """Canned responses covering every _send() call _action_fix_snmp makes."""
-    if command.startswith("sudo -n true"):
-        return ""  # passwordless sudo works
-    if command.startswith("command -v apt"):
-        return "/usr/bin/apt"
-    if command.startswith("command -v"):
-        return ""  # ufw/iptables not found, other pkg managers not found
-    if command.startswith("ss -tnp"):
-        return ""  # no netork_ip detected — skips firewall step
-    if command.startswith("cat /etc/snmp/snmpd.conf"):
-        return "agentAddress udp:161\nrocommunity public\n"
-    if command.startswith("snmpget"):
-        return "STRING: Linux test"
-    return ""
-
-
-class TestActionFixSnmp:
-    def test_runs_apt_get_update_before_install(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-        sudo_calls: list[str] = []
-
-        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
-            sudo_calls.append(command)
-            return ""
-
-        with (
-            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
-            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
-        ):
-            driver._action_fix_snmp()
-
-        update_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get update" in c)
-        install_idx = next(i for i, c in enumerate(sudo_calls) if "apt-get install" in c)
-        assert update_idx < install_idx
-
-    def test_install_exception_returns_failure_instead_of_raising(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-
-        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
-            if "apt-get install" in command:
-                raise TimeoutError("connection timed out")
-            return ""
-
-        with (
-            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
-            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
-        ):
-            result = driver._action_fix_snmp()
-
-        assert result["success"] is False
-        assert "install failed" in result["output"]
-
-    def test_apt_get_update_failure_is_non_fatal(self, driver):
-        """apt-get update failing (e.g. transient network issue) must not
-        abort the whole action — install is still attempted."""
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-
-        def _sudo_side_effect(command: str, read_timeout: float = 100) -> str:
-            if "apt-get update" in command:
-                raise TimeoutError("network unreachable")
-            return ""
-
-        with (
-            patch.object(driver, "_send", side_effect=_fix_snmp_send_side_effect),
-            patch.object(driver, "_sudo", side_effect=_sudo_side_effect),
-        ):
-            result = driver._action_fix_snmp()
-
-        assert "apt-get update failed" in result["output"]
-
-
-class TestActionAptUpdateUpgrade:
-    def test_skips_when_not_apt(self, driver):
-        driver._pkg_manager = "dnf"
-        result = driver._action_apt_update_upgrade()
-        assert result["success"] is True
-        assert "Skipped" in result["output"]
-
-    def test_fails_when_sudo_needs_password_and_none_configured(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = None
-        with patch.object(driver, "_send", return_value="__SUDO_NEEDS_PW__"):
-            result = driver._action_apt_update_upgrade()
-        assert result["success"] is False
-        assert "sudo requires a password" in result["output"]
-
-    def test_runs_update_then_upgrade_successfully(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = "secret"  # noqa: S105
-        with (
-            patch.object(driver, "_send", return_value=""),
-            patch.object(
-                driver,
-                "_sudo",
-                side_effect=["Reading package lists... Done", "0 upgraded, 0 newly installed"],
-            ) as mock_sudo,
-        ):
-            result = driver._action_apt_update_upgrade()
-
-        assert result["success"] is True
-        assert "[update]" in result["output"]
-        assert "[upgrade]" in result["output"]
-        update_call, upgrade_call = mock_sudo.call_args_list
-        assert "apt-get update" in update_call.args[0]
-        # full-upgrade (not plain upgrade) — plain upgrade silently holds back
-        # packages whose newer version needs to install/remove dependencies.
-        assert "apt-get full-upgrade" in upgrade_call.args[0]
-
-    def test_exception_during_upgrade_returns_failure(self, driver):
-        driver._pkg_manager = "apt"
-        driver._sudo_password = "secret"  # noqa: S105
-        with (
-            patch.object(driver, "_send", return_value=""),
-            patch.object(
-                driver, "_sudo", side_effect=["update ok", TimeoutError("connection lost")]
-            ),
-        ):
-            result = driver._action_apt_update_upgrade()
-
-        assert result["success"] is False
-        assert "[error]" in result["output"]
-
-
-class TestRunDeviceActionDispatch:
-    def test_apt_update_upgrade_action_dispatches(self, driver):
-        with patch.object(
-            driver, "_action_apt_update_upgrade", return_value={"success": True, "output": ""}
-        ) as mock_action:
-            driver.run_device_action("apt_update_upgrade")
-        mock_action.assert_called_once()
-
-
 class TestNoDockerOfItsOwn:
     """Containers are read and handled by netOrk over the Engine API, through
     ``open_container_engine`` (NetOrk/netork#765). The old CLI methods are gone;
-    only the OS action that lets the login user use Docker stays."""
+    only the fix that lets the login user use Docker stays."""
 
     def test_the_cli_methods_are_gone(self):
         for name in ("get_docker_info", "get_docker_outdated", "reconstruct_docker_run", "_docker_bin"):
@@ -959,7 +818,7 @@ class TestNoDockerOfItsOwn:
 
     def test_the_engine_access_and_the_permission_fix_stay(self):
         assert hasattr(LinuxDriver, "open_container_engine")
-        assert hasattr(LinuxDriver, "_action_fix_docker_permissions")
+        assert callable(LinuxDriver.grant_container_engine_access)
 
 
 class TestUninstallPackage:

tests/test_staged_root_writes.py

@@ -138,10 +138,16 @@ def _every_line(driver: LinuxDriver, host: Host) -> list[str]:
     return typed + [run.line for run in host.runs]
 
 
+_SETTINGS = {"community": "c0mm", "allowed_sources": []}
+
 ACTIONS = [
-    ("_action_fix_snmp", "/etc/snmp/snmpd.conf", b"agentAddress udp:161"),
     (
-        "_action_fix_apt_proxy",
+        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";',
     ),
@@ -160,7 +166,7 @@ class TestStagedWrite:
         host = Host()
         driver = _driver(host)
 
-        getattr(driver, action)()
+        action(driver)
 
         for planted in _PLANTED:
             assert not any(planted in line for line in _every_line(driver, host))
@@ -171,7 +177,7 @@ class TestStagedWrite:
         host = Host()
         driver = _driver(host)
 
-        getattr(driver, action)()
+        action(driver)
 
         stage = next(run for run in host.unprivileged if "mktemp" in run.line)
         assert "umask 077" in stage.line
@@ -182,7 +188,7 @@ class TestStagedWrite:
         host = Host()
         driver = _driver(host)
 
-        getattr(driver, action)()
+        action(driver)
 
         install = next(script for script in host.root_scripts if "install -m" in script)
         part = f"{target}.netork-tmp"
@@ -194,7 +200,7 @@ class TestStagedWrite:
         host = Host()
         driver = _driver(host)
 
-        getattr(driver, action)()
+        action(driver)
 
         assert not any(re.search(r"\bch(own|mod)\b", script) for script in host.root_scripts)
 
@@ -202,7 +208,7 @@ class TestStagedWrite:
         host = Host(stage_output="mktemp: No space left on device\n", stage_status=1)
         driver = _driver(host)
 
-        getattr(driver, action)()
+        action(driver)
 
         assert not any("install -m" in script for script in host.root_scripts)
 
@@ -210,7 +216,7 @@ class TestStagedWrite:
         host = Host(stage_output="/etc/shadow")
         driver = _driver(host)
 
-        getattr(driver, action)()
+        action(driver)
 
         assert not any("/etc/shadow" in script for script in host.root_scripts)
 
@@ -223,7 +229,7 @@ def test_a_failed_install_is_reported():
     driver = _driver(host)
     driver._device.send_command.side_effect = lambda command, **_k: ""
 
-    result = driver._action_fix_apt_proxy()
+    result = driver.configure_package_proxy("http://proxy:3142")
 
     assert result["success"] is False
     assert "install: cannot create" in result["output"]