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()
|