diff --git a/CHANGELOG.md b/CHANGELOG.md index 40f9182..a46fa2b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/napalm_bsd/base.py b/napalm_bsd/base.py index 5c0edd0..e0f4bd6 100644 --- a/napalm_bsd/base.py +++ b/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. -_SNMPD_CONF = ( - "agentAddress udp:161\n" - "rocommunity public\n" - "sysLocation Managed by netOrk\n" - "sysContact netork@localhost\n" -) +#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment. +_SNMP_WORD = re.compile(r'^[^\s"#]+$') +#: The SSH client side of this connection: the address the host sees it come from. +_PEER_COMMAND = 'printf "%s\\n" "$SSH_CONNECTION"' #: 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. -_SNMP_TYPES = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:") +def _source(value: str) -> str: + """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]: - """Execute a named action on the device; ``fix_snmp`` is the one there is.""" - if action == "fix_snmp": - return self._action_fix_snmp() - raise NotImplementedError(f"Unknown action: {action!r}") + def get_management_peer(self) -> Optional[str]: + """The address this host sees the SSH connection come from (``SSH_CONNECTION``).""" + out = self.run_command(_PEER_COMMAND, timeout=15).stdout.split() + try: + return str(ipaddress.ip_address(out[0])) if out else None + except ValueError: + return None - def _action_fix_snmp(self) -> dict[str, Any]: - """Install net-snmp, configure it as on Linux, start it, and ask it. + # The steps of napalm-device-types' configure_snmp_agent: net-snmp rather than + # the base daemons (FreeBSD bsnmpd, OpenBSD snmpd), as it answers UCD-SNMP-MIB. - net-snmp rather than the base daemons (FreeBSD bsnmpd, OpenBSD snmpd): - it answers UCD-SNMP-MIB, which netOrk's health metrics read - (NetOrk/netork#800). Stops at the first step that fails. - """ - lines: list[str] = [] - steps = ( - ("install", lambda: self.run_command( - self.INSTALL_COMMAND.format(name="net-snmp"), privileged=True, timeout=600)), - ("config", lambda: self._install_file_as_root(self.SNMPD_CONF, _SNMPD_CONF)), - ("service", lambda: self.run_command(self.SNMPD_START, privileged=True, timeout=600)), - ) - for label, step in steps: - result = step() - output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip()))) - lines.append(f"[{label}] {output[-300:]}".rstrip()) - if result.exit_code != 0: - return {"success": False, "output": "\n".join(lines)} - probe = self.run_command(self.SNMP_PROBE, timeout=30).stdout.strip() - lines.append(f"[probe] {probe}") - return {"success": any(t in probe for t in _SNMP_TYPES), "output": "\n".join(lines)} + def _step(self, command: str, done: str) -> FixStepOutcomeDict: + result = self.run_command(command, privileged=True, timeout=600) + 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:]} + + def _install_snmp_agent(self) -> FixStepOutcomeDict: + return self._step(self.INSTALL_COMMAND.format(name="net-snmp"), "net-snmp installed") + + def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: + written = self._install_file_as_root(self.SNMPD_CONF, render_snmpd_conf(settings)) + if written.exit_code: + return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]} + return {"ok": True, "output": f"wrote {self.SNMPD_CONF}"} + + def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict: + # pf and ipfw rule sets are the operator's; the agent itself answers only + # the allowed sources (one rocommunity line each). + return {"ok": True, "output": "pf/ipfw unchanged; snmpd answers only the allowed sources"} + + 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 ( diff --git a/pyproject.toml b/pyproject.toml index 6d60c8c..9a9f3d5 100644 --- a/pyproject.toml +++ b/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 ] diff --git a/tests/test_drivers.py b/tests/test_drivers.py index 1eb5615..ef5559b 100644 --- a/tests/test_drivers.py +++ b/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): - driver = _channel(cls, {base._STAGE: self.STAGED, BsdDriver.SNMP_PROBE: self.PROBE_OK}) + def test_installs_configures_and_starts_net_snmp_with_the_callers_values( + 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] == BsdDriver.SNMP_PROBE - assert all(p for c, p in driver.calls if c not in (BsdDriver.SNMP_PROBE, base._STAGE)) + assert commands[-1] == start # 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): - driver = _channel( - OpenBSDDriver, {BsdDriver.SNMP_PROBE: "Timeout: No Response from 127.0.0.1"} + def test_the_host_firewall_is_left_to_the_operator(self): + result = _channel(FreeBSDDriver, {base._STAGE: self.STAGED}).configure_snmp_agent( + 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): - with pytest.raises(NotImplementedError): - _channel(FreeBSDDriver, {}).run_device_action("fix_apt_proxy") + @pytest.mark.parametrize("community", ["two words", 'q"uote', ""]) + def test_a_community_that_would_break_the_file_is_refused(self, community): + 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