File size: 75,192 Bytes
5d97275 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 | """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}}
|