"""The protective layer: engine verdicts -> policy -> enforcement, with position awareness, posture audit, playbook names and a report an operator or an LLM can act on. One object per instance, whatever its role. L = Layer.load("models", policy="policy.json", inputs={"http": True, "flows": True}) L.calibrate_flows(normal_flows); L.calibrate_http(normal_requests) L.flows(flows) -> engine verdicts + actions taken (block / rate_limit / challenge) per source L.http(texts, client=..., status=..., t=...) -> same for requests (+ request-level playbook matches) L.rpc(bodies, client=..., t=...) -> JSON-RPC calls as requests + events (chain nodes) L.posture() -> findings + what was closed / proposed L.tick(now) -> lifts expired actions, re-calibrates on schedule L.report(since) -> incident report + enforcement + posture + playbook names (JSON, en, fa) L.fast_state() -> the blocklist / challenge / rate-limit map the Go fast path polls """ from __future__ import annotations import functools import ipaddress import os import re import time import zlib from collections import OrderedDict, deque from urllib.parse import unquote_plus import numpy as np from dibaguard import report as rep_mod from dibaguard.engine import Guard from dibaguard.flows import entity as _ent from dibaguard.http import sqlctx from dibaguard.shield import adapters, playbook, position, posture, rules as rules_mod from dibaguard.shield.database import DbGuard, _Decay from dibaguard.shield.enforce import Enforcer from dibaguard.shield.feedback import Feedback from dibaguard.shield.policy import Policy def _ua_of(text: str) -> str: for part in text.split(" | "): if part.startswith("h:user-agent="): return part[13:].lower() return "" def _machine_ua(ua: str) -> bool: """a client that says it is not a browser (no Mozilla token, or a crawler / bot token)""" return "mozilla/" not in ua or any(b in ua for b in ("bot", "crawl", "spider", "slurp")) @functools.lru_cache(maxsize=65536) def _ip_entity(ip: str, pfx: int) -> str: if ":" not in ip or "/" in ip or pfx >= 128: return ip try: a = ipaddress.ip_address(ip) except ValueError: return ip if a.version != 6 or a.ipv4_mapped is not None: return ip return str(ipaddress.ip_network(f"{ip}/{pfx}", strict=False)) class Layer: def __init__(self, guard: Guard, policy: Policy, inputs: dict | None = None, backend: str = "auto", audit_path: str | None = None, feedback_path: str | None = None, llm_interpret=None): self.guard, self.policy = guard, policy self.position = position.detect(inputs) self.enforcer = Enforcer(policy, backend, audit_path) self.feedback = Feedback(self, feedback_path) self.rules = rules_mod.RuleSet(policy.d.get("rules", [])) self.llm_interpret = llm_interpret # optional: text -> rule dict via an LLM (built-in parser otherwise) self.posture_findings: list = [] self.playbook_hits: dict = {} # playbook id -> count self.http_boot_n = int(policy.d.get("http_bootstrap_requests", 1000)) self.http_reprofile_s = float(policy.d.get("http_reprofile_s", 1800)) self.http_max_hosts = int(policy.d.get("http_max_hosts", 256)) # sites with a profile of their own (Host values) self._http_last = 0.0 self._http_boot_by: dict = {} # per-host bootstrap buffers self._http_recent_by: dict = {} # per-host non-alarmed requests of the current epoch self._http_last_by: dict = {} dbp = policy.d["database"] self.dbg = DbGuard(None, alpha=float(policy.d["alpha"]), auth_fail_floor=float(dbp["auth_fail_floor"]), exempt_shapes=set(dbp.get("exempt_shapes", [])), learn_new_clients=bool(dbp.get("learn_new_clients"))) self._http_ring: deque = deque(maxlen=50_000) # (t, client, decoded text) of recent requests: payload provenance for the DB tier self._req_rate = _Decay(60.0) # requests per client, ~per minute (rules: request_rate_over) self._ua_seen: "OrderedDict[str, OrderedDict]" = OrderedDict() # ip -> recent user-agent hashes (NAT / CGNAT detection) # requests per address and how many of them alarmed (~10 min): an address that looks shared but whose traffic is # mostly attack is not a NAT full of users - eight user-agents rotated by one stuffer must not buy a challenge self._addr_req, self._addr_bad = _Decay(600.0), _Decay(600.0) self._ua_by: dict = {} # entity -> user-agent of its latest request (this batch) self._sql_evidence: dict = {} self._sql_soft_src: dict = {} # (host, parameter) -> clients whose value only DbNet flagged self._sql_site_syntax: set = set() # (host, parameter) learned as the site's own syntax @classmethod def load(cls, model_dir: str, policy: str | dict | None = None, inputs: dict | None = None, backend: str = "auto", audit_path: str | None = None, feedback_path: str | None = None, llm_interpret=None, **guard_kw) -> "Layer": pol = Policy(policy) if isinstance(policy, dict) else Policy.load(policy) L = cls(Guard.load(model_dir, alpha=float(pol.d["alpha"]), **guard_kw), pol, inputs, backend, audit_path, feedback_path, llm_interpret) if os.path.exists(os.path.join(model_dir, "dbnet.npz")): from dibaguard.dbnet import DbNet L.dbg.net = DbNet.load(os.path.join(model_dir, "dbnet.npz")) if pol.d.get("rule_texts"): L.add_rules(pol.d["rule_texts"]) return L def add_rules(self, texts: list[str], interpret_only: bool = False) -> dict: """plain-language rules (fa/en) -> precise rules active immediately; returns each interpretation to confirm. interpret_only=True shows what it would do without installing them.""" interp = self.llm_interpret or rules_mod.parse out = [] for i, text in enumerate(texts): if interpret_only: d = interp(text) out.append({"text": text, "interpreted_as": d.get("note", ""), "rule": d}) else: r = self.rules.add(text, interp, len(self.rules.rules) + 1) out.append({"id": r.id, "text": text, "interpreted_as": r.note, "action": r.action, "scope": r.scope, "when": r.when}) return {"rules": out, "active": not interpret_only} # -- calibration passes straight through def calibrate_flows(self, flows: dict) -> dict: return self.guard.calibrate_flows(flows) def calibrate_http(self, texts, deepest=None) -> dict: return self.guard.calibrate_http(texts, deepest) # -- shared addresses (NAT, CGNAT, corporate proxies) _CGNAT = ipaddress.ip_network("100.64.0.0/10") def ip_entity(self, ip) -> str: """the address a source is judged and acted on as: an IPv6 address stands for its /64 (policy ipv6_entity_prefix) - one subscriber owns a /64, and a fresh address per request inside it must not make every request a newborn source (the enforcer already blocks the /64)""" return _ip_entity(str(ip), int(self.policy.d.get("ipv6_entity_prefix", 64))) def mostly_attack(self, ip: str, t: float) -> bool: """most of this address's recent requests alarmed or failed (>= 70% of >= 20 in ~10 min): a NAT's users are mostly served, a stuffer's guesses mostly fail""" n = self._addr_req.value(ip, t) return n >= 20.0 and self._addr_bad.value(ip, t) >= 0.7 * n def shared(self, ip: str, t: float | None = None) -> bool: """an address many people use: in the carrier-grade NAT range, or seen with >= nat_min_agents distinct user-agents within 15 minutes. Its users are judged apart (sub-entities) and it is never blocked as a whole""" ip = str(ip).split("#", 1)[0] try: if ipaddress.ip_address(ip) in self._CGNAT: return True except ValueError: if "/" not in ip: # (an IPv6 /64 entity is shared the same way: a household's devices) return False seen = self._ua_seen.get(ip) if not seen: return False if t is not None: return sum(1 for v in seen.values() if t - v <= 900.0) >= int(self.policy.d.get("nat_min_agents", 6)) return len(seen) >= int(self.policy.d.get("nat_min_agents", 6)) def _entity_keys(self, texts: list[str], client, t) -> list: """the key each request is judged under: the address, or address#agent when the address is shared""" if client is None: return None tt = np.zeros(len(texts)) if t is None else np.asarray(t, np.float64) keys = [] for i, x in enumerate(texts): ip = self.ip_entity(client[i]) ua = "" for part in x.split(" | "): if part.startswith("h:user-agent="): ua = part[13:] break h = zlib.crc32(ua.encode("utf-8", "replace")) % 4093 seen = self._ua_seen.get(ip) if seen is None: seen = self._ua_seen[ip] = OrderedDict() if len(self._ua_seen) > 200_000: self._ua_seen.popitem(last=False) seen[h] = float(tt[i]) seen.move_to_end(h) if len(seen) > 64: seen.popitem(last=False) # (an address whose recent traffic is mostly attack or failing is judged whole: rotating user-agents must # not split one stuffer into eight streams each too quiet to judge) keys.append(f"{ip}#{h}" if self.shared(ip, float(tt[i])) and not self.mostly_attack(ip, float(tt[i])) else ip) return keys # -- verdicts -> actions def _calibrated(self, kind: str) -> bool: """no enforcement before the site's own reference exists: an uncalibrated verdict is reported, never acted on""" g = self.guard if kind in ("http", "rpc"): return g.http is None or g.site.n("http") >= 200 # without a request model there is nothing to calibrate return g.ev_ref.n() > 0 def _act(self, out: dict, src, t, kind: str, skip: set | None = None) -> list[dict]: taken = [] skip = skip or set() alarm = np.asarray(out["alarm"]) block_ok = np.asarray(out.get("block_ok", alarm)) # soft newborn alarms are watched / challenged, not blocked if not alarm.any(): return taken hard = [bool(h) for h in out.get("playbook", [])] if not self._calibrated(kind) and not any(hard): return [{"ip": str(src[i]), "action": "monitor", "family": str(out["family"][i]), "score": int(out["score"][i]), "reason": f"{kind}: uncalibrated - reported only", "taken": False} for i in np.flatnonzero(alarm)[:50] if str(src[i]) not in skip] # a rule's decision stands if not self._calibrated(kind): # only hard signatures act before calibration alarm = alarm & np.array(hard + [False] * (len(alarm) - len(hard))) fam, score = np.asarray(out["family"], dtype=object), np.asarray(out["score"]).astype(np.int64) risk = out.get("entity_risk") if risk is not None: # a source's accumulated risk counts as much as its worst single event score = np.maximum(score, (100 * np.nan_to_num(np.asarray(risk, np.float64))).astype(np.int64)) beh = out.get("behaviour") if beh is not None: # a sustained behaviour judged at alpha is actionable on its own: it is score = np.where(np.asarray(beh), np.maximum(score, int(self.policy.d["min_action_score"])), score) tt = np.zeros(len(alarm)) if t is None else np.asarray(t, np.float64) best: dict = {} # one decision per source per batch: its strongest score and the fams: dict = {} # family its alarms name most often (a named family beats 'generic') fams_ok: dict = {} # the same over its block-safe alarms only for i in np.flatnonzero(alarm): s = str(src[i]) if s not in best or score[i] > best[s][0]: best[s] = (int(score[i]), float(tt[i])) fams.setdefault(s, {}) fams[s][fam[i]] = fams[s].get(fam[i], 0) + 1 if block_ok[i]: fams_ok.setdefault(s, {}) fams_ok[s][fam[i]] = fams_ok[s].get(fam[i], 0) + 1 confident = {str(src[i]) for i in np.flatnonzero(alarm & block_ok)} for s, (sc, ti) in best.items(): ip = s.split("#", 1)[0] # a sub-entity of a shared address acts on the address if s in skip or ip in skip: # a rule already decided this source this batch continue if self.feedback.is_allowed(ip, ti): # operator said this source is always fine continue # an enforced action takes its name from the alarms that make it enforceable: one event's confidence and # another's name (a benign client's generic alarm + the 'dos' its flooded server gave a later flow) is no verdict pool = fams_ok[s] if s in confident else fams[s] named = {k: v for k, v in pool.items() if k not in ("attack_generic", "benign")} f = max(named or pool, key=lambda k: (named or pool)[k]) action = self.policy.action_for(f, sc, ip) if s not in confident and action in ("block", "rate_limit"): # not enough to act on: watch (or challenge the browser) action = "challenge" if kind in ("http", "rpc") else "monitor" why = f"{kind}: {f} ({playbook.name_family(f) or 'learned'})" ua = self._ua_by.get(s, "") if kind == "http" else "" if ua and self.policy.verified_crawler(ip, ua): action, why = "monitor", why + " - verified crawler: watched, never acted on as an address" elif ua and action == "challenge" and s not in confident and _machine_ua(ua): action = self.policy.d.get("machine_challenge", "rate_limit") why += " - not a browser: a JavaScript challenge would cut it off" if (action in ("block", "rate_limit") and self.shared(ip, ti) and not self.policy.d.get("nat_block", False) and not (kind in ("http", "rpc") and self.mostly_attack(ip, ti))): # hundreds of people may stand behind this address: the attacker's own requests are refused inline # (signatures / challenge), the address as a whole is challenged - a per-browser cookie - not blocked action = "challenge" if kind in ("http", "rpc") else "monitor" why += " - shared address (NAT / CGNAT): challenged, not blocked" ev = self.enforcer.apply(ip, action, f, sc, reason=why, now=ti) taken.append(ev) return taken def _apply_rules(self, ctxs: list[dict], src, t) -> list[dict]: """operator rules run first, with no model and from the first second; an 'allow' rule shields the source""" taken, acted = [], {} tt = np.zeros(len(ctxs)) if t is None else np.asarray(t, np.float64) for i, ctx in enumerate(ctxs): r = self.rules.decide(ctx) if r is None or str(src[i]) in acted: continue acted[str(src[i])] = True if r.action == "allow": continue ev = self.enforcer.apply(str(src[i]), r.action, r.family, 100, reason=f"rule {r.id}: {r.note}", now=float(tt[i])) ev["rule"] = r.id taken.append(ev) return taken def flows(self, flows: dict) -> dict: if any(":" in str(x) for x in flows["src"]): # IPv6 sources are judged per /64 (see ip_entity) flows = dict(flows) flows["src"] = np.array([self.ip_entity(x) for x in flows["src"]], dtype=object) out = self.guard.flows(flows) src = np.asarray(flows["src"]) rule_acts = [] if self.rules.rules: import numpy as _np dport = _np.asarray(flows.get("dport", _np.full(len(src), -1))) proto = _np.asarray(flows.get("proto", _np.full(len(src), 6))) pmap = {6: "tcp", 17: "udp", 1: "icmp"} # the source's behaviour as rules speak of it (from the engine's entity counters, ~2-minute horizon) xe = out.get("entity_x") F = _ent._F beh = (lambda i: {}) if xe is None else (lambda i: { "flows_per_min": float(xe[i, F["src_flows_h2"]]) / 2.0, "new_ports": float(xe[i, F["src_newport_h2"]] * xe[i, F["src_flows_h2"]]) / 2.0, "noreply": float(xe[i, F["src_noreply_h2"]]), "repeat": float(xe[i, F["src_repeat_h2"]])}) ctxs = [{"scope": "flow", "src": str(src[i]), "port": int(dport[i]) if i < len(dport) else -1, "proto": pmap.get(int(proto[i]) if i < len(proto) else 0, "other"), "family": str(out["family"][i]), **beh(i)} for i in range(len(src))] rule_acts = self._apply_rules(ctxs, src, flows.get("t")) skip = {a["ip"] for a in rule_acts} out["actions"] = rule_acts + self._act(out, src, flows.get("t"), "flow", skip) return out def _http_profile(self, texts: list[str], hard: list[bool], t, host=None) -> None: """each site's HTTP reference: bootstrapped from its own first requests (hard-signature hits left out) and rebuilt every epoch from the requests that raised no alarm - per Host when several sites share the layer""" g = self.guard now = float(np.max(t)) if t is not None and len(t) else 0.0 hosts = [None] * len(texts) if host is None else [str(h) for h in host] for h in set(hosts): idx = [i for i in range(len(texts)) if hosts[i] == h and not hard[i]] if not idx: continue key = h or "" have = (h in g.site_http_by_host) if h is not None else g.site.n("http") >= 200 if not have: # the Host header is the client's to choose: a bounded number of sites get their own profile (the # rest are judged with the default one), and none learns from requests the default profile flags if key not in self._http_boot_by and len(self._http_boot_by) + len(g.site_http_by_host) >= self.http_max_hosts: continue buf = self._http_boot_by.setdefault(key, []) buf += [texts[i] for i in idx] if len(buf) >= self.http_boot_n: boot = buf[-self.http_boot_n * 4:] if h is not None and g.http is not None and g.site_http is not None and g.site.n("http") >= 200: z0 = g.http.logits(boot, [0] * len(boot), g.site_http)[:, 0] ok = z0 <= g.threshold("http")[0] if ok.mean() < 0.5: # mostly what the default profile calls attack: no profile of its own self._http_boot_by[key] = [] continue boot = [x for x, k in zip(boot, ok) if k] g.calibrate_http(boot, host=h) self._http_boot_by.pop(key, None) self._http_last_by[key] = now elif now - self._http_last_by.get(key, 0.0) >= self.http_reprofile_s and len(self._http_recent_by.get(key, [])) >= self.http_boot_n: if h is None: g.site.cal.pop("http", None) g.calibrate_http(self._http_recent_by[key][-self.http_boot_n * 4:], host=h) self._http_recent_by[key] = [] self._http_last_by[key] = now def _sql_context(self, texts: list[str], ids: list, host) -> list: """SQL-context analysis of every parameter (http/sqlctx.py) for requests no signature named: a value that breaks out of a query context into an attack's shape is a hard hit (A22); one only DbNet finds is soft, and a parameter whose DbNet-only hits come from many different clients with never a signature is this site's own syntax (a filter language, a template) and is learned as such""" net = self.dbg.net soft = [None] * len(texts) hosts = [""] * len(texts) if host is None else [str(h) for h in host] for i, x in enumerate(texts): if ids[i]: continue hits = sqlctx.analyse(x, net) if not hits: continue hard = [h for h in hits if not h["learned_only"]] if hard: ids[i] = ["A22"] self._sql_evidence[i] = hard[0] continue h0 = hits[0] key = (hosts[i], h0["field"]) if key in self._sql_site_syntax: continue soft[i] = h0 return soft def http(self, texts: list[str], deepest=None, client=None, host=None, status=None, t=None) -> dict: ids = [playbook.match_http(x) for x in texts] self._sql_evidence = {} soft_sql = self._sql_context(texts, ids, host) self._http_profile(texts, [bool(h) for h in ids], t, host) if client is not None: tt0 = np.zeros(len(texts)) if t is None else np.asarray(t, np.float64) for i, x in enumerate(texts): self._http_ring.append((float(tt0[i]), str(client[i]), unquote_plus(unquote_plus(x)).lower())) ekeys = self._entity_keys(texts, client, t) # shared addresses: one entity per user behind them out = self.guard.http_requests(texts, deepest, ekeys, host, status, t) hosts = [""] * len(texts) if host is None else [str(h) for h in host] for i, (x, a, hh) in enumerate(zip(texts, out["alarm"], ids)): if not a and not hh: lst = self._http_recent_by.setdefault(hosts[i], []) lst.append(x) if len(lst) > self.http_boot_n * 4: del lst[: len(lst) - self.http_boot_n * 4] # request-level playbook signatures (no statistics needed): named, forced to their family, and always in the # alarm log / review queue (also when the model itself did not alarm) tt = np.zeros(len(texts)) if t is None else np.asarray(t, np.float64) for i, hit in enumerate(ids): if hit: pb = playbook.BY_ID[hit[0]] if client is not None: ev = self._sql_evidence.get(i) why = f"{pb['id']} {pb['name']}" + (f": {ev['field']} breaks out of a {ev['context']} SQL context: {ev['value'][:160]}" if ev else "") named = False if out["alarm"][i]: # the model alarmed too: name its entry for e in reversed(self.guard.log[-4 * len(texts) - 8:]): if e["src"] in (str(client[i]), str(ekeys[i])) and e["t"] == float(tt[i]) and e["kind"].startswith("http"): e["evidence"] = why + " | " + str(e.get("evidence", "")) e["family"] = pb["family"] named = True break if not named: self.guard._log("http", float(tt[i]), str(client[i]), hosts[i], 0, pb["family"], 90, 0, evidence=why + " | " + texts[i][:300]) out["alarm"][i] = True out["family"][i] = pb["family"] out["score"][i] = max(int(out["score"][i]), 90) for h in hit: self.playbook_hits[h] = self.playbook_hits.get(h, 0) + 1 out["playbook"] = ids bok = np.asarray(out.get("block_ok", out["alarm"])).copy() for i, hit in enumerate(ids): if hit: bok[i] = True # a hard payload signature (JNDI, CMS exploit) blocks outright # SQL-context hits only the learned tier sees: an alarm and a challenge, never a block on their own sql_soft = {} for i, h in enumerate(soft_sql): if h is None: continue if client is not None: key = (hosts[i], h["field"]) seen = self._sql_soft_src.setdefault(key, set()) seen.add(str(client[i])) if len(seen) >= 5: # many unrelated clients send it: the site's own parameter syntax self._sql_site_syntax.add(key) continue if not out["alarm"][i]: self.guard._log("http", float(tt[i]), str(client[i]), hosts[i], 0, "sqli", 70, 0, soft=True, evidence=f"DbNet: {h['field']} reads as SQL injection in a {h['context']} context: {h['value'][:160]}") out["alarm"][i] = True out["family"][i] = "sqli" out["score"][i] = max(int(out["score"][i]), 70) sql_soft[i] = h out["sql_context"] = {i: self._sql_evidence[i] for i in self._sql_evidence} | sql_soft out["block_ok"] = bok if any(ids) and client is not None: # hard-signature hits always go to the review queue hit_src = {str(ekeys[i]) for i, h in enumerate(ids) if h} | {str(client[i]) for i, h in enumerate(ids) if h} tmin = float(np.min(t)) if t is not None and len(t) else -np.inf for e in reversed(self.guard.log): if e["t"] < tmin: break if e["src"] in hit_src: e["soft"] = False rule_acts = [] if self.rules.rules and client is not None: def _ua(txt): for part in txt.split(" | "): if part.startswith("h:user-agent="): return part[len("h:user-agent="):] return "" ctxs = [{"scope": "http", "src": str(client[i]), "path": texts[i].split(" | ")[0].split(" ", 1)[-1], "ua": _ua(texts[i]), "method": texts[i].split(" ", 1)[0], "status": int(status[i]) if status is not None else 0, "family": str(out["family"][i]), "flows_per_min": self._req_rate.add(str(client[i]), float(tt[i]), 1.0)} for i in range(len(texts))] rule_acts = self._apply_rules(ctxs, client, t) skip = {a["ip"] for a in rule_acts} if client is not None: self._ua_by = {str(k): _ua_of(texts[i]) for i, k in enumerate(ekeys)} for i, k in enumerate(ekeys): ip = str(k).split("#", 1)[0] self._addr_req.add(ip, float(tt[i]), 1.0) failed = status is not None and int(status[i] or 0) >= 400 self._addr_bad.add(ip, float(tt[i]), float(bool(out["alarm"][i]) or failed)) out["actions"] = rule_acts + (self._act(out, ekeys, t, "http", skip) if client is not None else []) return out def rpc(self, bodies: list, client: list[str], t=None, node: str = "node", errors=None) -> dict: """JSON-RPC calls (chain node): scored as requests and fed to the entity stream as events""" texts, deepest, hits, methods = [], [], [], [] for b in bodies: tx, d, h = adapters.rpc_text(b) texts.append(tx) deepest.append(d) hits.append(h) methods.append(tx.split("m:")[1].split(" |")[0] if "m:" in tx else "") tt = np.zeros(len(bodies)) if t is None else np.asarray(t, np.float64) err = [False] * len(bodies) if errors is None else list(errors) ev = [adapters.rpc_event(m, c, node, float(ti), len(b) if isinstance(b, (str, bytes)) else 0, e) for m, c, ti, b, e in zip(methods, client, tt, bodies, err)] flows = {k: np.array([e[k] for e in ev]) for k in ev[0]} if ev else None out = {"alarm": np.zeros(len(bodies), bool), "family": np.full(len(bodies), "benign", dtype=object), "score": np.zeros(len(bodies), np.int16)} if self.guard.http is not None and texts: o = self.guard.http_requests(texts, deepest, client, [node] * len(texts), None, tt) out.update({k: o[k] for k in ("alarm", "family", "score", "p_attack", "behaviour")}) if flows is not None: fo = self.guard.flows(flows) out["alarm"] = out["alarm"] | fo["alarm"] out["family"] = np.where(fo["alarm"], fo["family"], out["family"]) out["score"] = np.maximum(out["score"], fo["score"]) for i, h in enumerate(hits): for x in h: self.playbook_hits[x] = self.playbook_hits.get(x, 0) + 1 if h and out["alarm"][i]: out["family"][i] = playbook.BY_ID[h[0]]["family"] out["score"][i] = max(int(out["score"][i]), 80) # a named chain-abuse pattern at volume out["playbook"] = hits out["actions"] = self._act(out, client, tt, "rpc") return out # -- databases def calibrate_db(self, events: list[dict]) -> dict: """the applications' normal workload on their databases (query / audit log events)""" return self.dbg.calibrate(events) _TOK = re.compile(r"[a-z_@$][a-z0-9_@$.]{2,}|'|--|#|;|\(\s*select") _ATTACK_WORDS = {"union", "select", "sleep", "pg_sleep", "benchmark", "waitfor", "information_schema", "extractvalue", "updatexml", "from", "or", "and", "concat", "char", "substring", "substr", "ascii", "version", "database", "user", "$ne", "$gt", "$where", "$regex", "drop", "insert", "update", "delete", "exec", "xp_cmdshell", "load_file", "outfile", "pg_shadow", "mysql.user", "cast", "convert"} def _provenance(self, v: dict, query: str) -> list[tuple]: """which recent web requests carried this database query's payload: the literals and attack words of the query that also appear in a request (decoded) within the correlation window - the attacker behind the app server""" w = float(self.policy.d["database"]["correlate_window_s"]) q = query.lower() lits = {m for m in re.findall(r"'([^']{3,60})'|\"([^\"]{3,60})\"", q) for m in m if m} words = {x for x in self._TOK.findall(q) if x in self._ATTACK_WORDS or x in ("'", "--", "#", ";") or x.startswith("(")} best: dict = {} for (ti, src, txt) in reversed(self._http_ring): if ti < v["t"] - w: break if ti > v["t"] + 2: continue toks = set(self._TOK.findall(txt)) kw = len(words & toks) # whole tokens: 'or' must not match inside 'order' lit = sum(1 for x in lits if x in txt) sc = kw + 2 * lit if (kw >= 3 or (kw >= 1 and lit >= 2)) and sc > best.get(src, (0,))[0]: best[src] = (sc, ti) return sorted(((s, sc, ti) for s, (sc, ti) in best.items()), key=lambda x: -x[1])[:3] def db(self, events: list[dict]) -> dict: """database query / audit events (adapters in shield/database.py) -> verdicts + actions. An app server the database knows is not blocked at the network level (see policy.database); the web source whose request carried the payload is acted on instead, and the database-side remediation is attached to the alarm.""" out = self.dbg.score(events) pol = self.policy.d["database"] by_t = {} for e in events: by_t.setdefault((e["t"], e["client"]), e.get("query") or "") rank = {"monitor": 0, "challenge": 1, "rate_limit": 2, "block": 3} taken, decision, web = [], {}, {} for v in out: if not v["alarm"]: continue fam = "db_" + (v["families"][0] if v["families"] else "new_client") q = by_t.get((v["t"], v["client"]), "") self.guard._log("db", v["t"], v["client"], v["db"], 0, fam, v["score"], 3, evidence=" || ".join(v["evidence"])[:600], soft=False) self.guard.log[-1].update(query=q[:2000], engine=v["engine"], app_client=v["app_client"], remediation=v["remediation"]) if self.feedback.is_allowed(v["client"], v["t"]): continue action = self.policy.action_for(fam, v["score"], v["client"]) if v["learned_only"] and not (v["public"] and "new_client" in v["families"]) and action == "block": action = "monitor" # the model alone reports; a signature or behaviour acts reason = f"db {v['engine']}/{v['db']}: {', '.join(v['families'])}" if v["app_client"] and action in ("block", "rate_limit") and not pol["block_app_clients"]: action = "monitor" reason += " - application server, not blocked (it would take the app down); run the remediation on the database" for src, sc, ti in (self._provenance(v, q) if q else []): hf = "sqli" if v["families"][0] == "injection" else "webattack" if sc > web.get(src, (0,))[0]: web[src] = (sc, ti, hf, v, fam) # one decision per client per batch: its strongest verdict (a weak first alarm must not pre-empt a block) cur = decision.get(v["client"]) if cur is None or (rank[action], v["score"]) > (rank[cur[0]], cur[1]["score"]): decision[v["client"]] = (action, v, fam, reason) for src, (sc, ti, hf, v, fam) in web.items(): if src in decision: continue ev = self.enforcer.apply(src, self.policy.action_for(hf, 90, src), hf, 90, reason=f"payload provenance: its request reached {v['db']} through app {v['client']} as {fam}", now=ti) ev["provenance"] = {"db": v["db"], "app": v["client"], "match": sc} taken.append(ev) for client, (action, v, fam, reason) in decision.items(): ev = self.enforcer.apply(client, action, fam, v["score"], reason=reason, now=v["t"]) ev.update(remediation=v["remediation"], app_client=v["app_client"]) taken.append(ev) return {"verdicts": out, "actions": taken} # -- housekeeping def posture(self) -> dict: self.position = position.detect(self.position.get("inputs")) self.posture_findings = posture.audit(self.position, self.policy) done = posture.remediate(self.posture_findings, self.policy, self.enforcer) return {"roles": self.position["roles"], "firewall": self.position["firewall"], "findings": done} def tick(self, now: float | None = None) -> dict: now = time.time() if now is None else now return {"lifted": self.enforcer.expire(now), "active": len(self.enforcer.active)} def fast_state(self) -> dict: return self.enforcer.state() def report(self, since=None, until=None, top: int = 10) -> dict: r = rep_mod.build(self.guard, since, until, top) st = self.enforcer.state() r["enforcement"] = {"mode": self.policy.d["mode"], "backend": self.enforcer.backend.name, "blocked": len(st["block"]), "challenged": len(st["challenge"]), "rate_limited": len(st["rate_limit"]), "recent": self.enforcer.audit[-20:]} r["position"] = {"roles": self.position["roles"], "capabilities": self.position["capabilities"], "firewall": self.position["firewall"]} r["posture"] = self.posture_findings[:10] r["playbook"] = {k: {"count": v, "name": playbook.BY_ID[k]["name"], "action": playbook.BY_ID[k]["action"]} for k, v in self.playbook_hits.items()} for e in r["entities"]: e["playbook"] = playbook.name_family(e["families"][0]) if e["families"] else "" e["enforced"] = self.enforcer.lookup(e["entity"]) return r