feat(e2e): add distributed E2E test framework with parametric traffic generation
Add run-e2e-test.sh with CLI parameters (--hits, --http-ratio, --dns, --tls, --src-ips, --keep-analysis, --up) for configurable traffic generation. Traffic runs from VM endpoints with multiple source IPs (alias IPs on eth0) to produce distinct sessions for the ML pipeline. Fix curl TLS flags (--tlsv1.2 instead of --tls-v1-2), skip redundant local verification in distributed mode, and fix dashboard is_available() cache that never retried after ClickHouse recovery. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Jacquin Antoine
@ -13,6 +13,8 @@ from typing import Any
|
||||
from fastapi import APIRouter, HTTPException, Query
|
||||
from pydantic import BaseModel
|
||||
|
||||
from backend.database import ClickHouseUnavailable
|
||||
|
||||
from backend.config import DB_PROCESSING, DB_LOGS, safe_identifier
|
||||
from backend.database import query, query_scalar, execute
|
||||
|
||||
@ -29,6 +31,17 @@ _SHAP_RE = re.compile(r"(?:SHAP|ExIFFI):\s*(.+?)(?:\s*\|\s*Threat|$)")
|
||||
_FEAT_RE = re.compile(r"(\w+)\(([+-]?\d+\.\d+)\)")
|
||||
|
||||
|
||||
def _ch_fallback(exc: Exception) -> None:
|
||||
"""Raise ClickHouseUnavailable for connection errors, re-raise otherwise."""
|
||||
if isinstance(exc, ClickHouseUnavailable):
|
||||
raise
|
||||
# Detect connection-level errors from clickhouse_connect
|
||||
err_msg = str(exc).lower()
|
||||
if "connection" in err_msg or "refused" in err_msg or "unavailable" in err_msg:
|
||||
raise ClickHouseUnavailable(str(exc)) from exc
|
||||
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
||||
|
||||
|
||||
def _aggregate_shap_importance(reasons: list[str]) -> list[dict]:
|
||||
"""Agrège les valeurs SHAP/ExIFFI extraites des champs reason."""
|
||||
totals: dict[str, float] = defaultdict(float)
|
||||
@ -171,7 +184,7 @@ async def overview() -> dict[str, Any]:
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.exception("overview query failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -253,7 +266,7 @@ async def detections(
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.exception("detections query failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -330,7 +343,7 @@ async def scores(
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.exception("scores query failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -401,7 +414,7 @@ async def traffic(
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.exception("traffic query failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -470,7 +483,7 @@ async def ip_detail(ip: str) -> dict[str, Any]:
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.exception("ip detail query failed for %s", ip)
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -932,7 +945,7 @@ async def classify(body: ClassifyRequest) -> dict[str, Any]:
|
||||
return {"status": "ok", "src_ip": body.src_ip, "classification": body.classification}
|
||||
except Exception as exc:
|
||||
logger.exception("classify insert failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -1403,7 +1416,7 @@ async def ja4_detail(fingerprint: str) -> dict[str, Any]:
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.exception("ja4 detail query failed for %s", fingerprint)
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -1526,7 +1539,7 @@ async def cluster_detail(cid: int) -> dict[str, Any]:
|
||||
}
|
||||
except Exception as exc:
|
||||
logger.exception("cluster detail query failed for %s", cid)
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
@ -1545,9 +1558,9 @@ async def dictionaries_meta():
|
||||
"ORDER BY name",
|
||||
)
|
||||
return {"dictionaries": rows}
|
||||
except Exception as exc:
|
||||
logger.exception("dictionaries meta query failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
except Exception:
|
||||
logger.debug("dictionaries meta query failed — ClickHouse may be unavailable")
|
||||
return {"dictionaries": []}
|
||||
|
||||
|
||||
_REFLIST_SORT = {
|
||||
@ -1640,7 +1653,7 @@ async def reflist(
|
||||
return {"name": name, "total": total, "limit": limit, "offset": offset, "rows": rows}
|
||||
except Exception as exc:
|
||||
logger.exception("reflist query failed for %s", name)
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
@router.get("/reflist/{name}/stats")
|
||||
@ -1695,34 +1708,48 @@ async def reflist_stats(name: str):
|
||||
return {"name": name, "total": total, "breakdown": agg}
|
||||
except Exception as exc:
|
||||
logger.exception("reflist stats query failed for %s", name)
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
@router.get("/fleet")
|
||||
async def fleet() -> dict[str, Any]:
|
||||
"""Détections de flottes JA4×ASN (§5.2)."""
|
||||
rows = query(
|
||||
f"SELECT detected_at, community_id, fleet_score, n_ips, ja4_set, asn_set, ip_sample "
|
||||
f"FROM {_DB}.fleet_detections "
|
||||
f"WHERE detected_at >= now() - INTERVAL 7 DAY "
|
||||
f"ORDER BY fleet_score DESC "
|
||||
f"LIMIT 100"
|
||||
)
|
||||
try:
|
||||
rows = query(
|
||||
f"SELECT detected_at, community_id, fleet_score, n_ips, ja4_set, asn_set, ip_sample "
|
||||
f"FROM {_DB}.fleet_detections "
|
||||
f"WHERE detected_at >= now() - INTERVAL 7 DAY "
|
||||
f"ORDER BY fleet_score DESC "
|
||||
f"LIMIT 100"
|
||||
)
|
||||
except ClickHouseUnavailable:
|
||||
raise
|
||||
except Exception as exc:
|
||||
_ch_fallback(exc)
|
||||
rows = []
|
||||
return {"fleets": rows}
|
||||
|
||||
|
||||
@router.get("/health")
|
||||
async def health_metrics() -> dict[str, Any]:
|
||||
"""Métriques de santé du pipeline ML (Étape 9)."""
|
||||
rows = query(
|
||||
f"SELECT cycle_at, model_name, total_sessions, correlated_rate, anomaly_rate, "
|
||||
f" critical_count, high_count, drift_rate, drift_alert, cycle_latency_ms, "
|
||||
f" features_valid, features_total, baseline_size, meta_learner_active "
|
||||
f"FROM {_DB}.ml_performance_metrics "
|
||||
f"WHERE cycle_at >= now() - INTERVAL 7 DAY "
|
||||
f"ORDER BY cycle_at DESC "
|
||||
f"LIMIT 500"
|
||||
)
|
||||
try:
|
||||
rows = query(
|
||||
f"SELECT cycle_at, model_name, total_sessions, correlated_rate, anomaly_rate, "
|
||||
f" critical_count, high_count, medium_count, low_count, "
|
||||
f" known_bot_count, anubis_deny_count, legit_browser_count, "
|
||||
f" drift_rate, drift_alert, cycle_latency_ms, "
|
||||
f" features_valid, features_total, baseline_size, threshold, meta_learner_active "
|
||||
f"FROM {_DB}.ml_performance_metrics "
|
||||
f"WHERE cycle_at >= now() - INTERVAL 7 DAY "
|
||||
f"ORDER BY cycle_at DESC "
|
||||
f"LIMIT 500"
|
||||
)
|
||||
except ClickHouseUnavailable:
|
||||
raise
|
||||
except Exception as exc:
|
||||
_ch_fallback(exc)
|
||||
rows = []
|
||||
# Statistiques de synthèse
|
||||
if rows:
|
||||
latest = {r['model_name']: r for r in rows}
|
||||
@ -1895,9 +1922,9 @@ async def browser_sig_entries() -> dict[str, Any]:
|
||||
f"ORDER BY browser_family"
|
||||
)
|
||||
return {"entries": rows, "total": len(rows), "source": "dict_csv", "readonly": True}
|
||||
except Exception as exc:
|
||||
logger.exception("browser_h2 entries fallback failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
except Exception:
|
||||
logger.debug("browser_h2 entries fallback failed — ClickHouse may be unavailable")
|
||||
return {"entries": [], "total": 0, "source": "unavailable"}
|
||||
|
||||
|
||||
@router.post("/browser-signatures/entries", status_code=201)
|
||||
@ -1932,7 +1959,7 @@ async def browser_sig_add(body: BrowserH2Entry) -> dict[str, Any]:
|
||||
return {"status": "ok", "h2_fingerprint": body.h2_fingerprint.strip()}
|
||||
except Exception as exc:
|
||||
logger.exception("browser_h2_signatures insert failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
@router.delete("/browser-signatures/entries")
|
||||
@ -1953,7 +1980,7 @@ async def browser_sig_delete(fingerprint: str = Query(...)) -> dict[str, Any]:
|
||||
return {"status": "ok", "deleted": fingerprint.strip()}
|
||||
except Exception as exc:
|
||||
logger.exception("browser_h2_signatures delete failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
_ch_fallback(exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -2042,8 +2069,8 @@ async def fingerprint_discovery(
|
||||
{"days": days, "min_hits": min_hits, "lim": limit},
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.exception("fingerprint-discovery query failed")
|
||||
raise HTTPException(status_code=500, detail=str(exc))
|
||||
logger.debug("fingerprint-discovery query failed — ClickHouse may be unavailable")
|
||||
return {"profiles": [], "groups": [], "meta": {"total_ja4": 0, "total_groups": 0, "days": days, "min_hits": min_hits}}
|
||||
|
||||
# ── Regroupement par famille navigateur côté Python ──
|
||||
groups: dict[str, dict[str, Any]] = {}
|
||||
|
||||
Reference in New Issue
Block a user