From 897828f506bae39e5eeac88ce04e10ce829b26df Mon Sep 17 00:00:00 2001 From: DeepFal Date: Mon, 19 Oct 2026 12:51:04 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E6=B5=8B=EF=BC=9A=E8=A1=8C=E4=B8=BA?= =?UTF-8?q?=E6=A3=80=E6=B5=8B=E6=94=AF=E6=8C=81=E5=8E=86=E5=8F=B2=E5=8C=BA?= =?UTF-8?q?=E9=97=B4=EF=BC=9B=E7=89=B9=E5=BE=81=E6=89=AB=E6=8F=8F=E8=87=AA?= =?UTF-8?q?=E5=8A=A8=E6=97=B6=E9=97=B4=E5=88=87=E7=89=87=EF=BC=9B=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E7=94=BB=E5=83=8F=E4=BF=9D=E7=95=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- graylog_ops/behavior.py | 17 ++++- graylog_ops/watch.py | 43 +++++++++++-- tests/test_behavior.py | 27 ++++++++ tools/backfill_behavior.py | 126 +++++++++++++++++++++++++++++++++++++ tools/backfill_gap.py | 34 ++++++---- 5 files changed, 229 insertions(+), 18 deletions(-) create mode 100644 tools/backfill_behavior.py diff --git a/graylog_ops/behavior.py b/graylog_ops/behavior.py index e80c5f7..4c70b71 100644 --- a/graylog_ops/behavior.py +++ b/graylog_ops/behavior.py @@ -134,11 +134,20 @@ def asn_info(ip, root): return {"asn": "", "org": "未知", "shared": "unknown"} +RANGE_OVERRIDE = None + + def pivot(client, series, row_groups, column_groups=None, query="*", limit=300, timeout=180000): + if RANGE_OVERRIDE: + timerange = {"type": "absolute", + "from": RANGE_OVERRIDE["from"], + "to": RANGE_OVERRIDE["to"]} + else: + timerange = {"type": "relative", "range": WINDOW} body = {"queries": [{ "id": "q1", - "timerange": {"type": "relative", "range": WINDOW}, + "timerange": timerange, "query": {"type": "elasticsearch", "query_string": query}, "filter": {"type": "or", "filters": [{"type": "stream", "id": "6ac83059a6878e26256fcacc"}]}, @@ -401,9 +410,13 @@ class BehaviorAnalyzer: except GraylogError: pass try: + extra = {} + if RANGE_OVERRIDE: + extra = {"from_ts": RANGE_OVERRIDE["from"], + "to_ts": RANGE_OVERRIDE["to"]} hist = self.client.histogram(query=q, range_seconds=self.window, interval="5m", - filter_str=self.filter_str) + filter_str=self.filter_str, **extra) from .client import normalize_histogram detail["buckets5m"] = [b["count"] for b in normalize_histogram(hist)] diff --git a/graylog_ops/watch.py b/graylog_ops/watch.py index 3aecfb9..a82cebb 100644 --- a/graylog_ops/watch.py +++ b/graylog_ops/watch.py @@ -345,7 +345,8 @@ class Store: country_code TEXT, country_name TEXT, region TEXT, city TEXT )""" ) - for column in ("asn_brand TEXT", "asn_type TEXT", "asn_icon TEXT"): + for column in ("asn_brand TEXT", "asn_type TEXT", "asn_icon TEXT", + "historical INTEGER DEFAULT 0"): try: self.conn.execute( "ALTER TABLE behavior_profiles ADD COLUMN %s" % column) @@ -497,10 +498,40 @@ class Store: return item # ---------------------------------------------------------- behaviour + def upsert_historical_profiles(self, items): + """Insert backfilled behaviour profiles; live rows win on conflict.""" + stamp = now_iso() + with self.lock, self.conn: + for item in items: + self.conn.execute( + """INSERT OR IGNORE INTO behavior_profiles + (ip, score, verdict, signals, reason_text, share_note, + behavior_type, asn, asn_org, shared_type, asn_brand, + asn_type, asn_icon, total, pct4, uri_card, uri_templates, + ua_card, host_card, coverage, peak_share, top_uri, top_ua, + last_ts, updated_at, country_code, country_name, region, + city, historical) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,1)""", + (item["ip"], item.get("score"), item.get("verdict"), + json.dumps(item.get("signals") or [], ensure_ascii=False), + item.get("reason_text"), item.get("share_note"), + item.get("behavior_type"), item.get("asn"), + item.get("asn_org"), item.get("shared_type"), + item.get("asn_brand"), item.get("asn_type"), + item.get("asn_icon"), int(item.get("total") or 0), + item.get("pct4"), item.get("uri_card"), + item.get("uri_templates"), item.get("ua_card"), + item.get("host_card"), item.get("coverage"), + item.get("peak_share"), item.get("top_uri"), + item.get("top_ua"), item.get("last_ts") or stamp, stamp, + item.get("country_code"), item.get("country_name"), + item.get("region"), item.get("city"))) + def replace_behavior_profiles(self, items): stamp = now_iso() with self.lock, self.conn: - self.conn.execute("DELETE FROM behavior_profiles") + self.conn.execute( + "DELETE FROM behavior_profiles WHERE COALESCE(historical,0)=0") for item in items: self.conn.execute( """INSERT OR REPLACE INTO behavior_profiles @@ -509,8 +540,8 @@ class Store: asn_type, asn_icon, total, pct4, uri_card, uri_templates, ua_card, host_card, coverage, peak_share, top_uri, top_ua, last_ts, updated_at, country_code, country_name, region, - city) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + city, historical) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,0)""", (item["ip"], item.get("score"), item.get("verdict"), json.dumps(item.get("signals") or [], ensure_ascii=False), item.get("reason_text"), item.get("share_note"), @@ -649,11 +680,13 @@ class Store: if beh["verdict"] != "建议封禁" or ip in wanted: continue prev = existing.get(ip) + historical = bool(beh["historical"]) if "historical" in beh.keys() else False + fallback = (beh["last_ts"] if historical and beh["last_ts"] else now) wanted[ip] = { "sources": "behavior", "feature_score": None, "behavior_score": beh["score"], - "first_banned_at": prev["first_banned_at"] if prev else now, + "first_banned_at": prev["first_banned_at"] if prev else fallback, "last_activity_at": iso(beh["last_ts"]), "fused_reason": "行为表现：" + (beh["reason_text"] or ""), } diff --git a/tests/test_behavior.py b/tests/test_behavior.py index 9ffbb27..9c22adb 100644 --- a/tests/test_behavior.py +++ b/tests/test_behavior.py @@ -99,5 +99,32 @@ class FusedTests(unittest.TestCase): store.close() +class HistoricalBehaviourTests(unittest.TestCase): + def test_historical_profile_survives_live_refresh(self): + with tempfile.TemporaryDirectory() as directory: + store = watch.Store(os.path.join(directory, "ips.db")) + try: + store.upsert_historical_profiles([{ + "ip": "198.51.100.9", "score": 80, + "verdict": "建议封禁", "signals": ["历史补测"], + "last_ts": "2026-10-18T10:00:00Z", "total": 1000, + }]) + store.replace_behavior_profiles([{ + "ip": "203.0.113.9", "score": 70, + "verdict": "仅告警（共享出口）"}]) + rows = {r["ip"]: dict(r) for r in store.conn.execute( + "SELECT * FROM behavior_profiles")} + self.assertEqual(rows["198.51.100.9"]["historical"], 1) + self.assertEqual(rows["203.0.113.9"]["historical"], 0) + + store.sync_fused() + fused = {r["ip"]: dict(r) for r in store.conn.execute( + "SELECT * FROM ip_fused")} + self.assertEqual(fused["198.51.100.9"]["first_banned_at"], + "2026-10-18T10:00:00Z") + finally: + store.close() + + if __name__ == "__main__": unittest.main() diff --git a/tools/backfill_behavior.py b/tools/backfill_behavior.py new file mode 100644 index 0000000..0b9104c --- /dev/null +++ b/tools/backfill_behavior.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""Backfill behaviour detection for a historical range (service-off gap). + +Runs the independent behaviour analyzer against an absolute time range and +stores flagged profiles with historical=1 so later live runs keep them. +""" +import argparse +import datetime +import json +import os +import sys + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, ROOT) + +from graylog_ops import behavior, watch # noqa: E402 +from graylog_ops.client import GraylogClient # 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 main(): + parser = argparse.ArgumentParser() + parser.add_argument("--from", dest="from_ts", required=True) + parser.add_argument("--to", dest="to_ts", required=True) + parser.add_argument("--db", default=os.path.join(ROOT, "data", "ips.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("结束时间必须晚于开始时间") + + cfg = json.load(open(args.config, encoding="utf-8")) + client = GraylogClient(HttpClient(cfg["url"], username=cfg["token"], + password="token", timeout=120, retries=2)) + fields = build_fields(load_profile(args.profile)) + filter_str = "streams:" + args.stream + + behavior.RANGE_OVERRIDE = {"from": utc_iso(from_dt), "to": utc_iso(to_dt)} + store = watch.Store(args.db) + analyzer = behavior.BehaviorAnalyzer(client, store, fields, filter_str, + root=ROOT, notifier=None) + try: + items = analyzer.analyze() + finally: + behavior.RANGE_OVERRIDE = None + + flagged = [item for item in items if item.get("verdict") == "建议封禁"] + for item in items: + try: + last = client.search_absolute( + query='%s:"%s"' % (fields["client_ip"], item["ip"]), + from_ts=utc_iso(from_dt), to_ts=utc_iso(to_dt), + limit=1, sort="timestamp:desc", fields=["timestamp"], + filter_str=filter_str) + stamp = (last or [{}])[0].get("timestamp") + if stamp: + item["last_ts"] = datetime.datetime.fromisoformat( + str(stamp).replace("Z", "+00:00")).astimezone( + datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + except Exception as exc: + print("时间查询 %s 失败: %s" % (item["ip"], exc), file=sys.stderr) + + print("行为违规 %d 个，建议封禁 %d 个" % (len(items), len(flagged)), + file=sys.stderr) + for item in sorted(flagged, key=lambda x: -x["score"])[:20]: + print(" %-16s 评分 %-5s %-10s %s" % ( + item["ip"], item["score"], item.get("asn_brand") or "", + item.get("reason_text") or ""), file=sys.stderr) + + if args.dry_run: + print(json.dumps({"violations": len(items), + "flagged": len(flagged), + "ips": [i["ip"] for i in flagged]}, + ensure_ascii=False, indent=2)) + store.close() + return 0 + + store.upsert_historical_profiles(items) + before = {row["ip"] for row in store.conn.execute( + "SELECT ip FROM ip_fused")} + store.sync_fused() + added = [i["ip"] for i in flagged + if i["ip"] not in before] + stamp = watch.now_iso() + if added: + with store.lock, store.conn: + store.conn.executemany( + "UPDATE ip_fused SET last_pushed_at=? WHERE ip=?", + [(stamp, ip) for ip in added]) + print(json.dumps({ + "range": [utc_iso(from_dt), utc_iso(to_dt)], + "violations": len(items), + "flagged": len(flagged), + "newly_fused": len(added), + "new_ips": sorted(added), + }, ensure_ascii=False, indent=2)) + store.close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/backfill_gap.py b/tools/backfill_gap.py index c068c9a..1478249 100644 --- a/tools/backfill_gap.py +++ b/tools/backfill_gap.py @@ -67,6 +67,8 @@ def main(): 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("--size", type=int, default=10000, + help="单次聚合最多取多少个 IP（超过会自动按时间切片）") parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() @@ -84,25 +86,35 @@ def main(): counts = {} rule_totals = {} + + def collect(rule, query, start, end): + """Query one time slice; split recursively when the terms cap is hit.""" + result = client.terms(ip_field, query=query, size=args.size, + from_ts=utc_iso(start), to_ts=utc_iso(end), + filter_str=filter_str) + terms = result.get("terms") or [] + if len(terms) >= args.size and (end - start).total_seconds() > 300: + middle = start + (end - start) / 2 + return collect(rule, query, start, middle) + collect(rule, query, middle, end) + return [(str(item.get("term") or ""), int(item.get("count") or 0)) + for item in terms if item.get("term")] + 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) + pairs = collect(rule, query, from_dt, to_dt) 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) + merged = {} + for ip, count in pairs: + merged[ip] = merged.get(ip, 0) + count + rule_totals[rule["name"]] = sum(merged.values()) + for ip, count in merged.items(): + counts.setdefault(ip, {})[rule["name"]] = count + print("规则 %-14s %6d 个 IP" % (rule["name"], len(merged)), file=sys.stderr) now_iso = watch.now_iso() store = watch.Store(args.db)