diff --git a/README.md b/README.md index 86642b8..12cbb7d 100644 --- a/README.md +++ b/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 diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index f431060..37b1686 100644 --- a/napalm_device_types/__init__.py +++ b/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", diff --git a/napalm_device_types/kernel.py b/napalm_device_types/kernel.py new file mode 100644 index 0000000..5fd019c --- /dev/null +++ b/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)) diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index 0d20ce7..349dba3 100644 --- a/napalm_device_types/models.py +++ b/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. diff --git a/pyproject.toml b/pyproject.toml index 32eab9d..9ad75c2 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/test_kernel_facts.py b/tests/test_kernel_facts.py new file mode 100644 index 0000000..c402676 --- /dev/null +++ b/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")