| """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 |
| self.posture_findings: list = [] |
| self.playbook_hits: dict = {} |
| 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)) |
| self._http_last = 0.0 |
| self._http_boot_by: dict = {} |
| self._http_recent_by: dict = {} |
| 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) |
| self._req_rate = _Decay(60.0) |
| self._ua_seen: "OrderedDict[str, OrderedDict]" = OrderedDict() |
| |
| |
| self._addr_req, self._addr_bad = _Decay(600.0), _Decay(600.0) |
| self._ua_by: dict = {} |
| self._sql_evidence: dict = {} |
| self._sql_soft_src: dict = {} |
| self._sql_site_syntax: set = set() |
|
|
| @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} |
|
|
| |
| 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) |
|
|
| |
| _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: |
| 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) |
| |
| |
| 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 |
|
|
| |
| 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 |
| 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)) |
| 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] |
| if not self._calibrated(kind): |
| 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: |
| 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: |
| 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 = {} |
| fams: dict = {} |
| fams_ok: dict = {} |
| 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] |
| if s in skip or ip in skip: |
| continue |
| if self.feedback.is_allowed(ip, ti): |
| continue |
| |
| |
| 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"): |
| 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))): |
| |
| |
| 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"]): |
| 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"} |
| |
| 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: |
| |
| |
| 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: |
| 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) |
| 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] |
| |
| |
| 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]: |
| 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 |
| |
| 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: |
| 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: |
| 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) |
| out["playbook"] = hits |
| out["actions"] = self._act(out, client, tt, "rpc") |
| return out |
|
|
| |
| 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) |
| 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" |
| 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) |
| |
| 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} |
|
|
| |
| 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 |
|
|
