napalm-bsd/napalm_bsd/base.py

"""What FreeBSD and OpenBSD share: SSH, and the readers whose tools agree.

Every command runs on its own SSH exec channel (napalm-device-types'
``run_on_transport``): no PTY to parse a prompt from, stdout and stderr apart,
and a real exit status. Root comes from ``sudo -S`` with the sudo password on
stdin, as napalm-linux does it.
"""

from __future__ import annotations

import logging
import re
import socket
from shlex import quote
from typing import Any, Optional

import paramiko
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
from napalm_device_types import OSDriver
from napalm_device_types.channel import (
    ByteStream,
    CommandResult,
    open_stream_on_transport,
    run_on_transport,
)

from napalm_bsd import parse

logger = logging.getLogger(__name__)

#: The update-list entry for base-system patches (OpenBSD syspatch, classic
#: FreeBSD freebsd-update): one entry, as they are applied together (#799).
BASE_SYSTEM = "base-system"

_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_.+-]*$")


class BsdDriver(OSDriver):
    """Base for the BSD drivers; a concrete one names the commands that differ."""

    VENDOR = ""
    #: A full init sequence, as on any general-purpose host.
    REBOOT_SETTLE_SECONDS = 90

    #: Commands whose wording differs between the BSDs.
    HOSTNAME_COMMAND = "hostname"
    OS_VERSION_COMMAND = "uname -sr"
    KERNEL_COMMAND = "uname -r"
    #: Prints ``key=value`` lines; see :meth:`_platform`.
    PLATFORM_COMMAND = ""
    ROUTES_COMMAND = "netstat -rn"
    #: Lists listening sockets with their processes; as root it names them all.
    LISTENING_COMMAND = ""
    #: Lists them without root, if the command above needs it to see them all.
    LISTENING_FALLBACK_COMMAND = ""
    #: lstart is five words; :func:`parse.processes` reads it as one field.
    PS_COMMAND = "ps -axww -o user,pid,ppid,%cpu,%mem,vsz,rss,stat,lstart,command"

    def __init__(
        self,
        hostname: str,
        username: str,
        password: str,
        timeout: int = 60,
        optional_args: Optional[dict] = None,
    ) -> None:
        self.hostname = hostname
        self.username = username
        self.password = password
        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._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

    # -- connection --------------------------------------------------------------

    def open(self) -> None:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        try:
            client.connect(
                self.hostname,
                port=self.port,
                username=self.username,
                password=self.password or None,
                key_filename=self._key_file,
                allow_agent=self._allow_agent,
                look_for_keys=self._look_for_keys,
                timeout=self.timeout,
            )
        except (paramiko.SSHException, socket.error) as exc:
            client.close()
            raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc
        self._client = client
        self._root = None

    def close(self) -> None:
        if self._client:
            self._client.close()
            self._client = None

    def is_alive(self) -> dict[str, bool]:
        transport = self._client.get_transport() if self._client else None
        return {"is_alive": bool(transport and transport.is_active())}

    def _transport(self) -> Any:
        transport = self._client.get_transport() if self._client else None
        if transport is None or not transport.is_active():
            raise ConnectionClosedException("Not connected")
        return transport

    # -- command channel (napalm-device-types CommandChannelMixin) ---------------

    def _is_root(self) -> bool:
        if self._root is None:
            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]]:
        """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
        appears in a process list; without one, ``sudo -n`` fails at once
        where a prompt would hang.
        """
        if not privileged or self._is_root():
            return command, None
        if self._sudo_password:
            return f"sudo -S -p '' sh -c {quote(command)}", f"{self._sudo_password}\n".encode()
        return f"sudo -n sh -c {quote(command)}", None

    def run_command(
        self,
        command: str,
        *,
        privileged: bool = False,
        timeout: float = 60,
        stdin: Optional[bytes] = None,
    ) -> CommandResult:
        """Run *command* on an exec channel: no PTY, stderr apart, a real exit code."""
        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)

    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 _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()

    # -- facts -------------------------------------------------------------------

    def _platform(self) -> dict[str, str]:
        """``key=value`` lines from :attr:`PLATFORM_COMMAND`."""
        if not self.PLATFORM_COMMAND:
            return {}
        pairs = (line.partition("=") for line in self._out(self.PLATFORM_COMMAND).splitlines())
        return {key.strip(): value.strip() for key, sep, value in pairs if sep}

    def _hardware(self) -> tuple[str, str, str]:
        """``(vendor, model, serial)``; each concrete driver reads its own source."""
        return "", "", ""

    def _os_version(self) -> str:
        return self._out(self.OS_VERSION_COMMAND)

    def _uptime(self) -> int:
        booted = parse.boottime(self._out("sysctl -n kern.boottime"))
        now = self._out("date +%s")
        if booted is None or not now.isdigit():
            return -1
        return max(int(now) - booted, 0)

    def get_facts(self) -> dict[str, Any]:
        fqdn = self._out(self.HOSTNAME_COMMAND)
        vendor, model, serial = self._hardware()
        interfaces = parse.ifconfig(self._out("ifconfig -a"))
        return {
            "hostname": fqdn.split(".")[0],
            "fqdn": fqdn,
            "vendor": vendor or self.VENDOR,
            "model": model,
            "serial_number": serial,
            "os_version": self._os_version(),
            "uptime": self._uptime(),
            "interface_list": [n for n, i in interfaces.items() if not i["loopback"]],
            "running_kernel": self._out(self.KERNEL_COMMAND),
        }

    # -- interfaces, routes, neighbours ------------------------------------------

    def get_interfaces(self) -> dict[str, Any]:
        return {
            name: {
                "is_up": iface["up"],
                "is_enabled": iface["enabled"],
                "description": iface["description"],
                "last_flapped": -1.0,
                "speed": iface["speed"],
                "mtu": iface["mtu"],
                "mac_address": iface["mac"],
            }
            for name, iface in parse.ifconfig(self._out("ifconfig -a")).items()
        }

    def get_interfaces_ip(self) -> dict[str, Any]:
        result: dict[str, Any] = {}
        for name, iface in parse.ifconfig(self._out("ifconfig -a")).items():
            if not (iface["ipv4"] or iface["ipv6"]):
                continue
            result[name] = {
                family: {addr: {"prefix_length": prefix} for addr, prefix in iface[family].items()}
                for family in ("ipv4", "ipv6")
            }
        return result

    def get_route_to(
        self, destination: str = "", protocol: str = "", longer: bool = False
    ) -> dict[str, list[dict[str, Any]]]:
        """The routing table, keyed by network, in NAPALM's route shape."""
        routes: dict[str, list[dict[str, Any]]] = {}
        for route in parse.routes(self._out(self.ROUTES_COMMAND)):
            if destination and route["network"] != destination:
                continue
            if protocol and route["protocol"] != protocol.lower():
                continue
            routes.setdefault(route["network"], []).append(
                {
                    "protocol": route["protocol"],
                    "family": route["family"],
                    "current_active": True,
                    "last_active": False,
                    "age": -1,
                    "next_hop": route["next_hop"],
                    "outgoing_interface": route["interface"],
                    "selected_next_hop": True,
                    "preference": 0,
                    "routing_table": "global",
                    "protocol_attributes": {},
                }
            )
        return routes

    def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
        return parse.arp(self._out("arp -an"))

    def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
        """Neither BSD ships an LLDP daemon in its base system."""
        return {}

    # -- listening sockets -------------------------------------------------------

    def _parse_listening(self, output: str, *, attributed: bool) -> list[dict[str, Any]]:
        """Parse :attr:`LISTENING_COMMAND` (*attributed*) or its fallback."""
        raise NotImplementedError

    def get_listening_sockets(self) -> dict[str, Any]:
        """Every listening TCP and bound UDP socket, with the process holding it.

        Same shape as napalm-device-types' ``ListeningSocketsMixin`` -- whose
        ``ss``/cgroup reading is Linux's -- and the same rule: read as root
        first, and without root when that brings nothing back, which the
        reading then says (``attributed: False``). BSD has no systemd units or
        container IDs to name.
        """
        privileged = self.run_command(self.LISTENING_COMMAND, privileged=True)
        if privileged.exit_code == 0 and privileged.stdout.strip():
            return {
                "attributed": True,
                "sockets": self._parse_listening(privileged.stdout, attributed=True),
            }
        command = self.LISTENING_FALLBACK_COMMAND or self.LISTENING_COMMAND
        plain = self.run_command(command, privileged=False)
        return {
            "attributed": False,
            "sockets": self._parse_listening(plain.stdout, attributed=False),
        }

    # -- packages -----------------------------------------------------------------

    #: ``{name}`` is the package (with its version, when one is asked for).
    INSTALL_COMMAND = ""
    UNINSTALL_COMMAND = ""

    def _package_command(self, template: str, name: str) -> None:
        """Run a package command as root; a failure raises with what it printed."""
        result = self.run_command(template.format(name=name), privileged=True, timeout=600)
        if result.exit_code != 0:
            raise RuntimeError(
                (result.stderr or result.stdout).strip() or f"{name}: exit {result.exit_code}"
            )

    def install_package(self, name: str, version: str = "") -> None:
        if not _PACKAGE_NAME.match(name) or (version and not _PACKAGE_NAME.match(version)):
            raise ValueError(f"Not a package name: {name!r}")
        self._package_command(self.INSTALL_COMMAND, f"{name}-{version}" if version else name)

    def uninstall_package(self, name: str) -> None:
        if not _PACKAGE_NAME.match(name):
            raise ValueError(f"Not a package name: {name!r}")
        self._package_command(self.UNINSTALL_COMMAND, name)

    # -- services -----------------------------------------------------------------

    #: Prints ``name<TAB>status`` per enabled service; parsed by :meth:`_parse_services`.
    SERVICE_STATUS_COMMAND = ""
    #: ``{name}`` and ``{action}`` (start, stop, restart, enable, disable).
    SERVICE_ACTION_COMMAND = ""

    def _parse_services(self, output: str) -> list[dict[str, Any]]:
        raise NotImplementedError

    def get_services(self) -> list[dict[str, Any]]:
        """The enabled services, whether each runs, and its PID where known.

        Read as root where the status needs it, and without root otherwise.
        """
        result = self.run_command(self.SERVICE_STATUS_COMMAND, privileged=True, timeout=120)
        if result.exit_code != 0 or not result.stdout.strip():  # no root: sudo refused
            result = self.run_command(self.SERVICE_STATUS_COMMAND, timeout=120)
        return self._parse_services(result.stdout)

    def manage_service(self, name: str, action: str) -> dict[str, Any]:
        """Applies *action* to the service *name*, as root.

        :returns: ``{"success": bool, "output": str}``
        :raises ValueError: for an unknown action or an invalid name, before
            anything is sent.
        """
        if action not in _SERVICE_ACTIONS or not _SERVICE_NAME.match(name):
            raise ValueError(f"Cannot {action!r} service {name!r}")
        command = self.SERVICE_ACTION_COMMAND.format(name=name, action=action)
        result = self.run_command(command, privileged=True, timeout=120)
        output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
        return {"success": result.exit_code == 0, "output": output}

    # -- accounts, processes, cron -----------------------------------------------

    def get_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]]:
        return parse.processes(self._out(self.PS_COMMAND))

    def get_cron_jobs(self) -> list[dict[str, str]]:
        """The system crontab and the login user's own.

        Other users' crontabs (``/var/cron/tabs``) are root's to read.
        """
        return [
            *parse.crontab(self._out("cat /etc/crontab"), system=True),
            *parse.crontab(self._out("crontab -l"), user=self.username),
        ]