"""FreeBSDDriver and OpenBSDDriver: what netOrk's poll reads, on recorded output.""" from __future__ import annotations from pathlib import Path import pytest from napalm.base import NetworkDriver from napalm_device_types import OSDriver from napalm_bsd import FreeBSDDriver, OpenBSDDriver from napalm_bsd.base import BsdDriver FIXTURES = Path(__file__).parent / "fixtures" PS = "ps_axww_o_user_pid_ppid_cpu_mem_vsz_rss_stat_lstart_command.txt" def _read(system: str, name: str) -> str: return (FIXTURES / system / name).read_text() def _driver(cls, answers: dict[str, str]): """A driver whose commands are answered from *answers*; unknown ones print nothing.""" driver = cls("host", "netork", "") asked: list[str] = [] def out(command: str, *, privileged: bool = False, timeout: float = 60) -> str: asked.append(command) return answers.get(command, "").strip() driver._out = out # type: ignore[method-assign] driver.asked = asked # type: ignore[attr-defined] return driver def _freebsd_host(): s = "freebsd-host" return _driver( FreeBSDDriver, { "sysctl -n kern.hostname": "bsd-01.lab.example", "freebsd-version -u": "15.1-RELEASE", "freebsd-version -r": "15.1-RELEASE", FreeBSDDriver.PLATFORM_COMMAND: "maker=QEMU\nproduct=Standard PC (i440FX + PIIX, 1996)\nserial=", "sysctl -n kern.boottime": _read(s, "sysctl_n_kern.boottime.txt"), "date +%s": "1791408816", "ifconfig -a": _read(s, "ifconfig_a.txt"), "netstat -rnW": _read(s, "netstat_rn.txt"), "arp -an": _read(s, "arp_an.txt"), "cat /etc/passwd": _read(s, "cat_etc_passwd.txt"), "cat /etc/group": _read(s, "cat_etc_group.txt"), BsdDriver.PS_COMMAND: _read(s, PS), "cat /etc/crontab": _read(s, "cat_etc_crontab.txt"), "crontab -l": "0 4 * * * /home/netork/backup.sh\n", }, ) def _openbsd_vm(): s = "openbsd-vm" return _driver( OpenBSDDriver, { "hostname": "spike-openbsd", "uname -sr": "OpenBSD 7.9", "uname -v": "GENERIC.MP#449", OpenBSDDriver.PLATFORM_COMMAND: "vendor=QEMU\nproduct=Standard PC (i440FX + PIIX, 1996)\nserialno=", "sysctl -n kern.boottime": _read(s, "sysctl_n_kern.boottime.txt"), "date +%s": "1791406889", "ifconfig -a": _read(s, "ifconfig_a.txt"), "netstat -rn": _read(s, "netstat_rn.txt"), "arp -an": _read(s, "arp_an.txt"), "cat /etc/passwd": _read(s, "cat_etc_passwd.txt"), BsdDriver.PS_COMMAND: _read(s, PS), }, ) class TestIdentity: @pytest.mark.parametrize( ("cls", "name", "label"), [(FreeBSDDriver, "freebsd", "FreeBSD"), (OpenBSDDriver, "openbsd", "OpenBSD")], ) def test_names(self, cls, name, label): assert cls.DRIVER_NAME == name assert cls.TYPE_LABEL == label assert issubclass(cls, OSDriver) @pytest.mark.parametrize("cls", [FreeBSDDriver, OpenBSDDriver]) @pytest.mark.parametrize( "getter", [ "get_facts", "get_interfaces", "get_interfaces_ip", "get_route_to", "get_arp_table", "get_lldp_neighbors", "get_users", "get_processes", "get_cron_jobs", ], ) 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) class TestFacts: def test_freebsd(self): facts = _freebsd_host().get_facts() assert facts == { "hostname": "bsd-01", "fqdn": "bsd-01.lab.example", "vendor": "QEMU", "model": "Standard PC (i440FX + PIIX, 1996)", "serial_number": "", "os_version": "FreeBSD 15.1-RELEASE", "uptime": 1791408816 - 1791405216, "interface_list": ["em0"], "running_kernel": "15.1-RELEASE", } def test_openbsd(self): facts = _openbsd_vm().get_facts() assert facts["hostname"] == facts["fqdn"] == "spike-openbsd" assert facts["os_version"] == "OpenBSD 7.9" assert facts["running_kernel"] == "GENERIC.MP#449" assert facts["uptime"] == 100 assert facts["interface_list"] == ["vio0", "enc0", "pflog0"] def test_an_unreadable_boottime_is_no_uptime(self): driver = _freebsd_host() driver._out = lambda command, **kw: "" # type: ignore[method-assign] assert driver.get_facts()["uptime"] == -1 class TestInterfaces: def test_napalm_shape(self): interfaces = _freebsd_host().get_interfaces() assert interfaces["em0"] == { "is_up": True, "is_enabled": True, "description": "", "last_flapped": -1.0, "speed": 1000.0, "mtu": 1500, "mac_address": "bc:24:11:20:b0:20", } def test_addresses(self): assert _openbsd_vm().get_interfaces_ip()["vio0"] == { "ipv4": {"10.0.2.15": {"prefix_length": 24}}, "ipv6": {}, } class TestRoutesAndNeighbours: def test_routes_keyed_by_network(self): routes = _freebsd_host().get_route_to() default = routes["0.0.0.0/0"][0] assert default["next_hop"] == "192.0.2.1" assert default["outgoing_interface"] == "em0" assert default["protocol"] == "static" assert default["current_active"] is True def test_openbsd_reads_routes_without_W(self): driver = _openbsd_vm() assert "10.0.2.0/24" in driver.get_route_to() assert "netstat -rn" in driver.asked def test_arp(self): assert _openbsd_vm().get_arp_table()[0]["ip"] == "10.0.2.2" def test_no_lldp_daemon_means_no_neighbours(self): assert _freebsd_host().get_lldp_neighbors() == {} class TestAccountsAndProcesses: def test_users(self): users = _freebsd_host().get_users() assert next(u for u in users if u["username"] == "root")["groups"] == ["wheel", "operator"] def test_processes(self): init = next(p for p in _openbsd_vm().get_processes() if p["pid"] == 1) assert init["command"] == "/sbin/init" def test_cron_jobs_from_the_system_and_the_login_users_crontab(self): jobs = _freebsd_host().get_cron_jobs() assert {"user": "root", "schedule": "0 * * * *", "command": "newsyslog"} in jobs assert { "user": "netork", "schedule": "0 4 * * *", "command": "/home/netork/backup.sh", } in jobs class TestListeningSockets: def test_freebsd_as_root_is_attributed(self): driver = _driver(FreeBSDDriver, {}) sockstat = _read("freebsd-vm", "sudo_sockstat_46lq_P_tcp_udp.txt") driver.run_command = lambda command, privileged=False, timeout=60, stdin=None: ( # type: ignore[method-assign] type("R", (), {"stdout": sockstat, "exit_code": 0})() ) result = driver.get_listening_sockets() assert result["attributed"] is True assert len(result["sockets"]) == 7 def test_openbsd_without_root_falls_back_to_netstat(self): driver = _driver(OpenBSDDriver, {}) netstat = _read("openbsd-vm", "netstat_an.txt") calls = [] def run(command, privileged=False, timeout=60, stdin=None): calls.append((command, privileged)) if privileged: # sudo -n without a password: refused return type("R", (), {"stdout": "", "exit_code": 1})() return type("R", (), {"stdout": netstat, "exit_code": 0})() driver.run_command = run # type: ignore[method-assign] result = driver.get_listening_sockets() assert result["attributed"] is False assert ("tcp", "0.0.0.0", 22) in { (s["proto"], s["address"], s["port"]) for s in result["sockets"] } assert calls == [("fstat -n", True), (OpenBSDDriver.LISTENING_FALLBACK_COMMAND, False)] class _Result: def __init__(self, stdout: str = "", exit_code: int = 0, stderr: str = "") -> None: self.stdout, self.exit_code, self.stderr = stdout, exit_code, stderr def _channel(cls, answers: dict): """A driver whose run_command answers from *answers*: ``command -> stdout`` or ``command -> (stdout, exit_code)``; ``(command, True)`` keys answer only privileged calls. Every call is recorded as ``(command, privileged)``.""" driver = cls("host", "netork", "") calls: list[tuple[str, bool]] = [] def run(command, *, privileged=False, timeout=60, stdin=None): calls.append((command, privileged)) answer = answers.get((command, privileged), answers.get(command, "")) stdout, code = answer if isinstance(answer, tuple) else (answer, 0) return _Result(stdout, code) driver.run_command = run # type: ignore[method-assign] driver.calls = calls # type: ignore[attr-defined] return driver class TestPackages: def test_freebsd_packages_with_their_repository(self): driver = _channel( FreeBSDDriver, { "pkg -N": "", FreeBSDDriver.PKG_QUERY: _read("freebsd-vm", "pkg_query_n_v_R_sb_c.txt"), }, ) packages = driver.get_packages() assert {p["source"] for p in packages} >= {"FreeBSD-base", "FreeBSD-ports"} def test_freebsd_without_pkg_has_no_packages(self): """A hand-installed classic system may never have bootstrapped pkg.""" driver = _channel(FreeBSDDriver, {"pkg -N": ("pkg: pkg is not installed", 1)}) 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()) @pytest.mark.parametrize( ("cls", "install", "uninstall"), [ ( FreeBSDDriver, "env ASSUME_ALWAYS_YES=yes pkg install -y nginx", "pkg delete -y nginx", ), (OpenBSDDriver, "pkg_add -I nginx", "pkg_delete nginx"), ], ) def test_install_and_uninstall_as_root(self, cls, install, uninstall): driver = _channel(cls, {}) driver.install_package("nginx") driver.uninstall_package("nginx") assert driver.calls == [(install, True), (uninstall, True)] def test_a_version_is_part_of_the_name(self): driver = _channel(FreeBSDDriver, {}) driver.install_package("nginx", "1.28.0") assert driver.calls == [("env ASSUME_ALWAYS_YES=yes pkg install -y nginx-1.28.0", True)] def test_a_failed_install_raises(self): driver = _channel(OpenBSDDriver, {("pkg_add -I nope", True): ("Can't find nope", 1)}) with pytest.raises(RuntimeError, match="nope"): driver.install_package("nope") @pytest.mark.parametrize("name", ["", "nginx; reboot", "-f", "a b"]) def test_a_name_that_is_not_a_package_is_refused(self, name): driver = _channel(FreeBSDDriver, {}) with pytest.raises(ValueError): driver.install_package(name) assert driver.calls == [] class TestAvailableUpdates: def test_freebsd_marks_what_vuxml_knows(self): driver = _channel( FreeBSDDriver, { "pkg -N": "", ("pkg upgrade -n", True): _read("freebsd-vm", "sudo_pkg_upgrade_n_latest.txt"), ("pkg audit -Fq", True): ("python312-3.12.14\nexpat-2.8.2\n", 1), "pkg repos -l": _read("freebsd-vm", "pkg_repos_l.txt"), }, ) updates = {u["name"]: u for u in driver.get_available_updates()} assert len(updates) == 10 assert updates["expat"]["security"] is True assert updates["pkg"]["security"] is False assert updates["pkg"]["origin"] == "FreeBSD-ports" assert "base-system" not in updates # pkgbase: the base system is packages def test_freebsd_without_an_audit_cannot_tell(self): driver = _channel( FreeBSDDriver, { "pkg -N": "", ("pkg upgrade -n", True): _read("freebsd-vm", "sudo_pkg_upgrade_n_latest.txt"), ("pkg audit -Fq", True): ("pkg: cannot fetch", 3), "pkg repos -l": "FreeBSD-ports\nFreeBSD-base\n", }, ) assert {u["security"] for u in driver.get_available_updates()} == {None} def test_classic_freebsd_reports_fetched_base_updates_as_one_entry(self): driver = _channel( FreeBSDDriver, { "pkg -N": ("pkg: pkg is not installed", 1), ("freebsd-update --not-running-from-cron updatesready", True): ("", 0), "freebsd-version -u": "15.1-RELEASE", }, ) assert driver.get_available_updates() == [ { "name": "base-system", "current_version": "15.1-RELEASE", "new_version": "fetched by freebsd-update", "origin": "freebsd-update", "security": None, } ] def test_classic_freebsd_without_fetched_updates(self): driver = _channel( FreeBSDDriver, { "pkg -N": ("pkg: pkg is not installed", 1), ("freebsd-update --not-running-from-cron updatesready", True): ("", 2), }, ) assert driver.get_available_updates() == [] def test_openbsd_packages_and_syspatches(self): driver = _channel( OpenBSDDriver, { ("pkg_add -u -n -v", True): _read("openbsd-vm", "sudo_pkg_add_u_n_v.txt"), ("syspatch -c", True): _read("openbsd-vm", "syspatch_c.txt"), ("syspatch -l", True): "", "uname -r": "7.9", }, ) updates = {u["name"]: u for u in driver.get_available_updates()} assert updates["pcre2"]["new_version"] == "10.48" assert updates["pcre2"]["security"] is None base = updates["base-system"] assert base["current_version"] == "7.9" assert base["new_version"].startswith("34 patches: 001_xserver") assert base["origin"] == "syspatch" assert base["security"] is None def test_openbsd_fully_patched(self): driver = _channel( OpenBSDDriver, {("pkg_add -u -n -v", True): "", ("syspatch -c", True): ""} ) assert driver.get_available_updates() == [] class TestServices: def test_freebsd_reads_the_status_as_root(self): status = _read("freebsd-vm", "sudo_service_status_loop.txt") driver = _channel(FreeBSDDriver, {(FreeBSDDriver.SERVICE_STATUS_COMMAND, True): status}) services = {s["name"]: s for s in driver.get_services()} assert services["cron"]["running"] is True def test_freebsd_without_root_reads_what_it_can(self): status = _read("freebsd-vm", "service_status_loop.txt") driver = _channel( FreeBSDDriver, { (FreeBSDDriver.SERVICE_STATUS_COMMAND, True): ("sudo: a password is required", 1), (FreeBSDDriver.SERVICE_STATUS_COMMAND, False): status, }, ) assert {s["name"] for s in driver.get_services()} >= {"sshd", "cron"} def test_openbsd(self): driver = _channel( OpenBSDDriver, {OpenBSDDriver.SERVICE_STATUS_COMMAND: _read("openbsd-vm", "rcctl_check_loop.txt")}, ) assert {s["name"]: s["running"] for s in driver.get_services()}["smtpd"] is True @pytest.mark.parametrize( ("cls", "command"), [(FreeBSDDriver, "service sshd restart"), (OpenBSDDriver, "rcctl restart sshd")], ) def test_manage_as_root(self, cls, command): driver = _channel(cls, {}) assert driver.manage_service("sshd", "restart") == {"success": True, "output": ""} assert driver.calls == [(command, True)] def test_a_failed_action_says_so(self): driver = _channel( OpenBSDDriver, {("rcctl start nope", True): ("rcctl: service nope does not exist", 2)} ) result = driver.manage_service("nope", "start") assert result["success"] is False assert "does not exist" in result["output"] @pytest.mark.parametrize( ("name", "action"), [("sshd", "reload"), ("sshd;reboot", "start"), ("", "start")] ) def test_refused_before_anything_is_sent(self, name, action): driver = _channel(FreeBSDDriver, {}) with pytest.raises(ValueError): driver.manage_service(name, action) 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 == []