napalm-bsd/napalm_bsd/freebsd.py

"""FreeBSD: kenv for the hardware, freebsd-version for base 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 FreeBSDDriver(BsdDriver):
    """NAPALM driver for FreeBSD hosts, over SSH."""

    TYPE_LABEL = "FreeBSD"
    VENDOR = "FreeBSD"
    DRIVER_NAME = "freebsd"
    SNMP_FINGERPRINT = [FingerprintRule("freebsd", weight=5.0)]
    # OpenSSH in FreeBSD's base names it in its banner ("OpenSSH_9.9 FreeBSD-20250219").
    SSH_FINGERPRINT = [FingerprintRule("freebsd", weight=3.0)]

    HOSTNAME_COMMAND = "sysctl -n kern.hostname"
    # Userland and running kernel; both carry the patch level (15.1-RELEASE-p4).
    OS_VERSION_COMMAND = "freebsd-version -u"
    KERNEL_COMMAND = "freebsd-version -r"
    # kenv takes one variable per call, and a missing one fails the call.
    PLATFORM_COMMAND = (
        "for k in maker product serial; do "
        'printf "%s=%s\\n" "$k" "$(kenv -q smbios.system.$k)"; done'
    )
    # -W: FreeBSD cuts long destinations to the column width otherwise.
    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"

    def _parse_listening(self, output: str, *, attributed: bool) -> list[dict]:
        return parse.sockstat(output)

    PKG_QUERY = "pkg query '%n\t%v\t%R\t%sb\t%c'"
    INSTALL_COMMAND = "pkg install -y {name}"
    UNINSTALL_COMMAND = "pkg delete -y {name}"
    # Root reads every daemon's pidfile; one-shot scripts have no status.
    SERVICE_STATUS_COMMAND = (
        "for s in $(service -e); do n=${s##*/}; "
        'printf "%s\\t%s\\n" "$n" "$(service $n status 2>&1 | head -1)"; done'
    )
    SERVICE_ACTION_COMMAND = "service {name} {action}"

    def _parse_services(self, output: str) -> list[dict]:
        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

    def get_packages(self) -> list[dict]:
        return parse.pkg_query(self._out(self.PKG_QUERY)) if self._has_pkg() else []

    def get_available_updates(self) -> list[dict]:
        """Package updates from ``pkg upgrade -n``, and fetched base-system updates.

        ``security`` comes from VuXML (``pkg audit``): True for a package it
        lists as vulnerable, False for one it does not, None when the audit
        could not run. On pkgbase the base system is packages from the
        FreeBSD-base repository; a classic base reports one entry (#799).
        """
        updates: list[dict] = []
        if self._has_pkg():
            updates = parse.pkg_upgrades(self._out("pkg upgrade -n", privileged=True, timeout=300))
            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()
            for update in updates:
                update["security"] = (update["name"] in vulnerable) if known else None
            if "FreeBSD-base" in self._out("pkg repos -l").split():
                return updates
        return updates + self._base_updates()

    def _base_updates(self) -> list[dict]:
        """A classic base system's updates, as freebsd-update fetched them.

        ``updatesready`` exits 0 when fetched updates wait to be installed and
        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
        )
        if ready.exit_code != 0:
            return []
        return [
            {
                "name": BASE_SYSTEM,
                "current_version": self._out("freebsd-version -u"),
                "new_version": "fetched by freebsd-update",
                "origin": "freebsd-update",
                "security": None,
            }
        ]

    def _os_version(self) -> str:
        version = super()._os_version()
        return f"FreeBSD {version}" if version else ""

    def _hardware(self) -> tuple[str, str, str]:
        smbios = self._platform()
        return smbios.get("maker", ""), smbios.get("product", ""), smbios.get("serial", "")