diff --git a/README.md b/README.md index 4332e2a..20d9824 100644 --- a/README.md +++ b/README.md @@ -327,6 +327,12 @@ cd ../.. && python3 -m graylog_ops.geo \ [查看详情](http://179.253.234.124:12139/s/301b.../#/ip/112.32.1.211) ``` +### 自动补测 / 推送审计 / 监控 + +- **启动自动补测**：启动时读取上次成功检测时间，缺口超过 30 分钟自动后台回扫（可用 `--backfill-threshold`、`--backfill-max-gap`、`--no-auto-backfill` 控制） +- **推送审计**：`notifications` 表记录每个 IP 的推送成功/失败、错误信息；失败按 60s→120s→…→1h 指数退避重试；连续失败 3 次会记录告警（可选 `NOTIFY_ALERT_URL` 发到独立 webhook） +- **监控**：页面右上角「监控」入口，展示 Graylog 状态、检测轮次/耗时、行为分析、推送成功率、名单和补测记录；`/metrics` 输出 Prometheus 格式，可直接被任意监控系统抓取 + ### 状态指示灯 页面右上角有一个圆点，表示后端与 Graylog 的连接状态： diff --git a/docs/screenshots/monitor.png b/docs/screenshots/monitor.png new file mode 100644 index 0000000..fea4273 Binary files /dev/null and b/docs/screenshots/monitor.png differ diff --git a/graylog_ops/backfill.py b/graylog_ops/backfill.py new file mode 100644 index 0000000..e68ba16 --- /dev/null +++ b/graylog_ops/backfill.py @@ -0,0 +1,224 @@ +"""Backfill detection for a historical time window. + +Used by the startup auto-backfill thread and by tools/backfill_*.py. +""" +import datetime +import ipaddress +import time + +from . import behavior, watch +from .asn import AsnDB +from .geo import GeoIP + + +def parse_iso(value): + try: + return datetime.datetime.fromisoformat( + str(value).replace("Z", "+00:00")).astimezone(datetime.timezone.utc) + except (TypeError, ValueError): + return None + + +def iso(stamp): + return stamp.strftime("%Y-%m-%dT%H:%M:%S.000Z") + + +def _suppress(store, ips): + if not ips: + return + stamp = watch.now_iso() + with store.lock, store.conn: + store.conn.executemany( + "UPDATE ip_fused SET last_pushed_at=?, push_fail_count=0, " + "push_next_at=NULL WHERE ip=?", [(stamp, ip) for ip in ips]) + + +def feature_backfill(client, store, fields, filter_str, from_dt, to_dt, + geo=None, asn=None, size=10000): + """Signature-rule backfill with recursive time slicing.""" + ip_field = fields["client_ip"] + span = int((to_dt - from_dt).total_seconds()) + counts, rule_totals = {}, {} + + def collect(query, start, end): + result = client.terms(ip_field, query=query, size=size, + from_ts=iso(start), to_ts=iso(end), + filter_str=filter_str) + terms = result.get("terms") or [] + if len(terms) >= size and (end - start).total_seconds() > 300: + middle = start + (end - start) / 2 + return collect(query, start, middle) + collect(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: + pairs = collect(rule["query"](fields), from_dt, to_dt) + except Exception: + continue + 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 + + now_iso = watch.now_iso() + candidates = [] + for ip, hits in counts.items(): + try: + if not ipaddress.ip_address(ip).is_global: + continue + except ValueError: + 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, + "score": score, "reasons": reasons}) + + candidates.sort(key=lambda item: (-item["score"], -item["total"], item["ip"])) + touched = [] + created = upgraded = extended = 0 + for item in candidates: + ip = item["ip"] + first = last = None + try: + asc = client.search_absolute(query='%s:"%s"' % (ip_field, ip), + from_ts=iso(from_dt), to_ts=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=iso(from_dt), to_ts=iso(to_dt), + limit=1, sort="timestamp:desc", + fields=["timestamp"], filter_str=filter_str) + first = (asc or [{}])[0].get("timestamp") + last = (desc or [{}])[0].get("timestamp") + first = iso(parse_iso(first)) if parse_iso(first) else None + last = iso(parse_iso(last)) if parse_iso(last) else None + except Exception: + pass + first = first or iso(from_dt) + last = last or iso(to_dt) + record = { + "ip": ip, "tier": watch.TIER_MALICIOUS, "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 {}, + } + 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 touched: + store.sync_fused() + _suppress(store, touched) + with store.lock, store.conn: + store.conn.executemany( + """UPDATE ip_fused SET first_banned_at = ( + SELECT s.first_seen FROM ip_stats s WHERE s.ip = ip_fused.ip) + WHERE ip=? AND EXISTS ( + SELECT 1 FROM ip_stats s WHERE s.ip = ip_fused.ip + AND s.first_seen < ip_fused.first_banned_at)""", + [(ip,) for ip in touched]) + + return {"candidates": len(candidates), "created": created, + "upgraded": upgraded, "extended": extended, + "rule_totals": rule_totals, + "ips": [item["ip"] for item in candidates]} + + +def behavior_backfill(client, store, fields, filter_str, from_dt, to_dt, + root, geo=None, asn=None): + """Run the independent behaviour analyzer over an absolute range.""" + behavior.set_range(iso(from_dt), iso(to_dt)) + try: + analyzer = behavior.BehaviorAnalyzer(client, store, fields, filter_str, + root=root, notifier=None) + items = analyzer.analyze() + finally: + behavior.clear_range() + for item in items: + try: + last = client.search_absolute( + query='%s:"%s"' % (fields["client_ip"], item["ip"]), + from_ts=iso(from_dt), to_ts=iso(to_dt), limit=1, + sort="timestamp:desc", fields=["timestamp"], + filter_str=filter_str) + stamp = (last or [{}])[0].get("timestamp") + parsed = parse_iso(stamp) + if parsed: + item["last_ts"] = parsed.strftime("%Y-%m-%dT%H:%M:%SZ") + except Exception: + pass + flagged = [item for item in items if item.get("verdict") == "建议封禁"] + store.upsert_historical_profiles(items) + before = {row["ip"] for row in store.conn.execute("SELECT ip FROM ip_fused")} + store.sync_fused() + added = [item["ip"] for item in flagged if item["ip"] not in before] + _suppress(store, added) + return {"violations": len(items), "flagged": len(flagged), + "newly_fused": len(added), "new_ips": sorted(added)} + + +def auto_backfill_gap(store, client, fields, filter_str, from_dt, to_dt, + root, behavior_enabled=True): + started = time.time() + geo = asn = None + try: + import os + geo_path = os.path.join(root, "data", "geo", "geo.db") + asn_path = os.path.join(root, "data", "geo", "asn.db") + if os.path.exists(geo_path): + geo = GeoIP(geo_path) + if os.path.exists(asn_path): + asn = AsnDB(asn_path) + except Exception: + pass + summary = {"from": iso(from_dt), "to": iso(to_dt)} + try: + summary["feature"] = feature_backfill( + client, store, fields, filter_str, from_dt, to_dt, geo=geo, asn=asn) + gap_hours = (to_dt - from_dt).total_seconds() / 3600.0 + if behavior_enabled and 1 <= gap_hours <= 48: + summary["behavior"] = behavior_backfill( + client, store, fields, filter_str, from_dt, to_dt, root, + geo=geo, asn=asn) + store.incr_meta("backfill_runs_total") + store.set_meta("backfill_last_at", watch.now_iso()) + store.set_meta("backfill_last_from", iso(from_dt)) + store.set_meta("backfill_last_to", iso(to_dt)) + store.set_meta("backfill_last_added", + len(summary["feature"].get("ips") or [])) + store.set_meta("backfill_last_error", "") + except Exception as exc: + summary["error"] = str(exc) + store.set_meta("backfill_last_error", str(exc)) + summary["seconds"] = round(time.time() - started, 1) + return summary diff --git a/graylog_ops/behavior.py b/graylog_ops/behavior.py index 4c70b71..d20432a 100644 --- a/graylog_ops/behavior.py +++ b/graylog_ops/behavior.py @@ -134,15 +134,28 @@ def asn_info(ip, root): return {"asn": "", "org": "未知", "shared": "unknown"} -RANGE_OVERRIDE = None +_RANGE_LOCAL = threading.local() + + +def set_range(from_ts, to_ts): + """Set an absolute analysis range for the current thread only.""" + _RANGE_LOCAL.value = {"from": from_ts, "to": to_ts} + + +def clear_range(): + _RANGE_LOCAL.value = None + + +def current_range(): + return getattr(_RANGE_LOCAL, "value", None) def pivot(client, series, row_groups, column_groups=None, query="*", limit=300, timeout=180000): - if RANGE_OVERRIDE: + override = current_range() + if override: timerange = {"type": "absolute", - "from": RANGE_OVERRIDE["from"], - "to": RANGE_OVERRIDE["to"]} + "from": override["from"], "to": override["to"]} else: timerange = {"type": "relative", "range": WINDOW} body = {"queries": [{ @@ -367,6 +380,9 @@ class BehaviorAnalyzer: self.notifier = notifier self.last_run = None self.last_error = None + self.last_duration = None + self.last_count = 0 + self.error_count = 0 self.runs = 0 self.asn = None try: @@ -411,9 +427,10 @@ class BehaviorAnalyzer: pass try: extra = {} - if RANGE_OVERRIDE: - extra = {"from_ts": RANGE_OVERRIDE["from"], - "to_ts": RANGE_OVERRIDE["to"]} + override = current_range() + if override: + extra = {"from_ts": override["from"], + "to_ts": override["to"]} hist = self.client.histogram(query=q, range_seconds=self.window, interval="5m", filter_str=self.filter_str, **extra) @@ -623,7 +640,14 @@ class BehaviorAnalyzer: self.store.sync_fused() self.last_run = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) self.last_error = None + self.last_duration = round(time.time() - started, 2) + self.last_count = len(items) self.runs += 1 + self.store.set_meta("behavior_last_success", self.last_run) + self.store.set_meta("behavior_last_duration", self.last_duration) + self.store.set_meta("behavior_last_count", self.last_count) + self.store.set_meta("behavior_last_error", "") + self.store.incr_meta("behavior_runs_total") print("[%s] 行为分析: %d 个违规 IP，%d 个建议封禁，耗时 %.1fs" % (self.last_run, len(items), sum(1 for item in items if item["verdict"] == "建议封禁"), @@ -633,6 +657,9 @@ class BehaviorAnalyzer: return items except Exception as exc: # keep the service alive self.last_error = str(exc) + self.error_count += 1 + self.store.set_meta("behavior_last_error", str(exc)) + self.store.incr_meta("behavior_errors_total") print("[行为分析失败] %s" % exc, flush=True) return [] @@ -694,15 +721,46 @@ def notify_fused(store, notifier, root=None): if link: lines.append("[查看详情](%s)" % link) lines.append("") + ips = [row["ip"] for row in due[:20]] try: notifier.post("\n".join(lines).strip()) except Exception as exc: - # 失败时不刷屏：每 5 分钟最多提示一次；下一轮会自动重试 + store.mark_push_failed(ips, str(exc), stamp=now) + fails = store.incr_meta("push_consecutive_failures") + store.set_meta("push_last_error", str(exc)) + store.set_meta("push_last_error_at", now) + # 失败日志限流：每 5 分钟最多提示一次 global _PUSH_ERROR_AT - now = time.time() - if now - _PUSH_ERROR_AT > 300: - print("推送失败（每 5 分钟最多提示一次）: %s" % exc, flush=True) - _PUSH_ERROR_AT = now + now_epoch = time.time() + if now_epoch - _PUSH_ERROR_AT > 300: + print("推送失败（重试 #%d，每 5 分钟最多提示一次）: %s" + % (fails, exc), flush=True) + _PUSH_ERROR_AT = now_epoch + if fails == 3 or (fails >= 3 and fails % 10 == 0): + print("[推送告警] 连续 %d 次推送失败，请检查 webhook / 网络" % fails, + flush=True) + _send_alert("钉钉推送连续 %d 次失败：%s" % (fails, exc)) return 0 - store.mark_pushed([row["ip"] for row in due[:20]], now) - return len(due[:20]) + store.mark_pushed(ips, now, channel=getattr(notifier, "kind", "dingtalk")) + store.set_meta("push_consecutive_failures", 0) + store.set_meta("push_last_error", "") + return len(ips) + + +def _send_alert(text): + """Optional independent alert webhook (NOTIFY_ALERT_URL).""" + import os + url = (os.environ.get("NOTIFY_ALERT_URL") or "").strip() + if not url: + return + try: + from .notify import Notifier + alert = Notifier({ + "type": os.environ.get("NOTIFY_ALERT_TYPE", "dingtalk"), + "url": url, + "secret": os.environ.get("NOTIFY_ALERT_SECRET", ""), + "title": "推送告警", + }) + alert.post(text, title="推送告警") + except Exception as exc: + print("告警发送失败: %s" % exc, flush=True) diff --git a/graylog_ops/cli.py b/graylog_ops/cli.py index fccee6d..009b694 100644 --- a/graylog_ops/cli.py +++ b/graylog_ops/cli.py @@ -148,6 +148,12 @@ def build_parser(): watch.add_argument("--host", default="127.0.0.1", help="面板监听地址") watch.add_argument("--port", type=int, default=8088, help="面板端口，默认 8088") watch.add_argument("--db", default="data/ips.db", help="SQLite 路径") + watch.add_argument("--backfill-threshold", type=int, default=1800, + help="启动时检测缺口超过多少秒自动补测（默认 1800）") + watch.add_argument("--backfill-max-gap", type=int, default=7 * 86400, + help="自动补测允许的最大缺口秒数（默认 7 天）") + watch.add_argument("--no-auto-backfill", action="store_true", + help="启动时不自动补测历史缺口") watch.add_argument("--behavior-interval", type=int, default=900, help="行为检测刷新间隔（秒，默认 900）") watch.add_argument("--no-behavior", action="store_true", diff --git a/graylog_ops/watch.py b/graylog_ops/watch.py index 641872c..6eeb270 100644 --- a/graylog_ops/watch.py +++ b/graylog_ops/watch.py @@ -21,6 +21,7 @@ Design """ import csv +import datetime import html import io import ipaddress @@ -362,9 +363,31 @@ class Store: fused_reason TEXT, updated_at TEXT )""" ) + for column in ("push_fail_count INTEGER DEFAULT 0", + "push_next_at TEXT"): + try: + self.conn.execute( + "ALTER TABLE ip_fused ADD COLUMN %s" % column) + except sqlite3.OperationalError: + pass self.conn.execute( "CREATE INDEX IF NOT EXISTS idx_ip_fused_first " "ON ip_fused(first_banned_at DESC, ip)") + self.conn.execute( + """CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY, value TEXT + )""" + ) + self.conn.execute( + """CREATE TABLE IF NOT EXISTS notifications ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ip TEXT, channel TEXT, status TEXT, attempt INTEGER, + error TEXT, created_at TEXT + )""" + ) + self.conn.execute( + "CREATE INDEX IF NOT EXISTS idx_notifications_created " + "ON notifications(created_at DESC)") self.conn.execute( """CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -746,6 +769,85 @@ class Store: self.conn.execute("DELETE FROM ip_fused WHERE ip = ?", (ip,)) return len(wanted) + # ------------------------------------------------------------- meta + def get_meta(self, key, default=None): + with self.lock: + row = self.conn.execute( + "SELECT value FROM meta WHERE key=?", (key,)).fetchone() + return row["value"] if row else default + + def set_meta(self, key, value): + with self.lock, self.conn: + self.conn.execute( + "INSERT INTO meta(key, value) VALUES(?,?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (key, str(value))) + + def incr_meta(self, key, amount=1): + with self.lock, self.conn: + self.conn.execute( + "INSERT INTO meta(key, value) VALUES(?,?) " + "ON CONFLICT(key) DO UPDATE SET value=CAST(" + "CAST(value AS INTEGER)+? AS TEXT)", + (key, str(amount), int(amount))) + row = self.conn.execute( + "SELECT value FROM meta WHERE key=?", (key,)).fetchone() + return int(row["value"] or 0) + + def meta_dict(self): + with self.lock: + rows = self.conn.execute("SELECT key, value FROM meta").fetchall() + return {row["key"]: row["value"] for row in rows} + + # ------------------------------------------------------- notifications + def log_notification(self, ip, channel, status, attempt, error=None, + stamp=None): + stamp = stamp or now_iso() + with self.lock, self.conn: + self.conn.execute( + "INSERT INTO notifications(ip, channel, status, attempt, error, " + "created_at) VALUES(?,?,?,?,?,?)", + (ip, channel, status, int(attempt or 0), error, stamp)) + + def mark_push_failed(self, ips, error, stamp=None, base=60, cap=3600): + stamp = stamp or now_iso() + base_time = time.time() + with self.lock, self.conn: + for ip in ips: + row = self.conn.execute( + "SELECT COALESCE(push_fail_count,0) AS n FROM ip_fused " + "WHERE ip=?", (ip,)).fetchone() + fails = int(row["n"] if row else 0) + 1 + delay = min(cap, base * (2 ** (fails - 1))) + next_at = time.strftime( + "%Y-%m-%dT%H:%M:%SZ", + time.gmtime(base_time + delay)) + self.conn.execute( + "UPDATE ip_fused SET push_fail_count=?, push_next_at=? " + "WHERE ip=?", (fails, next_at, ip)) + self.conn.execute( + "INSERT INTO notifications(ip, channel, status, attempt, " + "error, created_at) VALUES(?,?,?,?,?,?)", + (ip, "dingtalk", "failed", fails, error, stamp)) + return len(ips) + + def notification_stats(self, hours=24): + since = time.strftime("%Y-%m-%dT%H:%M:%SZ", + time.gmtime(time.time() - hours * 3600)) + with self.lock: + rows = self.conn.execute( + "SELECT status, COUNT(*) AS n FROM notifications " + "WHERE created_at >= ? GROUP BY status", (since,)).fetchall() + recent = self.conn.execute( + "SELECT * FROM notifications ORDER BY id DESC LIMIT 30").fetchall() + counts = {row["status"]: int(row["n"]) for row in rows} + sent = counts.get("sent", 0) + failed = counts.get("failed", 0) + total = sent + failed + return {"sent": sent, "failed": failed, + "success_ratio": round(sent / total, 4) if total else None, + "recent": [dict(row) for row in recent]} + def due_pushes(self, active_hours=24, cooldown_hours=24): now = time.time() active_cut = time.strftime("%Y-%m-%dT%H:%M:%SZ", @@ -767,8 +869,9 @@ class Store: LEFT JOIN behavior_profiles b ON b.ip = f.ip WHERE f.last_activity_at >= ? AND (f.last_pushed_at IS NULL OR f.last_pushed_at <= ?) + AND (f.push_next_at IS NULL OR f.push_next_at <= ?) ORDER BY f.first_banned_at DESC LIMIT 50""", - (active_cut, push_cut)).fetchall() + (active_cut, push_cut, now_iso())).fetchall() items = [] for row in rows: item = dict(row) @@ -785,12 +888,18 @@ class Store: items.append(item) return items - def mark_pushed(self, ips, stamp): + def mark_pushed(self, ips, stamp, channel="dingtalk"): with self.lock, self.conn: for ip in ips: self.conn.execute( - "UPDATE ip_fused SET last_pushed_at=?, push_count=push_count+1 " - "WHERE ip=?", (stamp, ip)) + "UPDATE ip_fused SET last_pushed_at=?, " + "push_count=push_count+1, push_fail_count=0, " + "push_next_at=NULL WHERE ip=?", (stamp, ip)) + self.conn.execute( + "INSERT INTO notifications(ip, channel, status, attempt, " + "error, created_at) VALUES(?,?,?,?,?,?)", + (ip, channel, "sent", 1, None, stamp)) + self.incr_meta("push_sent_total", len(ips)) def stats(self): with self.lock: @@ -946,6 +1055,8 @@ class Detector: self.last_run = None self.last_success = None self.last_error = None + self.last_duration = None + self.error_count = 0 self.runs = 0 def _terms(self, query, window_cap=None): @@ -1232,7 +1343,13 @@ class Detector: self.last_run = stamp self.last_success = time.time() + self.last_duration = round(time.time() - started, 2) + self.last_error = None self.runs += 1 + self.store.set_meta("detect_last_success", stamp) + self.store.set_meta("detect_last_duration", self.last_duration) + self.store.set_meta("detect_last_error", "") + self.store.incr_meta("detect_rounds_total") return { "window": self.window, "request_seconds": round(time.time() - started, 2), @@ -1779,6 +1896,32 @@ a:hover{text-decoration:underline} .notes div{font-size:12.5px;color:var(--ink-2);background:var(--surface-2); border:1px solid var(--line);border-radius:9px;padding:9px 12px;line-height:1.55} .hint{color:var(--muted);font-size:12px;margin-top:11px;line-height:1.7} +.monitor-head{display:flex;align-items:center;gap:14px;margin:6px 0 18px} +.monitor-title{font-size:21px;font-weight:600;color:var(--ink);margin-right:auto} +.mgrid{display:grid;grid-template-columns:repeat(3,minmax(0,1fr)); + gap:12px;margin-bottom:18px} +@media (max-width:1100px){.mgrid{grid-template-columns:repeat(2,minmax(0,1fr))}} +@media (max-width:700px){.mgrid{grid-template-columns:1fr}} +.mcard{background:var(--surface);border:1px solid var(--line); + border-radius:14px;padding:15px 17px;box-shadow:0 1px 2px rgba(36,33,25,.03)} +.mcard h3{font-size:11.5px;color:var(--muted);font-weight:600; + letter-spacing:.06em;margin:0 0 7px} +.mbig{font-family:"Iowan Old Style",Palatino,Georgia,serif;font-size:27px; + color:var(--ink);line-height:1.15;margin-bottom:6px} +.mbig.mok{color:#3E8E63}.mbig.mbad{color:#C0503A}.mbig.mwarn{color:var(--amber)} +.mrow{display:flex;justify-content:space-between;gap:10px;font-size:12px; + color:var(--ink-2);margin-top:5px} +.mrow span:last-child{color:var(--muted);text-align:right} +.mtablewrap{overflow:auto;border:1px solid var(--line);border-radius:14px; + background:var(--surface);box-shadow:0 1px 2px rgba(36,33,25,.03)} +.mtable{width:100%;border-collapse:collapse;font-size:12.5px} +.mtable th{color:var(--muted);font-size:11.5px;font-weight:600;text-align:left; + padding:10px 14px;border-bottom:1px solid var(--line);background:var(--surface-2)} +.mtable td{padding:9px 14px;border-bottom:1px solid #EFE9DD;color:var(--ink-2)} +.mtable tr:last-child td{border-bottom:0} +.nbadge{display:inline-block;padding:2px 9px;border-radius:99px;font-size:11px} +.nbadge.ok{background:var(--green-soft,#E4EFE7);color:#2E6B4F} +.nbadge.bad{background:var(--red-soft);color:#93402D} .pager{display:flex;align-items:center;justify-content:center;gap:18px; margin:18px 0 2px;color:var(--muted);font-size:12.5px;flex-wrap:wrap} .pg-total b{color:var(--ink);font-weight:600} @@ -1895,6 +2038,7 @@ a:hover{text-decoration:underline} +
@@ -1956,7 +2100,7 @@ const RULE_TITLES = { bad_ua: '扫描器 UA', bot_ua: '爬虫 UA', not_found: '404 探测' }; const state = { q: '', items: [], page: 1, pageSize: 20, total: 0, pages: 1, - timer: null, inDetail: false, detailSamples: [] }; + timer: null, inDetail: false, inMonitor: false, detailSamples: [] }; const modal = $('#modal'); function openModal(title, text) { @@ -2020,8 +2164,9 @@ async function loadStats() {
${fmt(s.fused_both)}
双来源
${fmt(s.fused_behavior_only)}
行为新发现
`; $('#meta').innerHTML = `最近更新 ${localTime(s.updated_at)}
+ 监控 · JSON · - stats`; + metrics`; } catch (err) { $('#meta').textContent = '统计加载失败: ' + err.message; } @@ -2281,6 +2426,8 @@ function advice(item) { async function showDetail(ip) { state.inDetail = true; + state.inMonitor = false; + $('#monitorView').hidden = true; $('#listView').hidden = true; const view = $('#detailView'); view.hidden = false; @@ -2458,7 +2605,9 @@ async function showDetail(ip) { function showList() { state.inDetail = false; + state.inMonitor = false; document.title = '恶意 IP 清单'; + $('#monitorView').hidden = true; $('#detailView').hidden = true; $('#detailView').innerHTML = ''; $('#listView').hidden = false; @@ -2467,7 +2616,101 @@ function showList() { loadList(); } + +function monitorCard(title, value, rows, valueClass) { + const body = rows.map(row => + `
${esc(row[0])}${esc(row[1])}
` + ).join(''); + return `

${esc(title)}

+
${esc(value)}
${body}
`; +} + +function monitorHtml(m) { + const d = m.detect, b = m.behavior, p = m.push, l = m.list, bf = m.backfill; + const stateText = m.graylog === 'ok' ? '正常' : m.graylog === 'error' ? '失败' + : m.graylog === 'stale' ? '过期' : '未知'; + const stateCls = m.graylog === 'ok' ? 'mok' : m.graylog === 'error' ? 'mbad' + : 'mwarn'; + const ratio = p.success_ratio_24h == null ? '-' : + (p.success_ratio_24h * 100).toFixed(1) + '%'; + const cards = [ + monitorCard('Graylog 连接', stateText, [ + ['最近成功', d.last_success ? localTime(d.last_success) : '-'], + ['距上次成功', d.gap_seconds == null ? '-' : Math.round(d.gap_seconds) + ' 秒'], + ['最近失败', d.last_error ? d.last_error.slice(0, 24) : '无'], + ], stateCls), + monitorCard('特征检测', fmt(d.rounds) + ' 轮', [ + ['失败次数', fmt(d.errors)], + ['最近耗时', d.last_duration ? d.last_duration + ' 秒' : '-'], + ['检测间隔', '60 秒'], + ]), + monitorCard('行为分析', fmt(b.runs) + ' 次', [ + ['最近耗时', b.last_duration ? b.last_duration + ' 秒' : '-'], + ['最近违规', fmt(b.last_count)], + ['最近运行', b.last_success ? localTime(b.last_success) : '-'], + ], b.last_error ? 'mbad' : ''), + monitorCard('推送（24h）', ratio, [ + ['成功 / 失败', fmt(p.sent_24h) + ' / ' + fmt(p.failed_24h)], + ['连续失败', fmt(p.consecutive_failures)], + ['最近错误', p.last_error ? p.last_error.slice(0, 24) : '无'], + ], (p.consecutive_failures > 0) ? 'mbad' : 'mok'), + monitorCard('名单', fmt(l.fused_total) + ' 个', [ + ['今日新增', fmt(l.fused_new_24h)], + ['双来源', fmt(l.fused_both)], + ['行为新发现', fmt(l.fused_behavior_only)], + ]), + monitorCard('自动补测', fmt(bf.runs) + ' 次', [ + ['上次补测', bf.last_at ? localTime(bf.last_at) : '无'], + ['补测区间', bf.last_from ? (localTime(bf.last_from) + ' → ' + localTime(bf.last_to)) : '-'], + ['上次新增', fmt(bf.last_added)], + ], bf.last_error ? 'mbad' : ''), + ].join(''); + const rows = (m.recent_notifications || []).map(row => { + const cls = row.status === 'sent' ? 'ok' : 'bad'; + return `${localTime(row.created_at)} + ${esc(row.ip)} + ${row.status === 'sent' ? '成功' : '失败'} + ${esc(row.error || '-')}`; + }).join(''); + return `
+ ← 返回名单 +
运行监控
+ 每 30 秒刷新 +
+
${cards}
+
+ + + ${rows || ''} +
时间IP结果错误
暂无推送记录
+
`; +} + +async function showMonitor() { + state.inMonitor = true; + state.inDetail = false; + document.title = '运行监控'; + $('#listView').hidden = true; + $('#detailView').hidden = true; + $('#monitorView').hidden = false; + loadStatus(); + loadStats(); + await loadMonitor(); +} + +async function loadMonitor() { + const view = $('#monitorView'); + try { + const data = await jget(BASE + '/api/monitor'); + view.innerHTML = monitorHtml(data); + } catch (err) { + view.innerHTML = `← 返回名单 +
加载失败：${esc(err.message)}
`; + } +} + function route() { + if (location.hash === '#/monitor') { showMonitor(); return; } const match = location.hash.match(/^#\/ip\/(.+)$/); if (match) showDetail(decodeURIComponent(match[1])); else showList(); @@ -2524,7 +2767,8 @@ function startTimer() { clearInterval(state.timer); state.timer = setInterval(() => { loadStatus(); - if (!state.inDetail) { loadStats(); loadList(); } + if (state.inMonitor) { loadMonitor(); } + else if (!state.inDetail) { loadStats(); loadList(); } }, 30000); } window.addEventListener('hashchange', route); @@ -2584,6 +2828,133 @@ def make_handler(store, window, interval, detector=None, path_prefix=""): return self._send(200, {"ok": True}) if path == "/api/stats": return self._send(200, store.stats()) + if path in ("/metrics", "/api/monitor"): + stats = store.stats() + meta = store.meta_dict() + notif = store.notification_stats(24) + now_epoch = time.time() + + def stamp_epoch(value): + try: + return datetime.strptime( + value, "%Y-%m-%dT%H:%M:%SZ").replace( + tzinfo=timezone.utc).timestamp() + except (TypeError, ValueError): + return None + + last_success = stamp_epoch(meta.get("detect_last_success")) + last_behavior = stamp_epoch(meta.get("behavior_last_success")) + age = round(now_epoch - last_success, 1) if last_success else None + behavior_age = (round(now_epoch - last_behavior, 1) + if last_behavior else None) + if age is None: + graylog = "unknown" + elif meta.get("detect_last_error") and age > interval * 2: + graylog = "error" + elif age <= max(120, interval * 2.5): + graylog = "ok" + else: + graylog = "stale" + + payload = { + "time": now_iso(), + "graylog": graylog, + "detect": { + "rounds": int(meta.get("detect_rounds_total") or 0), + "errors": int(meta.get("detect_errors_total") or 0), + "last_success": meta.get("detect_last_success"), + "gap_seconds": age, + "last_duration": float(meta.get("detect_last_duration") or 0), + "last_error": meta.get("detect_last_error") or "", + }, + "behavior": { + "runs": int(meta.get("behavior_runs_total") or 0), + "errors": int(meta.get("behavior_errors_total") or 0), + "last_success": meta.get("behavior_last_success"), + "age_seconds": behavior_age, + "last_duration": float(meta.get("behavior_last_duration") or 0), + "last_count": int(meta.get("behavior_last_count") or 0), + "last_error": meta.get("behavior_last_error") or "", + }, + "push": { + "sent_total": int(meta.get("push_sent_total") or 0), + "failed_total": int(meta.get("push_failed_total") or 0), + "sent_24h": notif["sent"], + "failed_24h": notif["failed"], + "success_ratio_24h": notif["success_ratio"], + "consecutive_failures": int( + meta.get("push_consecutive_failures") or 0), + "last_error": meta.get("push_last_error") or "", + "last_error_at": meta.get("push_last_error_at") or "", + }, + "list": { + "fused_total": stats.get("fused_total"), + "fused_new_24h": stats.get("fused_new_24h"), + "fused_both": stats.get("fused_both"), + "fused_behavior_only": stats.get("fused_behavior_only"), + "updated_at": stats.get("updated_at"), + }, + "backfill": { + "runs": int(meta.get("backfill_runs_total") or 0), + "last_at": meta.get("backfill_last_at") or "", + "last_from": meta.get("backfill_last_from") or "", + "last_to": meta.get("backfill_last_to") or "", + "last_added": int(meta.get("backfill_last_added") or 0), + "last_error": meta.get("backfill_last_error") or "", + }, + "recent_notifications": notif["recent"], + } + if path == "/api/monitor": + return self._send(200, payload) + + lines = [ + "# HELP ipsec_detect_rounds_total Successful detection rounds", + "# TYPE ipsec_detect_rounds_total counter", + "ipsec_detect_rounds_total %d" % payload["detect"]["rounds"], + "# TYPE ipsec_detect_errors_total counter", + "ipsec_detect_errors_total %d" % payload["detect"]["errors"], + "# TYPE ipsec_detect_last_success_timestamp gauge", + "ipsec_detect_last_success_timestamp %s" + % (last_success or 0), + "# TYPE ipsec_detect_gap_seconds gauge", + "ipsec_detect_gap_seconds %s" % (age if age is not None else -1), + "# TYPE ipsec_detect_last_duration_seconds gauge", + "ipsec_detect_last_duration_seconds %s" + % payload["detect"]["last_duration"], + "# TYPE ipsec_behavior_runs_total counter", + "ipsec_behavior_runs_total %d" % payload["behavior"]["runs"], + "# TYPE ipsec_behavior_errors_total counter", + "ipsec_behavior_errors_total %d" % payload["behavior"]["errors"], + "# TYPE ipsec_behavior_last_duration_seconds gauge", + "ipsec_behavior_last_duration_seconds %s" + % payload["behavior"]["last_duration"], + "# TYPE ipsec_behavior_last_violations gauge", + "ipsec_behavior_last_violations %d" + % payload["behavior"]["last_count"], + "# TYPE ipsec_fused_total gauge", + "ipsec_fused_total %s" % (payload["list"]["fused_total"] or 0), + "# TYPE ipsec_fused_new_24h gauge", + "ipsec_fused_new_24h %s" % (payload["list"]["fused_new_24h"] or 0), + "# TYPE ipsec_fused_both gauge", + "ipsec_fused_both %s" % (payload["list"]["fused_both"] or 0), + "# TYPE ipsec_push_sent_total counter", + "ipsec_push_sent_total %d" % payload["push"]["sent_total"], + "# TYPE ipsec_push_failed_total counter", + "ipsec_push_failed_total %d" % payload["push"]["failed_total"], + "# TYPE ipsec_push_consecutive_failures gauge", + "ipsec_push_consecutive_failures %d" + % payload["push"]["consecutive_failures"], + "# TYPE ipsec_push_success_ratio_24h gauge", + "ipsec_push_success_ratio_24h %s" + % (payload["push"]["success_ratio_24h"] + if payload["push"]["success_ratio_24h"] is not None else -1), + "# TYPE ipsec_graylog_up gauge", + "ipsec_graylog_up %d" % (1 if graylog == "ok" else 0), + "", + ] + return self._send(200, "\n".join(lines), + content_type="text/plain; version=0.0.4; charset=utf-8") + if path == "/api/status": status = {"graylog": "unknown", "last_run": None, "last_error": None, "age_seconds": None} @@ -2838,6 +3209,47 @@ def run(ctx, args): # Fused block list: existing feature detections first (historical rows keep # their original discovery time), then the behaviour thread keeps adding. store.sync_fused() + filter_str = ("streams:" + ctx.stream_id) if ctx.stream_id else None + + def auto_backfill_worker(from_dt, to_dt): + from .backfill import auto_backfill_gap + summary = auto_backfill_gap( + store, ctx.client, fields, filter_str, from_dt, to_dt, + os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + print("[自动补测] %s" % json.dumps(summary, ensure_ascii=False), + flush=True) + + auto_enabled = not getattr(args, "no_auto_backfill", False) + threshold = int(getattr(args, "backfill_threshold", 1800) or 0) + max_gap = int(getattr(args, "backfill_max_gap", 7 * 86400) or 0) + if (auto_enabled and threshold > 0 + and not getattr(args, "serve_only", False)): + from .backfill import parse_iso + last_meta = store.get_meta("detect_last_success") + if not last_meta: + row = store.conn.execute( + "SELECT MAX(updated_at) AS t FROM ip_stats").fetchone() + last_meta = row["t"] if row and row["t"] else None + last_dt = parse_iso(last_meta) if last_meta else None + if last_dt is not None: + now_dt = datetime.now(timezone.utc) + gap = (now_dt - last_dt).total_seconds() + if gap > max_gap: + print("检测缺口 %.1f 小时，超过自动补测上限 %d 小时，已跳过" + % (gap / 3600.0, max_gap // 3600), flush=True) + elif gap > threshold: + print("检测缺口 %.1f 小时，后台自动补测 %s → %s" + % (gap / 3600.0, last_dt.strftime("%m-%d %H:%M"), + now_dt.astimezone().strftime("%m-%d %H:%M")), + flush=True) + threading.Thread(target=auto_backfill_worker, + args=(last_dt, now_dt), daemon=True).start() + else: + print("检测连续（上次成功 %s），无需补测" + % last_dt.strftime("%m-%d %H:%M"), flush=True) + else: + print("没有历史检测记录，跳过自动补测", flush=True) + behavior_analyzer = None if not getattr(args, "no_behavior", False): behavior_analyzer = BehaviorAnalyzer( @@ -2915,9 +3327,15 @@ def run(ctx, args): print(" 封禁名单: %d 个 -> %s" % (len(chosen), args.export), flush=True) except GraylogError as exc: detector.last_error = str(exc) + detector.error_count += 1 + detector.store.set_meta("detect_last_error", str(exc)) + detector.store.incr_meta("detect_errors_total") print("[%s] 检测失败: %s" % (now_iso(), exc), flush=True) except Exception as exc: detector.last_error = str(exc) + detector.error_count += 1 + detector.store.set_meta("detect_last_error", str(exc)) + detector.store.incr_meta("detect_errors_total") print("[%s] 检测异常: %s" % (now_iso(), exc), flush=True) time.sleep(max(5, int(args.interval))) except KeyboardInterrupt: diff --git a/tests/test_behavior.py b/tests/test_behavior.py index 9c22adb..986b6ba 100644 --- a/tests/test_behavior.py +++ b/tests/test_behavior.py @@ -8,6 +8,7 @@ HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.dirname(HERE)) from graylog_ops import watch # noqa: E402 +from graylog_ops.behavior import notify_fused # noqa: E402 NOW = "2026-10-17T16:00:00Z" @@ -99,6 +100,54 @@ class FusedTests(unittest.TestCase): store.close() +class PushRetryTests(unittest.TestCase): + def test_failure_backoff_and_audit(self): + with tempfile.TemporaryDirectory() as directory: + store = watch.Store(os.path.join(directory, "ips.db")) + try: + store.replace_behavior_profiles([_behavior("198.51.100.8")]) + store.sync_fused() + with store.lock, store.conn: + store.conn.execute( + "UPDATE ip_fused SET last_pushed_at=NULL WHERE ip=?", + ("198.51.100.8",)) + + class Failing: + kind = "dingtalk" + def link_for(self, ip): return "" + def post(self, text, title=None): raise RuntimeError("boom") + + notify_fused(store, Failing()) + notif = store.notification_stats(24) + self.assertEqual(notif["failed"], 1) + self.assertEqual(notif["sent"], 0) + row = store.fused_row("198.51.100.8") + self.assertEqual(row["push_fail_count"], 1) + self.assertTrue(row["push_next_at"]) + self.assertEqual(store.due_pushes(24, 24), []) # backoff active + self.assertEqual(store.get_meta("push_consecutive_failures"), "1") + + with store.lock, store.conn: + store.conn.execute( + "UPDATE ip_fused SET push_next_at=NULL WHERE ip=?", + ("198.51.100.8",)) + + class Working: + kind = "dingtalk" + def link_for(self, ip): return "" + def post(self, text, title=None): return None + + notify_fused(store, Working()) + row = store.fused_row("198.51.100.8") + self.assertEqual(row["push_fail_count"], 0) + self.assertIsNone(row["push_next_at"]) + self.assertEqual(store.get_meta("push_consecutive_failures"), "0") + notif = store.notification_stats(24) + self.assertEqual(notif["sent"], 1) + finally: + store.close() + + class HistoricalBehaviourTests(unittest.TestCase): def test_historical_profile_survives_live_refresh(self): with tempfile.TemporaryDirectory() as directory: diff --git a/tests/test_watch.py b/tests/test_watch.py index 8b2092d..ad0a25e 100644 --- a/tests/test_watch.py +++ b/tests/test_watch.py @@ -341,6 +341,53 @@ class PathPrefixTests(unittest.TestCase): store.close() +class MetricsTests(unittest.TestCase): + def test_metrics_and_monitor(self): + with tempfile.TemporaryDirectory() as directory: + store = watch.Store(os.path.join(directory, "ips.db")) + now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + store.set_meta("detect_last_success", now) + store.set_meta("detect_last_error", "") + store.incr_meta("detect_rounds_total", 3) + store.incr_meta("detect_errors_total", 1) + store.set_meta("behavior_last_success", now) + store.set_meta("behavior_last_duration", 12.5) + store.set_meta("behavior_last_count", 7) + store.incr_meta("behavior_runs_total", 2) + store.incr_meta("push_sent_total", 5) + store.incr_meta("push_failed_total", 2) + store.upsert({"ip": "203.0.113.9", "tier": watch.TIER_MALICIOUS, + "score": 91.0, "total": 120, "window": 3600, + "rules": {"probe_java": 2}, + "reasons": ["探测"]}, "2026-10-17T16:00:00Z") + store.sync_fused() + server = ThreadingHTTPServer(("127.0.0.1", 0), + watch.make_handler(store, 3600, 60)) + threading.Thread(target=server.serve_forever, daemon=True).start() + base = "http://127.0.0.1:%d" % server.server_address[1] + try: + with urllib.request.urlopen(base + "/metrics", timeout=5) as response: + text = response.read().decode("utf-8") + self.assertIn("ipsec_detect_rounds_total 3", text) + self.assertIn("ipsec_detect_errors_total 1", text) + self.assertIn("ipsec_behavior_last_duration_seconds 12.5", text) + self.assertIn("ipsec_fused_total 1", text) + self.assertIn("ipsec_push_sent_total 5", text) + self.assertIn("ipsec_graylog_up 1", text) + + with urllib.request.urlopen(base + "/api/monitor", timeout=5) as response: + payload = json.loads(response.read().decode("utf-8")) + self.assertEqual(payload["graylog"], "ok") + self.assertEqual(payload["detect"]["rounds"], 3) + self.assertEqual(payload["behavior"]["last_duration"], 12.5) + self.assertEqual(payload["push"]["sent_total"], 5) + self.assertEqual(payload["list"]["fused_total"], 1) + finally: + server.shutdown() + server.server_close() + store.close() + + class PaginationTests(unittest.TestCase): def test_pagination_and_search(self): with tempfile.TemporaryDirectory() as directory: diff --git a/tools/backfill_behavior.py b/tools/backfill_behavior.py index 0b9104c..01f3080 100644 --- a/tools/backfill_behavior.py +++ b/tools/backfill_behavior.py @@ -59,14 +59,14 @@ def main(): 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)} + 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.RANGE_OVERRIDE = None + behavior.clear_range() flagged = [item for item in items if item.get("verdict") == "建议封禁"] for item in items: