bot/esp32-cluster
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix dashboard latency and add DB pruning

Browse Source 
- Wrap all build_* route calls in asyncio.to_thread() so SQLite queries
  run in a thread pool instead of blocking the event loop. Fixes tab
  switch freezes and SSE stream stalls.
- Add 60s TTL caches to alerts, sessions, presence, and cross-node
  endpoints (previously uncached, queried on every request).
- Scope alerts queries (top targets, devices, reason breakdown, heatmap)
  to last 7 days instead of all-time full table scans.
- Add composite indexes: (bssid, received_at) on beacon_events and
  deauth_events, (src_mac, received_at) on probe_events.
- Add background pruning task: runs every 6 hours, deletes events older
  than HOT_DAYS from all tables, followed by WAL checkpoint.
- Document the full diagnosis and fixes in README.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
bot
2026-04-06 13:01:13 +03:00
parent fee23f0f66
commit 36c7248d0e
2 changed files with 241 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coordinator/dashboard.py
+167 -65
View File
@@ -19,6 +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
# ─── Deauth reason codes ─────────────────────────────────────────────────────
@@ -113,6 +114,10 @@ def query(sql: str, params: tuple = ()) -> list[dict]:
return [dict(r) for r in rows]
def _hot_cutoff() -> str:
"""ISO timestamp HOT_DAYS ago — used to scope dashboard queries."""
return (datetime.datetime.utcnow() - datetime.timedelta(days=HOT_DAYS)).isoformat(timespec="seconds")
def _node_status(last_seen: str | None, threshold: int = ONLINE_SECS) -> str:
if not last_seen:
return "offline"
@@ -125,7 +130,26 @@ def _node_status(last_seen: str | None, threshold: int = ONLINE_SECS) -> str:
# ─── API data builders ────────────────────────────────────────────────────────
_nodes_cache: tuple[float, list] | None = None
_networks_cache: tuple[float, list] | None = None
_clients_cache: tuple[float, list] | None = None
_cross_node_cache: tuple[float, dict] | None = None
_presence_cache: tuple[float, dict] | None = None
_alerts_cache: tuple[float, dict] | None = None
_sessions_cache: tuple[float, list] | None = None
_NODES_CACHE_TTL = 10 # seconds
_NETWORKS_CACHE_TTL = 30 # seconds
_CLIENTS_CACHE_TTL = 30 # seconds
_CROSS_NODE_CACHE_TTL = 60 # seconds
_PRESENCE_CACHE_TTL = 60 # seconds
_ALERTS_CACHE_TTL = 60 # seconds
_SESSIONS_CACHE_TTL = 60 # seconds
def build_nodes() -> list[dict]:
global _nodes_cache
now = datetime.datetime.utcnow().timestamp()
if _nodes_cache and (now - _nodes_cache[0]) < _NODES_CACHE_TTL:
return _nodes_cache[1]
rows = query("""
SELECT
node_id,
@@ -156,6 +180,7 @@ def build_nodes() -> list[dict]:
r["status"] = _node_status(r["last_heartbeat"], threshold=30)
else:
r["status"] = _node_status(r["last_beacon"], threshold=60)
_nodes_cache = (now, rows)
return rows
@@ -173,6 +198,10 @@ def build_events(limit: int = EVENT_CAP, node_id: str | None = None) -> list[dic
def build_networks() -> list[dict]:
global _networks_cache
now = datetime.datetime.utcnow().timestamp()
if _networks_cache and (now - _networks_cache[0]) < _NETWORKS_CACHE_TTL:
return _networks_cache[1]
rows = query("""
SELECT
ssid,
@@ -185,18 +214,24 @@ def build_networks() -> list[dict]:
COUNT(DISTINCT node_id) AS node_count,
MAX(received_at) AS last_seen
FROM beacon_events
WHERE bssid IS NOT NULL
WHERE bssid IS NOT NULL AND received_at > ?
GROUP BY bssid
ORDER BY times_seen DESC
LIMIT 300
""")
""", (_hot_cutoff(),))
_networks_cache = (now, rows)
return rows
def build_clients() -> list[dict]:
global _clients_cache
now = datetime.datetime.utcnow().timestamp()
if _clients_cache and (now - _clients_cache[0]) < _CLIENTS_CACHE_TTL:
return _clients_cache[1]
"""
Per unique src_mac: aggregated stats and the SSIDs they were probing for.
"""
cutoff = _hot_cutoff()
rows = query("""
SELECT
src_mac,
@@ -208,11 +243,11 @@ def build_clients() -> list[dict]:
MIN(received_at) AS first_seen,
MAX(received_at) AS last_seen
FROM probe_events
WHERE src_mac IS NOT NULL
WHERE src_mac IS NOT NULL AND received_at > ?
GROUP BY src_mac
ORDER BY times_seen DESC
LIMIT 300
""")
""", (cutoff,))
# Fetch SSIDs only for the MACs we actually returned
if not rows:
return []
@@ -221,8 +256,9 @@ def build_clients() -> list[dict]:
ssid_rows = query(f"""
SELECT DISTINCT src_mac, ssid FROM probe_events
WHERE src_mac IN ({placeholders}) AND ssid IS NOT NULL AND ssid != ''
AND received_at > ?
ORDER BY src_mac, ssid
""", tuple(mac_list))
""", tuple(mac_list) + (cutoff,))
ssids_by_mac: dict = {}
for s in ssid_rows:
ssids_by_mac.setdefault(s["src_mac"], []).append(s["ssid"])
@@ -232,6 +268,7 @@ def build_clients() -> list[dict]:
vendor, randomized = oui_lookup(r["src_mac"])
r["vendor"] = vendor
r["randomized"] = randomized
_clients_cache = (now, rows)
return rows
@@ -241,9 +278,15 @@ def build_cross_node() -> dict:
Result: { node_ids: [...], networks: [{bssid, ssid, channel, encryption,
node_count, nodes: {node_id: {best_rssi, avg_rssi, times_seen}}}] }
"""
global _cross_node_cache
now = datetime.datetime.utcnow().timestamp()
if _cross_node_cache and (now - _cross_node_cache[0]) < _CROSS_NODE_CACHE_TTL:
return _cross_node_cache[1]
node_rows = query("SELECT DISTINCT node_id FROM beacon_events ORDER BY node_id")
node_ids = [r["node_id"] for r in node_rows]
cutoff = _hot_cutoff()
rows = query("""
SELECT
b.bssid,
@@ -255,16 +298,16 @@ def build_cross_node() -> dict:
ROUND(AVG(b.rssi), 1) AS avg_rssi,
COUNT(*) AS times_seen
FROM beacon_events b
WHERE b.bssid IN (
WHERE b.received_at > ? AND b.bssid IN (
SELECT bssid FROM beacon_events
WHERE bssid IS NOT NULL
WHERE bssid IS NOT NULL AND received_at > ?
GROUP BY bssid
ORDER BY COUNT(DISTINCT node_id) DESC, COUNT(*) DESC
LIMIT 300
)
GROUP BY b.bssid, b.node_id
ORDER BY b.bssid, b.node_id
""")
""", (cutoff, cutoff))
networks: dict = {}
for r in rows:
@@ -291,10 +334,17 @@ def build_cross_node() -> dict:
for net in result:
net["node_count"] = len(net["nodes"])
return {"node_ids": node_ids, "networks": result}
result_dict = {"node_ids": node_ids, "networks": result}
_cross_node_cache = (now, result_dict)
return result_dict
def build_presence() -> dict:
global _presence_cache
_now_ts = datetime.datetime.utcnow().timestamp()
if _presence_cache and (_now_ts - _presence_cache[0]) < _PRESENCE_CACHE_TTL:
return _presence_cache[1]
now = datetime.datetime.utcnow()
cutoff_1h = (now - datetime.timedelta(hours=1)).isoformat(timespec="seconds")
cutoff_24h = (now - datetime.timedelta(hours=24)).isoformat(timespec="seconds")
@@ -341,6 +391,7 @@ def build_presence() -> dict:
r["randomized"] = randomized
new_macs.append(r)
cutoff_hot = _hot_cutoff()
new_networks = query("""
SELECT bssid,
MAX(ssid) AS ssid,
@@ -350,12 +401,12 @@ def build_presence() -> dict:
MAX(encryption) AS encryption,
COUNT(DISTINCT node_id) AS node_count
FROM beacon_events
WHERE bssid IS NOT NULL
WHERE bssid IS NOT NULL AND received_at > ?
GROUP BY bssid
HAVING MIN(received_at) > ?
ORDER BY first_seen DESC
LIMIT 100
""", (cutoff_24h,))
""", (cutoff_hot, cutoff_24h,))
# ── Regulars: seen on 2+ distinct calendar days ───────────────────────────
reg_mac_rows = query("""
@@ -398,13 +449,15 @@ def build_presence() -> dict:
LIMIT 50
""")
return {
result = {
"present": present,
"new_macs": new_macs,
"new_networks": new_networks,
"reg_macs": reg_macs,
"reg_networks": reg_networks,
}
_presence_cache = (_now_ts, result)
return result
SESSION_GAP_SECS = 120 # gap > 2 min between events from same src = new session
@@ -450,7 +503,12 @@ def _finalize_session(active: dict) -> dict:
def build_sessions() -> list[dict]:
cutoff = (datetime.datetime.utcnow() - datetime.timedelta(days=7)).isoformat(timespec="seconds")
global _sessions_cache
now = datetime.datetime.utcnow().timestamp()
if _sessions_cache and (now - _sessions_cache[0]) < _SESSIONS_CACHE_TTL:
return _sessions_cache[1]
cutoff = _hot_cutoff()
rows = query("""
SELECT src, dst, bssid, reason, rssi, received_at, node_id, subtype
@@ -505,13 +563,21 @@ def build_sessions() -> list[dict]:
sessions.append(_finalize_session(active))
sessions.sort(key=lambda s: s["start"], reverse=True)
return sessions[:100]
result = sessions[:100]
_sessions_cache = (now, result)
return result
def build_alerts() -> dict:
global _alerts_cache
_now_ts = datetime.datetime.utcnow().timestamp()
if _alerts_cache and (_now_ts - _alerts_cache[0]) < _ALERTS_CACHE_TTL:
return _alerts_cache[1]
now = datetime.datetime.utcnow()
cutoff_5m = (now - datetime.timedelta(minutes=5)).isoformat(timespec="seconds")
cutoff_1h = (now - datetime.timedelta(hours=1)).isoformat(timespec="seconds")
cutoff_7d = (now - datetime.timedelta(days=7)).isoformat(timespec="seconds")
# Recent raw deauth/disassoc events
recent = query("""
@@ -564,7 +630,7 @@ def build_alerts() -> dict:
WHERE received_at > ?
""", (cutoff_1h,))
# Most targeted BSSIDs — all time
# Most targeted BSSIDs — last 7 days
top_targets = query("""
SELECT
bssid,
@@ -574,10 +640,11 @@ def build_alerts() -> dict:
MIN(received_at) AS first_seen,
MAX(received_at) AS last_seen
FROM deauth_events
WHERE received_at > ?
GROUP BY bssid
ORDER BY total_frames DESC
LIMIT 10
""")
""", (cutoff_7d,))
if top_targets:
bssid_list = [t["bssid"] for t in top_targets]
placeholders = ",".join("?" * len(bssid_list))
@@ -590,7 +657,7 @@ def build_alerts() -> dict:
for t in top_targets:
t["ssid"] = ssid_map.get(t["bssid"])
# Most targeted devices — all time (grouped by dst, the actual victim)
# Most targeted devices — last 7 days (grouped by dst, the actual victim)
top_targeted_devices = query("""
SELECT
dst,
@@ -600,16 +667,17 @@ def build_alerts() -> dict:
MIN(received_at) AS first_seen,
MAX(received_at) AS last_seen
FROM deauth_events
WHERE received_at > ?
GROUP BY dst
ORDER BY total_frames DESC
LIMIT 10
""")
""", (cutoff_7d,))
for d in top_targeted_devices:
vendor, randomized = oui_lookup(d["dst"])
d["vendor"] = vendor
d["randomized"] = randomized
# Reason code breakdown — all time
# Reason code breakdown — last 7 days
reason_rows = query("""
SELECT reason,
COUNT(*) AS total_frames,
@@ -618,25 +686,27 @@ def build_alerts() -> dict:
MIN(received_at) AS first_seen,
MAX(received_at) AS last_seen
FROM deauth_events
WHERE received_at > ?
GROUP BY reason
ORDER BY total_frames DESC
""")
""", (cutoff_7d,))
for r in reason_rows:
desc, classification = deauth_reason_info(r["reason"])
r["description"] = desc
r["classification"] = classification
# Activity heatmap — frame count per day × hour
# Activity heatmap — frame count per day × hour, last 7 days
heatmap_rows = query("""
SELECT date(received_at) AS day,
CAST(strftime('%H', received_at) AS INTEGER) AS hour,
COUNT(*) AS frames
FROM deauth_events
WHERE received_at > ?
GROUP BY day, hour
ORDER BY day, hour
""")
""", (cutoff_7d,))
return {
result = {
"recent": recent,
"bursts": bursts,
"summary": summary[0] if summary else {},
@@ -645,6 +715,8 @@ def build_alerts() -> dict:
"reason_stats": reason_rows,
"heatmap": heatmap_rows,
}
_alerts_cache = (_now_ts, result)
return result
def build_search(q: str) -> dict:
@@ -893,30 +965,56 @@ def ensure_schema():
)
""")
conn.executescript("""
CREATE INDEX IF NOT EXISTS idx_beacon_received_at ON beacon_events (received_at);
CREATE INDEX IF NOT EXISTS idx_beacon_node_id ON beacon_events (node_id);
CREATE INDEX IF NOT EXISTS idx_beacon_bssid ON beacon_events (bssid);
CREATE INDEX IF NOT EXISTS idx_probe_received_at ON probe_events (received_at);
CREATE INDEX IF NOT EXISTS idx_probe_node_id ON probe_events (node_id);
CREATE INDEX IF NOT EXISTS idx_probe_src_mac ON probe_events (src_mac);
CREATE INDEX IF NOT EXISTS idx_heartbeat_node_id ON heartbeat_events (node_id);
CREATE INDEX IF NOT EXISTS idx_deauth_received_at ON deauth_events (received_at);
CREATE INDEX IF NOT EXISTS idx_deauth_node_id ON deauth_events (node_id);
CREATE INDEX IF NOT EXISTS idx_deauth_bssid ON deauth_events (bssid);
CREATE INDEX IF NOT EXISTS idx_deauth_src ON deauth_events (src);
CREATE INDEX IF NOT EXISTS idx_deauth_dst ON deauth_events (dst);
CREATE INDEX IF NOT EXISTS idx_assoc_received_at ON assoc_events (received_at);
CREATE INDEX IF NOT EXISTS idx_assoc_node_id ON assoc_events (node_id);
CREATE INDEX IF NOT EXISTS idx_assoc_src ON assoc_events (src);
CREATE INDEX IF NOT EXISTS idx_assoc_bssid ON assoc_events (bssid);
CREATE INDEX IF NOT EXISTS idx_beacon_received_at ON beacon_events (received_at);
CREATE INDEX IF NOT EXISTS idx_beacon_node_id ON beacon_events (node_id);
CREATE INDEX IF NOT EXISTS idx_beacon_bssid ON beacon_events (bssid);
CREATE INDEX IF NOT EXISTS idx_beacon_bssid_time ON beacon_events (bssid, received_at);
CREATE INDEX IF NOT EXISTS idx_probe_received_at ON probe_events (received_at);
CREATE INDEX IF NOT EXISTS idx_probe_node_id ON probe_events (node_id);
CREATE INDEX IF NOT EXISTS idx_probe_src_mac ON probe_events (src_mac);
CREATE INDEX IF NOT EXISTS idx_probe_src_time ON probe_events (src_mac, received_at);
CREATE INDEX IF NOT EXISTS idx_heartbeat_node_id ON heartbeat_events (node_id);
CREATE INDEX IF NOT EXISTS idx_deauth_received_at ON deauth_events (received_at);
CREATE INDEX IF NOT EXISTS idx_deauth_node_id ON deauth_events (node_id);
CREATE INDEX IF NOT EXISTS idx_deauth_bssid ON deauth_events (bssid);
CREATE INDEX IF NOT EXISTS idx_deauth_bssid_time ON deauth_events (bssid, received_at);
CREATE INDEX IF NOT EXISTS idx_deauth_src ON deauth_events (src);
CREATE INDEX IF NOT EXISTS idx_deauth_dst ON deauth_events (dst);
CREATE INDEX IF NOT EXISTS idx_assoc_received_at ON assoc_events (received_at);
CREATE INDEX IF NOT EXISTS idx_assoc_node_id ON assoc_events (node_id);
CREATE INDEX IF NOT EXISTS idx_assoc_src ON assoc_events (src);
CREATE INDEX IF NOT EXISTS idx_assoc_bssid ON assoc_events (bssid);
""")
conn.commit()
PRUNE_INTERVAL_SECS = 6 * 3600 # run pruning every 6 hours
def prune_old_events() -> None:
"""Delete events older than HOT_DAYS from all event tables."""
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()
app = FastAPI(title="ESP32 Recon Dashboard")
ensure_schema()
@app.on_event("startup")
async def start_pruning_task():
async def _pruner():
while True:
await asyncio.sleep(PRUNE_INTERVAL_SECS)
await asyncio.to_thread(prune_old_events)
asyncio.create_task(_pruner())
@app.get("/", response_class=HTMLResponse)
async def index():
return FileResponse(HTML_PATH)
@@ -934,73 +1032,77 @@ async def serve_js():
@app.get("/api/nodes")
async def api_nodes():
return build_nodes()
return await asyncio.to_thread(build_nodes)
@app.get("/api/events")
async def api_events(limit: int = EVENT_CAP):
limit = min(limit, EVENT_CAP)
return build_events(limit=limit)
return await asyncio.to_thread(build_events, limit)
@app.get("/api/networks")
async def api_networks():
return build_networks()
return await asyncio.to_thread(build_networks)
@app.get("/api/clients")
async def api_clients():
return build_clients()
return await asyncio.to_thread(build_clients)
@app.get("/api/cross-node")
async def api_cross_node():
return build_cross_node()
return await asyncio.to_thread(build_cross_node)
@app.get("/api/presence")
async def api_presence():
return build_presence()
return await asyncio.to_thread(build_presence)
@app.get("/api/rssi-history")
async def api_rssi_history(bssid: str, hours: int = 2):
hours = min(max(hours, 1), 48)
cutoff = (datetime.datetime.utcnow() - datetime.timedelta(hours=hours)).isoformat(timespec="seconds")
rows = query("""
SELECT node_id, received_at, rssi
FROM beacon_events
WHERE bssid = ? AND received_at >= ?
ORDER BY received_at ASC
""", (bssid, cutoff))
series: dict = {}
for r in rows:
nid = r["node_id"]
if nid not in series:
series[nid] = []
series[nid].append({"t": r["received_at"], "rssi": r["rssi"]})
return series
def _fetch():
rows = query("""
SELECT node_id, received_at, rssi
FROM beacon_events
WHERE bssid = ? AND received_at >= ?
ORDER BY received_at ASC
""", (bssid, cutoff))
series: dict = {}
for r in rows:
nid = r["node_id"]
if nid not in series:
series[nid] = []
series[nid].append({"t": r["received_at"], "rssi": r["rssi"]})
return series
return await asyncio.to_thread(_fetch)
@app.get("/api/sessions")
async def api_sessions():
return build_sessions()
return await asyncio.to_thread(build_sessions)
@app.get("/api/search")
async def api_search(q: str = ""):
return build_search(q)
return await asyncio.to_thread(build_search, q)
@app.get("/api/alerts")
async def api_alerts():
return build_alerts()
return await asyncio.to_thread(build_alerts)
@app.get("/api/heartbeats")
async def api_heartbeats():
"""Latest heartbeat per node."""
return query("""
return await asyncio.to_thread(query, """
SELECT node_id, received_at, uptime_ms, free_heap, wifi_rssi
FROM heartbeat_events
WHERE id IN (SELECT MAX(id) FROM heartbeat_events GROUP BY node_id)
@@ -1009,7 +1111,7 @@ async def api_heartbeats():
@app.get("/api/nodes/{node_id}")
async def api_node_detail(node_id: str):
detail = build_node_detail(node_id)
detail = await asyncio.to_thread(build_node_detail, node_id)
if detail is None:
raise HTTPException(status_code=404, detail="Node not found")
return detail
@@ -1022,13 +1124,13 @@ async def api_stream():
try:
last_id = 0
# Start from the current latest event so we don't replay history on connect
rows = query("SELECT MAX(id) AS max_id FROM beacon_events")
rows = await asyncio.to_thread(query, "SELECT MAX(id) AS max_id FROM beacon_events")
if rows and rows[0]["max_id"]:
last_id = rows[0]["max_id"]
while True:
await asyncio.sleep(2)
new_rows = query("""
new_rows = await asyncio.to_thread(query, """
SELECT * FROM beacon_events
WHERE id > ?
ORDER BY id ASC