From c5d00be8205f767c2746ca05942c28c5debc852d Mon Sep 17 00:00:00 2001 From: CyberMind Date: Mon, 18 May 2026 08:21:17 +0200 Subject: [PATCH] fix(secubox-waf): apply double-cache pattern on /stats + /alerts (closes #145) (#146) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Symptom (live-bench 2026-05-15, board admin.gk2.secubox.in) ========================================================== When the SOC dashboard tab is open, secubox-waf.service sustains ~17 % CPU, 1035 syscalls/sec, 8 MB/sec read on /var/log/secubox/waf-threats.log. nginx access.log confirms the SOC frontend polls /api/v1/waf/stats every 5 s and /api/v1/waf/alerts?limit=100&aggregate=true every 9 s. Each call re-scans the full threat log (lines-by-line + GeoIP per-IP) — there is no caching despite CLAUDE.md mandating the double-cache pattern for WAF dashboards explicitly. Combined with ~20 other non-conformant SecuBox daemons (separate audit), this contributed to load-avg 4+ and TCP wedge on sshd / HAProxy. Fix === Mirror packages/secubox-crowdsec/api/main.py:106-138 — a thread-safe StatsCache with 30 s TTL. Read-through: cached = stats_cache.get(key) if cached is not None: return cached ... compute ... stats_cache.set(key, result) return result Cache wired on: GET /stats → key "threat_stats" GET /alerts → key f"alerts:{limit}:{int(aggregate)}" (keyed per view so SOC + drill-in don't collide) Cache invalidated explicitly on: POST /ban → bans affect /stats top_ips and /alerts visibility POST /unban/{ip} POST /reload → rules changed → cached counts stale Expected impact (under SOC polling) =================================== syscalls/sec : 1035 → ~30 (-97 %) read MB/sec : 8 → <0.1 (one log scan per 30 s instead of per req) WAF CPU : 17 % → <1 % sustained load avg : ~4 → ~2.5 Tests ===== 13 pytest cases covering cache get/set/TTL-expiry/invalidate, endpoint-level cache hit (loader called exactly once within TTL), per-view keying (limit/aggregate don't collide), and explicit invalidation by mutating endpoints. All pass. Out of scope ============ ~20 other daemons are non-conformant (eye-remote /health 5× I/O, cdn /health, threats /alerts, admin /summary, netdiag, netmodes, tor, droplet, gotosocial). To be covered by a separate "check-dashboard-cache.py" lint issue. Co-authored-by: CyberMind-FR Co-authored-by: Claude Opus 4.7 (1M context) --- packages/secubox-waf/api/main.py | 83 ++++++- packages/secubox-waf/tests/__init__.py | 0 .../secubox-waf/tests/test_stats_cache.py | 216 ++++++++++++++++++ 3 files changed, 295 insertions(+), 4 deletions(-) create mode 100644 packages/secubox-waf/tests/__init__.py create mode 100644 packages/secubox-waf/tests/test_stats_cache.py diff --git a/packages/secubox-waf/api/main.py b/packages/secubox-waf/api/main.py index b44928ce..285b59f6 100644 --- a/packages/secubox-waf/api/main.py +++ b/packages/secubox-waf/api/main.py @@ -7,6 +7,8 @@ import os import json import re import subprocess +import threading +import time from pathlib import Path from datetime import datetime, timedelta from typing import Dict, List, Optional, Any @@ -31,6 +33,52 @@ _category_stats: Dict[str, dict] = {} _request_counts: Dict[str, List[float]] = defaultdict(list) +# ═══════════════════════════════════════════════════════════════════════ +# Stats Cache — double-caching pattern (CLAUDE.md § Performance Patterns) +# +# Threat-stats + alerts are expensive (full WAF log scan + GeoIP lookups). +# SOC dashboard polls every 5 s; the underlying data changes at human pace. +# A 30 s read-through cache drops sustained CPU from ~17 % to <1 % under +# polling load without user-visible staleness. +# +# Mirrors `packages/secubox-crowdsec/api/main.py:106-138`. +# ═══════════════════════════════════════════════════════════════════════ + +class StatsCache: + """Thread-safe stats cache with TTL.""" + def __init__(self, ttl_seconds: int = 30): + self.ttl = ttl_seconds + self._cache: Dict[str, Any] = {} + self._timestamps: Dict[str, float] = {} + self._lock = threading.Lock() + + def get(self, key: str) -> Optional[Any]: + with self._lock: + if key in self._cache: + if time.time() - self._timestamps[key] < self.ttl: + return self._cache[key] + del self._cache[key] + del self._timestamps[key] + return None + + def set(self, key: str, value: Any): + with self._lock: + self._cache[key] = value + self._timestamps[key] = time.time() + + def invalidate(self, key: Optional[str] = None): + with self._lock: + if key: + self._cache.pop(key, None) + self._timestamps.pop(key, None) + else: + self._cache.clear() + self._timestamps.clear() + + +stats_cache = StatsCache(ttl_seconds=30) + + def _cfg(): cfg = get_config("waf") return { @@ -398,7 +446,15 @@ async def toggle_category(category: str, req: ToggleCategoryRequest): @app.get("/stats") async def get_stats(): - """Get threat statistics (public).""" + """Get threat statistics (public). + + Read-through cache (TTL 30 s) — the underlying threat log is scanned + at most every 30 s even under dashboard polling. See `StatsCache`. + """ + cached = stats_cache.get("threat_stats") + if cached is not None: + return cached + stats = _get_threat_stats() # Add dashboard-friendly fields @@ -471,6 +527,7 @@ async def get_stats(): except Exception: pass + stats_cache.set("threat_stats", stats) return stats @@ -478,13 +535,24 @@ async def get_stats(): async def get_alerts(limit: int = 50, aggregate: bool = False): """Get recent threat alerts (public). + Read-through cache (TTL 30 s) keyed on (limit, aggregate). Tail-reading + the threat log is the expensive part; dashboard pollers hit a cached + response. See `StatsCache`. + Args: limit: Max alerts to return aggregate: If True, group alerts by IP with counts """ + cache_key = f"alerts:{limit}:{int(aggregate)}" + cached = stats_cache.get(cache_key) + if cached is not None: + return cached + log_path = Path(THREATS_LOG) if not log_path.exists(): - return {"alerts": []} + empty = {"alerts": []} + stats_cache.set(cache_key, empty) + return empty alerts = [] try: @@ -531,9 +599,13 @@ async def get_alerts(limit: int = 50, aggregate: bool = False): "last_seen": data.get("last_seen"), "latest_alert": data["alerts"][0] if data["alerts"] else None, }) - return {"alerts": aggregated, "aggregated": True} + response = {"alerts": aggregated, "aggregated": True} + stats_cache.set(cache_key, response) + return response - return {"alerts": alerts[:limit], "aggregated": False} + response = {"alerts": alerts[:limit], "aggregated": False} + stats_cache.set(cache_key, response) + return response @app.get("/bans") @@ -553,6 +625,7 @@ class BanRequest(BaseModel): async def ban_ip(req: BanRequest): """Manually ban an IP.""" _ban_ip(req.ip, req.duration, req.reason) + stats_cache.invalidate() # bans affect /stats top_ips + /alerts visibility return {"success": True, "ip": req.ip, "duration": req.duration} @@ -560,6 +633,7 @@ async def ban_ip(req: BanRequest): async def unban_ip(ip: str): """Remove IP ban.""" _unban_ip(ip) + stats_cache.invalidate() return {"success": True, "ip": ip} @@ -606,6 +680,7 @@ async def check_threat(req: CheckRequest): async def reload_rules(): """Reload WAF rules from file.""" _load_rules() + stats_cache.invalidate() # rules changed → cached stats stale total_rules = sum(len(p) for p in _compiled_patterns.values()) return { "success": True, diff --git a/packages/secubox-waf/tests/__init__.py b/packages/secubox-waf/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/packages/secubox-waf/tests/test_stats_cache.py b/packages/secubox-waf/tests/test_stats_cache.py new file mode 100644 index 00000000..52d13e5b --- /dev/null +++ b/packages/secubox-waf/tests/test_stats_cache.py @@ -0,0 +1,216 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gérald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. +"""Tests for StatsCache + endpoint cache wiring in secubox-waf. + +Pattern reference: CLAUDE.md § Performance Patterns — Double Caching. +Mirrors `packages/secubox-crowdsec/api/main.py:106-138`. + +Verifies: +- StatsCache.get/set roundtrip +- TTL expiry returns None +- invalidate(key) clears one key, invalidate() clears all +- /stats and /alerts wire through the cache (hit on second call, + underlying loader called exactly once within TTL) +- /alerts cache is keyed on (limit, aggregate) so different views + don't collide +- Mutating endpoints (ban/unban/reload) invalidate the cache +""" +from __future__ import annotations + +import sys +import time +from pathlib import Path +from unittest.mock import patch + +import pytest + +# Make the package importable without an install step +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +# Stub out secubox_core before importing waf.api — production deps +# (auth, config) aren't needed for these unit tests. +from types import ModuleType + +if "secubox_core" not in sys.modules: + secubox_core = ModuleType("secubox_core") + auth_mod = ModuleType("secubox_core.auth") + auth_mod.require_jwt = lambda: None + config_mod = ModuleType("secubox_core.config") + config_mod.get_config = lambda *_a, **_k: {} + secubox_core.auth = auth_mod + secubox_core.config = config_mod + sys.modules["secubox_core"] = secubox_core + sys.modules["secubox_core.auth"] = auth_mod + sys.modules["secubox_core.config"] = config_mod + +# geoip2 may or may not be installed in CI; stub it out. +if "geoip2" not in sys.modules: + geoip2 = ModuleType("geoip2") + geoip2_db = ModuleType("geoip2.database") + geoip2_db.Reader = object + geoip2_err = ModuleType("geoip2.errors") + class _AddressNotFoundError(Exception): ... + geoip2_err.AddressNotFoundError = _AddressNotFoundError + sys.modules["geoip2"] = geoip2 + sys.modules["geoip2.database"] = geoip2_db + sys.modules["geoip2.errors"] = geoip2_err + +from api import main as waf_main # noqa: E402 +from api.main import StatsCache # noqa: E402 + + +# ────────────────────────────────────────────────────────────────────── +# StatsCache unit tests +# ────────────────────────────────────────────────────────────────────── + +def test_get_miss_returns_none(): + c = StatsCache(ttl_seconds=10) + assert c.get("absent") is None + + +def test_set_then_get_returns_value(): + c = StatsCache(ttl_seconds=10) + c.set("k", {"a": 1}) + assert c.get("k") == {"a": 1} + + +def test_get_after_ttl_expires_returns_none(): + c = StatsCache(ttl_seconds=10) + c.set("k", "v") + # Fake elapsed time without sleep + c._timestamps["k"] = time.time() - 11 + assert c.get("k") is None + # And the entry is evicted on miss + assert "k" not in c._cache + + +def test_invalidate_single_key(): + c = StatsCache(ttl_seconds=10) + c.set("a", 1) + c.set("b", 2) + c.invalidate("a") + assert c.get("a") is None + assert c.get("b") == 2 + + +def test_invalidate_all(): + c = StatsCache(ttl_seconds=10) + c.set("a", 1) + c.set("b", 2) + c.invalidate() + assert c.get("a") is None + assert c.get("b") is None + + +def test_set_overwrites_and_refreshes_ttl(): + c = StatsCache(ttl_seconds=10) + c.set("k", "v1") + c._timestamps["k"] = time.time() - 9 # still fresh + c.set("k", "v2") + # Should still be fresh and updated + assert c.get("k") == "v2" + + +# ────────────────────────────────────────────────────────────────────── +# Endpoint integration: cache hit avoids second loader call +# ────────────────────────────────────────────────────────────────────── + +@pytest.fixture(autouse=True) +def _reset_module_cache(): + """Each test starts with an empty module-level cache.""" + waf_main.stats_cache.invalidate() + yield + waf_main.stats_cache.invalidate() + + +def _run(coro): + """Tiny sync runner so we don't need pytest-asyncio for one call.""" + import asyncio + return asyncio.new_event_loop().run_until_complete(coro) + + +def test_get_stats_caches_underlying_loader(): + """Second call within TTL must not re-invoke _get_threat_stats.""" + fake = {"total_threats": 42, "by_category": {}, "by_severity": {}, + "top_ips": {}, "top_countries": {}, "top_vhosts": {}, + "top_ips_countries": {}, "threats_today": 0} + with patch.object(waf_main, "_get_threat_stats", return_value=fake) as loader, \ + patch.object(waf_main, "_cfg", return_value={"enabled": True}), \ + patch.object(waf_main, "_compiled_patterns", {}): + r1 = _run(waf_main.get_stats()) + r2 = _run(waf_main.get_stats()) + assert loader.call_count == 1, "loader called more than once → cache bypassed" + assert r1 == r2 + assert r1["total_threats"] == 42 + + +def test_get_stats_recomputes_after_ttl(): + fake1 = {"total_threats": 1, "by_category": {}, "by_severity": {}, + "top_ips": {}, "top_countries": {}, "top_vhosts": {}, + "top_ips_countries": {}, "threats_today": 0} + fake2 = dict(fake1, total_threats=2) + with patch.object(waf_main, "_get_threat_stats", side_effect=[fake1, fake2]) as loader, \ + patch.object(waf_main, "_cfg", return_value={"enabled": True}), \ + patch.object(waf_main, "_compiled_patterns", {}): + r1 = _run(waf_main.get_stats()) + # Force TTL expiry without real sleep + waf_main.stats_cache._timestamps["threat_stats"] = time.time() - 31 + r2 = _run(waf_main.get_stats()) + assert loader.call_count == 2 + assert r1["total_threats"] == 1 + assert r2["total_threats"] == 2 + + +def test_get_alerts_cache_keyed_per_view(tmp_path): + """Different (limit, aggregate) params must not collide in the cache.""" + log = tmp_path / "waf-threats.log" + log.write_text('{"client_ip":"1.2.3.4","category":"xss","severity":"high","timestamp":"2026-05-15T15:00:00"}\n') + with patch.object(waf_main, "THREATS_LOG", str(log)): + r1 = _run(waf_main.get_alerts(limit=10, aggregate=False)) + r2 = _run(waf_main.get_alerts(limit=10, aggregate=True)) + r3 = _run(waf_main.get_alerts(limit=10, aggregate=False)) # hit + assert r1["aggregated"] is False + assert r2["aggregated"] is True + assert r1 == r3, "same (limit, aggregate) tuple must hit cache" + # All three keys recorded + assert "alerts:10:0" in waf_main.stats_cache._cache + assert "alerts:10:1" in waf_main.stats_cache._cache + + +def test_get_alerts_missing_log_caches_empty(): + """Even the empty-log fast path is cached so the .exists() check doesn't repeat.""" + with patch.object(waf_main, "THREATS_LOG", "/nonexistent/path/to/log"): + r1 = _run(waf_main.get_alerts(limit=50, aggregate=False)) + # Mutate the cache directly to detect re-execution + waf_main.stats_cache._cache["alerts:50:0"] = {"sentinel": True} + r2 = _run(waf_main.get_alerts(limit=50, aggregate=False)) + assert r1 == {"alerts": []} + assert r2 == {"sentinel": True}, "second call bypassed cache" + + +def test_ban_invalidates_cache(): + """Manually banning an IP must clear cached stats/alerts.""" + waf_main.stats_cache.set("threat_stats", {"stale": True}) + waf_main.stats_cache.set("alerts:50:0", {"stale": True}) + with patch.object(waf_main, "_ban_ip"): + req = waf_main.BanRequest(ip="1.2.3.4", duration="4h", reason="test") + _run(waf_main.ban_ip(req)) + assert waf_main.stats_cache.get("threat_stats") is None + assert waf_main.stats_cache.get("alerts:50:0") is None + + +def test_unban_invalidates_cache(): + waf_main.stats_cache.set("threat_stats", {"stale": True}) + with patch.object(waf_main, "_unban_ip"): + _run(waf_main.unban_ip("1.2.3.4")) + assert waf_main.stats_cache.get("threat_stats") is None + + +def test_reload_invalidates_cache(): + waf_main.stats_cache.set("threat_stats", {"stale": True}) + with patch.object(waf_main, "_load_rules"), \ + patch.object(waf_main, "_compiled_patterns", {}): + _run(waf_main.reload_rules()) + assert waf_main.stats_cache.get("threat_stats") is None