feat(iwooos): expose high-value config control coverage
Some checks failed
Code Review / ai-code-review (push) Successful in 15s
CD Pipeline / tests (push) Successful in 1m39s
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled

This commit is contained in:
Your Name
2026-06-27 13:14:23 +08:00
parent 9d2a443536
commit 121e5b8861
9 changed files with 564 additions and 5 deletions

View File

@@ -17,6 +17,9 @@ from fastapi.responses import JSONResponse
from src.services.iwooos_runtime_security_readback import (
load_latest_iwooos_runtime_security_readback,
)
from src.services.iwooos_high_value_config_control_coverage import (
load_latest_iwooos_high_value_config_control_coverage,
)
from src.services.iwooos_security_control_coverage import (
load_latest_iwooos_security_control_coverage,
)
@@ -168,3 +171,31 @@ async def get_iwooos_security_control_coverage() -> dict[str, Any]:
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"IwoooS security control coverage 無效:{exc}",
) from exc
@router.get(
"/api/v1/iwooos/high-value-config-control-coverage",
response_model=dict[str, Any],
summary="取得 IwoooS 高價值配置控管覆蓋矩陣",
description=(
"讀取已提交的高價值配置控管 snapshot回傳 Nginx、DNS / TLS、K8s、"
"Secrets、runner、Firewall、Backup、AI provider 與 agent-bounty runtime 的"
"公開安全只讀投影。此端點不查 live host、不讀 secret、不執行 nginx -t、"
"不 reload、不 sync、不啟動掃描、不開 runtime gate。"
),
)
async def get_iwooos_high_value_config_control_coverage() -> dict[str, Any]:
"""回傳高價值配置控管矩陣公開安全只讀狀態。"""
try:
payload = await asyncio.to_thread(load_latest_iwooos_high_value_config_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 high-value config control coverage 無效:{exc}",
) from exc

View File

@@ -0,0 +1,213 @@
"""
IwoooS 高價值配置控管覆蓋矩陣只讀讀回。
此服務只回傳已提交 snapshot 的公開安全投影,不查 live host、Nginx、
DNS、Kubernetes、Docker、runner、scanner、Telegram 或 secret store。
"""
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_high_value_config_control_coverage_v1"
_SOURCE_SCHEMA_VERSION = "high_value_config_control_coverage_v1"
_SNAPSHOT_NAME = "high-value-config-control-coverage.snapshot.json"
_PUBLIC_LABELS = {
"nginx_public_gateway": "Nginx / 公開入口 / 反向代理",
"dns_tls_certbot": "DNS / TLS / 憑證續期",
"k8s_production_gitops": "K8s / ArgoCD / 正式環境宣告",
"secret_metadata": "機密中繼資料 / 注入 / 遮罩",
"gitea_workflow_runner_source_control": "Gitea 工作流程 / 執行器 / 原始碼控制",
"public_admin_api_runtime_config": "公開 / 後台 / API runtime config",
"backup_restore_credential": "備份 / 還原 / 金庫",
"agent_bounty_protocol_runtime": "代理賞金協議 runtime / MCP / A2A / 金流邊界",
"monitoring_alerting_observability": "監控 / 告警 / 可觀測性",
"docker_compose_systemd_host_config": "Docker / systemd / 主機服務配置",
"ssh_firewall_network_access": "SSH / 防火牆 / 網路存取",
"ai_provider_model_routing": "AI 供應商 / 模型路由",
"product_surface_runtime_routes": "產品前後台 / 路由 / webhook",
"security_evidence_tooling": "資安證據 / snapshot / guard 工具",
}
_FALSE_BOUNDARY_KEYS = {
"runtime_execution_authorized",
"host_write_authorized",
"host_live_conf_read_authorized",
"nginx_test_authorized",
"nginx_reload_authorized",
"public_gateway_reload_authorized",
"dns_tls_change_authorized",
"certbot_renew_authorized",
"argocd_sync_authorized",
"kubectl_action_authorized",
"workflow_modification_authorized",
"runner_change_authorized",
"secret_value_collection_allowed",
"backup_run_authorized",
"restore_run_authorized",
"active_scan_authorized",
"agent_bounty_runtime_authorized",
"payout_or_withdrawal_authorized",
"action_buttons_allowed",
}
def load_latest_iwooos_high_value_config_control_coverage(
security_dir: Path | None = None,
) -> dict[str, Any]:
"""載入已提交的高價值配置控管 snapshot。"""
sec_dir = security_dir or _DEFAULT_SECURITY_DIR
snapshot = _load_json(sec_dir / _SNAPSHOT_NAME)
_validate_snapshot(snapshot)
categories = snapshot.get("coverage_categories") or []
category_by_id = {
category["category_id"]: _public_category(category)
for category in categories
if isinstance(category, dict) and category.get("category_id")
}
priority_lanes = [
category_by_id[category_id]
for category_id in snapshot.get("next_collection_order") or []
if category_id in category_by_id
]
return {
"schema_version": _SCHEMA_VERSION,
"source_schema_version": _SOURCE_SCHEMA_VERSION,
"status": "coverage_matrix_ready_no_runtime_action",
"mode": "committed_snapshot_projection_only_no_live_runtime_query",
"summary": snapshot["summary"],
"priority_lanes": priority_lanes,
"lowest_coverage_categories": [
category_by_id[item["category_id"]]
for item in snapshot.get("lowest_coverage_categories") or []
if isinstance(item, dict) and item.get("category_id") in category_by_id
],
"boundary_markers": _build_boundary_markers(snapshot),
"execution_boundaries": _public_boundaries(snapshot),
"no_false_green_rules": [
"高價值配置已註冊不代表 Nginx 可 reload、ArgoCD 可 sync、workflow 可修改或主機可寫入。",
"Nginx、DNS / TLS、K8s、secret、runner、防火牆、備份與 AI provider 只能先收 owner-provided redacted evidence。",
"前台顯示、API 200、snapshot 可讀或一般工作批准,都不能當成 runtime gate 已開。",
"owner response received / accepted、live evidence accepted、runtime gate 與 action button 必須維持 0直到獨立驗收通過。",
],
"source_refs": [
"docs/security/high-value-config-control-coverage.snapshot.json",
"scripts/security/high-value-config-control-coverage.py",
"scripts/security/iwooos-config-control-guard.py",
],
}
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")
return payload
def _validate_snapshot(snapshot: dict[str, Any]) -> None:
if snapshot.get("schema_version") != _SOURCE_SCHEMA_VERSION:
raise ValueError("high-value config coverage schema mismatch")
summary = snapshot.get("summary") or {}
required_summary = {
"category_count": 14,
"c0_category_count": 8,
"owner_response_received_count": 0,
"owner_response_accepted_count": 0,
"runtime_gate_count": 0,
"action_button_count": 0,
}
for key, expected in required_summary.items():
if summary.get(key) != expected:
raise ValueError(f"summary.{key} must be {expected!r}")
boundaries = snapshot.get("execution_boundaries") or {}
opened = sorted(key for key in _FALSE_BOUNDARY_KEYS if boundaries.get(key) is not False)
if opened:
raise ValueError(f"execution boundaries must stay false: {opened}")
categories = snapshot.get("coverage_categories") or []
if len(categories) != summary["category_count"]:
raise ValueError("coverage category count mismatch")
for category in categories:
if not isinstance(category, dict):
raise ValueError("coverage category must be an object")
category_id = category.get("category_id")
if category_id not in _PUBLIC_LABELS:
raise ValueError(f"unknown high-value config category: {category_id!r}")
for key in (
"owner_response_received",
"owner_response_accepted",
"runtime_gate_open",
"action_buttons_allowed",
):
if category.get(key) is not False:
raise ValueError(f"{category_id}.{key} must remain false")
def _public_category(category: dict[str, Any]) -> dict[str, Any]:
category_id = str(category["category_id"])
evidence_refs = category.get("evidence_refs") or []
required_validation = category.get("required_validation") or []
return {
"category_id": category_id,
"label": _PUBLIC_LABELS[category_id],
"priority": str(category.get("priority") or ""),
"control_tier": str(category.get("control_tier") or ""),
"coverage_percent": _as_int(category.get("coverage_percent")),
"coverage_status": str(category.get("coverage_status") or ""),
"current_gap": str(category.get("current_gap") or ""),
"next_owner_action": str(category.get("next_owner_action") or ""),
"owner_response_required": category.get("owner_response_required") is True,
"owner_response_received": False,
"owner_response_accepted": False,
"runtime_gate_open": False,
"action_buttons_allowed": False,
"evidence_ref_count": len(evidence_refs) if isinstance(evidence_refs, list) else 0,
"required_validation_count": len(required_validation) if isinstance(required_validation, list) else 0,
}
def _build_boundary_markers(snapshot: dict[str, Any]) -> list[str]:
summary = snapshot.get("summary") or {}
markers = [
f"high_value_config_control_coverage_category_count={summary.get('category_count', 0)}",
f"high_value_config_control_coverage_c0_category_count={summary.get('c0_category_count', 0)}",
f"high_value_config_control_coverage_c1_category_count={summary.get('c1_category_count', 0)}",
f"high_value_config_control_coverage_average_percent={summary.get('average_coverage_percent', 0)}",
f"high_value_config_control_coverage_needs_live_evidence_count={summary.get('needs_live_evidence_count', 0)}",
f"high_value_config_control_coverage_owner_response_required_count={summary.get('owner_response_required_count', 0)}",
"high_value_config_control_coverage_owner_response_received_count=0",
"high_value_config_control_coverage_owner_response_accepted_count=0",
"high_value_config_control_coverage_runtime_gate_count=0",
"high_value_config_control_coverage_action_button_count=0",
]
return markers
def _public_boundaries(snapshot: dict[str, Any]) -> dict[str, bool]:
source_boundaries = snapshot.get("execution_boundaries") or {}
boundaries = {
key: False for key in sorted(_FALSE_BOUNDARY_KEYS)
}
for key in boundaries:
if source_boundaries.get(key) is not False:
raise ValueError(f"execution boundary must stay false: {key}")
boundaries[key] = False
boundaries["not_authorization"] = True
return boundaries
def _as_int(value: Any) -> int:
return value if isinstance(value, int) else 0