Fix credential_db schema conflict and add resilience

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

@@ -170,9 +170,21 @@ class CaptureBus:
             except socket.timeout:
                 continue
             except OSError as e:
-                if self._running:
+                if not self._running:
-                    logger.error("Socket read error: %s", e)
                     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

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
 
     # ------------------------------------------------------------------

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