wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

feat(iwooos): add security control coverage board
Some checks failed
Code Review / ai-code-review (push) Successful in 14s
Details
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
Details
CD Pipeline / build-and-deploy (push) Has been cancelled
Details
CD Pipeline / post-deploy-checks (push) Has been cancelled
Details
CD Pipeline / tests (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Your Name
2026-06-26 18:18:45 +08:00
parent f8c6dd7284
commit c8912204ce
8 changed files with 1021 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
apps/api/src/api/v1/iwooos.py
View File

@@ -21,6 +21,9 @@ from fastapi.responses import JSONResponse
from src.services.iwooos_runtime_security_readback import (
load_latest_iwooos_runtime_security_readback,
)
from src.services.iwooos_security_control_coverage import (
load_latest_iwooos_security_control_coverage,
)
from src.services.public_redaction import redact_public_lan_topology
@@ -232,3 +235,30 @@ async def get_iwooos_runtime_security_readback() -> dict[str, Any]:
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"IwoooS runtime security readback 無效：{exc}",
) from exc
@router.get(
"/api/v1/iwooos/security-control-coverage",
response_model=dict[str, Any],
summary="取得 IwoooS 資安納管覆蓋總表",
description=(
"彙整已提交的主機、產品、服務、配置、監控、Wazuh、AI Agent 與 agent-bounty "
"資安納管 snapshot，形成只讀覆蓋總表。此端點不查 live host、不讀 secret、不啟動掃描、"
"不送告警、不開 runtime gate。"
),
)
async def get_iwooos_security_control_coverage() -> dict[str, Any]:
"""回傳 IwoooS 資安納管覆蓋只讀總表。"""
try:
payload = await asyncio.to_thread(load_latest_iwooos_security_control_coverage)
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 security control coverage 無效：{exc}",
) from exc

386
apps/api/src/services/iwooos_security_control_coverage.py Normal file
View File

@@ -0,0 +1,386 @@
"""
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)
return {
"schema_version": _SCHEMA_VERSION,
"status": "committed_scope_rollup_ready_runtime_control_blocked",
"mode": "committed_snapshot_rollup_only_no_live_runtime_query",
"summary": summary,
"domains": domains,
"p0_next_actions": _build_p0_next_actions(),
"no_false_green_rules": [
"納管覆蓋總表只代表 committed snapshot 可讀，不代表所有主機已被 Wazuh manager registry 驗收。",
"route 200、transport observed、UI 可見、一般工作批准都不能當成 runtime 授權。",
"owner response received / accepted、live evidence accepted、active scan、active response、Telegram send、host write 仍必須維持 0 / false，直到有脫敏證據與明確維護窗口。",
"Nginx、Firewall、Workflow、Secret、K8s、Docker、systemd、AI Agent provider 變更都必須先進 owner packet 與 guard，不得直接套用。",
],
"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 工作負載 / Secret / 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": 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": _as_int(wazuh_summary.get("expected_host_scope_count")),
"status": "waiting_manager_registry_readback",
"next_gate": "manager_registry_cross_check_all_expected_hosts",
"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": 0,
"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,
}
def _build_p0_next_actions() -> list[dict[str, str]]:
return [
{
"priority": "P0-01",
"title": "Wazuh manager registry 全主機交叉驗收",
"required_evidence": "manager registry、agent status、dashboard API / RBAC / TLS readback、缺席主機 owner decision。",
},
{
"priority": "P0-02",
"title": "Host / Docker / systemd live hash 與維護窗口",
"required_evidence": "live config hash、restart window、rollback owner、post-check 指標、drift disposition。",
},
{
"priority": "P0-03",
"title": "Nginx / 公開入口 / Firewall 變更控管",
"required_evidence": "脫敏 live conf export、rendered diff、nginx -t、route smoke、actor before / after。",
},
{
"priority": "P0-04",
"title": "監控告警卡片化與 receipt",
"required_evidence": "可讀卡片格式、receiver route、Telegram / webhook receipt、silence / dedup / inhibit review。",
},
{
"priority": "P0-05",
"title": "AI Agent runtime write gate",
"required_evidence": "人工批准、replay / shadow / canary、成本與安全邊界、post-write verifier。",
},
{
"priority": "P0-06",
"title": "agent-bounty-protocol 安全納管",
"required_evidence": "repo refs truth、data classification、MCP / A2A abuse boundary、deployment boundary、owner acceptance。",
},
]
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