127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
#!/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.set_range(utc_iso(from_dt), 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.clear_range()
|
|
|
|
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())
|