Author SHA1 Message Date
christianmanivong 2a92ec77e3 Merge pull request 'feat: say whether a host is still in its first boot' (#6) from feat/first-boot-pending into main
CI / test (3.10) (push) Successful in 38s
CI / test (3.11) (push) Successful in 27s
CI / test (3.12) (push) Successful in 30s
2026-10-08 11:01:34 +00:00
Christian Manivong 32a63411d3 feat: say whether a host is still in its first boot
CI / test (3.10) (push) Successful in 29s
CI / test (3.11) (push) Successful in 30s
CI / test (3.12) (push) Successful in 29s
CI / test (3.10) (pull_request) Successful in 29s
CI / test (3.11) (pull_request) Successful in 46s
CI / test (3.12) (pull_request) Successful in 42s
first_boot_pending() is true while /firstboot exists on FreeBSD. A FreeBSD
cloud image upgrades its base system on its first boot, starts sshd only
after that and restarts right away, and /etc/rc removes /firstboot just
before that restart. netOrk waits for this before it sets up a new VM
(NetOrk/netork#795).

OpenBSD has no such marker to ask yet, so it returns False there.
2026-10-08 12:58:00 +02:00
christianmanivong ae09e44345 Merge pull request 'fix: raise when an update reader cannot read, and report host status' (#5) from feat/host-status into main
CI / test (3.10) (push) Successful in 25s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 27s
2026-10-08 10:15:13 +00:00
Christian Manivong 25308bf747 fix: raise when an update reader cannot read, and report host status
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 25s
CI / test (3.12) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 25s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 26s
netOrk reads an empty update list as "no updates" and closes every patch
clock on the host. A refused sudo, a failing pkg, syspatch or freebsd-update,
or a pkg database pkg cannot read used to come back as that empty list. Each
of these now raises. Only pkg's "is not installed" still means no packages.

BsdDriver takes on napalm-device-types' HostStatusMixin. On FreeBSD it
reports a host whose installed kernel differs from the running one as
needing a reboot (device-types 4.1). OpenBSD stays unknown.

Closes #4
2026-10-08 12:08:44 +02:00
christianmanivong 1172b4d9d5 Merge pull request 'feat: set up SNMP with net-snmp' (#3) from feat/snmp into main
CI / test (3.10) (push) Successful in 27s
CI / test (3.11) (push) Successful in 26s
CI / test (3.12) (push) Successful in 28s
2026-10-08 08:08:46 +00:00
christianmanivong aef58ca161 feat: set up SNMP with net-snmp
CI / test (3.10) (pull_request) Successful in 30s
CI / test (3.11) (pull_request) Successful in 32s
CI / test (3.12) (pull_request) Successful in 36s
CI / test (3.10) (push) Successful in 45s
CI / test (3.11) (push) Successful in 30s
CI / test (3.12) (push) Successful in 29s
run_device_action("fix_snmp") and get_snmp_config for both drivers, as
decided in NetOrk/netork#800: net-snmp from packages rather than the base
daemons, because it answers UCD-SNMP-MIB, which netOrk's health metrics
read.

- fix_snmp installs net-snmp, writes the configuration netOrk uses on
  Linux (v2c, community "public", agentAddress udp:161) through stdin as
  root, enables and restarts the agent (FreeBSD `service snmpd`, OpenBSD
  `rcctl ... netsnmpd`, as `snmpd` is OpenBSD's own daemon) and asks it
  for sysDescr. It stops at the first step that fails and says why.
- get_snmp_config reports a running net-snmp's community and port.
- FreeBSD's install_package bootstraps pkg on a classic system that never
  had it instead of waiting for an answer that never comes.

Checked live: FreeBSD from a bare system, OpenBSD again over an existing
setup. Memory, swap and load answer on both; FreeBSD's ssCpuIdle about a
minute after start; OpenBSD's CPU figures from net-snmp are wrong (#800).
2026-10-08 10:05:09 +02:00
christianmanivong 7bf028ef4d Merge pull request 'feat: packages, services and updates' (#2) from feat/packages-services-updates into main
CI / test (3.10) (push) Successful in 28s
CI / test (3.11) (push) Successful in 26s
CI / test (3.12) (push) Successful in 28s
2026-10-08 08:02:25 +00:00
6 changed files with 392 additions and 25 deletions
+19 -9
View File
@@ -8,15 +8,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased] ## [Unreleased]
### Added ### Added
- Packages: `get_packages`, `install_package`, `uninstall_package`.
- Updates: `get_available_updates`, with VuXML's verdict as `security` on
FreeBSD and base-system patches as one `base-system` entry (OpenBSD
`syspatch`, classic FreeBSD `freebsd-update`).
- Services: `get_services` and `manage_service` (start, stop, restart,
enable, disable) through `service` and `rcctl`.
- `get_listening_sockets()` in the shape of `ListeningSocketsMixin`: FreeBSD
through `sockstat`, OpenBSD through `fstat` as root and `netstat -an`
without, which the reading reports as `attributed: False`.
- `FreeBSDDriver` (`freebsd`) and `OpenBSDDriver` (`openbsd`) on a shared - `FreeBSDDriver` (`freebsd`) and `OpenBSDDriver` (`openbsd`) on a shared
`BsdDriver`: SSH over exec channels (no PTY, real exit codes), root through `BsdDriver`: SSH over exec channels (no PTY, real exit codes), root through
`sudo -S` with the password on stdin. `sudo -S` with the password on stdin.
@@ -25,5 +16,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
their addresses (`ifconfig -a`), routes (`netstat -rn`), ARP (`arp -an`), their addresses (`ifconfig -a`), routes (`netstat -rn`), ARP (`arp -an`),
users and groups, processes (`ps … lstart`), cron jobs (system crontab and users and groups, processes (`ps … lstart`), cron jobs (system crontab and
the login user's). the login user's).
- `get_listening_sockets()` in the shape of `ListeningSocketsMixin`: FreeBSD
through `sockstat`, OpenBSD through `fstat` as root and `netstat -an`
without, which the reading reports as `attributed: False`.
- Packages: `get_packages`, `install_package`, `uninstall_package`. FreeBSD's
install bootstraps pkg on a classic system that never had it.
- Updates: `get_available_updates`, with VuXML's verdict as `security` on
FreeBSD and base-system patches as one `base-system` entry (OpenBSD
`syspatch`, classic FreeBSD `freebsd-update`). A reader that cannot read
raises instead of returning no updates (#4).
- Services: `get_services` and `manage_service` (start, stop, restart,
enable, disable) through `service` and `rcctl`.
- SNMP: `run_device_action("fix_snmp")` installs net-snmp, writes the same
configuration netOrk uses on Linux, starts the agent and asks it for
sysDescr; `get_snmp_config` reports a running agent's community and port.
- `get_host_status` (napalm-device-types' `HostStatusMixin`): on FreeBSD,
"reboot required" when the installed kernel differs from the running one.
- `first_boot_pending()`: whether a FreeBSD host is still in its first boot
(`/firstboot`), so netOrk can wait before setting up a new VM
(NetOrk/netork#795).
- Parsers tested on output recorded from FreeBSD 15.1 (hand-installed and - Parsers tested on output recorded from FreeBSD 15.1 (hand-installed and
cloud image) and OpenBSD 7.9 (NetOrk/netork#793). cloud image) and OpenBSD 7.9 (NetOrk/netork#793).
+23 -1
View File
@@ -36,6 +36,9 @@ over SSH, built on [napalm-device-types](https://git.netork.io/NAPALM/napalm-dev
| `get_packages`, `install_package`, `uninstall_package` | ✓ | ✓ | `pkg query` / `pkg install`, `pkg delete`; `pkg_info` / `pkg_add -I`, `pkg_delete` | | `get_packages`, `install_package`, `uninstall_package` | ✓ | ✓ | `pkg query` / `pkg install`, `pkg delete`; `pkg_info` / `pkg_add -I`, `pkg_delete` |
| `get_available_updates` | ✓ | ✓ | `pkg upgrade -n` + `pkg audit` (VuXML); `pkg_add -u -n -v` + `syspatch -c` | | `get_available_updates` | ✓ | ✓ | `pkg upgrade -n` + `pkg audit` (VuXML); `pkg_add -u -n -v` + `syspatch -c` |
| `get_services`, `manage_service` | ✓ | ✓ | `service -e` + `service … status` / `service … <action>`; `rcctl ls on` + `rcctl check` / `rcctl <action>` | | `get_services`, `manage_service` | ✓ | ✓ | `service -e` + `service … status` / `service … <action>`; `rcctl ls on` + `rcctl check` / `rcctl <action>` |
| `run_device_action("fix_snmp")`, `get_snmp_config` | ✓ | ✓ | net-snmp from packages: `/usr/local/etc/snmp/snmpd.conf` + `service snmpd`; `/etc/snmp/snmpd.conf` + `rcctl … netsnmpd` |
| `get_host_status` | ✓ | unknown | napalm-device-types' host status: on FreeBSD `freebsd-version -k` vs `-r` (4.1+) |
| `first_boot_pending` | ✓ | always `False` | whether `/firstboot` still exists |
`get_listening_sockets` has the shape of napalm-device-types' `get_listening_sockets` has the shape of napalm-device-types'
`ListeningSocketsMixin` (whose `ss`/cgroup reading is Linux's) and its rule: `ListeningSocketsMixin` (whose `ss`/cgroup reading is Linux's) and its rule:
@@ -57,11 +60,30 @@ reports what `freebsd-update` has fetched (`updatesready`). On FreeBSD with
pkgbase the base system is packages from the `FreeBSD-base` repository and pkgbase the base system is packages from the `FreeBSD-base` repository and
needs no extra entry. needs no extra entry.
A reader that cannot read **raises** rather than return an empty list: netOrk
takes `[]` as "no updates" and would close every patch clock on the host. That
covers a refused sudo, a failing `pkg`, `syspatch` or `freebsd-update`, and a pkg
database pkg cannot read (only pkg's "not installed" means no packages).
**Reboot.** `get_host_status` reports a FreeBSD host whose installed kernel
(`freebsd-version -k`) differs from the running one (`-r`) as needing a reboot.
OpenBSD has no such reading yet, so it stays unknown there.
**First boot.** A FreeBSD cloud image upgrades its base system on its first boot,
starts sshd only after that and restarts right away. `/etc/rc` removes `/firstboot`
just before the restart, so `first_boot_pending()` is true until then; netOrk waits
for it before it sets up a new VM. OpenBSD has no such marker to ask yet.
**Services** are the enabled ones. FreeBSD reads their status as root, as **Services** are the enabled ones. FreeBSD reads their status as root, as
root-only pidfiles hide a daemon from anyone else, and without root when sudo root-only pidfiles hide a daemon from anyone else, and without root when sudo
refuses; OpenBSD's `rcctl check` needs no root. Actions run as root. refuses; OpenBSD's `rcctl check` needs no root. Actions run as root.
SNMP (#800) follows. **SNMP** is net-snmp from packages (NetOrk/netork#800), configured as netOrk does
on Linux (v2c, community `public`, every address), so the agent answers
UCD-SNMP-MIB for netOrk's health metrics. Memory, swap and load are right on
both systems. CPU: FreeBSD reports `ssCpuIdle` about a minute after the
agent starts; on OpenBSD net-snmp's CPU figures are wrong (0 % idle on an idle
machine, also in `hrProcessorLoad`).
## Connection arguments ## Connection arguments
+115 -1
View File
@@ -17,6 +17,7 @@ from typing import Any, Optional
import paramiko import paramiko
from napalm.base.exceptions import ConnectionClosedException, ConnectionException from napalm.base.exceptions import ConnectionClosedException, ConnectionException
from napalm_device_types import OSDriver from napalm_device_types import OSDriver
from napalm_device_types.host_status import HostStatusMixin
from napalm_device_types.channel import ( from napalm_device_types.channel import (
ByteStream, ByteStream,
CommandResult, CommandResult,
@@ -36,8 +37,18 @@ _SERVICE_ACTIONS = frozenset({"start", "stop", "restart", "enable", "disable"})
_SERVICE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$") _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_.+-]*$") _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 a working agent answers sysDescr.0 with.
_SNMP_TYPES = ("STRING:", "INTEGER:", "OID:", "Timeticks:", "Hex-STRING:", "IpAddress:")
class BsdDriver(OSDriver):
class BsdDriver(HostStatusMixin, OSDriver):
"""Base for the BSD drivers; a concrete one names the commands that differ.""" """Base for the BSD drivers; a concrete one names the commands that differ."""
VENDOR = "" VENDOR = ""
@@ -153,11 +164,53 @@ class BsdDriver(OSDriver):
line, prefix = self._privileged(command, privileged) line, prefix = self._privileged(command, privileged)
return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix) return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix)
def _read(
self,
command: str,
*,
privileged: bool = False,
timeout: float = 60,
ok: tuple[int, ...] = (0,),
) -> CommandResult:
"""Run a reader's command; raise when it did not read.
A reader that cannot read raises rather than return "nothing": an empty
update list would close every patch clock netOrk keeps (napalm-bsd#4).
"""
result = self.run_command(command, privileged=privileged, timeout=timeout)
if result.exit_code not in ok:
reason = (result.stderr or result.stdout).strip() or f"exit {result.exit_code}"
raise RuntimeError(f"{command}: {reason}")
return result
def _out(self, command: str, *, privileged: bool = False, timeout: float = 60) -> str: def _out(self, command: str, *, privileged: bool = False, timeout: float = 60) -> str:
"""What *command* printed. A failing command prints nothing useful, """What *command* printed. A failing command prints nothing useful,
and the readers below treat empty output as "nothing there".""" and the readers below treat empty output as "nothing there"."""
return self.run_command(command, privileged=privileged, timeout=timeout).stdout.strip() return self.run_command(command, privileged=privileged, timeout=timeout).stdout.strip()
#: A file that exists until the host's first boot is over; "" when there is none to ask.
FIRST_BOOT_MARKER = ""
def first_boot_pending(self) -> bool:
"""Whether the host is still in its first boot.
A FreeBSD cloud image upgrades its base system on the first boot, starts
sshd only after that and restarts right away; ``/etc/rc`` removes
``/firstboot`` just before the restart. netOrk waits for this before it
sets up a new VM (NetOrk/netork#795).
"""
if not self.FIRST_BOOT_MARKER:
return False
return self.run_command(f"test -e {self.FIRST_BOOT_MARKER}").exit_code == 0
def _run_host_status_command(self, command: str) -> str:
"""The transport for ``HostStatusMixin.get_host_status``: read-only, no root.
On FreeBSD the report compares ``freebsd-version -k`` with ``-r``
(napalm-device-types 4.1); elsewhere "reboot required" stays unknown.
"""
return self.run_command(command).stdout
# -- facts ------------------------------------------------------------------- # -- facts -------------------------------------------------------------------
def _platform(self) -> dict[str, str]: def _platform(self) -> dict[str, str]:
@@ -344,6 +397,67 @@ class BsdDriver(OSDriver):
output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip()))) output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
return {"success": result.exit_code == 0, "output": output} return {"success": result.exit_code == 0, "output": output}
# -- SNMP (net-snmp from packages, NetOrk/netork#800) -------------------------
#: Where net-snmp reads its configuration, and how its daemon is started.
SNMPD_CONF = ""
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 _action_fix_snmp(self) -> dict[str, Any]:
"""Install net-snmp, configure it as on Linux, start it, and ask it.
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] = []
conf_dir = self.SNMPD_CONF.rsplit("/", 1)[0]
steps = (
("install", self.INSTALL_COMMAND.format(name="net-snmp"), None),
(
"config",
f"mkdir -p {conf_dir} && cat > {self.SNMPD_CONF} && chmod 644 {self.SNMPD_CONF}",
_SNMPD_CONF.encode(),
),
("service", self.SNMPD_START, None),
)
for label, command, stdin in steps:
result = self.run_command(command, privileged=True, timeout=600, stdin=stdin)
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 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
for line in self._out(f"cat {self.SNMPD_CONF}").splitlines():
words = line.split()
if len(words) >= 2 and words[0].lower() in (
"rocommunity",
"rwcommunity",
"rocommunity6",
):
community = words[1]
elif words and words[0] == "agentAddress":
found = re.search(r":(\d+)", line)
port = int(found[1]) if found else port
return {"running": True, "community": community, "port": port, "version": "2c"}
# -- accounts, processes, cron ----------------------------------------------- # -- accounts, processes, cron -----------------------------------------------
def get_users(self) -> list[dict[str, Any]]: def get_users(self) -> list[dict[str, Any]]:
+31 -9
View File
@@ -31,12 +31,15 @@ class FreeBSDDriver(BsdDriver):
ROUTES_COMMAND = "netstat -rnW" ROUTES_COMMAND = "netstat -rnW"
# sockstat names the process of every socket it can see; root sees them all. # sockstat names the process of every socket it can see; root sees them all.
LISTENING_COMMAND = "sockstat -46lq -P tcp,udp" LISTENING_COMMAND = "sockstat -46lq -P tcp,udp"
# /etc/rc removes it at the end of the first boot.
FIRST_BOOT_MARKER = "/firstboot"
def _parse_listening(self, output: str, *, attributed: bool) -> list[dict]: def _parse_listening(self, output: str, *, attributed: bool) -> list[dict]:
return parse.sockstat(output) return parse.sockstat(output)
PKG_QUERY = "pkg query '%n\t%v\t%R\t%sb\t%c'" PKG_QUERY = "pkg query '%n\t%v\t%R\t%sb\t%c'"
INSTALL_COMMAND = "pkg install -y {name}" # Bootstraps pkg on a classic system that never had it, instead of asking.
INSTALL_COMMAND = "env ASSUME_ALWAYS_YES=yes pkg install -y {name}"
UNINSTALL_COMMAND = "pkg delete -y {name}" UNINSTALL_COMMAND = "pkg delete -y {name}"
# Root reads every daemon's pidfile; one-shot scripts have no status. # Root reads every daemon's pidfile; one-shot scripts have no status.
SERVICE_STATUS_COMMAND = ( SERVICE_STATUS_COMMAND = (
@@ -44,13 +47,26 @@ class FreeBSDDriver(BsdDriver):
'printf "%s\\t%s\\n" "$n" "$(service $n status 2>&1 | head -1)"; done' 'printf "%s\\t%s\\n" "$n" "$(service $n status 2>&1 | head -1)"; done'
) )
SERVICE_ACTION_COMMAND = "service {name} {action}" SERVICE_ACTION_COMMAND = "service {name} {action}"
SNMPD_CONF = "/usr/local/etc/snmp/snmpd.conf"
# The rc script drops to the snmpd user, so the file stays readable (644).
SNMPD_START = "sysrc snmpd_enable=YES && service snmpd restart"
def _parse_services(self, output: str) -> list[dict]: def _parse_services(self, output: str) -> list[dict]:
return parse.service_status(output) return parse.service_status(output)
def _has_pkg(self) -> bool: def _has_pkg(self) -> bool:
"""pkg is bootstrapped; a hand-installed classic system may have only the stub.""" """pkg is bootstrapped; a hand-installed classic system may have only the stub.
return self.run_command("pkg -N").exit_code == 0
Only the stub's "not installed" means no pkg; any other failure (a
database pkg cannot read) raises rather than read as "no packages".
"""
result = self.run_command("pkg -N")
if result.exit_code == 0:
return True
message = (result.stderr or result.stdout).strip()
if "is not installed" in message:
return False
raise RuntimeError(f"pkg -N: {message or f'exit {result.exit_code}'}")
def get_packages(self) -> list[dict]: def get_packages(self) -> list[dict]:
return parse.pkg_query(self._out(self.PKG_QUERY)) if self._has_pkg() else [] return parse.pkg_query(self._out(self.PKG_QUERY)) if self._has_pkg() else []
@@ -65,10 +81,13 @@ class FreeBSDDriver(BsdDriver):
""" """
updates: list[dict] = [] updates: list[dict] = []
if self._has_pkg(): if self._has_pkg():
updates = parse.pkg_upgrades(self._out("pkg upgrade -n", privileged=True, timeout=300)) upgrade = self._read("pkg upgrade -n", privileged=True, timeout=300)
updates = parse.pkg_upgrades(upgrade.stdout)
audit = self.run_command("pkg audit -Fq", privileged=True, timeout=120) audit = self.run_command("pkg audit -Fq", privileged=True, timeout=120)
known = audit.exit_code in (0, 1) # 1: vulnerable packages found vulnerable = parse.pkg_audit(audit.stdout)
vulnerable = parse.pkg_audit(audit.stdout) if known else set() # 1 is "vulnerable packages found" and any error alike; only a list
# of packages makes it a verdict.
known = audit.exit_code == 0 or (audit.exit_code == 1 and bool(vulnerable))
for update in updates: for update in updates:
update["security"] = (update["name"] in vulnerable) if known else None update["security"] = (update["name"] in vulnerable) if known else None
if "FreeBSD-base" in self._out("pkg repos -l").split(): if "FreeBSD-base" in self._out("pkg repos -l").split():
@@ -82,10 +101,13 @@ class FreeBSDDriver(BsdDriver):
2 when there are none; it does not fetch, which ``freebsd-update cron`` 2 when there are none; it does not fetch, which ``freebsd-update cron``
does daily where it is enabled. does daily where it is enabled.
""" """
ready = self.run_command( ready = self._read(
"freebsd-update --not-running-from-cron updatesready", privileged=True, timeout=60 "freebsd-update --not-running-from-cron updatesready",
privileged=True,
timeout=60,
ok=(0, 2),
) )
if ready.exit_code != 0: if ready.exit_code == 2:
return [] return []
return [ return [
{ {
+5 -2
View File
@@ -42,6 +42,9 @@ class OpenBSDDriver(BsdDriver):
'printf "%s\\t%s\\n" "$s" "$r"; done' 'printf "%s\\t%s\\n" "$s" "$r"; done'
) )
SERVICE_ACTION_COMMAND = "rcctl {action} {name}" SERVICE_ACTION_COMMAND = "rcctl {action} {name}"
SNMPD_CONF = "/etc/snmp/snmpd.conf"
# net-snmp's rc script; "snmpd" is OpenBSD's own daemon.
SNMPD_START = "rcctl enable netsnmpd && rcctl restart netsnmpd"
def _parse_services(self, output: str) -> list[dict]: def _parse_services(self, output: str) -> list[dict]:
return parse.rcctl_check(output) return parse.rcctl_check(output)
@@ -61,10 +64,10 @@ class OpenBSDDriver(BsdDriver):
updates: list[dict] = [ updates: list[dict] = [
{**candidate, "origin": None, "security": None} {**candidate, "origin": None, "security": None}
for candidate in parse.pkg_add_candidates( for candidate in parse.pkg_add_candidates(
self._out("pkg_add -u -n -v", privileged=True, timeout=300) self._read("pkg_add -u -n -v", privileged=True, timeout=300).stdout
) )
] ]
patches = parse.syspatch(self._out("syspatch -c", privileged=True, timeout=120)) patches = parse.syspatch(self._read("syspatch -c", privileged=True, timeout=120).stdout)
if patches: if patches:
installed = parse.syspatch(self._out("syspatch -l", privileged=True)) installed = parse.syspatch(self._out("syspatch -l", privileged=True))
summary = ( summary = (
+199 -3
View File
@@ -267,6 +267,15 @@ class TestPackages:
assert driver.get_packages() == [] assert driver.get_packages() == []
assert driver.calls == [("pkg -N", False)] assert driver.calls == [("pkg -N", False)]
def test_a_pkg_that_cannot_read_its_database_raises(self):
"""Any other failure is "could not read", never "no packages"."""
driver = _channel(
FreeBSDDriver,
{"pkg -N": ("pkg: sqlite error ...: database disk image is malformed", 1)},
)
with pytest.raises(RuntimeError, match="malformed"):
driver.get_packages()
def test_openbsd_packages(self): def test_openbsd_packages(self):
driver = _channel(OpenBSDDriver, {"pkg_info": _read("openbsd-vm", "pkg_info_full.txt")}) driver = _channel(OpenBSDDriver, {"pkg_info": _read("openbsd-vm", "pkg_info_full.txt")})
assert any(p["name"] == "pcre2" for p in driver.get_packages()) assert any(p["name"] == "pcre2" for p in driver.get_packages())
@@ -274,7 +283,11 @@ class TestPackages:
@pytest.mark.parametrize( @pytest.mark.parametrize(
("cls", "install", "uninstall"), ("cls", "install", "uninstall"),
[ [
(FreeBSDDriver, "pkg install -y nginx", "pkg delete -y nginx"), (
FreeBSDDriver,
"env ASSUME_ALWAYS_YES=yes pkg install -y nginx",
"pkg delete -y nginx",
),
(OpenBSDDriver, "pkg_add -I nginx", "pkg_delete nginx"), (OpenBSDDriver, "pkg_add -I nginx", "pkg_delete nginx"),
], ],
) )
@@ -287,7 +300,7 @@ class TestPackages:
def test_a_version_is_part_of_the_name(self): def test_a_version_is_part_of_the_name(self):
driver = _channel(FreeBSDDriver, {}) driver = _channel(FreeBSDDriver, {})
driver.install_package("nginx", "1.28.0") driver.install_package("nginx", "1.28.0")
assert driver.calls == [("pkg install -y nginx-1.28.0", True)] assert driver.calls == [("env ASSUME_ALWAYS_YES=yes pkg install -y nginx-1.28.0", True)]
def test_a_failed_install_raises(self): def test_a_failed_install_raises(self):
driver = _channel(OpenBSDDriver, {("pkg_add -I nope", True): ("Can't find nope", 1)}) driver = _channel(OpenBSDDriver, {("pkg_add -I nope", True): ("Can't find nope", 1)})
@@ -355,7 +368,7 @@ class TestAvailableUpdates:
driver = _channel( driver = _channel(
FreeBSDDriver, FreeBSDDriver,
{ {
"pkg -N": ("", 1), "pkg -N": ("pkg: pkg is not installed", 1),
("freebsd-update --not-running-from-cron updatesready", True): ("", 2), ("freebsd-update --not-running-from-cron updatesready", True): ("", 2),
}, },
) )
@@ -437,3 +450,186 @@ class TestServices:
with pytest.raises(ValueError): with pytest.raises(ValueError):
driver.manage_service(name, action) driver.manage_service(name, action)
assert driver.calls == [] assert driver.calls == []
class TestSnmp:
"""net-snmp from packages, as decided in NetOrk/netork#800: UCD-SNMP-MIB,
the same health metrics as on Linux."""
PROBE_OK = "STRING: FreeBSD host 15.1-RELEASE-p4 FreeBSD 15.1-RELEASE-p4 GENERIC amd64"
@pytest.mark.parametrize(
("cls", "install", "conf", "start"),
[
(
FreeBSDDriver,
"env ASSUME_ALWAYS_YES=yes pkg install -y net-snmp",
"/usr/local/etc/snmp/snmpd.conf",
"sysrc snmpd_enable=YES && service snmpd restart",
),
(
OpenBSDDriver,
"pkg_add -I net-snmp",
"/etc/snmp/snmpd.conf",
"rcctl enable netsnmpd && rcctl restart netsnmpd",
),
],
)
def test_fix_snmp_installs_configures_and_starts_net_snmp(self, cls, install, conf, start):
driver = _channel(cls, {BsdDriver.SNMP_PROBE: self.PROBE_OK})
stdin_seen = {}
run = driver.run_command
def recording(command, *, privileged=False, timeout=60, stdin=None):
if stdin is not None:
stdin_seen[command] = stdin
return run(command, privileged=privileged, timeout=timeout, stdin=stdin)
driver.run_command = recording # type: ignore[method-assign]
result = driver.run_device_action("fix_snmp")
assert result["success"] is True
commands = [c for c, _ in driver.calls]
assert commands[0] == install
assert any(conf in c and "cat >" in c for c in commands)
assert start in commands
assert commands[-1] == BsdDriver.SNMP_PROBE
assert all(p for c, p in driver.calls if c != BsdDriver.SNMP_PROBE) # root for the setup
written = next(v for k, v in stdin_seen.items() if conf in k).decode()
assert "agentAddress udp:161" in written and "rocommunity public" in written
def test_a_failed_install_stops_and_says_so(self):
driver = _channel(
FreeBSDDriver,
{
("env ASSUME_ALWAYS_YES=yes pkg install -y net-snmp", True): (
"pkg: No packages available",
1,
)
},
)
result = driver.run_device_action("fix_snmp")
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"}
)
assert driver.run_device_action("fix_snmp")["success"] is False
def test_unknown_action(self):
with pytest.raises(NotImplementedError):
_channel(FreeBSDDriver, {}).run_device_action("fix_apt_proxy")
def test_snmp_config_of_a_running_agent(self):
driver = _channel(
FreeBSDDriver,
{
"pgrep -f /usr/local/sbin/snmpd": "1234",
"cat /usr/local/etc/snmp/snmpd.conf": "agentAddress udp:1161\nrocommunity s3cret\n",
},
)
assert driver.get_snmp_config() == {
"running": True,
"community": "s3cret",
"port": 1161,
"version": "2c",
}
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
class TestHostStatus:
"""napalm-device-types' host status, carried over the exec channel; on
FreeBSD it compares the installed with the running kernel (4.1.0)."""
REPORT = (
"HSTAT_BEGIN\n[kernel]\n15.1-RELEASE\n[modules]\n"
"[freebsd-kernel]\n15.1-RELEASE-p5\n[freebsd-running]\n15.1-RELEASE-p4\n"
"[timers]\nHSTAT_END\n"
)
def test_a_patched_kernel_waits_for_a_reboot(self):
from napalm_device_types.host_status import HOST_STATUS_COMMAND
driver = _channel(FreeBSDDriver, {HOST_STATUS_COMMAND: self.REPORT})
status = driver.get_host_status()
assert status["reboot_required"] is True
assert driver.calls == [(HOST_STATUS_COMMAND, False)] # read-only, no root
@pytest.mark.parametrize("cls", [FreeBSDDriver, OpenBSDDriver])
def test_both_have_it(self, cls):
assert callable(getattr(cls, "get_host_status", None))
class TestAFailedReadRaises:
"""A reader that could not read raises; [] would tell netOrk "no updates"
and close every patch clock on the host (napalm-bsd#4)."""
def test_freebsd_without_root(self):
driver = _channel(
FreeBSDDriver,
{"pkg -N": "", ("pkg upgrade -n", True): ("", 1)},
)
with pytest.raises(RuntimeError, match="pkg upgrade -n"):
driver.get_available_updates()
def test_a_failed_audit_leaves_security_unknown(self):
"""pkg audit exits 1 for "vulnerable packages found" and for errors alike;
only a list of packages makes the 1 a verdict."""
driver = _channel(
FreeBSDDriver,
{
"pkg -N": "",
("pkg upgrade -n", True): _read("freebsd-vm", "sudo_pkg_upgrade_n_latest.txt"),
("pkg audit -Fq", True): ("", 1),
"pkg repos -l": "FreeBSD-ports\nFreeBSD-base\n",
},
)
assert {u["security"] for u in driver.get_available_updates()} == {None}
def test_classic_freebsd_update_error(self):
driver = _channel(
FreeBSDDriver,
{
"pkg -N": ("pkg: pkg is not installed", 1),
("freebsd-update --not-running-from-cron updatesready", True): ("", 1),
},
)
with pytest.raises(RuntimeError, match="freebsd-update"):
driver.get_available_updates()
@pytest.mark.parametrize("failing", ["pkg_add -u -n -v", "syspatch -c"])
def test_openbsd_without_root(self, failing):
answers = {("pkg_add -u -n -v", True): "", ("syspatch -c", True): ""}
answers[(failing, True)] = ("", 1)
driver = _channel(OpenBSDDriver, answers)
with pytest.raises(RuntimeError, match=failing.split()[0]):
driver.get_available_updates()
class TestFirstBoot:
"""Whether the host is still in its first boot (NetOrk/netork#795).
A FreeBSD cloud image upgrades its base system on the first boot, starts
sshd only then and restarts right after; /etc/rc removes /firstboot just
before that restart. netOrk waits for it before setting up a new VM.
"""
def test_freebsd_in_its_first_boot(self):
driver = _channel(FreeBSDDriver, {"test -e /firstboot": ("", 0)})
assert driver.first_boot_pending() is True
assert driver.calls == [("test -e /firstboot", False)]
def test_freebsd_after_its_first_boot(self):
driver = _channel(FreeBSDDriver, {"test -e /firstboot": ("", 1)})
assert driver.first_boot_pending() is False
def test_openbsd_has_no_marker_to_ask(self):
driver = _channel(OpenBSDDriver, {})
assert driver.first_boot_pending() is False
assert driver.calls == []