From a8b81f38c97d5ada3536474422c6fd770fd1fddc Mon Sep 17 00:00:00 2001 From: Cobra Date: Tue, 7 Apr 2026 12:19:56 -0400 Subject: [PATCH] Fix credential_db schema conflict and add resilience MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit credential_sniffer and credential_db both wrote to the same credentials.db with different schemas; credential_db's executescript failed when creating the crack_status index on an existing table that lacked that column. Fix: split credential_db's schema init into table DDL + migration + indexes so missing columns are added before index creation runs. Migration also handles any future schema drift. state.py: add 30s timeout + busy_timeout PRAGMA to write connection so concurrent subprocess writes stop failing with 'database is locked'. capture_bus.py: reconnect on ENETDOWN instead of aborting the capture loop — mac_manager temporarily brings the interface down during MAC rotation which was silently killing all packet capture. --- core/capture_bus.py | 18 ++++++++++--- core/state.py | 3 ++- modules/intel/credential_db.py | 48 ++++++++++++++++++++++++++-------- 3 files changed, 54 insertions(+), 15 deletions(-) diff --git a/core/capture_bus.py b/core/capture_bus.py index adf8da3..722b3a5 100644 --- a/core/capture_bus.py +++ b/core/capture_bus.py @@ -170,9 +170,21 @@ class CaptureBus: except socket.timeout: continue except OSError as e: - if self._running: - logger.error("Socket read error: %s", e) - break + if not self._running: + break + logger.warning("Socket read error: %s — reconnecting in 3s", e) + try: + self._sock.close() + except Exception: + pass + time.sleep(3) + try: + self._sock = self._open_socket() + logger.info("CaptureBus reconnected on %s", self.interface) + except OSError as e2: + logger.error("CaptureBus reconnect failed: %s", e2) + time.sleep(10) + continue ts = time.time() self._total_packets += 1 diff --git a/core/state.py b/core/state.py index bff5b56..c050faf 100644 --- a/core/state.py +++ b/core/state.py @@ -139,9 +139,10 @@ class StateManager: def _get_write_conn(self) -> sqlite3.Connection: if self._write_conn is None: - self._write_conn = sqlite3.connect(self._db_path, check_same_thread=False) + self._write_conn = sqlite3.connect(self._db_path, check_same_thread=False, timeout=30) self._write_conn.execute("PRAGMA journal_mode=WAL") self._write_conn.execute("PRAGMA synchronous=NORMAL") + self._write_conn.execute("PRAGMA busy_timeout=30000") return self._write_conn # ------------------------------------------------------------------ diff --git a/modules/intel/credential_db.py b/modules/intel/credential_db.py index ab354e6..8954cd8 100644 --- a/modules/intel/credential_db.py +++ b/modules/intel/credential_db.py @@ -21,32 +21,37 @@ from utils.credential_encryption import emit_credential_found logger = logging.getLogger("sensor.intel.credential_db") -_SCHEMA = """ +_TABLE_DDL = """ CREATE TABLE IF NOT EXISTS credentials ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL NOT NULL, - source_module TEXT 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, + username TEXT NOT NULL DEFAULT '', domain TEXT DEFAULT '', cred_type TEXT NOT NULL, - cred_value 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) -); - -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); +) """ +_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, @@ -111,7 +116,12 @@ class CredentialDB(BaseModule): 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.executescript(_SCHEMA) + 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") @@ -124,6 +134,22 @@ class CredentialDB(BaseModule): 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