feat(api): Add Alertmanager native format webhook endpoint

- POST /api/v1/webhooks/alertmanager accepts Prometheus Alertmanager format
- Internal IPs (192.168.x.x, 10.x.x.x) bypass HMAC verification
- Converts Alertmanager alerts to Signal format → Redis Stream
- External IPs must use /signals with HMAC

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-24 18:43:18 +08:00
parent 2337a03dfa
commit 80b06e72a3

View File

@@ -977,6 +977,123 @@ async def receive_alert(
) from e
# =============================================================================
# Phase 10: Alertmanager 原生格式支援 (內網免 HMAC)
# =============================================================================
class AlertmanagerAlert(BaseModel):
"""Alertmanager 單一告警"""
status: str # firing, resolved
labels: dict = {}
annotations: dict = {}
startsAt: str | None = None
endsAt: str | None = None
generatorURL: str | None = None
fingerprint: str | None = None
class AlertmanagerPayload(BaseModel):
"""Alertmanager Webhook Payload"""
version: str | None = "4"
groupKey: str | None = None
status: str # firing, resolved
receiver: str | None = None
groupLabels: dict | None = {}
commonLabels: dict | None = {}
commonAnnotations: dict | None = {}
externalURL: str | None = None
alerts: list[AlertmanagerAlert]
def is_internal_ip(client_ip: str) -> bool:
"""檢查是否為內網 IP"""
import ipaddress
try:
ip = ipaddress.ip_address(client_ip)
# 私有網段: 10.x.x.x, 172.16-31.x.x, 192.168.x.x, localhost
return ip.is_private or ip.is_loopback
except ValueError:
return False
@router.post(
"/alertmanager",
response_model=SignalResponse,
summary="Phase 10: Alertmanager 原生格式 (內網免 HMAC)",
description="接收 Alertmanager Webhook內網來源免 HMAC 驗證。",
)
async def alertmanager_webhook(
request: Request,
payload: AlertmanagerPayload,
) -> SignalResponse:
"""
接收 Alertmanager 格式告警並轉換為 Signal
安全策略:
- 內網 IP (192.168.x.x, 10.x.x.x): 免 HMAC
- 外網 IP: 拒絕 (需使用 /signals 端點)
"""
# 取得客戶端 IP
client_ip = request.client.host if request.client else "unknown"
forwarded_for = request.headers.get("X-Forwarded-For", "").split(",")[0].strip()
actual_ip = forwarded_for or client_ip
# 內網檢查
if not is_internal_ip(actual_ip):
logger.warning(
"alertmanager_external_rejected",
client_ip=actual_ip,
reason="External IP must use /signals with HMAC",
)
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="External sources must use /signals endpoint with HMAC signature",
)
logger.info(
"alertmanager_webhook_received",
client_ip=actual_ip,
status=payload.status,
alert_count=len(payload.alerts),
)
# 處理每個告警
message_ids = []
for alert in payload.alerts:
if alert.status != "firing":
continue # 只處理 firing 狀態
# 轉換為 SignalPayload
severity_map = {"critical": "critical", "warning": "warning", "info": "info"}
severity = severity_map.get(
alert.labels.get("severity", "warning").lower(),
"warning"
)
signal = SignalPayload(
source="alertmanager",
alert_name=alert.labels.get("alertname", "UnknownAlert"),
severity=severity,
namespace=alert.labels.get("namespace", "default"),
target=alert.labels.get("pod", alert.labels.get("instance", "unknown")),
message=alert.annotations.get("summary", alert.annotations.get("description", "")),
labels=alert.labels,
annotations=alert.annotations,
)
try:
message_id = await produce_signal_to_stream(signal)
message_ids.append(message_id)
except Exception as e:
logger.error("alertmanager_signal_produce_error", error=str(e))
return SignalResponse(
success=True,
message_id=message_ids[0] if message_ids else None,
stream=SIGNAL_STREAM_KEY,
)
@router.get(
"/health",
summary="Webhook 健康檢查",