DibaGuard open release: licence-clean network models, runtime, docs (no HTTP model)
e02228b verified 4 days ago
75.2 kB
"""Guard: the one engine - flows and HTTP requests in, calibrated verdicts, entity risk and incident reports out.
Cascade per flow (time order is the only requirement):
  tier 0  EntityState: per-source / per-destination decayed counters (ns)            every flow
  tier 1  entity detector (oblivious trees on the stream features, ~2 us)             every flow
  tier 2  SeqNet: causal GRU over the source's recent flows, site-conditioned         warm sources only
A source is *warm* while its tier-1 logit was above the low-water mark recently; the GRU state of a source that
just warmed up starts from zero - which is exactly how the network was trained (fixed windows from a zero state),
so the cascade loses nothing a training window would have had. A cold source costs tiers 0 + 1 only.
HTTP: HttpModel2 (bag + MLP conditioned on this site's benign profile) per request, and the request also enters
tier 0 as a flow-like event (client -> host, 'port' = path hash, 'no reply' = 4xx/5xx), so enumeration, brute
force and scanners show up in the entity stream even when every single request looks harmless.
Decisions: each tier has its own conformal threshold at alpha / tiers from the site's own benign scores (shipped
thresholds until the site has enough), temperature-calibrated probabilities, severity from the family head, the
0-100 event score, decaying per-entity risk, site pressure, and a drift martingale on tier-1 p-values.
"""
from __future__ import annotations

import math
import os
import zlib
from collections import OrderedDict

import numpy as np

from dibaguard.campaign import AUTH_PORTS, CampaignTracker, http_fingerprint
from dibaguard.decide import _tol, ConformalFPR, DriftMartingale, EntityRisk, GlobalRate, RollingReference, SitePressure, SourceEvidence, event_score
from dibaguard.families import severity
from dibaguard.flows import entity as ent
from dibaguard.flows import seqnet
from dibaguard.flows.trees import Oblivious
from dibaguard.http.model import HttpModel2

ANOM_IDX = [ent.feature_names().index(n) for n in ent.ANOMALY_SITE]
DST_FEATS = ["dst_flows_h1", "dst_noreply_h1", "dst_newsrc_h1", "dst_flows_h2"]
DST_IDX = [ent.feature_names().index(n) for n in DST_FEATS]
DST_NEWSRC = ent.feature_names().index("dst_newsrc_h1")
SRC_RATE10 = ent.feature_names().index("src_flows_h1")                    # the source's own flows within ~10 s
# services where a steady stream of short repeated sessions is password guessing, never a busy client (directory and
# file services - SMB, LDAP, Kerberos - are chatty by nature and are left to the campaign / evidence tiers)
NEVER_EXCUSED = [21, 22, 23, 25, 110, 143, 465, 587, 993, 995, 1433, 1521, 3306, 3389, 5432, 5900, 6379, 9200, 27017]
SRC_NOREPLY2, SRC_SHORT2, SRC_RST2, SRC_SYN2 = (ent.feature_names().index(f"src_{k}_h2") for k in ("noreply", "short", "rst", "synonly"))
SRC_FLOWS3, SRC_REPEAT3 = (ent.feature_names().index(f"src_{k}_h3") for k in ("flows", "repeat"))
SRC_NEWPORT3, SRC_NOREPLY3, SRC_SYN3 = (ent.feature_names().index(f"src_{k}_h3") for k in ("newport", "noreply", "synonly"))
# interactive logins: no legitimate client opens short sessions to them many times a minute for many minutes (databases,
# mail and caches are left out - an application without a connection pool does exactly that)
INTERACTIVE_LOGIN = [21, 22, 23, 3389, 5900]
AGE_EDGES = (1.5, 5.5, 20.5, 100.5, 500.5)   # flows ever seen from the source: 1 | 2-5 | 6-20 | 21-100 | 101-500 | 500+


def _member(values, members) -> np.ndarray:
    """np.isin for keys (addresses, ids): one hash lookup per value. np.isin on object / string arrays falls back to
    an O(n x m) loop - measured 21 s for 400k buffer rows against a 5k-source set, 17.5 min inside one re-calibration"""
    s = members if isinstance(members, (set, frozenset)) else set(members)
    return np.fromiter((v in s for v in np.asarray(values, dtype=object).tolist()), bool, len(values))


def clip_upper(z, frac: float = 0.01) -> np.ndarray:
    """a reference's scores with its far upper outliers clipped to a fence. With n < 1/alpha the conformal threshold
    is the reference's maximum, so one attack request inside a bootstrap would *be* the threshold. Clipped: scores
    beyond q99 + 3 (q99 - q50) (at most `frac` of the rows, else the top `frac`), so an attacker can lift the
    threshold at most to a fence the benign bulk sets; the reference keeps its size (alpha stays reachable)"""
    z = np.asarray(z, np.float64)
    if len(z) < 50:
        return z
    q50, q99 = np.quantile(z, [0.5, 0.99])
    fence = q99 + 3.0 * max(q99 - q50, 1e-6)
    if (z > fence).sum() > frac * len(z):
        fence = float(np.quantile(z, 1.0 - frac))
    return np.minimum(z, fence)


def age_bucket(n_flows) -> np.ndarray:
    """true source age (flows ever seen, from EntityState) -> bucket; a client's first flows are compared only with
    other clients' first flows, a client with 500+ flows with other established clients"""
    return np.searchsorted(np.asarray(AGE_EDGES), np.asarray(n_flows, np.float64), side="right").astype(np.int8)
BOT_UA = ("gptbot", "bytespider", "ccbot", "claudebot", "anthropic-ai", "google-extended", "python-requests", "curl/", "wget/",
          "scrapy", "go-http-client", "java/", "libwww", "httpclient", "okhttp", "masscan", "zgrab", "nikto", "sqlmap", "nmap")
FLOW_FIELDS = ("t", "src", "dst", "dport", "proto", "bytes_in", "pkts_in", "bytes_out", "pkts_out", "client_flags",
               "server_flags", "flags", "duration_ms")
_NF_OF = {"dport": "L4_DST_PORT", "proto": "PROTOCOL", "bytes_in": "IN_BYTES", "pkts_in": "IN_PKTS", "bytes_out": "OUT_BYTES",
          "pkts_out": "OUT_PKTS", "client_flags": "CLIENT_TCP_FLAGS", "server_flags": "SERVER_TCP_FLAGS", "flags": "TCP_FLAGS",
          "duration_ms": "FLOW_DURATION_MILLISECONDS"}


def _col(flows, name, n, default=0.0):
    v = flows.get(name) if isinstance(flows, dict) else (flows[name] if name in flows else None)
    return np.full(n, default) if v is None else np.asarray(v)


class Guard:
    def __init__(self, tier1: Oblivious, seq: seqnet.SeqNet | None = None, http: HttpModel2 | None = None,
                 alpha: float = 1e-3, warm_logit: float | None = None, warm_flows: int = 64,
                 entity_half_life_s: float = 600.0, pressure_window_s: float = 300.0, max_entities: int = 500_000,
                 rolling_s: float = 1800.0, anomaly: bool = True, tau_s: float = 120.0, mondrian_min: int = 2000,
                 warmup_s: float = 3600.0, recal_s: float = 1800.0, recal_window_s: float = 7200.0, recal_max_rows: int = 400_000):
        self.tier1, self.seq, self.http, self.alpha = tier1, seq, http, alpha
        names = ent.feature_names()
        self.t1_idx = [names.index(n) for n in tier1.names] if tier1.names and set(tier1.names) <= set(names) else list(range(len(names)))
        self.state = ent.EntityState(max_entities=max_entities)
        self.temp1 = float(tier1.meta.get("temperature", 1.0))
        self.temp2 = float(seq.info.get("temperature", 1.0)) if seq else 1.0
        self.temph = float(http.info.get("temperature", 1.0)) if http else 1.0
        self.shipped = {k: {float(a): float(v) for a, v in m.items()} for k, m in (
            ("tier1", tier1.meta.get("thresholds", {})), ("seq", seq.info.get("thresholds", {}) if seq else {}),
            ("http", http.info.get("thresholds", {}) if http else {}))}
        self.site = ConformalFPR(max_per_group=200_000, min_n=1)
        # rolling self-cleaning references: previous chunks of non-alarmed traffic (scores per tier, entity features for
        # the site-only anomaly tier) - what the site looks like *now*, no labels needed
        self.rolling = {k: RollingReference(rolling_s) for k in ("tier1", "seq", "http", "sess")}
        self.anom = RollingReference(rolling_s) if anomaly else None       # entity features of normal traffic
        self.anom_s = RollingReference(rolling_s) if anomaly else None     # the anomaly score itself on normal traffic
        self.anom_age = [RollingReference(rolling_s) for _ in range(len(AGE_EDGES) + 1)] if anomaly else None   # per source age
        self.seq_age = [RollingReference(rolling_s) for _ in range(len(AGE_EDGES) + 1)]                          # per source age
        # real-world verdict unit: per-source accumulated evidence, judged per source-minute (see SourceEvidence)
        self.evidence = SourceEvidence(tau_s, max_entities)
        self.ev_ref = RollingReference(rolling_s)                            # benign source-minute maxima of E
        self.evidence_slow = SourceEvidence(1800.0, max_entities)            # long horizon: slow scans over ~30 min
        self.ev_slow_ref = RollingReference(rolling_s)
        self.evid_drift = 0.0                                                # benign mean of the per-flow increment (CUSUM baseline)
        # destination side: distributed / volumetric floods - no single source stands out, but one destination is
        # overwhelmed by many (often new) sources. A per-destination anomaly, accumulated per destination-minute.
        self.dst_anom = RollingReference(rolling_s)                          # benign dst-feature matrix (for the Fisher score)
        self.dst_anom_s = RollingReference(rolling_s)                        # benign dst Fisher scores (for p-values)
        self.dst_evidence = SourceEvidence(tau_s, max_entities)              # keyed by destination
        self.dst_drift = 0.0                                                 # CUSUM allowance for destination increments
        self.dst_rate_hi = np.inf                                            # benign per-destination rate, upper tail (floods are volumetric)
        self.dst_ev_ref = RollingReference(rolling_s)                        # benign destination-minute maxima
        self._anchor = None                                                 # first calibration, kept immutable
        self.drift_events: list = []                                        # drift guard verdicts per epoch
        self._alarm_src: "OrderedDict[str, list]" = OrderedDict()            # decayed alarm count per source (reference hygiene)
        self._contrib: "OrderedDict[str, list]" = OrderedDict()              # decayed reference contributions per source
        self.global_http = GlobalRate(120.0)                                # site-wide failure rate per HTTP path (distributed campaigns)
        self.global_floor = 10.0                                            # decayed failures to one path within ~tau -> a campaign
        self.global_scan = GlobalRate(120.0)                                # site-wide SYN-only / no-reply flow rate (distributed scans)
        self.global_scan_floor = 60.0
        self.campaign = CampaignTracker()                                   # coordinated low-and-slow campaigns, per fingerprint
        self.campaign_http = CampaignTracker()
        self.mondrian_min = mondrian_min
        self.warmup_s = warmup_s           # the counters need their longest horizon to reach steady state before they are a reference
        self._minute_max: dict = {}
        self.recal_s, self.recal_window_s, self.recal_max_rows = recal_s, recal_window_s, recal_max_rows
        self._buffer = {"t": np.zeros(0), "src": np.zeros(0, dtype=object), "dst": np.zeros(0, dtype=object), "z1": np.zeros(0),
                        "z2": np.zeros(0), "svc": np.zeros(0, np.int32), "xa": np.zeros((0, len(ANOM_IDX)), np.float32),
                        "xd": np.zeros((0, len(DST_IDX)), np.float32), "age": np.zeros(0, np.int8)}
        self.last_recal = 0.0
        self.roll_groups: dict = {}         # Mondrian (service x age) rolling references of tier-1 scores
        self.entities = EntityRisk(half_life_s=entity_half_life_s, max_entities=max_entities)
        self.pressure = SitePressure(window_s=pressure_window_s)
        self.drift = DriftMartingale()
        # src -> (state [layers, d], flows left warm), least recently warmed first. Bounded: a source that warms once
        # and never returns (a one-shot scanner among hundreds of thousands) is evicted, not kept forever
        self.gru: "OrderedDict[object, tuple]" = OrderedDict()
        self.gru_max = max(10_000, max_entities // 5)
        self.warm_flows = warm_flows
        self.warm_logit = warm_logit if warm_logit is not None else self.shipped["tier1"].get(1e-2, 0.0)
        self.site_http = http.site if http else None
        self.site_http_by_host: dict = {}            # many sites behind one layer: one HTTP profile + reference per Host
        self.http_ref_by_host: dict = {}
        self.gamma, self.beta = seq.context(seq.w.get("ctx_pool_default")) if seq else (None, None)
        self.log: list = []                       # recent alarms for the report
        self.counts = {"flows": 0, "http": 0, "alarms": 0, "seq_runs": 0}

    @classmethod
    def load(cls, model_dir: str, **kw) -> "Guard":
        mono = os.path.join(model_dir, "seq_v1", "nf_entity_mono.npz")      # monotone prior on attack-direction features
        t1 = Oblivious.load(mono if os.path.exists(mono) else os.path.join(model_dir, "entity_v1", "nf_entity_detector.npz"))
        sp = os.path.join(model_dir, "seq_v1", "seqnet.npz")
        hp = os.path.join(model_dir, "http_v2", "http_tier1_v2.npz")
        return cls(t1, seqnet.SeqNet.load(sp) if os.path.exists(sp) else None, HttpModel2.load(hp) if os.path.exists(hp) else None, **kw)

    # ------------------------------------------------------------ thresholds
    def n_tiers(self) -> int:
        return 1 + (self.seq is not None) + (self.anom is not None)

    def threshold(self, tier: str) -> tuple[float, str]:
        a = self.alpha / self.n_tiers() if tier != "http" else self.alpha
        need = ConformalFPR.n_needed(a, 0.5)
        if tier in self.rolling and self.rolling[tier].n() >= need:
            return self.rolling[tier].threshold(a), "site+rolling"
        if self.site.n(tier) >= need:
            return self.site.threshold(tier, a, fallback=None), "site"
        sh = self.shipped[tier]
        if a in sh:
            return sh[a], "shipped"
        return float(min(sh.values(), default=0.0)), "shipped-nearest"

    # ------------------------------------------------------------ flows
    def _features(self, flows: dict, n: int):
        t = _col(flows, "t", n).astype(np.float64)
        src, dst = _col(flows, "src", n, "?"), _col(flows, "dst", n, "?")
        cols = {"gap_s": None}
        nf = {_NF_OF[k]: _col(flows, k, n, 0.0).astype(np.float64) for k in _NF_OF}
        syn_only = ent.tcp_syn_only(nf["CLIENT_TCP_FLAGS"], nf["SERVER_TCP_FLAGS"]) & (nf["PROTOCOL"] == 6)
        rst = ent.tcp_rst(nf["TCP_FLAGS"])
        short = (nf["FLOW_DURATION_MILLISECONDS"] < ent.SHORT_MS) & (nf["IN_PKTS"] <= 2)
        out: list = []
        up, table = self.state.update, self.state.src
        ages = np.empty(n, np.int64)
        for i in range(n):
            up(float(t[i]), src[i], dst[i], int(nf["L4_DST_PORT"][i]), float(nf["IN_BYTES"][i]), float(nf["IN_PKTS"][i]),
               float(nf["OUT_PKTS"][i]), bool(syn_only[i]), bool(rst[i]), bool(short[i]), out)
            ages[i] = table[src[i]].n
        xe = np.asarray(out, np.float32).reshape(n, -1)
        cols["_age"] = ages
        for j, name in enumerate(ent.feature_names()):
            cols[name] = xe[:, j]
        cols.update(nf)
        return t, src, dst, xe, cols

    def calibrate_flows(self, benign_flows: dict) -> dict:
        """This site's own normal traffic (time-ordered, no attack labels): every reference of every tier is rebuilt
        from it. Entity state is updated as well (the site's baseline becomes the stream's memory). Later, the engine
        re-calibrates itself every `recal_s` seconds from the traffic it has seen (sources with accumulated evidence
        left out), so references never drift away from what the site looks like now - and every reference of an
        epoch is built from the same rows, so scores and their references always share one scale."""
        n = len(_col(benign_flows, "t", 0)) if "t" in benign_flows else len(next(iter(benign_flows.values())))
        t, src, dst, xe, cols = self._features(benign_flows, n)
        z1 = self.tier1.raw(xe[:, self.t1_idx])[:, 0]
        svc = ent.service_key(cols["L4_DST_PORT"], cols["PROTOCOL"])
        z2 = np.full(n, np.nan)
        if self.seq is not None:
            x, port, proto = seqnet.event_features(cols, "nf")
            run, _ = self._warm_mask(src, z1)                 # the same selection as live traffic
            z2 = self._seq_scores(src, x, port, proto, run) if run.any() else z2
            self.gru.clear()                  # calibration streams start every source from a zero state
        self.site.add("tier1", z1)
        if self.seq is not None and np.isfinite(z2).any():
            self.site.add("seq", z2[np.isfinite(z2)])
        ck, cf, _ = self._campaign_keys(cols)
        self.campaign.calibrate(ck, src, t, cf)
        self.global_scan.calibrate(["scan"] * n, t, self._scanlike(cols))     # the site's own failed-connect level
        self._pending, self._pending_n = [], 0
        self._buffer = {"t": t, "src": np.asarray(src, dtype=object), "dst": np.asarray(dst, dtype=object), "z1": z1, "z2": z2, "svc": svc,
                        "xa": xe[:, ANOM_IDX].astype(np.float32), "xd": xe[:, DST_IDX].astype(np.float32), "age": age_bucket(cols["_age"])}
        self.warmup_short = bool(n and (t >= t[0] + self.warmup_s).sum() < 1000)
        self._calibrate_refs(self._buffer, warmup=0.0 if self.warmup_short else self.warmup_s)
        if self._anchor is None and n:                 # the first clean calibration anchors every later epoch
            keep = np.sort(np.random.default_rng(0).choice(n, min(n, 80_000), replace=False))
            self._anchor = {k: (self._buffer[k][keep] if isinstance(self._buffer[k], np.ndarray) else self._buffer[k]) for k in self._buffer}
        self.last_recal = float(t[-1]) if n else 0.0
        return {k: self.site.n(k) for k in ("tier1", "seq")}

    def _calibrate_refs(self, buf: dict, warmup: float = 0.0) -> None:
        """Rebuild every reference from one window of (mostly benign) traffic: tier-1 by service x age, anomaly
        features and scores by age, sequence scores by age, and the evidence reference out of sample (the last
        quarter of the window scored against references from the first three quarters). Steady-state references
        (old sources) skip the first `warmup` seconds; young sources are reference material from the first flow."""
        t, src, z1, z2, svc, xa = (buf[k] for k in ("t", "src", "z1", "z2", "svc", "xa"))
        dst, xd = buf.get("dst"), buf.get("xd")
        n = len(t)
        if n == 0:
            return
        age = self._age_of(buf)
        keep = (t >= t[0] + warmup) | (age < len(AGE_EDGES))

        def build(m):
            self.rolling["tier1"].set_base(z1[m])
            self.roll_groups = {}
            for key, mm in self._group_keys(svc[m], age[m]):
                self.roll_groups[key] = RollingReference(self.rolling["tier1"].window_s)
                self.roll_groups[key].set_base(z1[m][mm])
            if self.anom is not None:
                self.anom.set_base(xa[m])
                sc = self.anom.feature_scores(xa[m])
                self.anom_s.set_base(sc)
                for b in range(len(AGE_EDGES) + 1):
                    self.anom_age[b].set_base(sc[age[m] == b] if (age[m] == b).any() else np.zeros(0))
            if self.seq is not None:
                ok = m & ~np.isnan(z2)
                self.rolling["seq"].set_base(z2[ok])
                for b in range(len(AGE_EDGES) + 1):
                    self.seq_age[b].set_base(z2[ok & (age == b)] if (ok & (age == b)).any() else np.zeros(0))

        q = t[0] + 0.75 * (t[-1] - t[0])
        build(keep & (t < q))
        self._calibrate_evidence(t, src, z1, svc, xa, age, t_from=q)
        if xd is not None and dst is not None:
            if len(xd):
                self.dst_rate_hi = float(np.quantile(xd[keep, 0], 0.999)) if keep.any() else float(np.quantile(xd[:, 0], 0.999))
            self.dst_anom.set_base(xd[keep & (t < q)])
            self.dst_anom_s.set_base(self._dst_scores(xd[keep & (t < q)]))
            self._calibrate_dst_evidence(t, dst, xd, t_from=q)
        build(keep)

    def _age_of(self, buf: dict) -> np.ndarray:
        return buf["age"] if "age" in buf else np.zeros(len(buf["t"]), np.int8) + len(AGE_EDGES)

    def _remember(self, t, src, dst, z1, z2, svc, xa, xd, age, ok) -> None:
        """keep the recent traffic (except flows that raised any tier) for the next re-calibration epoch. O(batch): the
        rows are queued and folded into the buffer (window cut, subsample, per-source cap) only every ~50k rows and
        before a re-calibration or a snapshot - doing it on every call cost ~20-50 ms per call (measured: 48 flows/s
        at batch 1)"""
        new = {"t": t[ok], "src": np.asarray(src, dtype=object)[ok], "dst": np.asarray(dst, dtype=object)[ok], "z1": z1[ok],
               "z2": z2[ok], "svc": svc[ok], "xa": xa[ok].astype(np.float32), "xd": xd[ok].astype(np.float32), "age": age[ok]}
        self.__dict__.setdefault("_pending", []).append(new)
        self._pending_n = self.__dict__.get("_pending_n", 0) + len(new["t"])
        if len(t):                                   # the stream's clock (its newest rows may all have been left out)
            self._buf_now = max(self.__dict__.get("_buf_now", -np.inf), float(t[-1]))
        if self._pending_n >= 50_000:
            self._flush_buffer()

    def _flush_buffer(self) -> None:
        pend = self.__dict__.get("_pending")
        if not pend:
            return
        self._pending, self._pending_n = [], 0
        b = self._buffer
        if "age" not in b:
            b["age"] = np.full(len(b["t"]), len(AGE_EDGES), np.int8)
        for k in b:
            b[k] = np.concatenate([b[k]] + [p[k] for p in pend])
        t = b["t"]
        now = max(float(t[-1]) if len(t) else -np.inf, self.__dict__.get("_buf_now", -np.inf))
        cutoff = now - self.recal_window_s
        m = b["t"] >= cutoff
        if m.sum() > self.recal_max_rows:            # bounded memory: a random subset, time order kept
            idx = np.flatnonzero(m)
            idx = np.sort(np.random.default_rng(int(now) % 100_003).choice(idx, self.recal_max_rows, replace=False))
            m = np.zeros(len(m), bool)
            m[idx] = True
        for k in b:
            b[k] = b[k][m]
        # no single source may write more than ~1% of the next reference (an un-alarmed slow attacker would otherwise
        # become its tail); a busy benign source (a NAT gateway, a CDN edge) is subsampled, not excluded
        n_b = len(b["src"])
        if n_b > 20_000:
            u, inv, cnt = np.unique(b["src"].astype(str), return_inverse=True, return_counts=True)
            cap = max(200, int(0.01 * n_b))
            if (cnt > cap).any():
                keep = np.ones(n_b, bool)
                rng = np.random.default_rng(n_b)
                for j in np.flatnonzero(cnt > cap):
                    idx = np.flatnonzero(inv == j)
                    keep[rng.choice(idx, len(idx) - cap, replace=False)] = False
                for k in b:
                    b[k] = b[k][keep]

    def recalibrate_now(self, now: float | None = None) -> bool:
        """operator-triggered: rebuild every reference from the recent window at once (accept the current traffic as normal)"""
        self._flush_buffer()
        b = self._buffer
        if len(b["t"]) < 1000:
            return False
        self.last_recal = -np.inf
        self._maybe_recalibrate(float(b["t"][-1]) if now is None else now)
        return True

    def _sustained(self, srcs, t, alarm, k: float = 3.0) -> set:
        """sources whose alarms keep coming (a decayed count over the re-calibration window reaching k): they are
        kept out of every reference, while a source with a single alarm - the benign tail - is not"""
        out, tau = set(), self.recal_window_s
        for i in np.flatnonzero(np.asarray(alarm)):
            s, ti = str(srcs[i]), float(t[i])
            st = self._alarm_src.get(s)
            if st is None:
                st = self._alarm_src[s] = [0.0, ti]
                if len(self._alarm_src) > 200_000:
                    self._alarm_src.popitem(last=False)
            st[0] = st[0] * math.exp(-max(0.0, ti - st[1]) / tau) + 1.0
            st[1] = ti
            if st[0] >= k:
                out.add(s)
        return out

    def _may_contribute(self, src: str, t: float, ref_n: int) -> bool:
        """no single source may write more than ~1% of a reference (at least 30 entries): a scraper or a scanner
        that is not alarmed yet would otherwise become the reference's tail within minutes and never alarm again;
        a very busy benign client is merely capped, not excluded"""
        st = self._contrib.get(src)
        tau = self.rolling["http"].window_s
        if st is None:
            st = self._contrib[src] = [0.0, t]
            if len(self._contrib) > 200_000:
                self._contrib.popitem(last=False)
        st[0] *= math.exp(-max(0.0, t - st[1]) / tau)
        st[1] = t
        if st[0] >= max(30.0, 0.01 * ref_n):
            return False
        st[0] += 1.0
        return True

    def sustained_now(self, src: str, t: float, k: float = 3.0) -> bool:
        st = self._alarm_src.get(str(src))
        return bool(st and st[0] * math.exp(-max(0.0, t - st[1]) / self.recal_window_s) >= k)

    def _drift_guard(self, b: dict, now: float) -> dict:
        """real drift or a slow poisoning? Before the recent window becomes the reference, its upper tail is compared
        with the anchor's: if more traffic than expected sits above the anchor's 99th percentile, the question is *who*
        put it there. A site that changed (a release, a new client population) shifts the tail across many sources -
        adopted, and reported. A handful of sources dragging the tail up epoch after epoch is an attacker teaching
        the reference to accept them (a 'boiling frog') - their rows are kept out and they are logged."""
        a = self._anchor
        if a is None or "z1" not in a or len(a["z1"]) < 1000 or len(b["t"]) < 1000:
            return b
        q = float(np.quantile(a["z1"], 0.99))
        # sources the anchor already knew are part of the site's normal: their tail is drift, not poisoning
        if not hasattr(self, "_anchor_src"):
            self._anchor_src = set(np.asarray(a["src"], dtype=object).tolist())
        known = _member(b["src"], self._anchor_src) if self._anchor_src else np.zeros(len(b["src"]), bool)
        tail = (b["z1"] > q) & ~known
        frac = float(tail.mean())
        ev = {"t": now, "tail_share": round(frac, 4), "expected": 0.01}
        if frac <= 0.03:
            return b
        srcs, counts = np.unique(np.asarray(b["src"], dtype=object)[tail], return_counts=True)
        order = np.argsort(-counts)
        top = srcs[order[:20]]
        conc = float(counts[order[:20]].sum() / max(1, counts.sum()))
        ev.update(tail_sources=int(len(srcs)), top20_share=round(conc, 3))
        if conc > 0.5 or len(srcs) < 30:
            keep = ~_member(b["src"], set(top.tolist()))
            b = {k: (v[keep] if isinstance(v, np.ndarray) and len(v) == len(keep) else v) for k, v in b.items()}
            ev["verdict"] = "concentrated: held out of the reference (suspected slow poisoning)"
            for s in top[:20]:
                self._log("flow", now, str(s), "-", 0, "attack_generic", 60, 7,
                          evidence=f"drift guard: this source drives {int(counts[srcs == s][0])} of the window's {int(tail.sum())} "
                                   f"tail flows (anchor q99); kept out of the reference")
        else:
            ev["verdict"] = "broad: adopted as site drift"
        self.drift_events.append(ev)
        if len(self.drift_events) > 500:
            del self.drift_events[:100]
        return b

    def _maybe_recalibrate(self, now: float) -> None:
        if now - self.last_recal < self.recal_s:
            return
        self._flush_buffer()
        b = self._buffer
        if len(b["t"]) >= 5000:
            b = self._drift_guard(b, now)
            if self._anchor is not None:            # anchor + recent window: the reference can track the site but
                merged = {k: (np.concatenate([self._anchor[k], b[k]]) if isinstance(b.get(k), np.ndarray) and k in self._anchor else b.get(k)) for k in b}
            else:
                merged = b
            self._calibrate_refs(merged, warmup=0.0)
            self.last_recal = now
            self.counts["recalibrations"] = self.counts.get("recalibrations", 0) + 1

    @staticmethod
    def _group_keys(svc, age):
        """(key, mask) for every Mondrian cell present: service x age, then service, then age"""
        out = []
        for g in np.unique(svc):
            for b in np.unique(age[svc == g]):
                out.append((f"{int(g)}/{int(b)}", (svc == g) & (age == b)))
            out.append((f"{int(g)}", svc == g))
        for b in np.unique(age):
            out.append((f"age{int(b)}", age == b))
        return out

    def _tier1_pvalues(self, z1, svc, age) -> np.ndarray:
        """tier-1 score -> p-value against the site's reference of the same service x source age; a client's first
        flows are judged only against other clients' first flows, and with no such reference yet they yield no
        evidence at all (abstain)"""
        n = len(z1)
        p1 = np.ones(n)
        for g in np.unique(svc):
            for b in np.unique(age[svc == g]):
                m = (svc == g) & (age == b)
                keys = (f"{int(g)}/{int(b)}", f"age{int(b)}") + (() if b < len(AGE_EDGES) else (f"{int(g)}", "*"))
                for key in keys:
                    ref_g = self.rolling["tier1"] if key == "*" else self.roll_groups.get(key)
                    if ref_g is not None and ref_g.n() >= self.mondrian_min:
                        p1[m] = ref_g.pvalues(z1[m])
                        break
        return p1

    def _anomaly_pvalues(self, sc, age) -> np.ndarray:
        """anomaly score -> p-value against sources of the same age. When a young bucket's own reference is too thin
        (sparsely-populated ages, e.g. sources with 2-20 flows), it is pooled with its neighbouring young buckets
        instead of abstaining - otherwise a low-and-slow scanner living in a thin bucket would never accumulate any
        evidence. Only a source's first flow (bucket 0) still abstains; newborn suppression + block_ok keep benign safe."""
        young = age < len(AGE_EDGES)
        pa = np.where(young, 1.0, self.anom_s.pvalues(sc)) if self.anom_s.n() else np.ones(len(sc))
        pooled_cache: dict = {}
        for b in range(len(AGE_EDGES) + 1):
            m = age == b
            if not m.any():
                continue
            if self.anom_age[b].n() >= self.mondrian_min:
                pa[m] = self.anom_age[b].pvalues(sc[m])
            elif 1 <= b < len(AGE_EDGES):
                lo, hi = max(1, b - 1), min(len(AGE_EDGES), b + 1)
                key = (lo, hi)
                if key not in pooled_cache:
                    refs = [self.anom_age[k].ref() for k in range(lo, hi + 1) if self.anom_age[k].n()]
                    pooled_cache[key] = np.sort(np.concatenate(refs)) if refs else None
                ref = pooled_cache[key]
                if ref is not None and len(ref) >= self.mondrian_min:
                    pa[m] = (1 + len(ref) - np.searchsorted(ref, sc[m] - _tol(sc[m]), side="left")) / (len(ref) + 1)
        return pa

    def _calibrate_evidence(self, t, src, z1, svc, xa, age, t_from: float = -np.inf) -> None:
        """benign source-minute maxima of the accumulated evidence over the window -> reference (evidence accumulates
        over every flow; only minutes at or after t_from - scored out of sample - enter the reference)"""
        p1 = self._tier1_pvalues(z1, svc, age)
        p = p1
        if self.anom is not None and self.anom_s.n() >= 200:
            p = np.minimum(1.0, 2.0 * np.minimum(p1, self._anomaly_pvalues(self.anom.feature_scores(xa), age)))
        lpall = -np.log(np.maximum(p, 1e-12))
        sus_all = np.where(p < 0.5, lpall, 0.0)
        _pos = sus_all[sus_all > 0]
        # CUSUM allowance k = mean + std of the benign increment: benign has a negative drift (bounded excursions at any
        # request rate); an attack's increments (-log p of ~7-12) still climb fast
        self.evid_drift = float(_pos.mean() + _pos.std()) if len(_pos) > 1 else 0.0
        ev = SourceEvidence(self.evidence.tau, self.evidence.max)
        ev_s = SourceEvidence(self.evidence_slow.tau, self.evidence_slow.max)
        mx: dict = {}
        mxs: dict = {}
        for i in range(len(t)):
            inc = (sus_all[i] - self.evid_drift) if sus_all[i] > 0 else -self.evid_drift * 0.0   # zero-mean under benign
            e = ev.update(src[i], float(t[i]), inc)
            es = ev_s.update(src[i], float(t[i]), inc)
            if t[i] < t_from:
                continue
            k = (src[i], int(t[i] // 60))
            mx[k] = max(mx.get(k, 0.0), e)
            ks = (src[i], int(t[i] // 1800))
            mxs[ks] = max(mxs.get(ks, 0.0), es)
        self.ev_ref.set_base(np.fromiter(mx.values(), np.float64))
        self.ev_slow_ref.set_base(np.fromiter(mxs.values(), np.float64))

    def _warm_mask(self, src, z1) -> tuple[np.ndarray, set]:
        """which flows the sequence net runs on: a source is warm for `warm_flows` flows after its tier-1 score last
        crossed the low-water mark. The same rule selects the calibration flows, so the GRU reference is drawn from
        the same population as the live scores (warm sources are not a random sample)."""
        n = len(src)
        run, cold = np.zeros(n, bool), set()
        for i in range(n):
            s = src[i]
            g = self.gru.get(s)
            if z1[i] > self.warm_logit:
                g = (g[0] if g is not None else np.zeros((self.seq.layers, self.seq.d), np.float32), self.warm_flows)
            if g is not None and g[1] > 0:
                run[i] = True
                self.gru[s] = (g[0], g[1] - 1)
                self.gru.move_to_end(s)
                if len(self.gru) > self.gru_max:
                    old, _ = self.gru.popitem(last=False)
                    cold.discard(old)
            elif g is not None:
                cold.add(s)             # deleted after this batch is scored (its earlier flows may still be pending)
        return run, cold

    def _dst_scores(self, xd) -> np.ndarray:
        """Fisher anomaly of a destination's (flows, no-reply share, new-source share, slow flows) vs the site's
        benign destinations - high when one destination is being flooded, especially by new sources."""
        from dibaguard.decide import feature_scores
        return feature_scores(self.dst_anom.ref(), np.asarray(xd, np.float64)) if self.dst_anom.n() else np.zeros(len(xd))

    def _calibrate_dst_evidence(self, t, dst, xd, t_from: float = -np.inf) -> None:
        sc = self._dst_scores(xd)
        ev = SourceEvidence(self.dst_evidence.tau, self.dst_evidence.max)
        mx: dict = {}
        pv = self.dst_anom_s.pvalues(sc) if self.dst_anom_s.n() else np.ones(len(sc))
        lp = -np.log(np.maximum(pv, 1e-12))
        pos = lp[pv < 0.5]
        self.dst_drift = float(pos.mean() + pos.std()) if len(pos) > 1 else 0.0
        for i in range(len(t)):
            e = ev.update(dst[i], float(t[i]), float(lp[i] - self.dst_drift) if pv[i] < 0.5 else 0.0)
            if t[i] < t_from:
                continue
            k = (dst[i], int(t[i] // 60))
            mx[k] = max(mx.get(k, 0.0), e)
        self.dst_ev_ref.set_base(np.fromiter(mx.values(), np.float64))

    def _seq_scores(self, src, x, port, proto, run: np.ndarray) -> np.ndarray:
        """GRU steps for the flows flagged in `run`, in time order per source; batched by rounds of distinct sources."""
        m = self.seq
        z = np.full(len(src), np.nan, np.float32)
        idx = np.flatnonzero(run)
        if len(idx) == 0:
            return z
        e = m.embed(x[idx], port[idx], proto[idx])
        pending: dict = {}
        for k, i in enumerate(idx):
            pending.setdefault(src[i], []).append(k)
        rounds = max(len(v) for v in pending.values())
        keys = list(pending)
        for r in range(rounds):
            live = [s for s in keys if r < len(pending[s])]
            ks = np.array([pending[s][r] for s in live])
            zero = np.zeros((m.layers, m.d), np.float32)
            st = np.stack([self.gru.get(s, (zero, 0))[0] for s in live], 1)          # [layers, n, d]
            st = m.gru_step(e[ks], st)
            zz = m.head(st[-1], self.gamma, self.beta)
            for j, s in enumerate(live):
                self.gru[s] = (st[:, j], self.gru.get(s, (zero, 0))[1])
            z[idx[ks]] = zz[:, 0]
            self._fam = getattr(self, "_fam", {})
            for j, s in enumerate(live):
                self._fam[idx[ks[j]]] = int(zz[j, 1:].argmax()) if zz.shape[1] > 1 else -1
        self.counts["seq_runs"] += len(idx)
        return z

    def flows(self, flows: dict) -> dict:
        """Time-ordered batch (dict of arrays with FLOW_FIELDS) -> per-flow verdicts."""
        n = len(_col(flows, "t", 0)) if "t" in flows else len(next(iter(flows.values())))
        t, src, dst, xe, cols = self._features(flows, n)
        z1 = self.tier1.raw(xe[:, self.t1_idx])[:, 0]
        p1 = 1 / (1 + np.exp(-z1 / self.temp1))
        thr1, src1 = self.threshold("tier1")
        a_tier = self.alpha / (2 * self.n_tiers())          # the per-flow tiers share half of the budget, the evidence tier has the other half
        svc = ent.service_key(cols["L4_DST_PORT"], cols["PROTOCOL"])
        ages = age_bucket(cols["_age"])
        pv1 = self._tier1_pvalues(z1, svc, ages)                    # Mondrian by service x age; abstains when uncalibrated
        alarm = pv1 < a_tier
        p, tier, fam = p1.copy(), np.full(n, 1, np.int8), np.full(n, "benign", dtype=object)
        z2 = np.full(n, np.nan, np.float32)
        if self.seq is not None:
            run, cold = self._warm_mask(src, z1)
            if run.any():
                self._fam = {}
                x, port, proto = seqnet.event_features(cols, "nf")
                z2 = self._seq_scores(src, x, port, proto, run)
                p2 = 1 / (1 + np.exp(-np.nan_to_num(z2, nan=-30.0) / self.temp2))
                zz = np.nan_to_num(z2, nan=-30.0)
                pv2 = np.ones(n)                                     # judged against sources of the same age, else abstain
                for b in range(len(AGE_EDGES) + 1):
                    m = run & (ages == b)
                    if m.any() and self.seq_age[b].n() >= self.mondrian_min:
                        pv2[m] = self.seq_age[b].pvalues(zz[m])
                hit = run & (pv2 < a_tier) & (ages >= 2)             # the sequence net needs a few flows of history
                alarm |= hit
                better = run & (p2 > p)
                p[better], tier[better] = p2[better], 2
                for i in np.flatnonzero(hit):
                    k = self._fam.get(i, -1)
                    fam[i] = self.seq.families[k + 1] if 0 <= k + 1 < len(self.seq.families) and self.seq.families else "attack_generic"
            for s in cold:
                self.gru.pop(s, None)
        anomaly_s = np.zeros(n)
        if self.anom is not None and self.anom_s.n() >= ConformalFPR.n_needed(a_tier, 0.5):
            # site-only tier: how far outside this site's own normal behaviour the source is; its score is calibrated
            # like every other tier, against sources of the same age (a client's first flows vs other first flows)
            anomaly_s = self.anom.feature_scores(xe[:, ANOM_IDX])
            pa = self._anomaly_pvalues(anomaly_s, ages)
            hit = pa < a_tier
            alarm |= hit
            tier[hit & (tier == 1) & ~(pv1 < a_tier)] = 3
            p = np.maximum(p, np.where(hit, 1.0 - pa, 0.0))
        # evidence tier (the real-world verdict): accumulate -log p per source, judge the source-minute
        pflow = np.minimum(1.0, 2.0 * np.minimum(pv1, self._anomaly_pvalues(anomaly_s, ages) if self.anom is not None and self.anom_s.n() >= 200 else 1.0))
        evid = np.zeros(n)
        evid_s = np.zeros(n)
        for i in range(n):
            inc = (float(-np.log(max(pflow[i], 1e-12))) - self.evid_drift) if pflow[i] < 0.5 else 0.0
            evid[i] = self.evidence.update(src[i], float(t[i]), inc)
            evid_s[i] = self.evidence_slow.update(src[i], float(t[i]), inc)
        a_ev = self.alpha / 2                              # the destination tier's share
        a_ev2 = self.alpha / 4                             # split between the fast and the slow source-evidence horizons
        thr_e = self.ev_ref.threshold(a_ev2) if self.ev_ref.n() >= ConformalFPR.n_needed(a_ev2, 0.5) else float("inf")
        thr_es = self.ev_slow_ref.threshold(a_ev2) if self.ev_slow_ref.n() >= max(200, ConformalFPR.n_needed(a_ev2, 0.5)) else float("inf")
        hit_e = (evid > thr_e) | (evid_s > thr_es)        # fast burst or slow accumulation over ~30 min
        tier[hit_e & ~alarm] = 4
        alarm |= hit_e
        p = np.maximum(p, np.where(hit_e, 1.0 - np.minimum(self.ev_ref.pvalues(evid), self.ev_slow_ref.pvalues(evid_s)), 0.0))
        # destination-flood tier: a destination overwhelmed by many (often new) sources - distributed / volumetric
        hit_d = np.zeros(n, bool)
        piling = np.zeros(n, bool)
        if self.dst_anom.n() >= self.mondrian_min and self.dst_ev_ref.n() >= ConformalFPR.n_needed(a_ev, 0.5):
            dsc = self._dst_scores(xe[:, DST_IDX])
            dpv = self.dst_anom_s.pvalues(dsc)
            dev = np.zeros(n)
            for i in range(n):
                dev[i] = self.dst_evidence.update(dst[i], float(t[i]),
                                                  float(-np.log(max(dpv[i], 1e-12)) - self.dst_drift) if dpv[i] < 0.5 else 0.0)
            thr_d = self.dst_ev_ref.threshold(a_ev)
            flooded = dev > thr_d                              # the destination is under a flood right now
            # flag only the flows piling on (new to the destination), so the destination's established benign clients
            # are not blamed for a flood aimed at their server
            volumetric = xe[:, DST_IDX[0]] > 1.5 * self.dst_rate_hi          # the destination's rate is far over its site norm
            # the *source* must be piling on itself (several flows within ~10 s - a flooding bot does, a visitor who
            # opened one page of the attacked server does not): the destination's new-source share is the same number
            # for every flow to it and cannot tell them apart. (A single off-profile flow is not enough: among 100k
            # benign flows a few reach any per-flow p-value by chance.)
            piling = xe[:, SRC_RATE10] >= 3.0
            # piling on = many new sources (a DDoS), or - for a pool of bots that keeps reconnecting, no longer 'new' after
            # its first contact - a volume far over the destination's norm carried by young sources (an old, trusted
            # machine's heavy job never qualifies)
            extreme = (xe[:, DST_IDX[0]] > 2.5 * self.dst_rate_hi) & (cols["_age"] < 100)
            many_new = xe[:, DST_NEWSRC] > 0.3
            hit_d = flooded & volumetric & (many_new | extreme) & (pflow < 0.5)
            # on volume alone (the bot pool) a source is acted on only if it floods itself; a new visitor who is busy for
            # a moment on the attacked server is watched
            piling = (piling & many_new) | (xe[:, SRC_RATE10] >= 10.0)
            tier[hit_d & ~alarm] = 5
            alarm |= hit_d
            p = np.maximum(p, np.where(hit_d, 1.0 - dpv, 0.0))
        elif np.isfinite(self.dst_rate_hi):
            # the destination evidence reference is still filling (a small network has few destinations): until then
            # only the unmistakable flood counts - three times the destination's calibrated peak rate, carried by young
            # sources that pile on themselves
            piling = xe[:, SRC_RATE10] >= 10.0
            hit_d = (xe[:, DST_IDX[0]] > 3.0 * self.dst_rate_hi) & (cols["_age"] < 100) & (pflow < 0.5)
            tier[hit_d & ~alarm] = 5
            alarm |= hit_d
            p = np.maximum(p, np.where(hit_d, 0.99, 0.0))
        # a source's first flows carry no behaviour to judge; behavioural tiers (1 Mondrian / 2 seq / 3 anomaly /
        # 4 evidence) abstain until it has a short history. The supervised prior firing hard (z1 well over its
        # shipped 1e-4 threshold) and the destination-flood tier still act, so blatant first-flow attacks are kept.
        newborn = cols["_age"] < 3
        prior_hard = z1 > self.shipped["tier1"].get(1e-4, np.inf)
        self_evident = pflow < a_tier            # a single flow so far off-profile it is actionable on its own
        suppress = newborn & ~prior_hard & ~self_evident & (tier != 5)
        alarm &= ~suppress
        tier[suppress] = 1
        # block_ok: alarms safe to auto-block. A newborn source raising only a soft behavioural tier (1 Mondrian /
        # 3 anomaly) is watched / challenged, never blocked on its first flows; everything with history or a hard
        # signal (evidence 4, sequence 2, flood 5, self-evident, or the prior firing hard) is block-safe.
        soft_newborn = alarm & newborn & np.isin(tier, [1, 3]) & ~self_evident & ~prior_hard
        block_ok = alarm & ~soft_newborn
        # a flow alarmed only because its destination is flooded is acted on only when its source piles on itself:
        # a visitor who opened one page of the attacked server is watched, never rate-limited
        block_ok &= ~(hit_d & ~piling)              # whichever tier raised it: its destination is flooded, the source did not flood
        if alarm.any():                           # transparent family from the stream counters; the GRU head only fills gaps
            rules = ent.family_rules(xe[alarm], np.asarray(cols["L4_DST_PORT"])[alarm])
            cur = fam[alarm]
            fam[alarm] = np.where(rules != "attack_generic", rules, np.where(cur == "benign", "attack_generic", cur))
        # a flooded destination names the flow only when the flood is what raised it, or the source itself piles on:
        # a slow scanner whose probe lands on a host someone else is flooding is still a scanner (named 'dos', it was
        # rate-limited instead of blocked)
        by_flood = hit_d & ((tier == 5) | piling)
        fam[by_flood] = np.where(xe[by_flood, DST_NEWSRC] > 0.3, "ddos", "dos")
        sev = np.array([severity(f) if a else 0.0 for f, a in zip(fam, alarm)])
        pe = np.where(alarm, p, np.minimum(p, 0.5))
        score = np.array([event_score(float(a), float(b)) for a, b in zip(pe, sev)], np.int16)
        # the traffic of this batch (sources with accumulated evidence left out) is tomorrow's reference: it is
        # remembered and, every recal_s seconds, every reference is rebuilt from the recent window at once
        # global distributed-scan tier: SYN-only / no-reply short flows are the scan signature (benign traffic has ~none);
        # when their site-wide decayed rate spikes, the cohort is a distributed scan even if each source is low-volume.
        scanlike = self._scanlike(cols)
        camp_scan, scan_rate = self.global_scan.score(["scan"] * n, t, scanlike, floor=self.global_scan_floor, rates=True)
        # a milder surge (over the floor and 1.5x the site's own level, short of the 4x the whole cohort needs): only a
        # young source that does nothing but fail is the scan's. Never the floor alone: a network whose normal is
        # full of young unanswered sources (DNS clients whose replies the sensor misses, internet background noise)
        # sits over it all day (measured on real benign traffic: ~half of all source-minutes alarmed)
        mild = scan_rate > self.global_scan_floor + 1.5 * self.global_scan.expected("scan", float(t[-1]) if n else 0.0)
        young_failing = (cols["_age"] < 20) & (np.maximum(xe[:, SRC_NOREPLY2], xe[:, SRC_SYN2]) > 0.8) & mild
        # even while a site-wide surge is on, the flows flagged are those of scan-shaped sources (young, or failing at
        # least half the time): an established client's one failed connect during someone else's scan is collateral
        camp_scan &= (cols["_age"] < 50) | (np.maximum(xe[:, SRC_NOREPLY2], xe[:, SRC_SYN2]) > 0.5)
        camp_scan |= scanlike & young_failing
        if camp_scan.any():
            newc = camp_scan & ~alarm
            tier[newc] = 5
            alarm |= camp_scan
            fam[camp_scan & (fam == "benign")] = "portscan"
        # campaign tier: a fingerprint (service x failed-connect / short-auth-session) with far more distinct failing
        # sources than its own history - members are the sources that failed on it and never succeeded
        ck, cf, authlike = self._campaign_keys(cols)
        # a completed connection to a port answers for that port's failed-connect fingerprint too (the two are separate
        # keys): without it every ordinary client with one failed connect became a member of a live scan campaign -
        # and block-safe (measured on SIMARGL 2021 benign traffic during its scan: 20.7% of benign source-minutes)
        okc = np.flatnonzero(~cf)
        if len(okc):
            self.campaign.credit([("probe", ck[i][1]) for i in okc], np.asarray(src)[okc], np.asarray(t)[okc])
        camp = self.campaign.update(ck, src, t, cf)
        camp_ev: dict = {}
        # a member is a source shaped like a campaign's: young, or failing at least half the time - an established
        # client that mostly succeeds is not one of a botnet's addresses because one connect of it failed meanwhile
        shaped = (cols["_age"] < 50) | (np.maximum(xe[:, SRC_NOREPLY2], xe[:, SRC_SYN2]) > 0.5)
        member = np.array([c is not None for c in camp], bool) & shaped
        if member.any():
            tier[member & ~alarm] = 6
            alarm |= member
            fam[member] = np.where(authlike[member], "bruteforce", "portscan")
            # a member of an ICMP campaign (no service, port 0) is alarmed, not block-safe: echo records without a reply
            # are common in benign traffic (one-way exports, filtered replies) and blocking a ping-sweep's cohort buys
            # little (measured on SIMARGL 2021 during its scan: ~12k benign ICMP records block-safe otherwise)
            block_ok = block_ok | (member & (np.asarray(cols["L4_DST_PORT"]).astype(np.int64) != 0))
            for i in np.flatnonzero(member):
                key, k_now, mu = camp[i]
                camp_ev[i] = (f"campaign: {k_now} distinct sources failing on {key[1]}/{key[0]} this window (normal ~{mu:.0f}); "
                              f"this source only ever failed there")
            score = np.where(member, np.maximum(score, 75), score).astype(np.int16)
        # slow scan: over the 30-minute horizon the source keeps reaching (host, port) pairs it never used and nearly
        # every attempt fails - port walking, however slow (a probe every 20 s never adds up on the 2-minute horizon).
        # Like the guessing rule below, only for a source the site has not known for long (< 500 flows): an established
        # host whose lookups go unanswered in the flow records (DNS to many resolvers, CICFlowMeter time-outs) walks
        # 'new' endpoints too (measured on CSE-CIC-IDS2018: ~11k benign records otherwise); the evidence tiers judge it
        slow_scan = ((xe[:, SRC_FLOWS3] >= 20.0) & (xe[:, SRC_NEWPORT3] >= 0.8)
                     & (np.maximum(xe[:, SRC_NOREPLY3], xe[:, SRC_SYN3]) >= 0.8) & self._scanlike(cols) & (cols["_age"] < 500))
        if slow_scan.any():
            tier[slow_scan & ~alarm] = 4
            alarm |= slow_scan
            fam[slow_scan & np.isin(fam, ["benign", "attack_generic"])] = "portscan"
            block_ok = block_ok | slow_scan
            pe = np.where(slow_scan, np.maximum(pe, 0.99), pe)
            sev = np.where(slow_scan, np.maximum(sev, severity("portscan")), sev)
            for i in np.flatnonzero(slow_scan):
                camp_ev.setdefault(i, f"slow scan: {xe[i, SRC_FLOWS3]:.0f} attempts in ~30 min, {100 * xe[i, SRC_NEWPORT3]:.0f}% to "
                                      f"endpoints it never used, {100 * max(xe[i, SRC_NOREPLY3], xe[i, SRC_SYN3]):.0f}% failing")
            score = np.where(slow_scan, np.maximum(score, 75), score).astype(np.int16)
        # sustained guessing: one source opening short sessions to an interactive login service again and again for
        # minutes - 15 tries a minute for half an hour - is password guessing even when no single flow is unusual and
        # the record shows no failure (the handshake completes, the server just says no). >= 120 flows on the 30-min
        # horizon is >= ~0.2 per second for >= 10 minutes; nearly all of them to an endpoint the source already used
        # Only for a source the site has not known for long (< 500 flows): a guesser is caught long before that (after
        # ~120 tries), while an established internal client whose long sessions the flow exporter cuts into many short
        # records is not a guesser (measured on CSE-CIC-IDS2018: ~82k benign RDP records otherwise)
        guess = (authlike & np.isin(np.asarray(cols["L4_DST_PORT"]).astype(np.int64), INTERACTIVE_LOGIN)
                 & (xe[:, SRC_FLOWS3] >= 120.0) & (xe[:, SRC_REPEAT3] >= 0.8) & (cols["_age"] < 500))
        if guess.any():
            tier[guess & ~alarm] = 4
            alarm |= guess
            fam[guess] = "bruteforce"
            block_ok = block_ok | guess
            pe = np.where(guess, np.maximum(pe, 0.99), pe)
            sev = np.where(guess, np.maximum(sev, severity("bruteforce")), sev)
            for i in np.flatnonzero(guess):
                camp_ev.setdefault(i, f"sustained guessing: {xe[i, SRC_FLOWS3]:.0f} short sessions in ~30 min, "
                                      f"{100 * xe[i, SRC_REPEAT3]:.0f}% to the same login service")
            score = np.where(guess, np.maximum(score, 80), score).astype(np.int16)
        # a source or destination that raised any alarm this batch is excluded from the reference *entirely* (every
        # one of its flows), so a partially-detected attack cannot seep into the self-training and loosen thresholds
        # whole-source exclusion only for sources whose alarms name a recognisable attack shape (port walking, repeated
        # failures, floods ...); an un-named alarm on a busy-but-ordinary source is more likely drift, and the reference
        # must be allowed to learn it (else a wrongly-flagged benign source stays flagged forever)
        # an established source (hundreds of flows of history) whose flows show no failure at all - every connect
        # answered, none cut short, none reset, none half-open - and that is only named for repeating itself is a
        # busy legitimate machine whose pace changed (a SCADA poller, a monitoring agent, a sync job), not a brute
        # forcer: it stays in the reference (else the alarm becomes permanent) and is watched, never auto-blocked
        # (a workstation's DNS / NTP exchanges are 'short' flows - under 10 ms, one or two packets - and no failure)
        fail_like = (xe[:, SRC_NOREPLY2] > 0.2) | (xe[:, SRC_RST2] > 0.2) | (xe[:, SRC_SYN2] > 0.2)
        # (never on an authentication service: a long-running password guesser completes its handshakes too)
        excusable = ((cols["_age"] >= 500) & ~fail_like & np.isin(fam, ["bruteforce", "attack_generic"]) & ~member
                     & ~np.isin(np.asarray(cols["L4_DST_PORT"]).astype(np.int64), NEVER_EXCUSED))
        block_ok &= ~excusable
        # and when only the statistical tiers (prior, sequence, evidence) object to it - its pace, not its shape - it is
        # not an alarm at all; a change of shape (tier 3), a flood (5), a campaign (6) still is
        quiet = excusable & np.isin(tier, [1, 2, 4])
        if quiet.any():
            alarm &= ~quiet
            fam[quiet] = "benign"
            pe = np.where(quiet, np.minimum(p, 0.5), pe)
            sev = np.where(quiet, 0.0, sev)
            score = np.where(quiet, 0, score).astype(np.int16)
        named = alarm & ~np.isin(fam, ["attack_generic", "benign"]) & ~excusable
        bad_src = set(np.asarray(src)[named].tolist())
        # what the next reference may learn from. A sporadic un-named alarm stays in: at the alpha budget it *is* the
        # benign tail - leaving every alarm out truncates the tail each epoch, the threshold drops, more benign flows
        # alarm, and the false-alarm rate ratchets up hour after hour (measured 1.8% -> 21% of benign source-minutes
        # over six hours). A source alarming again and again is not the tail but a weakly-detected attacker: all of
        # its flows stay out, or it would teach the reference to accept it (measured: an evolving scan 1.0 -> 0).
        bad_src |= self._sustained(src, t, alarm & ~excusable)
        ok = ~named & ~_member(src, bad_src) if bad_src else ~named
        self._remember(t, src, dst, z1, z2, svc, xe[:, ANOM_IDX], xe[:, DST_IDX], ages, ok)
        if n:
            self._maybe_recalibrate(float(t[-1]))
        risk = np.array([self.entities.update(str(s), float(ti), float(pi), float(si), bool(al))
                         for s, ti, pi, si, al in zip(src, t, pe, sev, alarm)])
        for i in range(n):
            self.pressure.update(float(t[i]), float(pe[i]) if alarm[i] else 0.0, fam[i] if alarm[i] else None)
            if self.site.n("tier1") >= 200:
                self.drift.update(self.site.pvalue("tier1", float(z1[i]), fallback=None))
            if alarm[i]:
                self._log("flow", float(t[i]), str(src[i]), str(dst[i]), int(cols["L4_DST_PORT"][i]), fam[i], int(score[i]), int(tier[i]),
                          evidence=camp_ev.get(i, ""), soft=not bool(block_ok[i]) or fam[i] == "attack_generic")   # un-named deviation: low priority
        self.counts["flows"] += n
        self.counts["alarms"] += int(alarm.sum())
        return {"p_attack": p, "alarm": alarm, "family": fam, "severity": sev, "score": score, "tier": tier,
                "entity_risk": risk, "z_tier1": z1, "z_seq": z2, "anomaly_score": anomaly_s, "evidence": evid, "threshold_evidence": thr_e,
                "block_ok": block_ok, "entity_x": xe,
                "flow_p": pflow, "threshold_tier1": thr1, "threshold_source": src1,
                "site_pressure": self.pressure.pressure, "drift": self.drift.drifted}

    @staticmethod
    def _scanlike(cols) -> np.ndarray:
        """SYN-only or unanswered short flows: the scan signature (benign traffic has few)"""
        synflag = ent.tcp_syn_only(cols["CLIENT_TCP_FLAGS"], cols["SERVER_TCP_FLAGS"]) & (np.asarray(cols["PROTOCOL"]) == 6)
        return (np.asarray(synflag) | (np.asarray(cols["OUT_PKTS"]) <= 0)) & (np.asarray(cols["IN_PKTS"]) <= 3)

    @staticmethod
    def _campaign_keys(cols) -> tuple[list, np.ndarray, np.ndarray]:
        """flow fingerprint for the campaign tier: (failure kind, service port); failed = a SYN-only / unanswered probe,
        or a short session on an authentication service (a guessed password closes the connection within seconds)"""
        port = np.asarray(cols["L4_DST_PORT"]).astype(np.int64)
        syn = ent.tcp_syn_only(cols["CLIENT_TCP_FLAGS"], cols["SERVER_TCP_FLAGS"]) & (np.asarray(cols["PROTOCOL"]) == 6)
        probe = (np.asarray(syn) | (np.asarray(cols["OUT_PKTS"]) <= 0)) & (np.asarray(cols["IN_PKTS"]) <= 3)
        authlike = ~probe & np.isin(port, list(AUTH_PORTS)) & (np.asarray(cols["FLOW_DURATION_MILLISECONDS"]) < 3000) & (np.asarray(cols["IN_PKTS"]) <= 40)
        keys = [("probe" if p else "auth", int(pt)) for p, pt in zip(probe, port)]
        return keys, probe | authlike, authlike

    # ------------------------------------------------------------ http
    def _http_key(self, text):
        return zlib.crc32(text.split(" | ")[0].encode("utf-8", "replace")) & 0xFFFFFFFF

    def calibrate_http(self, texts: list[str], deepest=None, host: str | None = None) -> dict:
        """Benign requests of a site (a few hundred): site profile + site threshold. With `host`, the profile and the
        reference belong to that Host only (a WAF in front of many sites keeps one per site); the first one calibrated
        also becomes the default for hosts not yet profiled."""
        if self.http is None:
            return {}
        deepest = deepest if deepest is not None else [0] * len(texts)
        prof = self.http.site_profile(texts, deepest)
        z = self.http.logits(texts, deepest, prof)[:, 0]
        z = clip_upper(z)
        if host is not None:
            self.site_http_by_host[host] = prof
            ref = RollingReference(self.rolling["http"].window_s)
            ref.set_base(z)
            self.http_ref_by_host[host] = ref
        if host is None or self.site.n("http") < 200:
            self.site_http = prof
            self.site.cal.pop("http", None)
            self.site.add("http", z)
            self.rolling["http"].set_base(self.site.cal["http"])
        return {"http": self.site.n("http"), "profile_rows": len(texts), "host": host, "hosts": len(self.site_http_by_host)}

    def http_requests(self, texts: list[str], deepest=None, client=None, host=None, status=None, t=None) -> dict:
        n = len(texts)
        deepest = deepest if deepest is not None else [0] * n
        hosts_in = ["" for _ in range(n)] if host is None else [str(h) for h in host]
        z = np.zeros(n, np.float32)
        if self.http is None:
            # no request model (an open build ships without one): the learned per-request tier abstains - the
            # signatures, the SQL-context analysis, DbNet and the session tier still judge; nothing is "benign by model"
            thr, srct = np.inf, "no-model"
            thr_row = np.full(n, np.inf)
        else:
            thr, srct = self.threshold("http")
            thr_row = np.full(n, thr, np.float64)
            by_host: dict = {}
            for i, h in enumerate(hosts_in):              # one pass (a loop per host was O(requests x hosts))
                by_host.setdefault(h, []).append(i)
            for h, idx in by_host.items():                # each request is judged against its own site's profile and reference
                prof = self.site_http_by_host.get(h, self.site_http)
                z[idx] = self.http.logits([texts[i] for i in idx], [deepest[i] for i in idx], prof)[:, 0]
                ref = self.http_ref_by_host.get(h)
                if ref is not None and ref.n() >= ConformalFPR.n_needed(self.alpha, 0.5):
                    thr_row[idx] = ref.threshold(self.alpha)
        p = 1 / (1 + np.exp(-z / self.temph))
        alarm = z > thr_row
        fam = np.where(alarm, "webattack", "benign").astype(object)
        sess = np.zeros(n, bool)
        z1_s, thr_s, fails_s, rate_s = None, 0.0, None, None
        if client is not None:
            # the request as an event of the client's stream (tier 0): enumeration / brute force / scanners show up here
            tt0 = np.zeros(n) if t is None else np.asarray(t, np.float64)
            hosts0 = ["web"] * n if host is None else list(host)
            st0 = [0] * n if status is None else list(status)
            out0: list = []
            ages = np.empty(n, np.int64)
            for i in range(n):
                port = zlib.crc32(texts[i].split(" | ")[0].encode("utf-8", "replace")) & 0xFFFF
                failed = int(st0[i]) >= 400 if st0[i] else False
                self.state.update(float(tt0[i]), client[i], hosts0[i], port, float(len(texts[i])), 1.0, 0.0 if failed else 1.0, False, False, failed, out0)
                ages[i] = self.state.src[client[i]].n
            xe = np.asarray(out0, np.float32).reshape(n, -1)
            z1 = self.tier1.raw(xe[:, self.t1_idx])[:, 0]
            # the session tier has its own reference: request streams are not flow streams (a visitor reading ten
            # product pages walks ten 'ports'), so the flow tier's threshold would call every browser a scanner.
            # Until the site's own session reference is filled, only streams that are mostly failing can alarm.
            fails = xe[:, ent.feature_names().index("src_noreply_h2")]
            a_s = self.alpha / 2
            ref_s = self.rolling["sess"]
            if ref_s.n() >= ConformalFPR.n_needed(a_s, 0.5):
                thr1 = ref_s.threshold(a_s)
                sess = z1 > thr1
            else:
                thr1, _ = self.threshold("tier1")
                sess = (z1 > thr1) & (fails > 0.5)
            # an established machine client polling the same endpoints at a modest pace with every request answered (a
            # partner integration, a monitor, a sync job) is not an attack shape: the session tier stays quiet for it
            # (the request model and the signatures still judge each request)
            f_ = ent.feature_names()
            steady = ((ages >= 30) & (fails <= 0.05) & (xe[:, f_.index("src_newport_h2")] <= 0.2)
                      & (xe[:, f_.index("src_flows_h1")] < 20.0) & (z <= thr_row))
            sess &= ~steady
            z1_s, thr_s, fails_s = z1, thr1, fails
            rate_s = xe[:, ent.feature_names().index("src_flows_h1")]
            alarm |= sess
            if sess.any():
                # in a request stream the 'port' is the path: port walking = enumeration / scraping, repetition of one
                # path with failures = credential stuffing, a flood = L7 DoS; a bot user-agent names itself
                # HTTP-specific family from the client's own stream: path diversity (the "new port" share is the share
                # of never-seen paths), failure share (4xx/5xx = no reply), repetition of one path, and rate.
                f = ent.feature_names()
                xs = xe[sess]
                newp, fail, rep, rate10 = (xs[:, f.index(k)] for k in ("src_newport_h2", "src_noreply_h2", "src_repeat_h2", "src_flows_h1"))
                ua_bot = np.array([any(b in texts[i].lower() for b in BOT_UA) for i in np.flatnonzero(sess)])
                hfam = np.full(len(xs), "bot", dtype=object)                      # default: automated client
                hfam[(newp > 0.5) & (fail > 0.1)] = "scan"                        # walking ids / paths and failing: enumeration
                hfam[(newp > 0.5) & (fail <= 0.1) & ~ua_bot] = "bot"              # walking paths, all served: scraping
                hfam[(rep > 0.5) & (fail > 0.3)] = "bruteforce"                   # the same endpoint failing again and again
                hfam[(rate10 >= 100) & (newp <= 0.2)] = "dos"                     # an L7 flood on few paths
                fam[sess] = hfam
                # confidence of a session verdict: how far past its calibrated threshold the stream is - against the
                # site's reference when there is one, else a margin over the threshold (never the single request's p)
                if self.site.n("tier1"):
                    p_s = np.array([1.0 - self.site.pvalue("tier1", float(v), fallback=None) for v in z1[sess]])
                else:
                    p_s = 0.5 + 0.5 / (1.0 + np.exp(-(z1[sess] - thr1)))
                p[sess] = np.maximum(p[sess], p_s)
        sev = np.array([severity(f) if a else 0.0 for f, a in zip(fam, alarm)])
        pe = np.where(alarm, p, np.minimum(p, 0.5))
        score = np.array([event_score(float(a), float(b)) for a, b in zip(pe, sev)], np.int16)
        # block_ok: a session-stream alarm (the client has a history) or a request far past threshold is block-safe;
        # a single request only mildly over the model threshold is challenged, not blocked (first-contact safety)
        # first-contact HTTP is challenged, never hard-blocked. A session alarm blocks only when the client's requests
        # are largely failing (4xx/5xx -> no reply): a benign user browsing many distinct pages (all 200) trips the
        # path-diversity ("new port") signal but gets replies, so it is challenged, not blocked; enumeration /
        # credential stuffing / scanners fail most requests and are blocked. Hard payload signatures block in the layer.
        _nr = ent.feature_names().index("src_noreply_h2")
        fail_share = xe[:, _nr] if sess.any() else np.zeros(n)
        block_ok = alarm & sess & (fail_share > 0.3)
        # global-aggregate tier: a path whose site-wide failure rate is far over its benign baseline is a distributed
        # campaign (credential stuffing / scanning spread over many low-volume IPs). The cohort is challenged (never
        # blocked), so a benign user of a briefly-hot path only sees a JS challenge. Self-calibrated from benign traffic.
        campaign = np.zeros(n, bool)
        if client is not None and status is not None:
            keys = [self._http_key(x) for x in texts]
            tt0 = np.zeros(n) if t is None else np.asarray(t, np.float64)
            failed0 = np.array([int(status[i]) >= 400 for i in range(n)])
            campaign = self.global_http.score(keys, tt0, failed0, floor=self.global_floor)
            # campaign tier (relative to each path's own history): distinct failing clients per window, CUSUM over windows
            hosts0 = [""] * n if host is None else [str(h) for h in host]
            camp = self.campaign_http.update([http_fingerprint(texts[i], hosts0[i]) for i in range(n)], client, tt0, failed0)
            member = np.array([c is not None for c in camp], bool)
            # only campaign-shaped clients are its cohort: brand new (a stuffer's address sends a request or two) or
            # failing most of the time - a user who hit one stale link or mistyped a password while a path was hot, and
            # is otherwise served, is not challenged for it (harm run: ~150 users per 1,000 per day before)
            campaign = (campaign | member) & ((ages < 3) | (fails > 0.5))
            alarm |= campaign
            fam[campaign & (fam == "benign")] = "bruteforce"
            # campaign members are left out of block_ok so the layer challenges (not blocks) them
        risk = None
        if client is not None:
            tt = np.zeros(n) if t is None else np.asarray(t, np.float64)
            hosts = ["web"] * n if host is None else list(host)
            risk = np.zeros(n)
            for i in range(n):
                port = zlib.crc32(texts[i].split(" | ")[0].encode("utf-8", "replace")) & 0xFFFF
                risk[i] = self.entities.update(str(client[i]), float(tt[i]), float(pe[i]), float(sev[i]), bool(alarm[i]))
                self.pressure.update(float(tt[i]), float(pe[i]) if alarm[i] else 0.0, fam[i] if alarm[i] else None)
                if alarm[i]:
                    self._log("http-session" if sess[i] and z[i] <= thr else "http", float(tt[i]), str(client[i]), hosts[i], port, fam[i], int(score[i]), 1,
                              texts[i][:200], soft=not bool(block_ok[i] or sess[i] or campaign[i]))
        if t is not None:
            # what the references learn: the benign tail stays in (requests just over a threshold are, at alpha, mostly
            # benign - excluding them ratchets the threshold down epoch after epoch); strong outliers and every request
            # of a client alarming again and again stay out (else a slow scraper teaches the reference to accept it)
            tt1 = np.asarray(t, np.float64)
            bad = self._sustained(client, tt1, alarm) if client is not None else set()
            for i in range(n):
                # a client alarming again and again stays out only while it also *behaves* like an attacker (most of
                # its requests failing, or a bot's rate): a person who hit a few stale links is the benign tail
                attacky = fails_s is not None and (fails_s[i] > 0.5 or rate_s[i] >= 5.0)
                if client is not None and ((attacky and (str(client[i]) in bad or self.sustained_now(str(client[i]), float(tt1[i]))))
                                           or not self._may_contribute(str(client[i]), float(tt1[i]), self.rolling["sess"].n())):
                    continue
                if z[i] <= thr_row[i] + 3.0:
                    self.rolling["http"].add(float(tt1[i]), float(z[i]))
                    r_h = self.http_ref_by_host.get(hosts_in[i])
                    if r_h is not None:
                        r_h.add(float(tt1[i]), float(z[i]))
                if client is not None and z1_s is not None and z1_s[i] <= thr_s + 3.0 and not (sess[i] and fails_s[i] > 0.3):
                    self.rolling["sess"].add(float(tt1[i]), float(z1_s[i]))
        self.counts["http"] += n
        self.counts["alarms"] += int(alarm.sum())
        return {"p_attack": p, "alarm": alarm, "family": fam, "severity": sev, "score": score, "entity_risk": risk, "block_ok": block_ok,
                "behaviour": sess | campaign,
                "threshold": thr, "threshold_source": srct, "site_pressure": self.pressure.pressure}

    # ------------------------------------------------------------ bookkeeping
    def _log(self, kind, t, src, dst, port, fam, score, tier, evidence: str = "", soft: bool = False) -> None:
        self.log.append({"kind": kind, "t": t, "src": src, "dst": dst, "port": port, "family": fam, "score": score,
                         "tier": tier, "evidence": evidence, "soft": soft})
        if len(self.log) > 50_000:
            del self.log[:10_000]

    def status(self) -> dict:
        return {"counts": dict(self.counts), "site_scores": {k: self.site.n(k) for k in ("tier1", "seq", "http")},
                "rolling_scores": {k: v.n() for k, v in self.rolling.items()}, "anomaly_reference": self.anom_s.n() if self.anom else 0,
                "evidence_reference": self.ev_ref.n(), "evidence_slow_reference": self.ev_slow_ref.n(), "dst_evidence_reference": self.dst_ev_ref.n(), "warmup_short": getattr(self, "warmup_short", False), "mondrian_groups": sum(1 for v in self.roll_groups.values() if v.n() >= self.mondrian_min),
                "thresholds": {k: self.threshold(k) for k in ("tier1",) + (("seq",) if self.seq else ()) + (("http",) if self.http else ())},
                "warm_sources": len(self.gru), "site_pressure": self.pressure.pressure, "drift": self.drift.drifted,
                "drift_guard": self.drift_events[-5:], "campaigns": self.campaign.active()[:5] + self.campaign_http.active()[:5],
                "models": {"tier1": self.tier1.meta.get("version"), "seq": self.seq.info.get("version") if self.seq else None,
                           "http": self.http.info.get("version") if self.http else None}}