diff --git a/apps/api/src/api/v1/iwooos.py b/apps/api/src/api/v1/iwooos.py index ed92e61b5..04405fbd8 100644 --- a/apps/api/src/api/v1/iwooos.py +++ b/apps/api/src/api/v1/iwooos.py @@ -74,6 +74,7 @@ from src.services.iwooos_wazuh_runtime_gate_owner_review_readback import ( from src.services.iwooos_wazuh_runtime_gate_owner_review_readback import ( validate_iwooos_wazuh_runtime_gate_owner_review_packet as validate_wazuh_runtime_gate_owner_review_packet_payload, ) +from src.services.platform_operator_service import list_recent_channel_events from src.services.public_redaction import redact_public_lan_topology router = APIRouter(tags=["IwoooS Security"]) @@ -84,6 +85,25 @@ async def _wazuh_readonly_status() -> JSONResponse: return JSONResponse(status_code=result.http_status, content=result.payload) +async def _alertmanager_receipt_readback() -> dict[str, Any]: + try: + payload = await list_recent_channel_events( + project_id="awoooi", + channel_type=None, + provider_prefix="alertmanager", + limit=20, + ) + return {**payload, "source_status": payload.get("source_status", "ready")} + except Exception as exc: + return { + "events": [], + "total": 0, + "limit": 20, + "source_status": "source_unavailable", + "source_error": exc.__class__.__name__, + } + + @router.get("/api/iwooos/wazuh") async def get_iwooos_wazuh_readonly_status_compat() -> JSONResponse: return await _wazuh_readonly_status() @@ -108,7 +128,11 @@ async def get_iwooos_wazuh_readonly_status_v1() -> JSONResponse: async def get_iwooos_security_operating_system() -> dict[str, Any]: """回傳 IwoooS 資安作業系統公開安全總入口。""" try: - payload = await asyncio.to_thread(load_latest_iwooos_security_operating_system) + alert_receipt_readback = await _alertmanager_receipt_readback() + payload = await asyncio.to_thread( + load_latest_iwooos_security_operating_system, + alert_receipt_readback=alert_receipt_readback, + ) return redact_public_lan_topology(payload) except FileNotFoundError as exc: raise HTTPException( diff --git a/apps/api/src/services/iwooos_security_operating_system.py b/apps/api/src/services/iwooos_security_operating_system.py index a1b9c0b07..5b6d14c63 100644 --- a/apps/api/src/services/iwooos_security_operating_system.py +++ b/apps/api/src/services/iwooos_security_operating_system.py @@ -156,6 +156,7 @@ _RUNTIME_ACTION_KEYS = { def load_latest_iwooos_security_operating_system( security_dir: Path | None = None, + alert_receipt_readback: dict[str, Any] | None = None, ) -> dict[str, Any]: """Load the committed IwoooS security operating system contract.""" directory = security_dir or _DEFAULT_SECURITY_DIR @@ -163,6 +164,7 @@ def load_latest_iwooos_security_operating_system( _require_boundaries(snapshot) summary = _summary(snapshot) + alert_receipt_summary = _alert_receipt_summary(alert_receipt_readback) merged_summary = { "reference_framework_count": _int(summary.get("reference_framework_count")), "operating_role_count": _int(summary.get("operating_role_count")), @@ -192,9 +194,13 @@ def load_latest_iwooos_security_operating_system( "wazuh_registry_accepted_count": _int( summary.get("wazuh_registry_accepted_count") ), - "alert_receipt_accepted_count": _int( - summary.get("alert_receipt_accepted_count") + "alert_receipt_accepted_count": max( + _int(summary.get("alert_receipt_accepted_count")), + alert_receipt_summary["accepted_count"], ), + "alert_receipt_observed_count": alert_receipt_summary["observed_count"], + "alert_receipt_source_status": alert_receipt_summary["source_status"], + "alert_receipt_source_error": alert_receipt_summary["source_error"], "incident_case_accepted_count": _int( summary.get("incident_case_accepted_count") ), @@ -461,6 +467,37 @@ def _int(value: Any) -> int: return value if isinstance(value, int) else 0 +def _alert_receipt_summary(readback: dict[str, Any] | None) -> dict[str, Any]: + if not isinstance(readback, dict): + return { + "accepted_count": 0, + "observed_count": 0, + "source_status": "not_connected", + "source_error": None, + } + + source_status = str(readback.get("source_status") or "ready") + source_error = readback.get("source_error") + events = readback.get("events") + total = readback.get("total") + observed_count = ( + _int(total) + if total is not None + else len(events) + if isinstance(events, list) + else 0 + ) + accepted_count = ( + observed_count if source_status == "ready" and observed_count > 0 else 0 + ) + return { + "accepted_count": accepted_count, + "observed_count": observed_count, + "source_status": source_status, + "source_error": str(source_error) if source_error else None, + } + + def _strings(value: Any) -> list[str]: if not isinstance(value, list): return [] @@ -612,7 +649,7 @@ def _checkpoints(value: Any) -> list[dict[str, Any]]: return items -def _boundary_markers(summary: dict[str, int]) -> list[str]: +def _boundary_markers(summary: dict[str, Any]) -> list[str]: return [ "iwooos_security_operating_system_api_visible=true", "iwooos_security_operation_packet_validation_api_available=true", @@ -627,6 +664,8 @@ def _boundary_markers(summary: dict[str, int]) -> list[str]: f"iwooos_security_operating_system_operation_packet_required_field_count={summary['operation_packet_required_field_count']}", f"iwooos_security_operating_system_evidence_weighted_percent={summary['evidence_weighted_security_operating_system_percent']}", f"iwooos_security_operating_system_wazuh_registry_accepted_count={summary['wazuh_registry_accepted_count']}", + f"iwooos_security_operating_system_alert_receipt_accepted_count={summary['alert_receipt_accepted_count']}", + f"iwooos_security_operating_system_alert_receipt_observed_count={summary['alert_receipt_observed_count']}", "iwooos_security_operating_system_operation_packet_validation_no_persist=true", "iwooos_security_operating_system_runtime_gate_count=0", "runtime_execution_authorized=false", diff --git a/apps/api/tests/test_iwooos_security_operating_system.py b/apps/api/tests/test_iwooos_security_operating_system.py index 30ab94947..1a8ff2582 100644 --- a/apps/api/tests/test_iwooos_security_operating_system.py +++ b/apps/api/tests/test_iwooos_security_operating_system.py @@ -71,6 +71,9 @@ def test_iwooos_security_operating_system_readback_exposes_api_validator() -> No assert payload["summary"]["operation_packet_validator_available_count"] == 1 assert payload["summary"]["operation_packet_required_field_count"] == 24 assert payload["summary"]["wazuh_registry_accepted_count"] == 6 + assert payload["summary"]["alert_receipt_accepted_count"] == 0 + assert payload["summary"]["alert_receipt_observed_count"] == 0 + assert payload["summary"]["alert_receipt_source_status"] == "not_connected" assert payload["summary"]["runtime_gate_count"] == 0 assert payload["operation_packet_validation_endpoint"] == ( "/api/v1/iwooos/security-operating-system/validate-operation-packet" @@ -87,6 +90,26 @@ def test_iwooos_security_operating_system_readback_exposes_api_validator() -> No ) +def test_iwooos_security_operating_system_accepts_alert_receipt_readback() -> None: + payload = load_latest_iwooos_security_operating_system( + alert_receipt_readback={ + "events": [{"provider_event_id": "alertmanager:received:alert-a"}], + "total": 1, + "limit": 20, + "source_status": "ready", + } + ) + + assert payload["summary"]["alert_receipt_accepted_count"] == 1 + assert payload["summary"]["alert_receipt_observed_count"] == 1 + assert payload["summary"]["alert_receipt_source_status"] == "ready" + assert payload["summary"]["runtime_gate_count"] == 0 + assert any( + marker == "iwooos_security_operating_system_alert_receipt_accepted_count=1" + for marker in payload["boundary_markers"] + ) + + def test_iwooos_security_operating_system_api_is_public_safe() -> None: response = _client().get("/api/v1/iwooos/security-operating-system") @@ -103,6 +126,31 @@ def test_iwooos_security_operating_system_api_is_public_safe() -> None: assert "WAZUH_API_PASSWORD" not in response.text +def test_iwooos_security_operating_system_api_counts_alertmanager_receipts( + monkeypatch, +) -> None: + async def fake_recent_events(**_: object) -> dict[str, object]: + return { + "events": [{"provider_event_id": "alertmanager:received:alert-a"}], + "total": 1, + "limit": 20, + } + + monkeypatch.setattr( + "src.api.v1.iwooos.list_recent_channel_events", + fake_recent_events, + ) + + response = _client().get("/api/v1/iwooos/security-operating-system") + + assert response.status_code == 200 + data = response.json() + assert data["summary"]["alert_receipt_accepted_count"] == 1 + assert data["summary"]["alert_receipt_observed_count"] == 1 + assert data["summary"]["alert_receipt_source_status"] == "ready" + assert data["summary"]["runtime_gate_count"] == 0 + + def test_iwooos_security_operation_packet_validator_accepts_redacted_loop() -> None: payload = validate_iwooos_security_operation_packet(_valid_operation_packet())