From 9fd97e6925652451bff1fe138dbc1c161323890c Mon Sep 17 00:00:00 2001 From: bot Date: Wed, 8 Apr 2026 08:34:52 +0300 Subject: [PATCH] Fix recurring WAL corruption with RESTART checkpointing Root cause: SQLite's default PASSIVE checkpoint mode cannot complete under sustained write load (4 nodes, hundreds of events/min) because active readers always block it. The WAL grows unbounded and eventually corrupts the main DB file. - Add dedicated checkpoint_wal() using PRAGMA wal_checkpoint(RESTART), which waits for readers to finish then checkpoints fully and resets the WAL back to the start. - Add background checkpointer task running every 30 minutes, separate from the 6-hour prune cycle. - Remove PASSIVE checkpoint from prune_old_events(), replace with call to checkpoint_wal() after pruning completes. - Set wal_autocheckpoint=500 on every connection (~2MB threshold) so SQLite's own auto-checkpoint also triggers more frequently. - Reduce HOT_DAYS from 30 to 7: at current data rate (~500k beacons/day) 30-day retention would grow to several GB before first prune runs. 7 days keeps the DB under ~200MB and queries fast permanently. Co-Authored-By: Claude Sonnet 4.6 --- coordinator/dashboard.py | 115 ++++++++++++++++++++++++++++++++------- 1 file changed, 94 insertions(+), 21 deletions(-) diff --git a/coordinator/dashboard.py b/coordinator/dashboard.py index 2b10a5d..e6614b8 100644 --- a/coordinator/dashboard.py +++ b/coordinator/dashboard.py @@ -19,7 +19,7 @@ JS_PATH = os.path.join(os.path.dirname(__file__), "static", "dashboard.js") OUI_PATH = os.path.join(os.path.dirname(__file__), "oui.txt") ONLINE_SECS = 30 # node considered online if heartbeat within this many seconds EVENT_CAP = 100 # max events returned / shown in dashboard -HOT_DAYS = 30 # dashboard queries cover this many days; Search is always all-time +HOT_DAYS = 7 # dashboard queries cover this many days; Search is always all-time # ─── Deauth reason codes ───────────────────────────────────────────────────── @@ -105,6 +105,7 @@ def get_conn() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA wal_autocheckpoint=500") # auto-checkpoint every 500 pages (~2MB) return conn @@ -481,24 +482,27 @@ def _finalize_session(active: dict) -> dict: else: severity = "normal" + desc, _ = deauth_reason_info(dominant) src_vendor, src_rand = oui_lookup(active["src"]) return { - "src": active["src"], - "src_vendor": src_vendor, - "src_randomized": src_rand, - "start": active["start"], - "end": active["end"], - "duration_secs": duration, - "frame_count": frames, - "unique_bssids": list(active["bssids"]), - "unique_targets": list(active["targets"]), - "node_count": len(active["nodes"]), - "nodes": list(active["nodes"]), - "peak_rssi": max(rssi_vals) if rssi_vals else None, - "avg_rssi": round(sum(rssi_vals) / len(rssi_vals), 1) if rssi_vals else None, - "dominant_reason": dominant, - "severity": severity, - "events": active["frames"][:50], + "src": active["src"], + "src_vendor": src_vendor, + "src_randomized": src_rand, + "start": active["start"], + "end": active["end"], + "duration_secs": duration, + "frame_count": frames, + "unique_bssids": list(active["bssids"]), + "unique_targets": list(active["targets"]), + "node_count": len(active["nodes"]), + "nodes": list(active["nodes"]), + "peak_rssi": max(rssi_vals) if rssi_vals else None, + "avg_rssi": round(sum(rssi_vals) / len(rssi_vals), 1) if rssi_vals else None, + "dominant_reason": dominant, + "dominant_reason_desc": desc, + "dominant_reason_class": classification, + "severity": severity, + "events": active["frames"][:50], } @@ -654,8 +658,30 @@ def build_alerts() -> dict: GROUP BY bssid """, tuple(bssid_list)) ssid_map = {r["bssid"]: r["ssid"] for r in ssid_rows} + # Dominant reason per impersonated BSSID + reason_rows_t = query(f""" + SELECT bssid, reason, COUNT(*) AS cnt + FROM deauth_events + WHERE received_at > ? AND bssid IN ({placeholders}) + GROUP BY bssid, reason + ORDER BY bssid, cnt DESC + """, (cutoff_7d,) + tuple(bssid_list)) + dom_reason_by_bssid: dict = {} + for r in reason_rows_t: + if r["bssid"] not in dom_reason_by_bssid: + dom_reason_by_bssid[r["bssid"]] = r["reason"] for t in top_targets: t["ssid"] = ssid_map.get(t["bssid"]) + code = dom_reason_by_bssid.get(t["bssid"]) + if code is not None: + desc, cls = deauth_reason_info(code) + t["dominant_reason"] = code + t["dominant_reason_desc"] = desc + t["dominant_reason_class"] = cls + else: + t["dominant_reason"] = None + t["dominant_reason_desc"] = None + t["dominant_reason_class"] = None # Most targeted devices — last 7 days (grouped by dst, the actual victim) top_targeted_devices = query(""" @@ -672,10 +698,38 @@ def build_alerts() -> dict: ORDER BY total_frames DESC LIMIT 10 """, (cutoff_7d,)) + if top_targeted_devices: + dst_list = [d["dst"] for d in top_targeted_devices] + placeholders = ",".join("?" * len(dst_list)) + # Dominant reason per targeted dst MAC + reason_rows_d = query(f""" + SELECT dst, reason, COUNT(*) AS cnt + FROM deauth_events + WHERE received_at > ? AND dst IN ({placeholders}) + GROUP BY dst, reason + ORDER BY dst, cnt DESC + """, (cutoff_7d,) + tuple(dst_list)) + dom_reason_by_dst: dict = {} + for r in reason_rows_d: + if r["dst"] not in dom_reason_by_dst: + dom_reason_by_dst[r["dst"]] = r["reason"] + else: + dom_reason_by_dst = {} + for d in top_targeted_devices: vendor, randomized = oui_lookup(d["dst"]) d["vendor"] = vendor d["randomized"] = randomized + code = dom_reason_by_dst.get(d["dst"]) + if code is not None: + desc, cls = deauth_reason_info(code) + d["dominant_reason"] = code + d["dominant_reason_desc"] = desc + d["dominant_reason_class"] = cls + else: + d["dominant_reason"] = None + d["dominant_reason_desc"] = None + d["dominant_reason_class"] = None # Reason code breakdown — last 7 days reason_rows = query(""" @@ -988,18 +1042,30 @@ def ensure_schema(): conn.commit() -PRUNE_INTERVAL_SECS = 6 * 3600 # run pruning every 6 hours +PRUNE_INTERVAL_SECS = 6 * 3600 # run pruning every 6 hours +CHECKPOINT_INTERVAL_SECS = 30 * 60 # force WAL checkpoint every 30 minutes def prune_old_events() -> None: - """Delete events older than HOT_DAYS from all event tables.""" + """Delete events older than HOT_DAYS from all event tables, then checkpoint.""" cutoff = _hot_cutoff() tables = ["beacon_events", "probe_events", "deauth_events", "assoc_events"] with get_conn() as conn: for table in tables: conn.execute(f"DELETE FROM {table} WHERE received_at < ?", (cutoff,)) - conn.execute("PRAGMA wal_checkpoint(PASSIVE)") conn.commit() + checkpoint_wal() + + +def checkpoint_wal() -> None: + """ + Force a RESTART checkpoint — writes all WAL frames to the main DB file + and resets the WAL back to the beginning. Prevents the WAL from growing + unbounded under sustained write load, which causes DB corruption over time. + RESTART blocks until all current readers finish, then checkpoints fully. + """ + with get_conn() as conn: + conn.execute("PRAGMA wal_checkpoint(RESTART)") app = FastAPI(title="ESP32 Recon Dashboard") @@ -1007,12 +1073,19 @@ ensure_schema() @app.on_event("startup") -async def start_pruning_task(): +async def start_background_tasks(): async def _pruner(): while True: await asyncio.sleep(PRUNE_INTERVAL_SECS) await asyncio.to_thread(prune_old_events) + + async def _checkpointer(): + while True: + await asyncio.sleep(CHECKPOINT_INTERVAL_SECS) + await asyncio.to_thread(checkpoint_wal) + asyncio.create_task(_pruner()) + asyncio.create_task(_checkpointer()) @app.get("/", response_class=HTMLResponse)