From 176da06dc93718b6b8098eee235548cc4e9941d1 Mon Sep 17 00:00:00 2001 From: n0mad1k Date: Wed, 18 Mar 2026 13:33:17 -0400 Subject: [PATCH] =?UTF-8?q?Add=20Phase=202=20intel=20modules=20=E2=80=94?= =?UTF-8?q?=209=20intelligence=20modules=20for=20on-device=20analysis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - credential_db: Central credential store with dedup, hashcat/CSV/JSON export, crack tracking - topology_mapper: Network graph from HOST_DISCOVERED/VLAN_DETECTED events, Graphviz DOT/SVG output - net_intel: Beacon detection (stddev/mean analysis), comm graph, DNS tunneling, service deps - user_timeline: Per-user activity timelines, work hours, admin/service account identification - supply_chain_detect: Passive detection of PyPI/npm mirrors, WSUS, CI/CD, SCCM, container registries - change_detector: Continuous baseline diff for burn detection, scan/security tool alerts - security_posture: EDR/SIEM/NAC/honeypot/scanner detection by DNS/port/UA, risk gating - operator_audit: Append-only HMAC chain audit log for engagement deconfliction - tool_output_parser: Unified parser for bettercap events, Responder logs, mitmproxy flows --- modules/intel/__init__.py | 23 + modules/intel/change_detector.py | 604 ++++++++++++++++++++ modules/intel/credential_db.py | 492 ++++++++++++++++ modules/intel/net_intel.py | 480 ++++++++++++++++ modules/intel/operator_audit.py | 412 ++++++++++++++ modules/intel/security_posture.py | 577 +++++++++++++++++++ modules/intel/supply_chain_detect.py | 491 ++++++++++++++++ modules/intel/tool_output_parser.py | 800 +++++++++++++++++++++++++++ modules/intel/topology_mapper.py | 528 ++++++++++++++++++ modules/intel/user_timeline.py | 584 +++++++++++++++++++ 10 files changed, 4991 insertions(+) create mode 100644 modules/intel/__init__.py create mode 100644 modules/intel/change_detector.py create mode 100644 modules/intel/credential_db.py create mode 100644 modules/intel/net_intel.py create mode 100644 modules/intel/operator_audit.py create mode 100644 modules/intel/security_posture.py create mode 100644 modules/intel/supply_chain_detect.py create mode 100644 modules/intel/tool_output_parser.py create mode 100644 modules/intel/topology_mapper.py create mode 100644 modules/intel/user_timeline.py diff --git a/modules/intel/__init__.py b/modules/intel/__init__.py new file mode 100644 index 0000000..0cc930e --- /dev/null +++ b/modules/intel/__init__.py @@ -0,0 +1,23 @@ +"""BigBrother intelligence modules — on-device analysis and data aggregation.""" + +from modules.intel.credential_db import CredentialDB +from modules.intel.topology_mapper import TopologyMapper +from modules.intel.net_intel import NetIntel +from modules.intel.user_timeline import UserTimeline +from modules.intel.supply_chain_detect import SupplyChainDetect +from modules.intel.change_detector import ChangeDetector +from modules.intel.security_posture import SecurityPosture +from modules.intel.operator_audit import OperatorAudit +from modules.intel.tool_output_parser import ToolOutputParser + +__all__ = [ + "CredentialDB", + "TopologyMapper", + "NetIntel", + "UserTimeline", + "SupplyChainDetect", + "ChangeDetector", + "SecurityPosture", + "OperatorAudit", + "ToolOutputParser", +] diff --git a/modules/intel/change_detector.py b/modules/intel/change_detector.py new file mode 100644 index 0000000..1c27f39 --- /dev/null +++ b/modules/intel/change_detector.py @@ -0,0 +1,604 @@ +#!/usr/bin/env python3 +"""Network change detection — continuous diff against baseline state. + +Critical for BURN DETECTION. Alerts on new hosts, disappeared hosts, +new services, new open ports, security tool traffic, scan activity +targeting the implant subnet, and unusual auth patterns. + +Publishes CHANGE_DETECTED events with severity levels. +""" + +import json +import logging +import os +import sqlite3 +import threading +import time +from collections import defaultdict +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.change_detector") + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS changes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + change_type TEXT NOT NULL, + severity TEXT NOT NULL DEFAULT 'info', + description TEXT NOT NULL, + old_value TEXT DEFAULT '', + new_value TEXT DEFAULT '', + target_ip TEXT DEFAULT '', + acknowledged INTEGER DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS idx_changes_ts ON changes(timestamp); +CREATE INDEX IF NOT EXISTS idx_changes_sev ON changes(severity); +CREATE INDEX IF NOT EXISTS idx_changes_type ON changes(change_type); + +CREATE TABLE IF NOT EXISTS baseline_hosts ( + ip TEXT PRIMARY KEY, + hostname TEXT DEFAULT '', + mac TEXT DEFAULT '', + os_family TEXT DEFAULT '', + open_ports TEXT DEFAULT '[]', + services TEXT DEFAULT '[]', + first_seen REAL NOT NULL, + last_seen REAL NOT NULL +); + +CREATE TABLE IF NOT EXISTS baseline_services ( + ip TEXT NOT NULL, + port INTEGER NOT NULL, + protocol TEXT DEFAULT 'tcp', + service TEXT DEFAULT '', + banner TEXT DEFAULT '', + first_seen REAL NOT NULL, + PRIMARY KEY (ip, port, protocol) +); +""" + +# Change types and their default severity +_CHANGE_SEVERITY = { + "new_host": "info", + "host_disappeared": "warning", + "new_service": "info", + "new_port": "info", + "service_disappeared": "info", + "security_tool_traffic": "critical", + "scan_detected": "critical", + "unusual_auth": "warning", + "mac_change": "warning", + "os_change": "warning", + "mass_port_scan": "critical", + "arp_anomaly": "critical", +} + +# Ports/services that indicate security tool deployment +_SECURITY_TOOL_PORTS = { + 8088: "splunk_hec", + 8089: "splunk_mgmt", + 9997: "splunk_forwarder", + 1514: "wazuh_agent", + 1515: "wazuh_enrollment", + 55000: "wazuh_api", + 514: "syslog", + 6514: "syslog_tls", + 443: None, # too generic, needs hostname corroboration +} + +# DNS patterns suggesting security tool investigation +_SECURITY_DNS_PATTERNS = ( + "nessus", "qualys", "rapid7", "tenable", "crowdstrike", + "sentinelone", "carbonblack", "cybereason", "sophos", + "defender", "wireshark", "zeek", "suricata", "snort", +) + +# Scan detection: many ports from a single source in short time +_SCAN_THRESHOLD_PORTS = 50 # ports in window = scan +_SCAN_THRESHOLD_WINDOW = 60.0 # seconds + + +class ChangeDetector(BaseModule): + """Detect network changes against baseline — critical for burn detection.""" + + name = "change_detector" + module_type = "intel" + priority = 50 # Critical priority + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._db_path = "" + self._conn: Optional[sqlite3.Connection] = None + self._lock = threading.Lock() + self._monitor_thread: Optional[threading.Thread] = None + self._baseline_built = False + self._change_count = 0 + + # Scan detection tracking: source_ip -> [(timestamp, dst_port)] + self._port_attempts: dict[str, list] = defaultdict(list) + + # Our implant's IPs for self-protection monitoring + self._implant_ips: set = set() + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "change_detector.db") + os.makedirs(os.path.dirname(self._db_path), exist_ok=True) + + self._conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=NORMAL") + self._conn.row_factory = sqlite3.Row + self._conn.executescript(_SCHEMA) + + # Load implant IPs from config + self._implant_ips = set(self.config.get("implant_ips", [])) + + # Check if baseline exists + row = self._conn.execute("SELECT COUNT(*) FROM baseline_hosts").fetchone() + self._baseline_built = row[0] > 0 + + # Subscribe to events + self.bus.subscribe(self._on_host_discovered, "HOST_DISCOVERED") + self.bus.subscribe(self._on_credential_found, "CREDENTIAL_FOUND") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + # Monitor thread for periodic diffing + self._monitor_thread = threading.Thread( + target=self._monitor_loop, daemon=True, + name="bb-change-detector", + ) + self._monitor_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info( + "ChangeDetector started — baseline=%s, db=%s", + "loaded" if self._baseline_built else "empty", self._db_path, + ) + + def stop(self) -> None: + if not self._running: + return + self._running = False + + self.bus.unsubscribe(self._on_host_discovered, "HOST_DISCOVERED") + self.bus.unsubscribe(self._on_credential_found, "CREDENTIAL_FOUND") + + if self._conn: + self._conn.close() + self._conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("ChangeDetector stopped — %d changes recorded", self._change_count) + + def status(self) -> dict: + change_count = 0 + critical_count = 0 + if self._conn: + try: + with self._lock: + row = self._conn.execute( + "SELECT COUNT(*) FROM changes" + ).fetchone() + change_count = row[0] if row else 0 + row = self._conn.execute( + "SELECT COUNT(*) FROM changes WHERE severity='critical'" + ).fetchone() + critical_count = row[0] if row else 0 + except Exception: + pass + + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "baseline_built": self._baseline_built, + "total_changes": change_count, + "critical_changes": critical_count, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + self._implant_ips = set(config.get("implant_ips", self._implant_ips)) + + # ------------------------------------------------------------------ + # Baseline management + # ------------------------------------------------------------------ + + def build_baseline(self) -> int: + """Snapshot current known hosts and services as the baseline. + + Returns: + Number of hosts in the baseline. + """ + hosts_json = self.state.get("host_discovery", "discovered_hosts") + if not hosts_json: + logger.warning("No host discovery data available for baseline") + return 0 + + try: + hosts = json.loads(hosts_json) + except (json.JSONDecodeError, TypeError): + return 0 + + with self._lock: + # Clear existing baseline + self._conn.execute("DELETE FROM baseline_hosts") + self._conn.execute("DELETE FROM baseline_services") + + now = time.time() + for host in hosts: + ip = host.get("ip", "") + if not ip: + continue + + ports = host.get("open_ports", []) + services = host.get("services", []) + + self._conn.execute( + """INSERT OR REPLACE INTO baseline_hosts + (ip, hostname, mac, os_family, open_ports, services, + first_seen, last_seen) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", + (ip, host.get("hostname", ""), host.get("mac", ""), + host.get("os_family", ""), json.dumps(ports), + json.dumps(services), now, now), + ) + + for port in ports: + self._conn.execute( + """INSERT OR REPLACE INTO baseline_services + (ip, port, protocol, service, first_seen) + VALUES (?, ?, 'tcp', '', ?)""", + (ip, port, now), + ) + + self._conn.commit() + self._baseline_built = True + + count = len(hosts) + logger.info("Baseline built with %d hosts", count) + return count + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + def _on_host_discovered(self, event) -> None: + """Check a newly discovered host against baseline.""" + if not self._baseline_built: + return + + p = event.payload + ip = p.get("ip", "") + if not ip: + return + + with self._lock: + # Check if host is in baseline + row = self._conn.execute( + "SELECT * FROM baseline_hosts WHERE ip = ?", (ip,) + ).fetchone() + + if not row: + # NEW HOST + self._record_change( + change_type="new_host", + description=f"New host discovered: {ip} " + f"(hostname={p.get('hostname', '')}, " + f"mac={p.get('mac', '')})", + new_value=json.dumps(p), + target_ip=ip, + ) + else: + # Existing host — check for changes + baseline_mac = row["mac"] + new_mac = p.get("mac", "") + if baseline_mac and new_mac and baseline_mac != new_mac: + self._record_change( + change_type="mac_change", + description=f"MAC changed on {ip}: " + f"{baseline_mac} -> {new_mac}", + old_value=baseline_mac, + new_value=new_mac, + target_ip=ip, + severity="warning", + ) + + baseline_os = row["os_family"] + new_os = p.get("os_family", "") + if baseline_os and new_os and baseline_os != new_os: + self._record_change( + change_type="os_change", + description=f"OS changed on {ip}: " + f"{baseline_os} -> {new_os}", + old_value=baseline_os, + new_value=new_os, + target_ip=ip, + ) + + # Check for new ports + try: + baseline_ports = set(json.loads(row["open_ports"])) + except (json.JSONDecodeError, TypeError): + baseline_ports = set() + + new_ports = set(p.get("open_ports", [])) + added_ports = new_ports - baseline_ports + + for port in added_ports: + sev = "info" + desc = f"New port {port} on {ip}" + + # Check if it's a security tool port + if port in _SECURITY_TOOL_PORTS: + tool = _SECURITY_TOOL_PORTS[port] + if tool: + sev = "critical" + desc = f"SECURITY TOOL: {tool} port {port} on {ip}" + + self._record_change( + change_type="new_port", + description=desc, + new_value=str(port), + target_ip=ip, + severity=sev, + ) + + def _on_credential_found(self, event) -> None: + """Monitor for unusual authentication patterns.""" + p = event.payload + # Track rapid auth failures from unknown sources as investigation indicator + if p.get("cred_type") == "auth_failure": + src = p.get("source_ip", "") + target = p.get("target_ip", "") + self._record_change( + change_type="unusual_auth", + description=f"Auth failure: {src} -> {target} " + f"(user={p.get('username', '')})", + target_ip=target, + severity="warning", + ) + + # ------------------------------------------------------------------ + # Connection / scan monitoring + # ------------------------------------------------------------------ + + def ingest_connection_attempt(self, src_ip: str, dst_ip: str, + dst_port: int, timestamp: float = None) -> None: + """Track connection attempts for scan detection. + + Called by passive modules watching network traffic. + """ + ts = timestamp or time.time() + + # Check for scans targeting our implant + if dst_ip in self._implant_ips: + self._record_change( + change_type="scan_detected", + description=f"Connection to implant IP {dst_ip}:{dst_port} " + f"from {src_ip}", + target_ip=dst_ip, + severity="critical", + ) + + # Track port scan patterns + with self._lock: + attempts = self._port_attempts[src_ip] + attempts.append((ts, dst_port)) + + # Prune old entries + cutoff = ts - _SCAN_THRESHOLD_WINDOW + self._port_attempts[src_ip] = [ + (t, p) for t, p in attempts if t > cutoff + ] + + # Check for scan pattern + recent = self._port_attempts[src_ip] + unique_ports = len(set(p for _, p in recent)) + + if unique_ports >= _SCAN_THRESHOLD_PORTS: + self._record_change( + change_type="mass_port_scan", + description=f"Port scan from {src_ip}: " + f"{unique_ports} unique ports in " + f"{_SCAN_THRESHOLD_WINDOW}s", + target_ip=src_ip, + severity="critical", + ) + # Reset to avoid flood + self._port_attempts[src_ip] = [] + + def ingest_dns_query(self, query_name: str, client_ip: str = "") -> None: + """Check DNS queries for security tool investigation indicators.""" + query_lower = query_name.lower() + for pattern in _SECURITY_DNS_PATTERNS: + if pattern in query_lower: + self._record_change( + change_type="security_tool_traffic", + description=f"Security tool DNS query: {query_name} " + f"from {client_ip}", + new_value=query_name, + target_ip=client_ip, + severity="critical", + ) + break + + # ------------------------------------------------------------------ + # Monitor loop + # ------------------------------------------------------------------ + + def _monitor_loop(self) -> None: + """Periodic monitoring: check for disappeared hosts, etc.""" + interval = self.config.get("change_check_interval", 300) + while self._running: + time.sleep(interval) + try: + if self._baseline_built: + self._check_disappeared_hosts() + self._prune_scan_tracking() + except Exception: + logger.exception("Change detector monitor cycle failed") + + def _check_disappeared_hosts(self) -> None: + """Check if any baseline hosts have disappeared.""" + # Get current known hosts from host_discovery + current_json = self.state.get("host_discovery", "discovered_hosts") + if not current_json: + return + + try: + current_hosts = json.loads(current_json) + current_ips = {h.get("ip") for h in current_hosts if h.get("ip")} + except (json.JSONDecodeError, TypeError): + return + + with self._lock: + baseline_rows = self._conn.execute( + "SELECT ip, hostname FROM baseline_hosts" + ).fetchall() + + for row in baseline_rows: + ip = row["ip"] + if ip not in current_ips: + # Check if we already recorded this disappearance recently + recent = self._conn.execute( + """SELECT id FROM changes + WHERE change_type='host_disappeared' AND target_ip=? + AND timestamp > ?""", + (ip, time.time() - 3600), + ).fetchone() + + if not recent: + hostname = row["hostname"] + self._record_change( + change_type="host_disappeared", + description=f"Baseline host disappeared: {ip} " + f"({hostname})", + old_value=ip, + target_ip=ip, + ) + + def _prune_scan_tracking(self) -> None: + """Remove stale scan tracking data to limit memory usage.""" + cutoff = time.time() - (_SCAN_THRESHOLD_WINDOW * 2) + with self._lock: + stale = [ + ip for ip, attempts in self._port_attempts.items() + if not attempts or attempts[-1][0] < cutoff + ] + for ip in stale: + del self._port_attempts[ip] + + # ------------------------------------------------------------------ + # Change recording + # ------------------------------------------------------------------ + + def _record_change(self, change_type: str, description: str, + old_value: str = "", new_value: str = "", + target_ip: str = "", severity: str = None) -> None: + """Record a detected change and publish an event.""" + if severity is None: + severity = _CHANGE_SEVERITY.get(change_type, "info") + + now = time.time() + self._change_count += 1 + + try: + self._conn.execute( + """INSERT INTO changes + (timestamp, change_type, severity, description, + old_value, new_value, target_ip) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (now, change_type, severity, description, + old_value, new_value, target_ip), + ) + self._conn.commit() + except Exception: + logger.exception("Failed to record change") + return + + # Publish event + self.bus.emit("CHANGE_DETECTED", { + "change_type": change_type, + "severity": severity, + "description": description, + "target_ip": target_ip, + "old_value": old_value, + "new_value": new_value, + }, source_module=self.name) + + if severity == "critical": + logger.critical("BURN RISK: %s", description) + elif severity == "warning": + logger.warning("Change: %s", description) + else: + logger.info("Change: %s", description) + + # ------------------------------------------------------------------ + # Query interface + # ------------------------------------------------------------------ + + def get_changes(self, severity: str = None, change_type: str = None, + since: float = 0, limit: int = 100) -> list: + """Query recorded changes with optional filters.""" + clauses = ["timestamp > ?"] + params: list = [since] + + if severity: + clauses.append("severity = ?") + params.append(severity) + if change_type: + clauses.append("change_type = ?") + params.append(change_type) + + where = " AND ".join(clauses) + params.append(limit) + + with self._lock: + rows = self._conn.execute( + f"""SELECT * FROM changes + WHERE {where} + ORDER BY timestamp DESC LIMIT ?""", + params, + ).fetchall() + + return [dict(r) for r in rows] + + def get_critical_changes(self, hours: float = 24.0) -> list: + """Get critical changes in the last N hours.""" + since = time.time() - (hours * 3600) + return self.get_changes(severity="critical", since=since) + + def get_summary(self) -> dict: + """Return change detection summary.""" + with self._lock: + rows = self._conn.execute( + """SELECT severity, COUNT(*) FROM changes + GROUP BY severity""" + ).fetchall() + by_type = self._conn.execute( + """SELECT change_type, COUNT(*) FROM changes + GROUP BY change_type""" + ).fetchall() + + return { + "total": sum(c for _, c in rows), + "by_severity": {s: c for s, c in rows}, + "by_type": {t: c for t, c in by_type}, + "baseline_built": self._baseline_built, + } diff --git a/modules/intel/credential_db.py b/modules/intel/credential_db.py new file mode 100644 index 0000000..3adc422 --- /dev/null +++ b/modules/intel/credential_db.py @@ -0,0 +1,492 @@ +#!/usr/bin/env python3 +"""Central credential store — aggregates credentials from all capture sources. + +Subscribes to CREDENTIAL_FOUND, TICKET_HARVESTED, CLOUD_TOKEN_FOUND events. +Deduplicates on (service, username, cred_type, cred_value). Exports in +hashcat, CSV, and JSON formats. Tracks crack status per credential. +""" + +import csv +import io +import json +import logging +import os +import sqlite3 +import threading +import time +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.credential_db") + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS credentials ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + source_module TEXT NOT NULL, + source_ip TEXT, + target_ip TEXT, + target_port INTEGER, + service TEXT NOT NULL, + username TEXT NOT NULL, + domain TEXT DEFAULT '', + cred_type TEXT NOT NULL, + cred_value TEXT NOT NULL, + hashcat_mode INTEGER, + cracked_value TEXT, + crack_status TEXT DEFAULT 'uncracked', + notes TEXT DEFAULT '', + UNIQUE(service, username, cred_type, cred_value) +); + +CREATE INDEX IF NOT EXISTS idx_creds_service ON credentials(service); +CREATE INDEX IF NOT EXISTS idx_creds_user ON credentials(username); +CREATE INDEX IF NOT EXISTS idx_creds_type ON credentials(cred_type); +CREATE INDEX IF NOT EXISTS idx_creds_status ON credentials(crack_status); +""" + +# Hashcat mode mapping for common credential types +_HASHCAT_MODES = { + "ntlmv2": 5600, + "ntlmv1": 5500, + "ntlm": 1000, + "net-ntlmv2": 5600, + "net-ntlmv1": 5500, + "kerberos_tgs": 13100, # Kerberoast — TGS-REP (RC4) + "kerberos_as": 18200, # AS-REP roast + "kerberos_tgs_aes128": 19600, + "kerberos_tgs_aes256": 19700, + "http_basic": None, # plaintext + "http_digest": 11400, + "snmp_community": None, # plaintext + "mssql": 131, + "mysql": 300, + "postgres": None, + "ftp": None, # plaintext + "telnet": None, # plaintext + "smtp": None, # plaintext + "ldap": None, # plaintext + "vnc": None, # plaintext + "rdp_ntlm": 5600, + "wpa_pmkid": 22000, + "wpa_handshake": 22000, + "cloud_token": None, # not hashable + "ssh_key": None, + "cookie": None, + "jwt": None, + "bearer_token": None, +} + + +class CredentialDB(BaseModule): + """Central credential database — ingests, deduplicates, and exports credentials.""" + + name = "credential_db" + module_type = "intel" + priority = 50 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._db_path = "" + self._conn: Optional[sqlite3.Connection] = None + self._lock = threading.Lock() + self._ingest_count = 0 + self._dedup_count = 0 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "credentials.db") + os.makedirs(os.path.dirname(self._db_path), exist_ok=True) + + self._conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=NORMAL") + self._conn.executescript(_SCHEMA) + + # Subscribe to credential events + self.bus.subscribe(self._on_credential_found, "CREDENTIAL_FOUND") + self.bus.subscribe(self._on_ticket_harvested, "TICKET_HARVESTED") + self.bus.subscribe(self._on_cloud_token, "CLOUD_TOKEN_FOUND") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info("CredentialDB started — db=%s", self._db_path) + + def stop(self) -> None: + if not self._running: + return + self._running = False + + self.bus.unsubscribe(self._on_credential_found, "CREDENTIAL_FOUND") + self.bus.unsubscribe(self._on_ticket_harvested, "TICKET_HARVESTED") + self.bus.unsubscribe(self._on_cloud_token, "CLOUD_TOKEN_FOUND") + + if self._conn: + self._conn.close() + self._conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info( + "CredentialDB stopped — ingested=%d, deduped=%d", + self._ingest_count, self._dedup_count, + ) + + def status(self) -> dict: + total = 0 + cracked = 0 + if self._conn: + try: + with self._lock: + row = self._conn.execute( + "SELECT COUNT(*) FROM credentials" + ).fetchone() + total = row[0] if row else 0 + row = self._conn.execute( + "SELECT COUNT(*) FROM credentials WHERE crack_status='cracked'" + ).fetchone() + cracked = row[0] if row else 0 + except Exception: + pass + + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_credentials": total, + "cracked": cracked, + "ingested": self._ingest_count, + "deduped": self._dedup_count, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + def _on_credential_found(self, event) -> None: + """Handle CREDENTIAL_FOUND events from any capture module.""" + p = event.payload + self._ingest_credential( + source_module=event.source_module or p.get("source_module", "unknown"), + source_ip=p.get("source_ip", ""), + target_ip=p.get("target_ip", ""), + target_port=p.get("target_port"), + service=p.get("service", "unknown"), + username=p.get("username", ""), + domain=p.get("domain", ""), + cred_type=p.get("cred_type", "unknown"), + cred_value=p.get("cred_value", ""), + hashcat_mode=p.get("hashcat_mode"), + notes=p.get("notes", ""), + ) + + def _on_ticket_harvested(self, event) -> None: + """Handle TICKET_HARVESTED events (Kerberos TGS/AS-REP).""" + p = event.payload + cred_type = p.get("ticket_type", "kerberos_tgs") + self._ingest_credential( + source_module=event.source_module or "kerberos_harvester", + source_ip=p.get("source_ip", ""), + target_ip=p.get("target_ip", ""), + target_port=p.get("target_port", 88), + service=p.get("spn", "krbtgt"), + username=p.get("username", ""), + domain=p.get("domain", ""), + cred_type=cred_type, + cred_value=p.get("ticket_hash", ""), + hashcat_mode=_HASHCAT_MODES.get(cred_type), + notes=p.get("notes", f"SPN: {p.get('spn', '')}"), + ) + + def _on_cloud_token(self, event) -> None: + """Handle CLOUD_TOKEN_FOUND events (AWS/Azure/GCP tokens).""" + p = event.payload + self._ingest_credential( + source_module=event.source_module or "cloud_token_harvester", + source_ip=p.get("source_ip", ""), + target_ip=p.get("target_ip", ""), + target_port=p.get("target_port"), + service=p.get("cloud_provider", "cloud"), + username=p.get("identity", ""), + domain=p.get("account_id", ""), + cred_type="cloud_token", + cred_value=p.get("token", ""), + hashcat_mode=None, + notes=p.get("notes", f"Provider: {p.get('cloud_provider', '')}"), + ) + + # ------------------------------------------------------------------ + # Core credential storage + # ------------------------------------------------------------------ + + def _ingest_credential(self, source_module: str, source_ip: str, + target_ip: str, target_port: Optional[int], + service: str, username: str, domain: str, + cred_type: str, cred_value: str, + hashcat_mode: Optional[int] = None, + notes: str = "") -> bool: + """Insert a credential, deduplicating on (service, username, cred_type, cred_value). + + Returns True if a new credential was inserted, False if duplicate. + """ + if not cred_value: + return False + + # Auto-resolve hashcat mode if not provided + if hashcat_mode is None: + hashcat_mode = _HASHCAT_MODES.get(cred_type.lower()) + + self._ingest_count += 1 + + with self._lock: + try: + self._conn.execute( + """INSERT INTO credentials + (timestamp, source_module, source_ip, target_ip, target_port, + service, username, domain, cred_type, cred_value, + hashcat_mode, crack_status, notes) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'uncracked', ?)""", + (time.time(), source_module, source_ip, target_ip, target_port, + service, username, domain, cred_type, cred_value, + hashcat_mode, notes), + ) + self._conn.commit() + logger.info( + "New credential: %s\\%s@%s (%s via %s)", + domain, username, service, cred_type, source_module, + ) + return True + except sqlite3.IntegrityError: + # Duplicate — update notes/source if new info + self._dedup_count += 1 + self._conn.execute( + """UPDATE credentials + SET notes = CASE WHEN notes = '' THEN ? ELSE notes || '; ' || ? END, + source_module = CASE WHEN source_module NOT LIKE '%' || ? || '%' + THEN source_module || ',' || ? ELSE source_module END + WHERE service=? AND username=? AND cred_type=? AND cred_value=?""", + (notes, notes, source_module, source_module, + service, username, cred_type, cred_value), + ) + self._conn.commit() + return False + except Exception: + logger.exception("Failed to ingest credential") + return False + + # ------------------------------------------------------------------ + # Crack status management + # ------------------------------------------------------------------ + + def update_crack_status(self, cred_id: int, status: str, + cracked_value: str = "") -> None: + """Update crack status for a credential. + + Args: + cred_id: Credential ID. + status: One of 'uncracked', 'cracking', 'cracked'. + cracked_value: The plaintext password if cracked. + """ + if status not in ("uncracked", "cracking", "cracked"): + return + + with self._lock: + try: + self._conn.execute( + """UPDATE credentials + SET crack_status=?, cracked_value=? + WHERE id=?""", + (status, cracked_value, cred_id), + ) + self._conn.commit() + except Exception: + logger.exception("Failed to update crack status for ID %d", cred_id) + + def bulk_update_cracked(self, results: dict) -> int: + """Bulk update cracked passwords from hashcat output. + + Args: + results: {hash_value: plaintext_password, ...} + + Returns: + Number of credentials updated. + """ + updated = 0 + with self._lock: + try: + for hash_val, plaintext in results.items(): + cur = self._conn.execute( + """UPDATE credentials + SET crack_status='cracked', cracked_value=? + WHERE cred_value=? AND crack_status != 'cracked'""", + (plaintext, hash_val), + ) + updated += cur.rowcount + self._conn.commit() + except Exception: + logger.exception("Failed to bulk update cracked credentials") + return updated + + # ------------------------------------------------------------------ + # Export methods + # ------------------------------------------------------------------ + + def to_hashcat(self, mode: Optional[int] = None) -> str: + """Export credentials in hashcat format, optionally filtered by mode. + + Args: + mode: Hashcat mode number. If None, export all hashable credentials. + + Returns: + Newline-separated hash strings ready for hashcat. + """ + with self._lock: + if mode is not None: + rows = self._conn.execute( + """SELECT username, domain, cred_value, cred_type + FROM credentials + WHERE hashcat_mode=? AND crack_status='uncracked'""", + (mode,), + ).fetchall() + else: + rows = self._conn.execute( + """SELECT username, domain, cred_value, cred_type + FROM credentials + WHERE hashcat_mode IS NOT NULL AND crack_status='uncracked'""", + ).fetchall() + + lines = [] + for username, domain, cred_value, cred_type in rows: + # NTLMv2 and similar Net-NTLM hashes are already in hashcat format + if cred_type.lower() in ("ntlmv2", "net-ntlmv2", "ntlmv1", "net-ntlmv1"): + lines.append(cred_value) + elif cred_type.lower() in ("kerberos_tgs", "kerberos_as", + "kerberos_tgs_aes128", "kerberos_tgs_aes256"): + lines.append(cred_value) + else: + # Generic: just the hash value + lines.append(cred_value) + + return "\n".join(lines) + + def to_csv(self) -> str: + """Export all credentials as CSV.""" + with self._lock: + rows = self._conn.execute( + """SELECT id, timestamp, source_module, source_ip, target_ip, + target_port, service, username, domain, cred_type, + cred_value, hashcat_mode, cracked_value, crack_status, notes + FROM credentials ORDER BY timestamp""" + ).fetchall() + + output = io.StringIO() + writer = csv.writer(output) + writer.writerow([ + "id", "timestamp", "source_module", "source_ip", "target_ip", + "target_port", "service", "username", "domain", "cred_type", + "cred_value", "hashcat_mode", "cracked_value", "crack_status", "notes", + ]) + for row in rows: + writer.writerow(row) + + return output.getvalue() + + def to_json(self) -> str: + """Export all credentials as JSON.""" + with self._lock: + self._conn.row_factory = sqlite3.Row + rows = self._conn.execute( + """SELECT * FROM credentials ORDER BY timestamp""" + ).fetchall() + self._conn.row_factory = None + + return json.dumps([dict(r) for r in rows], indent=2, default=str) + + def summary(self) -> dict: + """Return a summary of stored credentials.""" + with self._lock: + total = self._conn.execute("SELECT COUNT(*) FROM credentials").fetchone()[0] + by_type = self._conn.execute( + "SELECT cred_type, COUNT(*) FROM credentials GROUP BY cred_type" + ).fetchall() + by_service = self._conn.execute( + "SELECT service, COUNT(*) FROM credentials GROUP BY service" + ).fetchall() + by_status = self._conn.execute( + "SELECT crack_status, COUNT(*) FROM credentials GROUP BY crack_status" + ).fetchall() + unique_users = self._conn.execute( + "SELECT COUNT(DISTINCT username) FROM credentials" + ).fetchone()[0] + unique_hosts = self._conn.execute( + "SELECT COUNT(DISTINCT target_ip) FROM credentials WHERE target_ip != ''" + ).fetchone()[0] + + return { + "total": total, + "unique_users": unique_users, + "unique_hosts": unique_hosts, + "by_type": {t: c for t, c in by_type}, + "by_service": {s: c for s, c in by_service}, + "by_status": {s: c for s, c in by_status}, + } + + # ------------------------------------------------------------------ + # Query helpers + # ------------------------------------------------------------------ + + def get_credentials(self, service: str = None, username: str = None, + cred_type: str = None, limit: int = 100) -> list: + """Query credentials with optional filters.""" + clauses = [] + params = [] + + if service: + clauses.append("service = ?") + params.append(service) + if username: + clauses.append("username = ?") + params.append(username) + if cred_type: + clauses.append("cred_type = ?") + params.append(cred_type) + + where = " AND ".join(clauses) if clauses else "1=1" + params.append(limit) + + with self._lock: + self._conn.row_factory = sqlite3.Row + rows = self._conn.execute( + f"SELECT * FROM credentials WHERE {where} ORDER BY timestamp DESC LIMIT ?", + params, + ).fetchall() + self._conn.row_factory = None + + return [dict(r) for r in rows] + + def get_cracked(self) -> list: + """Return all cracked credentials with plaintext values.""" + with self._lock: + self._conn.row_factory = sqlite3.Row + rows = self._conn.execute( + """SELECT username, domain, service, cracked_value, target_ip + FROM credentials + WHERE crack_status='cracked' AND cracked_value IS NOT NULL + ORDER BY timestamp DESC""" + ).fetchall() + self._conn.row_factory = None + + return [dict(r) for r in rows] diff --git a/modules/intel/net_intel.py b/modules/intel/net_intel.py new file mode 100644 index 0000000..de58d40 --- /dev/null +++ b/modules/intel/net_intel.py @@ -0,0 +1,480 @@ +#!/usr/bin/env python3 +"""Network intelligence — beacon detection, communication graph analysis, +existing C2/malware detection, and service dependency mapping. + +Analyzes traffic patterns to identify beaconing connections, central/isolated +nodes, known-bad indicators, and service dependencies. Feeds anomaly +baselines to traffic_mimicry. +""" + +import json +import logging +import math +import os +import sqlite3 +import threading +import time +from collections import defaultdict +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.net_intel") + +# Known malicious infrastructure patterns +_KNOWN_BAD_PORTS = { + 4444, 5555, 1234, 31337, 8888, 6666, 6667, 6668, 6669, # common RAT/IRC + 4443, 8443, 2222, # common C2 alt ports +} + +# DNS tunneling detection thresholds +_DNS_TUNNEL_MIN_SUBDOMAIN_LEN = 30 # base64-encoded data tends to be long +_DNS_TUNNEL_QUERY_RATE = 50 # queries/min to a single domain = suspicious + +# Beacon detection defaults +_BEACON_INTERVAL_RATIO = 0.15 # stddev/mean < 0.15 = likely beacon +_BEACON_MIN_CONNECTIONS = 10 # need enough samples to be meaningful + + +class NetIntel(BaseModule): + """Analyze network traffic patterns for beacons, C2, anomalies, and dependencies.""" + + name = "net_intel" + module_type = "intel" + priority = 150 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._lock = threading.Lock() + self._analysis_thread: Optional[threading.Thread] = None + + # Connection tracking: (src, dst, port) -> [timestamps] + self._conn_times: dict[tuple, list] = defaultdict(list) + + # Communication graph: ip -> {peers: set, bytes_in, bytes_out} + self._comm_graph: dict[str, dict] = defaultdict( + lambda: {"peers": set(), "bytes_in": 0, "bytes_out": 0, "conn_count": 0} + ) + + # DNS query tracking: domain -> [timestamps] + self._dns_queries: dict[str, list] = defaultdict(list) + + # Detected beacons + self._beacons: list[dict] = [] + + # Service dependency map: service_ip:port -> [client_ips] + self._service_deps: dict[str, set] = defaultdict(set) + + # Anomalies / suspicious findings + self._findings: list[dict] = [] + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self.bus.subscribe(self._on_host_discovered, "HOST_DISCOVERED") + self.bus.subscribe(self._on_beacon_detected, "BEACON_DETECTED") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + # Periodic analysis thread + self._analysis_thread = threading.Thread( + target=self._analysis_loop, daemon=True, + name="bb-netintel-analysis", + ) + self._analysis_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info("NetIntel started") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + self.bus.unsubscribe(self._on_host_discovered, "HOST_DISCOVERED") + self.bus.unsubscribe(self._on_beacon_detected, "BEACON_DETECTED") + + # Persist findings + self._save_findings() + + self.state.set_module_status(self.name, "stopped") + logger.info( + "NetIntel stopped — %d beacons, %d findings", + len(self._beacons), len(self._findings), + ) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "tracked_flows": len(self._conn_times), + "graph_nodes": len(self._comm_graph), + "beacons_detected": len(self._beacons), + "findings": len(self._findings), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Data ingestion + # ------------------------------------------------------------------ + + def ingest_connection(self, src_ip: str, dst_ip: str, dst_port: int, + protocol: str = "tcp", bytes_sent: int = 0, + bytes_recv: int = 0, timestamp: float = None) -> None: + """Ingest a connection record for analysis. + + Called by other modules (dns_logger, packet analyzers) or directly. + """ + ts = timestamp or time.time() + + with self._lock: + # Connection timing for beacon detection + flow_key = (src_ip, dst_ip, dst_port) + self._conn_times[flow_key].append(ts) + + # Communication graph + self._comm_graph[src_ip]["peers"].add(dst_ip) + self._comm_graph[src_ip]["bytes_out"] += bytes_sent + self._comm_graph[src_ip]["conn_count"] += 1 + + self._comm_graph[dst_ip]["peers"].add(src_ip) + self._comm_graph[dst_ip]["bytes_in"] += bytes_recv + self._comm_graph[dst_ip]["conn_count"] += 1 + + # Service dependency + service_key = f"{dst_ip}:{dst_port}" + self._service_deps[service_key].add(src_ip) + + # Known-bad port check + if dst_port in _KNOWN_BAD_PORTS: + self._add_finding( + "suspicious_port", + f"{src_ip} -> {dst_ip}:{dst_port} ({protocol})", + severity="warning", + details={"src": src_ip, "dst": dst_ip, "port": dst_port}, + ) + + def ingest_dns_query(self, client_ip: str, query_name: str, + query_type: str = "A", timestamp: float = None) -> None: + """Ingest a DNS query for tunneling and pattern analysis.""" + ts = timestamp or time.time() + + with self._lock: + self._dns_queries[query_name].append(ts) + + # Check for DNS tunneling indicators + parts = query_name.split(".") + if len(parts) >= 3: + # Long subdomain labels suggest encoding + subdomain = ".".join(parts[:-2]) + if len(subdomain) >= _DNS_TUNNEL_MIN_SUBDOMAIN_LEN: + base_domain = ".".join(parts[-2:]) + self._add_finding( + "dns_tunnel_suspect", + f"Long subdomain to {base_domain} from {client_ip}: " + f"{len(subdomain)} chars", + severity="warning", + details={ + "client": client_ip, + "base_domain": base_domain, + "subdomain_len": len(subdomain), + "query_type": query_type, + }, + ) + + def _on_host_discovered(self, event) -> None: + """Track new hosts for the communication graph.""" + p = event.payload + ip = p.get("ip", "") + if ip and ip not in self._comm_graph: + with self._lock: + self._comm_graph[ip]["peers"] = set() + + def _on_beacon_detected(self, event) -> None: + """Record externally detected beacons.""" + p = event.payload + with self._lock: + self._beacons.append({ + "src": p.get("src_ip", ""), + "dst": p.get("dst_ip", ""), + "port": p.get("port", 0), + "interval": p.get("interval", 0), + "jitter": p.get("jitter", 0), + "detected_at": time.time(), + "source": event.source_module, + }) + + # ------------------------------------------------------------------ + # Analysis engine + # ------------------------------------------------------------------ + + def _analysis_loop(self) -> None: + """Periodic analysis of accumulated traffic data.""" + interval = self.config.get("analysis_interval", 120) + while self._running: + time.sleep(interval) + try: + self._detect_beacons() + self._detect_dns_tunneling() + self._analyze_communication_graph() + self._save_findings() + except Exception: + logger.exception("NetIntel analysis cycle failed") + + def _detect_beacons(self) -> None: + """Detect beaconing connections by analyzing inter-connection intervals. + + A beacon has a regular interval: stddev/mean ratio < threshold. + """ + threshold = self.config.get( + "beacon_ratio_threshold", _BEACON_INTERVAL_RATIO + ) + min_samples = self.config.get( + "beacon_min_connections", _BEACON_MIN_CONNECTIONS + ) + + with self._lock: + for flow_key, timestamps in self._conn_times.items(): + if len(timestamps) < min_samples: + continue + + # Calculate inter-arrival intervals + sorted_ts = sorted(timestamps) + intervals = [ + sorted_ts[i + 1] - sorted_ts[i] + for i in range(len(sorted_ts) - 1) + ] + + if not intervals: + continue + + mean_interval = sum(intervals) / len(intervals) + if mean_interval <= 0: + continue + + variance = sum((x - mean_interval) ** 2 for x in intervals) / len(intervals) + stddev = math.sqrt(variance) + ratio = stddev / mean_interval + + if ratio < threshold: + src, dst, port = flow_key + beacon_info = { + "src": src, + "dst": dst, + "port": port, + "mean_interval": round(mean_interval, 2), + "stddev": round(stddev, 2), + "ratio": round(ratio, 4), + "sample_count": len(timestamps), + "detected_at": time.time(), + } + + # Avoid duplicate beacon alerts + is_dup = any( + b["src"] == src and b["dst"] == dst and b["port"] == port + for b in self._beacons + ) + if not is_dup: + self._beacons.append(beacon_info) + self.bus.emit("BEACON_DETECTED", { + "src_ip": src, + "dst_ip": dst, + "port": port, + "interval": mean_interval, + "jitter": stddev, + "ratio": ratio, + }, source_module=self.name) + + logger.warning( + "BEACON: %s -> %s:%d every %.1fs " + "(ratio=%.4f, n=%d)", + src, dst, port, mean_interval, ratio, + len(timestamps), + ) + + def _detect_dns_tunneling(self) -> None: + """Detect potential DNS tunneling by query rate to specific domains.""" + now = time.time() + window = 60.0 # 1-minute window + + with self._lock: + # Aggregate by base domain (last 2 labels) + domain_rates: dict[str, int] = defaultdict(int) + for query_name, timestamps in self._dns_queries.items(): + parts = query_name.split(".") + if len(parts) >= 2: + base = ".".join(parts[-2:]) + recent = sum(1 for t in timestamps if now - t < window) + domain_rates[base] += recent + + for domain, rate in domain_rates.items(): + if rate >= _DNS_TUNNEL_QUERY_RATE: + self._add_finding( + "dns_tunnel_rate", + f"High DNS query rate to {domain}: {rate}/min", + severity="critical", + details={"domain": domain, "rate_per_min": rate}, + ) + + def _analyze_communication_graph(self) -> None: + """Identify central nodes, isolated hosts, and unusual patterns.""" + with self._lock: + if len(self._comm_graph) < 3: + return + + # Find most-connected nodes (potential central servers / DCs) + by_peers = sorted( + self._comm_graph.items(), + key=lambda x: len(x[1]["peers"]), + reverse=True, + ) + + total_nodes = len(self._comm_graph) + central_threshold = max(total_nodes * 0.3, 5) + + for ip, data in by_peers[:10]: + peer_count = len(data["peers"]) + if peer_count >= central_threshold: + self.state.set( + self.name, + f"central_node_{ip}", + json.dumps({ + "ip": ip, + "peer_count": peer_count, + "bytes_in": data["bytes_in"], + "bytes_out": data["bytes_out"], + }), + ) + + # Find isolated hosts (single or no peers — unusual) + for ip, data in self._comm_graph.items(): + if len(data["peers"]) <= 1 and data["conn_count"] > 20: + self._add_finding( + "isolated_talker", + f"{ip} has {data['conn_count']} connections " + f"but only {len(data['peers'])} peer(s)", + severity="info", + details={"ip": ip, "peers": list(data["peers"])}, + ) + + # ------------------------------------------------------------------ + # Findings management + # ------------------------------------------------------------------ + + def _add_finding(self, finding_type: str, description: str, + severity: str = "info", details: dict = None) -> None: + """Record a finding, deduplicating by type + description.""" + # Simple dedup: don't re-add identical findings + for f in self._findings: + if f["type"] == finding_type and f["description"] == description: + f["last_seen"] = time.time() + f["count"] = f.get("count", 1) + 1 + return + + self._findings.append({ + "type": finding_type, + "description": description, + "severity": severity, + "details": details or {}, + "first_seen": time.time(), + "last_seen": time.time(), + "count": 1, + }) + + def _save_findings(self) -> None: + """Persist findings and beacons to state.""" + with self._lock: + beacons_ser = list(self._beacons) + findings_ser = list(self._findings) + + # Serialize service deps + deps_ser = {k: list(v) for k, v in self._service_deps.items()} + + self.state.set(self.name, "beacons", json.dumps(beacons_ser)) + self.state.set(self.name, "findings", json.dumps(findings_ser)) + self.state.set(self.name, "service_deps", json.dumps(deps_ser)) + + # ------------------------------------------------------------------ + # Public query interface + # ------------------------------------------------------------------ + + def get_beacons(self) -> list: + """Return all detected beacons.""" + with self._lock: + return list(self._beacons) + + def get_findings(self, severity: str = None) -> list: + """Return findings, optionally filtered by severity.""" + with self._lock: + if severity: + return [f for f in self._findings if f["severity"] == severity] + return list(self._findings) + + def get_central_nodes(self, min_peers: int = 5) -> list: + """Return the most-connected nodes in the communication graph.""" + with self._lock: + results = [] + for ip, data in self._comm_graph.items(): + peer_count = len(data["peers"]) + if peer_count >= min_peers: + results.append({ + "ip": ip, + "peer_count": peer_count, + "bytes_in": data["bytes_in"], + "bytes_out": data["bytes_out"], + "conn_count": data["conn_count"], + }) + results.sort(key=lambda x: x["peer_count"], reverse=True) + return results + + def get_service_dependencies(self) -> dict: + """Return service dependency map: service -> [dependent clients].""" + with self._lock: + return {k: list(v) for k, v in self._service_deps.items()} + + def get_anomaly_baseline(self) -> dict: + """Return traffic pattern baseline for traffic_mimicry module. + + Provides: + - Typical connection intervals per flow + - Byte distribution per protocol + - Peak/quiet hour patterns + """ + with self._lock: + flow_intervals = {} + for flow_key, timestamps in self._conn_times.items(): + if len(timestamps) < 3: + continue + sorted_ts = sorted(timestamps) + intervals = [sorted_ts[i + 1] - sorted_ts[i] + for i in range(len(sorted_ts) - 1)] + mean = sum(intervals) / len(intervals) + flow_intervals[f"{flow_key[0]}->{flow_key[1]}:{flow_key[2]}"] = { + "mean_interval": round(mean, 2), + "sample_count": len(timestamps), + } + + # Hour-of-day connection distribution + hourly = defaultdict(int) + for timestamps in self._conn_times.values(): + for ts in timestamps: + hour = time.localtime(ts).tm_hour + hourly[hour] += 1 + + return { + "flow_intervals": flow_intervals, + "hourly_distribution": dict(hourly), + "total_flows": len(self._conn_times), + "total_nodes": len(self._comm_graph), + } diff --git a/modules/intel/operator_audit.py b/modules/intel/operator_audit.py new file mode 100644 index 0000000..e887aa8 --- /dev/null +++ b/modules/intel/operator_audit.py @@ -0,0 +1,412 @@ +#!/usr/bin/env python3 +"""Operator audit trail — append-only log with HMAC integrity chain. + +Logs all operator actions: SSH sessions, CLI commands, module activations, +config changes. Each row includes an HMAC of the previous row, forming +a tamper-evident chain. Subscribes to MODULE_STARTED, MODULE_STOPPED, +and KILL_SWITCH events. +""" + +import hashlib +import hmac +import json +import logging +import os +import sqlite3 +import subprocess +import threading +import time +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.operator_audit") + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + action TEXT NOT NULL, + operator TEXT DEFAULT '', + details TEXT DEFAULT '', + hmac TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(timestamp); +CREATE INDEX IF NOT EXISTS idx_audit_action ON audit_log(action); +""" + +# Action types +ACTION_MODULE_START = "module_start" +ACTION_MODULE_STOP = "module_stop" +ACTION_KILL_SWITCH = "kill_switch" +ACTION_CONFIG_CHANGE = "config_change" +ACTION_SSH_SESSION = "ssh_session" +ACTION_CLI_COMMAND = "cli_command" +ACTION_CRED_EXPORT = "cred_export" +ACTION_DATA_EXFIL = "data_exfil" +ACTION_BASELINE_BUILD = "baseline_build" +ACTION_MANUAL = "manual" + + +class OperatorAudit(BaseModule): + """Append-only operator audit trail with HMAC integrity chain.""" + + name = "operator_audit" + module_type = "intel" + priority = 50 # Critical — always runs + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._db_path = "" + self._conn: Optional[sqlite3.Connection] = None + self._lock = threading.Lock() + self._hmac_key = b"" + self._last_hmac = "" + self._entry_count = 0 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "operator_audit.db") + os.makedirs(os.path.dirname(self._db_path), exist_ok=True) + + self._conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=NORMAL") + self._conn.row_factory = sqlite3.Row + self._conn.executescript(_SCHEMA) + + # HMAC key: from config or derive from machine ID + key_config = self.config.get("audit_hmac_key", "") + if key_config: + self._hmac_key = key_config.encode() if isinstance(key_config, str) else key_config + else: + self._hmac_key = self._derive_machine_key() + + # Load last HMAC from chain + self._last_hmac = self._get_last_hmac() + + # Subscribe to events + self.bus.subscribe(self._on_module_started, "MODULE_STARTED") + self.bus.subscribe(self._on_module_stopped, "MODULE_STOPPED") + self.bus.subscribe(self._on_kill_switch, "KILL_SWITCH") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + self.state.set_module_status(self.name, "running", pid=self._pid) + + # Log our own start + self.log_action( + ACTION_MODULE_START, + details="OperatorAudit started", + ) + + logger.info( + "OperatorAudit started — db=%s, chain_length=%d", + self._db_path, self._entry_count, + ) + + def stop(self) -> None: + if not self._running: + return + + # Log our own stop before shutting down + self.log_action( + ACTION_MODULE_STOP, + details="OperatorAudit stopping", + ) + + self._running = False + + self.bus.unsubscribe(self._on_module_started, "MODULE_STARTED") + self.bus.unsubscribe(self._on_module_stopped, "MODULE_STOPPED") + self.bus.unsubscribe(self._on_kill_switch, "KILL_SWITCH") + + if self._conn: + self._conn.close() + self._conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info( + "OperatorAudit stopped — %d entries in chain", self._entry_count + ) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "entries": self._entry_count, + "chain_valid": self.verify_chain(), + } + + def configure(self, config: dict) -> None: + old_config = json.dumps(self.config, sort_keys=True, default=str) + self.config.update(config) + new_config = json.dumps(self.config, sort_keys=True, default=str) + + if old_config != new_config: + self.log_action( + ACTION_CONFIG_CHANGE, + details=f"Config updated", + ) + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + def _on_module_started(self, event) -> None: + p = event.payload + module_name = p.get("module", event.source_module or "unknown") + if module_name == self.name: + return # Skip our own start event (already logged) + self.log_action( + ACTION_MODULE_START, + details=f"Module started: {module_name} (pid={p.get('pid', '?')})", + ) + + def _on_module_stopped(self, event) -> None: + p = event.payload + module_name = p.get("module", event.source_module or "unknown") + if module_name == self.name: + return + self.log_action( + ACTION_MODULE_STOP, + details=f"Module stopped: {module_name}", + ) + + def _on_kill_switch(self, event) -> None: + p = event.payload + self.log_action( + ACTION_KILL_SWITCH, + details=f"KILL SWITCH activated: reason={p.get('reason', 'unknown')}", + ) + + # ------------------------------------------------------------------ + # Core logging + # ------------------------------------------------------------------ + + def log_action(self, action: str, operator: str = "", + details: str = "") -> None: + """Log an operator action with HMAC chain integrity. + + Args: + action: Action type (use ACTION_* constants). + operator: Operator identifier (SSH key fingerprint, username, etc). + details: Free-text description of the action. + """ + if not self._conn: + return + + if not operator: + operator = self._detect_operator() + + ts = time.time() + + # Compute HMAC: H(key, previous_hmac | timestamp | action | operator | details) + msg = f"{self._last_hmac}|{ts}|{action}|{operator}|{details}" + entry_hmac = hmac.new( + self._hmac_key, msg.encode(), hashlib.sha256 + ).hexdigest() + + with self._lock: + try: + self._conn.execute( + """INSERT INTO audit_log + (timestamp, action, operator, details, hmac) + VALUES (?, ?, ?, ?, ?)""", + (ts, action, operator, details, entry_hmac), + ) + self._conn.commit() + self._last_hmac = entry_hmac + self._entry_count += 1 + except Exception: + logger.exception("Failed to log audit action: %s", action) + + def log_ssh_session(self, remote_ip: str, key_fingerprint: str = "", + event: str = "connect") -> None: + """Log an SSH session event.""" + self.log_action( + ACTION_SSH_SESSION, + operator=key_fingerprint or remote_ip, + details=f"SSH {event} from {remote_ip} " + f"(key={key_fingerprint or 'password'})", + ) + + def log_cli_command(self, command: str, operator: str = "") -> None: + """Log a CLI command execution.""" + self.log_action( + ACTION_CLI_COMMAND, + operator=operator, + details=f"Command: {command}", + ) + + # ------------------------------------------------------------------ + # HMAC chain + # ------------------------------------------------------------------ + + def _derive_machine_key(self) -> bytes: + """Derive HMAC key from machine-specific identifiers.""" + components = [] + + # CPU serial (RPi) + try: + with open("/proc/cpuinfo", "r") as f: + for line in f: + if line.startswith("Serial"): + components.append(line.strip()) + break + except (IOError, PermissionError): + pass + + # Machine ID + try: + with open("/etc/machine-id", "r") as f: + components.append(f.read().strip()) + except (IOError, PermissionError): + pass + + if not components: + components.append("bigbrother-default-key") + + combined = "|".join(components) + return hashlib.sha256(combined.encode()).digest() + + def _get_last_hmac(self) -> str: + """Get the HMAC of the last entry in the chain.""" + with self._lock: + row = self._conn.execute( + "SELECT hmac FROM audit_log ORDER BY id DESC LIMIT 1" + ).fetchone() + + count_row = self._conn.execute( + "SELECT COUNT(*) FROM audit_log" + ).fetchone() + self._entry_count = count_row[0] if count_row else 0 + + if row: + return row["hmac"] + return "" # Genesis — empty for first entry + + def _detect_operator(self) -> str: + """Attempt to detect the current operator identity.""" + # Check SSH connection + ssh_client = os.environ.get("SSH_CLIENT", "") + if ssh_client: + parts = ssh_client.split() + remote_ip = parts[0] if parts else "unknown" + + # Try to get SSH key fingerprint + try: + result = subprocess.run( + ["who", "-m"], + capture_output=True, timeout=2, + ) + who = result.stdout.decode().strip() + if who: + return f"ssh:{remote_ip}:{who.split()[0]}" + except Exception: + pass + + return f"ssh:{remote_ip}" + + # Local operator + user = os.environ.get("USER", os.environ.get("LOGNAME", "local")) + return f"local:{user}" + + # ------------------------------------------------------------------ + # Chain verification + # ------------------------------------------------------------------ + + def verify_chain(self) -> bool: + """Verify the entire HMAC chain integrity. + + Returns: + True if the chain is valid, False if tampered. + """ + if not self._conn: + return False + + with self._lock: + rows = self._conn.execute( + "SELECT timestamp, action, operator, details, hmac " + "FROM audit_log ORDER BY id ASC" + ).fetchall() + + if not rows: + return True # Empty chain is valid + + prev_hmac = "" + for row in rows: + ts = row["timestamp"] + action = row["action"] + operator = row["operator"] + details = row["details"] + stored_hmac = row["hmac"] + + msg = f"{prev_hmac}|{ts}|{action}|{operator}|{details}" + computed = hmac.new( + self._hmac_key, msg.encode(), hashlib.sha256 + ).hexdigest() + + if computed != stored_hmac: + logger.error( + "HMAC chain broken at ts=%.3f action=%s — " + "expected %s, got %s", + ts, action, computed[:16], stored_hmac[:16], + ) + return False + + prev_hmac = stored_hmac + + return True + + # ------------------------------------------------------------------ + # Export + # ------------------------------------------------------------------ + + def export_log(self, since: float = 0, limit: int = 10000) -> list: + """Export audit log entries for engagement reports.""" + with self._lock: + rows = self._conn.execute( + """SELECT * FROM audit_log + WHERE timestamp > ? + ORDER BY id ASC LIMIT ?""", + (since, limit), + ).fetchall() + + return [dict(r) for r in rows] + + def export_json(self, since: float = 0) -> str: + """Export audit log as JSON string.""" + entries = self.export_log(since) + return json.dumps(entries, indent=2, default=str) + + def get_summary(self) -> dict: + """Return audit log summary.""" + with self._lock: + rows = self._conn.execute( + """SELECT action, COUNT(*) FROM audit_log + GROUP BY action ORDER BY COUNT(*) DESC""" + ).fetchall() + + operators = self._conn.execute( + """SELECT DISTINCT operator FROM audit_log + WHERE operator != ''""" + ).fetchall() + + return { + "total_entries": self._entry_count, + "chain_valid": self.verify_chain(), + "by_action": {a: c for a, c in rows}, + "operators": [r[0] for r in operators], + } diff --git a/modules/intel/security_posture.py b/modules/intel/security_posture.py new file mode 100644 index 0000000..a5343c5 --- /dev/null +++ b/modules/intel/security_posture.py @@ -0,0 +1,577 @@ +#!/usr/bin/env python3 +"""Security posture mapper — detect security infrastructure by traffic patterns. + +Identifies EDR, SIEM, vulnerability scanners, honeypots, NAC, and network +monitoring tools by observing DNS queries, HTTP traffic, and port patterns. +Gates active module risk assessment — if strong security tooling is detected, +active operations should increase caution. +""" + +import json +import logging +import os +import re +import sqlite3 +import threading +import time +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.security_posture") + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS security_tools ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + tool_type TEXT NOT NULL, + tool_name TEXT NOT NULL, + evidence TEXT NOT NULL, + confidence REAL DEFAULT 0.5, + target_ip TEXT DEFAULT '', + target_hostname TEXT DEFAULT '', + first_seen REAL NOT NULL, + last_seen REAL NOT NULL, + impacts_active_ops INTEGER DEFAULT 1, + details TEXT DEFAULT '', + UNIQUE(tool_type, tool_name, target_ip) +); + +CREATE INDEX IF NOT EXISTS idx_st_type ON security_tools(tool_type); +CREATE INDEX IF NOT EXISTS idx_st_name ON security_tools(tool_name); +""" + +# ----------------------------------------------------------------------- +# Detection signatures +# ----------------------------------------------------------------------- + +# DNS patterns: (regex, tool_type, tool_name, confidence, impacts_active) +_DNS_SIGNATURES = [ + # EDR / Endpoint Protection + (re.compile(r"ts01-[a-z0-9]+\.cloudsink\.net$", re.I), + "edr", "CrowdStrike Falcon", 0.95, True), + (re.compile(r"(lfodown|lfoup)\.crowdstrike\.com$", re.I), + "edr", "CrowdStrike Falcon", 0.95, True), + (re.compile(r"\.crowdstrike\.com$", re.I), + "edr", "CrowdStrike Falcon", 0.9, True), + (re.compile(r"[a-z0-9]+\.sentinelone\.net$", re.I), + "edr", "SentinelOne", 0.95, True), + (re.compile(r"management.*\.sentinelone\.net$", re.I), + "edr", "SentinelOne", 0.95, True), + (re.compile(r"\.cybereason\.com$", re.I), + "edr", "Cybereason", 0.9, True), + (re.compile(r"\.carbonblack\.(com|io)$", re.I), + "edr", "Carbon Black", 0.9, True), + (re.compile(r"\.cb\.carbonblack\.io$", re.I), + "edr", "Carbon Black Cloud", 0.95, True), + (re.compile(r"(wdcp|wdcpalt)\.microsoft\.com$", re.I), + "edr", "Microsoft Defender ATP", 0.85, True), + (re.compile(r"\.securitycenter\.windows\.com$", re.I), + "edr", "Microsoft Defender ATP", 0.9, True), + (re.compile(r"\.endpoint\.security\.microsoft\.com$", re.I), + "edr", "Microsoft Defender for Endpoint", 0.95, True), + (re.compile(r"\.sophos\.(com|net)$", re.I), + "edr", "Sophos", 0.8, True), + (re.compile(r"\.cylance\.(com|io)$", re.I), + "edr", "Cylance/BlackBerry Protect", 0.85, True), + (re.compile(r"\.trellix\.com$", re.I), + "edr", "Trellix (McAfee/FireEye)", 0.85, True), + (re.compile(r"\.fireeye\.com$", re.I), + "edr", "FireEye/Mandiant", 0.85, True), + + # SIEM + (re.compile(r"\.splunkcloud\.com$", re.I), + "siem", "Splunk Cloud", 0.9, False), + (re.compile(r"input.*\.splunkcloud\.com$", re.I), + "siem", "Splunk Cloud HEC", 0.95, False), + (re.compile(r"\.sumologic\.com$", re.I), + "siem", "Sumo Logic", 0.85, False), + (re.compile(r"\.logz\.io$", re.I), + "siem", "Logz.io", 0.85, False), + (re.compile(r"\.elastic-cloud\.com$", re.I), + "siem", "Elastic Cloud", 0.85, False), + (re.compile(r"\.datadoghq\.com$", re.I), + "siem", "Datadog", 0.8, False), + + # Vulnerability Scanners + (re.compile(r"(plugins|sensor)\.nessus\.org$", re.I), + "vuln_scanner", "Nessus", 0.9, False), + (re.compile(r"\.tenable\.(com|io)$", re.I), + "vuln_scanner", "Tenable", 0.85, False), + (re.compile(r"\.qualys\.com$", re.I), + "vuln_scanner", "Qualys", 0.9, False), + (re.compile(r"\.rapid7\.com$", re.I), + "vuln_scanner", "Rapid7 InsightVM", 0.85, False), + + # Honeypots / Deception + (re.compile(r"\.canary\.tools$", re.I), + "honeypot", "Thinkst Canary", 0.95, True), + (re.compile(r"canarytokens\..*\.com$", re.I), + "honeypot", "Canary Tokens", 0.9, True), + (re.compile(r"\.illusive-networks\.com$", re.I), + "honeypot", "Illusive Networks", 0.85, True), + (re.compile(r"\.attivo\.com$", re.I), + "honeypot", "Attivo Networks", 0.85, True), + + # NAC + (re.compile(r"\.forescout\.com$", re.I), + "nac", "Forescout", 0.85, True), + (re.compile(r"ise[\w-]*\.(local|internal|corp|lan)$", re.I), + "nac", "Cisco ISE", 0.8, True), + + # Threat Intelligence + (re.compile(r"\.virustotal\.com$", re.I), + "threat_intel", "VirusTotal", 0.7, False), + (re.compile(r"\.shodan\.io$", re.I), + "threat_intel", "Shodan", 0.7, False), +] + +# Port-based signatures: port -> (tool_type, tool_name, confidence, impacts_active) +_PORT_SIGNATURES = { + 8088: ("siem", "Splunk HEC", 0.7, False), + 8089: ("siem", "Splunk Management", 0.7, False), + 9997: ("siem", "Splunk Forwarder", 0.75, False), + 1514: ("siem", "Wazuh Agent", 0.7, False), + 1515: ("siem", "Wazuh Enrollment", 0.75, False), + 55000: ("siem", "Wazuh API", 0.8, False), + 8834: ("vuln_scanner", "Nessus Scanner", 0.85, False), + 3790: ("vuln_scanner", "Rapid7 Metasploit Pro/Nexpose", 0.75, False), + 9390: ("vuln_scanner", "OpenVAS/Greenbone", 0.8, False), + 9392: ("vuln_scanner", "Greenbone GSA", 0.8, False), + 1812: ("nac", "RADIUS Authentication", 0.6, True), + 1813: ("nac", "RADIUS Accounting", 0.6, True), + 3799: ("nac", "RADIUS CoA (Change of Authorization)", 0.7, True), +} + +# HTTP User-Agent patterns +_UA_SIGNATURES = [ + (re.compile(r"Nessus", re.I), "vuln_scanner", "Nessus", 0.9), + (re.compile(r"Qualys", re.I), "vuln_scanner", "Qualys", 0.9), + (re.compile(r"InsightVM|Rapid7", re.I), "vuln_scanner", "Rapid7", 0.85), + (re.compile(r"Nikto", re.I), "vuln_scanner", "Nikto", 0.8), + (re.compile(r"OpenVAS", re.I), "vuln_scanner", "OpenVAS", 0.85), + (re.compile(r"CrowdStrike", re.I), "edr", "CrowdStrike", 0.7), + (re.compile(r"Splunk", re.I), "siem", "Splunk", 0.7), +] + +# T-Pot honeypot: multiple decoy services on unusual port ranges +_TPOT_PORT_CLUSTERS = { + # T-Pot maps real honeypots to high ports; seeing many of these is a sign + 64295, 64297, 64300, # T-Pot management +} + + +class SecurityPosture(BaseModule): + """Map the target network's security tooling and posture.""" + + name = "security_posture" + module_type = "intel" + priority = 100 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._db_path = "" + self._conn: Optional[sqlite3.Connection] = None + self._lock = threading.Lock() + self._scan_thread: Optional[threading.Thread] = None + self._detection_count = 0 + + # Cache of risk assessment + self._risk_level = "unknown" # low, medium, high, critical + self._active_ops_safe = True + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "security_posture.db") + os.makedirs(os.path.dirname(self._db_path), exist_ok=True) + + self._conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=NORMAL") + self._conn.row_factory = sqlite3.Row + self._conn.executescript(_SCHEMA) + + # Subscribe to host discovery for port-based detection + self.bus.subscribe(self._on_host_discovered, "HOST_DISCOVERED") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._scan_thread = threading.Thread( + target=self._periodic_scan, daemon=True, + name="bb-security-posture", + ) + self._scan_thread.start() + + # Initial risk assessment from any previously stored data + self._reassess_risk() + + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info("SecurityPosture started — db=%s", self._db_path) + + def stop(self) -> None: + if not self._running: + return + self._running = False + + self.bus.unsubscribe(self._on_host_discovered, "HOST_DISCOVERED") + + if self._conn: + self._conn.close() + self._conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info( + "SecurityPosture stopped — %d tools detected, risk=%s", + self._detection_count, self._risk_level, + ) + + def status(self) -> dict: + tool_count = 0 + if self._conn: + try: + with self._lock: + row = self._conn.execute( + "SELECT COUNT(*) FROM security_tools" + ).fetchone() + tool_count = row[0] if row else 0 + except Exception: + pass + + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "detected_tools": tool_count, + "risk_level": self._risk_level, + "active_ops_safe": self._active_ops_safe, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Detection functions + # ------------------------------------------------------------------ + + def analyze_dns_query(self, query_name: str, client_ip: str = "", + resolved_ip: str = "") -> Optional[dict]: + """Analyze a DNS query for security tool signatures.""" + for pattern, tool_type, tool_name, confidence, impacts in _DNS_SIGNATURES: + if pattern.search(query_name): + det = self._record_detection( + tool_type=tool_type, + tool_name=tool_name, + evidence=f"DNS: {query_name}", + confidence=confidence, + target_ip=client_ip, + target_hostname=query_name, + impacts_active=impacts, + details=json.dumps({ + "query": query_name, + "client_ip": client_ip, + "resolved_ip": resolved_ip, + }), + ) + return det + return None + + def analyze_port(self, target_ip: str, port: int, + hostname: str = "") -> Optional[dict]: + """Check if a port suggests security tooling.""" + if port in _PORT_SIGNATURES: + tool_type, tool_name, confidence, impacts = _PORT_SIGNATURES[port] + return self._record_detection( + tool_type=tool_type, + tool_name=tool_name, + evidence=f"Port {port} open on {target_ip}", + confidence=confidence, + target_ip=target_ip, + target_hostname=hostname, + impacts_active=impacts, + ) + + # T-Pot cluster detection + if port in _TPOT_PORT_CLUSTERS: + return self._record_detection( + tool_type="honeypot", + tool_name="T-Pot", + evidence=f"T-Pot management port {port} on {target_ip}", + confidence=0.8, + target_ip=target_ip, + target_hostname=hostname, + impacts_active=True, + ) + + return None + + def analyze_user_agent(self, user_agent: str, source_ip: str = "") -> Optional[dict]: + """Check HTTP User-Agent for security scanner signatures.""" + for pattern, tool_type, tool_name, confidence in _UA_SIGNATURES: + if pattern.search(user_agent): + return self._record_detection( + tool_type=tool_type, + tool_name=tool_name, + evidence=f"User-Agent: {user_agent[:100]}", + confidence=confidence, + target_ip=source_ip, + impacts_active=False, + details=json.dumps({"user_agent": user_agent}), + ) + return None + + def analyze_radius_traffic(self, src_ip: str, dst_ip: str, + dst_port: int) -> Optional[dict]: + """Detect 802.1X/RADIUS NAC infrastructure.""" + if dst_port in (1812, 1813, 3799): + return self._record_detection( + tool_type="nac", + tool_name="RADIUS/802.1X", + evidence=f"RADIUS traffic {src_ip} -> {dst_ip}:{dst_port}", + confidence=0.8, + target_ip=dst_ip, + impacts_active=True, + details=json.dumps({ + "src": src_ip, "dst": dst_ip, "port": dst_port + }), + ) + return None + + def _on_host_discovered(self, event) -> None: + """Check discovered host ports for security tools.""" + p = event.payload + ip = p.get("ip", "") + hostname = p.get("hostname", "") + ports = p.get("open_ports", []) + + for port in ports: + self.analyze_port(ip, port, hostname) + + # ------------------------------------------------------------------ + # Database operations + # ------------------------------------------------------------------ + + def _record_detection(self, tool_type: str, tool_name: str, + evidence: str, confidence: float, + target_ip: str = "", target_hostname: str = "", + impacts_active: bool = True, + details: str = "") -> dict: + """Record a security tool detection.""" + now = time.time() + detection = { + "tool_type": tool_type, + "tool_name": tool_name, + "evidence": evidence, + "confidence": confidence, + "target_ip": target_ip, + "impacts_active_ops": impacts_active, + } + + with self._lock: + try: + self._conn.execute( + """INSERT INTO security_tools + (tool_type, tool_name, evidence, confidence, + target_ip, target_hostname, first_seen, last_seen, + impacts_active_ops, details) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(tool_type, tool_name, target_ip) DO UPDATE SET + last_seen = excluded.last_seen, + confidence = MAX(security_tools.confidence, excluded.confidence), + evidence = security_tools.evidence || '; ' || excluded.evidence + """, + (tool_type, tool_name, evidence, confidence, + target_ip, target_hostname, now, now, + 1 if impacts_active else 0, details), + ) + self._conn.commit() + self._detection_count += 1 + except Exception: + logger.exception("Failed to record security tool detection") + return detection + + logger.warning( + "Security tool detected: %s (%s) at %s — confidence=%.2f, " + "impacts_active=%s", + tool_name, tool_type, target_ip or "unknown", confidence, + impacts_active, + ) + + # Reassess risk after each new detection + self._reassess_risk() + + return detection + + # ------------------------------------------------------------------ + # Risk assessment + # ------------------------------------------------------------------ + + def _reassess_risk(self) -> None: + """Recalculate overall security risk level based on detected tools.""" + if not self._conn: + return + + with self._lock: + try: + tools = self._conn.execute( + "SELECT tool_type, tool_name, confidence, impacts_active_ops " + "FROM security_tools" + ).fetchall() + except Exception: + return + + if not tools: + self._risk_level = "low" + self._active_ops_safe = True + return + + has_edr = any(t["tool_type"] == "edr" and t["confidence"] >= 0.7 + for t in tools) + has_siem = any(t["tool_type"] == "siem" and t["confidence"] >= 0.7 + for t in tools) + has_nac = any(t["tool_type"] == "nac" and t["confidence"] >= 0.7 + for t in tools) + has_honeypot = any(t["tool_type"] == "honeypot" and t["confidence"] >= 0.7 + for t in tools) + has_vuln_scanner = any(t["tool_type"] == "vuln_scanner" + and t["confidence"] >= 0.7 for t in tools) + + # Score-based risk assessment + score = 0 + if has_edr: + score += 40 + if has_siem: + score += 20 + if has_nac: + score += 25 + if has_honeypot: + score += 30 + if has_vuln_scanner: + score += 10 + + if score >= 60: + self._risk_level = "critical" + self._active_ops_safe = False + elif score >= 40: + self._risk_level = "high" + self._active_ops_safe = False + elif score >= 20: + self._risk_level = "medium" + self._active_ops_safe = True # with caution + else: + self._risk_level = "low" + self._active_ops_safe = True + + # Persist for other modules + self.state.set(self.name, "risk_level", self._risk_level) + self.state.set(self.name, "active_ops_safe", json.dumps(self._active_ops_safe)) + self.state.set(self.name, "has_edr", json.dumps(has_edr)) + self.state.set(self.name, "has_siem", json.dumps(has_siem)) + self.state.set(self.name, "has_nac", json.dumps(has_nac)) + self.state.set(self.name, "has_honeypot", json.dumps(has_honeypot)) + + # ------------------------------------------------------------------ + # Periodic scanning + # ------------------------------------------------------------------ + + def _periodic_scan(self) -> None: + """Pull DNS data from other modules and analyze.""" + interval = self.config.get("posture_scan_interval", 120) + while self._running: + time.sleep(interval) + try: + self._scan_dns_data() + self._reassess_risk() + except Exception: + logger.exception("Security posture scan failed") + + def _scan_dns_data(self) -> None: + """Pull recent DNS queries from dns_logger and analyze for security tools.""" + last_ts = self.state.get(self.name, "dns_scan_last_ts") + last_ts = float(last_ts) if last_ts else 0 + + dns_json = self.state.get("dns_logger", "recent_queries") + if not dns_json: + return + + try: + queries = json.loads(dns_json) + max_ts = last_ts + for q in queries: + ts = q.get("timestamp", 0) + if ts <= last_ts: + continue + self.analyze_dns_query( + query_name=q.get("query_name", ""), + client_ip=q.get("client_ip", ""), + resolved_ip=q.get("resolved_ip", ""), + ) + max_ts = max(max_ts, ts) + + self.state.set(self.name, "dns_scan_last_ts", str(max_ts)) + except (json.JSONDecodeError, TypeError): + pass + + # ------------------------------------------------------------------ + # Query interface + # ------------------------------------------------------------------ + + def get_risk_level(self) -> str: + """Return current overall risk level.""" + return self._risk_level + + def is_active_ops_safe(self) -> bool: + """Return whether active operations are considered safe.""" + return self._active_ops_safe + + def get_detected_tools(self, tool_type: str = None) -> list: + """Get all detected security tools.""" + with self._lock: + if tool_type: + rows = self._conn.execute( + """SELECT * FROM security_tools + WHERE tool_type = ? + ORDER BY confidence DESC""", + (tool_type,), + ).fetchall() + else: + rows = self._conn.execute( + """SELECT * FROM security_tools + ORDER BY confidence DESC""" + ).fetchall() + + return [dict(r) for r in rows] + + def get_posture_summary(self) -> dict: + """Return a full security posture summary.""" + with self._lock: + rows = self._conn.execute( + """SELECT tool_type, COUNT(*) as count, + AVG(confidence) as avg_confidence, + MAX(impacts_active_ops) as any_impacts_active + FROM security_tools + GROUP BY tool_type + ORDER BY count DESC""" + ).fetchall() + + return { + "risk_level": self._risk_level, + "active_ops_safe": self._active_ops_safe, + "total_tools": sum(r[1] for r in rows), + "by_type": { + r[0]: { + "count": r[1], + "avg_confidence": round(r[2], 2), + "impacts_active": bool(r[3]), + } + for r in rows + }, + } diff --git a/modules/intel/supply_chain_detect.py b/modules/intel/supply_chain_detect.py new file mode 100644 index 0000000..c9a235f --- /dev/null +++ b/modules/intel/supply_chain_detect.py @@ -0,0 +1,491 @@ +#!/usr/bin/env python3 +"""Supply chain infrastructure detection — passively identifies internal +package repos, update servers, config management, CI/CD, and container +registries by analyzing DNS and HTTP traffic patterns. + +Detection only — no exploitation or active probing. +""" + +import json +import logging +import os +import re +import sqlite3 +import threading +import time +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.supply_chain_detect") + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS supply_chain ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + type TEXT NOT NULL, + target_ip TEXT NOT NULL, + target_hostname TEXT DEFAULT '', + target_port INTEGER, + evidence TEXT NOT NULL, + confidence REAL DEFAULT 0.5, + first_seen REAL NOT NULL, + last_seen REAL NOT NULL, + details TEXT DEFAULT '', + UNIQUE(type, target_ip, target_port) +); + +CREATE INDEX IF NOT EXISTS idx_sc_type ON supply_chain(type); +CREATE INDEX IF NOT EXISTS idx_sc_ip ON supply_chain(target_ip); +""" + +# ----------------------------------------------------------------------- +# Detection signatures +# ----------------------------------------------------------------------- + +# HTTP path patterns -> (infra_type, description, confidence) +_HTTP_PATH_PATTERNS = [ + # Package repositories + (re.compile(r"/simple/[\w-]+/"), "pypi_mirror", "PyPI mirror (PEP 503 simple API)", 0.9), + (re.compile(r"/pypi/[\w-]+/json"), "pypi_mirror", "PyPI mirror (JSON API)", 0.9), + (re.compile(r"/-/npm/v1/"), "npm_registry", "npm registry (v1 API)", 0.9), + (re.compile(r"/@[\w-]+/[\w-]+/-/"), "npm_registry", "npm scoped package download", 0.85), + (re.compile(r"/repository/maven-"), "maven_repo", "Maven/Nexus repository", 0.85), + (re.compile(r"/artifactory/"), "artifactory", "JFrog Artifactory", 0.9), + (re.compile(r"/content/repositories/"), "nexus_repo", "Sonatype Nexus repository", 0.85), + (re.compile(r"/api/v4/projects/\d+/packages/"), "gitlab_packages", "GitLab Package Registry", 0.9), + (re.compile(r"/v2/[\w.-]+/[\w.-]+/manifests/"), "container_registry", "Docker/OCI registry (manifest pull)", 0.95), + (re.compile(r"/v2/[\w.-]+/[\w.-]+/blobs/"), "container_registry", "Docker/OCI registry (blob pull)", 0.95), + (re.compile(r"/v2/_catalog"), "container_registry", "Docker registry catalog", 0.95), + + # WSUS + (re.compile(r"/ClientWebService/"), "wsus", "WSUS client web service", 0.9), + (re.compile(r"/SimpleAuthWebService/"), "wsus", "WSUS auth service", 0.9), + (re.compile(r"/ApiRemoting30/"), "wsus", "WSUS API remoting", 0.9), + (re.compile(r"/Content/[\w-]+\.cab"), "wsus", "WSUS content download", 0.8), + + # CI/CD + (re.compile(r"/job/[\w-]+/build"), "jenkins", "Jenkins build trigger", 0.9), + (re.compile(r"/api/json\?tree="), "jenkins", "Jenkins API query", 0.85), + (re.compile(r"/queue/api/json"), "jenkins", "Jenkins queue API", 0.85), + (re.compile(r"/api/v4/runners/"), "gitlab_ci", "GitLab CI runner API", 0.9), + (re.compile(r"/api/v3/repos/.+/actions/"), "github_actions", "GitHub Actions API", 0.85), + (re.compile(r"/_apis/build/"), "azure_devops", "Azure DevOps build API", 0.9), + + # Config management + (re.compile(r"/puppet/v3/"), "puppet", "Puppet v3 API", 0.9), + (re.compile(r"/puppet/v4/"), "puppet", "Puppet v4 API", 0.9), + (re.compile(r"/organizations/[\w-]+/cookbooks/"), "chef", "Chef cookbook API", 0.9), + (re.compile(r"/nodes/[\w.-]+"), "chef", "Chef node API", 0.8), + + # Internal CA + (re.compile(r"/certsrv/"), "internal_ca", "Microsoft AD CS certificate services", 0.9), + (re.compile(r"/acme/directory"), "internal_ca", "ACME CA directory (internal Let's Encrypt/step-ca)", 0.8), + (re.compile(r"/v1/pki/"), "vault_pki", "HashiCorp Vault PKI engine", 0.9), + + # SCCM / MECM + (re.compile(r"/SMS_MP/"), "sccm", "SCCM Management Point", 0.9), + (re.compile(r"/CCM_Client/"), "sccm", "SCCM client endpoint", 0.9), + (re.compile(r"/CMGateway/"), "sccm", "SCCM Cloud Management Gateway", 0.85), +] + +# DNS patterns -> (infra_type, description, confidence) +_DNS_PATTERNS = [ + (re.compile(r"(pypi|pip)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "pypi_mirror", "Internal PyPI mirror hostname", 0.85), + (re.compile(r"(npm|registry)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "npm_registry", "Internal npm registry hostname", 0.85), + (re.compile(r"(nexus|artifactory|repo)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "artifact_repo", "Internal artifact repository", 0.8), + (re.compile(r"(docker|registry|harbor|quay)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "container_registry", "Internal container registry", 0.85), + (re.compile(r"(wsus|update)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "wsus", "WSUS/update server", 0.8), + (re.compile(r"(jenkins|ci|build|drone|bamboo|teamcity|concourse)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "ci_cd", "CI/CD server", 0.8), + (re.compile(r"(gitlab)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "gitlab", "GitLab instance", 0.9), + (re.compile(r"(puppet|puppetdb|foreman)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "puppet", "Puppet infrastructure", 0.85), + (re.compile(r"(chef|supermarket)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "chef", "Chef infrastructure", 0.85), + (re.compile(r"(ansible|tower|awx)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "ansible", "Ansible Tower/AWX", 0.85), + (re.compile(r"(sccm|mecm|configmgr)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "sccm", "SCCM/MECM infrastructure", 0.9), + (re.compile(r"(ca|pki|cert|issuing)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "internal_ca", "Internal CA/PKI", 0.7), + (re.compile(r"(vault)[\w-]*\.[\w.]+\.(local|internal|corp|lan)$", re.I), + "hashicorp_vault", "HashiCorp Vault", 0.8), +] + +# Port patterns -> (infra_type, description, confidence) +_PORT_PATTERNS = { + 8140: ("puppet", "Puppet master default port", 0.7), + 8443: ("ci_cd", "Common CI/CD HTTPS port", 0.3), # low confidence — many things use 8443 + 8080: ("jenkins", "Jenkins default HTTP port", 0.3), + 50000: ("jenkins", "Jenkins agent JNLP port", 0.7), + 5000: ("container_registry", "Docker registry default port", 0.5), + 443: ("generic_https", "HTTPS", 0.1), # too common for high confidence alone + 8140: ("puppet", "Puppet master", 0.7), + 4433: ("chef", "Chef Automate", 0.6), +} + + +class SupplyChainDetect(BaseModule): + """Passively detect internal supply chain infrastructure.""" + + name = "supply_chain_detect" + module_type = "intel" + priority = 200 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._db_path = "" + self._conn: Optional[sqlite3.Connection] = None + self._lock = threading.Lock() + self._scan_thread: Optional[threading.Thread] = None + self._detection_count = 0 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "supply_chain.db") + os.makedirs(os.path.dirname(self._db_path), exist_ok=True) + + self._conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=NORMAL") + self._conn.row_factory = sqlite3.Row + self._conn.executescript(_SCHEMA) + + # Subscribe to host discovery for port-based detection + self.bus.subscribe(self._on_host_discovered, "HOST_DISCOVERED") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + # Periodic scan of accumulated DNS and HTTP data + self._scan_thread = threading.Thread( + target=self._periodic_scan, daemon=True, + name="bb-supply-chain-scan", + ) + self._scan_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info("SupplyChainDetect started — db=%s", self._db_path) + + def stop(self) -> None: + if not self._running: + return + self._running = False + + self.bus.unsubscribe(self._on_host_discovered, "HOST_DISCOVERED") + + if self._conn: + self._conn.close() + self._conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("SupplyChainDetect stopped — %d detections", self._detection_count) + + def status(self) -> dict: + det_count = 0 + if self._conn: + try: + with self._lock: + row = self._conn.execute( + "SELECT COUNT(*) FROM supply_chain" + ).fetchone() + det_count = row[0] if row else 0 + except Exception: + pass + + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "detections": det_count, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Detection functions + # ------------------------------------------------------------------ + + def analyze_http_request(self, target_ip: str, target_port: int, + method: str, path: str, host_header: str = "", + user_agent: str = "") -> Optional[dict]: + """Analyze an HTTP request for supply chain infrastructure indicators. + + Called by tool_output_parser or directly by HTTP monitoring modules. + + Returns: + Detection dict if found, None otherwise. + """ + for pattern, infra_type, description, confidence in _HTTP_PATH_PATTERNS: + if pattern.search(path): + det = self._record_detection( + infra_type=infra_type, + target_ip=target_ip, + target_hostname=host_header, + target_port=target_port, + evidence=f"{method} {path}", + confidence=confidence, + details=json.dumps({ + "method": method, + "path": path, + "host": host_header, + "user_agent": user_agent, + "description": description, + }), + ) + return det + return None + + def analyze_dns_query(self, query_name: str, resolved_ip: str = "", + client_ip: str = "") -> Optional[dict]: + """Analyze a DNS query for supply chain infrastructure indicators. + + Returns: + Detection dict if found, None otherwise. + """ + for pattern, infra_type, description, confidence in _DNS_PATTERNS: + if pattern.search(query_name): + det = self._record_detection( + infra_type=infra_type, + target_ip=resolved_ip or "unresolved", + target_hostname=query_name, + target_port=None, + evidence=f"DNS: {query_name}", + confidence=confidence, + details=json.dumps({ + "query": query_name, + "resolved_ip": resolved_ip, + "client_ip": client_ip, + "description": description, + }), + ) + return det + return None + + def analyze_port(self, target_ip: str, port: int, + hostname: str = "") -> Optional[dict]: + """Check if a discovered open port suggests supply chain infrastructure.""" + if port not in _PORT_PATTERNS: + return None + + infra_type, description, confidence = _PORT_PATTERNS[port] + if confidence < 0.5: + # Low-confidence port-only detections need corroboration + # Check if we already have DNS or HTTP evidence for this IP + existing = self._get_detections_for_ip(target_ip) + if existing: + # Boost confidence if corroborated + confidence = min(confidence + 0.3, 0.95) + else: + return None + + return self._record_detection( + infra_type=infra_type, + target_ip=target_ip, + target_hostname=hostname, + target_port=port, + evidence=f"Open port {port}", + confidence=confidence, + details=json.dumps({"description": description, "port": port}), + ) + + def _on_host_discovered(self, event) -> None: + """Check discovered host ports for supply chain indicators.""" + p = event.payload + ip = p.get("ip", "") + hostname = p.get("hostname", "") + ports = p.get("open_ports", []) + + for port in ports: + self.analyze_port(ip, port, hostname) + + # ------------------------------------------------------------------ + # Database operations + # ------------------------------------------------------------------ + + def _record_detection(self, infra_type: str, target_ip: str, + target_hostname: str, target_port: Optional[int], + evidence: str, confidence: float, + details: str = "") -> dict: + """Record a supply chain detection, deduplicating by type+ip+port.""" + now = time.time() + detection = { + "type": infra_type, + "target_ip": target_ip, + "target_hostname": target_hostname, + "target_port": target_port, + "evidence": evidence, + "confidence": confidence, + "details": details, + } + + with self._lock: + try: + self._conn.execute( + """INSERT INTO supply_chain + (type, target_ip, target_hostname, target_port, + evidence, confidence, first_seen, last_seen, details) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(type, target_ip, target_port) DO UPDATE SET + last_seen = excluded.last_seen, + confidence = MAX(supply_chain.confidence, excluded.confidence), + evidence = supply_chain.evidence || '; ' || excluded.evidence, + target_hostname = CASE WHEN excluded.target_hostname != '' + THEN excluded.target_hostname + ELSE supply_chain.target_hostname END + """, + (infra_type, target_ip, target_hostname, target_port, + evidence, confidence, now, now, details), + ) + self._conn.commit() + self._detection_count += 1 + + logger.info( + "Supply chain detection: %s at %s:%s (%s, confidence=%.2f)", + infra_type, target_ip, target_port, evidence, confidence, + ) + except Exception: + logger.exception("Failed to record supply chain detection") + + return detection + + def _get_detections_for_ip(self, ip: str) -> list: + """Get all detections for a given IP.""" + with self._lock: + try: + rows = self._conn.execute( + "SELECT * FROM supply_chain WHERE target_ip = ?", (ip,) + ).fetchall() + return [dict(r) for r in rows] + except Exception: + return [] + + # ------------------------------------------------------------------ + # Periodic scanning + # ------------------------------------------------------------------ + + def _periodic_scan(self) -> None: + """Periodically pull DNS and HTTP data from other modules' state.""" + interval = self.config.get("supply_chain_scan_interval", 180) + while self._running: + time.sleep(interval) + try: + self._scan_dns_data() + self._scan_http_data() + except Exception: + logger.exception("Periodic supply chain scan failed") + + def _scan_dns_data(self) -> None: + """Pull recent DNS queries from dns_logger state and analyze.""" + last_ts = self.state.get(self.name, "dns_scan_last_ts") + last_ts = float(last_ts) if last_ts else 0 + + dns_json = self.state.get("dns_logger", "recent_queries") + if not dns_json: + return + + try: + queries = json.loads(dns_json) + max_ts = last_ts + for q in queries: + ts = q.get("timestamp", 0) + if ts <= last_ts: + continue + self.analyze_dns_query( + query_name=q.get("query_name", ""), + resolved_ip=q.get("resolved_ip", ""), + client_ip=q.get("client_ip", ""), + ) + max_ts = max(max_ts, ts) + + self.state.set(self.name, "dns_scan_last_ts", str(max_ts)) + except (json.JSONDecodeError, TypeError): + pass + + def _scan_http_data(self) -> None: + """Pull recent HTTP log data from state and analyze.""" + last_ts = self.state.get(self.name, "http_scan_last_ts") + last_ts = float(last_ts) if last_ts else 0 + + http_json = self.state.get("http_monitor", "recent_requests") + if not http_json: + return + + try: + requests = json.loads(http_json) + max_ts = last_ts + for req in requests: + ts = req.get("timestamp", 0) + if ts <= last_ts: + continue + self.analyze_http_request( + target_ip=req.get("target_ip", ""), + target_port=req.get("target_port", 80), + method=req.get("method", "GET"), + path=req.get("path", ""), + host_header=req.get("host", ""), + user_agent=req.get("user_agent", ""), + ) + max_ts = max(max_ts, ts) + + self.state.set(self.name, "http_scan_last_ts", str(max_ts)) + except (json.JSONDecodeError, TypeError): + pass + + # ------------------------------------------------------------------ + # Query interface + # ------------------------------------------------------------------ + + def get_detections(self, infra_type: str = None, + min_confidence: float = 0.0) -> list: + """Get supply chain detections, optionally filtered.""" + with self._lock: + if infra_type: + rows = self._conn.execute( + """SELECT * FROM supply_chain + WHERE type = ? AND confidence >= ? + ORDER BY confidence DESC""", + (infra_type, min_confidence), + ).fetchall() + else: + rows = self._conn.execute( + """SELECT * FROM supply_chain + WHERE confidence >= ? + ORDER BY confidence DESC""", + (min_confidence,), + ).fetchall() + + return [dict(r) for r in rows] + + def get_summary(self) -> dict: + """Return a summary of detected supply chain infrastructure.""" + with self._lock: + rows = self._conn.execute( + """SELECT type, COUNT(*) as count, + AVG(confidence) as avg_confidence + FROM supply_chain + GROUP BY type + ORDER BY count DESC""" + ).fetchall() + + return { + "total_detections": sum(r[1] for r in rows), + "by_type": {r[0]: {"count": r[1], "avg_confidence": round(r[2], 2)} + for r in rows}, + } diff --git a/modules/intel/tool_output_parser.py b/modules/intel/tool_output_parser.py new file mode 100644 index 0000000..d4dd1b2 --- /dev/null +++ b/modules/intel/tool_output_parser.py @@ -0,0 +1,800 @@ +#!/usr/bin/env python3 +"""Unified tool output parser — ingests output from bettercap, Responder, +and mitmproxy, extracting credentials and host data into the event bus. + +Periodically scans tool log directories for new data and publishes +CREDENTIAL_FOUND and HOST_DISCOVERED events for other intel modules +to consume. +""" + +import glob +import json +import logging +import os +import re +import threading +import time +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.tool_output_parser") + +# ----------------------------------------------------------------------- +# Responder log patterns +# ----------------------------------------------------------------------- + +# NTLMv2 hash format: user::domain:challenge:response:blob +_NTLMV2_RE = re.compile( + r"^(?P[^:]+?)::(?P[^:]*?):(?P[0-9a-fA-F]+):" + r"(?P[0-9a-fA-F]+):(?P[0-9a-fA-F]+)\s*$", + re.MULTILINE, +) + +# NTLMv1 hash format: user::domain:lm_response:nt_response:challenge +_NTLMV1_RE = re.compile( + r"^(?P[^:]+?)::(?P[^:]*?):(?P[0-9a-fA-F]+):" + r"(?P[0-9a-fA-F]+):(?P[0-9a-fA-F]+)\s*$", + re.MULTILINE, +) + +# Responder session log: credential captures with timestamps +# Format: [*] [TIMESTAMP] PROTOCOL - user:password from source_ip +_RESPONDER_SESSION_RE = re.compile( + r"^\[\*\]\s+\[(?P[^\]]+)\]\s+(?P\w+)\s*[-:]\s*" + r"(?P[^:]+):(?P[^\s]+)\s+from\s+(?P[\d.]+)", + re.MULTILINE, +) + +# HTTP Basic Auth captured by Responder +_RESPONDER_HTTP_RE = re.compile( + r"^\[\+\].*HTTP.*Username\s*:\s*(?P\S+).*Password\s*:\s*(?P\S+)", + re.MULTILINE | re.IGNORECASE, +) + +# ----------------------------------------------------------------------- +# bettercap event patterns +# ----------------------------------------------------------------------- + +# bettercap credential event types +_BETTERCAP_CRED_EVENTS = { + "net.sniff.credentials", + "http.proxy.credentials", + "https.proxy.credentials", +} + +_BETTERCAP_HOST_EVENTS = { + "endpoint.new", + "endpoint.lost", +} + +# ----------------------------------------------------------------------- +# mitmproxy flow patterns +# ----------------------------------------------------------------------- + +# Authorization header patterns +_AUTH_BASIC_RE = re.compile(r"Basic\s+([A-Za-z0-9+/=]+)", re.I) +_AUTH_BEARER_RE = re.compile(r"Bearer\s+(\S+)", re.I) + +# Cookie patterns with session identifiers +_SESSION_COOKIE_NAMES = { + "sessionid", "session_id", "phpsessid", "jsessionid", + "asp.net_sessionid", "connect.sid", "laravel_session", + "csrf_token", "_csrf", "xsrf-token", +} + +# Interesting HTTP headers +_INTERESTING_HEADERS = { + "authorization", "x-api-key", "x-auth-token", "x-csrf-token", + "cookie", "set-cookie", "www-authenticate", +} + + +class ToolOutputParser(BaseModule): + """Parse output from bettercap, Responder, and mitmproxy into structured events.""" + + name = "tool_output_parser" + module_type = "intel" + priority = 80 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._lock = threading.Lock() + self._scan_thread: Optional[threading.Thread] = None + self._bettercap_thread: Optional[threading.Thread] = None + + # Track file positions to avoid re-reading + self._file_positions: dict[str, int] = {} + + # Dedup: track recently emitted credential hashes + self._emitted_hashes: set = set() + + # Counters + self._creds_parsed = 0 + self._hosts_parsed = 0 + self._files_scanned = 0 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + # Load saved file positions + positions_json = self.state.get(self.name, "file_positions") + if positions_json: + try: + self._file_positions = json.loads(positions_json) + except (json.JSONDecodeError, TypeError): + pass + + # Periodic log directory scanner + self._scan_thread = threading.Thread( + target=self._scan_loop, daemon=True, + name="bb-tool-parser-scan", + ) + self._scan_thread.start() + + # bettercap event stream reader (if bettercap API available) + bettercap_cfg = self.config.get("bettercap", {}) + if bettercap_cfg.get("enabled", True): + self._bettercap_thread = threading.Thread( + target=self._bettercap_event_loop, daemon=True, + name="bb-tool-parser-bettercap", + ) + self._bettercap_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info("ToolOutputParser started") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + # Save file positions for resume + self.state.set(self.name, "file_positions", + json.dumps(self._file_positions)) + + self.state.set_module_status(self.name, "stopped") + logger.info( + "ToolOutputParser stopped — creds=%d, hosts=%d, files=%d", + self._creds_parsed, self._hosts_parsed, self._files_scanned, + ) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "credentials_parsed": self._creds_parsed, + "hosts_parsed": self._hosts_parsed, + "files_scanned": self._files_scanned, + "tracked_files": len(self._file_positions), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Log directory scanner + # ------------------------------------------------------------------ + + def _scan_loop(self) -> None: + """Periodically scan tool output directories for new data.""" + interval = self.config.get("scan_interval", 30) + while self._running: + try: + self._scan_responder_logs() + self._scan_mitmproxy_flows() + self._scan_bettercap_logs() + + # Persist positions periodically + self.state.set(self.name, "file_positions", + json.dumps(self._file_positions)) + except Exception: + logger.exception("Tool output scan cycle failed") + + time.sleep(interval) + + # ------------------------------------------------------------------ + # Responder parser + # ------------------------------------------------------------------ + + def _scan_responder_logs(self) -> None: + """Parse Responder log files for captured credentials.""" + log_dirs = self.config.get("responder_log_dirs", [ + "/opt/.cache/bb/responder/logs", + "/tmp/responder/logs", + os.path.expanduser("~/.bigbrother/responder"), + ]) + + for log_dir in log_dirs: + if not os.path.isdir(log_dir): + continue + + # Parse NTLMv2 hash files + for path in glob.glob(os.path.join(log_dir, "*-NTLMv2-*.txt")): + self._parse_responder_hash_file(path, "ntlmv2", 5600) + + # Parse NTLMv1 hash files + for path in glob.glob(os.path.join(log_dir, "*-NTLMv1-*.txt")): + self._parse_responder_hash_file(path, "ntlmv1", 5500) + + # Parse session log + session_log = os.path.join(log_dir, "Responder-Session.log") + if os.path.isfile(session_log): + self._parse_responder_session(session_log) + + def _parse_responder_hash_file(self, path: str, cred_type: str, + hashcat_mode: int) -> None: + """Parse a Responder NTLMv1/v2 hash file.""" + new_data = self._read_new_data(path) + if not new_data: + return + + self._files_scanned += 1 + regex = _NTLMV2_RE if cred_type == "ntlmv2" else _NTLMV1_RE + + for match in regex.finditer(new_data): + full_hash = match.group(0).strip() + username = match.group("username") + domain = match.group("domain") + + # Dedup check + hash_key = f"{cred_type}:{username}:{domain}:{full_hash[:32]}" + if hash_key in self._emitted_hashes: + continue + self._emitted_hashes.add(hash_key) + + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": "smb", + "username": username, + "domain": domain, + "cred_type": cred_type, + "cred_value": full_hash, + "hashcat_mode": hashcat_mode, + "source_ip": "", + "target_ip": "", + "notes": f"Captured by Responder from {os.path.basename(path)}", + }, source_module=self.name) + + def _parse_responder_session(self, path: str) -> None: + """Parse Responder-Session.log for cleartext credentials.""" + new_data = self._read_new_data(path) + if not new_data: + return + + self._files_scanned += 1 + + for match in _RESPONDER_SESSION_RE.finditer(new_data): + protocol = match.group("protocol").lower() + username = match.group("username") + password = match.group("password") + source_ip = match.group("source_ip") + + hash_key = f"responder_session:{protocol}:{username}:{source_ip}" + if hash_key in self._emitted_hashes: + continue + self._emitted_hashes.add(hash_key) + + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": protocol, + "username": username, + "domain": "", + "cred_type": f"{protocol}_cleartext", + "cred_value": password, + "hashcat_mode": None, + "source_ip": source_ip, + "target_ip": "", + "notes": "Cleartext capture by Responder", + }, source_module=self.name) + + # Also check for HTTP basic auth in session log + for match in _RESPONDER_HTTP_RE.finditer(new_data): + username = match.group("username") + password = match.group("password") + + hash_key = f"responder_http:{username}:{password[:8]}" + if hash_key in self._emitted_hashes: + continue + self._emitted_hashes.add(hash_key) + + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": "http", + "username": username, + "domain": "", + "cred_type": "http_basic", + "cred_value": password, + "hashcat_mode": None, + "source_ip": "", + "target_ip": "", + "notes": "HTTP Basic Auth captured by Responder", + }, source_module=self.name) + + # ------------------------------------------------------------------ + # bettercap parser + # ------------------------------------------------------------------ + + def _scan_bettercap_logs(self) -> None: + """Scan bettercap log files for credential and host events.""" + log_dirs = self.config.get("bettercap_log_dirs", [ + "/opt/.cache/bb/bettercap/logs", + os.path.expanduser("~/.bigbrother/bettercap"), + ]) + + for log_dir in log_dirs: + if not os.path.isdir(log_dir): + continue + + for path in glob.glob(os.path.join(log_dir, "events*.json")): + self._parse_bettercap_event_file(path) + + def _parse_bettercap_event_file(self, path: str) -> None: + """Parse a bettercap JSON event log file.""" + new_data = self._read_new_data(path) + if not new_data: + return + + self._files_scanned += 1 + + # bettercap event files: one JSON object per line + for line in new_data.splitlines(): + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + + self._process_bettercap_event(event) + + def _bettercap_event_loop(self) -> None: + """Connect to bettercap REST API and stream events.""" + # Import here to avoid hard dependency + try: + import urllib.request + import urllib.error + except ImportError: + return + + api_host = self.config.get("bettercap", {}).get("host", "127.0.0.1") + api_port = self.config.get("bettercap", {}).get("port", 8083) + api_user = self.config.get("bettercap", {}).get("user", "bb") + api_pass = self.config.get("bettercap", {}).get("password", "bb") + + base_url = f"http://{api_host}:{api_port}" + last_event_id = 0 + + while self._running: + try: + url = f"{base_url}/api/events?n=50" + + # Create auth handler + auth = f"{api_user}:{api_pass}" + import base64 + auth_b64 = base64.b64encode(auth.encode()).decode() + req = urllib.request.Request(url) + req.add_header("Authorization", f"Basic {auth_b64}") + + with urllib.request.urlopen(req, timeout=10) as resp: + data = json.loads(resp.read().decode()) + + if isinstance(data, list): + for event in data: + self._process_bettercap_event(event) + + except urllib.error.URLError: + pass # bettercap not running — expected + except Exception: + logger.debug("bettercap API poll failed", exc_info=True) + + time.sleep(5) + + def _process_bettercap_event(self, event: dict) -> None: + """Process a single bettercap event.""" + tag = event.get("tag", "") + data = event.get("data", {}) + + if not isinstance(data, dict): + return + + # Credential events + if tag in _BETTERCAP_CRED_EVENTS or tag == "net.sniff.credentials": + self._handle_bettercap_credential(data) + + # Host discovery events + elif tag in _BETTERCAP_HOST_EVENTS or tag == "endpoint.new": + self._handle_bettercap_host(data) + + # HTTP request events (for header analysis) + elif tag in ("net.sniff.http.request", "http.proxy.request"): + self._handle_bettercap_http(data) + + def _handle_bettercap_credential(self, data: dict) -> None: + """Extract credentials from a bettercap sniff event.""" + protocol = data.get("protocol", "unknown").lower() + username = data.get("username", data.get("user", "")) + password = data.get("password", data.get("pass", "")) + source = data.get("from", data.get("src", "")) + target = data.get("to", data.get("dst", "")) + + if not (username or password): + # Check for raw hash data + hash_val = data.get("hash", data.get("data", "")) + if hash_val: + cred_type = data.get("type", "unknown_hash") + hash_key = f"bettercap:{cred_type}:{hash_val[:32]}" + if hash_key in self._emitted_hashes: + return + self._emitted_hashes.add(hash_key) + + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": protocol, + "username": "", + "domain": "", + "cred_type": cred_type, + "cred_value": hash_val, + "source_ip": source, + "target_ip": target, + "notes": "Captured by bettercap", + }, source_module=self.name) + return + + hash_key = f"bettercap:{protocol}:{username}:{password[:16]}" + if hash_key in self._emitted_hashes: + return + self._emitted_hashes.add(hash_key) + + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": protocol, + "username": username, + "domain": "", + "cred_type": f"{protocol}_cleartext", + "cred_value": password, + "hashcat_mode": None, + "source_ip": source, + "target_ip": target, + "notes": "Captured by bettercap sniffer", + }, source_module=self.name) + + def _handle_bettercap_host(self, data: dict) -> None: + """Extract host info from a bettercap endpoint event.""" + ip = data.get("ipv4", data.get("addr", "")) + if not ip: + return + + hash_key = f"bettercap:host:{ip}" + if hash_key in self._emitted_hashes: + return + self._emitted_hashes.add(hash_key) + + self._hosts_parsed += 1 + self.bus.emit("HOST_DISCOVERED", { + "ip": ip, + "mac": data.get("mac", ""), + "hostname": data.get("hostname", data.get("name", "")), + "os_family": data.get("os", ""), + "vendor": data.get("vendor", ""), + }, source_module=self.name) + + def _handle_bettercap_http(self, data: dict) -> None: + """Extract credentials from HTTP request headers.""" + headers = data.get("headers", {}) + if not isinstance(headers, dict): + return + + host = data.get("host", "") + path = data.get("path", data.get("url", "")) + source = data.get("from", "") + + for header_name, header_value in headers.items(): + header_lower = header_name.lower() + + if header_lower == "authorization": + self._parse_auth_header(header_value, host, source) + + elif header_lower == "cookie": + self._parse_cookie_header(header_value, host, source) + + def _parse_auth_header(self, value: str, host: str, + source_ip: str) -> None: + """Parse Authorization header for credentials.""" + # Basic Auth + basic_match = _AUTH_BASIC_RE.match(value) + if basic_match: + try: + import base64 + decoded = base64.b64decode(basic_match.group(1)).decode("utf-8", errors="replace") + if ":" in decoded: + username, password = decoded.split(":", 1) + + hash_key = f"http_auth:{host}:{username}" + if hash_key not in self._emitted_hashes: + self._emitted_hashes.add(hash_key) + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": f"http://{host}", + "username": username, + "domain": "", + "cred_type": "http_basic", + "cred_value": password, + "hashcat_mode": None, + "source_ip": source_ip, + "target_ip": host, + "notes": "HTTP Basic Auth from traffic", + }, source_module=self.name) + except Exception: + pass + return + + # Bearer token + bearer_match = _AUTH_BEARER_RE.match(value) + if bearer_match: + token = bearer_match.group(1) + hash_key = f"bearer:{host}:{token[:16]}" + if hash_key not in self._emitted_hashes: + self._emitted_hashes.add(hash_key) + self._creds_parsed += 1 + + # Check if it looks like a JWT + cred_type = "jwt" if token.count(".") == 2 else "bearer_token" + + self.bus.emit("CREDENTIAL_FOUND", { + "service": f"http://{host}", + "username": "", + "domain": "", + "cred_type": cred_type, + "cred_value": token, + "hashcat_mode": None, + "source_ip": source_ip, + "target_ip": host, + "notes": f"{cred_type.upper()} from HTTP traffic", + }, source_module=self.name) + + def _parse_cookie_header(self, value: str, host: str, + source_ip: str) -> None: + """Parse Cookie header for session tokens.""" + for cookie_pair in value.split(";"): + cookie_pair = cookie_pair.strip() + if "=" not in cookie_pair: + continue + + name, cookie_val = cookie_pair.split("=", 1) + name = name.strip().lower() + + if name in _SESSION_COOKIE_NAMES: + hash_key = f"cookie:{host}:{name}:{cookie_val[:16]}" + if hash_key not in self._emitted_hashes: + self._emitted_hashes.add(hash_key) + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": f"http://{host}", + "username": "", + "domain": "", + "cred_type": "cookie", + "cred_value": f"{name}={cookie_val}", + "hashcat_mode": None, + "source_ip": source_ip, + "target_ip": host, + "notes": f"Session cookie '{name}' from HTTP traffic", + }, source_module=self.name) + + # ------------------------------------------------------------------ + # mitmproxy parser + # ------------------------------------------------------------------ + + def _scan_mitmproxy_flows(self) -> None: + """Scan mitmproxy flow dump files for credentials.""" + log_dirs = self.config.get("mitmproxy_log_dirs", [ + "/opt/.cache/bb/mitmproxy", + os.path.expanduser("~/.bigbrother/mitmproxy"), + ]) + + for log_dir in log_dirs: + if not os.path.isdir(log_dir): + continue + + # mitmproxy dumped flows (JSON format via mitmdump --set flow_detail=3) + for path in glob.glob(os.path.join(log_dir, "flows*.json")): + self._parse_mitmproxy_flow_file(path) + + # Also check for plaintext credential logs + for path in glob.glob(os.path.join(log_dir, "credentials*.log")): + self._parse_mitmproxy_cred_log(path) + + def _parse_mitmproxy_flow_file(self, path: str) -> None: + """Parse a mitmproxy JSON flow dump.""" + new_data = self._read_new_data(path) + if not new_data: + return + + self._files_scanned += 1 + + for line in new_data.splitlines(): + line = line.strip() + if not line: + continue + try: + flow = json.loads(line) + except json.JSONDecodeError: + continue + + request = flow.get("request", {}) + headers = request.get("headers", {}) + host = request.get("host", "") + path_url = request.get("path", "") + method = request.get("method", "") + + if isinstance(headers, dict): + for hdr_name, hdr_value in headers.items(): + hdr_lower = hdr_name.lower() + if hdr_lower == "authorization": + self._parse_auth_header(hdr_value, host, "") + elif hdr_lower == "cookie": + self._parse_cookie_header(hdr_value, host, "") + + # Check POST body for credentials + content = request.get("content", "") + if method == "POST" and content: + self._parse_post_body(content, host) + + def _parse_mitmproxy_cred_log(self, path: str) -> None: + """Parse a mitmproxy addon credential log (custom format).""" + new_data = self._read_new_data(path) + if not new_data: + return + + self._files_scanned += 1 + + for line in new_data.splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + + username = entry.get("username", "") + password = entry.get("password", "") + host = entry.get("host", "") + + if username and password: + hash_key = f"mitmproxy:{host}:{username}:{password[:16]}" + if hash_key in self._emitted_hashes: + continue + self._emitted_hashes.add(hash_key) + + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": f"https://{host}", + "username": username, + "domain": "", + "cred_type": "http_form", + "cred_value": password, + "hashcat_mode": None, + "source_ip": "", + "target_ip": host, + "notes": "Form credentials captured by mitmproxy", + }, source_module=self.name) + + def _parse_post_body(self, content: str, host: str) -> None: + """Look for credentials in HTTP POST bodies.""" + # URL-encoded form data + cred_fields = { + "password", "passwd", "pass", "pwd", "secret", + "token", "api_key", "apikey", + } + user_fields = { + "username", "user", "login", "email", "account", + } + + # Try URL-encoded + try: + from urllib.parse import parse_qs + params = parse_qs(content, keep_blank_values=True) + + username = "" + password = "" + for field in user_fields: + if field in params: + username = params[field][0] + break + for field in cred_fields: + if field in params: + password = params[field][0] + break + + if username and password: + hash_key = f"post:{host}:{username}:{password[:16]}" + if hash_key not in self._emitted_hashes: + self._emitted_hashes.add(hash_key) + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": f"http://{host}", + "username": username, + "domain": "", + "cred_type": "http_form", + "cred_value": password, + "hashcat_mode": None, + "source_ip": "", + "target_ip": host, + "notes": "Form POST credentials from HTTP traffic", + }, source_module=self.name) + except Exception: + pass + + # Try JSON body + try: + data = json.loads(content) + if isinstance(data, dict): + username = "" + password = "" + for field in user_fields: + if field in data: + username = str(data[field]) + break + for field in cred_fields: + if field in data: + password = str(data[field]) + break + + if username and password: + hash_key = f"json_post:{host}:{username}:{password[:16]}" + if hash_key not in self._emitted_hashes: + self._emitted_hashes.add(hash_key) + self._creds_parsed += 1 + self.bus.emit("CREDENTIAL_FOUND", { + "service": f"http://{host}", + "username": username, + "domain": "", + "cred_type": "http_json", + "cred_value": password, + "hashcat_mode": None, + "source_ip": "", + "target_ip": host, + "notes": "JSON POST credentials from HTTP traffic", + }, source_module=self.name) + except (json.JSONDecodeError, TypeError): + pass + + # ------------------------------------------------------------------ + # File reading utilities + # ------------------------------------------------------------------ + + def _read_new_data(self, path: str) -> str: + """Read only new data from a file since the last read. + + Returns the new data, or empty string if nothing new. + """ + try: + size = os.path.getsize(path) + except OSError: + return "" + + last_pos = self._file_positions.get(path, 0) + if size <= last_pos: + return "" + + try: + with open(path, "r", errors="replace") as f: + f.seek(last_pos) + data = f.read() + self._file_positions[path] = size + return data + except (IOError, PermissionError): + return "" diff --git a/modules/intel/topology_mapper.py b/modules/intel/topology_mapper.py new file mode 100644 index 0000000..de4f74c --- /dev/null +++ b/modules/intel/topology_mapper.py @@ -0,0 +1,528 @@ +#!/usr/bin/env python3 +"""Network topology mapper — aggregates host and VLAN data into graph output. + +Subscribes to HOST_DISCOVERED and VLAN_DETECTED events. Queries the state +database for host_discovery, network_mapper, and vlan_discovery data. +Generates Graphviz DOT output with nodes color-coded by OS family and +shaped by role (workstation, server, printer, router, switch, AP). +""" + +import json +import logging +import os +import subprocess +import threading +import time +from collections import defaultdict +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.topology_mapper") + +# OS family -> Graphviz fill color +_OS_COLORS = { + "windows": "#4488cc", + "linux": "#44aa44", + "macos": "#888888", + "ios": "#aaaaaa", + "android": "#88bb44", + "freebsd": "#cc4444", + "printer": "#ddaa44", + "network": "#cc8844", + "iot": "#aa66cc", + "unknown": "#cccccc", +} + +# Device role -> Graphviz node shape +_ROLE_SHAPES = { + "workstation": "box", + "server": "box3d", + "printer": "tab", + "router": "diamond", + "switch": "hexagon", + "ap": "trapezium", + "phone": "ellipse", + "iot": "component", + "unknown": "ellipse", +} + +# Protocol -> edge color +_PROTO_COLORS = { + "tcp": "#333333", + "udp": "#666666", + "smb": "#4488cc", + "http": "#44aa44", + "https": "#228822", + "dns": "#cc8844", + "ssh": "#cc4444", + "rdp": "#4444cc", + "ldap": "#aa66cc", + "kerberos": "#ddaa44", +} + + +class TopologyMapper(BaseModule): + """Build and export a network topology graph from discovered hosts and links.""" + + name = "topology_mapper" + module_type = "intel" + priority = 150 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._lock = threading.Lock() + # In-memory graph: nodes keyed by IP, edges as (src, dst) -> {protocols, bytes} + self._nodes: dict[str, dict] = {} + self._edges: dict[tuple, dict] = defaultdict(lambda: { + "protocols": set(), "bytes_total": 0, "conn_count": 0, + }) + self._vlans: dict[int, dict] = {} + self._update_thread: Optional[threading.Thread] = None + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self.bus.subscribe(self._on_host_discovered, "HOST_DISCOVERED") + self.bus.subscribe(self._on_vlan_detected, "VLAN_DETECTED") + + # Load any existing topology data from state + self._load_from_state() + + # Periodic refresh from state database + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._update_thread = threading.Thread( + target=self._periodic_update, daemon=True, + name="bb-topology-update", + ) + self._update_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info( + "TopologyMapper started — %d nodes, %d edges loaded from state", + len(self._nodes), len(self._edges), + ) + + def stop(self) -> None: + if not self._running: + return + self._running = False + + self.bus.unsubscribe(self._on_host_discovered, "HOST_DISCOVERED") + self.bus.unsubscribe(self._on_vlan_detected, "VLAN_DETECTED") + + # Persist current topology + self._save_to_state() + + self.state.set_module_status(self.name, "stopped") + logger.info( + "TopologyMapper stopped — %d nodes, %d edges", + len(self._nodes), len(self._edges), + ) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "nodes": len(self._nodes), + "edges": len(self._edges), + "vlans": len(self._vlans), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + def _on_host_discovered(self, event) -> None: + """Ingest a newly discovered host.""" + p = event.payload + ip = p.get("ip", "") + if not ip: + return + + with self._lock: + if ip in self._nodes: + # Merge new data into existing node + node = self._nodes[ip] + node["last_seen"] = time.time() + if p.get("hostname"): + node["hostname"] = p["hostname"] + if p.get("os_family"): + node["os_family"] = p["os_family"].lower() + if p.get("role"): + node["role"] = p["role"].lower() + if p.get("mac"): + node["mac"] = p["mac"] + if p.get("open_ports"): + node.setdefault("open_ports", set()).update(p["open_ports"]) + if p.get("services"): + node.setdefault("services", []).extend(p["services"]) + else: + open_ports = p.get("open_ports", []) + self._nodes[ip] = { + "ip": ip, + "hostname": p.get("hostname", ""), + "mac": p.get("mac", ""), + "os_family": p.get("os_family", "unknown").lower(), + "role": p.get("role", "unknown").lower(), + "vlan": p.get("vlan"), + "open_ports": set(open_ports) if isinstance(open_ports, list) else open_ports, + "services": p.get("services", []), + "first_seen": time.time(), + "last_seen": time.time(), + } + + def _on_vlan_detected(self, event) -> None: + """Record a detected VLAN.""" + p = event.payload + vlan_id = p.get("vlan_id") + if vlan_id is None: + return + + with self._lock: + self._vlans[vlan_id] = { + "vlan_id": vlan_id, + "name": p.get("name", ""), + "subnet": p.get("subnet", ""), + "gateway": p.get("gateway", ""), + "first_seen": time.time(), + } + + # ------------------------------------------------------------------ + # Graph data aggregation + # ------------------------------------------------------------------ + + def add_edge(self, src_ip: str, dst_ip: str, protocol: str = "tcp", + bytes_transferred: int = 0) -> None: + """Add or update an edge in the topology graph.""" + key = (src_ip, dst_ip) if src_ip < dst_ip else (dst_ip, src_ip) + with self._lock: + edge = self._edges[key] + edge["protocols"].add(protocol.lower()) + edge["bytes_total"] += bytes_transferred + edge["conn_count"] += 1 + + def _periodic_update(self) -> None: + """Periodically refresh topology from state database.""" + interval = self.config.get("topology_refresh_interval", 300) + while self._running: + time.sleep(interval) + try: + self._refresh_from_state() + self._save_to_state() + except Exception: + logger.exception("Periodic topology update failed") + + def _refresh_from_state(self) -> None: + """Pull host and network data from state key-value store.""" + # Pull host list from host_discovery module state + hosts_json = self.state.get("host_discovery", "discovered_hosts") + if hosts_json: + try: + hosts = json.loads(hosts_json) + for host in hosts: + ip = host.get("ip", "") + if ip and ip not in self._nodes: + with self._lock: + self._nodes[ip] = { + "ip": ip, + "hostname": host.get("hostname", ""), + "mac": host.get("mac", ""), + "os_family": host.get("os_family", "unknown").lower(), + "role": self._infer_role(host), + "vlan": host.get("vlan"), + "open_ports": set(host.get("open_ports", [])), + "services": host.get("services", []), + "first_seen": host.get("first_seen", time.time()), + "last_seen": host.get("last_seen", time.time()), + } + except (json.JSONDecodeError, TypeError): + pass + + # Pull VLAN data from vlan_discovery module state + vlans_json = self.state.get("vlan_discovery", "vlans") + if vlans_json: + try: + vlans = json.loads(vlans_json) + for vlan in vlans: + vid = vlan.get("vlan_id") + if vid is not None: + with self._lock: + self._vlans[vid] = vlan + except (json.JSONDecodeError, TypeError): + pass + + def _infer_role(self, host: dict) -> str: + """Infer device role from open ports and services.""" + ports = set(host.get("open_ports", [])) + hostname = host.get("hostname", "").lower() + + # Router/gateway indicators + if ports & {179, 520, 521}: # BGP, RIP + return "router" + if any(kw in hostname for kw in ("gw", "gateway", "router", "rtr")): + return "router" + + # Switch indicators + if any(kw in hostname for kw in ("switch", "sw-", "sw0")): + return "switch" + + # AP indicators + if any(kw in hostname for kw in ("ap-", "wap", "access-point")): + return "ap" + + # Printer indicators + if ports & {515, 631, 9100}: # LPD, IPP, RAW + return "printer" + if any(kw in hostname for kw in ("print", "prn", "hp", "canon", "brother")): + return "printer" + + # Server indicators (multiple service ports) + server_ports = {22, 25, 53, 80, 443, 389, 636, 88, 445, 3389, 5432, 3306, 1433, 8080, 8443} + if len(ports & server_ports) >= 3: + return "server" + if any(kw in hostname for kw in ("srv", "server", "dc", "dns", "mail", "web", "db")): + return "server" + + return "workstation" + + # ------------------------------------------------------------------ + # State persistence + # ------------------------------------------------------------------ + + def _load_from_state(self) -> None: + """Load saved topology from state on startup.""" + nodes_json = self.state.get(self.name, "nodes") + if nodes_json: + try: + nodes = json.loads(nodes_json) + for ip, node in nodes.items(): + node["open_ports"] = set(node.get("open_ports", [])) + self._nodes[ip] = node + except (json.JSONDecodeError, TypeError): + pass + + edges_json = self.state.get(self.name, "edges") + if edges_json: + try: + edges = json.loads(edges_json) + for key_str, edge_data in edges.items(): + parts = key_str.split("|") + if len(parts) == 2: + key = (parts[0], parts[1]) + edge_data["protocols"] = set(edge_data.get("protocols", [])) + self._edges[key] = edge_data + except (json.JSONDecodeError, TypeError): + pass + + vlans_json = self.state.get(self.name, "vlans") + if vlans_json: + try: + self._vlans = json.loads(vlans_json) + except (json.JSONDecodeError, TypeError): + pass + + def _save_to_state(self) -> None: + """Persist topology to state database.""" + with self._lock: + # Serialize nodes (convert sets to lists) + nodes_ser = {} + for ip, node in self._nodes.items(): + n = dict(node) + n["open_ports"] = list(n.get("open_ports", set())) + nodes_ser[ip] = n + + # Serialize edges (convert tuple keys and sets) + edges_ser = {} + for (src, dst), edge in self._edges.items(): + key = f"{src}|{dst}" + edges_ser[key] = { + "protocols": list(edge.get("protocols", set())), + "bytes_total": edge.get("bytes_total", 0), + "conn_count": edge.get("conn_count", 0), + } + + self.state.set(self.name, "nodes", json.dumps(nodes_ser)) + self.state.set(self.name, "edges", json.dumps(edges_ser)) + self.state.set(self.name, "vlans", json.dumps(self._vlans)) + + # ------------------------------------------------------------------ + # DOT / SVG generation + # ------------------------------------------------------------------ + + def generate_dot(self) -> str: + """Generate Graphviz DOT representation of the network topology.""" + lines = [ + "digraph BigBrotherTopology {", + ' rankdir=LR;', + ' bgcolor="#1a1a2e";', + ' node [style=filled, fontcolor=white, fontname="Courier"];', + ' edge [fontname="Courier", fontsize=9];', + "", + ] + + # Group nodes by VLAN + vlan_groups = defaultdict(list) + no_vlan = [] + + with self._lock: + for ip, node in self._nodes.items(): + vlan = node.get("vlan") + if vlan is not None: + vlan_groups[vlan].append((ip, node)) + else: + no_vlan.append((ip, node)) + + # VLAN subgraphs + for vlan_id, members in sorted(vlan_groups.items()): + vlan_info = self._vlans.get(vlan_id, {}) + vlan_label = vlan_info.get("name", f"VLAN {vlan_id}") + subnet = vlan_info.get("subnet", "") + label = f"{vlan_label}" + if subnet: + label += f"\\n{subnet}" + + lines.append(f" subgraph cluster_vlan{vlan_id} {{") + lines.append(f' label="{label}";') + lines.append(' style=dashed;') + lines.append(' color="#666666";') + lines.append(' fontcolor="#aaaaaa";') + + for ip, node in members: + lines.append(f" {self._node_dot(ip, node)}") + lines.append(" }") + lines.append("") + + # Ungrouped nodes + for ip, node in no_vlan: + lines.append(f" {self._node_dot(ip, node)}") + + lines.append("") + + # Edges + for (src, dst), edge in self._edges.items(): + protos = ", ".join(sorted(edge.get("protocols", set()))) + count = edge.get("conn_count", 0) + # Thicker line for higher traffic + penwidth = min(1.0 + (count / 100.0), 5.0) + # Color by primary protocol + primary_proto = next(iter(edge.get("protocols", {"tcp"})), "tcp") + color = _PROTO_COLORS.get(primary_proto, "#666666") + + label = protos + if count > 10: + label += f"\\n({count})" + + src_id = self._sanitize_id(src) + dst_id = self._sanitize_id(dst) + lines.append( + f' {src_id} -> {dst_id} ' + f'[label="{label}", color="{color}", ' + f'penwidth={penwidth:.1f}];' + ) + + lines.append("}") + return "\n".join(lines) + + def _node_dot(self, ip: str, node: dict) -> str: + """Generate DOT for a single node.""" + os_family = node.get("os_family", "unknown") + role = node.get("role", "unknown") + hostname = node.get("hostname", "") + color = _OS_COLORS.get(os_family, _OS_COLORS["unknown"]) + shape = _ROLE_SHAPES.get(role, _ROLE_SHAPES["unknown"]) + + label = ip + if hostname: + label = f"{hostname}\\n{ip}" + + ports = node.get("open_ports", set()) + if ports: + port_str = ", ".join(str(p) for p in sorted(ports)[:8]) + if len(ports) > 8: + port_str += f" (+{len(ports) - 8})" + label += f"\\n[{port_str}]" + + node_id = self._sanitize_id(ip) + return ( + f'{node_id} [label="{label}", shape={shape}, ' + f'fillcolor="{color}", tooltip="{os_family}/{role}"];' + ) + + @staticmethod + def _sanitize_id(ip: str) -> str: + """Convert IP to valid DOT node ID.""" + return "n_" + ip.replace(".", "_").replace(":", "_") + + def generate_svg(self, output_path: str = None) -> Optional[str]: + """Generate SVG from the DOT graph (requires graphviz installed). + + Args: + output_path: Path to write SVG file. If None, returns SVG as string. + + Returns: + SVG string if output_path is None, else the output path. + """ + dot = self.generate_dot() + + try: + result = subprocess.run( + ["dot", "-Tsvg"], + input=dot.encode(), + capture_output=True, + timeout=30, + ) + if result.returncode != 0: + logger.error("Graphviz dot failed: %s", result.stderr.decode()) + return None + + svg = result.stdout.decode() + + if output_path: + with open(output_path, "w") as f: + f.write(svg) + return output_path + return svg + + except FileNotFoundError: + logger.warning("Graphviz not installed — SVG generation unavailable") + return None + except subprocess.TimeoutExpired: + logger.error("Graphviz timed out generating SVG") + return None + + def get_topology_summary(self) -> dict: + """Return a summary of the current network topology.""" + with self._lock: + roles = defaultdict(int) + os_families = defaultdict(int) + for node in self._nodes.values(): + roles[node.get("role", "unknown")] += 1 + os_families[node.get("os_family", "unknown")] += 1 + + protocols_seen = set() + for edge in self._edges.values(): + protocols_seen.update(edge.get("protocols", set())) + + return { + "total_nodes": len(self._nodes), + "total_edges": len(self._edges), + "total_vlans": len(self._vlans), + "by_role": dict(roles), + "by_os": dict(os_families), + "protocols": sorted(protocols_seen), + "vlan_ids": sorted(self._vlans.keys()), + } diff --git a/modules/intel/user_timeline.py b/modules/intel/user_timeline.py new file mode 100644 index 0000000..9577183 --- /dev/null +++ b/modules/intel/user_timeline.py @@ -0,0 +1,584 @@ +#!/usr/bin/env python3 +"""Per-user activity timeline — aggregates login events, service access, +file operations, and web activity into chronological user profiles. + +Ingests data from dns_logger, auth_flow_tracker, smb_monitor, and +credential_sniffer via the event bus and periodic state queries. +Identifies work hours, admin activity windows, and service account behavior. +""" + +import json +import logging +import os +import sqlite3 +import threading +import time +from collections import defaultdict +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.intel.user_timeline") + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS user_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + username TEXT NOT NULL, + domain TEXT DEFAULT '', + event_type TEXT NOT NULL, + source_ip TEXT, + target_ip TEXT, + service TEXT, + detail TEXT DEFAULT '', + source_module TEXT DEFAULT '' +); + +CREATE INDEX IF NOT EXISTS idx_ue_user ON user_events(username); +CREATE INDEX IF NOT EXISTS idx_ue_ts ON user_events(timestamp); +CREATE INDEX IF NOT EXISTS idx_ue_type ON user_events(event_type); + +CREATE TABLE IF NOT EXISTS user_profiles ( + username TEXT PRIMARY KEY, + domain TEXT DEFAULT '', + first_seen REAL, + last_seen REAL, + event_count INTEGER DEFAULT 0, + is_admin INTEGER DEFAULT 0, + is_service_acct INTEGER DEFAULT 0, + typical_hours TEXT DEFAULT '', + workstations TEXT DEFAULT '', + services_used TEXT DEFAULT '', + notes TEXT DEFAULT '' +); +""" + +# Event types +EVENT_LOGIN = "login" +EVENT_LOGOUT = "logout" +EVENT_AUTH_FAIL = "auth_fail" +EVENT_SERVICE_ACCESS = "service_access" +EVENT_FILE_ACCESS = "file_access" +EVENT_WEB_BROWSE = "web_browse" +EVENT_ADMIN_ACTION = "admin_action" +EVENT_CRED_CAPTURED = "cred_captured" + +# Admin indicators: services or patterns that suggest admin activity +_ADMIN_SERVICES = { + "rdp", "ssh", "winrm", "wmi", "psremoting", "dcom", + "ldap", "kerberos", "smb_admin", "rpc", +} + +_ADMIN_USERNAMES = { + "administrator", "admin", "root", "domain admin", +} + +# Service account patterns +_SVC_PATTERNS = ( + "svc_", "svc-", "service_", "sa_", "task_", "app_", + "sql_", "iis_", "backup_", "scan_", "nessus", "splunk", + "crowdstrike", "sccm", "wsus", +) + + +class UserTimeline(BaseModule): + """Build per-user activity timelines from network observation data.""" + + name = "user_timeline" + module_type = "intel" + priority = 200 + requires_root = False + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._db_path = "" + self._conn: Optional[sqlite3.Connection] = None + self._lock = threading.Lock() + self._ingest_thread: Optional[threading.Thread] = None + self._event_count = 0 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "user_timeline.db") + os.makedirs(os.path.dirname(self._db_path), exist_ok=True) + + self._conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=NORMAL") + self._conn.row_factory = sqlite3.Row + self._conn.executescript(_SCHEMA) + + # Subscribe to credential and host events for user correlation + self.bus.subscribe(self._on_credential_found, "CREDENTIAL_FOUND") + self.bus.subscribe(self._on_host_discovered, "HOST_DISCOVERED") + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + # Periodic ingestion from other module state + self._ingest_thread = threading.Thread( + target=self._periodic_ingest, daemon=True, + name="bb-user-timeline-ingest", + ) + self._ingest_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._pid) + logger.info("UserTimeline started — db=%s", self._db_path) + + def stop(self) -> None: + if not self._running: + return + self._running = False + + self.bus.unsubscribe(self._on_credential_found, "CREDENTIAL_FOUND") + self.bus.unsubscribe(self._on_host_discovered, "HOST_DISCOVERED") + + if self._conn: + self._conn.close() + self._conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("UserTimeline stopped — %d events recorded", self._event_count) + + def status(self) -> dict: + user_count = 0 + if self._conn: + try: + with self._lock: + row = self._conn.execute( + "SELECT COUNT(DISTINCT username) FROM user_events" + ).fetchone() + user_count = row[0] if row else 0 + except Exception: + pass + + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_events": self._event_count, + "tracked_users": user_count, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Event handlers + # ------------------------------------------------------------------ + + def _on_credential_found(self, event) -> None: + """Record credential capture as a user event.""" + p = event.payload + username = p.get("username", "") + if not username: + return + + self._record_event( + username=username, + domain=p.get("domain", ""), + event_type=EVENT_CRED_CAPTURED, + source_ip=p.get("source_ip", ""), + target_ip=p.get("target_ip", ""), + service=p.get("service", ""), + detail=f"Type: {p.get('cred_type', 'unknown')}", + source_module=event.source_module, + ) + + def _on_host_discovered(self, event) -> None: + """Check for user information in host discovery data.""" + p = event.payload + # Some host discovery may include logged-in user information + username = p.get("logged_in_user", "") + if username: + self._record_event( + username=username, + domain=p.get("domain", ""), + event_type=EVENT_LOGIN, + source_ip=p.get("ip", ""), + target_ip="", + service="workstation", + detail=f"Active on {p.get('hostname', p.get('ip', ''))}", + source_module=event.source_module, + ) + + # ------------------------------------------------------------------ + # Event recording + # ------------------------------------------------------------------ + + def _record_event(self, username: str, domain: str, event_type: str, + source_ip: str = "", target_ip: str = "", + service: str = "", detail: str = "", + source_module: str = "", + timestamp: float = None) -> None: + """Record a user activity event.""" + if not username: + return + + ts = timestamp or time.time() + self._event_count += 1 + + with self._lock: + try: + self._conn.execute( + """INSERT INTO user_events + (timestamp, username, domain, event_type, source_ip, + target_ip, service, detail, source_module) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (ts, username, domain, event_type, source_ip, + target_ip, service, detail, source_module), + ) + + # Update user profile + is_admin = 1 if ( + service.lower() in _ADMIN_SERVICES + or username.lower() in _ADMIN_USERNAMES + or event_type == EVENT_ADMIN_ACTION + ) else 0 + + is_svc = 1 if any( + username.lower().startswith(p) for p in _SVC_PATTERNS + ) else 0 + + self._conn.execute( + """INSERT INTO user_profiles + (username, domain, first_seen, last_seen, event_count, + is_admin, is_service_acct) + VALUES (?, ?, ?, ?, 1, ?, ?) + ON CONFLICT(username) DO UPDATE SET + last_seen = MAX(excluded.last_seen, user_profiles.last_seen), + event_count = user_profiles.event_count + 1, + is_admin = MAX(excluded.is_admin, user_profiles.is_admin), + is_service_acct = MAX(excluded.is_service_acct, user_profiles.is_service_acct) + """, + (username, domain, ts, ts, is_admin, is_svc), + ) + + self._conn.commit() + except Exception: + logger.exception("Failed to record user event for %s", username) + + def ingest_auth_event(self, username: str, domain: str, source_ip: str, + target_ip: str, service: str, success: bool, + timestamp: float = None) -> None: + """Ingest an authentication event from external modules.""" + event_type = EVENT_LOGIN if success else EVENT_AUTH_FAIL + detail = "success" if success else "failure" + self._record_event( + username=username, domain=domain, event_type=event_type, + source_ip=source_ip, target_ip=target_ip, service=service, + detail=detail, source_module="auth_flow_tracker", + timestamp=timestamp, + ) + + def ingest_file_access(self, username: str, source_ip: str, + file_path: str, action: str = "read", + timestamp: float = None) -> None: + """Ingest a file access event (from smb_monitor).""" + self._record_event( + username=username, domain="", event_type=EVENT_FILE_ACCESS, + source_ip=source_ip, target_ip="", service="smb", + detail=f"{action}: {file_path}", + source_module="smb_monitor", timestamp=timestamp, + ) + + def ingest_web_activity(self, source_ip: str, domain: str, + url: str = "", timestamp: float = None) -> None: + """Ingest web browsing activity (correlated to user by IP).""" + # Look up username by source_ip from recent events + username = self._resolve_user_by_ip(source_ip) + if not username: + username = f"host:{source_ip}" + + self._record_event( + username=username, domain="", event_type=EVENT_WEB_BROWSE, + source_ip=source_ip, target_ip="", service="web", + detail=url or domain, + source_module="dns_logger", timestamp=timestamp, + ) + + def _resolve_user_by_ip(self, ip: str) -> str: + """Try to resolve a username from a source IP using recent login events.""" + with self._lock: + try: + row = self._conn.execute( + """SELECT username FROM user_events + WHERE source_ip = ? AND event_type = ? + ORDER BY timestamp DESC LIMIT 1""", + (ip, EVENT_LOGIN), + ).fetchone() + return row["username"] if row else "" + except Exception: + return "" + + # ------------------------------------------------------------------ + # Periodic state ingestion + # ------------------------------------------------------------------ + + def _periodic_ingest(self) -> None: + """Periodically pull data from other module state stores.""" + interval = self.config.get("timeline_ingest_interval", 120) + while self._running: + time.sleep(interval) + try: + self._ingest_from_auth_tracker() + self._ingest_from_smb_monitor() + self._ingest_from_dns_logger() + self._update_user_profiles() + except Exception: + logger.exception("Periodic user timeline ingest failed") + + def _ingest_from_auth_tracker(self) -> None: + """Pull authentication events from auth_flow_tracker state.""" + last_ts = self.state.get(self.name, "auth_tracker_last_ts") + last_ts = float(last_ts) if last_ts else 0 + + auth_json = self.state.get("auth_flow_tracker", "recent_events") + if not auth_json: + return + + try: + events = json.loads(auth_json) + for evt in events: + ts = evt.get("timestamp", 0) + if ts <= last_ts: + continue + self.ingest_auth_event( + username=evt.get("username", ""), + domain=evt.get("domain", ""), + source_ip=evt.get("source_ip", ""), + target_ip=evt.get("target_ip", ""), + service=evt.get("service", ""), + success=evt.get("success", True), + timestamp=ts, + ) + last_ts = max(last_ts, ts) + + self.state.set(self.name, "auth_tracker_last_ts", str(last_ts)) + except (json.JSONDecodeError, TypeError): + pass + + def _ingest_from_smb_monitor(self) -> None: + """Pull SMB file access events from smb_monitor state.""" + last_ts = self.state.get(self.name, "smb_monitor_last_ts") + last_ts = float(last_ts) if last_ts else 0 + + smb_json = self.state.get("smb_monitor", "recent_file_ops") + if not smb_json: + return + + try: + events = json.loads(smb_json) + for evt in events: + ts = evt.get("timestamp", 0) + if ts <= last_ts: + continue + self.ingest_file_access( + username=evt.get("username", ""), + source_ip=evt.get("source_ip", ""), + file_path=evt.get("path", ""), + action=evt.get("action", "read"), + timestamp=ts, + ) + last_ts = max(last_ts, ts) + + self.state.set(self.name, "smb_monitor_last_ts", str(last_ts)) + except (json.JSONDecodeError, TypeError): + pass + + def _ingest_from_dns_logger(self) -> None: + """Pull DNS query data for web activity correlation.""" + last_ts = self.state.get(self.name, "dns_logger_last_ts") + last_ts = float(last_ts) if last_ts else 0 + + dns_json = self.state.get("dns_logger", "recent_queries") + if not dns_json: + return + + try: + queries = json.loads(dns_json) + for q in queries: + ts = q.get("timestamp", 0) + if ts <= last_ts: + continue + self.ingest_web_activity( + source_ip=q.get("client_ip", ""), + domain=q.get("query_name", ""), + timestamp=ts, + ) + last_ts = max(last_ts, ts) + + self.state.set(self.name, "dns_logger_last_ts", str(last_ts)) + except (json.JSONDecodeError, TypeError): + pass + + def _update_user_profiles(self) -> None: + """Refresh computed fields on user profiles (work hours, workstations, etc).""" + with self._lock: + try: + users = self._conn.execute( + "SELECT DISTINCT username FROM user_events" + ).fetchall() + + for (username,) in users: + # Compute typical hours + rows = self._conn.execute( + """SELECT timestamp FROM user_events + WHERE username = ?""", + (username,), + ).fetchall() + + hours = defaultdict(int) + for (ts,) in rows: + hour = time.localtime(ts).tm_hour + hours[hour] += 1 + + # Top active hours (>10% of activity) + total = sum(hours.values()) + if total > 0: + typical = sorted( + [h for h, c in hours.items() if c / total > 0.1] + ) + else: + typical = [] + + # Workstations (source IPs used for login) + ws_rows = self._conn.execute( + """SELECT DISTINCT source_ip FROM user_events + WHERE username = ? AND event_type = ? AND source_ip != ''""", + (username, EVENT_LOGIN), + ).fetchall() + workstations = [r[0] for r in ws_rows] + + # Services used + svc_rows = self._conn.execute( + """SELECT DISTINCT service FROM user_events + WHERE username = ? AND service != ''""", + (username,), + ).fetchall() + services = [r[0] for r in svc_rows] + + self._conn.execute( + """UPDATE user_profiles + SET typical_hours = ?, workstations = ?, services_used = ? + WHERE username = ?""", + (json.dumps(typical), json.dumps(workstations), + json.dumps(services), username), + ) + + self._conn.commit() + except Exception: + logger.exception("Failed to update user profiles") + + # ------------------------------------------------------------------ + # Query interface + # ------------------------------------------------------------------ + + def get_timeline(self, username: str, limit: int = 200, + since: float = 0) -> list: + """Get chronological activity timeline for a user.""" + with self._lock: + rows = self._conn.execute( + """SELECT * FROM user_events + WHERE username = ? AND timestamp > ? + ORDER BY timestamp DESC LIMIT ?""", + (username, since, limit), + ).fetchall() + + return [dict(r) for r in rows] + + def get_active_users(self, hours: float = 1.0) -> list: + """Get users active within the given time window.""" + since = time.time() - (hours * 3600) + with self._lock: + rows = self._conn.execute( + """SELECT username, domain, COUNT(*) as event_count, + MAX(timestamp) as last_active + FROM user_events + WHERE timestamp > ? + GROUP BY username + ORDER BY last_active DESC""", + (since,), + ).fetchall() + + return [dict(r) for r in rows] + + def get_admin_users(self) -> list: + """Get all users identified as having admin privileges.""" + with self._lock: + rows = self._conn.execute( + """SELECT * FROM user_profiles + WHERE is_admin = 1 + ORDER BY last_seen DESC""" + ).fetchall() + + return [dict(r) for r in rows] + + def get_service_accounts(self) -> list: + """Get all detected service accounts.""" + with self._lock: + rows = self._conn.execute( + """SELECT * FROM user_profiles + WHERE is_service_acct = 1 + ORDER BY event_count DESC""" + ).fetchall() + + return [dict(r) for r in rows] + + def get_user_profile(self, username: str) -> Optional[dict]: + """Get computed profile for a user.""" + with self._lock: + row = self._conn.execute( + "SELECT * FROM user_profiles WHERE username = ?", + (username,), + ).fetchone() + + if not row: + return None + + profile = dict(row) + # Parse JSON fields + for field in ("typical_hours", "workstations", "services_used"): + try: + profile[field] = json.loads(profile.get(field, "[]")) + except (json.JSONDecodeError, TypeError): + profile[field] = [] + + return profile + + def get_user_work_hours(self, username: str) -> dict: + """Analyze and return a user's typical work hours.""" + with self._lock: + rows = self._conn.execute( + """SELECT timestamp FROM user_events + WHERE username = ?""", + (username,), + ).fetchall() + + if not rows: + return {"username": username, "hours": {}, "typical_start": None, "typical_end": None} + + hours = defaultdict(int) + for (ts,) in rows: + hour = time.localtime(ts).tm_hour + hours[hour] += 1 + + total = sum(hours.values()) + active_hours = sorted([h for h, c in hours.items() if c / total > 0.05]) + + return { + "username": username, + "hours": dict(hours), + "active_hours": active_hours, + "typical_start": active_hours[0] if active_hours else None, + "typical_end": active_hours[-1] if active_hours else None, + "total_events": total, + }