#!/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())