CHANGELOG.md

@@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.3.0] - 2026-10-17
+
+### Changed
+- **Breaking:** `run_device_action("fix_snmp")` is gone. The drivers implement
+  napalm-device-types 4.13's `configure_snmp_agent(settings)` template instead, and
+  the caller supplies every value: one `rocommunity` line per allowed source, contact
+  and location as given. No community, contact or location of an application is built
+  in any more. pf and ipfw rules are left to the operator. Requires
+  napalm-device-types 4.13.0 or later.
+- `get_snmp_config` no longer assumes the community `public` when the file names none.
+
+### Added
+- `get_management_peer()`: the address the host sees the SSH connection come from.
+- `ensure_package(name)` from the base package.
+
 ## [0.2.1] - 2026-10-16
 
 ### Fixed

napalm_bsd/base.py

@@ -8,6 +8,7 @@ stdin, as napalm-linux does it.
 
 from __future__ import annotations
 
+import ipaddress
 import logging
 import re
 import socket
@@ -16,7 +17,8 @@ from typing import Any, Optional
 
 import paramiko
 from napalm.base.exceptions import ConnectionClosedException, ConnectionException
-from napalm_device_types import OSDriver
+from napalm_device_types import EnsurePackageMixin, OSDriver, SnmpAgentMixin
+from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict
 from napalm_device_types.host_status import HostStatusMixin
 from napalm_device_types.channel import (
     ByteStream,
@@ -37,13 +39,10 @@ _SERVICE_ACTIONS = frozenset({"start", "stop", "restart", "enable", "disable"})
 _SERVICE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
 _PACKAGE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.+-]*$")
 
-#: The agent netOrk sets up, as on Linux: v2c, community "public", every address.
+#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
-_SNMPD_CONF = (
+_SNMP_WORD = re.compile(r'^[^\s"#]+$')
-    "agentAddress udp:161\n"
+#: The SSH client side of this connection: the address the host sees it come from.
-    "rocommunity public\n"
+_PEER_COMMAND = 'printf "%s\\n" "$SSH_CONNECTION"'
-    "sysLocation    Managed by netOrk\n"
-    "sysContact     netork@localhost\n"
-)
 #: Why ``run_command`` refuses stdin to a privileged command while a sudo
 #: password is set (#8).
 _PRIVILEGED_STDIN = (
@@ -63,11 +62,30 @@ _STAGE = (
     '{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
 )
 
-#: What a working agent answers sysDescr.0 with.
+def _source(value: str) -> str:
-_SNMP_TYPES = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
+    """An allowed source as an address or network, or ValueError."""
+    return str(ipaddress.ip_network(value.strip(), strict=False))
 
 
-class BsdDriver(HostStatusMixin, OSDriver):
+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 BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
     """Base for the BSD drivers; a concrete one names the commands that differ."""
 
     VENDOR = ""
@@ -451,43 +469,46 @@ class BsdDriver(HostStatusMixin, OSDriver):
     SNMPD_START = ""
     #: net-snmp's daemon; OpenBSD's base snmpd is /usr/sbin/snmpd.
     SNMPD_DAEMON = "/usr/local/sbin/snmpd"
-    SNMP_PROBE = "snmpget -v2c -cpublic -t2 -r0 -Ov 127.0.0.1 1.3.6.1.2.1.1.1.0"
 
-    def run_device_action(self, action: str) -> dict[str, Any]:
+    def get_management_peer(self) -> Optional[str]:
-        """Execute a named action on the device; ``fix_snmp`` is the one there is."""
+        """The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
-        if action == "fix_snmp":
+        out = self.run_command(_PEER_COMMAND, timeout=15).stdout.split()
-            return self._action_fix_snmp()
+        try:
-        raise NotImplementedError(f"Unknown action: {action!r}")
+            return str(ipaddress.ip_address(out[0])) if out else None
+        except ValueError:
+            return None
 
-    def _action_fix_snmp(self) -> dict[str, Any]:
+    # The steps of napalm-device-types' configure_snmp_agent: net-snmp rather than
-        """Install net-snmp, configure it as on Linux, start it, and ask it.
+    # the base daemons (FreeBSD bsnmpd, OpenBSD snmpd), as it answers UCD-SNMP-MIB.
 
-        net-snmp rather than the base daemons (FreeBSD bsnmpd, OpenBSD snmpd):
+    def _step(self, command: str, done: str) -> FixStepOutcomeDict:
-        it answers UCD-SNMP-MIB, which netOrk's health metrics read
+        result = self.run_command(command, privileged=True, timeout=600)
-        (NetOrk/netork#800). Stops at the first step that fails.
+        output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
-        """
+        return {"ok": result.exit_code == 0, "output": done if result.exit_code == 0 else output[-300:]}
-        lines: list[str] = []
+
-        steps = (
+    def _install_snmp_agent(self) -> FixStepOutcomeDict:
-            ("install", lambda: self.run_command(
+        return self._step(self.INSTALL_COMMAND.format(name="net-snmp"), "net-snmp installed")
-                self.INSTALL_COMMAND.format(name="net-snmp"), privileged=True, timeout=600)),
+
-            ("config", lambda: self._install_file_as_root(self.SNMPD_CONF, _SNMPD_CONF)),
+    def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
-            ("service", lambda: self.run_command(self.SNMPD_START, privileged=True, timeout=600)),
+        written = self._install_file_as_root(self.SNMPD_CONF, render_snmpd_conf(settings))
-        )
+        if written.exit_code:
-        for label, step in steps:
+            return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
-            result = step()
+        return {"ok": True, "output": f"wrote {self.SNMPD_CONF}"}
-            output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
+
-            lines.append(f"[{label}] {output[-300:]}".rstrip())
+    def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
-            if result.exit_code != 0:
+        # pf and ipfw rule sets are the operator's; the agent itself answers only
-                return {"success": False, "output": "\n".join(lines)}
+        # the allowed sources (one rocommunity line each).
-        probe = self.run_command(self.SNMP_PROBE, timeout=30).stdout.strip()
+        return {"ok": True, "output": "pf/ipfw unchanged; snmpd answers only the allowed sources"}
-        lines.append(f"[probe] {probe}")
+
-        return {"success": any(t in probe for t in _SNMP_TYPES), "output": "\n".join(lines)}
+    def _start_snmp_agent(self) -> FixStepOutcomeDict:
+        return self._step(self.SNMPD_START, "snmpd enabled and running")
 
     def get_snmp_config(self) -> Optional[dict[str, Any]]:
         """net-snmp's community and port, if its daemon runs; None otherwise."""
         if not self._out(f"pgrep -f {self.SNMPD_DAEMON}"):
             return None
-        community, port = "public", 161
+        # A community that is not in the file is not known; no default is assumed.
+        community, port = "", 161
         for line in self._out(f"cat {self.SNMPD_CONF}").splitlines():
             words = line.split()
             if len(words) >= 2 and words[0].lower() in (

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-bsd"
-version = "0.2.1"
+version = "0.3.0"
 description = "NAPALM drivers for FreeBSD and OpenBSD hosts via SSH"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -26,7 +26,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=4.7.1",  # channel.run_on_transport; OSDriver.get_os_users
+    "napalm-device-types>=4.13.0",  # fixes: SnmpAgentMixin, EnsurePackageMixin
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]
 

tests/test_drivers.py

@@ -462,10 +462,11 @@ class TestServices:
 
 class TestSnmp:
     """net-snmp from packages, as decided in NetOrk/netork#800: UCD-SNMP-MIB,
-    the same health metrics as on Linux."""
+    the same health metrics as on Linux. Set up through napalm-device-types'
+    ``configure_snmp_agent`` template; every value is the caller's (#12)."""
 
     STAGED = "/tmp/napalm-bsd-stage.Ab12Cd"
-    PROBE_OK = "STRING: FreeBSD host 15.1-RELEASE-p4 FreeBSD 15.1-RELEASE-p4 GENERIC amd64"
+    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "contact": "ops"}
 
     @pytest.mark.parametrize(
         ("cls", "install", "conf", "start"),
@@ -484,8 +485,10 @@ class TestSnmp:
             ),
         ],
     )
-    def test_fix_snmp_installs_configures_and_starts_net_snmp(self, cls, install, conf, start):
+    def test_installs_configures_and_starts_net_snmp_with_the_callers_values(
-        driver = _channel(cls, {base._STAGE: self.STAGED, BsdDriver.SNMP_PROBE: self.PROBE_OK})
+        self, cls, install, conf, start
+    ):
+        driver = _channel(cls, {base._STAGE: self.STAGED})
         stdin_seen = []
         run = driver.run_command
 
@@ -495,24 +498,25 @@ class TestSnmp:
             return run(command, privileged=privileged, timeout=timeout, stdin=stdin)
 
         driver.run_command = recording  # type: ignore[method-assign]
-        result = driver.run_device_action("fix_snmp")
+        result = driver.configure_snmp_agent(self.SETTINGS)
 
         assert result["success"] is True
+        assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
         commands = [c for c, _ in driver.calls]
         assert commands[0] == install
         installed = next(c for c, p in driver.calls if p and "install -m 644" in c)
         assert conf in installed and self.STAGED in installed
-        assert start in commands
+        assert commands[-1] == start
-        assert commands[-1] == BsdDriver.SNMP_PROBE
-        assert all(p for c, p in driver.calls if c not in (BsdDriver.SNMP_PROBE, base._STAGE))
         # The configuration reaches the host only through the unprivileged stage (#8).
         assert [(c, p) for c, p, _ in stdin_seen] == [(base._STAGE, False)]
         written = stdin_seen[0][2].decode()
-        assert "agentAddress udp:161" in written and "rocommunity public" in written
+        assert "rocommunity c0mm 10.7.224.20/32" in written
+        assert "sysContact    ops" in written
+        assert "public" not in written and "netork" not in written.lower()
 
     def test_a_failed_stage_runs_nothing_as_root(self):
         driver = _channel(FreeBSDDriver, {base._STAGE: ("mktemp: permission denied", 1)})
-        result = driver.run_device_action("fix_snmp")
+        result = driver.configure_snmp_agent(self.SETTINGS)
         assert result["success"] is False
         assert not any(p and "install -m 644" in c for c, p in driver.calls)
 
@@ -526,20 +530,35 @@ class TestSnmp:
                 )
             },
         )
-        result = driver.run_device_action("fix_snmp")
+        result = driver.configure_snmp_agent(self.SETTINGS)
         assert result["success"] is False
         assert "No packages available" in result["output"]
         assert len(driver.calls) == 1
 
-    def test_no_answer_from_the_agent_is_a_failure(self):
+    def test_the_host_firewall_is_left_to_the_operator(self):
-        driver = _channel(
+        result = _channel(FreeBSDDriver, {base._STAGE: self.STAGED}).configure_snmp_agent(
-            OpenBSDDriver, {BsdDriver.SNMP_PROBE: "Timeout: No Response from 127.0.0.1"}
+            self.SETTINGS
         )
-        assert driver.run_device_action("fix_snmp")["success"] is False
+        firewall = next(s for s in result["steps"] if s["step"] == "firewall")
+        assert firewall["ok"] is True and "pf" in firewall["output"]
 
-    def test_unknown_action(self):
+    @pytest.mark.parametrize("community", ["two words", 'q"uote', ""])
-        with pytest.raises(NotImplementedError):
+    def test_a_community_that_would_break_the_file_is_refused(self, community):
-            _channel(FreeBSDDriver, {}).run_device_action("fix_apt_proxy")
+        driver = _channel(FreeBSDDriver, {base._STAGE: self.STAGED})
+        result = driver.configure_snmp_agent({"community": community, "allowed_sources": []})
+        assert result["success"] is False
+        assert not any(p and "install -m 644" in c for c, p in driver.calls)
+
+    def test_run_device_action_is_gone(self):
+        assert not hasattr(BsdDriver, "run_device_action")
+        assert not hasattr(BsdDriver, "_action_fix_snmp")
+
+    def test_the_management_peer_is_the_ssh_client(self):
+        driver = _channel(FreeBSDDriver, {base._PEER_COMMAND: "10.7.224.20 51234 10.0.0.9 22"})
+        assert driver.get_management_peer() == "10.7.224.20"
+
+    def test_no_ssh_connection_variable_no_peer(self):
+        assert _channel(FreeBSDDriver, {}).get_management_peer() is None
 
     def test_snmp_config_of_a_running_agent(self):
         driver = _channel(
@@ -556,6 +575,16 @@ class TestSnmp:
             "version": "2c",
         }
 
+    def test_a_community_that_is_not_in_the_file_is_not_assumed(self):
+        driver = _channel(
+            FreeBSDDriver,
+            {
+                "pgrep -f /usr/local/sbin/snmpd": "1234",
+                "cat /usr/local/etc/snmp/snmpd.conf": "agentAddress udp:161\n",
+            },
+        )
+        assert driver.get_snmp_config()["community"] == ""
+
     def test_no_agent_running_means_no_config(self):
         driver = _channel(OpenBSDDriver, {"pgrep -f /usr/local/sbin/snmpd": ("", 1)})
         assert driver.get_snmp_config() is None