#!/usr/bin/env python3 """Central credential store — aggregates credentials from all capture sources. Subscribes to CREDENTIAL_FOUND, TICKET_HARVESTED, CLOUD_TOKEN_FOUND events. Deduplicates on (service, username, cred_type, cred_value). Exports in hashcat, CSV, and JSON formats. Tracks crack status per credential. """ import csv import io import json import logging import os import sqlite3 import threading import time from typing import Optional from modules.base import BaseModule from utils.credential_encryption import emit_credential_found logger = logging.getLogger(__name__) _TABLE_DDL = """ CREATE TABLE IF NOT EXISTS credentials ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL NOT NULL, source_module TEXT NOT NULL DEFAULT '', source_ip TEXT, target_ip TEXT, target_port INTEGER, service TEXT NOT NULL, username TEXT NOT NULL DEFAULT '', domain TEXT DEFAULT '', cred_type TEXT NOT NULL, cred_value TEXT NOT NULL DEFAULT '', hashcat_mode INTEGER, cracked_value TEXT, crack_status TEXT DEFAULT 'uncracked', notes TEXT DEFAULT '', UNIQUE(service, username, cred_type, cred_value) ) """ _INDEXES = [ "CREATE INDEX IF NOT EXISTS idx_creds_service ON credentials(service)", "CREATE INDEX IF NOT EXISTS idx_creds_user ON credentials(username)", "CREATE INDEX IF NOT EXISTS idx_creds_type ON credentials(cred_type)", "CREATE INDEX IF NOT EXISTS idx_creds_status ON credentials(crack_status)", ] # Legacy alias so any external code importing _SCHEMA still works _SCHEMA = _TABLE_DDL # Hashcat mode mapping for common credential types _HASHCAT_MODES = { "ntlmv2": 5600, "ntlmv1": 5500, "ntlm": 1000, "net-ntlmv2": 5600, "net-ntlmv1": 5500, "kerberos_tgs": 13100, # Kerberoast — TGS-REP (RC4) "kerberos_as": 18200, # AS-REP roast "kerberos_tgs_aes128": 19600, "kerberos_tgs_aes256": 19700, "http_basic": None, # plaintext "http_digest": 11400, "snmp_community": None, # plaintext "mssql": 131, "mysql": 300, "postgres": None, "ftp": None, # plaintext "telnet": None, # plaintext "smtp": None, # plaintext "ldap": None, # plaintext "vnc": None, # plaintext "rdp_ntlm": 5600, "wpa_pmkid": 22000, "wpa_handshake": 22000, "cloud_token": None, # not hashable "ssh_key": None, "cookie": None, "jwt": None, "bearer_token": None, } class CredentialDB(BaseModule): """Central credential database — ingests, deduplicates, and exports credentials.""" name = "credential_db" module_type = "intel" priority = 50 requires_root = False def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._db_path = "" self._conn: Optional[sqlite3.Connection] = None self._lock = threading.Lock() self._ingest_count = 0 self._dedup_count = 0 # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: if self._running: return base_dir = self.config.get("data_dir", os.path.expanduser("~/.implant")) self._db_path = os.path.join(base_dir, "credentials.db") os.makedirs(os.path.dirname(self._db_path), exist_ok=True) self._conn = sqlite3.connect(self._db_path, check_same_thread=False) self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA synchronous=NORMAL") self._conn.execute(_TABLE_DDL) self._conn.commit() self._migrate_schema() for idx_sql in _INDEXES: self._conn.execute(idx_sql) self._conn.commit() # Subscribe to credential events self.bus.subscribe(self._on_credential_found, "CREDENTIAL_FOUND") self.bus.subscribe(self._on_ticket_harvested, "TICKET_HARVESTED") self.bus.subscribe(self._on_cloud_token, "CLOUD_TOKEN_FOUND") self._running = True self._pid = os.getpid() self._start_time = time.time() self.state.set_module_status(self.name, "running", pid=self._pid) logger.info("CredentialDB started — db=%s", self._db_path) def _migrate_schema(self) -> None: """Add columns that exist in _SCHEMA but are missing from an older table.""" required = { "source_module": "TEXT NOT NULL DEFAULT ''", "cracked_value": "TEXT", "crack_status": "TEXT DEFAULT 'uncracked'", "notes": "TEXT DEFAULT ''", } cur = self._conn.execute("PRAGMA table_info(credentials)") existing = {row[1] for row in cur.fetchall()} for col, coldef in required.items(): if col not in existing: self._conn.execute(f"ALTER TABLE credentials ADD COLUMN {col} {coldef}") logger.info("Schema migrated: added column %s", col) self._conn.commit() def stop(self) -> None: if not self._running: return self._running = False self.bus.unsubscribe(self._on_credential_found, "CREDENTIAL_FOUND") self.bus.unsubscribe(self._on_ticket_harvested, "TICKET_HARVESTED") self.bus.unsubscribe(self._on_cloud_token, "CLOUD_TOKEN_FOUND") if self._conn: self._conn.close() self._conn = None self.state.set_module_status(self.name, "stopped") logger.info( "CredentialDB stopped — ingested=%d, deduped=%d", self._ingest_count, self._dedup_count, ) def status(self) -> dict: total = 0 cracked = 0 if self._conn: try: with self._lock: row = self._conn.execute( "SELECT COUNT(*) FROM credentials" ).fetchone() total = row[0] if row else 0 row = self._conn.execute( "SELECT COUNT(*) FROM credentials WHERE crack_status='cracked'" ).fetchone() cracked = row[0] if row else 0 except Exception: pass return { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, "total_credentials": total, "cracked": cracked, "ingested": self._ingest_count, "deduped": self._dedup_count, } def configure(self, config: dict) -> None: self.config.update(config) # ------------------------------------------------------------------ # Event handlers # ------------------------------------------------------------------ def _on_credential_found(self, event) -> None: """Handle CREDENTIAL_FOUND events from any capture module.""" p = event.payload from utils.credential_encryption import decrypt_credential_payload p = decrypt_credential_payload(p) self._ingest_credential( source_module=event.source_module or p.get("source_module", "unknown"), source_ip=p.get("source_ip", ""), target_ip=p.get("target_ip", ""), target_port=p.get("target_port"), service=p.get("service", "unknown"), username=p.get("username", ""), domain=p.get("domain", ""), cred_type=p.get("cred_type", "unknown"), cred_value=p.get("cred_value", ""), hashcat_mode=p.get("hashcat_mode"), notes=p.get("notes", ""), ) def _on_ticket_harvested(self, event) -> None: """Handle TICKET_HARVESTED events (Kerberos TGS/AS-REP).""" p = event.payload cred_type = p.get("ticket_type", "kerberos_tgs") self._ingest_credential( source_module=event.source_module or "kerberos_harvester", source_ip=p.get("source_ip", ""), target_ip=p.get("target_ip", ""), target_port=p.get("target_port", 88), service=p.get("spn", "krbtgt"), username=p.get("username", ""), domain=p.get("domain", ""), cred_type=cred_type, cred_value=p.get("ticket_hash", ""), hashcat_mode=_HASHCAT_MODES.get(cred_type), notes=p.get("notes", f"SPN: {p.get('spn', '')}"), ) def _on_cloud_token(self, event) -> None: """Handle CLOUD_TOKEN_FOUND events (AWS/Azure/GCP tokens).""" p = event.payload self._ingest_credential( source_module=event.source_module or "cloud_token_harvester", source_ip=p.get("source_ip", ""), target_ip=p.get("target_ip", ""), target_port=p.get("target_port"), service=p.get("cloud_provider", "cloud"), username=p.get("identity", ""), domain=p.get("account_id", ""), cred_type="cloud_token", cred_value=p.get("token", ""), hashcat_mode=None, notes=p.get("notes", f"Provider: {p.get('cloud_provider', '')}"), ) # ------------------------------------------------------------------ # Core credential storage # ------------------------------------------------------------------ def _ingest_credential(self, source_module: str, source_ip: str, target_ip: str, target_port: Optional[int], service: str, username: str, domain: str, cred_type: str, cred_value: str, hashcat_mode: Optional[int] = None, notes: str = "") -> bool: """Insert a credential, deduplicating on (service, username, cred_type, cred_value). Returns True if a new credential was inserted, False if duplicate. """ if not cred_value: return False # Auto-resolve hashcat mode if not provided if hashcat_mode is None: hashcat_mode = _HASHCAT_MODES.get(cred_type.lower()) self._ingest_count += 1 with self._lock: try: self._conn.execute( """INSERT INTO credentials (timestamp, source_module, source_ip, target_ip, target_port, service, username, domain, cred_type, cred_value, hashcat_mode, crack_status, notes) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'uncracked', ?)""", (time.time(), source_module, source_ip, target_ip, target_port, service, username, domain, cred_type, cred_value, hashcat_mode, notes), ) self._conn.commit() logger.info( "New credential: %s\\%s@%s (%s via %s)", domain, username, service, cred_type, source_module, ) return True except sqlite3.IntegrityError: # Duplicate — update notes/source if new info self._dedup_count += 1 self._conn.execute( """UPDATE credentials SET notes = CASE WHEN notes = '' THEN ? ELSE notes || '; ' || ? END, source_module = CASE WHEN source_module NOT LIKE '%' || ? || '%' THEN source_module || ',' || ? ELSE source_module END WHERE service=? AND username=? AND cred_type=? AND cred_value=?""", (notes, notes, source_module, source_module, service, username, cred_type, cred_value), ) self._conn.commit() return False except Exception: logger.exception("Failed to ingest credential") return False # ------------------------------------------------------------------ # Crack status management # ------------------------------------------------------------------ def update_crack_status(self, cred_id: int, status: str, cracked_value: str = "") -> None: """Update crack status for a credential. Args: cred_id: Credential ID. status: One of 'uncracked', 'cracking', 'cracked'. cracked_value: The plaintext password if cracked. """ if status not in ("uncracked", "cracking", "cracked"): return with self._lock: try: self._conn.execute( """UPDATE credentials SET crack_status=?, cracked_value=? WHERE id=?""", (status, cracked_value, cred_id), ) self._conn.commit() except Exception: logger.exception("Failed to update crack status for ID %d", cred_id) def bulk_update_cracked(self, results: dict) -> int: """Bulk update cracked passwords from hashcat output. Args: results: {hash_value: plaintext_password, ...} Returns: Number of credentials updated. """ updated = 0 with self._lock: try: for hash_val, plaintext in results.items(): cur = self._conn.execute( """UPDATE credentials SET crack_status='cracked', cracked_value=? WHERE cred_value=? AND crack_status != 'cracked'""", (plaintext, hash_val), ) updated += cur.rowcount self._conn.commit() except Exception: logger.exception("Failed to bulk update cracked credentials") return updated # ------------------------------------------------------------------ # Export methods # ------------------------------------------------------------------ def to_hashcat(self, mode: Optional[int] = None) -> str: """Export credentials in hashcat format, optionally filtered by mode. Args: mode: Hashcat mode number. If None, export all hashable credentials. Returns: Newline-separated hash strings ready for hashcat. """ with self._lock: if mode is not None: rows = self._conn.execute( """SELECT username, domain, cred_value, cred_type FROM credentials WHERE hashcat_mode=? AND crack_status='uncracked'""", (mode,), ).fetchall() else: rows = self._conn.execute( """SELECT username, domain, cred_value, cred_type FROM credentials WHERE hashcat_mode IS NOT NULL AND crack_status='uncracked'""", ).fetchall() lines = [] for username, domain, cred_value, cred_type in rows: # NTLMv2 and similar Net-NTLM hashes are already in hashcat format if cred_type.lower() in ("ntlmv2", "net-ntlmv2", "ntlmv1", "net-ntlmv1"): lines.append(cred_value) elif cred_type.lower() in ("kerberos_tgs", "kerberos_as", "kerberos_tgs_aes128", "kerberos_tgs_aes256"): lines.append(cred_value) else: # Generic: just the hash value lines.append(cred_value) return "\n".join(lines) def to_csv(self) -> str: """Export all credentials as CSV.""" with self._lock: rows = self._conn.execute( """SELECT id, timestamp, source_module, source_ip, target_ip, target_port, service, username, domain, cred_type, cred_value, hashcat_mode, cracked_value, crack_status, notes FROM credentials ORDER BY timestamp""" ).fetchall() output = io.StringIO() writer = csv.writer(output) writer.writerow([ "id", "timestamp", "source_module", "source_ip", "target_ip", "target_port", "service", "username", "domain", "cred_type", "cred_value", "hashcat_mode", "cracked_value", "crack_status", "notes", ]) for row in rows: writer.writerow(row) return output.getvalue() def to_json(self) -> str: """Export all credentials as JSON.""" with self._lock: self._conn.row_factory = sqlite3.Row rows = self._conn.execute( """SELECT * FROM credentials ORDER BY timestamp""" ).fetchall() self._conn.row_factory = None return json.dumps([dict(r) for r in rows], indent=2, default=str) def summary(self) -> dict: """Return a summary of stored credentials.""" with self._lock: total = self._conn.execute("SELECT COUNT(*) FROM credentials").fetchone()[0] by_type = self._conn.execute( "SELECT cred_type, COUNT(*) FROM credentials GROUP BY cred_type" ).fetchall() by_service = self._conn.execute( "SELECT service, COUNT(*) FROM credentials GROUP BY service" ).fetchall() by_status = self._conn.execute( "SELECT crack_status, COUNT(*) FROM credentials GROUP BY crack_status" ).fetchall() unique_users = self._conn.execute( "SELECT COUNT(DISTINCT username) FROM credentials" ).fetchone()[0] unique_hosts = self._conn.execute( "SELECT COUNT(DISTINCT target_ip) FROM credentials WHERE target_ip != ''" ).fetchone()[0] return { "total": total, "unique_users": unique_users, "unique_hosts": unique_hosts, "by_type": {t: c for t, c in by_type}, "by_service": {s: c for s, c in by_service}, "by_status": {s: c for s, c in by_status}, } # ------------------------------------------------------------------ # Query helpers # ------------------------------------------------------------------ def get_credentials(self, service: str = None, username: str = None, cred_type: str = None, limit: int = 100) -> list: """Query credentials with optional filters.""" clauses = [] params = [] if service: clauses.append("service = ?") params.append(service) if username: clauses.append("username = ?") params.append(username) if cred_type: clauses.append("cred_type = ?") params.append(cred_type) where = " AND ".join(clauses) if clauses else "1=1" params.append(limit) with self._lock: self._conn.row_factory = sqlite3.Row rows = self._conn.execute( f"SELECT * FROM credentials WHERE {where} ORDER BY timestamp DESC LIMIT ?", params, ).fetchall() self._conn.row_factory = None return [dict(r) for r in rows] def get_cracked(self) -> list: """Return all cracked credentials with plaintext values.""" with self._lock: self._conn.row_factory = sqlite3.Row rows = self._conn.execute( """SELECT username, domain, service, cracked_value, target_ip FROM credentials WHERE crack_status='cracked' AND cracked_value IS NOT NULL ORDER BY timestamp DESC""" ).fetchall() self._conn.row_factory = None return [dict(r) for r in rows]