File size: 10,338 Bytes
5d97275 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """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 " # list separator of the Persian report text
_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)
# the recommended action weighs what the source did in the window (alarms x severity), not only its decayed risk
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)
|