File size: 15,927 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 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 | """Feedback: the operator's word on alarms, and what the layer does with it - at once and later.
fb = Feedback(layer, path="feedback.jsonl")
fb.pending(limit=50) -> alarms awaiting a verdict (grouped per source, newest first)
fb.decide(entity, verdict, note) -> verdict in {"confirm", "reject", "benign_source"}
confirm the alarm was right: the policy's action is taken now (escalated to block if the policy allows)
reject this alarm was wrong: the action is lifted, the source's recent flows / requests join the site's
benign reference (the layer learns from it immediately), and the case is kept for retraining
benign_source this source is always fine (a scanner of yours, a monitoring probe): allowlisted with a TTL
fb.export(path) -> labelled rows (features / texts + verdict) for offline retraining
fb.stats() -> precision of the alarms as the operator sees them
Every verdict is appended to a jsonl file (who, when, what, why) and survives restarts; nothing updates model
weights online - that is the offline retraining job's input (guard_jobs/feedback_train.py).
"""
from __future__ import annotations
import json
import os
import time
import numpy as np
VERDICTS = ("confirm", "reject", "benign_source")
class Feedback:
def __init__(self, layer, path: str | None = None, allow_ttl_s: float = 7 * 86400, reject_ttl_s: float = 6 * 3600,
sustained_soft: int = 25):
self.layer, self.path, self.allow_ttl = layer, path, allow_ttl_s
self.reject_ttl = reject_ttl_s # a rejected case stays out of the queue this long, then may come back
self.sustained_soft = sustained_soft # soft (auto-challenged) alarms reach a human only when sustained
self.records: list = []
self.allow: dict = {} # entity -> until (benign_source)
if path and os.path.exists(path):
with open(path, encoding="utf-8") as f:
for line in f:
try:
r = json.loads(line)
self.records.append(r)
if r.get("verdict") == "benign_source":
self.allow[r["entity"]] = r.get("until", 0)
except ValueError:
continue
# an always-ok verdict outlives a restart: put the unexpired ones back into the policy's allowlist
pol = getattr(layer, "policy", None)
if pol is not None and self.allow:
import ipaddress
now = time.time()
for ent, until in self.allow.items():
if until > now:
net = ent if "/" in ent else (f"{ent}/128" if ":" in ent else f"{ent}/32")
if net not in pol.d.setdefault("allow", []):
pol.d["allow"].append(net)
pol._allow = [ipaddress.ip_network(a, strict=False) for a in pol.d["allow"]]
# -- what needs a verdict
def pending(self, limit: int = 50, since: float | None = None, now: float | None = None) -> list[dict]:
"""the review queue: sources whose alarms need a human. Soft alarms (automatically challenged / monitored
first-contact verdicts) only qualify when sustained; hard, behavioural and signature verdicts always do."""
now = time.time() if now is None else now
decided = {r["entity"] for r in self.records if r.get("verdict") in ("reject", "benign_source")
and r.get("until", r.get("ts", 0) + self.reject_ttl) > now}
by: dict = {}
for e in reversed(self.layer.guard.log):
if since is not None and e["t"] < since:
break
if e["src"] in decided:
continue
r = by.setdefault(e["src"], {"entity": e["src"], "alarms": 0, "hard": 0, "families": {}, "first": e["t"], "last": e["t"], "max_score": 0,
"targets": {}, "evidence": [], "enforced": self.layer.enforcer.lookup(e["src"])})
r["alarms"] += 1
r["hard"] += 0 if e.get("soft") else 1
r["families"][e["family"]] = r["families"].get(e["family"], 0) + 1
r["targets"][e["dst"]] = r["targets"].get(e["dst"], 0) + 1
r["first"], r["last"] = min(r["first"], e["t"]), 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"])
if len(by) >= 5000: # cap the scan, not the result: rank everything, then return the top `limit`
break
hard = [r for r in by.values() if r["hard"] > 0]
soft = [r for r in by.values() if r["hard"] == 0 and r["alarms"] >= self.sustained_soft]
out = hard
if len(soft) > 10 and "site:drift" not in decided:
# many sources deviating at once with no recognisable attack shape = the site's behaviour shifted (a busier hour,
# a new client population, a release). One item for a human, not one ticket per source.
out = hard + [{"entity": "site:drift", "alarms": sum(r["alarms"] for r in soft), "hard": 0,
"families": {"site_drift": len(soft)}, "first": min(r["first"] for r in soft), "last": max(r["last"] for r in soft),
"max_score": max(r["max_score"] for r in soft), "targets": {}, "evidence": [f"{len(soft)} sources deviating without an attack shape"],
"enforced": "monitor", "sources": len(soft)}]
else:
out = hard + soft
for r in out:
r["families"] = sorted(r["families"], key=lambda k: -r["families"][k])
r["targets"] = sorted(r["targets"], key=lambda k: -r["targets"][k])[:3]
out.sort(key=lambda r: (-r["max_score"], -r["alarms"]))
return out[:limit]
# -- guards against poisoning the loop (a wrong click, a hijacked console, an LLM talked into approving an attacker)
HARD_EVIDENCE = ("A0", "A1", "A2", "campaign:", "drift guard", "signature: rce", "signature: privilege", "signature: destructive",
"payload provenance")
HARD_FAMILIES = ("db_rce", "db_privilege", "db_destructive", "rce", "sqli", "nosqli", "xss", "traversal", "ssti", "xxe", "ssrf")
MAX_CLEARS_PER_HOUR = 50
# behaviour-only verdicts (the usual false alarms) stay one-click; content / database verdicts are judged by severity
BEHAVIOURAL = {"dos", "ddos", "portscan", "bruteforce", "recon", "bot", "scan", "attack_generic", "lateral", "benign",
"db_new_client", "db_recon", "db_unbounded", "db_bruteforce"}
DB_AS = {"db_injection": "sqli", "db_exfil": "exfil", "db_rce": "rce", "db_privilege": "rce", "db_destructive": "rce"}
def _guard(self, entity: str, verdict: str, case: list, now: float, force: bool) -> str | None:
"""-> why this verdict needs an explicit second confirmation (force=True), or None"""
if verdict == "confirm" or entity == "site:drift":
return None
if verdict == "benign_source" and "/" in entity:
import ipaddress
try:
net = ipaddress.ip_network(entity, strict=False)
except ValueError:
return "not an address or network"
if net.prefixlen < (24 if net.version == 4 else 48):
return f"{entity} is wider than a /24 (/48): allowlist networks in the policy, not through alarm feedback"
if force:
return None
from dibaguard.families import severity
# hard = what would cost most if it were wrongly cleared: a severe family (db_* judged as its family) or a named
# signature / campaign / provenance in the evidence. Behavioural verdicts (the usual false alarms) stay
# one-click - clearing a wrongly-blocked benign source is what this button is for - under the hourly cap.
hard = [e for e in case if e["family"] in self.HARD_FAMILIES or (str(e["family"]) not in self.BEHAVIOURAL
and severity(self.DB_AS.get(str(e["family"]), str(e["family"]).removeprefix("db_"))) >= 0.7)
or any(h in str(e.get("evidence", "")) for h in self.HARD_EVIDENCE)]
if hard:
return (f"{len(hard)} of this source's alarms carry hard evidence ({hard[-1]['family']}: {str(hard[-1].get('evidence', ''))[:120]}) - "
f"clearing it needs a second, explicit confirmation")
recent = sum(1 for r in self.records if r.get("verdict") in ("reject", "benign_source") and now - r.get("ts", 0) < 3600)
if recent >= self.MAX_CLEARS_PER_HOUR:
return f"{recent} alarms cleared within the last hour - more need an explicit confirmation (a burst of clears is how a loop gets poisoned)"
return None
def _append(self, rec: dict) -> None:
"""append-only journal with a hash chain: every record names the hash of the one before it, so an edited or
removed verdict breaks the chain (verify())"""
import hashlib
prev = self.records[-1].get("hash", "") if self.records else ""
rec["prev"] = prev
body = json.dumps({k: v for k, v in rec.items() if k != "hash"}, ensure_ascii=False, sort_keys=True, default=str)
rec["hash"] = hashlib.sha256((prev + body).encode("utf-8")).hexdigest()[:32]
self.records.append(rec)
if self.path:
with open(self.path, "a", encoding="utf-8") as f:
f.write(json.dumps(rec, ensure_ascii=False, default=str) + "\n")
def verify(self) -> dict:
"""check the journal's hash chain"""
import hashlib
prev, bad = "", []
for i, r in enumerate(self.records):
if "hash" not in r:
continue
body = json.dumps({k: v for k, v in r.items() if k != "hash"}, ensure_ascii=False, sort_keys=True, default=str)
if r.get("prev", "") != prev or hashlib.sha256((prev + body).encode("utf-8")).hexdigest()[:32] != r["hash"]:
bad.append(i)
prev = r["hash"]
return {"records": len(self.records), "chain_ok": not bad, "broken_at": bad[:10]}
# -- the verdict and its immediate effect
def decide(self, entity: str, verdict: str, note: str = "", who: str = "operator", now: float | None = None, force: bool = False) -> dict:
if verdict not in VERDICTS:
raise ValueError(f"verdict must be one of {VERDICTS}")
now = time.time() if now is None else now
L = self.layer
case = [e for e in L.guard.log if e["src"] == entity][-200:]
why = self._guard(entity, verdict, case, now, force)
if why:
return {"entity": entity, "verdict": verdict, "applied": False, "needs_confirmation": True, "why": why}
rec = {"ts": now, "who": who, "entity": entity, "verdict": verdict, "note": note, "alarms": len(case),
"families": sorted({e["family"] for e in case}), "evidence": [e["evidence"] for e in case if e.get("evidence")][:5],
"enforced_before": L.enforcer.lookup(entity), "forced": bool(force)}
if verdict == "reject":
rec["until"] = now + self.reject_ttl
if entity == "site:drift":
# the operator says the shifted behaviour is normal: rebuild the references from the recent traffic now
rec["recalibrated"] = bool(verdict != "confirm" and L.guard.recalibrate_now())
self._append(rec)
return rec
if verdict == "confirm":
fam = max((e["family"] for e in case), key=lambda f: sum(1 for e in case if e["family"] == f), default="attack_generic")
action = L.policy.action_for(fam, 100, entity)
if action == "monitor":
action = "block" if L.policy.enforcing else "monitor"
rec["taken"] = L.enforcer.apply(entity, action, fam, 100, reason=f"operator confirmed ({note})", now=now)
else:
if entity in L.enforcer.active: # lift whatever was done
st = L.enforcer.active.pop(entity)
if L.policy.enforcing and st["action"] == "block":
L.enforcer.backend.unblock(entity)
L.enforcer._record({"ip": entity, "action": "lift", "family": st["family"], "score": st["score"], "reason": f"operator {verdict}", "taken": L.policy.enforcing})
L.guard.entities.s.pop(entity, None) # risk back to zero
L.guard.evidence.e.pop(entity, None) # evidence back to zero
L.guard.log = [e for e in L.guard.log if e["src"] != entity]
if verdict == "benign_source":
rec["until"] = now + self.allow_ttl
self.allow[entity] = rec["until"]
L.policy.d.setdefault("allow", []).append(entity if ":" in entity else f"{entity}/32")
L.policy._allow = [__import__("ipaddress").ip_network(a, strict=False) for a in L.policy.d["allow"]]
# the rejected alarms' requests are normal for this site: they join the next HTTP re-profile; a rejected
# database query becomes one of that database's application shapes (a hard family only by this decision).
# One verdict teaches at most 20 requests: a decision is a correction, not a training set.
taught = 0
for e in case:
if e.get("kind", "").startswith("http") and e.get("evidence") and taught < 20:
taught += 1
L._http_recent_by.setdefault(str(e.get("dst", "")), []).append(e["evidence"])
elif e.get("kind") == "db" and e.get("query") and getattr(L, "dbg", None) is not None:
from dibaguard.shield.database import shape
shp = shape(e["query"])
L.dbg.known_shapes.setdefault(str(e.get("dst", "")), set()).add(shp)
if e["family"] in ("db_rce", "db_privilege", "db_destructive"):
L.dbg.exempt_shapes.add(shp)
rec.setdefault("learned_shapes", []).append(shp[:200])
self._append(rec)
return rec
def is_allowed(self, entity: str, now: float | None = None) -> bool:
now = time.time() if now is None else now
return self.allow.get(entity, 0) > now
# -- for retraining
def export(self, path: str) -> dict:
"""labelled cases for offline retraining: one row per (entity, verdict) with its evidence texts and families"""
rows = [{"entity": r["entity"], "label": 1 if r["verdict"] == "confirm" else 0, "verdict": r["verdict"], "families": r["families"],
"texts": r.get("evidence", []), "ts": r["ts"], "note": r.get("note", "")} for r in self.records]
with open(path, "w", encoding="utf-8") as f:
for r in rows:
f.write(json.dumps(r, ensure_ascii=False) + "\n")
return {"rows": len(rows), "confirmed": sum(r["label"] for r in rows), "rejected": sum(1 - r["label"] for r in rows), "path": path}
def stats(self) -> dict:
c = sum(1 for r in self.records if r["verdict"] == "confirm")
n = len(self.records)
return {"decided": n, "confirmed": c, "rejected": n - c, "precision_as_judged": round(c / n, 3) if n else None,
"allowlisted": sum(1 for u in self.allow.values() if u > time.time())}
# the review page is part of the operator console now (dibaguard/console.py, GET /console); kept for old links
DASHBOARD = """<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="0;url=/console"><title>DibaGuard</title>
</head><body><a href="/console">DibaGuard console</a></body></html>"""
|