christianmanivong
e8468a292a
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 24s
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 25s
The write side NetOrk/netork#799 needs, on both drivers. - Packages: get_packages (FreeBSD `pkg query` with the repository as source, nothing on a classic system without pkg; OpenBSD `pkg_info`), install_package / uninstall_package as root (`pkg install`/`pkg delete`, `pkg_add -I`/`pkg_delete`), names checked before anything is sent, a failure raised with what the tool printed. - Updates: get_available_updates. FreeBSD `pkg upgrade -n`, with `security` from VuXML (`pkg audit`: True for a listed package, False for any other, None when the audit could not run); OpenBSD `pkg_add -u -n -v` (no security verdict). Base-system patches are one `base-system` entry, as decided in #799: OpenBSD's `syspatch -c`, and on a classic FreeBSD what `freebsd-update` has fetched; on pkgbase the base system is packages from FreeBSD-base. - Services: get_services lists the enabled ones (FreeBSD `service ... status` as root, as root-only pidfiles hide daemons otherwise, without root when sudo refuses; OpenBSD `rcctl check`), manage_service runs start/stop/restart/enable/disable as root and returns success and output. Fixtures recorded on the FreeBSD 15.1 and OpenBSD 7.9 VMs with genuinely outdated packages (FreeBSD pointed at the latest branch, an OpenBSD package taken back to its release build). Checked live on both, including a service restart and installing and removing a package.
358 lines
14 KiB
Python
358 lines
14 KiB
Python
"""Parsers for BSD command output, on output recorded from real systems.
|
|
|
|
``freebsd-host``: a hand-installed FreeBSD 15.1 (classic base, no pkg).
|
|
``freebsd-vm``: the FreeBSD 15.1 BASIC-CLOUDINIT image (pkgbase).
|
|
``openbsd-vm``: the OpenBSD 7.9 cloud image from hcartiaux/openbsd-cloud-image.
|
|
Recorded for NetOrk/netork#793; ``COMMANDS.txt`` in each directory lists them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from napalm_bsd import parse
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def fixture(system: str, name: str) -> str:
|
|
return (FIXTURES / system / name).read_text()
|
|
|
|
|
|
class TestIfconfig:
|
|
def test_freebsd_ethernet(self):
|
|
em0 = parse.ifconfig(fixture("freebsd-host", "ifconfig_a.txt"))["em0"]
|
|
assert em0["mac"] == "bc:24:11:20:b0:20"
|
|
assert em0["mtu"] == 1500
|
|
assert em0["enabled"] and em0["up"]
|
|
assert em0["speed"] == 1000.0
|
|
assert em0["ipv4"] == {"192.0.2.104": 24}
|
|
assert em0["ipv6"] == {"fe80::be24:11ff:fe20:b020": 64}
|
|
|
|
def test_freebsd_loopback(self):
|
|
lo0 = parse.ifconfig(fixture("freebsd-host", "ifconfig_a.txt"))["lo0"]
|
|
assert lo0["loopback"]
|
|
assert lo0["ipv4"] == {"127.0.0.1": 8}
|
|
assert lo0["ipv6"] == {"::1": 128, "fe80::1": 64}
|
|
|
|
def test_freebsd_vm_reports_ten_gigabit(self):
|
|
vtnet0 = parse.ifconfig(fixture("freebsd-vm", "ifconfig_a.txt"))["vtnet0"]
|
|
assert vtnet0["speed"] == 10000.0
|
|
assert "10.0.2.15" in vtnet0["ipv4"]
|
|
|
|
def test_openbsd_calls_the_mac_lladdr(self):
|
|
interfaces = parse.ifconfig(fixture("openbsd-vm", "ifconfig_a.txt"))
|
|
assert list(interfaces) == ["lo0", "vio0", "enc0", "pflog0"]
|
|
assert interfaces["vio0"]["mac"] == "bc:24:11:aa:bb:04"
|
|
assert interfaces["vio0"]["ipv4"] == {"10.0.2.15": 24}
|
|
assert interfaces["vio0"]["up"]
|
|
|
|
def test_an_interface_without_flags_or_mtu_is_down(self):
|
|
enc0 = parse.ifconfig(fixture("openbsd-vm", "ifconfig_a.txt"))["enc0"]
|
|
assert not enc0["enabled"] and not enc0["up"]
|
|
assert enc0["mtu"] == 0
|
|
assert enc0["mac"] == ""
|
|
|
|
|
|
class TestRoutes:
|
|
def test_freebsd_default_and_connected(self):
|
|
routes = parse.routes(fixture("freebsd-host", "netstat_rn.txt"))
|
|
assert routes == [
|
|
{
|
|
"network": "0.0.0.0/0",
|
|
"next_hop": "192.0.2.1",
|
|
"interface": "em0",
|
|
"protocol": "static",
|
|
"family": "ipv4",
|
|
},
|
|
{
|
|
"network": "192.0.2.0/24",
|
|
"next_hop": "",
|
|
"interface": "em0",
|
|
"protocol": "connected",
|
|
"family": "ipv4",
|
|
},
|
|
{
|
|
"network": "fe80::/64",
|
|
"next_hop": "",
|
|
"interface": "em0",
|
|
"protocol": "connected",
|
|
"family": "ipv6",
|
|
},
|
|
]
|
|
|
|
def test_openbsd_abbreviated_networks_are_written_out(self):
|
|
routes = parse.routes(fixture("openbsd-vm", "netstat_rn.txt"))
|
|
networks = {r["network"]: r for r in routes}
|
|
assert networks["0.0.0.0/0"]["next_hop"] == "10.0.2.2"
|
|
assert networks["10.0.2.0/24"]["protocol"] == "connected"
|
|
assert networks["10.0.2.0/24"]["next_hop"] == ""
|
|
assert all(r["interface"] != "lo0" for r in routes)
|
|
|
|
def test_host_and_broadcast_entries_are_not_routes(self):
|
|
routes = parse.routes(fixture("openbsd-vm", "netstat_rn.txt"))
|
|
assert "10.0.2.15/32" not in {r["network"] for r in routes}
|
|
assert "10.0.2.255/32" not in {r["network"] for r in routes}
|
|
|
|
|
|
class TestArp:
|
|
def test_freebsd(self):
|
|
entries = parse.arp(fixture("freebsd-host", "arp_an.txt"))
|
|
assert entries[0] == {
|
|
"interface": "em0",
|
|
"mac": "bc:24:11:3c:e2:d0",
|
|
"ip": "192.0.2.254",
|
|
"age": 1176.0,
|
|
}
|
|
assert entries[2]["age"] == -1.0 # permanent: its own address
|
|
|
|
def test_openbsd_table(self):
|
|
entries = parse.arp(fixture("openbsd-vm", "arp_an.txt"))
|
|
assert entries[0] == {
|
|
"interface": "vio0",
|
|
"mac": "52:55:0a:00:02:02",
|
|
"ip": "10.0.2.2",
|
|
"age": 18 * 60 + 58.0,
|
|
}
|
|
assert entries[2]["age"] == -1.0
|
|
|
|
@pytest.mark.parametrize(
|
|
("text", "seconds"), [("18m58s", 1138.0), ("1h2m3s", 3723.0), ("45s", 45.0)]
|
|
)
|
|
def test_openbsd_durations(self, text, seconds):
|
|
assert parse.duration(text) == seconds
|
|
|
|
|
|
class TestUsers:
|
|
def test_passwd_with_supplementary_groups(self):
|
|
users = parse.users(
|
|
fixture("freebsd-host", "cat_etc_passwd.txt"),
|
|
fixture("freebsd-host", "cat_etc_group.txt"),
|
|
)
|
|
root = next(u for u in users if u["username"] == "root")
|
|
assert root == {
|
|
"username": "root",
|
|
"uid": 0,
|
|
"gid": 0,
|
|
"home": "/root",
|
|
"shell": "/bin/sh",
|
|
"groups": ["wheel", "operator"],
|
|
}
|
|
assert any(u["username"] == "netork" for u in users)
|
|
|
|
def test_comment_lines_are_skipped(self):
|
|
assert (
|
|
parse.users("# $FreeBSD$\n#\nroot:*:0:0:Charlie &:/root:/bin/sh\n", "")[0]["username"]
|
|
== "root"
|
|
)
|
|
|
|
|
|
class TestProcesses:
|
|
@pytest.mark.parametrize("system", ["freebsd-host", "freebsd-vm", "openbsd-vm"])
|
|
def test_the_start_date_is_kept_whole(self, system):
|
|
"""lstart is five words; napalm-linux cut it to the weekday (napalm-linux#21)."""
|
|
processes = parse.processes(
|
|
fixture(system, "ps_axww_o_user_pid_ppid_cpu_mem_vsz_rss_stat_lstart_command.txt")
|
|
)
|
|
init = next(p for p in processes if p["pid"] == 1)
|
|
assert init["command"] == "/sbin/init"
|
|
assert init["user"] == "root"
|
|
assert init["ppid"] == 0
|
|
assert len(init["started"].split()) == 5
|
|
|
|
def test_fields(self):
|
|
processes = parse.processes(
|
|
fixture("openbsd-vm", "ps_axww_o_user_pid_ppid_cpu_mem_vsz_rss_stat_lstart_command.txt")
|
|
)
|
|
assert processes[0] == {
|
|
"pid": 1,
|
|
"ppid": 0,
|
|
"user": "root",
|
|
"cpu": 0.0,
|
|
"memory": 0.0,
|
|
"vsz": 956,
|
|
"rss": 504,
|
|
"tty": "",
|
|
"state": "I",
|
|
"started": "Wed Oct 7 15:59:49 2026",
|
|
"command": "/sbin/init",
|
|
}
|
|
|
|
|
|
class TestCrontab:
|
|
def test_system_crontab_names_the_user(self):
|
|
jobs = parse.crontab(fixture("freebsd-host", "cat_etc_crontab.txt"), system=True)
|
|
assert {"user": "root", "schedule": "0 * * * *", "command": "newsyslog"} in jobs
|
|
assert {
|
|
"user": "operator",
|
|
"schedule": "*/11 * * * *",
|
|
"command": "/usr/libexec/save-entropy",
|
|
} in jobs
|
|
|
|
def test_user_crontab(self):
|
|
jobs = parse.crontab(fixture("openbsd-vm", "crontab_l.txt"), user="root")
|
|
assert jobs
|
|
assert all(j["user"] == "root" for j in jobs)
|
|
assert all(len(j["schedule"].split()) == 5 for j in jobs)
|
|
|
|
def test_special_schedules_and_environment_lines(self):
|
|
text = "SHELL=/bin/sh\n@reboot root /usr/local/bin/start\n# comment\n"
|
|
assert parse.crontab(text, system=True) == [
|
|
{"user": "root", "schedule": "@reboot", "command": "/usr/local/bin/start"}
|
|
]
|
|
|
|
|
|
class TestBoottime:
|
|
def test_freebsd(self):
|
|
assert parse.boottime(fixture("freebsd-host", "sysctl_n_kern.boottime.txt")) == 1791405216
|
|
|
|
def test_openbsd(self):
|
|
assert parse.boottime(fixture("openbsd-vm", "sysctl_n_kern.boottime.txt")) == 1791406789
|
|
|
|
def test_garbage(self):
|
|
assert parse.boottime("sysctl: unknown oid") is None
|
|
|
|
|
|
class TestListeningSockets:
|
|
"""In the shape of napalm-device-types' ListeningSocketDict: ``0.0.0.0`` and
|
|
``::`` are every address of their family, as ``ss`` prints them."""
|
|
|
|
@staticmethod
|
|
def _keys(sockets):
|
|
return {(s["proto"], s["address"], s["port"], s["interface"]) for s in sockets}
|
|
|
|
def test_freebsd_sockstat(self):
|
|
sockets = parse.sockstat(fixture("freebsd-vm", "sudo_sockstat_46lq_P_tcp_udp.txt"))
|
|
assert self._keys(sockets) == {
|
|
("udp", "127.0.0.1", 7779, None),
|
|
("tcp", "::1", 7777, None),
|
|
("tcp", "127.0.0.1", 7778, None),
|
|
("tcp", "::", 22, None),
|
|
("tcp", "0.0.0.0", 22, None),
|
|
("udp", "::", 514, None),
|
|
("udp", "0.0.0.0", 514, None),
|
|
}
|
|
sshd = next(s for s in sockets if s["port"] == 22 and s["address"] == "0.0.0.0")
|
|
assert (sshd["process"], sshd["pid"], sshd["unit"], sshd["container_id"]) == (
|
|
"sshd",
|
|
1282,
|
|
None,
|
|
None,
|
|
)
|
|
|
|
def test_openbsd_fstat_keeps_only_listening_and_bound_sockets(self):
|
|
sockets = parse.fstat_sockets(fixture("openbsd-vm", "sudo_fstat_n.txt"))
|
|
keys = self._keys(sockets)
|
|
assert ("tcp", "0.0.0.0", 22, None) in keys
|
|
assert ("tcp", "::", 22, None) in keys
|
|
assert ("tcp", "::1", 25, None) in keys
|
|
assert ("tcp", "fe80::1", 25, "lo0") in keys
|
|
assert ("udp", "10.0.2.15", 68, None) in keys
|
|
assert ("udp", "127.0.0.1", 7779, None) in keys
|
|
# connected (ntpd, ssh sessions) and unbound (*:0) sockets are not listening
|
|
assert not any(s["port"] in (0, 123) or s["process"] == "sshd-session" for s in sockets)
|
|
smtpd = next(s for s in sockets if s["port"] == 25 and s["address"] == "127.0.0.1")
|
|
assert (smtpd["process"], smtpd["pid"]) == ("smtpd", 69044)
|
|
|
|
def test_openbsd_netstat_without_root_names_no_process(self):
|
|
sockets = parse.netstat_listening(fixture("openbsd-vm", "netstat_an.txt"))
|
|
keys = self._keys(sockets)
|
|
assert ("tcp", "0.0.0.0", 22, None) in keys
|
|
assert ("tcp", "::", 22, None) in keys
|
|
assert ("tcp", "fe80::1", 25, "lo0") in keys
|
|
assert ("udp", "127.0.0.1", 7779, None) in keys
|
|
assert all(s["process"] is None and s["pid"] is None for s in sockets)
|
|
assert not any(s["port"] == 123 for s in sockets) # ntpd's connected sockets
|
|
|
|
def test_a_socket_held_by_two_processes_is_listed_once(self):
|
|
text = "www nginx 901 6 tcp4 *:80 *:*\nwww nginx 900 6 tcp4 *:80 *:*\n"
|
|
assert [(s["pid"]) for s in parse.sockstat(text)] == [901]
|
|
|
|
|
|
class TestPackages:
|
|
def test_freebsd_pkg_query_names_the_repository(self):
|
|
packages = parse.pkg_query(fixture("freebsd-vm", "pkg_query_n_v_R_sb_c.txt"))
|
|
acct = next(p for p in packages if p["name"] == "FreeBSD-acct")
|
|
assert acct == {
|
|
"name": "FreeBSD-acct",
|
|
"version": "15.1p4",
|
|
"installed": True,
|
|
"description": "System resource accounting",
|
|
"size": 118976,
|
|
"source": "FreeBSD-base",
|
|
}
|
|
assert any(p["source"] == "FreeBSD-ports" for p in packages)
|
|
|
|
@pytest.mark.parametrize(
|
|
("full", "name", "version"),
|
|
[
|
|
("pcre2-10.44", "pcre2", "10.44"),
|
|
("bash-completion-2.17.0", "bash-completion", "2.17.0"),
|
|
("py3-tzdata-2026.3", "py3-tzdata", "2026.3"),
|
|
("argon2-20190702p0", "argon2", "20190702p0"),
|
|
("gettext-runtime-1.0", "gettext-runtime", "1.0"),
|
|
],
|
|
)
|
|
def test_openbsd_package_names(self, full, name, version):
|
|
assert parse.split_package(full) == (name, version)
|
|
|
|
def test_openbsd_pkg_info(self):
|
|
packages = parse.pkg_info(fixture("openbsd-vm", "pkg_info_full.txt"))
|
|
assert packages[1] == {
|
|
"name": "bash",
|
|
"version": "5.3.9",
|
|
"installed": True,
|
|
"description": "GNU Bourne Again Shell",
|
|
"size": 0,
|
|
"source": "pkg_add",
|
|
}
|
|
assert any(p["name"] == "bash-completion" for p in packages)
|
|
|
|
|
|
class TestUpdates:
|
|
def test_freebsd_pkg_upgrade_dry_run(self):
|
|
updates = parse.pkg_upgrades(fixture("freebsd-vm", "sudo_pkg_upgrade_n_latest.txt"))
|
|
assert len(updates) == 10
|
|
assert updates[0] == {
|
|
"name": "expat",
|
|
"current_version": "2.8.2",
|
|
"new_version": "2.8.5",
|
|
"origin": "FreeBSD-ports",
|
|
}
|
|
|
|
def test_freebsd_pkg_audit_names_the_vulnerable_packages(self):
|
|
text = fixture("freebsd-vm", "sudo_pkg_audit_Fq.txt").replace("rc=1\n", "")
|
|
assert parse.pkg_audit(text) == {"python312", "expat"}
|
|
|
|
def test_openbsd_update_candidates_that_change(self):
|
|
updates = parse.pkg_add_candidates(fixture("openbsd-vm", "sudo_pkg_add_u_n_v.txt"))
|
|
assert updates == [
|
|
{"name": "pcre2", "current_version": "10.44", "new_version": "10.48"},
|
|
{"name": "py3-tzdata", "current_version": "2026.3", "new_version": "2026.4"},
|
|
]
|
|
|
|
def test_openbsd_syspatch(self):
|
|
patches = parse.syspatch(fixture("openbsd-vm", "syspatch_c.txt"))
|
|
assert len(patches) == 34
|
|
assert patches[0] == "001_xserver"
|
|
|
|
|
|
class TestServices:
|
|
def test_freebsd_status_as_root(self):
|
|
services = {
|
|
s["name"]: s
|
|
for s in parse.service_status(fixture("freebsd-vm", "sudo_service_status_loop.txt"))
|
|
}
|
|
assert services["sshd"] == {"name": "sshd", "running": True, "enabled": True, "pid": 1282}
|
|
assert services["qemu-guest-agent"]["running"] is False
|
|
assert services["hostid"]["running"] is False # a one-shot script without a status
|
|
|
|
def test_openbsd_rcctl_check(self):
|
|
services = {
|
|
s["name"]: s for s in parse.rcctl_check(fixture("openbsd-vm", "rcctl_check_loop.txt"))
|
|
}
|
|
assert services["sshd"] == {"name": "sshd", "running": True, "enabled": True, "pid": 0}
|
|
assert services["check_quotas"]["running"] is False
|