feat(iwooos): expose owner evidence intake preflight

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

@@ -20,6 +20,9 @@ from src.services.iwooos_runtime_security_readback import (
 from src.services.iwooos_high_value_config_control_coverage import (
     load_latest_iwooos_high_value_config_control_coverage,
 )
+from src.services.iwooos_owner_evidence_intake_preflight import (
+    load_latest_iwooos_owner_evidence_intake_preflight,
+)
 from src.services.iwooos_security_control_coverage import (
     load_latest_iwooos_security_control_coverage,
 )
@@ -199,3 +202,31 @@ async def get_iwooos_high_value_config_control_coverage() -> dict[str, Any]:
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"IwoooS high-value config control coverage 無效：{exc}",
         ) from exc
+
+
+@router.get(
+    "/api/v1/iwooos/owner-evidence-intake-preflight",
+    response_model=dict[str, Any],
+    summary="取得 IwoooS 負責人脫敏證據收件預檢",
+    description=(
+        "整合 high-value config owner packet、配置覆蓋矩陣與 Wazuh 負責人證據預檢，"
+        "回傳 Nginx、DNS / TLS、K8s、secret / runner、public runtime config 與 Wazuh registry "
+        "的公開安全收件欄位、拒收規則與 0 / false 邊界。此端點不送 owner request、不收回覆、"
+        "不寫 reviewer queue、不讀 secret、不查 live host、不查 Wazuh API、不啟動 runtime action。"
+    ),
+)
+async def get_iwooos_owner_evidence_intake_preflight() -> dict[str, Any]:
+    """回傳 IwoooS 負責人脫敏證據收件預檢公開安全只讀狀態。"""
+    try:
+        payload = await asyncio.to_thread(load_latest_iwooos_owner_evidence_intake_preflight)
+        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 owner evidence intake preflight 無效：{exc}",
+        ) from exc

apps/api/src/services/iwooos_owner_evidence_intake_preflight.py

@@ -0,0 +1,454 @@
+"""
+IwoooS 負責人脫敏證據收件預檢讀回。
+
+此服務只整合已提交 snapshot 的公開安全欄位，不查 live host、不讀
+Nginx、不查 Wazuh API、不收 secret、不送 owner request，也不建立
+reviewer queue 或 runtime action。
+"""
+
+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__))
+_SCHEMA_VERSION = "iwooos_owner_evidence_intake_preflight_v1"
+_INTAKE_SCHEMA = "high_value_config_owner_packet_intake_preflight_v1"
+_COVERAGE_SCHEMA = "high_value_config_control_coverage_v1"
+_WAZUH_SCHEMA = "wazuh_agent_visibility_owner_evidence_preflight_v1"
+
+_SNAPSHOTS = {
+    "intake": "high-value-config-owner-packet-intake-preflight.snapshot.json",
+    "coverage": "high-value-config-control-coverage.snapshot.json",
+    "wazuh": "wazuh-agent-visibility-owner-evidence-preflight.snapshot.json",
+}
+
+_REQUIRED_FALSE_BOUNDARIES = {
+    "action_buttons_allowed",
+    "active_scan_authorized",
+    "dispatch_authorized",
+    "dns_tls_change_authorized",
+    "host_write_authorized",
+    "nginx_reload_authorized",
+    "production_write_authorized",
+    "request_sent",
+    "reviewer_queue_write",
+    "runtime_execution_authorized",
+    "secret_value_collection_allowed",
+    "workflow_modification_authorized",
+}
+
+_OWNER_FIELDS = [
+    "owner_role_or_team",
+    "decision",
+    "decision_reason",
+    "affected_scope",
+    "redacted_evidence_refs",
+    "followup_owner",
+    "rollback_owner",
+    "maintenance_window",
+    "validation_plan",
+]
+
+_LANE_ORDER = [
+    "nginx_public_gateway",
+    "dns_tls_certbot",
+    "k8s_production_gitops",
+    "secret_workflow_runner_source_control",
+    "public_admin_api_runtime_config",
+    "wazuh_manager_registry",
+]
+
+_PUBLIC_LABELS = {
+    "nginx_public_gateway": "Nginx / 公開入口 / 反向代理",
+    "dns_tls_certbot": "DNS / TLS / 憑證續期",
+    "k8s_production_gitops": "K8s / ArgoCD / 正式環境宣告",
+    "secret_workflow_runner_source_control": "機密中繼資料 / 工作流程 / 執行器",
+    "public_admin_api_runtime_config": "公開 / 後台 / API runtime config",
+    "wazuh_manager_registry": "Wazuh manager registry / 逐主機矩陣",
+}
+
+_REQUIRED_EVIDENCE = {
+    "nginx_public_gateway": [
+        "owner-provided live conf 脫敏參照",
+        "source-to-live rendered diff 參照",
+        "nginx -t 結果參照",
+        "affected route smoke 參照",
+        "maintenance window 與 rollback owner",
+    ],
+    "dns_tls_certbot": [
+        "domain / certificate path 對照參照",
+        "ACME challenge route smoke 參照",
+        "renewal window 與 rollback owner",
+        "public HTTPS smoke 參照",
+    ],
+    "k8s_production_gitops": [
+        "ArgoCD app health / sync / revision 參照",
+        "manifest diff 與 rollback revision",
+        "Pending / image pull / scheduling 摘要",
+        "route / AI / monitoring impact 參照",
+    ],
+    "secret_workflow_runner_source_control": [
+        "secret name parity state 參照",
+        "workflow diff state 參照",
+        "runner owner attestation",
+        "deploy marker / run readback",
+        "log redaction readback",
+    ],
+    "public_admin_api_runtime_config": [
+        "affected route refs",
+        "admin / auth boundary",
+        "API contract readback",
+        "frontend env / i18n redaction review",
+        "desktop / mobile production smoke",
+    ],
+    "wazuh_manager_registry": [
+        "manager registry 脫敏計數",
+        "逐主機 alias matrix",
+        "Dashboard API 狀態參照",
+        "manager API version / health 參照",
+        "缺席主機 owner decision",
+    ],
+}
+
+_BLOCKED_RUNTIME_ACTIONS = {
+    "nginx_public_gateway": ["nginx_reload", "host_write", "dns_tls_modify", "secret_value_submit"],
+    "dns_tls_certbot": ["certbot_renew", "dns_tls_modify", "nginx_reload", "secret_value_submit"],
+    "k8s_production_gitops": ["argocd_sync", "kubectl_apply", "secret_value_submit", "production_write"],
+    "secret_workflow_runner_source_control": [
+        "workflow_modify",
+        "runner_enable",
+        "secret_value_submit",
+        "webhook_secret_submit",
+    ],
+    "public_admin_api_runtime_config": ["production_write", "cors_change", "route_change", "secret_value_submit"],
+    "wazuh_manager_registry": [
+        "wazuh_api_live_query",
+        "wazuh_active_response",
+        "host_write",
+        "active_scan",
+        "raw_payload_store",
+    ],
+}
+
+
+def load_latest_iwooos_owner_evidence_intake_preflight(
+    security_dir: Path | None = None,
+) -> dict[str, Any]:
+    """載入 IwoooS 統一負責人脫敏證據收件預檢。"""
+    sec_dir = security_dir or _DEFAULT_SECURITY_DIR
+    intake = _load_json(sec_dir / _SNAPSHOTS["intake"], _INTAKE_SCHEMA)
+    coverage = _load_json(sec_dir / _SNAPSHOTS["coverage"], _COVERAGE_SCHEMA)
+    wazuh = _load_json(sec_dir / _SNAPSHOTS["wazuh"], _WAZUH_SCHEMA)
+    _validate_intake(intake)
+    _validate_coverage(coverage)
+    _validate_wazuh(wazuh)
+
+    packet_by_category = {
+        packet.get("category_id"): packet
+        for packet in intake.get("intake_packets") or []
+        if isinstance(packet, dict) and packet.get("category_id")
+    }
+    coverage_by_category = {
+        category.get("category_id"): category
+        for category in coverage.get("coverage_categories") or []
+        if isinstance(category, dict) and category.get("category_id")
+    }
+    lanes = [
+        _build_lane(lane_id, packet_by_category, coverage_by_category, wazuh)
+        for lane_id in _LANE_ORDER
+    ]
+    summary = _summary(intake, wazuh, lanes)
+
+    return {
+        "schema_version": _SCHEMA_VERSION,
+        "source_schema_versions": {
+            "high_value_config_owner_packet_intake_preflight": _INTAKE_SCHEMA,
+            "high_value_config_control_coverage": _COVERAGE_SCHEMA,
+            "wazuh_owner_evidence_preflight": _WAZUH_SCHEMA,
+        },
+        "status": "owner_evidence_intake_preflight_ready_no_runtime_action",
+        "mode": "committed_snapshot_projection_only_no_live_runtime_query",
+        "summary": summary,
+        "lanes": lanes,
+        "dispatch_preflight_checks": _public_checks(intake.get("dispatch_preflight_checks")),
+        "reviewer_intake_lanes": _public_reviewer_lanes(intake.get("reviewer_intake_lanes")),
+        "blocked_requests": _blocked_requests(intake, wazuh),
+        "execution_boundaries": _execution_boundaries(),
+        "boundary_markers": _boundary_markers(summary),
+        "no_false_green_rules": [
+            "收件預檢 ready 不代表 request 已送出、owner response 已收到、reviewer 已接受或 runtime gate 已開。",
+            "只收脫敏 evidence ref、ticket、commit、hash 或 metadata pointer；secret value、raw payload、token、private key、cookie、session 一律拒收或隔離。",
+            "Nginx reload、DNS / TLS 變更、ArgoCD sync、kubectl apply、workflow 修改、runner enable、Wazuh active response、Kali active scan 都不是此預檢授權。",
+            "前台卡片、API 200、一般工作批准或 Dashboard 可開，不能替代 owner-provided redacted evidence 與 reviewer validation。",
+        ],
+        "source_refs": [
+            f"docs/security/{_SNAPSHOTS['intake']}",
+            f"docs/security/{_SNAPSHOTS['coverage']}",
+            f"docs/security/{_SNAPSHOTS['wazuh']}",
+            "scripts/security/high-value-config-owner-packet-intake-preflight.py",
+            "scripts/security/wazuh-agent-visibility-owner-evidence-preflight.py",
+        ],
+    }
+
+
+def _load_json(path: Path, schema: str) -> dict[str, Any]:
+    if not path.is_file():
+        raise FileNotFoundError(f"{path}: snapshot 不存在")
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(payload, dict):
+        raise ValueError(f"{path}: expected JSON object")
+    if payload.get("schema_version") != schema:
+        raise ValueError(f"{path}: schema_version 必須是 {schema}")
+    return payload
+
+
+def _validate_intake(payload: dict[str, Any]) -> None:
+    summary = payload.get("summary") or {}
+    expected_zero = (
+        "request_sent_count",
+        "received_response_count",
+        "accepted_response_count",
+        "rejected_response_count",
+        "reviewer_queue_write_count",
+        "runtime_gate_count",
+        "action_button_count",
+    )
+    for key in expected_zero:
+        if _int(summary.get(key)) != 0:
+            raise ValueError(f"intake.summary.{key} 必須維持 0")
+    boundaries = payload.get("execution_boundaries") or {}
+    for key in _REQUIRED_FALSE_BOUNDARIES:
+        if boundaries.get(key) is not False:
+            raise ValueError(f"intake.execution_boundaries.{key} 必須維持 false")
+    if boundaries.get("not_authorization") is not True:
+        raise ValueError("intake.execution_boundaries.not_authorization 必須是 true")
+
+
+def _validate_coverage(payload: dict[str, Any]) -> None:
+    summary = payload.get("summary") or {}
+    if _int(summary.get("owner_response_received_count")) != 0:
+        raise ValueError("coverage owner_response_received_count 必須維持 0")
+    if _int(summary.get("owner_response_accepted_count")) != 0:
+        raise ValueError("coverage owner_response_accepted_count 必須維持 0")
+    if _int(summary.get("runtime_gate_count")) != 0:
+        raise ValueError("coverage runtime_gate_count 必須維持 0")
+    if _int(summary.get("action_button_count")) != 0:
+        raise ValueError("coverage action_button_count 必須維持 0")
+
+
+def _validate_wazuh(payload: dict[str, Any]) -> None:
+    summary = payload.get("summary") or {}
+    for key in (
+        "registry_export_received_count",
+        "registry_export_accepted_count",
+        "owner_evidence_received_count",
+        "owner_evidence_accepted_count",
+        "runtime_gate_count",
+        "active_response_authorized_count",
+        "host_write_authorized_count",
+        "secret_value_collection_allowed_count",
+    ):
+        if _int(summary.get(key)) != 0:
+            raise ValueError(f"wazuh.summary.{key} 必須維持 0")
+    boundaries = payload.get("execution_boundaries") or {}
+    for key, value in boundaries.items():
+        if key == "not_authorization":
+            if value is not True:
+                raise ValueError("wazuh.execution_boundaries.not_authorization 必須是 true")
+        elif value is not False:
+            raise ValueError(f"wazuh.execution_boundaries.{key} 必須維持 false")
+
+
+def _build_lane(
+    lane_id: str,
+    packet_by_category: dict[str, dict[str, Any]],
+    coverage_by_category: dict[str, dict[str, Any]],
+    wazuh: dict[str, Any],
+) -> dict[str, Any]:
+    if lane_id == "secret_workflow_runner_source_control":
+        categories = [
+            coverage_by_category.get("secret_metadata") or {},
+            coverage_by_category.get("gitea_workflow_runner_source_control") or {},
+        ]
+        priority = "P0"
+        control_tier = "C0"
+        status = "coverage_derived_waiting_owner_packet"
+        evidence_ref_count = sum(_len(category.get("evidence_refs")) for category in categories)
+        required_validation_count = sum(_len(category.get("required_validation")) for category in categories)
+        source_kind = "coverage_derived_intake_candidate"
+    elif lane_id == "wazuh_manager_registry":
+        summary = wazuh.get("summary") or {}
+        priority = "P0"
+        control_tier = "C0"
+        status = "wazuh_owner_evidence_preflight_ready"
+        evidence_ref_count = 0
+        required_validation_count = _int(summary.get("reviewer_check_count"))
+        source_kind = "wazuh_owner_evidence_preflight"
+    else:
+        packet = packet_by_category.get(lane_id)
+        category = coverage_by_category.get(lane_id) or {}
+        priority = str((packet or category).get("priority") or "P0")
+        control_tier = str((packet or category).get("control_tier") or "C0")
+        status = "owner_packet_ready_waiting_dispatch_preflight" if packet else "coverage_derived_waiting_owner_packet"
+        evidence_ref_count = _len(category.get("evidence_refs"))
+        required_validation_count = _len((packet or {}).get("required_validation")) or _len(
+            category.get("required_validation")
+        )
+        source_kind = "owner_packet_intake_preflight" if packet else "coverage_derived_intake_candidate"
+
+    return {
+        "lane_id": lane_id,
+        "label": _PUBLIC_LABELS[lane_id],
+        "priority": priority,
+        "control_tier": control_tier,
+        "status": status,
+        "source_kind": source_kind,
+        "required_owner_fields": _OWNER_FIELDS,
+        "required_owner_field_count": len(_OWNER_FIELDS),
+        "required_evidence": _REQUIRED_EVIDENCE[lane_id],
+        "required_evidence_count": len(_REQUIRED_EVIDENCE[lane_id]),
+        "required_validation_count": required_validation_count,
+        "blocked_runtime_actions": _BLOCKED_RUNTIME_ACTIONS[lane_id],
+        "blocked_runtime_action_count": len(_BLOCKED_RUNTIME_ACTIONS[lane_id]),
+        "evidence_ref_count": evidence_ref_count,
+        "request_sent": False,
+        "owner_response_received": False,
+        "owner_response_accepted": False,
+        "owner_response_rejected": False,
+        "owner_response_quarantined": False,
+        "reviewer_queue_write": False,
+        "runtime_gate_open": False,
+        "action_buttons_allowed": False,
+        "secret_value_collection_allowed": False,
+        "not_authorization": True,
+    }
+
+
+def _summary(intake: dict[str, Any], wazuh: dict[str, Any], lanes: list[dict[str, Any]]) -> dict[str, int]:
+    intake_summary = intake.get("summary") or {}
+    wazuh_summary = wazuh.get("summary") or {}
+    return {
+        "lane_count": len(lanes),
+        "owner_packet_source_lane_count": sum(1 for lane in lanes if lane["source_kind"] == "owner_packet_intake_preflight"),
+        "coverage_derived_lane_count": sum(1 for lane in lanes if lane["source_kind"] == "coverage_derived_intake_candidate"),
+        "wazuh_lane_count": sum(1 for lane in lanes if lane["source_kind"] == "wazuh_owner_evidence_preflight"),
+        "canonical_owner_field_count": len(_OWNER_FIELDS),
+        "required_owner_field_total": len(_OWNER_FIELDS) * len(lanes),
+        "dispatch_preflight_check_count": _int(intake_summary.get("dispatch_preflight_check_count")),
+        "reviewer_intake_lane_count": _int(intake_summary.get("reviewer_intake_lane_count")),
+        "blocked_request_count": len(_blocked_requests(intake, wazuh)),
+        "wazuh_required_field_count": _int(wazuh_summary.get("required_field_count")),
+        "wazuh_reviewer_check_count": _int(wazuh_summary.get("reviewer_check_count")),
+        "request_sent_count": 0,
+        "owner_response_received_count": 0,
+        "owner_response_accepted_count": 0,
+        "owner_response_rejected_count": 0,
+        "owner_response_quarantined_count": 0,
+        "reviewer_queue_write_count": 0,
+        "runtime_gate_count": 0,
+        "action_button_count": 0,
+        "host_write_authorized_count": 0,
+        "active_scan_authorized_count": 0,
+        "wazuh_live_query_authorized_count": 0,
+        "wazuh_active_response_authorized_count": 0,
+        "secret_value_collection_allowed_count": 0,
+    }
+
+
+def _public_checks(items: Any) -> list[dict[str, Any]]:
+    checks = items if isinstance(items, list) else []
+    return [
+        {
+            "check_id": str(item.get("check_id") or ""),
+            "label": str(item.get("label") or ""),
+            "instruction": str(item.get("instruction") or ""),
+            "required": item.get("required") is True,
+            "gate_effect": "不增加 request / received / accepted / runtime gate。",
+        }
+        for item in checks
+        if isinstance(item, dict)
+    ]
+
+
+def _public_reviewer_lanes(items: Any) -> list[dict[str, Any]]:
+    lanes = items if isinstance(items, list) else []
+    return [
+        {
+            "lane_id": str(item.get("lane_id") or ""),
+            "instruction": str(item.get("instruction") or ""),
+            "gate_effect": str(item.get("gate_effect") or ""),
+        }
+        for item in lanes
+        if isinstance(item, dict)
+    ]
+
+
+def _blocked_requests(intake: dict[str, Any], wazuh: dict[str, Any]) -> list[str]:
+    blocked = set()
+    for item in intake.get("blocked_requests") or []:
+        if isinstance(item, str):
+            blocked.add(item)
+    for item in wazuh.get("forbidden_payloads") or []:
+        if isinstance(item, str):
+            blocked.add(item)
+    blocked.update(
+        {
+            "wazuh_api_live_query",
+            "wazuh_active_response",
+            "raw_wazuh_payload_store",
+            "internal_ip_public_display",
+            "agent_identity_public_display",
+        }
+    )
+    return sorted(blocked)
+
+
+def _execution_boundaries() -> dict[str, bool]:
+    boundaries = {key: False for key in sorted(_REQUIRED_FALSE_BOUNDARIES)}
+    boundaries.update(
+        {
+            "wazuh_api_live_query_authorized": False,
+            "wazuh_active_response_authorized": False,
+            "agent_identity_public_display_allowed": False,
+            "internal_ip_public_display_allowed": False,
+            "raw_wazuh_payload_storage_allowed": False,
+            "not_authorization": True,
+        }
+    )
+    return boundaries
+
+
+def _boundary_markers(summary: dict[str, int]) -> list[str]:
+    return [
+        f"owner_evidence_intake_lane_count={summary['lane_count']}",
+        f"canonical_owner_field_count={summary['canonical_owner_field_count']}",
+        f"required_owner_field_total={summary['required_owner_field_total']}",
+        f"dispatch_preflight_check_count={summary['dispatch_preflight_check_count']}",
+        f"reviewer_intake_lane_count={summary['reviewer_intake_lane_count']}",
+        f"blocked_request_count={summary['blocked_request_count']}",
+        "request_sent_count=0",
+        "owner_response_received_count=0",
+        "owner_response_accepted_count=0",
+        "owner_response_quarantined_count=0",
+        "reviewer_queue_write_count=0",
+        "runtime_gate_count=0",
+        "action_button_count=0",
+        "host_write_authorized_count=0",
+        "active_scan_authorized_count=0",
+        "wazuh_live_query_authorized_count=0",
+        "wazuh_active_response_authorized_count=0",
+        "secret_value_collection_allowed_count=0",
+        "not_authorization=true",
+    ]
+
+
+def _int(value: Any) -> int:
+    return value if isinstance(value, int) else 0
+
+
+def _len(value: Any) -> int:
+    return len(value) if isinstance(value, list) else 0