From d78eabc45814632eb6cfe99e490471929870e554 Mon Sep 17 00:00:00 2001 From: DeepFal Date: Mon, 19 Oct 2026 12:44:46 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E6=B5=8B=E8=84=9A=E6=9C=AC=EF=BC=9A?= =?UTF-8?q?=E5=9B=9E=E6=89=AB=E6=9C=8D=E5=8A=A1=E7=A6=BB=E7=BA=BF=E6=97=B6?= =?UTF-8?q?=E6=AE=B5=EF=BC=9B=E9=92=89=E9=92=89=E5=8A=A0=E7=AD=BE=E4=B8=8E?= =?UTF-8?q?=E6=8E=A8=E9=80=81=E5=A4=B1=E8=B4=A5=E9=99=90=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/backfill_gap.py | 209 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 209 insertions(+) create mode 100644 tools/backfill_gap.py diff --git a/tools/backfill_gap.py b/tools/backfill_gap.py new file mode 100644 index 0000000..c068c9a --- /dev/null +++ b/tools/backfill_gap.py @@ -0,0 +1,209 @@ +#!/usr/bin/env python3 +"""Backfill feature detection for a historical time range. + +Scans the Graylog F5 stream for every signature rule over [--from, --to], +scores the IPs exactly like the live detector and writes them into the local +SQLite database. Backfilled rows are marked as already pushed so they do not +flood the DingTalk webhook. +""" +import argparse +import datetime +import ipaddress +import json +import os +import sys +import time + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, ROOT) + +from graylog_ops import watch # noqa: E402 +from graylog_ops.asn import AsnDB # noqa: E402 +from graylog_ops.client import GraylogClient # noqa: E402 +from graylog_ops.geo import GeoIP # noqa: E402 +from graylog_ops.http import HttpClient # noqa: E402 +from graylog_ops.judge import build_fields, load_profile # noqa: E402 + +STREAM_ID = "6ac83059a6878e26256fcacc" + + +def parse_local(value): + text = value.strip() + if text.endswith("Z"): + stamp = datetime.datetime.fromisoformat(text.replace("Z", "+00:00")) + elif "T" in text and ("+" in text[10:] or text.endswith("00")): + stamp = datetime.datetime.fromisoformat(text) + else: + stamp = datetime.datetime.strptime(text, "%Y-%m-%d %H:%M").astimezone() + if stamp.tzinfo is None: + stamp = stamp.astimezone() + return stamp.astimezone(datetime.timezone.utc) + + +def utc_iso(stamp): + return stamp.strftime("%Y-%m-%dT%H:%M:%S.000Z") + + +def normalise_timestamp(value): + text = str(value or "") + if not text: + return None + try: + stamp = datetime.datetime.fromisoformat(text.replace("Z", "+00:00")) + except ValueError: + return None + return utc_iso(stamp.astimezone(datetime.timezone.utc)) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--from", dest="from_ts", required=True, + help='开始时间，例如 "2026-10-18 04:00"') + parser.add_argument("--to", dest="to_ts", required=True, + help='结束时间，例如 "2026-10-19 03:55"') + parser.add_argument("--db", default=os.path.join(ROOT, "data", "ips.db")) + parser.add_argument("--geo-db", default=os.path.join(ROOT, "data", "geo", "geo.db")) + parser.add_argument("--asn-db", default=os.path.join(ROOT, "data", "geo", "asn.db")) + parser.add_argument("--profile", default=os.path.join(ROOT, "profiles", "f5.json")) + parser.add_argument("--config", default=os.path.join(ROOT, "graylog-ops.json")) + parser.add_argument("--stream", default=STREAM_ID) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + + from_dt, to_dt = parse_local(args.from_ts), parse_local(args.to_ts) + if to_dt <= from_dt: + raise SystemExit("结束时间必须晚于开始时间") + span = int((to_dt - from_dt).total_seconds()) + + cfg = json.load(open(args.config, encoding="utf-8")) + client = GraylogClient(HttpClient(cfg["url"], username=cfg["token"], + password="token", timeout=60, retries=2)) + fields = build_fields(load_profile(args.profile)) + ip_field = fields["client_ip"] + filter_str = "streams:" + args.stream + + counts = {} + rule_totals = {} + for rule in watch.RULE_DEFS: + if rule.get("weight", 0) <= 0 and rule["name"] != "not_found": + continue + try: + query = rule["query"](fields) + result = client.terms(ip_field, query=query, size=2000, + from_ts=utc_iso(from_dt), to_ts=utc_iso(to_dt), + filter_str=filter_str) + except Exception as exc: # keep going; one broken rule must not stop us + print("规则 %s 查询失败: %s" % (rule["name"], exc), file=sys.stderr) + continue + terms = result.get("terms") or [] + rule_totals[rule["name"]] = sum(int(item.get("count") or 0) for item in terms) + for item in terms: + ip = str(item.get("term") or "") + if not ip: + continue + counts.setdefault(ip, {})[rule["name"]] = int(item.get("count") or 0) + print("规则 %-14s %6d 个 IP" % (rule["name"], len(terms)), file=sys.stderr) + + now_iso = watch.now_iso() + store = watch.Store(args.db) + geo = GeoIP(args.geo_db) if args.geo_db and os.path.exists(args.geo_db) else None + asn = AsnDB(args.asn_db) if args.asn_db and os.path.exists(args.asn_db) else None + + candidates = [] + for ip, hits in counts.items(): + try: + addr = ipaddress.ip_address(ip) + except ValueError: + continue + if not addr.is_global: + continue + total = sum(hits.values()) + tier, score, reasons = watch.classify(total, hits, window_scale=span / 900.0) + if tier != watch.TIER_MALICIOUS: + continue + candidates.append({"ip": ip, "hits": hits, "total": total, + "tier": tier, "score": score, "reasons": reasons}) + + candidates.sort(key=lambda item: (-item["score"], -item["total"], item["ip"])) + created = upgraded = extended = 0 + touched = [] + for item in candidates: + ip = item["ip"] + first = last = None + try: + asc = client.search_absolute(query='%s:"%s"' % (ip_field, ip), + from_ts=utc_iso(from_dt), to_ts=utc_iso(to_dt), + limit=1, sort="timestamp:asc", + fields=["timestamp"], filter_str=filter_str) + desc = client.search_absolute(query='%s:"%s"' % (ip_field, ip), + from_ts=utc_iso(from_dt), to_ts=utc_iso(to_dt), + limit=1, sort="timestamp:desc", + fields=["timestamp"], filter_str=filter_str) + first = normalise_timestamp((asc or [{}])[0].get("timestamp")) + last = normalise_timestamp((desc or [{}])[0].get("timestamp")) + except Exception as exc: + print("时间查询 %s 失败: %s" % (ip, exc), file=sys.stderr) + first = first or utc_iso(from_dt) + last = last or utc_iso(to_dt) + + record = { + "ip": ip, "tier": item["tier"], "score": item["score"], + "total": item["total"], "rules": item["hits"], + "reasons": item["reasons"], "first_seen": first, "last_seen": last, + "last_malicious_at": last, "risk": "high" if item["score"] >= 85 else "medium", + "window": span, + "geo": geo.lookup(ip) if geo else {}, "asn": asn.lookup(ip) if asn else {}, + } + if args.dry_run: + candidates[0] = candidates[0] # no-op + print("[dry-run] %-16s 评分 %5.1f 命中 %d 次 规则 %s 首次 %s" + % (ip, item["score"], item["total"], + ",".join(item["hits"]), first), file=sys.stderr) + continue + + existing = store.get(ip) + if existing is None: + store.upsert(record, now_iso) + created += 1 + touched.append(ip) + elif existing.get("tier") == watch.TIER_MALICIOUS: + with store.lock, store.conn: + store.conn.execute( + "UPDATE ip_stats SET first_seen=?, last_seen=?, last_malicious_at=?, " + "ever_malicious=1 WHERE ip=?", + (min(first, existing.get("first_seen") or first), + max(last, existing.get("last_seen") or last), + max(last, existing.get("last_malicious_at") or last), ip)) + extended += 1 + touched.append(ip) + elif float(existing.get("score") or 0) < item["score"]: + store.upsert(record, now_iso) + upgraded += 1 + touched.append(ip) + else: + extended += 1 + touched.append(ip) + + if not args.dry_run and touched: + store.sync_fused() + stamp = watch.now_iso() + with store.lock, store.conn: + store.conn.executemany( + "UPDATE ip_fused SET last_pushed_at=? WHERE ip=?", + [(stamp, ip) for ip in touched]) + + print(json.dumps({ + "range": [utc_iso(from_dt), utc_iso(to_dt)], + "span_hours": round(span / 3600.0, 1), + "candidates": len(candidates), + "created": created, "upgraded": upgraded, "extended": extended, + "rule_totals": rule_totals, + "top": [{"ip": c["ip"], "score": c["score"], "total": c["total"], + "rules": c["hits"]} for c in candidates[:15]], + }, ensure_ascii=False, indent=2)) + store.close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())