| """Campaign tier: coordinated attacks spread over many sources, each too quiet to be judged on its own. |
| |
| A residential-proxy credential-stuffing run (3,000 addresses, one guess each per half hour), a botnet brute-forcing |
| SSH from 400 machines (one try each per 20-40 minutes), a slow horizontal scan of database ports from 1,500 hosts: |
| no single source ever accumulates evidence, but the *fingerprint* they share suddenly has far more distinct failing |
| sources than it ever has. Internet background radiation (Shodan, Censys, the botnets' own scanning) is part of every |
| public site's normal; so the reference is the site's own, per fingerprint, and the test is relative to it. |
| |
| fingerprint flows: (service port, failed-connect | short-auth-session); HTTP: (host, method, path template) |
| statistic distinct sources with a failure on the fingerprint per window (5 min); the fingerprint's own benign |
| history gives its mean and spread (median / MAD over the last 24 h of windows, a pooled prior while |
| young); a one-sided CUSUM over windows - S = max(0, S + z - k) - lets a slow campaign add up over |
| hours, and a strong spike inside a window fires at once |
| members sources that failed on an active fingerprint and never succeeded on it (a user who mistyped a |
| password and then logged in is not a member; nor is anyone the site knows as normal) |
| reference only windows outside a campaign feed the history at full weight, so a campaign cannot become the |
| baseline within hours; one active window in four is admitted too, so a level that persists for many |
| hours (a growing audience, a new client population) is adopted instead of being flagged forever, and |
| the CUSUM is capped so an ended campaign stops within ~2 h |
| |
| The tier names the campaign (fingerprint, distinct sources now vs normal, since when) for the report and the operator. |
| """ |
| from __future__ import annotations |
|
|
| import math |
| import re |
| from collections import OrderedDict, deque |
|
|
| from dibaguard.flows.entity import AUTH_PORTS as _AUTH |
|
|
| AUTH_PORTS = set(_AUTH) |
| _NUMSEG = re.compile(r"/\d+(?=/|$)") |
| _HEXSEG = re.compile(r"/[0-9a-f]{8,}(?=/|$)|/[0-9a-f-]{36}(?=/|$)", re.I) |
|
|
|
|
| def http_fingerprint(text: str, host: str = "") -> str: |
| """(host, method, path template): ids -> {n} / {h}; the query string is not part of the fingerprint""" |
| head = text.split(" | ", 1)[0] |
| method, _, path = head.partition(" ") |
| path = path.split("?", 1)[0].lower() |
| path = _HEXSEG.sub("/{h}", _NUMSEG.sub("/{n}", path)) |
| return f"{host}|{method.lower()}|{path[:120]}" |
|
|
|
|
| class _Key: |
| __slots__ = ("w", "cur", "hist", "S", "active", "since", "peak", "members", "clean", "act_n") |
|
|
| def __init__(self, hist_len: int): |
| self.w = None |
| self.cur: set = set() |
| self.hist: deque = deque(maxlen=hist_len) |
| self.S = 0.0 |
| self.active = False |
| self.since = None |
| self.peak = 0 |
| self.members: "OrderedDict[str, float]" = OrderedDict() |
| self.clean: "OrderedDict[str, float]" = OrderedDict() |
| self.act_n = 0 |
|
|
|
|
| class CampaignTracker: |
| s_cap, adopt_every = 20.0, 4 |
| |
| |
| |
| young_end = False |
|
|
| def __init__(self, window_s: float = 300.0, k: float = 0.75, h: float = 10.0, spike: float = 8.0, min_sources: int = 8, |
| hist_len: int = 288, max_keys: int = 50_000, max_set: int = 20_000, member_ttl_s: float = 3600.0, warm_windows: int = 6): |
| self.W, self.k, self.h, self.spike = window_s, k, h, spike |
| self.min_sources, self.hist_len, self.max_keys, self.max_set = min_sources, hist_len, max_keys, max_set |
| self.member_ttl, self.warm = member_ttl_s, warm_windows |
| self.clean_ttl = 3600.0 |
| self.s_cap = 2.0 * h |
| self.adopt_every = 4 |
| self.keys: "OrderedDict[object, _Key]" = OrderedDict() |
| self.pooled: deque = deque(maxlen=5000) |
| self.events: list = [] |
|
|
| |
| def _stats(self, st: _Key) -> tuple[float, float, bool]: |
| """(mean, spread, own): the fingerprint's own benign history once it has `warm` windows, else the pooled prior""" |
| own = len(st.hist) >= self.warm |
| h = st.hist if own else self.pooled |
| if len(h) < self.warm: |
| return 0.0, 1.0, False |
| v = sorted(h) |
| med = v[len(v) // 2] |
| mad = sorted(abs(x - med) for x in v)[len(v) // 2] |
| sd = max(1.4826 * mad, math.sqrt(max(med, 1.0)), 1.0) |
| return float(med), float(sd), own |
|
|
| def _young_bar(self) -> float: |
| """a fingerprint without its own history fires only far above what any fingerprint shows in a window""" |
| if len(self.pooled) < 20: |
| return math.inf |
| v = sorted(self.pooled) |
| return max(2.0 * self.min_sources, 2.0 * v[min(len(v) - 1, int(0.999 * len(v)))]) |
|
|
| def _get(self, key) -> _Key: |
| st = self.keys.get(key) |
| if st is None: |
| st = _Key(self.hist_len) |
| self.keys[key] = st |
| if len(self.keys) > self.max_keys: |
| self.keys.popitem(last=False) |
| else: |
| self.keys.move_to_end(key) |
| return st |
|
|
| def _close(self, key, st: _Key, w: int, detect: bool) -> None: |
| """close the windows from st.w up to (not including) w: judge each, feed the reference with benign ones""" |
| gap = 0 |
| while st.w is not None and st.w < w and gap < self.hist_len: |
| n = len(st.cur) |
| mu, sd, ok = self._stats(st) |
| z = (n - mu) / sd |
| if detect and ok: |
| st.S = min(self.s_cap, max(0.0, st.S + z - self.k)) |
| if not st.active and st.S > self.h and n >= self.min_sources: |
| st.active, st.since = True, st.w * self.W |
| self.events.append({"key": key, "start": st.since, "sources": n, "normal": round(mu, 1)}) |
| elif st.active and st.S == 0.0: |
| st.active = False |
| self.events.append({"key": key, "end": (st.w + 1) * self.W, "peak": st.peak}) |
| st.peak = 0 |
| elif detect and st.active and self.young_end and n < 0.5 * self._young_bar(): |
| |
| st.active, st.S = False, 0.0 |
| self.events.append({"key": key, "end": (st.w + 1) * self.W, "peak": st.peak}) |
| st.peak = 0 |
| if not st.active: |
| st.hist.append(n) |
| self.pooled.append(n) |
| st.act_n = 0 |
| else: |
| st.peak = max(st.peak, n) |
| |
| |
| st.act_n = getattr(st, "act_n", 0) + 1 |
| if st.act_n % self.adopt_every == 0: |
| st.hist.append(n) |
| st.cur = set() |
| st.w += 1 |
| gap += 1 |
| st.w = w |
|
|
| def update(self, keys, srcs, t, failed, detect: bool = True) -> list: |
| """time-ordered events -> per event None or (campaign key, sources this window, normal mean) when the event is a |
| failure on an active fingerprint by a source that never succeeded on it (a campaign member)""" |
| out = [None] * len(keys) |
| for i in range(len(keys)): |
| key, src, ti = keys[i], str(srcs[i]), float(t[i]) |
| st = self._get(key) |
| w = int(ti // self.W) |
| if st.w is None: |
| st.w = w |
| elif w > st.w: |
| self._close(key, st, w, detect) |
| c = st.clean.get(src) |
| if not failed[i]: |
| if c is None: |
| c = st.clean[src] = [0, 0, ti, 0] |
| if len(st.clean) > self.max_set: |
| st.clean.popitem(last=False) |
| c[0] += 1 |
| c[2] = ti |
| c[3] = 0 |
| st.clean.move_to_end(src) |
| if c[1] < 3 or c[1] < 4 * c[0]: |
| st.members.pop(src, None) |
| continue |
| |
| |
| |
| shielded = False |
| if c is not None: |
| c[1] += 1 |
| c[3] += 1 |
| shielded = ti - c[2] <= self.clean_ttl and c[3] < 2 and (c[1] < 3 or c[1] < 4 * c[0]) |
| if len(st.cur) < self.max_set: |
| st.cur.add(src) |
| if detect and not st.active: |
| mu, sd, ok = self._stats(st) |
| young = not ok and len(self.pooled) >= self.warm and len(st.cur) >= self._young_bar() |
| if young or (ok and len(st.cur) >= self.min_sources and (len(st.cur) - mu) / sd > self.spike): |
| st.active, st.since, st.S = True, w * self.W, self.h + 1.0 |
| self.events.append({"key": key, "start": st.since, "sources": len(st.cur), "normal": round(mu, 1), "spike": True}) |
| if st.active and not shielded: |
| st.members[src] = ti |
| if len(st.members) > self.max_set: |
| st.members.popitem(last=False) |
| mu, _, _ = self._stats(st) |
| out[i] = (key, len(st.cur), mu) |
| return out |
|
|
| def credit(self, keys, srcs, t) -> None: |
| """successes seen under another fingerprint of the same service (a completed connection to a port also answers |
| for the failed-connect fingerprint of that port): they clear the source like a success on the key itself. Only |
| fingerprints already tracked are touched (no new state)""" |
| for i in range(len(keys)): |
| st = self.keys.get(keys[i]) |
| if st is None: |
| continue |
| src, ti = str(srcs[i]), float(t[i]) |
| c = st.clean.get(src) |
| if c is None: |
| c = st.clean[src] = [0, 0, ti, 0] |
| if len(st.clean) > self.max_set: |
| st.clean.popitem(last=False) |
| c[0] += 1 |
| c[2] = ti |
| c[3] = 0 |
| st.clean.move_to_end(src) |
| if c[1] < 3 or c[1] < 4 * c[0]: |
| st.members.pop(src, None) |
|
|
| def calibrate(self, keys, srcs, t, failed) -> None: |
| """benign traffic -> every fingerprint's history (no detection)""" |
| self.update(keys, srcs, t, failed, detect=False) |
|
|
| def is_member(self, key, src: str, t: float) -> bool: |
| st = self.keys.get(key) |
| return bool(st and st.active and src in st.members and t - st.members[src] <= self.member_ttl) |
|
|
| def active(self) -> list[dict]: |
| out = [] |
| for key, st in self.keys.items(): |
| if st.active: |
| mu, sd, _ = self._stats(st) |
| out.append({"key": key, "since": st.since, "members": len(st.members), "sources_this_window": len(st.cur), |
| "normal": round(mu, 1), "spread": round(sd, 1), "cusum": round(st.S, 1)}) |
| return sorted(out, key=lambda d: -d["members"]) |
|
|
