109 lines
4.2 KiB
Python
109 lines
4.2 KiB
Python
# ============================================================
|
||
# 审计完整性告警通知 v1（moduleId: core-audit-notify, 可重生 ✅）
|
||
# plan.md §3.6 / 矩阵「全审计」剩余项：告警通知渠道
|
||
# build_alerts 聚合出告警后，本模块按配置渠道投递（console/log/jsonl/webhook），
|
||
# critical/warning 默认投递、info 可选；同告警码+消息防抖（时间窗内不重复发）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger("aps.audit.notify")
|
||
|
||
# 防抖窗口（秒）：同 code+message 在该窗口内只投递一次，避免 gov/audit 轮询刷屏
|
||
_DEDUP_WINDOW_SECONDS = float(os.environ.get("APS_AUDIT_ALERT_DEDUP_SECONDS") or "300")
|
||
# 默认投递渠道：console（日志）；可配 jsonl:路径 或 webhook:url（多个用逗号分隔）
|
||
_DEFAULT_CHANNELS = (os.environ.get("APS_AUDIT_ALERT_CHANNELS") or "console").strip()
|
||
# info 级告警是否投递（默认否，仅 critical/warning）
|
||
_INCLUDE_INFO = (os.environ.get("APS_AUDIT_ALERT_INCLUDE_INFO") or "0") == "1"
|
||
|
||
_seen: dict[tuple[str, str], float] = {}
|
||
|
||
|
||
def _channel_list() -> list[str]:
|
||
return [c.strip() for c in _DEFAULT_CHANNELS.split(",") if c.strip()]
|
||
|
||
|
||
def _deliver_console(alert: dict[str, Any]) -> bool:
|
||
level = logging.CRITICAL if alert.get("severity") == "critical" else logging.WARNING
|
||
logger.log(level, "[audit-alert] %s %s: %s", alert.get("code"), alert.get("severity"), alert.get("message"))
|
||
return True
|
||
|
||
|
||
def _deliver_jsonl(alert: dict[str, Any], target: str) -> bool:
|
||
path = Path(target)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with path.open("a", encoding="utf-8") as fh:
|
||
fh.write(json.dumps(alert, ensure_ascii=False, sort_keys=True) + "\n")
|
||
return True
|
||
|
||
|
||
def _deliver_webhook(alert: dict[str, Any], url: str, timeout: float = 8.0) -> bool:
|
||
payload = json.dumps(alert, ensure_ascii=False).encode("utf-8")
|
||
req = urllib.request.Request(url, data=payload, headers={"Content-Type": "application/json"})
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310（url 来自显式配置）
|
||
return resp.status < 400
|
||
except Exception:
|
||
logger.exception("audit alert webhook failed: %s", url)
|
||
return False
|
||
|
||
|
||
def _should_send(alert: dict[str, Any]) -> bool:
|
||
severity = str(alert.get("severity") or "info")
|
||
if severity == "info" and not _INCLUDE_INFO:
|
||
return False
|
||
code = str(alert.get("code") or "?")
|
||
message = str(alert.get("message") or "")
|
||
now = time.monotonic()
|
||
key = (code, message)
|
||
last = _seen.get(key, -1e9)
|
||
if now - last < _DEDUP_WINDOW_SECONDS:
|
||
return False
|
||
_seen[key] = now
|
||
return True
|
||
|
||
|
||
def notify_alerts(alerts: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""按配置渠道投递审计告警；返回投递统计（尽力而为，不抛异常）。
|
||
|
||
Returns: {"delivered": int, "skipped": int, "channels": [...]}
|
||
"""
|
||
channels = _channel_list()
|
||
if not channels:
|
||
channels = ["console"]
|
||
delivered = 0
|
||
skipped = 0
|
||
for alert in alerts or []:
|
||
if not _should_send(alert):
|
||
skipped += 1
|
||
continue
|
||
ok = True
|
||
for channel in channels:
|
||
try:
|
||
if channel == "console":
|
||
ok = _deliver_console(alert) and ok
|
||
elif channel.startswith("jsonl:"):
|
||
ok = _deliver_jsonl(alert, channel[len("jsonl:"):]) and ok
|
||
elif channel.startswith("webhook:"):
|
||
ok = _deliver_webhook(alert, channel[len("webhook:"):]) and ok
|
||
else:
|
||
logger.warning("unknown audit alert channel: %s", channel)
|
||
except Exception:
|
||
logger.exception("audit alert delivery failed on channel %s", channel)
|
||
ok = False
|
||
if ok:
|
||
delivered += 1
|
||
return {"delivered": delivered, "skipped": skipped, "channels": channels}
|
||
|
||
|
||
def reset_notify_state() -> None:
|
||
"""清空防抖缓存（测试隔离用）。"""
|
||
_seen.clear()
|