Blame for runtime/python/dibaguard/shield/layer.py

"""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