feat(iwooos): add runtime security readback board

apps/api/src/api/v1/iwooos.py

@@ -7,15 +7,22 @@ Wazuh 接線採用只讀 metadata 模式：預設關閉、不保存 raw payload
 
 from __future__ import annotations
 
+import asyncio
+import json
 import os
 from base64 import b64encode
 from typing import Any
 from urllib.parse import urljoin, urlparse
 
 import httpx
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException, status
 from fastapi.responses import JSONResponse
 
+from src.services.iwooos_runtime_security_readback import (
+    load_latest_iwooos_runtime_security_readback,
+)
+from src.services.public_redaction import redact_public_lan_topology
+
 
 router = APIRouter(tags=["IwoooS Security"])
 REQUEST_TIMEOUT_SECONDS = 5.0
@@ -198,3 +205,30 @@ async def get_iwooos_wazuh_readonly_status_compat() -> JSONResponse:
 @router.get("/api/v1/iwooos/wazuh")
 async def get_iwooos_wazuh_readonly_status_v1() -> JSONResponse:
     return await _wazuh_readonly_status()
+
+
+@router.get(
+    "/api/v1/iwooos/runtime-security-readback",
+    response_model=dict[str, Any],
+    summary="取得 IwoooS runtime security readback",
+    description=(
+        "讀取最新已提交的 IwoooS 資安只讀快照，彙總 Wazuh、Kali、SOC/SIEM、"
+        "告警可讀性、owner dispatch 與外部入侵防護 Gate。此端點不呼叫 Wazuh / Kali / "
+        "主機 / Docker / Nginx / firewall / Telegram，不收集 secret，不授權 runtime 寫入。"
+    ),
+)
+async def get_iwooos_runtime_security_readback() -> dict[str, Any]:
+    """回傳 IwoooS 資安 runtime readback 只讀總板。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_iwooos_runtime_security_readback)
+        return redact_public_lan_topology(payload)
+    except FileNotFoundError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=str(exc),
+        ) from exc
+    except (json.JSONDecodeError, ValueError) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"IwoooS runtime security readback 無效：{exc}",
+        ) from exc

apps/api/src/services/iwooos_runtime_security_readback.py

@@ -0,0 +1,322 @@
+"""
+IwoooS runtime security readback.
+
+Loads committed security snapshots and exposes a public-safe, read-only runtime
+security board. This module never queries Wazuh, Kali, hosts, Docker, Nginx,
+firewalls, databases, Telegram, or secrets.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from src.services.snapshot_paths import default_security_dir
+
+_DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__))
+
+_SNAPSHOT_FILES = {
+    "owner_gap": "s4-9-owner-response-gap-audit.snapshot.json",
+    "wazuh_coverage": "wazuh-managed-host-coverage-gate.snapshot.json",
+    "wazuh_runtime": "wazuh-agent-visibility-runtime-gate.snapshot.json",
+    "kali_status": "kali-integration-status.snapshot.json",
+    "soc_control": "soc-siem-kali-wazuh-integration-control.snapshot.json",
+    "alert_readability": "telegram-alert-readability-guard.snapshot.json",
+    "owner_dispatch": "monitoring-owner-request-draft.snapshot.json",
+    "intrusion_prevention": "external-host-intrusion-prevention-control.snapshot.json",
+}
+
+_EXPECTED_SCHEMAS = {
+    "owner_gap": "s4_9_owner_response_gap_audit_v1",
+    "wazuh_coverage": "wazuh_managed_host_coverage_gate_v1",
+    "wazuh_runtime": "wazuh_agent_visibility_runtime_gate_v1",
+    "kali_status": "kali_integration_status_v1",
+    "soc_control": "soc_siem_kali_wazuh_integration_control_v1",
+    "alert_readability": "telegram_alert_readability_guard_v1",
+    "owner_dispatch": "monitoring_owner_request_draft_v1",
+    "intrusion_prevention": "external_host_intrusion_prevention_control_v1",
+}
+
+_FALSE_BOUNDARY_KEYS = {
+    "active_scan_authorized",
+    "alertmanager_reload_authorized",
+    "auto_block_authorized",
+    "credentialed_scan_authorized",
+    "firewall_change_authorized",
+    "host_write_authorized",
+    "kali_execute_authorized",
+    "kali_scan_authorized",
+    "nginx_reload_authorized",
+    "production_write_authorized",
+    "runtime_execution_authorized",
+    "runtime_gate_open",
+    "secret_value_collection_allowed",
+    "telegram_send_authorized",
+    "wazuh_active_response_authorized",
+    "wazuh_api_live_query_authorized",
+}
+
+
+def load_latest_iwooos_runtime_security_readback(
+    security_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load and normalize the current IwoooS runtime security readback."""
+    directory = security_dir or _DEFAULT_SECURITY_DIR
+    snapshots = {key: _load_snapshot(directory, key, filename) for key, filename in _SNAPSHOT_FILES.items()}
+    _require_runtime_boundaries(snapshots)
+
+    owner_gap_summary = _summary(snapshots["owner_gap"])
+    wazuh_summary = _summary(snapshots["wazuh_coverage"])
+    soc_summary = _summary(snapshots["soc_control"])
+    alert_summary = _summary(snapshots["alert_readability"])
+    dispatch_summary = _summary(snapshots["owner_dispatch"])
+    intrusion_summary = _summary(snapshots["intrusion_prevention"])
+
+    source_refs = [f"docs/security/{filename}" for filename in _SNAPSHOT_FILES.values()]
+    runtime_gate_count = _max_summary_count(
+        snapshots,
+        "runtime_gate_count",
+        "active_response_authorized_count",
+        "kali_active_scan_authorized_count",
+        "telegram_send_authorized_count",
+    )
+
+    return {
+        "schema_version": "iwooos_runtime_security_readback_v1",
+        "status": "blocked_waiting_owner_evidence_and_runtime_gates",
+        "mode": "committed_snapshot_readback_only_no_runtime_query",
+        "source_refs": source_refs,
+        "summary": {
+            "source_snapshot_count": len(source_refs),
+            "p0_lane_count": 6,
+            "control_plane_visibility_percent": _average_percent(
+                soc_summary.get("coverage_percent_after_soc_integration_control"),
+                intrusion_summary.get("coverage_percent_after_prevention_control"),
+                _alert_contract_percent(alert_summary),
+            ),
+            "actual_runtime_acceptance_percent": 0,
+            "owner_response_received_count": _int(owner_gap_summary.get("owner_response_received_count")),
+            "owner_response_accepted_count": _int(owner_gap_summary.get("owner_response_accepted_count")),
+            "redacted_evidence_refs_received_count": 0,
+            "request_sent_count": _int(dispatch_summary.get("request_sent_count")),
+            "wazuh_expected_host_scope_count": _int(wazuh_summary.get("expected_host_scope_count")),
+            "wazuh_manager_registry_accepted_count": _int(wazuh_summary.get("manager_registry_accepted_count")),
+            "wazuh_transport_observed_count": _int(wazuh_summary.get("manager_transport_established_connection_count")),
+            "wazuh_dashboard_api_degraded_observed_count": _int(
+                wazuh_summary.get("dashboard_api_degraded_observed_count")
+            ),
+            "kali_active_scan_authorized_count": _int(soc_summary.get("kali_active_scan_authorized_count")),
+            "kali_execute_authorized_count": _int(soc_summary.get("kali_execute_authorized_count")),
+            "kali_finding_envelope_accepted_count": _int(soc_summary.get("kali_finding_envelope_accepted_count")),
+            "alert_formatter_contract_marker_count": _int(alert_summary.get("source_formatter_marker_count")),
+            "alert_receipt_runtime_send_count": _int(alert_summary.get("telegram_send_authorized_count")),
+            "intrusion_prevention_candidate_count": _int(intrusion_summary.get("urgent_prevention_candidate_count")),
+            "runtime_gate_count": runtime_gate_count,
+        },
+        "lanes": [
+            _lane(
+                "wazuh_registry",
+                "blocked_waiting_manager_registry",
+                0,
+                "locked",
+                "manager_registry_cross_check",
+                {
+                    "expected_hosts": wazuh_summary.get("expected_host_scope_count", 0),
+                    "transport_observed": wazuh_summary.get("manager_transport_established_connection_count", 0),
+                    "registry_accepted": wazuh_summary.get("manager_registry_accepted_count", 0),
+                },
+                ["docs/security/wazuh-managed-host-coverage-gate.snapshot.json"],
+            ),
+            _lane(
+                "wazuh_dashboard_api",
+                "degraded_api_connection_not_green",
+                0,
+                "warn",
+                "dashboard_api_rbac_tls_repair_readback",
+                {
+                    "dashboard_api_degraded": wazuh_summary.get("dashboard_api_degraded_observed_count", 0),
+                    "runtime_gate": wazuh_summary.get("runtime_gate_count", 0),
+                    "accepted_evidence": _accepted_evidence_count(snapshots["wazuh_runtime"]),
+                },
+                [
+                    "docs/security/wazuh-managed-host-coverage-gate.snapshot.json",
+                    "docs/security/wazuh-agent-visibility-runtime-gate.snapshot.json",
+                ],
+            ),
+            _lane(
+                "kali_intake",
+                snapshots["kali_status"].get("status", "blocked_waiting_kali_scope"),
+                0,
+                "locked",
+                "kali_scope_and_finding_envelope_accepted",
+                {
+                    "active_scan_authorized": soc_summary.get("kali_active_scan_authorized_count", 0),
+                    "execute_authorized": soc_summary.get("kali_execute_authorized_count", 0),
+                    "finding_envelope_accepted": soc_summary.get("kali_finding_envelope_accepted_count", 0),
+                },
+                [
+                    "docs/security/kali-integration-status.snapshot.json",
+                    "docs/security/soc-siem-kali-wazuh-integration-control.snapshot.json",
+                ],
+            ),
+            _lane(
+                "alert_readability",
+                "contract_ready_no_send_receipt",
+                _alert_contract_percent(alert_summary),
+                "warn",
+                "alert_route_receipt_available",
+                {
+                    "formatter_markers": alert_summary.get("source_formatter_marker_count", 0),
+                    "required_markers": alert_summary.get("required_output_marker_count", 0),
+                    "telegram_send": alert_summary.get("telegram_send_authorized_count", 0),
+                },
+                ["docs/security/telegram-alert-readability-guard.snapshot.json"],
+            ),
+            _lane(
+                "owner_dispatch",
+                snapshots["owner_dispatch"].get("status", "owner_request_draft_ready_not_dispatched"),
+                0,
+                "locked",
+                "owner_response_packet_delivery",
+                {
+                    "request_drafts": dispatch_summary.get("request_draft_count", 0),
+                    "request_sent": dispatch_summary.get("request_sent_count", 0),
+                    "owner_accepted": dispatch_summary.get("owner_response_accepted_count", 0),
+                },
+                ["docs/security/monitoring-owner-request-draft.snapshot.json"],
+            ),
+            _lane(
+                "intrusion_prevention",
+                "candidate_only_no_runtime_containment",
+                _int(intrusion_summary.get("coverage_percent_after_prevention_control")),
+                "warn",
+                "redacted_evidence_refs_and_maintenance_window",
+                {
+                    "urgent_candidates": intrusion_summary.get("urgent_prevention_candidate_count", 0),
+                    "evidence_received": intrusion_summary.get("evidence_ref_received_count", 0),
+                    "containment_accepted": intrusion_summary.get("containment_decision_accepted_count", 0),
+                },
+                ["docs/security/external-host-intrusion-prevention-control.snapshot.json"],
+            ),
+        ],
+        "boundaries": {
+            "active_response_authorized": False,
+            "active_scan_authorized": False,
+            "action_buttons_allowed": False,
+            "host_write_authorized": False,
+            "kali_execute_authorized": False,
+            "nginx_reload_authorized": False,
+            "raw_payload_storage_allowed": False,
+            "runtime_execution_authorized": False,
+            "secret_value_collection_allowed": False,
+            "telegram_send_authorized": False,
+            "wazuh_active_response_authorized": False,
+            "wazuh_api_live_query_authorized": False,
+            "workflow_modification_authorized": False,
+            "not_authorization": True,
+        },
+        "no_false_green_rules": [
+            "dashboard_route_200_is_not_wazuh_registry_recovery",
+            "transport_count_is_not_full_host_management",
+            "ui_visible_is_not_runtime_authorization",
+            "owner_request_draft_is_not_owner_acceptance",
+            "kali_health_is_not_active_scan_authorization",
+            "alert_format_contract_is_not_telegram_send_receipt",
+        ],
+    }
+
+
+def _load_snapshot(directory: Path, key: str, filename: str) -> dict[str, Any]:
+    path = directory / filename
+    if not path.is_file():
+        raise FileNotFoundError(f"{path}: security snapshot not found")
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    expected_schema = _EXPECTED_SCHEMAS[key]
+    if payload.get("schema_version") != expected_schema:
+        raise ValueError(f"{path}: expected schema_version={expected_schema}")
+    return payload
+
+
+def _summary(payload: dict[str, Any]) -> dict[str, Any]:
+    summary = payload.get("summary")
+    return summary if isinstance(summary, dict) else {}
+
+
+def _int(value: Any) -> int:
+    return value if isinstance(value, int) else 0
+
+
+def _average_percent(*values: Any) -> int:
+    percents = [_int(value) for value in values if isinstance(value, int)]
+    return int(round(sum(percents) / len(percents))) if percents else 0
+
+
+def _alert_contract_percent(summary: dict[str, Any]) -> int:
+    source_markers = _int(summary.get("source_formatter_marker_count"))
+    required_markers = max(1, _int(summary.get("required_output_marker_count")))
+    return min(100, int(round((source_markers / required_markers) * 100)))
+
+
+def _accepted_evidence_count(payload: dict[str, Any]) -> int:
+    evidence = payload.get("required_evidence_before_green")
+    if not isinstance(evidence, list):
+        return 0
+    return sum(1 for item in evidence if isinstance(item, dict) and item.get("accepted") is True)
+
+
+def _lane(
+    lane_id: str,
+    status_text: Any,
+    completion_percent: int,
+    tone: str,
+    next_gate: str,
+    metrics: dict[str, Any],
+    source_refs: list[str],
+) -> dict[str, Any]:
+    return {
+        "lane_id": lane_id,
+        "status": str(status_text),
+        "completion_percent": completion_percent,
+        "tone": tone,
+        "next_gate": next_gate,
+        "metrics": {key: _int(value) for key, value in metrics.items()},
+        "source_refs": source_refs,
+    }
+
+
+def _max_summary_count(
+    snapshots: dict[str, dict[str, Any]],
+    *keys: str,
+) -> int:
+    return max((_int(_summary(payload).get(key)) for payload in snapshots.values() for key in keys), default=0)
+
+
+def _require_runtime_boundaries(snapshots: dict[str, dict[str, Any]]) -> None:
+    for name, payload in snapshots.items():
+        summary = _summary(payload)
+        if _int(summary.get("runtime_gate_count")) != 0:
+            raise ValueError(f"{name}: runtime_gate_count must remain 0")
+        for key in (
+            "owner_response_accepted_count",
+            "wazuh_active_response_enabled_count",
+            "active_response_enabled_count",
+            "active_scan_authorized_count",
+            "kali_active_scan_authorized_count",
+            "telegram_send_authorized_count",
+            "host_write_authorized_count",
+            "secret_value_collection_allowed_count",
+        ):
+            if key in summary and _int(summary.get(key)) != 0:
+                raise ValueError(f"{name}: {key} must remain 0")
+
+        boundaries = payload.get("execution_boundaries")
+        if isinstance(boundaries, dict):
+            invalid = sorted(
+                key for key in _FALSE_BOUNDARY_KEYS if key in boundaries and boundaries.get(key) is not False
+            )
+            if invalid:
+                raise ValueError(f"{name}: execution boundaries must remain false: {invalid}")

apps/api/src/services/snapshot_paths.py

@@ -35,3 +35,8 @@ def default_evaluations_dir(anchor: Path) -> Path:
 def default_operations_dir(anchor: Path) -> Path:
     """Resolve the default committed operations snapshot directory."""
     return resolve_repo_root(anchor) / "docs" / "operations"
+
+
+def default_security_dir(anchor: Path) -> Path:
+    """Resolve the default committed security snapshot directory."""
+    return resolve_repo_root(anchor) / "docs" / "security"