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