"""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 = """DibaGuard
DibaGuard console"""