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
92 lines
3.6 KiB
Python
92 lines
3.6 KiB
Python
"""OpenBSD: hw.* sysctls for the hardware, uname for release and kernel."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from napalm_device_types import FingerprintRule
|
|
|
|
from napalm_bsd import parse
|
|
from napalm_bsd.base import BASE_SYSTEM, BsdDriver
|
|
|
|
|
|
class OpenBSDDriver(BsdDriver):
|
|
"""NAPALM driver for OpenBSD hosts, over SSH."""
|
|
|
|
TYPE_LABEL = "OpenBSD"
|
|
VENDOR = "OpenBSD"
|
|
DRIVER_NAME = "openbsd"
|
|
# OpenBSD's SSH banner names no OS; its sysDescr does ("OpenBSD host 7.9 GENERIC.MP#449").
|
|
SNMP_FINGERPRINT = [FingerprintRule("openbsd", weight=5.0)]
|
|
|
|
OS_VERSION_COMMAND = "uname -sr"
|
|
# The kernel's build, e.g. GENERIC.MP#449; the release is in uname -r.
|
|
KERNEL_COMMAND = "uname -v"
|
|
# A sysctl node the machine lacks fails on its own and prints nothing.
|
|
PLATFORM_COMMAND = (
|
|
"for k in vendor product serialno; do "
|
|
'printf "%s=%s\\n" "$k" "$(sysctl -n hw.$k 2>/dev/null)"; done'
|
|
)
|
|
|
|
# fstat shows a user only their own processes; netstat shows every socket, unnamed.
|
|
LISTENING_COMMAND = "fstat -n"
|
|
LISTENING_FALLBACK_COMMAND = "netstat -an -f inet; netstat -an -f inet6"
|
|
|
|
def _parse_listening(self, output: str, *, attributed: bool) -> list[dict]:
|
|
return parse.fstat_sockets(output) if attributed else parse.netstat_listening(output)
|
|
|
|
INSTALL_COMMAND = "pkg_add -I {name}"
|
|
UNINSTALL_COMMAND = "pkg_delete {name}"
|
|
# rcctl check needs no root; "rcctl ls started" does.
|
|
SERVICE_STATUS_COMMAND = (
|
|
"for s in $(rcctl ls on); do "
|
|
"if rcctl check $s >/dev/null 2>&1; then r=1; else r=0; fi; "
|
|
'printf "%s\\t%s\\n" "$s" "$r"; done'
|
|
)
|
|
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]:
|
|
return parse.rcctl_check(output)
|
|
|
|
def get_services(self) -> list[dict]:
|
|
return self._parse_services(self._out(self.SERVICE_STATUS_COMMAND, timeout=120))
|
|
|
|
def get_packages(self) -> list[dict]:
|
|
return parse.pkg_info(self._out("pkg_info"))
|
|
|
|
def get_available_updates(self) -> list[dict]:
|
|
"""Package updates from ``pkg_add -u -n -v``, and the base system's syspatches.
|
|
|
|
syspatch applies its patches together and in order, so they are one
|
|
entry (#799). Neither tool says which update is a security fix.
|
|
"""
|
|
updates: list[dict] = [
|
|
{**candidate, "origin": None, "security": None}
|
|
for candidate in parse.pkg_add_candidates(
|
|
self._read("pkg_add -u -n -v", privileged=True, timeout=300).stdout
|
|
)
|
|
]
|
|
patches = parse.syspatch(self._read("syspatch -c", privileged=True, timeout=120).stdout)
|
|
if patches:
|
|
installed = parse.syspatch(self._out("syspatch -l", privileged=True))
|
|
summary = (
|
|
", ".join(patches)
|
|
if len(patches) <= 3
|
|
else f"{len(patches)} patches: {patches[0]} … {patches[-1]}"
|
|
)
|
|
updates.append(
|
|
{
|
|
"name": BASE_SYSTEM,
|
|
"current_version": installed[-1] if installed else self._out("uname -r"),
|
|
"new_version": summary,
|
|
"origin": "syspatch",
|
|
"security": None,
|
|
}
|
|
)
|
|
return updates
|
|
|
|
def _hardware(self) -> tuple[str, str, str]:
|
|
hw = self._platform()
|
|
return hw.get("vendor", ""), hw.get("product", ""), hw.get("serialno", "")
|