feat(iwooos): expose wazuh manager registry reviewer validation

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

@@ -35,6 +35,9 @@ from src.services.iwooos_wazuh_live_metadata_gate import (
 from src.services.iwooos_wazuh_managed_host_coverage import (
     load_latest_iwooos_wazuh_managed_host_coverage,
 )
+from src.services.iwooos_wazuh_manager_registry_reviewer_validation import (
+    load_latest_iwooos_wazuh_manager_registry_reviewer_validation,
+)
 from src.services.iwooos_wazuh_owner_evidence_preflight import (
     load_latest_iwooos_wazuh_owner_evidence_preflight,
 )
@@ -147,6 +150,34 @@ async def get_iwooos_wazuh_managed_host_coverage() -> dict[str, Any]:
         ) from exc
 
 
+@router.get(
+    "/api/v1/iwooos/wazuh-manager-registry-reviewer-validation",
+    response_model=dict[str, Any],
+    summary="取得 Wazuh manager registry reviewer validation 只讀讀回",
+    description=(
+        "讀取已提交的 Wazuh manager registry reviewer validation contract，回傳 owner export "
+        "必要欄位、reviewer 檢查、evidence slots、結果分流、拒收內容與 0 / false 邊界。"
+        "此端點不收 raw payload、不查 Wazuh API、不讀主機、不重新註冊 agent、不重啟服務、"
+        "不保存機密、不啟用主動回應、不改 Nginx / Docker / K8s / firewall。"
+    ),
+)
+async def get_iwooos_wazuh_manager_registry_reviewer_validation() -> dict[str, Any]:
+    """回傳 Wazuh manager registry reviewer validation 公開安全只讀狀態。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_iwooos_wazuh_manager_registry_reviewer_validation)
+        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 Wazuh manager registry reviewer validation 無效：{exc}",
+        ) from exc
+
+
 @router.get(
     "/api/v1/iwooos/runtime-security-readback",
     response_model=dict[str, Any],

apps/api/src/services/iwooos_wazuh_manager_registry_reviewer_validation.py

@@ -0,0 +1,227 @@
+"""
+IwoooS Wazuh manager registry reviewer validation readback.
+
+This service exposes a committed reviewer-validation contract for future
+owner-provided redacted Wazuh manager registry exports. It never receives raw
+payloads, queries Wazuh, reads host data, reads secrets, or authorizes runtime
+actions.
+"""
+
+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_FILE = "wazuh-manager-registry-reviewer-validation.snapshot.json"
+_EXPECTED_SCHEMA = "wazuh_manager_registry_reviewer_validation_v1"
+
+_REQUIRED_FALSE_BOUNDARIES = {
+    "agent_identity_public_display_allowed",
+    "host_write_authorized",
+    "internal_ip_public_display_allowed",
+    "kali_active_scan_authorized",
+    "raw_wazuh_payload_storage_allowed",
+    "runtime_execution_authorized",
+    "secret_value_collection_allowed",
+    "wazuh_active_response_authorized",
+    "wazuh_agent_reenroll_authorized",
+    "wazuh_agent_restart_authorized",
+    "wazuh_api_live_query_authorized",
+    "wazuh_manager_restart_authorized",
+}
+
+
+def load_latest_iwooos_wazuh_manager_registry_reviewer_validation(
+    security_dir: Path | None = None,
+) -> dict[str, Any]:
+    """Load the public-safe Wazuh manager registry reviewer-validation contract."""
+    directory = security_dir or _DEFAULT_SECURITY_DIR
+    snapshot = _load_snapshot(directory)
+    _require_boundaries(snapshot)
+
+    summary = _summary(snapshot)
+    merged_summary = {
+        "expected_scope_alias_count": _int(summary.get("expected_scope_alias_count")),
+        "required_owner_field_count": _int(summary.get("required_owner_field_count")),
+        "per_host_required_field_count": _int(summary.get("per_host_required_field_count")),
+        "reviewer_validation_check_count": _int(summary.get("reviewer_validation_check_count")),
+        "outcome_lane_count": _int(summary.get("outcome_lane_count")),
+        "evidence_slot_count": _int(summary.get("evidence_slot_count")),
+        "forbidden_payload_count": _int(summary.get("forbidden_payload_count")),
+        "forbidden_action_count": _int(summary.get("forbidden_action_count")),
+        "owner_registry_export_received_count": _int(summary.get("owner_registry_export_received_count")),
+        "owner_registry_export_accepted_count": _int(summary.get("owner_registry_export_accepted_count")),
+        "reviewer_validation_ready_count": _int(summary.get("reviewer_validation_ready_count")),
+        "reviewer_validation_passed_count": _int(summary.get("reviewer_validation_passed_count")),
+        "reviewer_validation_failed_count": _int(summary.get("reviewer_validation_failed_count")),
+        "reviewer_validation_quarantined_count": _int(summary.get("reviewer_validation_quarantined_count")),
+        "manager_registry_accepted_count": _int(summary.get("manager_registry_accepted_count")),
+        "post_enable_readback_passed_count": _int(summary.get("post_enable_readback_passed_count")),
+        "runtime_gate_count": _int(summary.get("runtime_gate_count")),
+        "host_write_authorized_count": _int(summary.get("host_write_authorized_count")),
+        "active_response_authorized_count": _int(summary.get("active_response_authorized_count")),
+        "secret_value_collection_allowed_count": _int(summary.get("secret_value_collection_allowed_count")),
+    }
+
+    return {
+        "schema_version": "iwooos_wazuh_manager_registry_reviewer_validation_readback_v1",
+        "source_schema_version": snapshot["schema_version"],
+        "status": snapshot.get("status", "waiting_owner_registry_export_for_reviewer_validation"),
+        "mode": "committed_validation_contract_readback_no_runtime_no_secret_collection",
+        "source_refs": [
+            f"docs/security/{_SNAPSHOT_FILE}",
+            "scripts/security/wazuh-manager-registry-reviewer-validation.py",
+        ],
+        "summary": merged_summary,
+        "expected_scope_aliases": _strings(snapshot.get("expected_scope_aliases")),
+        "reviewer_validation_checks": _checks(snapshot.get("reviewer_validation_checks")),
+        "outcome_lanes": _strings(snapshot.get("outcome_lanes")),
+        "evidence_slots": _evidence_slots(snapshot.get("evidence_slots")),
+        "forbidden_payloads": _strings(snapshot.get("forbidden_payloads")),
+        "forbidden_actions": _strings(snapshot.get("forbidden_actions")),
+        "boundary_markers": _boundary_markers(merged_summary),
+        "boundaries": {
+            "wazuh_api_live_query_authorized": False,
+            "wazuh_agent_reenroll_authorized": False,
+            "wazuh_agent_restart_authorized": False,
+            "wazuh_manager_restart_authorized": False,
+            "wazuh_active_response_authorized": False,
+            "host_write_authorized": False,
+            "secret_value_collection_allowed": False,
+            "raw_wazuh_payload_storage_allowed": False,
+            "agent_identity_public_display_allowed": False,
+            "internal_ip_public_display_allowed": False,
+            "kali_active_scan_authorized": False,
+            "runtime_execution_authorized": False,
+            "not_authorization": True,
+        },
+        "no_false_green_rules": _strings(snapshot.get("no_false_green_rules")),
+    }
+
+
+def _load_snapshot(directory: Path) -> dict[str, Any]:
+    path = directory / _SNAPSHOT_FILE
+    if not path.is_file():
+        raise FileNotFoundError(f"{path}: Wazuh manager registry reviewer validation 快照不存在")
+    with path.open(encoding="utf-8") as handle:
+        payload = json.load(handle)
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    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 _strings(value: Any) -> list[str]:
+    if not isinstance(value, list):
+        return []
+    return [item for item in value if isinstance(item, str)]
+
+
+def _checks(value: Any) -> list[dict[str, str]]:
+    if not isinstance(value, list):
+        return []
+    checks: list[dict[str, str]] = []
+    for item in value:
+        if not isinstance(item, dict):
+            continue
+        checks.append(
+            {
+                "check_id": str(item.get("check_id", "")),
+                "title": str(item.get("title", "")),
+                "required_evidence": str(item.get("required_evidence", "")),
+                "failure_lane": str(item.get("failure_lane", "")),
+            }
+        )
+    return checks
+
+
+def _evidence_slots(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list):
+        return []
+    slots: list[dict[str, Any]] = []
+    for item in value:
+        if not isinstance(item, dict):
+            continue
+        slots.append(
+            {
+                "slot_id": str(item.get("slot_id", "")),
+                "title": str(item.get("title", "")),
+                "required_fields": _strings(item.get("required_fields")),
+                "received": item.get("received") is True,
+                "accepted": item.get("accepted") is True,
+                "quarantined": item.get("quarantined") is True,
+                "next_gate": str(item.get("next_gate", "")),
+            }
+        )
+    return slots
+
+
+def _boundary_markers(summary: dict[str, int]) -> list[str]:
+    return [
+        "wazuh_manager_registry_reviewer_validation_visible=true",
+        f"wazuh_manager_registry_reviewer_validation_expected_scope_alias_count={summary['expected_scope_alias_count']}",
+        f"wazuh_manager_registry_reviewer_validation_required_owner_field_count={summary['required_owner_field_count']}",
+        f"wazuh_manager_registry_reviewer_validation_per_host_required_field_count={summary['per_host_required_field_count']}",
+        f"wazuh_manager_registry_reviewer_validation_check_count={summary['reviewer_validation_check_count']}",
+        f"wazuh_manager_registry_reviewer_validation_outcome_lane_count={summary['outcome_lane_count']}",
+        f"wazuh_manager_registry_reviewer_validation_evidence_slot_count={summary['evidence_slot_count']}",
+        f"wazuh_manager_registry_reviewer_validation_forbidden_payload_count={summary['forbidden_payload_count']}",
+        f"wazuh_manager_registry_reviewer_validation_owner_registry_export_received_count={summary['owner_registry_export_received_count']}",
+        f"wazuh_manager_registry_reviewer_validation_owner_registry_export_accepted_count={summary['owner_registry_export_accepted_count']}",
+        f"wazuh_manager_registry_reviewer_validation_passed_count={summary['reviewer_validation_passed_count']}",
+        f"wazuh_manager_registry_reviewer_validation_quarantined_count={summary['reviewer_validation_quarantined_count']}",
+        f"wazuh_manager_registry_reviewer_validation_manager_registry_accepted_count={summary['manager_registry_accepted_count']}",
+        f"wazuh_manager_registry_reviewer_validation_post_enable_readback_passed_count={summary['post_enable_readback_passed_count']}",
+        f"wazuh_manager_registry_reviewer_validation_runtime_gate_count={summary['runtime_gate_count']}",
+        "wazuh_api_live_query_authorized=false",
+        "wazuh_agent_reenroll_authorized=false",
+        "wazuh_agent_restart_authorized=false",
+        "wazuh_active_response_authorized=false",
+        "host_write_authorized=false",
+        "raw_wazuh_payload_storage_allowed=false",
+        "secret_value_collection_allowed=false",
+        "not_authorization=true",
+    ]
+
+
+def _require_boundaries(payload: dict[str, Any]) -> None:
+    summary = _summary(payload)
+    for key in (
+        "owner_registry_export_received_count",
+        "owner_registry_export_accepted_count",
+        "reviewer_validation_ready_count",
+        "reviewer_validation_passed_count",
+        "reviewer_validation_failed_count",
+        "reviewer_validation_quarantined_count",
+        "manager_registry_accepted_count",
+        "post_enable_readback_passed_count",
+        "runtime_gate_count",
+        "host_write_authorized_count",
+        "active_response_authorized_count",
+        "secret_value_collection_allowed_count",
+    ):
+        if _int(summary.get(key)) != 0:
+            raise ValueError(f"Wazuh manager registry reviewer validation summary.{key} 必須維持 0")
+
+    boundaries = payload.get("execution_boundaries")
+    if not isinstance(boundaries, dict):
+        raise ValueError("Wazuh manager registry reviewer validation execution_boundaries 缺失")
+    for key in _REQUIRED_FALSE_BOUNDARIES:
+        if boundaries.get(key) is not False:
+            raise ValueError(f"Wazuh manager registry reviewer validation execution_boundaries.{key} 必須維持 false")
+    if boundaries.get("not_authorization") is not True:
+        raise ValueError("Wazuh manager registry reviewer validation not_authorization 必須維持 true")