from __future__ import annotations import asyncio import hashlib import json import re import urllib.error import urllib.parse import urllib.request import uuid from pathlib import Path from typing import Any from src.core.config import settings from src.core.logging import get_logger from src.utils.timezone import now_taipei logger = get_logger("awoooi.agent99_sre_bridge") AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS = 600 _SENSITIVE_KEY_RE = re.compile( r"(token|secret|password|passwd|authorization|cookie|session|private[_-]?key)", re.IGNORECASE, ) _SAFE_TOKEN_RE = re.compile(r"[^A-Za-z0-9_.-]+") def _safe_text(value: Any, *, max_length: int = 1200) -> str: text = str(value or "") text = " ".join(text.split()) if len(text) > max_length: return text[: max_length - 14] + "...(truncated)" return text def _safe_label_value(key: str, value: Any) -> str: if _SENSITIVE_KEY_RE.search(key): return "[redacted]" return _safe_text(value, max_length=240) def _safe_file_token(value: str) -> str: safe = _SAFE_TOKEN_RE.sub("_", value).strip("._-") return (safe or "alert")[:96] def _safe_url_for_log(value: str) -> str: if not value: return "" parsed = urllib.parse.urlsplit(value) return urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", "")) def _alert_text( *, alertname: str, severity: str, target_resource: str, namespace: str, message: str, labels: dict[str, Any], annotations: dict[str, Any], ) -> str: return " ".join( [ alertname, severity, target_resource, namespace, message, " ".join(f"{k}={v}" for k, v in labels.items()), " ".join(f"{k}={v}" for k, v in annotations.items()), ] ).lower() def resolve_agent99_alert_kind( *, alertname: str, severity: str, target_resource: str, namespace: str, message: str, labels: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None, ) -> str: text = _alert_text( alertname=alertname, severity=severity, target_resource=target_resource, namespace=namespace, message=message, labels=labels or {}, annotations=annotations or {}, ) if re.search( r"provider_freshness_signal|source_provider_freshness_triage|" r"raw_signal_normalized|controlled_runtime_candidate|freshness|" r"last_seen|last seen|ingestion", text, ): return "provider_freshness_signal" if re.search( r"wazuh|siem|security|intrusion|malware|auth_failed|bruteforce|" r"privilege|vulnerability|\bcve\b|\bioc\b|suspicious|unauthorized", text, ): return "security_health" if re.search(r"backup|snapshot|restore drill|restore_drill|backup cron", text): return "backup_health" if re.search( r"postgres|postgresql|mysql|mariadb|database|db_connection|" r"pg_isready|connection pool|replication lag", text, ): return "database_health" if re.search( r"alertchain|alert_chain|alert chain|alertmanager.*(broken|down|failed)|" r"monitoring pipeline|notification pipeline", text, ): return "alert_chain_health" if re.search( r"reboot-auto-recovery|reboot_auto_recovery|reboot recovery|" r"cold[-_ ]start(?:[-_ ]gate)?|full[-_ ]stack[-_ ]cold[-_ ]start|" r"host_down|hostdown|host_reboot|vm_not_running|powered_off|" r"all_required_hosts_not_in_10_minute_reboot_window", text, ): return "host_recovery" if re.search(r"502|site_502|public_route|website|route_down|nginx upstream", text): return "public_route_502" if re.search(r"cpu|load|memory|mem_|disk|pressure|performance", text): return "performance_pressure" if re.search(r"harbor|registry", text): return "harbor_registry" if re.search( r"\bci/?cd\b|\bcicd\b|build-and-deploy|build_and_deploy|" r"deploy(?:ment)?[_ -](?:failed|failure)|build[_ -](?:failed|failure)|" r"部署失敗|構建失敗", text, ): return "cicd_failure" if re.search( r"k3s|imagepullbackoff|errimagepull|crashloopbackoff|pod|deployment", text ): return "awoooi_k3s" if re.search( r"missing_key|e_ai_missing_key|application_ai_config_error|" r"locale guard|ai support", text, ): return "application_ai_config_error" return "monitoring_alert" def _suggested_mode_for_kind(kind: str) -> str | None: return { "provider_freshness_signal": "ProviderFreshness", "public_route_502": "Recover", "performance_pressure": "Perf", "backup_health": "BackupCheck", "security_health": "SecurityTriage", "database_health": "Status", "alert_chain_health": "Status", "harbor_registry": "HarborRepair", "awoooi_k3s": "AwoooRepair", "host_recovery": "Recover", "application_ai_config_error": "Status", "cicd_failure": "Status", "monitoring_alert": "Status", }.get(kind) def _resolution_policy_for_kind(kind: str) -> str: if kind in { "provider_freshness_signal", "public_route_502", "performance_pressure", "backup_health", "harbor_registry", "awoooi_k3s", "host_recovery", }: return "mode_verifier_can_resolve" return "external_source_receipt_required" def _agent99_route_group_key( *, kind: str, target_resource: str, host: str, ) -> str: parts = ( kind.strip().lower() or "monitoring_alert", target_resource.strip().lower() or "unknown", host.strip().lower() or "no-host", ) return ":".join(_safe_file_token(part) for part in parts) def agent99_sre_single_flight_key(fingerprint: str) -> str: digest = hashlib.sha256(fingerprint.encode("utf-8")).hexdigest() return f"agent99:sre_dispatch:single_flight:{digest}" async def try_acquire_agent99_sre_single_flight( fingerprint: str, alert_id: str, *, ttl_seconds: int = AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS, ) -> bool: """Grant one Agent99 dispatch per fingerprint inside the controlled window.""" if not fingerprint: return True try: from src.core.redis_client import get_redis redis = get_redis() acquired = await redis.set( agent99_sre_single_flight_key(fingerprint), alert_id, ex=ttl_seconds, nx=True, ) return bool(acquired) except Exception as exc: logger.warning( "agent99_sre_single_flight_failed_fail_open", fingerprint=fingerprint, alert_id=alert_id, error=str(exc), ) return True async def release_agent99_sre_single_flight(fingerprint: str, alert_id: str) -> None: """Release only the caller-owned lock when transport did not accept the alert.""" if not fingerprint: return try: from src.core.redis_client import get_redis redis = get_redis() await redis.eval( """ if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) end return 0 """, 1, agent99_sre_single_flight_key(fingerprint), alert_id, ) except Exception as exc: logger.warning( "agent99_sre_single_flight_release_failed", fingerprint=fingerprint, alert_id=alert_id, error=str(exc), ) def build_agent99_sre_alert( *, alert_id: str, alertname: str, severity: str, namespace: str, target_resource: str, message: str, labels: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None, fingerprint: str = "", alert_category: str = "", notification_type: str = "", source_url: str | None = None, ) -> dict[str, Any]: labels = labels or {} annotations = annotations or {} normalized_severity = (severity or "warning").lower() kind = resolve_agent99_alert_kind( alertname=alertname, severity=normalized_severity, target_resource=target_resource, namespace=namespace, message=message, labels=labels, annotations=annotations, ) suggested_mode = _suggested_mode_for_kind(kind) host = str(labels.get("host") or labels.get("node") or labels.get("instance") or "") if ":" in host: host = host.split(":", 1)[0] service = str( labels.get("service") or labels.get("app") or labels.get("component") or target_resource or "awoooi" ) route_group_key = _agent99_route_group_key( kind=kind, target_resource=target_resource, host=host, ) safe_labels = [ "alertmanager", "awoooi-api", f"alertname={_safe_label_value('alertname', alertname)}", f"namespace={_safe_label_value('namespace', namespace)}", f"target={_safe_label_value('target', target_resource)}", ] if fingerprint: safe_labels.append( f"fingerprint={_safe_label_value('fingerprint', fingerprint)}" ) for key, value in sorted(labels.items()): safe_labels.append( f"{_safe_text(key, max_length=80)}={_safe_label_value(key, value)}" ) if kind == "provider_freshness_signal": safe_labels.extend(["provider-freshness", "no-false-green"]) instruction = None if kind == "provider_freshness_signal": instruction = ( "Provider freshness candidate required; require providerWindow,lastSeen," "directReference,verifierReadback; mark missing,expired,timeout," "no_false_green; do not switch provider, call paid model, edit env, or reload." ) payload: dict[str, Any] = { "id": f"awoooi-alertmanager-{_safe_file_token(alert_id)}", "source": "awoooi-api-alertmanager", "severity": normalized_severity, "kind": kind, "service": _safe_text(service, max_length=160), "host": _safe_text(host, max_length=160), "title": _safe_text(f"Alertmanager {alertname}", max_length=240), "message": _safe_text(message, max_length=2000), "labels": safe_labels, "force": True, "controlledApply": suggested_mode in { "Recover", "StartVMs", "HarborRepair", "AwoooRepair", "Perf", "LoadShed", }, "createdAt": now_taipei().isoformat(), "routing": { "schemaVersion": "agent99_alert_route_v1", "kind": kind, "suggestedMode": suggested_mode or "Status", "routeSource": "structured_kind", "groupKey": route_group_key, "correlationKey": route_group_key, "sourceFingerprint": _safe_text(fingerprint, max_length=160), "singleFlightWindowSeconds": AGENT99_SRE_SINGLE_FLIGHT_TTL_SECONDS, }, "resolution": { "schemaVersion": "agent99_source_event_resolution_v1", "policy": _resolution_policy_for_kind(kind), "required": True, }, "awoooi": { "alertId": _safe_text(alert_id, max_length=160), "alertname": _safe_text(alertname, max_length=160), "namespace": _safe_text(namespace, max_length=160), "targetResource": _safe_text(target_resource, max_length=240), "fingerprint": _safe_text(fingerprint, max_length=160), "alertCategory": _safe_text(alert_category, max_length=160), "notificationType": _safe_text(notification_type, max_length=160), "sourceUrl": _safe_text(source_url or "", max_length=600), }, } if suggested_mode: payload["suggestedMode"] = suggested_mode if instruction: payload["instruction"] = instruction return payload def write_agent99_sre_alert( payload: dict[str, Any], *, inbox_path: str | None = None, ) -> Path | None: target_dir_text = ( inbox_path if inbox_path is not None else settings.AGENT99_SRE_ALERT_INBOX_PATH ) if not target_dir_text: logger.info( "agent99_sre_bridge_disabled", reason="AGENT99_SRE_ALERT_INBOX_PATH not configured", alert_id=payload.get("id"), kind=payload.get("kind"), ) return None target_dir = Path(target_dir_text) target_dir.mkdir(parents=True, exist_ok=True) alert_id = _safe_file_token(str(payload.get("id") or uuid.uuid4().hex)) final_path = target_dir / f"{alert_id}.json" tmp_path = target_dir / f".{alert_id}.{uuid.uuid4().hex}.tmp" tmp_path.write_text( json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8", ) tmp_path.replace(final_path) logger.info( "agent99_sre_alert_written", path=str(final_path), alert_id=payload.get("id"), kind=payload.get("kind"), ) return final_path def post_agent99_sre_alert( payload: dict[str, Any], *, relay_url: str | None = None, relay_token: str | None = None, timeout_seconds: float | None = None, ) -> dict[str, Any] | None: target_url = ( relay_url if relay_url is not None else settings.AGENT99_SRE_ALERT_RELAY_URL ) if not target_url: return None token = ( relay_token if relay_token is not None else settings.AGENT99_SRE_ALERT_RELAY_TOKEN ) timeout = ( timeout_seconds if timeout_seconds is not None else settings.AGENT99_SRE_ALERT_RELAY_TIMEOUT_SECONDS ) data = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8") headers = { "Content-Type": "application/json; charset=utf-8", "User-Agent": "awoooi-agent99-sre-bridge/1", } if token: headers["X-Agent99-Relay-Token"] = token request = urllib.request.Request( target_url, data=data, headers=headers, method="POST" ) try: with urllib.request.urlopen(request, timeout=timeout) as response: status = int(response.getcode()) body = response.read(4096).decode("utf-8", errors="replace") except urllib.error.HTTPError as exc: body = exc.read(4096).decode("utf-8", errors="replace") raise RuntimeError(f"Agent99 relay HTTP {exc.code}: {body[:300]}") from exc if status >= 400: raise RuntimeError(f"Agent99 relay HTTP {status}: {body[:300]}") logger.info( "agent99_sre_alert_relay_posted", relay_url=_safe_url_for_log(target_url), alert_id=payload.get("id"), kind=payload.get("kind"), status=status, ) return {"status": status, "body": body} def dispatch_agent99_sre_alert(payload: dict[str, Any]) -> str: receipt = dispatch_agent99_sre_alert_with_receipt(payload) return str(receipt["transport"]) def dispatch_agent99_sre_alert_with_receipt( payload: dict[str, Any], ) -> dict[str, Any]: """Dispatch one alert and retain only public-safe acceptance evidence.""" alert_id = _safe_text(payload.get("id"), max_length=160) kind = _safe_text(payload.get("kind"), max_length=120) if settings.AGENT99_SRE_ALERT_RELAY_URL: raw_receipt = post_agent99_sre_alert(payload) or {} response_payload: dict[str, Any] = {} try: parsed = json.loads(str(raw_receipt.get("body") or "")) if isinstance(parsed, dict): response_payload = parsed except json.JSONDecodeError: response_payload = {} http_status = int(raw_receipt.get("status") or 0) accepted = bool(200 <= http_status < 300 and response_payload.get("ok") is True) inbox_triggered = bool(response_payload.get("inboxTriggered") is True) return { "schema_version": "agent99_sre_dispatch_receipt_v1", "status": ( "accepted_inbox_triggered" if accepted and inbox_triggered else "accepted_inbox_pending" if accepted else "relay_response_not_accepted" ), "transport": "relay", "alert_id": alert_id, "kind": kind, "http_status": http_status, "accepted": accepted, "inbox_triggered": inbox_triggered, "stores_raw_response": False, } if settings.AGENT99_SRE_ALERT_INBOX_PATH: written = write_agent99_sre_alert(payload) if written is not None: return { "schema_version": "agent99_sre_dispatch_receipt_v1", "status": "file_written", "transport": "file", "alert_id": alert_id, "kind": kind, "accepted": True, "inbox_triggered": False, "stores_raw_response": False, } logger.info( "agent99_sre_bridge_disabled", reason="no relay URL or mounted inbox path configured", alert_id=payload.get("id"), kind=payload.get("kind"), ) return { "schema_version": "agent99_sre_dispatch_receipt_v1", "status": "bridge_disabled", "transport": "disabled", "alert_id": alert_id, "kind": kind, "accepted": False, "inbox_triggered": False, "stores_raw_response": False, } async def bridge_alertmanager_to_agent99( *, alert_id: str, alertname: str, severity: str, namespace: str, target_resource: str, message: str, labels: dict[str, Any] | None = None, annotations: dict[str, Any] | None = None, fingerprint: str = "", alert_category: str = "", notification_type: str = "", source_url: str | None = None, ) -> dict[str, Any]: single_flight_fingerprint = "" single_flight_acquired = False try: payload = build_agent99_sre_alert( alert_id=alert_id, alertname=alertname, severity=severity, namespace=namespace, target_resource=target_resource, message=message, labels=labels, annotations=annotations, fingerprint=fingerprint, alert_category=alert_category, notification_type=notification_type, source_url=source_url, ) routing = ( payload.get("routing") if isinstance(payload.get("routing"), dict) else {} ) single_flight_fingerprint = fingerprint or str( routing.get("correlationKey") or "" ) single_flight_acquired = await try_acquire_agent99_sre_single_flight( single_flight_fingerprint, alert_id, ) if not single_flight_acquired: logger.info( "agent99_sre_dispatch_single_flight_suppressed", alert_id=alert_id, alertname=alertname, fingerprint=single_flight_fingerprint, group_key=routing.get("groupKey"), ) return { "status": "suppressed", "reason": "single_flight_duplicate", "fingerprint": single_flight_fingerprint, "groupKey": routing.get("groupKey"), } dispatch_receipt = await asyncio.to_thread( dispatch_agent99_sre_alert_with_receipt, payload ) if not dispatch_receipt.get("accepted"): await release_agent99_sre_single_flight(single_flight_fingerprint, alert_id) single_flight_acquired = False return { "status": "failed", "reason": dispatch_receipt.get("status") or "dispatch_not_accepted", "dispatchReceipt": dispatch_receipt, "fingerprint": single_flight_fingerprint, "groupKey": routing.get("groupKey"), } return { "status": "dispatched", "dispatch": dispatch_receipt.get("transport"), "dispatchReceipt": dispatch_receipt, "fingerprint": single_flight_fingerprint, "groupKey": routing.get("groupKey"), "kind": payload.get("kind"), "suggestedMode": payload.get("suggestedMode"), } except Exception as exc: if single_flight_acquired: await release_agent99_sre_single_flight(single_flight_fingerprint, alert_id) logger.warning( "agent99_sre_bridge_failed_fail_open", alert_id=alert_id, alertname=alertname, error=str(exc), ) return {"status": "failed", "reason": type(exc).__name__}