"""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 ipaddress
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 EnsurePackageMixin, OSDriver, SnmpAgentMixin
from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict
from napalm_device_types.host_status import HostStatusMixin
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_.+-]*$")

#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
_SNMP_WORD = re.compile(r'^[^\s"#]+$')
#: The SSH client side of this connection: the address the host sees it come from.
_PEER_COMMAND = 'printf "%s\\n" "$SSH_CONNECTION"'
#: Why ``run_command`` refuses stdin to a privileged command while a sudo
#: password is set (#8).
_PRIVILEGED_STDIN = (
    "run_command(privileged=True) takes no stdin while a sudo password is set: "
    "sudo -S reads the password from stdin, and a sudo that does not prompt "
    "(NOPASSWD, a cached timestamp) leaves it for the command, which would read it "
    "as the first line of the data. Stage the data without privilege and have root "
    "install that file."
)

#: Stages content without privilege: a fresh ``mktemp`` file only the login user
#: may read, never a fixed name another user could create first. Prints its path.
_STAGE_TEMPLATE = "/tmp/napalm-bsd-stage.XXXXXX"
_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
_STAGE = (
    f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
    '{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
)

def _source(value: str) -> str:
    """An allowed source as an address or network, or ValueError."""
    return str(ipaddress.ip_network(value.strip(), strict=False))


def render_snmpd_conf(settings: SnmpAgentSettingsDict) -> str:
    """net-snmp's configuration for *settings*: read-only v2c, per allowed source."""
    community = settings["community"]
    if not _SNMP_WORD.match(community):
        raise ValueError("the community must be one word without quotes or #")
    listen = settings.get("listen")
    lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"]
    sources = [_source(s) for s in settings.get("allowed_sources", [])]
    lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"]
    for key, directive in (("location", "sysLocation"), ("contact", "sysContact")):
        value = settings.get(key, "")
        if value:
            if "\n" in value:
                raise ValueError(f"{key} must be one line")
            lines.append(f"{directive}    {value}")
    return "\n".join(lines) + "\n"


class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, 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.

        :raises ValueError: for *stdin* to a privileged command while a sudo
            password is set (see ``_PRIVILEGED_STDIN``).
        """
        line, prefix = self._privileged(command, privileged)
        if prefix and stdin is not None:
            raise ValueError(_PRIVILEGED_STDIN)
        return run_on_transport(self._transport(), line, stdin=prefix or stdin, timeout=timeout)

    def _install_file_as_root(self, path: str, content: str) -> CommandResult:
        """Replace *path* with *content*, as a new file owned by root, mode 644.

        The content is staged without privilege (``_STAGE``), so it never follows
        a sudo password on stdin. Root then installs the staged file next to *path*
        and renames it over *path*; nothing root runs follows a symlink at *path*.
        Nothing runs as root when staging failed.
        """
        staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
        lines = staged.stdout.strip().splitlines()
        tmp = lines[-1] if lines else ""
        if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]:
            reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
            return CommandResult(reason, "", staged.exit_code or 1)
        tmp_q, part_q, path_q = (quote(p) for p in (tmp, f"{path}.napalm-tmp", path))
        directory = quote(path.rsplit("/", 1)[0] or "/")
        return self.run_command(
            f"mkdir -p {directory} && install -m 644 -o root -g wheel {tmp_q} {part_q}"
            f" && mv -f {part_q} {path_q}; s=$?; rm -f {tmp_q} {part_q}; exit $s",
            privileged=True,
            timeout=30,
        )

    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 _read(
        self,
        command: str,
        *,
        privileged: bool = False,
        timeout: float = 60,
        ok: tuple[int, ...] = (0,),
    ) -> CommandResult:
        """Run a reader's command; raise when it did not read.

        A reader that cannot read raises rather than return "nothing": an empty
        update list would close every patch clock netOrk keeps (napalm-bsd#4).
        """
        result = self.run_command(command, privileged=privileged, timeout=timeout)
        if result.exit_code not in ok:
            reason = (result.stderr or result.stdout).strip() or f"exit {result.exit_code}"
            raise RuntimeError(f"{command}: {reason}")
        return result

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

    #: A file that exists until the host's first boot is over; "" when there is none to ask.
    FIRST_BOOT_MARKER = ""

    def first_boot_pending(self) -> bool:
        """Whether the host is still in its first boot.

        A FreeBSD cloud image upgrades its base system on the first boot, starts
        sshd only after that and restarts right away; ``/etc/rc`` removes
        ``/firstboot`` just before the restart. netOrk waits for this before it
        sets up a new VM (NetOrk/netork#795).
        """
        if not self.FIRST_BOOT_MARKER:
            return False
        return self.run_command(f"test -e {self.FIRST_BOOT_MARKER}").exit_code == 0

    def _run_host_status_command(self, command: str) -> str:
        """The transport for ``HostStatusMixin.get_host_status``: read-only, no root.

        On FreeBSD the report compares ``freebsd-version -k`` with ``-r``
        (napalm-device-types 4.1); elsewhere "reboot required" stays unknown.
        """
        return self.run_command(command).stdout

    # -- 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}

    # -- SNMP (net-snmp from packages, NetOrk/netork#800) -------------------------

    #: Where net-snmp reads its configuration, and how its daemon is started.
    SNMPD_CONF = ""
    SNMPD_START = ""
    #: net-snmp's daemon; OpenBSD's base snmpd is /usr/sbin/snmpd.
    SNMPD_DAEMON = "/usr/local/sbin/snmpd"

    def get_management_peer(self) -> Optional[str]:
        """The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
        out = self.run_command(_PEER_COMMAND, timeout=15).stdout.split()
        try:
            return str(ipaddress.ip_address(out[0])) if out else None
        except ValueError:
            return None

    # The steps of napalm-device-types' configure_snmp_agent: net-snmp rather than
    # the base daemons (FreeBSD bsnmpd, OpenBSD snmpd), as it answers UCD-SNMP-MIB.

    def _step(self, command: str, done: str) -> FixStepOutcomeDict:
        result = self.run_command(command, privileged=True, timeout=600)
        output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
        return {"ok": result.exit_code == 0, "output": done if result.exit_code == 0 else output[-300:]}

    def _install_snmp_agent(self) -> FixStepOutcomeDict:
        return self._step(self.INSTALL_COMMAND.format(name="net-snmp"), "net-snmp installed")

    def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
        written = self._install_file_as_root(self.SNMPD_CONF, render_snmpd_conf(settings))
        if written.exit_code:
            return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
        return {"ok": True, "output": f"wrote {self.SNMPD_CONF}"}

    def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
        # pf and ipfw rule sets are the operator's; the agent itself answers only
        # the allowed sources (one rocommunity line each).
        return {"ok": True, "output": "pf/ipfw unchanged; snmpd answers only the allowed sources"}

    def _start_snmp_agent(self) -> FixStepOutcomeDict:
        return self._step(self.SNMPD_START, "snmpd enabled and running")

    def get_snmp_config(self) -> Optional[dict[str, Any]]:
        """net-snmp's community and port, if its daemon runs; None otherwise."""
        if not self._out(f"pgrep -f {self.SNMPD_DAEMON}"):
            return None
        # A community that is not in the file is not known; no default is assumed.
        community, port = "", 161
        for line in self._out(f"cat {self.SNMPD_CONF}").splitlines():
            words = line.split()
            if len(words) >= 2 and words[0].lower() in (
                "rocommunity",
                "rwcommunity",
                "rocommunity6",
            ):
                community = words[1]
            elif words and words[0] == "agentAddress":
                found = re.search(r":(\d+)", line)
                port = int(found[1]) if found else port
        return {"running": True, "community": community, "port": port, "version": "2c"}

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

    def get_os_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),
        ]