feat: FreeBSD and OpenBSD drivers, read side

FreeBSDDriver (`freebsd`) and OpenBSDDriver (`openbsd`) on a shared
BsdDriver, for the BSD VMs netOrk provisions and BSD hosts added by hand
(NetOrk/netork#798).

- SSH over exec channels (napalm-device-types' run_on_transport): no PTY,
  stdout and stderr apart, real exit codes; run_command()/open_stream()
  are public. Root through `sudo -S` with the password on stdin, or
  `sudo -n` without one.
- Facts: hostname, release and running kernel (freebsd-version / uname),
  hardware from SMBIOS (kenv) or hw.* sysctls, uptime from kern.boottime.
- Interfaces and addresses (ifconfig -a: ether/lladdr, hex netmasks,
  scoped IPv6), routes (netstat -rn, OpenBSD's abbreviated networks
  written out, host/broadcast/loopback entries left out), ARP (FreeBSD's
  sentences, OpenBSD's table), users and groups, processes (lstart read as
  one five-word field), cron jobs (system crontab and the login user's).
- Parsers are pure functions, tested on output recorded from FreeBSD 15.1
  (hand-installed and cloud image) and OpenBSD 7.9 (NetOrk/netork#793),
  with internal addresses replaced by documentation ranges.

napalm_bsd/__init__.py

@@ -0,0 +1,6 @@
+"""NAPALM drivers for BSD hosts: FreeBSD and OpenBSD, over SSH."""
+
+from napalm_bsd.freebsd import FreeBSDDriver
+from napalm_bsd.openbsd import OpenBSDDriver
+
+__all__ = ["FreeBSDDriver", "OpenBSDDriver"]

napalm_bsd/base.py

@@ -0,0 +1,264 @@
+"""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 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__)
+
+
+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"
+    #: 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 {}
+
+    # -- 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),
+        ]

napalm_bsd/freebsd.py

@@ -0,0 +1,38 @@
+"""FreeBSD: kenv for the hardware, freebsd-version for base and kernel."""
+
+from __future__ import annotations
+
+from napalm_device_types import FingerprintRule
+
+from napalm_bsd.base import 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"
+
+    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", "")

napalm_bsd/openbsd.py

@@ -0,0 +1,30 @@
+"""OpenBSD: hw.* sysctls for the hardware, uname for release and kernel."""
+
+from __future__ import annotations
+
+from napalm_device_types import FingerprintRule
+
+from napalm_bsd.base import 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'
+    )
+
+    def _hardware(self) -> tuple[str, str, str]:
+        hw = self._platform()
+        return hw.get("vendor", ""), hw.get("product", ""), hw.get("serialno", "")

napalm_bsd/parse.py

@@ -0,0 +1,259 @@
+"""Parsers for the output of BSD commands. Pure functions, no I/O.
+
+FreeBSD and OpenBSD share most tools; where their output differs (``arp``,
+``kern.boottime``, how ``netstat`` abbreviates networks, ``ether`` vs
+``lladdr``) one parser reads both.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import re
+from typing import Any, Optional
+
+_IFACE_HEADER = re.compile(
+    r"^(?P<name>[^\s:]+): flags=[0-9a-f]+<(?P<flags>[^>]*)>(?:.*?\bmtu (?P<mtu>\d+))?"
+)
+_SPEED = re.compile(r"\((\d+)(G?)base", re.IGNORECASE)
+
+
+def _prefix_from_netmask(mask: str) -> int:
+    """``0xffffff00`` (as ifconfig prints it) or ``255.255.255.0`` -> 24."""
+    value = int(mask, 16) if mask.startswith("0x") else int(ipaddress.IPv4Address(mask))
+    return bin(value).count("1")
+
+
+def ifconfig(text: str) -> dict[str, dict[str, Any]]:
+    """``ifconfig -a`` -> per interface: flags, MTU, MAC, state, speed, addresses."""
+    interfaces: dict[str, dict[str, Any]] = {}
+    current: Optional[dict[str, Any]] = None
+    for line in text.splitlines():
+        header = _IFACE_HEADER.match(line)
+        if header:
+            flags = set(filter(None, header["flags"].split(",")))
+            current = {
+                "flags": flags,
+                "enabled": "UP" in flags,
+                "up": "UP" in flags and "RUNNING" in flags,
+                "loopback": "LOOPBACK" in flags,
+                "mtu": int(header["mtu"] or 0),
+                "mac": "",
+                "description": "",
+                "speed": -1.0,
+                "ipv4": {},
+                "ipv6": {},
+            }
+            interfaces[header["name"]] = current
+            continue
+        if current is None:
+            continue
+        words = line.split()
+        if not words:
+            continue
+        key = words[0]
+        if key in ("ether", "lladdr") and len(words) > 1:
+            current["mac"] = words[1].lower()
+        elif key == "inet" and len(words) > 3 and words[2] == "netmask":
+            current["ipv4"][words[1]] = _prefix_from_netmask(words[3])
+        elif key == "inet6" and len(words) > 3 and words[2] == "prefixlen":
+            current["ipv6"][words[1].split("%")[0]] = int(words[3])
+        elif key == "description:":
+            current["description"] = line.split("description:", 1)[1].strip()
+        elif key == "status:":
+            if line.split("status:", 1)[1].strip() == "no carrier":
+                current["up"] = False
+        elif key == "media:":
+            speed = _SPEED.search(line)
+            if speed:
+                current["speed"] = float(int(speed[1]) * (1000 if speed[2] else 1))
+    return interfaces
+
+
+def _network(destination: str, family: str) -> Optional[str]:
+    """A netstat destination as a network: ``default``, ``10.0.2/24``, ``fe80::%em0/64``."""
+    if destination == "default":
+        return "0.0.0.0/0" if family == "ipv4" else "::/0"
+    address, slash, prefix = destination.partition("/")
+    address = address.split("%")[0]
+    if family == "ipv4":
+        octets = address.split(".")
+        address = ".".join(octets + ["0"] * (4 - len(octets)))
+    try:
+        return str(ipaddress.ip_network(f"{address}/{prefix}" if slash else address, strict=False))
+    except ValueError:
+        return None
+
+
+def routes(text: str) -> list[dict[str, str]]:
+    """``netstat -rn`` -> the routes, without host, broadcast and loopback entries.
+
+    FreeBSD prints ``Netif``, OpenBSD ``Iface``; the column is found from the
+    header. A route through a gateway (``G``) has a next hop, any other is
+    connected; ``S`` marks a static one.
+    """
+    result: list[dict[str, str]] = []
+    family = ""
+    iface_column: Optional[int] = None
+    for line in text.splitlines():
+        words = line.split()
+        if not words:
+            continue
+        if words[0] in ("Internet:", "Internet6:"):
+            family = "ipv4" if words[0] == "Internet:" else "ipv6"
+            continue
+        if words[0] == "Destination":
+            names = [w for w in words if w in ("Netif", "Iface")]
+            iface_column = words.index(names[0]) if names else None
+            continue
+        if not family or iface_column is None or len(words) <= iface_column:
+            continue
+        destination, gateway, flags, iface = words[0], words[1], words[2], words[iface_column]
+        if iface.startswith("lo") or ("H" in flags and "G" not in flags) or "b" in flags:
+            continue
+        network = _network(destination, family)
+        if network is None or ipaddress.ip_network(network).is_multicast:
+            continue
+        gateway_route = "G" in flags
+        result.append(
+            {
+                "network": network,
+                "next_hop": gateway if gateway_route else "",
+                "interface": iface,
+                "protocol": ("static" if "S" in flags else "dynamic")
+                if gateway_route
+                else "connected",
+                "family": family,
+            }
+        )
+    return result
+
+
+_DURATION = re.compile(r"(?:(\d+)h)?(?:(\d+)m)?(?:(\d+)s)?$")
+
+
+def duration(text: str) -> float:
+    """OpenBSD's ``1h2m3s`` in seconds; -1.0 for anything else (``permanent``)."""
+    match = _DURATION.match(text)
+    if not text or not match or not any(match.groups()):
+        return -1.0
+    hours, minutes, seconds = (int(g or 0) for g in match.groups())
+    return float(hours * 3600 + minutes * 60 + seconds)
+
+
+_FREEBSD_ARP = re.compile(
+    r"^\S+ \((?P<ip>[\d.]+)\) at (?P<mac>[0-9a-f:]{17}) on (?P<iface>\S+)"
+    r"(?: expires in (?P<expires>\d+) seconds)?"
+)
+
+
+def arp(text: str) -> list[dict[str, Any]]:
+    """``arp -an`` -> ARP entries; FreeBSD prints sentences, OpenBSD a table."""
+    entries: list[dict[str, Any]] = []
+    for line in text.splitlines():
+        sentence = _FREEBSD_ARP.match(line)
+        if sentence:
+            expires = sentence["expires"]
+            entries.append(
+                {
+                    "interface": sentence["iface"],
+                    "mac": sentence["mac"],
+                    "ip": sentence["ip"],
+                    "age": float(expires) if expires else -1.0,
+                }
+            )
+            continue
+        words = line.split()
+        if len(words) >= 4 and re.fullmatch(r"[0-9a-f:]{17}", words[1]):
+            entries.append(
+                {"interface": words[2], "mac": words[1], "ip": words[0], "age": duration(words[3])}
+            )
+    return entries
+
+
+def users(passwd: str, group: str) -> list[dict[str, Any]]:
+    """``/etc/passwd`` plus the supplementary groups ``/etc/group`` lists."""
+    memberships: dict[str, list[str]] = {}
+    for line in group.splitlines():
+        parts = line.split(":")
+        if len(parts) < 4 or line.startswith("#"):
+            continue
+        for member in filter(None, (m.strip() for m in parts[3].split(","))):
+            memberships.setdefault(member, []).append(parts[0])
+    result: list[dict[str, Any]] = []
+    for line in passwd.splitlines():
+        parts = line.split(":")
+        if len(parts) < 7 or line.startswith("#"):
+            continue
+        name, _, uid, gid, _, home, shell = parts[:7]
+        if not (uid.isdigit() and gid.isdigit()):
+            continue
+        result.append(
+            {
+                "username": name,
+                "uid": int(uid),
+                "gid": int(gid),
+                "home": home,
+                "shell": shell,
+                "groups": memberships.get(name, []),
+            }
+        )
+    return result
+
+
+_PROCESS = re.compile(
+    r"^(?P<user>\S+)\s+(?P<pid>\d+)\s+(?P<ppid>\d+)\s+(?P<cpu>[\d.]+)\s+(?P<mem>[\d.]+)\s+"
+    r"(?P<vsz>\d+)\s+(?P<rss>\d+)\s+(?P<stat>\S+)\s+"
+    r"(?P<started>\w{3}\s+\w{3}\s+\d+\s+\d\d:\d\d:\d\d\s+\d{4})\s+(?P<command>.*)$"
+)
+
+
+def processes(text: str) -> list[dict[str, Any]]:
+    """``ps -axww -o user,pid,ppid,%cpu,%mem,vsz,rss,stat,lstart,command``.
+
+    ``lstart`` is five words; it is matched as one field.
+    """
+    result: list[dict[str, Any]] = []
+    for line in text.splitlines():
+        match = _PROCESS.match(line)
+        if not match:
+            continue
+        result.append(
+            {
+                "pid": int(match["pid"]),
+                "ppid": int(match["ppid"]),
+                "user": match["user"],
+                "cpu": float(match["cpu"]),
+                "memory": float(match["mem"]),
+                "vsz": int(match["vsz"]),
+                "rss": int(match["rss"]),
+                "tty": "",
+                "state": match["stat"][0],
+                "started": match["started"],
+                "command": match["command"].strip(),
+            }
+        )
+    return result
+
+
+def crontab(text: str, *, system: bool = False, user: str = "") -> list[dict[str, str]]:
+    """A crontab -> its jobs. A system crontab names the user in the sixth field."""
+    jobs: list[dict[str, str]] = []
+    for line in text.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or re.match(r"^[A-Za-z_][A-Za-z0-9_]*\s*=", line):
+            continue
+        fields = 1 if line.startswith("@") else 5
+        parts = line.split(None, fields + (1 if system else 0))
+        if len(parts) < fields + (2 if system else 1):
+            continue
+        schedule = " ".join(parts[:fields])
+        owner = parts[fields] if system else user
+        command = parts[-1]
+        jobs.append({"user": owner, "schedule": schedule, "command": command})
+    return jobs
+
+
+def boottime(text: str) -> Optional[int]:
+    """``sysctl -n kern.boottime``: FreeBSD ``{ sec = N, … }``, OpenBSD ``N``."""
+    match = re.search(r"sec = (\d+)", text) or re.fullmatch(r"\s*(\d+)\s*", text)
+    return int(match[1]) if match else None