""" IwoooS security control coverage rollup. This service consolidates committed security inventory snapshots into one read-only control-plane view. It never queries live hosts, Wazuh, Kali, Kubernetes, Docker, Nginx, Telegram, or secret stores. """ from __future__ import annotations import json from pathlib import Path from typing import Any from src.services.snapshot_paths import default_evaluations_dir, default_security_dir _DEFAULT_SECURITY_DIR = default_security_dir(Path(__file__)) _DEFAULT_EVALUATIONS_DIR = default_evaluations_dir(Path(__file__)) _SCHEMA_VERSION = "iwooos_security_control_coverage_v1" _SECURITY_SNAPSHOTS = { "asset_ledger": "security-asset-control-ledger.snapshot.json", "host_service": "host-service-config-inventory.snapshot.json", "monitoring": "monitoring-alerting-observability-inventory.snapshot.json", "ssh_network": "ssh-network-access-inventory.snapshot.json", "wazuh_hosts": "wazuh-managed-host-coverage-gate.snapshot.json", "agent_bounty": "agent-bounty-iwooos-onboarding-handoff.snapshot.json", } def load_latest_iwooos_security_control_coverage( security_dir: Path | None = None, evaluations_dir: Path | None = None, ) -> dict[str, Any]: """Load committed security-control inventory snapshots as one rollup.""" sec_dir = security_dir or _DEFAULT_SECURITY_DIR eval_dir = evaluations_dir or _DEFAULT_EVALUATIONS_DIR snapshots = { key: _load_json(sec_dir / filename) for key, filename in _SECURITY_SNAPSHOTS.items() } runtime_inventory = _load_latest(eval_dir, "runtime_surface_inventory_*.json") ai_inventory = _load_latest(eval_dir, "ai_agent_automation_inventory_snapshot_*.json") snapshots["runtime_surface"] = runtime_inventory snapshots["ai_agent_automation"] = ai_inventory _require_snapshot_schemas(snapshots) _require_zero_runtime_boundaries(snapshots) domains = _build_domains(snapshots) summary = _build_summary(snapshots, domains) p0_next_actions = _build_p0_next_actions() professional_review = _build_professional_security_governance_review( summary, domains, p0_next_actions, ) summary.update( { "professional_review_finding_count": len( professional_review["root_cause_summary"] ), "professional_review_p0_work_item_count": len( professional_review["priority_work_items"] ), "external_security_tool_track_count": len( professional_review["security_tool_integration_matrix"] ), "ai_security_automation_stage_count": len( professional_review["ai_automation_closure_loop"] ), "full_scope_contract_count": len(professional_review["scope_contract"]), "ui_ux_control_room_requirement_count": len( professional_review["ui_ux_control_room_requirements"] ), } ) return { "schema_version": _SCHEMA_VERSION, "status": "committed_scope_rollup_ready_with_controlled_apply_exception", "mode": "committed_snapshot_rollup_only_no_live_runtime_query", "summary": summary, "domains": domains, "professional_review": professional_review, "p0_next_actions": p0_next_actions, "no_false_green_rules": [ "納管覆蓋總表只代表 committed snapshot 可讀，不代表所有主機已被 Wazuh manager registry 驗收。", "route 200、transport observed、UI 可見、一般工作批准都不能當成 runtime 授權。", "IwoooS ledger 的 owner response received / accepted、live evidence accepted、active scan、active response、Telegram send、host write 仍維持 0 / false；這不阻擋 AwoooP allowlisted controlled apply。", "Nginx、Firewall、Workflow、Secret、K8s、Docker、systemd、AI Agent provider 變更都必須先進 execution packet、check-mode、rollback、verifier 與 KM / PlayBook trust，不得繞過受控路由。", ], "source_refs": [ "docs/security/security-asset-control-ledger.snapshot.json", "docs/security/host-service-config-inventory.snapshot.json", "docs/security/monitoring-alerting-observability-inventory.snapshot.json", "docs/security/ssh-network-access-inventory.snapshot.json", "docs/security/wazuh-managed-host-coverage-gate.snapshot.json", "docs/security/agent-bounty-iwooos-onboarding-handoff.snapshot.json", f"docs/evaluations/{runtime_inventory['_snapshot_name']}", f"docs/evaluations/{ai_inventory['_snapshot_name']}", ], } def _load_json(path: Path) -> dict[str, Any]: with path.open(encoding="utf-8") as handle: payload = json.load(handle) if not isinstance(payload, dict): raise ValueError(f"{path}: expected JSON object") payload["_snapshot_name"] = path.name return payload def _load_latest(directory: Path, pattern: str) -> dict[str, Any]: candidates = sorted(directory.glob(pattern)) if not candidates: raise FileNotFoundError(f"no snapshots found: {directory / pattern}") return _load_json(candidates[-1]) def _require_snapshot_schemas(snapshots: dict[str, dict[str, Any]]) -> None: expected = { "asset_ledger": "security_asset_control_ledger_v1", "host_service": "host_service_config_inventory_v1", "monitoring": "monitoring_alerting_observability_inventory_v1", "ssh_network": "ssh_network_access_inventory_v1", "wazuh_hosts": "wazuh_managed_host_coverage_gate_v1", "agent_bounty": "agent_bounty_iwooos_onboarding_handoff_v1", "runtime_surface": "runtime_surface_inventory_v1", "ai_agent_automation": "ai_agent_automation_inventory_snapshot_v1", } for key, schema in expected.items(): actual = snapshots[key].get("schema_version") if actual != schema: raise ValueError(f"{key}: expected schema_version={schema}, got {actual!r}") def _require_zero_runtime_boundaries(snapshots: dict[str, dict[str, Any]]) -> None: for key in ("asset_ledger", "host_service", "monitoring", "ssh_network", "wazuh_hosts"): boundaries = snapshots[key].get("execution_boundaries") or {} allowed = sorted( name for name, value in boundaries.items() if name != "not_authorization" and value is not False ) if allowed: raise ValueError(f"{key}: execution boundaries must stay false: {allowed}") runtime_approvals = snapshots["runtime_surface"].get("approval_boundaries") or {} allowed_runtime = sorted(name for name, value in runtime_approvals.items() if value is not False) if allowed_runtime: raise ValueError(f"runtime_surface: approval boundaries must stay false: {allowed_runtime}") runtime_operations = snapshots["runtime_surface"].get("operation_boundaries") or {} if runtime_operations.get("read_only_api_allowed") is not True: raise ValueError("runtime_surface: read_only_api_allowed must remain true") blocked_runtime_ops = { "live_k8s_query_allowed", "kubectl_allowed", "rollout_allowed", "restart_allowed", "scale_allowed", "delete_allowed", "secret_read_allowed", "secret_plaintext_allowed", "active_scan_allowed", "production_route_change_allowed", } allowed_ops = sorted(name for name in blocked_runtime_ops if runtime_operations.get(name) is not False) if allowed_ops: raise ValueError(f"runtime_surface: operation boundaries must stay false: {allowed_ops}") ai_program = snapshots["ai_agent_automation"].get("program_status") or {} if ai_program.get("read_only_mode") is not True: raise ValueError("ai_agent_automation: read_only_mode must remain true") ai_approvals = snapshots["ai_agent_automation"].get("approval_boundaries") or {} allowed_ai = sorted(name for name, value in ai_approvals.items() if value is not False) if allowed_ai: raise ValueError(f"ai_agent_automation: approval boundaries must stay false: {allowed_ai}") agent_bounty_summary = snapshots["agent_bounty"].get("summary") or {} blocked_agent_bounty = { "owner_response_received", "owner_response_accepted", "repo_refs_truth_accepted", "data_classification_accepted", "deployment_boundary_accepted", "external_agent_boundary_accepted", "runtime_execution_authorized", "runtime_gate_open", "production_deploy_authorized", "workflow_modification_authorized", "secret_value_collection_authorized", "external_agent_autonomy_authorized", "auto_claim_submit_authorized", "bounty_payout_authorized", } allowed_agent_bounty = sorted( name for name in blocked_agent_bounty if agent_bounty_summary.get(name) is not False ) if allowed_agent_bounty: raise ValueError(f"agent_bounty: blocked summary flags must stay false: {allowed_agent_bounty}") def _build_domains(snapshots: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: asset_summary = snapshots["asset_ledger"].get("summary") or {} host_summary = snapshots["host_service"].get("summary") or {} monitoring_summary = snapshots["monitoring"].get("summary") or {} ssh_summary = snapshots["ssh_network"].get("summary") or {} wazuh_summary = snapshots["wazuh_hosts"].get("summary") or {} runtime_rollups = snapshots["runtime_surface"].get("rollups") or {} runtime_program = snapshots["runtime_surface"].get("program_status") or {} agent_bounty_summary = snapshots["agent_bounty"].get("summary") or {} ai_program = snapshots["ai_agent_automation"].get("program_status") or {} ai_assets = snapshots["ai_agent_automation"].get("assets") or [] return [ { "domain_id": "high_value_asset_control", "label": "高價值資產與配置總帳", "priority": "P0", "coverage_percent": _as_int(asset_summary.get("security_asset_control_ledger_completion_percent")), "scope_count": _as_int(asset_summary.get("asset_group_count")), "write_capable_count": _as_int(asset_summary.get("c0_asset_group_count")), "accepted_count": _as_int(asset_summary.get("owner_response_accepted_count")), "blocked_count": _as_int(asset_summary.get("owner_packet_required_count")), "status": "owner_packet_required", "next_gate": "owner_packet_and_redacted_live_evidence", "source_refs": ["docs/security/security-asset-control-ledger.snapshot.json"], }, { "domain_id": "host_service_runtime", "label": "主機服務 / Docker / systemd", "priority": "P0", "coverage_percent": _as_int(host_summary.get("coverage_percent_after_inventory")), "scope_count": _as_int(host_summary.get("surface_count")), "write_capable_count": _as_int(host_summary.get("write_capable_surface_count")), "accepted_count": _as_int(host_summary.get("owner_response_accepted_count")), "blocked_count": _as_int(host_summary.get("surfaces_requiring_owner_response_count")), "status": "waiting_live_hash_and_owner_response", "next_gate": "host_live_hash_restart_window_rollback_owner", "source_refs": ["docs/security/host-service-config-inventory.snapshot.json"], }, { "domain_id": "monitoring_alerting_observability", "label": "監控 / 告警 / 可觀測性 / 通知", "priority": "P0", "coverage_percent": _as_int(monitoring_summary.get("coverage_percent_after_inventory")), "scope_count": _as_int(monitoring_summary.get("surface_count")), "write_capable_count": _as_int(monitoring_summary.get("write_capable_surface_count")), "accepted_count": _as_int(monitoring_summary.get("owner_response_accepted_count")), "blocked_count": _as_int(monitoring_summary.get("surfaces_requiring_owner_response_count")), "status": "waiting_receiver_route_and_receipt_evidence", "next_gate": "alert_readability_receiver_receipt_reload_owner", "source_refs": ["docs/security/monitoring-alerting-observability-inventory.snapshot.json"], }, { "domain_id": "ssh_firewall_network_access", "label": "SSH / Firewall / WireGuard / NodePort / NetworkPolicy", "priority": "P0", "coverage_percent": _as_int(ssh_summary.get("coverage_percent_after_inventory")), "scope_count": _as_int(ssh_summary.get("surface_count")), "write_capable_count": _as_int(ssh_summary.get("write_capable_surface_count")), "accepted_count": _as_int(ssh_summary.get("owner_response_accepted_count")), "blocked_count": _as_int(ssh_summary.get("surfaces_requiring_owner_response_count")), "status": "waiting_actor_before_after_and_recurrence_guard", "next_gate": "post_incident_network_change_readback", "source_refs": ["docs/security/ssh-network-access-inventory.snapshot.json"], }, { "domain_id": "awoooi_runtime_surfaces", "label": "AWOOOI runtime 工作負載 / 敏感設定 / Ingress / CronJob", "priority": "P0", "coverage_percent": _as_int(runtime_program.get("overall_completion_percent")), "scope_count": _as_int(runtime_rollups.get("total_surfaces")), "write_capable_count": len(runtime_rollups.get("secret_surface_ids") or []), "accepted_count": 0, "blocked_count": len(runtime_rollups.get("action_required_surface_ids") or []), "status": "manifest_mapped_read_only_runtime_gate_closed", "next_gate": "live_runtime_binding_owner_acceptance", "source_refs": [f"docs/evaluations/{snapshots['runtime_surface']['_snapshot_name']}"], }, { "domain_id": "wazuh_managed_host_coverage", "label": "Wazuh 主機納管與 manager registry", "priority": "P0", "coverage_percent": 70 if _as_int(wazuh_summary.get("manager_registry_accepted_count")) else 0, "scope_count": _as_int(wazuh_summary.get("expected_host_scope_count")), "write_capable_count": _as_int(wazuh_summary.get("agent_reenroll_authorized_count")), "accepted_count": _as_int(wazuh_summary.get("manager_registry_accepted_count")), "blocked_count": max( _as_int(wazuh_summary.get("expected_host_scope_count")) - _as_int(wazuh_summary.get("manager_registry_accepted_count")), 0, ), "status": ( "manager_registry_readback_accepted_runtime_gate_closed" if _as_int(wazuh_summary.get("manager_registry_accepted_count")) else "waiting_manager_registry_readback" ), "next_gate": "runtime_gate_owner_review_and_postcheck", "source_refs": ["docs/security/wazuh-managed-host-coverage-gate.snapshot.json"], }, { "domain_id": "agent_bounty_protocol", "label": "agent-bounty-protocol 產品邊界 / MCP / A2A", "priority": "P0", "coverage_percent": _as_int(agent_bounty_summary.get("onboarding_handoff_completion_percent")), "scope_count": len(snapshots["agent_bounty"].get("product_surfaces") or []), "write_capable_count": 0, "accepted_count": 0, "blocked_count": len(snapshots["agent_bounty"].get("product_surfaces") or []), "status": "draft_waiting_owner_review_runtime_gate_closed", "next_gate": "repo_refs_data_classification_deployment_boundary_owner_acceptance", "source_refs": ["docs/security/agent-bounty-iwooos-onboarding-handoff.snapshot.json"], }, { "domain_id": "ai_agent_automation", "label": "AI Agent / Provider / 自動化資產", "priority": "P0", "coverage_percent": _as_int(ai_program.get("overall_completion_percent")), "scope_count": len(ai_assets), "write_capable_count": 0, "accepted_count": 0, "blocked_count": len( [ asset for asset in ai_assets if isinstance(asset, dict) and asset.get("status") != "done" ] ), "status": "read_only_inventory_runtime_write_gate_closed", "next_gate": "owner_approved_runtime_write_gate_and_replay_shadow_canary", "source_refs": [f"docs/evaluations/{snapshots['ai_agent_automation']['_snapshot_name']}"], }, ] def _build_summary(snapshots: dict[str, dict[str, Any]], domains: list[dict[str, Any]]) -> dict[str, Any]: visible_scope_count = sum(_as_int(domain.get("scope_count")) for domain in domains) write_capable_count = sum(_as_int(domain.get("write_capable_count")) for domain in domains) blocked_count = sum(_as_int(domain.get("blocked_count")) for domain in domains) accepted_count = sum(_as_int(domain.get("accepted_count")) for domain in domains) coverage_values = [_as_int(domain.get("coverage_percent")) for domain in domains] return { "source_snapshot_count": 8, "control_domain_count": len(domains), "visible_scope_unit_count": visible_scope_count, "write_capable_scope_count": write_capable_count, "blocked_scope_count": blocked_count, "accepted_scope_count": accepted_count, "control_plane_visibility_percent": round(sum(coverage_values) / len(coverage_values)), "actual_runtime_acceptance_percent": 0, "runtime_gate_count": 0, "owner_response_received_count": 0, "owner_response_accepted_count": 0, "live_evidence_accepted_count": 0, "wazuh_manager_registry_accepted_count": _summary_int( snapshots["wazuh_hosts"], "manager_registry_accepted_count", ), "active_scan_authorized_count": 0, "active_response_authorized_count": 0, "telegram_send_authorized_count": 0, "host_write_authorized_count": 0, "secret_value_collected_count": 0, "agent_bounty_runtime_gate_open_count": 0, "ai_agent_runtime_write_gate_open_count": 0, "asset_group_count": _summary_int(snapshots["asset_ledger"], "asset_group_count"), "host_service_surface_count": _summary_int(snapshots["host_service"], "surface_count"), "monitoring_surface_count": _summary_int(snapshots["monitoring"], "surface_count"), "ssh_network_surface_count": _summary_int(snapshots["ssh_network"], "surface_count"), "runtime_surface_count": _as_int((snapshots["runtime_surface"].get("rollups") or {}).get("total_surfaces")), "wazuh_expected_host_scope_count": _summary_int(snapshots["wazuh_hosts"], "expected_host_scope_count"), "agent_bounty_product_surface_count": len(snapshots["agent_bounty"].get("product_surfaces") or []), "ai_agent_asset_count": len(snapshots["ai_agent_automation"].get("assets") or []), "all_scope_runtime_controlled": False, "allowlisted_controlled_apply_bypasses_iwooos_ledger": True, "controlled_apply_policy": "low_medium_high_allowed_after_allowlist_check_mode_rollback_verifier_km", "critical_break_glass_required": True, } def _build_p0_next_actions() -> list[dict[str, str]]: return [ { "priority": "P0-01", "title": "全域資安完成口徑與 owner closure", "required_evidence": "每個主機、服務、產品、網站、工具、套件與 AI Agent 都要有 scope、owner、risk、verifier、runtime acceptance 欄位；UI / API / CD 綠燈不得算完成。", }, { "priority": "P0-02", "title": "全資產 / 全主機 / 全服務 scope truth", "required_evidence": "Gitea source truth、runtime surface inventory、host/service live hash、公開入口、runner、backup、AI tool 權限與缺口清單；不得用單一 dashboard 代表全域。", }, { "priority": "P0-03", "title": "Wazuh manager registry、FIM、弱點與告警 receipt", "required_evidence": "manager registry parity、agent active/disconnected rollup、FIM baseline、vulnerability detection rollup、alert sample receipt、TLS/RBAC readback、缺席主機 decision。", }, { "priority": "P0-04", "title": "Host / Docker / systemd 入侵與鑑識基線", "required_evidence": "auth、sudo、process、network、persistence、package、service、Docker event 的脫敏 readback；所有 remediation 先進 check-mode、rollback、post-verifier。", }, { "priority": "P0-05", "title": "Nginx / Gateway / SSH / Firewall / K8s network", "required_evidence": "source-to-live diff、nginx -t 或等價檢核、route smoke、before/after actor、NetworkPolicy / firewall diff、rollback owner。", }, { "priority": "P0-06", "title": "Web / API / AppSec / webhook 防護", "required_evidence": "auth、authorization、session、rate limit、CORS、security headers、webhook abuse case、ASVS mapping、desktop/mobile smoke 與 no sensitive copy。", }, { "priority": "P0-07", "title": "套件 / container / workflow supply-chain", "required_evidence": "SBOM、lockfile diff、image scan、KEV/EPSS priority、runner/workflow diff、artifact provenance、signature verification、package SLA。", }, { "priority": "P0-08", "title": "SIEM / SOAR / Monitoring 告警閉環", "required_evidence": "Wazuh / Alertmanager / Telegram / OCSF / Sigma normalized event、dedupe、noise budget、case id、receiver receipt、operator acknowledgment。", }, { "priority": "P0-09", "title": "Backup / restore / forensic retention", "required_evidence": "restore drill、offsite/escrow、chain of custody、retention policy、backup freshness、rollback proof、post-restore smoke。", }, { "priority": "P0-10", "title": "AI Agent / MCP / RAG permission gate", "required_evidence": "tool allowlist、prompt redaction、secret boundary、RAG source trust、MCP connector scope、cost cap、replay/shadow/canary、KM / PlayBook writeback。", }, { "priority": "P0-11", "title": "AISOC 管理者 UI / UX 控制室", "required_evidence": "單一風險地圖、目前 P0、blocked reason、next executable action、tool coverage、host/service/product/package drilldown；避免以大段文字取代狀態。", }, { "priority": "P0-12", "title": "Controlled apply deployment 與 production readback", "required_evidence": "target selector、source-of-truth diff、check-mode/dry-run、rollback、post-apply verifier、runtime readback、KM / PlayBook trust、deploy marker。", }, ] def _build_professional_security_governance_review( summary: dict[str, Any], domains: list[dict[str, Any]], priority_work_items: list[dict[str, str]], ) -> dict[str, Any]: runtime_acceptance = _as_int(summary.get("actual_runtime_acceptance_percent")) owner_accepted = _as_int(summary.get("owner_response_accepted_count")) runtime_gate_count = _as_int(summary.get("runtime_gate_count")) visible_scope_count = _as_int(summary.get("visible_scope_unit_count")) domain_count = len(domains) root_causes = [ { "finding_id": "F01", "severity": "P0", "title": "控制面可視不等於 runtime 防護生效", "current_evidence": ( f"已可讀 {visible_scope_count} 個納管單位與 {domain_count} 個控制域；" f"runtime acceptance 仍為 {runtime_acceptance}%。" ), "required_closure": "每個 control domain 必須補 live/public-safe readback、post-verifier 與 accepted runtime gate。", }, { "finding_id": "F02", "severity": "P0", "title": "完成定義混用了 UI / API / CD 成功與資安完成", "current_evidence": "Dashboard 200、frontend visible、CD success、route smoke 只能算 delivery evidence。", "required_closure": "把 delivery evidence 與 security runtime acceptance 拆開顯示，所有綠燈必須附 source、runtime、verifier 三段證據。", }, { "finding_id": "F03", "severity": "P0", "title": "Wazuh 已有 registry snapshot，但尚未形成偵測與回應閉環", "current_evidence": "Manager registry accepted 可以計數，但 live metadata、FIM、弱點 rollup、alert receipt 與 active-response gate 仍是不同閉環。", "required_closure": "先完成 Wazuh registry parity、FIM / vulnerability readback、alert receipt，再讓 active response 保持 gated candidate。", }, { "finding_id": "F04", "severity": "P0", "title": "主機、服務、產品、網站、工具與套件缺少同一張優先序", "current_evidence": "既有 domain 涵蓋資產、主機、監控、網路、runtime、Wazuh、產品與 AI Agent，但套件 / 網站 / 工具閉環沒有集中成同一條順序。", "required_closure": "用同一個 P0 list 管理 hosts, services, products, websites, tools, packages and AI automation，不再讓支線互相搶主線。", }, { "finding_id": "F05", "severity": "P0", "title": "AI Agent 自動化還沒有變成可執行的安全閉環", "current_evidence": "AI assets are inventoried, while runtime write gates and action buttons remain closed.", "required_closure": "每個 AI action 必須有 tool allowlist、source diff、check-mode、rollback、post-verifier、cost/privacy gate 與 learning writeback。", }, { "finding_id": "F06", "severity": "P1", "title": "管理者 UI 仍偏文字報告，不是 AISOC 控制台", "current_evidence": "The current board exposes correct boundaries but does not yet compress risk, blockers, next action and tool coverage into one glance.", "required_closure": "新增風險地圖、P0 queue、tool coverage、blocked reason、next executable action and drilldown，讓文字退到 details。", }, ] tool_matrix = [ { "tool_id": "wazuh", "priority": "P0", "role": "XDR / SIEM, agent registry, FIM, vulnerability detection, alert normalization", "integration_state": "partial_registry_readback_runtime_gate_closed", "required_closure": "registry parity、FIM baseline、vulnerability rollup、alert receipt、SOAR candidate gate。", }, { "tool_id": "trivy", "priority": "P0", "role": "container image, filesystem, IaC, dependency and SBOM vulnerability scanning", "integration_state": "required_not_closed", "required_closure": "CI/check-mode scan、KEV/EPSS priority、artifact report、no-secret output guard。", }, { "tool_id": "syft_grype", "priority": "P0", "role": "SBOM generation and vulnerability matching for packages and images", "integration_state": "required_not_closed", "required_closure": "SPDX/CycloneDX SBOM、lockfile diff、package SLA、VEX / accepted-risk register。", }, { "tool_id": "openssf_scorecard", "priority": "P1", "role": "source, build, dependency, testing and maintenance risk scoring", "integration_state": "required_not_closed", "required_closure": "Gitea/source-side hygiene mapping、branch protection equivalent、dependency/update policy readback。", }, { "tool_id": "kyverno_cosign", "priority": "P1", "role": "Kubernetes policy, image signature and attestation verification", "integration_state": "required_when_k8s_runtime_gate_opens", "required_closure": "policy check-mode、image verification、exception register、rollback and post-apply verifier。", }, { "tool_id": "falco", "priority": "P1", "role": "runtime threat detection for Linux and Kubernetes workloads", "integration_state": "required_not_closed", "required_closure": "read-only rule dry-run、event redaction、Wazuh/OCSF correlation、no auto-block until reviewer gate。", }, { "tool_id": "owasp_zap_or_equivalent_dast", "priority": "P1", "role": "web/API dynamic testing with bounded scope", "integration_state": "controlled_scan_required", "required_closure": "explicit target selector、rate limit、test window、baseline scan, no credentialed attack unless break-glass approved。", }, { "tool_id": "sigma_ocsf", "priority": "P1", "role": "portable detection rules and normalized event schema", "integration_state": "required_for_ai_triage_quality", "required_closure": "map Wazuh / Alertmanager / app events into normalized cards and regression-test false positives。", }, ] scope_contract = [ {"scope_id": "hosts", "included": True, "closure_signal": "agent / process / service / network readback"}, {"scope_id": "services", "included": True, "closure_signal": "systemd / Docker / API health and config hash"}, {"scope_id": "products", "included": True, "closure_signal": "owner, data classification, route and deploy boundary"}, {"scope_id": "websites", "included": True, "closure_signal": "desktop/mobile smoke, headers, auth and webhook abuse review"}, {"scope_id": "tools", "included": True, "closure_signal": "security tool health, version, rule pack and output redaction"}, {"scope_id": "packages", "included": True, "closure_signal": "SBOM, lockfile diff, vulnerability SLA and provenance"}, {"scope_id": "ai_agents", "included": True, "closure_signal": "tool allowlist, MCP/RAG trust, replay/shadow/canary"}, {"scope_id": "runtime_actions", "included": True, "closure_signal": "target selector, check-mode, rollback and verifier"}, ] ai_loop = [ {"stage_id": "collect", "gate": "public_safe_readback_only", "runtime_write": False}, {"stage_id": "normalize", "gate": "redaction_and_schema_guard", "runtime_write": False}, {"stage_id": "triage", "gate": "severity_scope_confidence_scoring", "runtime_write": False}, {"stage_id": "plan", "gate": "target_selector_source_diff_check_mode_rollback", "runtime_write": False}, {"stage_id": "apply", "gate": "allowlisted_controlled_apply_only", "runtime_write": runtime_gate_count > 0}, {"stage_id": "verify", "gate": "post_apply_readback_and_no_false_green", "runtime_write": False}, {"stage_id": "learn", "gate": "KM_PlayBook_trust_writeback", "runtime_write": False}, ] ui_requirements = [ "one_glance_current_p0_blocker_next_action", "host_service_product_package_tool_coverage_map", "wazuh_registry_fim_vulnerability_alert_receipt_status", "runtime_acceptance_not_confused_with_deploy_success", "ai_agent_action_queue_with_check_mode_verifier_state", "details_only_for_long_boundary_text", ] acceptance_definition = [ "all_scope_items_have_owner_scope_risk_and_verifier", "wazuh_registry_fim_vulnerability_and_alert_receipt_readback_green", "critical_tools_have_health_version_rule_and_redaction_readback", "packages_and_images_have_sbom_vulnerability_and_provenance_receipts", "ai_agent_actions_have_allowlist_check_mode_rollback_post_verifier", "production_dashboard_shows_runtime_acceptance_separate_from_delivery", ] return { "schema_version": "iwooos_security_governance_professional_review_v1", "status": "review_updated_runtime_closure_required", "review_scope_statement": ( "Covers hosts, services, products, websites, tools, packages, security tools, " "runtime actions, AI Agents, MCP/RAG and production readback." ), "completion_diagnosis": { "control_plane_visibility_percent": _as_int( summary.get("control_plane_visibility_percent") ), "actual_runtime_acceptance_percent": runtime_acceptance, "owner_response_accepted_count": owner_accepted, "runtime_gate_count": runtime_gate_count, "visible_scope_unit_count": visible_scope_count, "control_domain_count": domain_count, "diagnosis": "control_plane_exists_runtime_security_closure_not_complete", }, "root_cause_summary": root_causes, "scope_contract": scope_contract, "priority_work_items": priority_work_items, "security_tool_integration_matrix": tool_matrix, "ai_automation_closure_loop": ai_loop, "ui_ux_control_room_requirements": ui_requirements, "acceptance_definition": acceptance_definition, "boundaries": { "secret_value_collection_allowed": False, "raw_session_or_sqlite_read_allowed": False, "active_scan_without_target_selector_allowed": False, "auto_block_without_reviewer_allowed": False, "runtime_write_without_post_verifier_allowed": False, }, } def _summary_int(snapshot: dict[str, Any], key: str) -> int: return _as_int((snapshot.get("summary") or {}).get(key)) def _as_int(value: Any) -> int: return value if isinstance(value, int) else 0