a5ff039803
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m36s
CD Pipeline / build-and-deploy (push) Successful in 5m15s
CD Pipeline / post-deploy-checks (push) Successful in 3m44s
697 lines
37 KiB
Python
697 lines
37 KiB
Python
"""
|
||
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"
|
||
_REQUIRED_CONTROLLED_APPLY_STAGES = [
|
||
"sensor_source_receipt",
|
||
"normalized_asset_identity",
|
||
"source_truth_diff",
|
||
"ai_decision_candidate_action",
|
||
"risk_policy_decision",
|
||
"check_mode_dry_run_receipt",
|
||
"bounded_execution_receipt",
|
||
"independent_post_verifier",
|
||
"km_rag_mcp_playbook_write_ack",
|
||
]
|
||
|
||
_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 執行候選，但任何閉環宣稱都必須寫回 IWOOOS / AISOC ledger receipt。",
|
||
"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": False,
|
||
"allowlisted_controlled_apply_requires_iwooos_ledger_receipt": True,
|
||
"controlled_apply_ledger_contract": "same_run_iwooos_aisoc_trace_required",
|
||
"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, Any]]:
|
||
actions: list[dict[str, Any]] = [
|
||
{
|
||
"priority": "P0-01",
|
||
"title": "全域資安完成口徑與 owner closure",
|
||
"required_evidence": "每個主機、服務、產品、網站、工具、套件與 AI Agent 都要有 scope、owner、risk、verifier、runtime acceptance 欄位；UI / API / CD 綠燈不得算完成。",
|
||
"asset_scope_ids": ["hosts", "services", "products", "websites", "tools", "packages", "ai_agents"],
|
||
"next_executable_action": "bind_security_program_queue_to_same_run_runtime_receipt_contract",
|
||
},
|
||
{
|
||
"priority": "P0-02",
|
||
"title": "全資產 / 全主機 / 全服務 scope truth",
|
||
"required_evidence": "Gitea source truth、runtime surface inventory、host/service live hash、公開入口、runner、backup、AI tool 權限與缺口清單；不得用單一 dashboard 代表全域。",
|
||
"asset_scope_ids": ["hosts", "services", "products", "websites"],
|
||
"next_executable_action": "reconcile_committed_inventory_with_production_readback_and_create_gap_work_items",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["hosts", "services", "tools", "observability_and_security"],
|
||
"next_executable_action": "attach_wazuh_fim_vulnerability_alert_case_and_post_verifier_receipts",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["hosts", "services", "packages"],
|
||
"next_executable_action": "build_public_safe_host_service_forensics_baseline_and_ansible_check_mode_candidates",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["hosts", "services", "websites"],
|
||
"next_executable_action": "stage_network_gateway_source_live_diff_check_mode_and_rollback_verifiers",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["products", "websites", "services"],
|
||
"next_executable_action": "create_bounded_appsec_canary_and_webhook_abuse_regression_receipts",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["products", "packages", "tools"],
|
||
"next_executable_action": "attach_sbom_image_scan_workflow_provenance_and_no_secret_output_receipts",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["observability_and_security", "tools", "ai_agents"],
|
||
"next_executable_action": "normalize_alert_events_to_cases_and_queue_controlled_response_candidates",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["data_and_backup", "products", "services"],
|
||
"next_executable_action": "bind_backup_restore_drill_receipts_to_security_incident_closure",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["ai_agents", "tools", "data_and_backup"],
|
||
"next_executable_action": "build_cross_product_mcp_rag_capability_registry_and_runtime_receipt_gate",
|
||
},
|
||
{
|
||
"priority": "P0-11",
|
||
"title": "AISOC 管理者 UI / UX 控制室",
|
||
"required_evidence": "單一風險地圖、目前 P0、blocked reason、next executable action、tool coverage、host/service/product/package drilldown；避免以大段文字取代狀態。",
|
||
"asset_scope_ids": ["websites", "products", "observability_and_security"],
|
||
"next_executable_action": "replace_text_wall_with_first_viewport_risk_map_queue_and_drilldown_cockpit",
|
||
},
|
||
{
|
||
"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。",
|
||
"asset_scope_ids": ["hosts", "services", "products", "tools", "ai_agents"],
|
||
"next_executable_action": "run_allowlisted_same_run_controlled_apply_chain_and_write_iwooos_ledger_receipt",
|
||
},
|
||
]
|
||
for action in actions:
|
||
action["work_item_id"] = f"IWOOOS-SEC-{action['priority']}"
|
||
action["status"] = "runtime_closure_required"
|
||
action["source_truth_layer"] = "production_runtime_receipt_first"
|
||
action["runtime_credit_policy"] = "no_credit_until_same_run_iwooos_aisoc_receipts_exist"
|
||
action["requires_iwooos_ledger_receipt"] = True
|
||
action["required_controlled_apply_stages"] = list(_REQUIRED_CONTROLLED_APPLY_STAGES)
|
||
action["critical_break_glass_required"] = True
|
||
action["controlled_apply_allowed_risk_levels"] = ["low", "medium", "high"]
|
||
return actions
|
||
|
||
|
||
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
|