| """Incident reports: what the engine saw, in a fixed JSON shape (schema below) and in plain English / Persian. |
| |
| A report is the hand-over point to people and to language models: every number in it comes from the engine's |
| own state (no model is asked to narrate), the text is rendered from the same dict, and the schema is what an LLM |
| tool call receives. `REPORT_SCHEMA` is a JSON Schema so that any agent framework can validate it. |
| """ |
| from __future__ import annotations |
|
|
| import math |
| import time |
| from collections import Counter, defaultdict |
|
|
| from dibaguard.families import severity |
|
|
| REPORT_SCHEMA = { |
| "$schema": "https://json-schema.org/draft/2020-12/schema", |
| "title": "DibaGuard incident report", |
| "type": "object", |
| "properties": { |
| "generated_at": {"type": "string"}, |
| "window": {"type": "object", "properties": {"since": {"type": "number"}, "until": {"type": "number"}}}, |
| "level": {"type": "string", "enum": ["quiet", "elevated", "attack", "severe"]}, |
| "site_pressure": {"type": "number", "description": "share of recent traffic judged attack, 0-1 (decayed)"}, |
| "drift": {"type": "boolean", "description": "benign score distribution has drifted: recalibrate"}, |
| "counts": {"type": "object", "additionalProperties": {"type": "integer"}}, |
| "families": {"type": "object", "additionalProperties": {"type": "integer"}}, |
| "entities": {"type": "array", "items": {"type": "object", "properties": { |
| "entity": {"type": "string"}, "risk": {"type": "number"}, "alarms": {"type": "integer"}, |
| "families": {"type": "array", "items": {"type": "string"}}, "targets": {"type": "array", "items": {"type": "string"}}, |
| "ports": {"type": "array", "items": {"type": "integer"}}, "first": {"type": "number"}, "last": {"type": "number"}, |
| "max_score": {"type": "integer"}, "evidence": {"type": "array", "items": {"type": "string"}}, |
| "action": {"type": "string", "enum": ["monitor", "rate_limit", "challenge", "block"]}}}}, |
| "thresholds": {"type": "object"}, |
| "recommendations": {"type": "array", "items": {"type": "string"}}, |
| "text": {"type": "object", "properties": {"en": {"type": "string"}, "fa": {"type": "string"}}}, |
| }, |
| "required": ["generated_at", "level", "site_pressure", "entities", "text"], |
| } |
|
|
| _ACTION = [(0.9, "block"), (0.6, "challenge"), (0.3, "rate_limit"), (0.0, "monitor")] |
| _SEP_FA = "\u060c " |
| _FA_FAMILY = {"portscan": "\u0627\u0633\u06a9\u0646 \u067e\u0648\u0631\u062a", "dos": "DoS", "ddos": "DDoS", "bruteforce": "\u062d\u0645\u0644\u0647 brute force", "webattack": "\u062d\u0645\u0644\u0647 \u0648\u0628", |
| "botnet": "\u0628\u0627\u062a\u200c\u0646\u062a", "infiltration": "\u0646\u0641\u0648\u0630", "exploit": "\u0627\u06a9\u0633\u067e\u0644\u0648\u06cc\u062a", "backdoor": "\u062f\u0631\u0628 \u067e\u0634\u062a\u06cc", "recon": "\u0634\u0646\u0627\u0633\u0627\u06cc\u06cc", |
| "attack_generic": "\u062d\u0645\u0644\u0647 (\u062e\u0627\u0646\u0648\u0627\u062f\u0647 \u0646\u0627\u0645\u0634\u062e\u0635)", "sqli": "\u062a\u0632\u0631\u06cc\u0642 SQL", "xss": "XSS", "cmdi": "\u062a\u0632\u0631\u06cc\u0642 \u0641\u0631\u0645\u0627\u0646", |
| "traversal": "\u067e\u06cc\u0645\u0627\u06cc\u0634 \u0645\u0633\u06cc\u0631", "scan": "\u0627\u0633\u06a9\u0646\u0631", "bot": "\u0628\u0627\u062a", "rce": "\u0627\u062c\u0631\u0627\u06cc \u06a9\u062f \u0627\u0632 \u0631\u0627\u0647 \u062f\u0648\u0631"} |
|
|
|
|
| def _action(risk: float) -> str: |
| return next(a for lo, a in _ACTION if risk >= lo) |
|
|
|
|
| def build(guard, since: float | None = None, until: float | None = None, top: int = 10) -> dict: |
| """Report over the engine's alarm log in [since, until] (engine time units, usually unix seconds).""" |
| log = [e for e in guard.log if (since is None or e["t"] >= since) and (until is None or e["t"] <= until)] |
| now = until if until is not None else (max((e["t"] for e in log), default=time.time())) |
| per = defaultdict(lambda: {"alarms": 0, "families": Counter(), "targets": Counter(), "ports": Counter(), "first": None, |
| "last": None, "max_score": 0, "evidence": []}) |
| fam_all: Counter = Counter() |
| for e in log: |
| r = per[e["src"]] |
| r["alarms"] += 1 |
| r["families"][e["family"]] += 1 |
| fam_all[e["family"]] += 1 |
| r["targets"][e["dst"]] += 1 |
| r["ports"][e["port"]] += 1 |
| r["first"] = e["t"] if r["first"] is None else min(r["first"], e["t"]) |
| r["last"] = e["t"] if r["last"] is None else max(r["last"], e["t"]) |
| r["max_score"] = max(r["max_score"], e["score"]) |
| if e.get("evidence") and len(r["evidence"]) < 3: |
| r["evidence"].append(e["evidence"]) |
| ents = [] |
| for src, r in per.items(): |
| risk = guard.entities.risk(src, now) |
| |
| acted = max(risk, (1.0 - math.exp(-r["alarms"] / 20.0)) * max((severity(f) for f in r["families"]), default=0.6)) |
| ents.append({"entity": src, "risk": round(risk, 4), "alarms": r["alarms"], "families": [f for f, _ in r["families"].most_common(3)], |
| "targets": [t for t, _ in r["targets"].most_common(3)], "ports": [int(p) for p, _ in r["ports"].most_common(5)], |
| "first": r["first"], "last": r["last"], "max_score": int(r["max_score"]), "evidence": r["evidence"], |
| "action": _action(acted)}) |
| ents.sort(key=lambda e: (-(e["max_score"] * min(e["alarms"], 50)), -e["risk"])) |
| pressure = float(guard.pressure.pressure) |
| worst = max((severity(f) for f in fam_all), default=0.0) |
| level = ("severe" if pressure > 0.5 or (ents and ents[0]["risk"] > 0.9 and worst >= 0.8) |
| else "attack" if pressure > 0.1 or any(e["risk"] > 0.6 for e in ents) |
| else "elevated" if ents else "quiet") |
| rec = [] |
| if guard.drift.drifted: |
| rec.append("Benign score distribution drifted: re-run site calibration with fresh normal traffic.") |
| st = guard.status() |
| if any(v == 0 for k, v in st["site_scores"].items() if k in st["thresholds"]): |
| rec.append("No site calibration yet: thresholds are the shipped ones; feed 30 minutes of normal traffic.") |
| for e in ents[:top]: |
| if e["action"] in ("block", "challenge"): |
| rec.append(f"{e['action']} {e['entity']} ({', '.join(e['families'])}, risk {e['risk']:.2f})") |
| rep = {"generated_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "window": {"since": since, "until": until}, |
| "level": level, "site_pressure": round(pressure, 5), "drift": bool(guard.drift.drifted), |
| "counts": {**guard.counts, "alarms_in_window": len(log), "entities_in_window": len(ents)}, |
| "families": dict(fam_all.most_common()), "entities": ents[:top], |
| "thresholds": {k: {"value": round(float(v[0]), 4), "source": v[1]} for k, v in st["thresholds"].items()}, |
| "recommendations": rec} |
| rep["text"] = {"en": render_en(rep), "fa": render_fa(rep)} |
| return rep |
|
|
|
|
| def render_en(r: dict) -> str: |
| lines = [f"DibaGuard report - level {r['level'].upper()}, attack pressure {100 * r['site_pressure']:.1f}% of recent traffic" |
| + (", DRIFT detected" if r["drift"] else "") + "."] |
| c = r["counts"] |
| lines.append(f"Seen: {c.get('flows', 0):,} flows, {c.get('http', 0):,} HTTP requests; {c.get('alarms_in_window', 0):,} alarms " |
| f"from {c.get('entities_in_window', 0)} entities in this window.") |
| if r["families"]: |
| lines.append("Families: " + ", ".join(f"{f} x{n}" for f, n in r["families"].items()) + ".") |
| for e in r["entities"]: |
| lines.append(f"- {e['entity']}: risk {e['risk']:.2f}, {e['alarms']} alarms ({', '.join(e['families'])}), " |
| f"targets {', '.join(e['targets'])}, ports {', '.join(map(str, e['ports']))}, max score {e['max_score']} -> {e['action']}.") |
| for ev in e["evidence"]: |
| lines.append(f" evidence: {ev}") |
| for x in r["recommendations"]: |
| lines.append(f"* {x}") |
| return "\n".join(lines) |
|
|
|
|
| def render_fa(r: dict) -> str: |
| lv = {"quiet": "\u0622\u0631\u0627\u0645", "elevated": "\u0628\u0627\u0644\u0627", "attack": "\u062d\u0645\u0644\u0647", "severe": "\u0634\u062f\u06cc\u062f"}[r["level"]] |
| lines = [f"\u06af\u0632\u0627\u0631\u0634 DibaGuard - \u0633\u0637\u062d {lv}\u060c \u0641\u0634\u0627\u0631 \u062d\u0645\u0644\u0647 {100 * r['site_pressure']:.1f}\u066a \u0627\u0632 \u062a\u0631\u0627\u0641\u06cc\u06a9 \u0627\u062e\u06cc\u0631" + ("\u060c \u0631\u0627\u0646\u0634 (drift) \u062f\u06cc\u062f\u0647 \u0634\u062f" if r["drift"] else "") + "."] |
| c = r["counts"] |
| lines.append(f"\u0645\u0634\u0627\u0647\u062f\u0647\u200c\u0634\u062f\u0647: {c.get('flows', 0):,} \u062c\u0631\u06cc\u0627\u0646\u060c {c.get('http', 0):,} \u062f\u0631\u062e\u0648\u0627\u0633\u062a HTTP\u061b {c.get('alarms_in_window', 0):,} \u0647\u0634\u062f\u0627\u0631 \u0627\u0632 " |
| f"{c.get('entities_in_window', 0)} \u0645\u0648\u062c\u0648\u062f\u06cc\u062a \u062f\u0631 \u0627\u06cc\u0646 \u0628\u0627\u0632\u0647.") |
| if r["families"]: |
| lines.append("\u062e\u0627\u0646\u0648\u0627\u062f\u0647\u200c\u0647\u0627: " + "\u060c ".join(f"{_FA_FAMILY.get(f, f)} ×{n}" for f, n in r["families"].items()) + ".") |
| act = {"block": "\u0645\u0633\u062f\u0648\u062f \u0634\u0648\u062f", "challenge": "\u0686\u0627\u0644\u0634 (captcha/JS)", "rate_limit": "\u0645\u062d\u062f\u0648\u062f\u06cc\u062a \u0646\u0631\u062e", "monitor": "\u067e\u0627\u06cc\u0634"} |
| for e in r["entities"]: |
| lines.append(f"- {e['entity']}: \u0631\u06cc\u0633\u06a9 {e['risk']:.2f}\u060c {e['alarms']} \u0647\u0634\u062f\u0627\u0631 ({_SEP_FA.join(_FA_FAMILY.get(f, f) for f in e['families'])})\u060c " |
| f"\u0627\u0647\u062f\u0627\u0641 {_SEP_FA.join(e['targets'])}\u060c \u067e\u0648\u0631\u062a\u200c\u0647\u0627 {_SEP_FA.join(map(str, e['ports']))}\u060c \u0628\u06cc\u0634\u06cc\u0646\u0647 \u0627\u0645\u062a\u06cc\u0627\u0632 {e['max_score']} ← {act[e['action']]}.") |
| for x in r["recommendations"]: |
| lines.append(f"* {x}") |
| return "\n".join(lines) |
|
|
