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

- CI moves to .gitea/workflows/ci.yml and calls the shared driver workflow
  of NAPALM/driver-standard; .github/ is gone.
- napalm-device-types >=4.21.1 from PyPI (the README no longer installs
  it from git), py.typed, SPDX license string, setuptools>=77, version
  floors on the dev extra, Issues and Changelog links on git.netork.io.
- mypy strict for the package; napalm_bsd.base, napalm_bsd.freebsd and
  napalm_bsd.openbsd are exempt for now (28 errors, mostly getters typed
  as plain dicts rather than the device-types TypedDicts), recorded in the
  quality baseline with three complex parsers. parse.py is strict after a
  one-line annotation.
- Coverage floor 94 % (measured 94.63 %).

.gitea/workflows/ci.yml

@@ -0,0 +1,15 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+    tags: ["v*"]
+  pull_request:
+
+jobs:
+  driver:
+    uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
+    with:
+      standard-ref: main
+    secrets:
+      PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}

.github/workflows/ci.yml

@@ -1,48 +0,0 @@
-name: CI
-
-on:
-  push:
-    branches: ["**"]
-  pull_request:
-    branches: ["**"]
-
-jobs:
-  test:
-    runs-on: ubuntu-latest
-    strategy:
-      fail-fast: false
-      matrix:
-        python-version: ["3.10", "3.11", "3.12"]
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v4
-
-      - name: Setup Python
-        uses: actions/setup-python@v5
-        with:
-          python-version: ${{ matrix.python-version }}
-          cache: pip
-
-      - name: Install package with dev extras
-        run: |
-          python -m pip install --upgrade pip
-          # napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
-          # pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
-          python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
-          python -m pip install -e ".[dev]"
-
-      - name: Run unit tests
-        run: |
-          python -m pytest -q --tb=short
-
-      - name: Build wheel and sdist
-        run: |
-          python -m pip install build
-          python -m build
-
-      - name: Upload dist artifacts
-        # v4 refuses to run on Gitea ("not currently supported on GHES").
-        uses: actions/upload-artifact@v3
-        with:
-          name: dist-${{ matrix.python-version }}
-          path: dist/*

CHANGELOG.md

@@ -7,23 +7,57 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
-### Added
-- 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.
+## [0.5.0] - 2026-10-23
+
+The first release on PyPI.
 
 ### Changed
-- FreeBSD's `install_package` bootstraps pkg on a classic system that never
-  had it (`ASSUME_ALWAYS_YES`) instead of waiting for an answer.
-- 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`.
+- The drivers follow the NAPALM driver standard
+  (https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
+  tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
+  a build check, and a release tag publishes to PyPI. The package now ships `py.typed`.
+  Python 3.10 stays the floor. napalm-device-types now comes from PyPI, 4.21.1 or later.
+
+### Security
+- The SSH host key is verified (NetOrk/netork#1167). `open()` takes the key to expect from `optional_args["ssh_host_key"]` and refuses any other one before it logs in (`SshHostKeyMismatch`); without one it accepts the first key. Every key it sees goes to `optional_args["ssh_host_key_observer"]`. Replaces paramiko's `AutoAddPolicy`. Requires napalm-device-types 4.19.0; version 0.4.0.
+
+## [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
+- A sudo password no longer ends up in `snmpd.conf`. `run_command` sent the
+  password and the data on the same stdin, and a sudo that does not prompt
+  (NOPASSWD, a cached timestamp) left the password for `cat`, which wrote it
+  as the first line of a file then made world-readable (#8). `run_command` now
+  refuses stdin to a privileged command while a sudo password is set, and
+  `fix_snmp` stages the configuration without privilege before root installs
+  it, without following a symlink at the target.
+
+## [0.2.0] - 2026-10-13
+
+### Changed
+- `get_users` is now `get_os_users`, as napalm-device-types 4.7 names it on
+  `OSDriver`; the return shape is unchanged. NAPALM's own `get_users` has
+  another shape and a stub on every driver, so a poll asking `hasattr` could
+  not tell the two apart. Requires napalm-device-types 4.7.1 or later.
+
+## [0.1.0]
+
+### Added
 - `FreeBSDDriver` (`freebsd`) and `OpenBSDDriver` (`openbsd`) on a shared
   `BsdDriver`: SSH over exec channels (no PTY, real exit codes), root through
   `sudo -S` with the password on stdin.
@@ -32,5 +66,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`),
   users and groups, processes (`ps … lstart`), cron jobs (system crontab and
   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
   cloud image) and OpenBSD 7.9 (NetOrk/netork#793).

README.md

@@ -31,12 +31,14 @@ over SSH, built on [napalm-device-types](https://git.netork.io/NAPALM/napalm-dev
 | `get_route_to` | ✓ | ✓ | `netstat -rnW` / `netstat -rn` |
 | `get_arp_table` | ✓ | ✓ | `arp -an` |
 | `get_lldp_neighbors` | `{}` | `{}` | no LLDP daemon in either base system |
-| `get_users`, `get_processes`, `get_cron_jobs` | ✓ | ✓ | `/etc/passwd`+`/etc/group`, `ps … lstart`, system crontab + `crontab -l` |
+| `get_os_users`, `get_processes`, `get_cron_jobs` | ✓ | ✓ | `/etc/passwd`+`/etc/group`, `ps … lstart`, system crontab + `crontab -l` |
 | `get_listening_sockets` | ✓ | ✓ | `sockstat -46lq` / `fstat -n` as root, `netstat -an` without |
 | `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_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'
 `ListeningSocketsMixin` (whose `ss`/cgroup reading is Linux's) and its rule:
@@ -58,6 +60,20 @@ reports what `freebsd-update` has fetched (`updatesready`). On FreeBSD with
 pkgbase the base system is packages from the `FreeBSD-base` repository and
 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
 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.
@@ -77,7 +93,6 @@ machine, also in `hrProcessorLoad`).
 ## Development
 
 ```bash
-pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
 pip install -e ".[dev]"
 pytest
 ```

napalm_bsd/base.py

@@ -8,21 +8,24 @@ stdin, as napalm-linux does it.
 
 from __future__ import annotations
 
+import ipaddress
 import logging
 import re
-import socket
 from shlex import quote
-from typing import Any, Optional
+from typing import Any
 
 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.channel import (
     ByteStream,
     CommandResult,
     open_stream_on_transport,
     run_on_transport,
 )
+from napalm_device_types.host_status import HostStatusMixin
+from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict
+from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
 
 from napalm_bsd import parse
 
@@ -36,18 +39,55 @@ _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 = (
+    "run_command(privileged=True) takes no stdin while a sudo password is set: "
+    "sudo -S reads the password from stdin, and a sudo that does not prompt "
+    "(NOPASSWD, a cached timestamp) leaves it for the command, which would read it "
+    "as the first line of the data. Stage the data without privilege and have root "
+    "install that file."
+)
+
+#: Stages content without privilege: a fresh ``mktemp`` file only the login user
+#: may read, never a fixed name another user could create first. Prints its path.
+#: The path is on the device, not a local temp file.
+_STAGE_TEMPLATE = "/tmp/napalm-bsd-stage.XXXXXX"  # nosec B108
+_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
+_STAGE = (
+    f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
+    '{ 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:")
 
 
-class BsdDriver(OSDriver):
+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 BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
     """Base for the BSD drivers; a concrete one names the commands that differ."""
 
     VENDOR = ""
@@ -74,7 +114,7 @@ class BsdDriver(OSDriver):
         username: str,
         password: str,
         timeout: int = 60,
-        optional_args: Optional[dict] = None,
+        optional_args: dict | None = None,
     ) -> None:
         self.hostname = hostname
         self.username = username
@@ -82,22 +122,22 @@ class BsdDriver(OSDriver):
         self.timeout = timeout
         args = optional_args or {}
         self.port: int = int(args.get("port", 22))
-        self._key_file: Optional[str] = args.get("key_file")
-        self._sudo_password: Optional[str] = args.get("sudo_password")
+        self._key_file: str | None = args.get("key_file")
+        self._sudo_password: str | None = args.get("sudo_password")
         self._allow_agent: bool = bool(args.get("allow_agent", False))
         self._look_for_keys: bool = bool(args.get("look_for_keys", False))
-        self._client: Optional[paramiko.SSHClient] = None
-        self._root: Optional[bool] = None
+        # The host key to expect, and where the one seen is reported (netork#1167).
+        self._host_key = HostKeyCheck(args, hostname, self.port)
+        self._client: paramiko.SSHClient | None = None
+        self._root: bool | None = None
 
     # -- connection --------------------------------------------------------------
 
     def open(self) -> None:
         client = paramiko.SSHClient()
-        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
         try:
-            client.connect(
-                self.hostname,
-                port=self.port,
+            self._host_key.connect(
+                client,
                 username=self.username,
                 password=self.password or None,
                 key_filename=self._key_file,
@@ -105,7 +145,10 @@ class BsdDriver(OSDriver):
                 look_for_keys=self._look_for_keys,
                 timeout=self.timeout,
             )
-        except (paramiko.SSHException, socket.error) as exc:
+        except SshHostKeyMismatch:
+            client.close()  # refused before any login; nothing was sent
+            raise
+        except (OSError, paramiko.SSHException) as exc:
             client.close()
             raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc
         self._client = client
@@ -133,7 +176,7 @@ class BsdDriver(OSDriver):
             self._root = run_on_transport(self._transport(), "id -u").stdout.strip() == "0"
         return self._root
 
-    def _privileged(self, command: str, privileged: bool) -> tuple[str, Optional[bytes]]:
+    def _privileged(self, command: str, privileged: bool) -> tuple[str, bytes | None]:
         """The command line that runs *command* as asked, and what goes to stdin first.
 
         With a sudo password, ``sudo -S`` reads it from stdin, so it never
@@ -152,22 +195,96 @@ class BsdDriver(OSDriver):
         *,
         privileged: bool = False,
         timeout: float = 60,
-        stdin: Optional[bytes] = None,
+        stdin: bytes | None = None,
     ) -> CommandResult:
-        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code."""
+        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code.
+
+        :raises ValueError: for *stdin* to a privileged command while a sudo
+            password is set (see ``_PRIVILEGED_STDIN``).
+        """
         line, prefix = self._privileged(command, privileged)
-        data = (prefix or b"") + (stdin or b"") if (prefix or stdin) else None
-        return run_on_transport(self._transport(), line, stdin=data, timeout=timeout)
+        if prefix and stdin is not None:
+            raise ValueError(_PRIVILEGED_STDIN)
+        return run_on_transport(self._transport(), line, stdin=prefix or stdin, timeout=timeout)
+
+    def _install_file_as_root(self, path: str, content: str) -> CommandResult:
+        """Replace *path* with *content*, as a new file owned by root, mode 644.
+
+        The content is staged without privilege (``_STAGE``), so it never follows
+        a sudo password on stdin. Root then installs the staged file next to *path*
+        and renames it over *path*; nothing root runs follows a symlink at *path*.
+        Nothing runs as root when staging failed.
+        """
+        staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
+        lines = staged.stdout.strip().splitlines()
+        tmp = lines[-1] if lines else ""
+        if (
+            staged.exit_code != 0
+            or not tmp.startswith(_STAGE_PREFIX)
+            or "/" in tmp[len(_STAGE_PREFIX) :]
+        ):
+            reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
+            return CommandResult(reason, "", staged.exit_code or 1)
+        tmp_q, part_q, path_q = (quote(p) for p in (tmp, f"{path}.napalm-tmp", path))
+        directory = quote(path.rsplit("/", 1)[0] or "/")
+        return self.run_command(
+            f"mkdir -p {directory} && install -m 644 -o root -g wheel {tmp_q} {part_q}"
+            f" && mv -f {part_q} {path_q}; s=$?; rm -f {tmp_q} {part_q}; exit $s",
+            privileged=True,
+            timeout=30,
+        )
 
     def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
         line, prefix = self._privileged(command, privileged)
         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:
         """What *command* printed. A failing command prints nothing useful,
         and the readers below treat empty output as "nothing there"."""
         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 -------------------------------------------------------------------
 
     def _platform(self) -> dict[str, str]:
@@ -361,47 +478,49 @@ class BsdDriver(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) -> str | None:
+        """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] = []
-        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)
+    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())))
-            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)}
+        return {
+            "ok": result.exit_code == 0,
+            "output": done if result.exit_code == 0 else output[-300:],
+        }
 
-    def get_snmp_config(self) -> Optional[dict[str, Any]]:
+    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) -> dict[str, Any] | None:
         """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 (
@@ -417,7 +536,7 @@ class BsdDriver(OSDriver):
 
     # -- accounts, processes, cron -----------------------------------------------
 
-    def get_users(self) -> list[dict[str, Any]]:
+    def get_os_users(self) -> list[dict[str, Any]]:
         return parse.users(self._out("cat /etc/passwd"), self._out("cat /etc/group"))
 
     def get_processes(self) -> list[dict[str, Any]]:

napalm_bsd/freebsd.py

@@ -31,6 +31,8 @@ class FreeBSDDriver(BsdDriver):
     ROUTES_COMMAND = "netstat -rnW"
     # sockstat names the process of every socket it can see; root sees them all.
     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]:
         return parse.sockstat(output)
@@ -53,8 +55,18 @@ class FreeBSDDriver(BsdDriver):
         return parse.service_status(output)
 
     def _has_pkg(self) -> bool:
-        """pkg is bootstrapped; a hand-installed classic system may have only the stub."""
-        return self.run_command("pkg -N").exit_code == 0
+        """pkg is bootstrapped; a hand-installed classic system may have only the stub.
+
+        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]:
         return parse.pkg_query(self._out(self.PKG_QUERY)) if self._has_pkg() else []
@@ -69,10 +81,13 @@ class FreeBSDDriver(BsdDriver):
         """
         updates: list[dict] = []
         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)
-            known = audit.exit_code in (0, 1)  # 1: vulnerable packages found
-            vulnerable = parse.pkg_audit(audit.stdout) if known else set()
+            vulnerable = parse.pkg_audit(audit.stdout)
+            # 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:
                 update["security"] = (update["name"] in vulnerable) if known else None
             if "FreeBSD-base" in self._out("pkg repos -l").split():
@@ -86,10 +101,13 @@ class FreeBSDDriver(BsdDriver):
         2 when there are none; it does not fetch, which ``freebsd-update cron``
         does daily where it is enabled.
         """
-        ready = self.run_command(
-            "freebsd-update --not-running-from-cron updatesready", privileged=True, timeout=60
+        ready = self._read(
+            "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 [
             {

napalm_bsd/openbsd.py

@@ -64,10 +64,10 @@ class OpenBSDDriver(BsdDriver):
         updates: list[dict] = [
             {**candidate, "origin": None, "security": None}
             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:
             installed = parse.syspatch(self._out("syspatch -l", privileged=True))
             summary = (

napalm_bsd/parse.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 import ipaddress
 import re
-from typing import Any, Optional
+from typing import Any
 
 _IFACE_HEADER = re.compile(
     r"^(?P<name>[^\s:]+): flags=[0-9a-f]+<(?P<flags>[^>]*)>(?:.*?\bmtu (?P<mtu>\d+))?"
@@ -26,7 +26,7 @@ def _prefix_from_netmask(mask: str) -> int:
 def ifconfig(text: str) -> dict[str, dict[str, Any]]:
     """``ifconfig -a`` -> per interface: flags, MTU, MAC, state, speed, addresses."""
     interfaces: dict[str, dict[str, Any]] = {}
-    current: Optional[dict[str, Any]] = None
+    current: dict[str, Any] | None = None
     for line in text.splitlines():
         header = _IFACE_HEADER.match(line)
         if header:
@@ -69,7 +69,7 @@ def ifconfig(text: str) -> dict[str, dict[str, Any]]:
     return interfaces
 
 
-def _network(destination: str, family: str) -> Optional[str]:
+def _network(destination: str, family: str) -> str | None:
     """A netstat destination as a network: ``default``, ``10.0.2/24``, ``fe80::%em0/64``."""
     if destination == "default":
         return "0.0.0.0/0" if family == "ipv4" else "::/0"
@@ -93,7 +93,7 @@ def routes(text: str) -> list[dict[str, str]]:
     """
     result: list[dict[str, str]] = []
     family = ""
-    iface_column: Optional[int] = None
+    iface_column: int | None = None
     for line in text.splitlines():
         words = line.split()
         if not words:
@@ -253,15 +253,20 @@ def crontab(text: str, *, system: bool = False, user: str = "") -> list[dict[str
     return jobs
 
 
-def boottime(text: str) -> Optional[int]:
+def boottime(text: str) -> int | None:
     """``sysctl -n kern.boottime``: FreeBSD ``{ sec = N, … }``, OpenBSD ``N``."""
     match = re.search(r"sec = (\d+)", text) or re.fullmatch(r"\s*(\d+)\s*", text)
     return int(match[1]) if match else None
 
 
+#: The wildcard addresses a socket listens on: reported as data, never bound to.
+_ANY_V4 = "0.0.0.0"  # nosec B104
+_ANY_V6 = "::"
+
+
 def _socket(
-    proto: str, local: str, family: str, process: Optional[str], pid: Optional[int]
-) -> Optional[dict[str, Any]]:
+    proto: str, local: str, family: str, process: str | None, pid: int | None
+) -> dict[str, Any] | None:
     """A ListeningSocketDict from a local endpoint: ``*:22``, ``[::1]:25``,
     ``127.0.0.1:25``, ``[fe80::1%lo0]:25`` (sockstat, fstat) or ``*.22``,
     ``::1.25``, ``fe80::1%lo0.25`` (netstat)."""
@@ -275,7 +280,7 @@ def _socket(
         return None
     address, _, zone = address.partition("%")
     if address == "*":
-        address = "0.0.0.0" if family == "4" else "::"
+        address = _ANY_V4 if family == "4" else _ANY_V6
     return {
         "proto": proto,
         "address": address,
@@ -288,9 +293,9 @@ def _socket(
     }
 
 
-def _unique(sockets: list[Optional[dict[str, Any]]]) -> list[dict[str, Any]]:
+def _unique(sockets: list[dict[str, Any] | None]) -> list[dict[str, Any]]:
     """One entry per socket, the first process holding it, as ss reports it."""
-    seen: set[tuple] = set()
+    seen: set[tuple[Any, ...]] = set()
     result = []
     for s in sockets:
         if s is None:

pyproject.toml

@@ -1,14 +1,14 @@
 [build-system]
-requires = ["setuptools>=68", "wheel"]
+requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-bsd"
-version = "0.1.0"
+version = "0.5.0"
 description = "NAPALM drivers for FreeBSD and OpenBSD hosts via SSH"
 readme = "README.md"
 requires-python = ">=3.10"
-license = { text = "Apache-2.0" }
+license = "Apache-2.0"
 authors = [
     { name = "Christian Manivong", email = "christian@manivong.de" },
 ]
@@ -16,22 +16,23 @@ keywords = ["napalm", "network", "automation", "freebsd", "openbsd", "bsd", "ssh
 classifiers = [
     "Development Status :: 3 - Alpha",
     "Intended Audience :: System Administrators",
-    "License :: OSI Approved :: Apache Software License",
     "Operating System :: OS Independent",
     "Programming Language :: Python :: 3",
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "Topic :: System :: Systems Administration",
+    "Typing :: Typed",
 ]
 dependencies = [
     "napalm>=4.0",
-    "napalm-device-types>=2.6.0",  # channel.run_on_transport
+    "napalm-device-types>=4.21.1",
     "paramiko>=5.0.0",  # CVE-2026-44405
 ]
 
 [project.optional-dependencies]
-dev = ["pytest", "pytest-cov", "ruff", "mypy"]
+dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"]
 
 [project.entry-points."napalm.drivers"]
 freebsd = "napalm_bsd:FreeBSDDriver"
@@ -39,11 +40,47 @@ openbsd = "napalm_bsd:OpenBSDDriver"
 
 [project.urls]
 Repository = "https://git.netork.io/NAPALM/napalm-bsd"
+Issues = "https://git.netork.io/NAPALM/napalm-bsd/issues"
+Changelog = "https://git.netork.io/NAPALM/napalm-bsd/src/branch/main/CHANGELOG.md"
 
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["napalm_bsd*"]
 
+[tool.setuptools.package-data]
+napalm_bsd = ["py.typed"]
+
 [tool.ruff]
 line-length = 100
 target-version = "py310"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B"]
+
+[tool.ruff.lint.per-file-ignores]
+# Recorded device output keeps its lines.
+"tests/**" = ["E501"]
+
+[tool.mypy]
+python_version = "3.10"
+strict = true
+# napalm itself is untyped, so every driver subclasses Any.
+disallow_subclassing_any = false
+
+[[tool.mypy.overrides]]
+module = ["napalm.*", "paramiko.*"]
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+# Not yet under strict mypy (quality baseline); may only shrink.
+module = ["napalm_bsd.base", "napalm_bsd.freebsd", "napalm_bsd.openbsd"]
+ignore_errors = true
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.coverage.run]
+source = ["napalm_bsd"]
+
+[tool.coverage.report]
+fail_under = 94

tests/quality_baseline.json

@@ -0,0 +1,16 @@
+{
+ "coverage_floor": 94.0,
+ "function_complexity": {
+  "napalm_bsd/parse.py::ifconfig": 21,
+  "napalm_bsd/parse.py::routes": 21,
+  "napalm_bsd/parse.py::users": 11
+ },
+ "function_lines": {},
+ "mypy_relaxed": [
+  "napalm_bsd.base",
+  "napalm_bsd.freebsd",
+  "napalm_bsd.openbsd"
+ ],
+ "python_file_lines": {},
+ "untyped_public": []
+}

tests/test_drivers.py

@@ -8,7 +8,7 @@ import pytest
 from napalm.base import NetworkDriver
 from napalm_device_types import OSDriver
 
-from napalm_bsd import FreeBSDDriver, OpenBSDDriver
+from napalm_bsd import FreeBSDDriver, OpenBSDDriver, base
 from napalm_bsd.base import BsdDriver
 
 FIXTURES = Path(__file__).parent / "fixtures"
@@ -96,7 +96,7 @@ class TestIdentity:
             "get_route_to",
             "get_arp_table",
             "get_lldp_neighbors",
-            "get_users",
+            "get_os_users",
             "get_processes",
             "get_cron_jobs",
         ],
@@ -104,6 +104,13 @@ class TestIdentity:
     def test_implements_what_the_poll_reads(self, cls, getter):
         """netOrk asks whether a getter is overridden, not merely present."""
         assert getattr(cls, getter) is not getattr(NetworkDriver, getter, None)
+        assert getattr(cls, getter) is not getattr(OSDriver, getter, None)
+
+    @pytest.mark.parametrize("cls", [FreeBSDDriver, OpenBSDDriver])
+    def test_leaves_napalms_get_users_alone(self, cls):
+        """NAPALM's get_users is a different getter; the host's accounts are get_os_users."""
+        assert "get_users" not in vars(BsdDriver)
+        assert "get_users" not in vars(cls)
 
 
 class TestFacts:
@@ -177,8 +184,8 @@ class TestRoutesAndNeighbours:
 
 
 class TestAccountsAndProcesses:
-    def test_users(self):
-        users = _freebsd_host().get_users()
+    def test_os_users(self):
+        users = _freebsd_host().get_os_users()
         assert next(u for u in users if u["username"] == "root")["groups"] == ["wheel", "operator"]
 
     def test_processes(self):
@@ -267,6 +274,15 @@ class TestPackages:
         assert driver.get_packages() == []
         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):
         driver = _channel(OpenBSDDriver, {"pkg_info": _read("openbsd-vm", "pkg_info_full.txt")})
         assert any(p["name"] == "pcre2" for p in driver.get_packages())
@@ -359,7 +375,7 @@ class TestAvailableUpdates:
         driver = _channel(
             FreeBSDDriver,
             {
-                "pkg -N": ("", 1),
+                "pkg -N": ("pkg: pkg is not installed", 1),
                 ("freebsd-update --not-running-from-cron updatesready", True): ("", 2),
             },
         )
@@ -445,9 +461,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)."""
 
-    PROBE_OK = "STRING: FreeBSD host 15.1-RELEASE-p4 FreeBSD 15.1-RELEASE-p4 GENERIC amd64"
+    STAGED = "/tmp/napalm-bsd-stage.Ab12Cd"
+    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "contact": "ops"}
 
     @pytest.mark.parametrize(
         ("cls", "install", "conf", "start"),
@@ -466,28 +484,40 @@ class TestSnmp:
             ),
         ],
     )
-    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 = {}
+    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
 
         def recording(command, *, privileged=False, timeout=60, stdin=None):
             if stdin is not None:
-                stdin_seen[command] = stdin
+                stdin_seen.append((command, privileged, 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")
+        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
-        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
+        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 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 "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.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)
 
     def test_a_failed_install_stops_and_says_so(self):
         driver = _channel(
@@ -499,20 +529,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(
@@ -529,6 +574,146 @@ 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
+
+
+class TestPrivilegedStdin:
+    """A privileged command never gets stdin while a sudo password is set (#8).
+
+    ``sudo -S`` reads the password from stdin; a sudo that does not prompt
+    (NOPASSWD, a cached timestamp) leaves it for the command, which would write
+    it into whatever the data goes to."""
+
+    def _driver(self, monkeypatch, *, sudo_password, root=False):
+        sent = []
+        monkeypatch.setattr(
+            base,
+            "run_on_transport",
+            lambda transport, line, stdin=None, timeout=60: sent.append((line, stdin)) or _Result(),
+        )
+        args = {"sudo_password": sudo_password} if sudo_password else {}
+        driver = FreeBSDDriver("host", "netork", "", optional_args=args)
+        driver._transport = lambda: object()  # type: ignore[method-assign]
+        driver._root = root
+        return driver, sent
+
+    def test_refuses_stdin_to_a_privileged_command_with_a_sudo_password(self, monkeypatch):
+        driver, sent = self._driver(monkeypatch, sudo_password="s3cret")
+        with pytest.raises(ValueError, match="no stdin"):
+            driver.run_command("cat > /etc/x", privileged=True, stdin=b"data")
+        assert sent == []
+
+    def test_a_privileged_command_without_stdin_gets_only_the_password(self, monkeypatch):
+        driver, sent = self._driver(monkeypatch, sudo_password="s3cret")
+        driver.run_command("id -u", privileged=True)
+        assert sent == [("sudo -S -p '' sh -c 'id -u'", b"s3cret\n")]
+
+    @pytest.mark.parametrize("sudo_password, root", [(None, False), ("s3cret", True)])
+    def test_stdin_is_fine_where_no_password_goes_first(self, monkeypatch, sudo_password, root):
+        driver, sent = self._driver(monkeypatch, sudo_password=sudo_password, root=root)
+        driver.run_command("cat > /tmp/x", privileged=True, stdin=b"data")
+        assert sent[0][1] == b"data"
+
+
+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 == []

tests/test_ssh_host_key.py

@@ -0,0 +1,109 @@
+"""open() verifies the host key it is given, and reports the one it sees (NetOrk/netork#1167)."""
+
+from __future__ import annotations
+
+import paramiko
+import pytest
+from napalm_device_types.ssh_host_keys import (
+    KNOWN_KEYS_ARG,
+    OBSERVER_ARG,
+    SshHostKey,
+    SshHostKeyMismatch,
+)
+
+from napalm_bsd import FreeBSDDriver, base
+
+SERVER_KEY = paramiko.ECDSAKey.generate()
+OTHER_KEY = paramiko.ECDSAKey.generate()
+
+
+class _HostKeys:
+    def __init__(self) -> None:
+        self.lines: list[str] = []
+
+    def load(self, path: str) -> None:
+        with open(path) as f:
+            self.lines += f.read().splitlines()
+
+
+class _Transport:
+    def get_remote_server_key(self):
+        return SERVER_KEY
+
+    def is_active(self) -> bool:
+        return True
+
+
+class _Client:
+    """paramiko.SSHClient as far as open() uses it; refuses like paramiko would."""
+
+    def __init__(self) -> None:
+        self.host_keys = _HostKeys()
+        self.policy = None
+        self.connected_with: dict | None = None
+        self.closed = False
+
+    def set_missing_host_key_policy(self, policy) -> None:
+        self.policy = policy
+
+    def get_host_keys(self) -> _HostKeys:
+        return self.host_keys
+
+    def connect(self, hostname, **kwargs) -> None:
+        if self.host_keys.lines and SshHostKey.of(SERVER_KEY).key not in self.host_keys.lines[0]:
+            raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY)
+        if not self.host_keys.lines:
+            self.policy.missing_host_key(self, hostname, SERVER_KEY)
+        self.connected_with = {"hostname": hostname, **kwargs}
+
+    def get_transport(self) -> _Transport:
+        return _Transport()
+
+    def close(self) -> None:
+        self.closed = True
+
+
+@pytest.fixture
+def client(monkeypatch) -> _Client:
+    fake = _Client()
+    monkeypatch.setattr(base.paramiko, "SSHClient", lambda: fake)
+    return fake
+
+
+def _driver(seen: list, known: SshHostKey | None = None) -> FreeBSDDriver:
+    args: dict = {"port": 2222, OBSERVER_ARG: seen.append}
+    if known is not None:
+        args[KNOWN_KEYS_ARG] = {"2222": known.as_dict()}
+    return FreeBSDDriver("10.0.0.7", "netork", "s3cret", optional_args=args)
+
+
+def test_first_use_connects_and_reports_the_key(client):
+    seen: list = []
+    _driver(seen).open()
+    assert client.connected_with["password"] == "s3cret"
+    assert client.connected_with["port"] == 2222
+    assert seen[0]["accepted"] is True
+    assert seen[0]["seen"]["fingerprint"] == SshHostKey.of(SERVER_KEY).fingerprint
+    assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.7", 2222)
+
+
+def test_the_known_key_connects(client):
+    seen: list = []
+    _driver(seen, SshHostKey.of(SERVER_KEY)).open()
+    assert client.host_keys.lines == [SshHostKey.of(SERVER_KEY).known_hosts_line("10.0.0.7", 2222)]
+    assert client.connected_with is not None
+    assert seen[0]["accepted"] is True
+
+
+def test_a_changed_key_is_refused_and_the_client_closed(client):
+    seen: list = []
+    with pytest.raises(SshHostKeyMismatch):
+        _driver(seen, SshHostKey.of(OTHER_KEY)).open()
+    assert client.connected_with is None
+    assert client.closed is True
+    assert seen[0]["accepted"] is False
+
+
+def test_without_host_key_args_it_connects_as_before(client):
+    FreeBSDDriver("10.0.0.7", "netork", "s3cret").open()
+    assert client.connected_with is not None