diff --git a/bigbrother.py b/bigbrother.py index 92adb50..d86bcfb 100755 --- a/bigbrother.py +++ b/bigbrother.py @@ -12,6 +12,9 @@ import signal import logging import inspect import importlib +import json +import sqlite3 +from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Type @@ -50,25 +53,28 @@ logger = logging.getLogger("bb.cli") # PID file for daemon mode PID_FILE = "/tmp/.bb.pid" +# All module categories to discover +MODULE_CATEGORIES = ["stealth", "passive", "active", "intel", "connectivity"] + # --------------------------------------------------------------------------- # Module discovery # --------------------------------------------------------------------------- -def discover_stealth_modules() -> Dict[str, Type[BaseModule]]: - """Scan modules/stealth/ for BaseModule subclasses.""" +def discover_modules_in_category(category: str) -> Dict[str, Type[BaseModule]]: + """Scan modules// for BaseModule subclasses.""" modules: Dict[str, Type[BaseModule]] = {} - stealth_dir = os.path.join(PROJECT_ROOT, "modules", "stealth") + cat_dir = os.path.join(PROJECT_ROOT, "modules", category) - if not os.path.isdir(stealth_dir): + if not os.path.isdir(cat_dir): return modules - for filename in sorted(os.listdir(stealth_dir)): + for filename in sorted(os.listdir(cat_dir)): if filename.startswith("_") or not filename.endswith(".py"): continue module_name = filename[:-3] - dotted = f"modules.stealth.{module_name}" + dotted = f"modules.{category}.{module_name}" try: mod = importlib.import_module(dotted) @@ -90,18 +96,53 @@ def discover_stealth_modules() -> Dict[str, Type[BaseModule]]: return modules +def discover_all_modules() -> Dict[str, Type[BaseModule]]: + """Discover all modules across every category.""" + all_modules: Dict[str, Type[BaseModule]] = {} + for category in MODULE_CATEGORIES: + all_modules.update(discover_modules_in_category(category)) + return all_modules + + +def discover_stealth_modules() -> Dict[str, Type[BaseModule]]: + """Scan modules/stealth/ for BaseModule subclasses (backward compat).""" + return discover_modules_in_category("stealth") + + def get_module_config(module_name: str, config: dict) -> dict: """Extract module-specific config from the master config.""" - # Check for module-specific section + # Check for module-specific section in each category + for category in MODULE_CATEGORIES: + cat_cfg = config.get(category, {}) + if module_name in cat_cfg: + merged = dict(config) + merged.update(cat_cfg[module_name]) + return merged + + # Fallback: check top-level modules section mod_cfg = config.get("modules", {}).get(module_name, {}) if mod_cfg: - # Merge with top-level config for access to device, network, etc. merged = dict(config) merged.update(mod_cfg) return merged return dict(config) +def is_module_enabled(module_name: str, module_type: str, config: dict) -> bool: + """Check if a module is enabled in the config (modules.yaml).""" + # Look in the category-level config + cat_cfg = config.get(module_type, {}) + if module_name in cat_cfg: + return cat_cfg[module_name].get("enabled", True) + # Default: stealth/passive/intel enabled, active disabled + if module_type in ("stealth", "passive", "intel"): + return True + if module_type == "connectivity": + # Connectivity has mixed defaults; check config + return cat_cfg.get(module_name, {}).get("enabled", False) + return False # active modules default to disabled + + # --------------------------------------------------------------------------- # Click CLI # --------------------------------------------------------------------------- @@ -145,15 +186,26 @@ def start(daemon, passive_only): tool_manager = ToolManager(bus=bus) tool_manager.start_monitoring() - # Discover and register stealth modules - discovered = discover_stealth_modules() - console.print(f" Discovered [cyan]{len(discovered)}[/cyan] stealth modules") - - enabled_modules = config.get("modules", {}).get("enabled", list(discovered.keys())) + # Discover all modules across every category + discovered = discover_all_modules() + by_type: Dict[str, int] = {} for name, cls in discovered.items(): - if name in enabled_modules or not config.get("modules", {}).get("enabled"): + mod_type = getattr(cls, "module_type", "unknown") + by_type[mod_type] = by_type.get(mod_type, 0) + 1 + + console.print(f" Discovered [cyan]{len(discovered)}[/cyan] modules: " + + ", ".join(f"{t}={c}" for t, c in sorted(by_type.items()))) + + # Register enabled modules + registered_count = 0 + for name, cls in discovered.items(): + mod_type = getattr(cls, "module_type", "unknown") + if is_module_enabled(name, mod_type, config): mod_config = get_module_config(name, config) engine.register(cls, mod_config) + registered_count += 1 + + console.print(f" Registered [cyan]{registered_count}[/cyan] enabled modules") # Start modules in dependency order results = engine.start_all() @@ -272,7 +324,7 @@ def activate(module_name): console.print(f"[red]Config error:[/red] {exc}") return - discovered = discover_stealth_modules() + discovered = discover_all_modules() if module_name not in discovered: console.print(f"[red]Module '{module_name}' not found.[/red]") console.print(f"Available: {', '.join(sorted(discovered.keys()))}") @@ -443,10 +495,11 @@ def selftest(): else: console.print(f" [yellow]SKIP[/yellow] Binary not found: {binary}") - # 7. Stealth modules - discovered = discover_stealth_modules() - console.print(f" [green]PASS[/green] Stealth modules discovered: {len(discovered)}") - passed += 1 + # 7. Module discovery — all categories + for category in MODULE_CATEGORIES: + discovered = discover_modules_in_category(category) + console.print(f" [green]PASS[/green] {category.capitalize()} modules: {len(discovered)}") + passed += 1 console.print(f"\n[bold]Results: {passed} passed, {failed} failed[/bold]") @@ -459,8 +512,7 @@ def list_modules(): except Exception: config = {} - discovered = discover_stealth_modules() - enabled_list = config.get("modules", {}).get("enabled") + discovered = discover_all_modules() table = Table(title="Available Modules", show_lines=True) table.add_column("Module", style="cyan", min_width=22) @@ -477,16 +529,340 @@ def list_modules(): requires_root = "Yes" if getattr(cls, "requires_root", False) else "No" deps = ", ".join(getattr(cls, "dependencies", [])) or "-" - if enabled_list is None: - enabled = "[green]Yes[/green]" - elif name in enabled_list: - enabled = "[green]Yes[/green]" - else: - enabled = "[dim]No[/dim]" + enabled = is_module_enabled(name, mod_type, config) + enabled_str = "[green]Yes[/green]" if enabled else "[dim]No[/dim]" - table.add_row(name, mod_type, priority, requires_root, enabled, deps) + table.add_row(name, mod_type, priority, requires_root, enabled_str, deps) console.print(table) + console.print(f"\n Total: [cyan]{len(discovered)}[/cyan] modules across " + f"{len(MODULE_CATEGORIES)} categories") + + +# --------------------------------------------------------------------------- +# Interactive menu — view functions +# --------------------------------------------------------------------------- + +def _view_credentials(): + """Query the credential database and display a Rich table.""" + base_dir = os.path.expanduser("~/.bigbrother") + db_path = os.path.join(base_dir, "credentials.db") + + if not os.path.isfile(db_path): + console.print("\n[yellow]No credential database found.[/yellow]") + console.print(f" Expected: {db_path}") + console.print(" (Credentials are stored when modules capture them at runtime)") + return + + try: + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + # Summary stats + total = conn.execute("SELECT COUNT(*) FROM credentials").fetchone()[0] + cracked = conn.execute( + "SELECT COUNT(*) FROM credentials WHERE crack_status='cracked'" + ).fetchone()[0] + unique_users = conn.execute( + "SELECT COUNT(DISTINCT username) FROM credentials" + ).fetchone()[0] + unique_hosts = conn.execute( + "SELECT COUNT(DISTINCT target_ip) FROM credentials WHERE target_ip != ''" + ).fetchone()[0] + + console.print(f"\n Total credentials: [cyan]{total}[/cyan] | " + f"Cracked: [green]{cracked}[/green] | " + f"Unique users: [yellow]{unique_users}[/yellow] | " + f"Unique hosts: [magenta]{unique_hosts}[/magenta]") + + if total == 0: + conn.close() + return + + # Show recent credentials + rows = conn.execute( + """SELECT id, timestamp, source_module, target_ip, service, + username, domain, cred_type, crack_status, cracked_value + FROM credentials + ORDER BY timestamp DESC LIMIT 50""" + ).fetchall() + + table = Table(title="Credentials (most recent 50)", show_lines=True) + table.add_column("ID", justify="right", min_width=4) + table.add_column("Time", min_width=16) + table.add_column("Source", min_width=14) + table.add_column("Target", min_width=14) + table.add_column("Service", min_width=10) + table.add_column("Domain\\User", min_width=20) + table.add_column("Type", min_width=10) + table.add_column("Status", min_width=8) + + for row in rows: + ts = datetime.fromtimestamp(row["timestamp"]).strftime("%Y-%m-%d %H:%M") + domain_user = row["username"] + if row["domain"]: + domain_user = f"{row['domain']}\\{row['username']}" + + crack_status = row["crack_status"] + if crack_status == "cracked": + status_str = f"[green]cracked[/green]" + elif crack_status == "cracking": + status_str = f"[yellow]cracking[/yellow]" + else: + status_str = f"[dim]uncracked[/dim]" + + table.add_row( + str(row["id"]), + ts, + row["source_module"] or "-", + row["target_ip"] or "-", + row["service"] or "-", + domain_user, + row["cred_type"] or "-", + status_str, + ) + + console.print(table) + + # Type breakdown + type_rows = conn.execute( + "SELECT cred_type, COUNT(*) as cnt FROM credentials GROUP BY cred_type ORDER BY cnt DESC" + ).fetchall() + if type_rows: + console.print("\n By type: " + " | ".join( + f"{r['cred_type']}={r['cnt']}" for r in type_rows + )) + + conn.close() + + except Exception as exc: + console.print(f"\n[red]Error reading credential database:[/red] {exc}") + + +def _view_intel(): + """Show an intelligence summary: hosts, credentials, modules, security tools.""" + base_dir = os.path.expanduser("~/.bigbrother") + state = StateManager() + + console.print("\n") + intel_table = Table(title="Intelligence Summary", show_lines=True, min_width=60) + intel_table.add_column("Category", style="cyan", min_width=25) + intel_table.add_column("Value", min_width=30) + + # Host count from topology mapper state + nodes_json = state.get("topology_mapper", "nodes") + host_count = 0 + if nodes_json: + try: + nodes = json.loads(nodes_json) + host_count = len(nodes) + except (json.JSONDecodeError, TypeError): + pass + intel_table.add_row("Discovered Hosts", str(host_count)) + + # Credential count + cred_db_path = os.path.join(base_dir, "credentials.db") + cred_count = 0 + cracked_count = 0 + if os.path.isfile(cred_db_path): + try: + conn = sqlite3.connect(cred_db_path) + cred_count = conn.execute("SELECT COUNT(*) FROM credentials").fetchone()[0] + cracked_count = conn.execute( + "SELECT COUNT(*) FROM credentials WHERE crack_status='cracked'" + ).fetchone()[0] + conn.close() + except Exception: + pass + intel_table.add_row("Credentials (total/cracked)", f"{cred_count} / {cracked_count}") + + # Active modules + all_status = state.get_all_module_status() + running_modules = [n for n, s in all_status.items() if s.get("status") == "running"] + intel_table.add_row("Running Modules", str(len(running_modules))) + if running_modules: + intel_table.add_row(" Active", ", ".join(sorted(running_modules)[:10])) + + # Security posture + risk_level = state.get("security_posture", "risk_level") + if risk_level: + risk_color = {"low": "green", "medium": "yellow", "high": "red", "critical": "bold red" + }.get(risk_level, "dim") + intel_table.add_row("Security Risk Level", f"[{risk_color}]{risk_level}[/{risk_color}]") + else: + intel_table.add_row("Security Risk Level", "[dim]not assessed[/dim]") + + has_edr = state.get("security_posture", "has_edr") + if has_edr: + intel_table.add_row("EDR Detected", "Yes" if json.loads(has_edr) else "No") + + has_siem = state.get("security_posture", "has_siem") + if has_siem: + intel_table.add_row("SIEM Detected", "Yes" if json.loads(has_siem) else "No") + + has_nac = state.get("security_posture", "has_nac") + if has_nac: + intel_table.add_row("NAC Detected", "Yes" if json.loads(has_nac) else "No") + + has_honeypot = state.get("security_posture", "has_honeypot") + if has_honeypot: + intel_table.add_row("Honeypots Detected", "Yes" if json.loads(has_honeypot) else "No") + + # VLANs + vlans_json = state.get("topology_mapper", "vlans") + if vlans_json: + try: + vlans = json.loads(vlans_json) + intel_table.add_row("VLANs Discovered", str(len(vlans))) + except (json.JSONDecodeError, TypeError): + pass + + console.print(intel_table) + + +def _view_topology(): + """Show network topology from topology_mapper state data.""" + state = StateManager() + nodes_json = state.get("topology_mapper", "nodes") + edges_json = state.get("topology_mapper", "edges") + vlans_json = state.get("topology_mapper", "vlans") + + nodes = {} + edges = {} + vlans = {} + + if nodes_json: + try: + nodes = json.loads(nodes_json) + except (json.JSONDecodeError, TypeError): + pass + + if edges_json: + try: + edges = json.loads(edges_json) + except (json.JSONDecodeError, TypeError): + pass + + if vlans_json: + try: + vlans = json.loads(vlans_json) + except (json.JSONDecodeError, TypeError): + pass + + if not nodes: + console.print("\n[yellow]No topology data available yet.[/yellow]") + console.print(" Topology is built from passive host discovery and network observation.") + return + + # Host table + table = Table(title=f"Network Topology ({len(nodes)} hosts, {len(edges)} links, {len(vlans)} VLANs)", + show_lines=True) + table.add_column("IP", style="cyan", min_width=15) + table.add_column("Hostname", min_width=18) + table.add_column("MAC", min_width=17) + table.add_column("OS", min_width=10) + table.add_column("Role", min_width=12) + table.add_column("VLAN", justify="right", min_width=5) + table.add_column("Ports", min_width=15) + + for ip in sorted(nodes.keys()): + node = nodes[ip] + hostname = node.get("hostname", "") + mac = node.get("mac", "") + os_family = node.get("os_family", "unknown") + role = node.get("role", "unknown") + vlan = str(node.get("vlan", "")) if node.get("vlan") is not None else "-" + ports = node.get("open_ports", []) + if isinstance(ports, list): + ports_str = ", ".join(str(p) for p in sorted(ports)[:8]) + if len(ports) > 8: + ports_str += f" (+{len(ports) - 8})" + else: + ports_str = str(ports) + + table.add_row(ip, hostname or "-", mac or "-", os_family, role, vlan, ports_str or "-") + + console.print(f"\n") + console.print(table) + + # VLAN summary + if vlans: + console.print("\n VLANs:") + for vid, vinfo in sorted(vlans.items(), key=lambda x: str(x[0])): + name = vinfo.get("name", "") + subnet = vinfo.get("subnet", "") + console.print(f" VLAN {vid}: {name} ({subnet})" if name + else f" VLAN {vid}: {subnet}") + + +def _view_timeline(): + """Show recent events from the operator audit log.""" + base_dir = os.path.expanduser("~/.bigbrother") + db_path = os.path.join(base_dir, "operator_audit.db") + + if not os.path.isfile(db_path): + console.print("\n[yellow]No audit log found.[/yellow]") + console.print(f" Expected: {db_path}") + console.print(" (Audit log is created when operator_audit module starts)") + return + + try: + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + + total = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()[0] + + rows = conn.execute( + """SELECT id, timestamp, action, operator, details + FROM audit_log + ORDER BY id DESC LIMIT 50""" + ).fetchall() + + if not rows: + console.print("\n[yellow]Audit log is empty.[/yellow]") + conn.close() + return + + console.print(f"\n Total audit entries: [cyan]{total}[/cyan] (showing most recent 50)") + + table = Table(title="Operator Timeline", show_lines=True) + table.add_column("ID", justify="right", min_width=5) + table.add_column("Time", min_width=19) + table.add_column("Action", min_width=16) + table.add_column("Operator", min_width=16) + table.add_column("Details", min_width=30) + + for row in rows: + ts = datetime.fromtimestamp(row["timestamp"]).strftime("%Y-%m-%d %H:%M:%S") + action = row["action"] + + # Color code actions + action_colors = { + "module_start": "green", + "module_stop": "yellow", + "kill_switch": "bold red", + "config_change": "magenta", + "ssh_session": "cyan", + "cli_command": "blue", + "cred_export": "green", + "data_exfil": "green", + } + color = action_colors.get(action, "dim") + action_str = f"[{color}]{action}[/{color}]" + + table.add_row( + str(row["id"]), + ts, + action_str, + row["operator"] or "-", + (row["details"] or "-")[:60], + ) + + console.print(table) + conn.close() + + except Exception as exc: + console.print(f"\n[red]Error reading audit log:[/red] {exc}") # --------------------------------------------------------------------------- @@ -521,10 +897,10 @@ def interactive_menu(): console.print(" [bold]2.[/bold] Stop all modules") console.print(" [bold]3.[/bold] Module status") console.print(" [bold]4.[/bold] List modules") - console.print(" [bold]5.[/bold] Credentials [dim](Available in Phase 2)[/dim]") - console.print(" [bold]6.[/bold] Intelligence [dim](Available in Phase 2)[/dim]") - console.print(" [bold]7.[/bold] Topology [dim](Available in Phase 2)[/dim]") - console.print(" [bold]8.[/bold] Timeline [dim](Available in Phase 2)[/dim]") + console.print(" [bold]5.[/bold] View Credentials") + console.print(" [bold]6.[/bold] View Intelligence") + console.print(" [bold]7.[/bold] Network Topology") + console.print(" [bold]8.[/bold] Timeline") console.print(" [bold]9.[/bold] Self-test") console.print(" [bold]C.[/bold] Configuration") console.print(" [bold]K.[/bold] Kill switch") @@ -546,9 +922,14 @@ def interactive_menu(): elif choice == "4": ctx = click.Context(list_modules) ctx.invoke(list_modules) - elif choice in ("5", "6", "7", "8"): - labels = {"5": "Credentials", "6": "Intelligence", "7": "Topology", "8": "Timeline"} - console.print(f"\n[yellow]{labels[choice]} viewer is available in Phase 2.[/yellow]") + elif choice == "5": + _view_credentials() + elif choice == "6": + _view_intel() + elif choice == "7": + _view_topology() + elif choice == "8": + _view_timeline() elif choice == "9": ctx = click.Context(selftest) ctx.invoke(selftest) diff --git a/config/modules.yaml b/config/modules.yaml index 21f822f..9d8d8f2 100644 --- a/config/modules.yaml +++ b/config/modules.yaml @@ -3,7 +3,7 @@ # Modules gated by hardware tier and engagement phase. # --------------------------------------------------------------------------- -# Passive Modules (16) — metadata extraction, zero network noise +# Passive Modules (17) — metadata extraction, zero network noise # --------------------------------------------------------------------------- passive: packet_capture: @@ -103,12 +103,12 @@ passive: - mongodb # --------------------------------------------------------------------------- -# Active Modules (8) — operator-enabled, bettercap + tool wrappers +# Active Modules (9) — operator-enabled, bettercap + tool wrappers # Gated by engagement_phase.mode == active_allowed # --------------------------------------------------------------------------- active: bettercap_mgr: - enabled: true + enabled: false description: "Central bettercap orchestrator (REST API lifecycle)" max_restarts: 3 health_poll_interval_s: 10 @@ -220,7 +220,7 @@ stealth: description: "Read-only root FS + tmpfs overlay (zero SD writes)" # --------------------------------------------------------------------------- -# Intelligence Modules (8) — on-device analysis + aggregation +# Intelligence Modules (9) — on-device analysis + aggregation # --------------------------------------------------------------------------- intel: credential_db: diff --git a/modules/connectivity/cellular_backup.py b/modules/connectivity/cellular_backup.py index 8b47004..244761c 100644 --- a/modules/connectivity/cellular_backup.py +++ b/modules/connectivity/cellular_backup.py @@ -11,11 +11,15 @@ OPi Zero 3+ only — skips on pi_zero tier (no USB bandwidth for modem). import logging import os import re -import serial import subprocess import time from typing import Optional +try: + import serial +except ImportError: + serial = None + from modules.base import BaseModule logger = logging.getLogger("bb.connectivity.cellular_backup") @@ -51,6 +55,12 @@ class CellularBackup(BaseModule): if self._running: return + if serial is None: + logger.error("pyserial not installed — cellular backup unavailable") + self.state.set_module_status(self.name, "error", + extra={"reason": "pyserial_missing"}) + return + # Check hardware tier device_tier = self.config.get("device", {}).get("tier", "") if device_tier == "pi_zero": diff --git a/tests/test_all_modules.py b/tests/test_all_modules.py new file mode 100644 index 0000000..dbaa609 --- /dev/null +++ b/tests/test_all_modules.py @@ -0,0 +1,416 @@ +"""Tests for ALL module categories — import, BaseModule subclass, and attribute validation. + +Covers: passive (17), active (9), intel (9), connectivity (8). +Stealth modules are already covered by test_stealth_modules.py. +""" + +import importlib + +import pytest + +from modules.base import BaseModule + + +# --------------------------------------------------------------------------- +# Passive modules (17) +# --------------------------------------------------------------------------- + +PASSIVE_MODULE_DOTTED = [ + "modules.passive.packet_capture", + "modules.passive.dns_logger", + "modules.passive.tls_sni_extractor", + "modules.passive.credential_sniffer", + "modules.passive.kerberos_harvester", + "modules.passive.host_discovery", + "modules.passive.os_fingerprint", + "modules.passive.traffic_analyzer", + "modules.passive.vlan_discovery", + "modules.passive.network_mapper", + "modules.passive.auth_flow_tracker", + "modules.passive.smb_monitor", + "modules.passive.cloud_token_harvester", + "modules.passive.ldap_harvester", + "modules.passive.rdp_monitor", + "modules.passive.quic_analyzer", + "modules.passive.db_interceptor", +] + +PASSIVE_CLASS_NAMES = [ + "PacketCapture", + "DNSLogger", + "TLSSNIExtractor", + "CredentialSniffer", + "KerberosHarvester", + "HostDiscovery", + "OSFingerprint", + "TrafficAnalyzer", + "VLANDiscovery", + "NetworkMapper", + "AuthFlowTracker", + "SMBMonitor", + "CloudTokenHarvester", + "LDAPHarvester", + "RDPMonitor", + "QUICAnalyzer", + "DBInterceptor", +] + + +class TestPassiveModulesImportable: + + @pytest.mark.parametrize("dotted_path", PASSIVE_MODULE_DOTTED) + def test_passive_module_importable(self, dotted_path): + """Each passive module file can be imported without errors.""" + mod = importlib.import_module(dotted_path) + assert mod is not None + + +class TestPassiveModulesAreBaseModule: + + @pytest.mark.parametrize("class_name", PASSIVE_CLASS_NAMES) + def test_passive_modules_are_base_module(self, class_name): + """Each passive module class is a subclass of BaseModule.""" + import modules.passive as passive_pkg + cls = getattr(passive_pkg, class_name) + assert issubclass(cls, BaseModule), f"{class_name} is not a BaseModule subclass" + + +class TestPassiveModuleAttributes: + + @pytest.mark.parametrize("class_name", PASSIVE_CLASS_NAMES) + def test_passive_module_attributes(self, class_name): + """Each passive module has required class attributes.""" + import modules.passive as passive_pkg + cls = getattr(passive_pkg, class_name) + + assert hasattr(cls, "name") and cls.name != "unnamed" + assert hasattr(cls, "module_type") and cls.module_type == "passive" + assert hasattr(cls, "priority") and isinstance(cls.priority, int) + assert hasattr(cls, "dependencies") and isinstance(cls.dependencies, list) + assert hasattr(cls, "requires_root") + + for method_name in ("start", "stop", "status", "configure"): + assert callable(getattr(cls, method_name, None)) + + +class TestPassiveModuleInstantiate: + + @pytest.mark.parametrize("class_name", PASSIVE_CLASS_NAMES) + def test_passive_module_instantiate(self, class_name, mock_bus, mock_state, mock_config): + """Each passive module can be instantiated with mock bus/state/config.""" + import modules.passive as passive_pkg + cls = getattr(passive_pkg, class_name) + instance = cls(bus=mock_bus, state=mock_state, config=mock_config) + assert instance is not None + assert instance._running is False + + +# --------------------------------------------------------------------------- +# Active modules (9) +# --------------------------------------------------------------------------- + +ACTIVE_MODULE_DOTTED = [ + "modules.active.bettercap_mgr", + "modules.active.arp_spoof", + "modules.active.dns_poison", + "modules.active.dhcp_spoof", + "modules.active.evil_twin", + "modules.active.ipv6_slaac", + "modules.active.responder_mgr", + "modules.active.mitmproxy_mgr", + "modules.active.ntlm_relay", +] + +ACTIVE_CLASS_NAMES = [ + "BettercapManager", + "ARPSpoof", + "DNSPoison", + "DHCPSpoof", + "EvilTwin", + "IPv6SLAAC", + "ResponderManager", + "MitmproxyManager", + "NTLMRelay", +] + + +class TestActiveModulesImportable: + + @pytest.mark.parametrize("dotted_path", ACTIVE_MODULE_DOTTED) + def test_active_module_importable(self, dotted_path): + """Each active module file can be imported without errors.""" + mod = importlib.import_module(dotted_path) + assert mod is not None + + +class TestActiveModulesAreBaseModule: + + @pytest.mark.parametrize("class_name", ACTIVE_CLASS_NAMES) + def test_active_modules_are_base_module(self, class_name): + """Each active module class is a subclass of BaseModule.""" + import modules.active as active_pkg + cls = getattr(active_pkg, class_name) + assert issubclass(cls, BaseModule), f"{class_name} is not a BaseModule subclass" + + +class TestActiveModuleAttributes: + + @pytest.mark.parametrize("class_name", ACTIVE_CLASS_NAMES) + def test_active_module_attributes(self, class_name): + """Each active module has required class attributes.""" + import modules.active as active_pkg + cls = getattr(active_pkg, class_name) + + assert hasattr(cls, "name") and cls.name != "unnamed" + assert hasattr(cls, "module_type") and cls.module_type == "active" + assert hasattr(cls, "priority") and isinstance(cls.priority, int) + assert hasattr(cls, "dependencies") and isinstance(cls.dependencies, list) + assert hasattr(cls, "requires_root") + + for method_name in ("start", "stop", "status", "configure"): + assert callable(getattr(cls, method_name, None)) + + +class TestActiveModuleInstantiate: + + @pytest.mark.parametrize("class_name", ACTIVE_CLASS_NAMES) + def test_active_module_instantiate(self, class_name, mock_bus, mock_state, mock_config): + """Each active module can be instantiated with mock bus/state/config.""" + import modules.active as active_pkg + cls = getattr(active_pkg, class_name) + instance = cls(bus=mock_bus, state=mock_state, config=mock_config) + assert instance is not None + assert instance._running is False + + +# --------------------------------------------------------------------------- +# Intel modules (9) +# --------------------------------------------------------------------------- + +INTEL_MODULE_DOTTED = [ + "modules.intel.credential_db", + "modules.intel.topology_mapper", + "modules.intel.net_intel", + "modules.intel.user_timeline", + "modules.intel.supply_chain_detect", + "modules.intel.change_detector", + "modules.intel.security_posture", + "modules.intel.operator_audit", + "modules.intel.tool_output_parser", +] + +INTEL_CLASS_NAMES = [ + "CredentialDB", + "TopologyMapper", + "NetIntel", + "UserTimeline", + "SupplyChainDetect", + "ChangeDetector", + "SecurityPosture", + "OperatorAudit", + "ToolOutputParser", +] + + +class TestIntelModulesImportable: + + @pytest.mark.parametrize("dotted_path", INTEL_MODULE_DOTTED) + def test_intel_module_importable(self, dotted_path): + """Each intel module file can be imported without errors.""" + mod = importlib.import_module(dotted_path) + assert mod is not None + + +class TestIntelModulesAreBaseModule: + + @pytest.mark.parametrize("class_name", INTEL_CLASS_NAMES) + def test_intel_modules_are_base_module(self, class_name): + """Each intel module class is a subclass of BaseModule.""" + import modules.intel as intel_pkg + cls = getattr(intel_pkg, class_name) + assert issubclass(cls, BaseModule), f"{class_name} is not a BaseModule subclass" + + +class TestIntelModuleAttributes: + + @pytest.mark.parametrize("class_name", INTEL_CLASS_NAMES) + def test_intel_module_attributes(self, class_name): + """Each intel module has required class attributes.""" + import modules.intel as intel_pkg + cls = getattr(intel_pkg, class_name) + + assert hasattr(cls, "name") and cls.name != "unnamed" + assert hasattr(cls, "module_type") and cls.module_type == "intel" + assert hasattr(cls, "priority") and isinstance(cls.priority, int) + assert hasattr(cls, "dependencies") and isinstance(cls.dependencies, list) + assert hasattr(cls, "requires_root") + + for method_name in ("start", "stop", "status", "configure"): + assert callable(getattr(cls, method_name, None)) + + +class TestIntelModuleInstantiate: + + @pytest.mark.parametrize("class_name", INTEL_CLASS_NAMES) + def test_intel_module_instantiate(self, class_name, mock_bus, mock_state, mock_config): + """Each intel module can be instantiated with mock bus/state/config.""" + import modules.intel as intel_pkg + cls = getattr(intel_pkg, class_name) + instance = cls(bus=mock_bus, state=mock_state, config=mock_config) + assert instance is not None + assert instance._running is False + + +# --------------------------------------------------------------------------- +# Connectivity modules (8) +# --------------------------------------------------------------------------- + +CONNECTIVITY_MODULE_DOTTED = [ + "modules.connectivity.wireguard", + "modules.connectivity.tailscale", + "modules.connectivity.bridge", + "modules.connectivity.wifi_client", + "modules.connectivity.reverse_tunnel", + "modules.connectivity.cellular_backup", + "modules.connectivity.ble_emergency", + "modules.connectivity.data_exfil", +] + +CONNECTIVITY_CLASS_NAMES = [ + "WireGuard", + "Tailscale", + "Bridge", + "WiFiClient", + "ReverseTunnel", + "CellularBackup", + "BLEEmergency", + "DataExfil", +] + + +class TestConnectivityModulesImportable: + + @pytest.mark.parametrize("dotted_path", CONNECTIVITY_MODULE_DOTTED) + def test_connectivity_module_importable(self, dotted_path): + """Each connectivity module file can be imported without errors.""" + mod = importlib.import_module(dotted_path) + assert mod is not None + + +class TestConnectivityModulesAreBaseModule: + + @pytest.mark.parametrize("class_name", CONNECTIVITY_CLASS_NAMES) + def test_connectivity_modules_are_base_module(self, class_name): + """Each connectivity module class is a subclass of BaseModule.""" + import modules.connectivity as conn_pkg + cls = getattr(conn_pkg, class_name) + assert issubclass(cls, BaseModule), f"{class_name} is not a BaseModule subclass" + + +class TestConnectivityModuleAttributes: + + @pytest.mark.parametrize("class_name", CONNECTIVITY_CLASS_NAMES) + def test_connectivity_module_attributes(self, class_name): + """Each connectivity module has required class attributes.""" + import modules.connectivity as conn_pkg + cls = getattr(conn_pkg, class_name) + + assert hasattr(cls, "name") and cls.name != "unnamed" + assert hasattr(cls, "module_type") and cls.module_type == "connectivity" + assert hasattr(cls, "priority") and isinstance(cls.priority, int) + assert hasattr(cls, "dependencies") and isinstance(cls.dependencies, list) + assert hasattr(cls, "requires_root") + + for method_name in ("start", "stop", "status", "configure"): + assert callable(getattr(cls, method_name, None)) + + +class TestConnectivityModuleInstantiate: + + @pytest.mark.parametrize("class_name", CONNECTIVITY_CLASS_NAMES) + def test_connectivity_module_instantiate(self, class_name, mock_bus, mock_state, mock_config): + """Each connectivity module can be instantiated with mock bus/state/config.""" + import modules.connectivity as conn_pkg + cls = getattr(conn_pkg, class_name) + instance = cls(bus=mock_bus, state=mock_state, config=mock_config) + assert instance is not None + assert instance._running is False + + +# --------------------------------------------------------------------------- +# Cross-category: package __init__.py exports +# --------------------------------------------------------------------------- + +class TestPackageExports: + + def test_passive_init_exports_all(self): + """modules.passive.__init__.py exports all 17 classes.""" + import modules.passive as pkg + assert len(pkg.__all__) == 17 + for name in pkg.__all__: + cls = getattr(pkg, name) + assert issubclass(cls, BaseModule) + + def test_active_init_exports_all(self): + """modules.active.__init__.py exports all 9 classes.""" + import modules.active as pkg + assert len(pkg.__all__) == 9 + for name in pkg.__all__: + cls = getattr(pkg, name) + assert issubclass(cls, BaseModule) + + def test_intel_init_exports_all(self): + """modules.intel.__init__.py exports all 9 classes.""" + import modules.intel as pkg + assert len(pkg.__all__) == 9 + for name in pkg.__all__: + cls = getattr(pkg, name) + assert issubclass(cls, BaseModule) + + def test_connectivity_init_exports_all(self): + """modules.connectivity.__init__.py exports all 8 classes.""" + import modules.connectivity as pkg + assert len(pkg.__all__) == 8 + for name in pkg.__all__: + cls = getattr(pkg, name) + assert issubclass(cls, BaseModule) + + def test_stealth_init_exports_all(self): + """modules.stealth.__init__.py exports all 12 classes.""" + import modules.stealth as pkg + assert len(pkg.__all__) == 12 + for name in pkg.__all__: + cls = getattr(pkg, name) + assert issubclass(cls, BaseModule) + + +# --------------------------------------------------------------------------- +# Discovery function test +# --------------------------------------------------------------------------- + +class TestModuleDiscovery: + + def test_discover_all_modules(self): + """discover_all_modules() finds all 55 modules across 5 categories.""" + from pathlib import Path + import sys + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + from bigbrother import discover_all_modules, discover_modules_in_category + + all_mods = discover_all_modules() + # Stealth: 12, Passive: 17, Active: 9, Intel: 9, Connectivity: 8 = 55 + assert len(all_mods) == 55, f"Expected 55 modules, found {len(all_mods)}" + + # Verify per-category counts + assert len(discover_modules_in_category("stealth")) == 12 + assert len(discover_modules_in_category("passive")) == 17 + assert len(discover_modules_in_category("active")) == 9 + assert len(discover_modules_in_category("intel")) == 9 + assert len(discover_modules_in_category("connectivity")) == 8 + + def test_all_discovered_are_base_module(self): + """Every discovered module is a BaseModule subclass.""" + from bigbrother import discover_all_modules + + for name, cls in discover_all_modules().items(): + assert issubclass(cls, BaseModule), f"{name} ({cls}) is not a BaseModule subclass"