Merge pull request 'feat: read what a Linux kernel has built and loaded, once for every driver' (#4) from feat/kernel-facts into main

README.md

@@ -73,6 +73,12 @@ is a thin bundle over them — `PackageManagementMixin`, `HealthMetricsMixin`,
 `HostRebootMixin` (`reboot_host`) is mixed into `DeviceTypeDriver` itself, since any
 device may be restartable; like the others it only declares.
 
+`KernelFactsMixin` (`get_kernel_facts`) is the exception that is mixed in by a driver
+rather than by a role base: what a Linux kernel has built and loaded is read the same way
+everywhere, so the command and its parse are concrete here and a driver supplies only
+`_run_kernel_facts_command`. `OSDriver` does not carry it — a Windows host is an OS driver
+too, and `hasattr(driver, "get_kernel_facts")` has to stay truthful.
+
 A function class may use the **template form** — public method concrete, the
 device-specific part a `_hook` declared under `if TYPE_CHECKING` — *when the base
 genuinely does work* on the result: normalising, sorting, validating, or orchestrating

napalm_device_types/__init__.py

@@ -40,6 +40,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.health_metrics.HealthMetricsMixin`
 * :class:`~napalm_device_types.host_reboot.HostRebootMixin`
 * :class:`~napalm_device_types.interface_filter.InterfaceFilterMixin`
+* :class:`~napalm_device_types.kernel.KernelFactsMixin`
 * :class:`~napalm_device_types.mac_acl.MacAclMixin`
 * :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
 * :class:`~napalm_device_types.packages.PackageManagementMixin`
@@ -63,6 +64,7 @@ from napalm_device_types.firewall_rules import FirewallRuleMixin
 from napalm_device_types.health_metrics import HealthMetricsMixin
 from napalm_device_types.host_reboot import HostRebootMixin
 from napalm_device_types.interface_filter import InterfaceFilterMixin
+from napalm_device_types.kernel import KERNEL_FACTS_COMMAND, KernelFactsMixin, parse_kernel_facts
 from napalm_device_types.lag import add_lag_interfaces
 from napalm_device_types.mac_acl import MacAclMixin
 from napalm_device_types.media import MediaDriver
@@ -94,6 +96,9 @@ __all__ = [
     "NatVpnMixin",
     "OSDriver",
     "PackageManagementMixin",
+    "KernelFactsMixin",
+    "KERNEL_FACTS_COMMAND",
+    "parse_kernel_facts",
     "PhoneDriver",
     "PingSweepMixin",
     "PortSpec",

napalm_device_types/kernel.py

@@ -0,0 +1,168 @@
+# -*- coding: utf-8 -*-
+"""What the running kernel has built and loaded.
+
+A kernel CVE's exploitability often hangs on code that is simply not there --
+a module that is neither loaded nor shipped, a subsystem the kernel was built
+without. Reading that is identical on every Linux host, so the command and its
+parse live here once and a driver only carries the command across: SSH,
+an API's exec endpoint, whatever it has.
+
+The command is read-only and needs no privileges. It frames its report and
+sends it gzipped and base64-encoded, for two reasons: nothing in the payload can
+then look like a shell prompt to a screen-scraping transport, and a kernel's
+build configuration (~300 kB on a distribution kernel) crosses as a fifth of
+that.
+
+What the four lists mean for a module, and why "not loaded" alone is never
+"absent": a module that is not loaded can still be loaded on demand -- by an
+attacker too, where autoloading reaches it. Only a module that is neither
+loaded, nor compiled in, nor shipped for this kernel is one it cannot have.
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import gzip
+import zlib
+from typing import Dict, List, Optional, TYPE_CHECKING
+
+from napalm_device_types.models import KernelFactsDict
+
+_BEGIN = "KFACTS_BEGIN"
+_END = "KFACTS_END"
+
+#: One line, POSIX ``sh``, read-only. The frame markers are printed in two
+#: halves so that a transport which echoes the command does not show them early.
+KERNEL_FACTS_COMMAND = (
+    "r=$(uname -r); m=/lib/modules/$r; "
+    "printf '%s%s\\n' KFACTS_ BEGIN; "
+    "{ echo '[release]'; echo \"$r\"; "
+    "if [ -r /proc/modules ]; then echo '[loaded]'; cut -d' ' -f1 /proc/modules; fi; "
+    "if [ -r $m/modules.builtin ]; then echo '[builtin]'; cat $m/modules.builtin; fi; "
+    "if [ -r $m/modules.dep ]; then echo '[available]'; cut -d: -f1 $m/modules.dep; fi; "
+    "if [ -r /boot/config-$r ]; then echo '[config]'; grep '^CONFIG_' /boot/config-$r; "
+    "elif [ -r /proc/config.gz ]; then echo '[config]'; zcat /proc/config.gz | grep '^CONFIG_'; fi; "
+    "} 2>/dev/null | gzip -c | base64; "
+    "printf '%s%s\\n' KFACTS_ END"
+)
+
+
+def module_name(raw: str) -> str:
+    """A module as the kernel names it: no path, no ``.ko`` suffix, ``_`` for ``-``.
+
+    ``kernel/net/can/can-raw.ko.zst`` and ``can_raw`` are the same module; the
+    kernel itself treats dash and underscore alike.
+    """
+    base = raw.strip().rsplit("/", 1)[-1]
+    suffix = base.find(".ko")
+    if suffix != -1:
+        base = base[:suffix]
+    return base.replace("-", "_").lower()
+
+
+def _report(output: str) -> str:
+    lines = [line.strip() for line in output.splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no kernel facts in the output") from None
+    try:
+        packed = base64.b64decode("".join(lines[start + 1 : end]), validate=True)
+        return gzip.decompress(packed).decode()
+    except (binascii.Error, OSError, EOFError, zlib.error, UnicodeDecodeError) as exc:
+        raise ValueError(f"the kernel facts could not be decoded: {exc}") from exc
+
+
+def _sections(report: str) -> Dict[str, List[str]]:
+    sections: Dict[str, List[str]] = {}
+    current: Optional[List[str]] = None
+    for line in report.splitlines():
+        line = line.strip()
+        if line.startswith("[") and line.endswith("]"):
+            current = sections.setdefault(line[1:-1], [])
+        elif line and current is not None:
+            current.append(line)
+    return sections
+
+
+def _config(lines: List[str]) -> Dict[str, str]:
+    config: Dict[str, str] = {}
+    for line in lines:
+        option, sep, value = line.partition("=")
+        if not sep:
+            continue
+        if len(value) >= 2 and value[0] == value[-1] == '"':
+            value = value[1:-1]
+        config[option] = value
+    return config
+
+
+def parse_kernel_facts(output: str) -> KernelFactsDict:
+    """Parse what :data:`KERNEL_FACTS_COMMAND` printed.
+
+    A section the command did not print -- the file was missing or unreadable --
+    comes back ``None``, never empty.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_report(output))
+
+    def names(key: str) -> Optional[List[str]]:
+        if key not in sections:
+            return None
+        return sorted({module_name(line) for line in sections[key]})
+
+    release = sections.get("release") or [""]
+    return {
+        "release": release[0],
+        "loaded": names("loaded"),
+        "builtin": names("builtin"),
+        "available": names("available"),
+        "config": _config(sections["config"]) if "config" in sections else None,
+    }
+
+
+class KernelFactsMixin:
+    """Adds :meth:`get_kernel_facts` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the reading and its parse are
+    the same everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_kernel_facts_command` -- how a command reaches its host. Mixed in
+    by the drivers that can, not by :class:`~napalm_device_types.os.OSDriver`:
+    a Windows host is an OS driver too and has no Linux kernel to read, and
+    ``hasattr(driver, "get_kernel_facts")`` has to stay a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_kernel_facts_command(self, command: str) -> str:
+            """Run *command* on the host with ``sh`` and return what it printed."""
+            ...
+
+    def get_kernel_facts(self) -> KernelFactsDict:
+        """
+        Returns what the running kernel has built and loaded.
+
+        * release (string) - ``uname -r``
+        * loaded (list or None) - loaded modules, from ``/proc/modules``
+        * builtin (list or None) - modules compiled into the kernel image
+        * available (list or None) - modules shipped for this kernel
+        * config (dict or None) - the build configuration's set options
+
+        ``None`` means the source could not be read.
+
+        Example::
+
+            {
+                "release": "6.1.0-25-amd64",
+                "loaded": ["nf_tables", "tipc"],
+                "builtin": ["tcp_cubic"],
+                "available": ["can_raw", "nf_tables", "tipc"],
+                "config": {"CONFIG_TIPC": "m", "CONFIG_HZ": "250"},
+            }
+
+        :raises ValueError: if the host's output carried no intact report.
+        """
+        return parse_kernel_facts(self._run_kernel_facts_command(KERNEL_FACTS_COMMAND))

napalm_device_types/models.py

@@ -298,6 +298,24 @@ class NATTranslationDict(TypedDict):
     age: float
 
 
+class KernelFactsDict(TypedDict):
+    """What the running kernel has built and loaded (``KernelFactsMixin.get_kernel_facts``).
+
+    ``None`` means *could not be read*; an empty list means *read, and there is
+    nothing*. The difference is what lets a consumer say "this module cannot be
+    loaded on this kernel" rather than "we did not look".
+
+    Module names are normalised by :func:`napalm_device_types.kernel.module_name`:
+    no path, no ``.ko`` suffix, ``-`` folded to ``_``.
+    """
+
+    release: str  # uname -r
+    loaded: Optional[List[str]]  # /proc/modules
+    builtin: Optional[List[str]]  # modules.builtin -- compiled into the kernel image
+    available: Optional[List[str]]  # modules.dep -- shipped as loadable modules
+    config: Optional[Dict[str, str]]  # build configuration, set options only; quotes stripped
+
+
 class PortForwardDict(TypedDict):
     """A port the WAN side can reach, forwarded to a host inside.
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "2.0.0"
+version = "2.1.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_kernel_facts.py

@@ -0,0 +1,151 @@
+"""get_kernel_facts: what the running kernel has built and loaded.
+
+A kernel CVE's preconditions ask whether a module is loaded or a build option
+set. Reading that is the same on every Linux host -- one read-only command and
+its parse -- so both live here once, and a driver only carries the command
+across (#268 in netOrk).
+"""
+
+from __future__ import annotations
+
+import base64
+import gzip
+import os
+import subprocess
+
+import pytest
+
+from napalm_device_types import OSDriver
+from napalm_device_types.kernel import (
+    KERNEL_FACTS_COMMAND,
+    KernelFactsMixin,
+    module_name,
+    parse_kernel_facts,
+)
+
+REPORT = """[release]
+6.1.0-25-amd64
+[loaded]
+tipc
+nf_tables
+[builtin]
+kernel/net/ipv4/tcp_cubic.ko
+kernel/drivers/char/tpm/tpm-tis.ko
+[available]
+kernel/net/tipc/tipc.ko.xz
+kernel/net/can/can-raw.ko.zst
+kernel/net/netfilter/nf_tables.ko
+[config]
+CONFIG_TIPC=m
+CONFIG_BPF_JIT=y
+CONFIG_DEFAULT_HOSTNAME="(none)"
+CONFIG_HZ=250
+"""
+
+
+def _wire(report: str, *, noise: str = "") -> str:
+    """The report as the command prints it: framed, gzipped, base64 in lines."""
+    payload = base64.encodebytes(gzip.compress(report.encode())).decode()
+    return f"{noise}KFACTS_BEGIN\n{payload}KFACTS_END\n"
+
+
+class TestParsing:
+    def test_every_section_is_read(self):
+        facts = parse_kernel_facts(_wire(REPORT))
+
+        assert facts["release"] == "6.1.0-25-amd64"
+        assert facts["loaded"] == ["nf_tables", "tipc"]
+        assert facts["builtin"] == ["tcp_cubic", "tpm_tis"]
+        assert facts["available"] == ["can_raw", "nf_tables", "tipc"]
+        assert facts["config"] == {
+            "CONFIG_TIPC": "m",
+            "CONFIG_BPF_JIT": "y",
+            "CONFIG_DEFAULT_HOSTNAME": "(none)",
+            "CONFIG_HZ": "250",
+        }
+
+    def test_a_section_never_printed_is_none_not_empty(self):
+        """``None`` is "could not read"; an empty list would claim "read it,
+        and there is nothing" -- and that is what turns a module into
+        ``not_met`` downstream."""
+        facts = parse_kernel_facts(_wire("[release]\n6.1.0\n[loaded]\n"))
+
+        assert facts["loaded"] == []
+        assert facts["builtin"] is None
+        assert facts["available"] is None
+        assert facts["config"] is None
+
+    def test_whatever_surrounds_the_frame_is_ignored(self):
+        """A screen-scraping transport may echo the command or a banner."""
+        noise = "Last login: today\nprintf '%s%s\\n' KFACTS_ BEGIN; ...\n"
+
+        assert parse_kernel_facts(_wire(REPORT, noise=noise))["release"] == "6.1.0-25-amd64"
+
+    def test_output_without_the_frame_raises(self):
+        with pytest.raises(ValueError):
+            parse_kernel_facts("sh: gzip: not found\n")
+
+    def test_a_damaged_payload_raises(self):
+        with pytest.raises(ValueError):
+            parse_kernel_facts("KFACTS_BEGIN\nnot base64 at all!\nKFACTS_END\n")
+
+
+class TestModuleNames:
+    @pytest.mark.parametrize(
+        "raw, name",
+        [
+            ("tipc", "tipc"),
+            ("kernel/net/tipc/tipc.ko", "tipc"),
+            ("kernel/net/tipc/tipc.ko.zst", "tipc"),
+            ("kernel/net/can/can-raw.ko.xz", "can_raw"),
+            ("CAN-RAW", "can_raw"),
+            ("  nf_tables  ", "nf_tables"),
+        ],
+    )
+    def test_dash_and_underscore_are_one_name(self, raw, name):
+        """The kernel treats ``-`` and ``_`` in module names as the same."""
+        assert module_name(raw) == name
+
+
+class TestTheCommand:
+    def test_the_frame_is_not_in_the_command_itself(self):
+        """An echoing transport prints the command back; the markers must only
+        appear once the command has run."""
+        assert "KFACTS_BEGIN" not in KERNEL_FACTS_COMMAND
+        assert "KFACTS_END" not in KERNEL_FACTS_COMMAND
+
+    def test_it_writes_nothing(self):
+        for verb in ("modprobe", "insmod", "rmmod", "sudo", " > ", ">>"):
+            assert verb not in KERNEL_FACTS_COMMAND
+
+    @pytest.mark.skipif(os.uname().sysname != "Linux", reason="reads a Linux kernel")
+    def test_it_runs_and_parses_on_this_host(self):
+        out = subprocess.run(
+            ["sh", "-c", KERNEL_FACTS_COMMAND], capture_output=True, text=True, timeout=60
+        ).stdout
+
+        facts = parse_kernel_facts(out)
+
+        assert facts["release"] == os.uname().release
+        assert facts["loaded"] is None or all(isinstance(m, str) for m in facts["loaded"])
+
+
+class TestTheTemplate:
+    def test_a_driver_supplies_only_the_transport(self):
+        class Driver(KernelFactsMixin):
+            def _run_kernel_facts_command(self, command: str) -> str:
+                self.sent = command
+                return _wire(REPORT)
+
+        driver = Driver()
+        facts = driver.get_kernel_facts()
+
+        assert driver.sent == KERNEL_FACTS_COMMAND
+        assert facts["release"] == "6.1.0-25-amd64"
+
+    def test_not_every_os_driver_has_it(self):
+        """A Windows host is an OSDriver too, and has no Linux kernel to read:
+        ``hasattr`` has to stay a truthful answer, so the drivers that can mix
+        this in themselves."""
+        assert not issubclass(OSDriver, KernelFactsMixin)
+        assert not hasattr(OSDriver, "get_kernel_facts")