From d73436276835557fb5a62bbbc2cbd3eaee8fedbf Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 12:42:33 +0800 Subject: [PATCH] feat(security): expand canonical asset coverage --- apps/api/src/jobs/asset_scanner_job.py | 748 +++++++++++++++--- .../iwooos_security_asset_control_plane.py | 66 +- apps/api/tests/test_asset_scanner_job.py | 383 ++++++++- ...est_iwooos_security_asset_control_plane.py | 23 + .../security-asset-control-plane-cockpit.tsx | 14 +- .../security-asset-control-plane.test.ts | 15 + .../src/lib/security-asset-control-plane.ts | 15 + 7 files changed, 1134 insertions(+), 130 deletions(-) diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 7875c94ea..7ba0c5b04 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -28,8 +28,10 @@ ADR-090 監控盲區治理 § Phase 7 Asset Inventory Foundation from __future__ import annotations import asyncio +import hashlib import json as _json import time as _time +from pathlib import Path from typing import Any import structlog @@ -39,16 +41,21 @@ logger = structlog.get_logger(__name__) # ============================================================================ # 排程參數 # ============================================================================ -_SCAN_INTERVAL_SEC = 3600 # 每 1 小時 -_FIRST_DELAY_SEC = 60 # 啟動後等 60s 再首掃 (其他 service init) +_SCAN_INTERVAL_SEC = 3600 # 每 1 小時 +_FIRST_DELAY_SEC = 60 # 啟動後等 60s 再首掃 (其他 service init) _KUBECTL_TIMEOUT_SEC = 30 -_LOOP_BACKOFF_SEC = 300 # 異常時 backoff 5 分鐘 +_LOOP_BACKOFF_SEC = 300 # 異常時 backoff 5 分鐘 _PROJECT_ID = "awoooi" # 7 個自動化覆蓋維度 (ADR-090 §3.5) _COVERAGE_DIMENSIONS = ( - "auto_monitoring", "auto_alerting", "auto_rule_creation", - "auto_rule_matching", "auto_playbook", "auto_remediation", "auto_km_creation", + "auto_monitoring", + "auto_alerting", + "auto_rule_creation", + "auto_rule_matching", + "auto_playbook", + "auto_remediation", + "auto_km_creation", ) # K8s asset_type 對應 @@ -67,6 +74,7 @@ _K8S_RESOURCE_TO_ASSET_TYPE = { # Public entry — main.py lifespan 呼叫 # ============================================================================ + async def run_asset_scanner_loop() -> None: """ 永久迴圈：每 _SCAN_INTERVAL_SEC 秒做一次資產盤點。 @@ -90,7 +98,13 @@ async def run_asset_scanner_loop() -> None: async def scan_once( triggered_by: str = "cron", - scope: tuple[str, ...] = ("k8s", "prometheus", "database_catalog"), + scope: tuple[str, ...] = ( + "k8s", + "prometheus", + "database_catalog", + "ai_catalog", + "model_registry", + ), scan_depth: str = "shallow", ) -> str: """ @@ -150,11 +164,11 @@ async def scan_once( # 為每個 active asset 寫 7 維 coverage (預設 unknown,後續其他 service 升級為 green/yellow/red) await _write_coverage_snapshots(run_id) - logger.info("asset_scan_relationships_written", run_id=run_id, relationships=rel_written) + logger.info( + "asset_scan_relationships_written", run_id=run_id, relationships=rel_written + ) if collector_errors: - raise RuntimeError( - "asset_collector_failures=" + ",".join(collector_errors) - ) + raise RuntimeError("asset_collector_failures=" + ",".join(collector_errors)) except Exception as e: error_msg = f"{type(e).__name__}: {e}"[:1000] @@ -196,7 +210,10 @@ async def scan_once( # K8s 資產收集 # ============================================================================ -async def _fetch_kubectl_json(resource: str, all_namespaces: bool = True) -> dict[str, Any]: + +async def _fetch_kubectl_json( + resource: str, all_namespaces: bool = True +) -> dict[str, Any]: """ subprocess 執行 kubectl get --all-namespaces -o json (或 nodes 不帶 ns). 回傳 parse 後的 payload dict ({'items': [...]}). @@ -212,9 +229,13 @@ async def _fetch_kubectl_json(resource: str, all_namespaces: bool = True) -> dic ), timeout=_KUBECTL_TIMEOUT_SEC, ) - stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=_KUBECTL_TIMEOUT_SEC) + stdout, stderr = await asyncio.wait_for( + proc.communicate(), timeout=_KUBECTL_TIMEOUT_SEC + ) if proc.returncode != 0: - raise RuntimeError(f"kubectl {resource} failed rc={proc.returncode}: {stderr.decode('utf-8', errors='replace')[:300]}") + raise RuntimeError( + f"kubectl {resource} failed rc={proc.returncode}: {stderr.decode('utf-8', errors='replace')[:300]}" + ) try: return _json.loads(stdout.decode("utf-8", errors="replace")) except _json.JSONDecodeError as e: @@ -282,7 +303,9 @@ def _build_workload_asset(item: dict[str, Any], kind: str) -> dict[str, Any]: "ready_replicas": status.get("readyReplicas"), "selector": (spec.get("selector", {}) or {}).get("matchLabels", {}), }, - "tags": [f"kind:{kind}"] + [f"app:{labels['app']}"] if labels.get("app") else [f"kind:{kind}"], + "tags": [f"kind:{kind}"] + [f"app:{labels['app']}"] + if labels.get("app") + else [f"kind:{kind}"], } @@ -317,7 +340,9 @@ def _build_node_asset(item: dict[str, Any]) -> dict[str, Any]: labels = meta.get("labels", {}) or {} status = item.get("status", {}) or {} addresses = status.get("addresses", []) or [] - internal_ip = next((a["address"] for a in addresses if a.get("type") == "InternalIP"), "") + internal_ip = next( + (a["address"] for a in addresses if a.get("type") == "InternalIP"), "" + ) return { "asset_key": f"k8s/node/{name}", @@ -361,12 +386,211 @@ def _build_configmap_asset(item: dict[str, Any]) -> dict[str, Any]: } -async def _collect_runtime_assets() -> tuple[ - list[dict[str, Any]], - list[dict[str, str]], - tuple[str, ...], - tuple[str, ...], -]: +def _build_pvc_asset(item: dict[str, Any]) -> dict[str, Any]: + """PersistentVolumeClaim metadata -> volume asset without reading data.""" + meta = item.get("metadata", {}) or {} + spec = item.get("spec", {}) or {} + status = item.get("status", {}) or {} + ns = meta.get("namespace") or "default" + name = meta.get("name") or "unknown" + + return { + "asset_key": f"k8s/pvc/{ns}/{name}", + "asset_type": "volume", + "host": None, + "namespace": ns, + "name": name, + "metadata": { + "kind": "PersistentVolumeClaim", + "storage_class": spec.get("storageClassName"), + "access_modes": spec.get("accessModes", []), + "volume_mode": spec.get("volumeMode"), + "requested_storage": ( + (spec.get("resources", {}) or {}).get("requests", {}) or {} + ).get("storage"), + "phase": status.get("phase"), + "data_read": False, + }, + "tags": ["kind:PersistentVolumeClaim", "source:k8s_metadata"], + } + + +def _build_cronjob_asset(item: dict[str, Any]) -> dict[str, Any]: + """CronJob scheduling metadata -> scheduled_job asset.""" + meta = item.get("metadata", {}) or {} + spec = item.get("spec", {}) or {} + status = item.get("status", {}) or {} + ns = meta.get("namespace") or "default" + name = meta.get("name") or "unknown" + + return { + "asset_key": f"k8s/cronjob/{ns}/{name}", + "asset_type": "scheduled_job", + "host": None, + "namespace": ns, + "name": name, + "metadata": { + "kind": "CronJob", + "schedule": spec.get("schedule"), + "suspend": bool(spec.get("suspend", False)), + "concurrency_policy": spec.get("concurrencyPolicy"), + "last_schedule_time": status.get("lastScheduleTime"), + "active_job_count": len(status.get("active", []) or []), + "command_or_env_read": False, + }, + "tags": ["kind:CronJob", "source:k8s_metadata"], + } + + +def _build_secret_reference_asset( + namespace: str, + name: str, + *, + reference_kind: str, +) -> dict[str, Any]: + """Create a secret identity from a reference; values are never requested.""" + return { + "asset_key": f"k8s/secret-ref/{namespace}/{name}", + "asset_type": "secret", + "host": None, + "namespace": namespace, + "name": name, + "metadata": { + "kind": "SecretReference", + "reference_kind": reference_kind, + "secret_object_queried": False, + "secret_value_read": False, + }, + "tags": ["classification:secret", "source:k8s_reference"], + } + + +def _collect_ingress_assets( + payload: dict[str, Any], +) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + """Build website/API/certificate assets from Ingress metadata only.""" + assets_by_key: dict[str, dict[str, Any]] = {} + relationships: list[dict[str, str]] = [] + + def route_key(namespace: str, ingress_name: str, host: str, path: str) -> str: + identity = "\x1f".join((namespace, ingress_name, host, path)) + fingerprint = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:20] + return f"k8s/ingress-route/{namespace}/{ingress_name}/{fingerprint}" + + def ensure_website(host: str, namespace: str, ingress_name: str) -> str: + website_key = f"k8s/website/{host}" + website = assets_by_key.get(website_key) + ingress_ref = f"{namespace}/{ingress_name}" + if website is None: + website = { + "asset_key": website_key, + "asset_type": "website", + "host": host, + "namespace": namespace, + "name": host, + "metadata": { + "kind": "IngressHost", + "ingress_refs": [ingress_ref], + "source": "k8s_ingress", + }, + "tags": ["source:k8s_ingress"], + } + assets_by_key[website_key] = website + elif ingress_ref not in website["metadata"]["ingress_refs"]: + website["metadata"]["ingress_refs"].append(ingress_ref) + return website_key + + for item in payload.get("items", []) or []: + meta = item.get("metadata", {}) or {} + spec = item.get("spec", {}) or {} + namespace = meta.get("namespace") or "default" + ingress_name = meta.get("name") or "unknown" + + for rule in spec.get("rules", []) or []: + host = str(rule.get("host") or "").strip() + if not host: + continue + website_key = ensure_website(host, namespace, ingress_name) + http = rule.get("http", {}) or {} + for path_row in http.get("paths", []) or []: + path = str(path_row.get("path") or "/") + backend = path_row.get("backend", {}) or {} + service = backend.get("service", {}) or {} + service_name = service.get("name") or backend.get("serviceName") + route_asset_key = route_key(namespace, ingress_name, host, path) + assets_by_key[route_asset_key] = { + "asset_key": route_asset_key, + "asset_type": "api_endpoint", + "host": host, + "namespace": namespace, + "name": f"{host}{path}", + "metadata": { + "kind": "IngressRoute", + "ingress": ingress_name, + "path": path, + "path_type": path_row.get("pathType"), + "backend_service": service_name, + "request_or_response_body_read": False, + }, + "tags": ["source:k8s_ingress", f"host:{host}"], + } + relationships.append( + { + "from_key": website_key, + "to_key": route_asset_key, + "relationship_type": "routes_to", + } + ) + if service_name: + relationships.append( + { + "from_key": route_asset_key, + "to_key": f"k8s/service/{namespace}/{service_name}", + "relationship_type": "routes_to", + } + ) + + for tls in spec.get("tls", []) or []: + secret_name = str(tls.get("secretName") or "").strip() + if not secret_name: + continue + hosts = sorted({str(host) for host in tls.get("hosts", []) or [] if host}) + certificate_key = f"k8s/certificate-ref/{namespace}/{secret_name}" + assets_by_key[certificate_key] = { + "asset_key": certificate_key, + "asset_type": "certificate", + "host": hosts[0] if len(hosts) == 1 else None, + "namespace": namespace, + "name": secret_name, + "metadata": { + "kind": "IngressTLSReference", + "hosts": hosts, + "secret_object_queried": False, + "certificate_material_read": False, + }, + "tags": ["source:k8s_ingress_tls", "material:not_read"], + } + for host in hosts: + website_key = ensure_website(host, namespace, ingress_name) + relationships.append( + { + "from_key": website_key, + "to_key": certificate_key, + "relationship_type": "authenticates_via", + } + ) + + return list(assets_by_key.values()), relationships + + +async def _collect_runtime_assets() -> ( + tuple[ + list[dict[str, Any]], + list[dict[str, str]], + tuple[str, ...], + tuple[str, ...], + ] +): """ 掃多種 K8s 資源類型 + 提取 relationship. @@ -419,6 +643,7 @@ async def _collect_runtime_assets() -> tuple[ # 2. Pods — 主體 + 從 ownerReferences 建 relationship pod_by_key: dict[str, dict[str, Any]] = {} + secret_reference_assets: dict[str, dict[str, Any]] = {} try: payload = await _fetch_kubectl_json("pods") for item in payload.get("items", []) or []: @@ -435,44 +660,73 @@ async def _collect_runtime_assets() -> tuple[ continue # StatefulSet/DaemonSet 直接 owner Pod,直接建 relationship if owner_kind in ("statefulset", "daemonset"): - relationships.append({ - "from_key": a["asset_key"], - "to_key": f"k8s/{owner_kind}/{ns}/{owner_name}", - "relationship_type": "depends_on", - }) + relationships.append( + { + "from_key": a["asset_key"], + "to_key": f"k8s/{owner_kind}/{ns}/{owner_name}", + "relationship_type": "depends_on", + } + ) # ReplicaSet 中介: 用 rs_to_deployment map 反查 Deployment elif owner_kind == "replicaset": deploy_name = rs_to_deployment.get(f"{ns}/{owner_name}") if deploy_name: - relationships.append({ - "from_key": a["asset_key"], - "to_key": f"k8s/deployment/{ns}/{deploy_name}", - "relationship_type": "depends_on", - }) + relationships.append( + { + "from_key": a["asset_key"], + "to_key": f"k8s/deployment/{ns}/{deploy_name}", + "relationship_type": "depends_on", + } + ) # 極少數直接是 Deployment owner (舊版 K8s) elif owner_kind == "deployment": - relationships.append({ - "from_key": a["asset_key"], - "to_key": f"k8s/deployment/{ns}/{owner_name}", - "relationship_type": "depends_on", - }) + relationships.append( + { + "from_key": a["asset_key"], + "to_key": f"k8s/deployment/{ns}/{owner_name}", + "relationship_type": "depends_on", + } + ) # Pod volumes → ConfigMap relationship for v in (item.get("spec", {}) or {}).get("volumes", []) or []: cm = (v.get("configMap") or {}).get("name") if cm: - relationships.append({ - "from_key": a["asset_key"], - "to_key": f"k8s/configmap/{ns}/{cm}", - "relationship_type": "depends_on", - }) + relationships.append( + { + "from_key": a["asset_key"], + "to_key": f"k8s/configmap/{ns}/{cm}", + "relationship_type": "depends_on", + } + ) + secret_name = (v.get("secret") or {}).get("secretName") + if secret_name: + secret_asset = _build_secret_reference_asset( + ns, + str(secret_name), + reference_kind="PodVolume", + ) + secret_reference_assets[secret_asset["asset_key"]] = secret_asset + relationships.append( + { + "from_key": a["asset_key"], + "to_key": secret_asset["asset_key"], + "relationship_type": "depends_on", + } + ) + assets.extend(secret_reference_assets.values()) reconciled_prefixes.add("k8s/pod/") + reconciled_prefixes.add("k8s/secret-ref/") except Exception as e: logger.warning("collect_pods_failed", error=str(e)) collector_errors.append("k8s_pods") # 3. Deployments / StatefulSets / DaemonSets - for kind, resource in (("Deployment", "deployments"), ("StatefulSet", "statefulsets"), ("DaemonSet", "daemonsets")): + for kind, resource in ( + ("Deployment", "deployments"), + ("StatefulSet", "statefulsets"), + ("DaemonSet", "daemonsets"), + ): try: payload = await _fetch_kubectl_json(resource) for item in payload.get("items", []) or []: @@ -497,14 +751,18 @@ async def _collect_runtime_assets() -> tuple[ for pod_key, pod_item in pod_by_key.items(): if not pod_key.startswith(f"k8s/pod/{svc_ns}/"): continue - pod_labels = (pod_item.get("metadata", {}) or {}).get("labels", {}) or {} + pod_labels = (pod_item.get("metadata", {}) or {}).get( + "labels", {} + ) or {} # selector 所有 kv 必須都在 pod labels 內 if all(pod_labels.get(k) == v for k, v in selector.items()): - relationships.append({ - "from_key": svc["asset_key"], - "to_key": pod_key, - "relationship_type": "routes_to", - }) + relationships.append( + { + "from_key": svc["asset_key"], + "to_key": pod_key, + "relationship_type": "routes_to", + } + ) reconciled_prefixes.add("k8s/service/") except Exception as e: logger.warning("collect_services_failed", error=str(e)) @@ -520,7 +778,40 @@ async def _collect_runtime_assets() -> tuple[ logger.warning("collect_configmaps_failed", error=str(e)) collector_errors.append("k8s_configmaps") - # 6. Prometheus targets — 補齊 host-install services (110/112/188/125 等非 K8s) + # 6. PersistentVolumeClaims — metadata only, never volume contents. + try: + payload = await _fetch_kubectl_json("persistentvolumeclaims") + for item in payload.get("items", []) or []: + assets.append(_build_pvc_asset(item)) + reconciled_prefixes.add("k8s/pvc/") + except Exception as e: + logger.warning("collect_persistentvolumeclaims_failed", error=str(e)) + collector_errors.append("k8s_persistentvolumeclaims") + + # 7. Ingresses — website/API/TLS reference identities, no request or key data. + try: + payload = await _fetch_kubectl_json("ingresses") + ingress_assets, ingress_relationships = _collect_ingress_assets(payload) + assets.extend(ingress_assets) + relationships.extend(ingress_relationships) + reconciled_prefixes.update( + ("k8s/website/", "k8s/ingress-route/", "k8s/certificate-ref/") + ) + except Exception as e: + logger.warning("collect_ingresses_failed", error=str(e)) + collector_errors.append("k8s_ingresses") + + # 8. CronJobs — schedule metadata only, no command or environment values. + try: + payload = await _fetch_kubectl_json("cronjobs") + for item in payload.get("items", []) or []: + assets.append(_build_cronjob_asset(item)) + reconciled_prefixes.add("k8s/cronjob/") + except Exception as e: + logger.warning("collect_cronjobs_failed", error=str(e)) + collector_errors.append("k8s_cronjobs") + + # 9. Prometheus targets — 補齊 host-install services (110/112/188/125 等非 K8s) # Gap 1 修補 (2026-04-19 audit): 原本 asset_inventory 只涵蓋 K8s, # 110 Harbor/Gitea/監控 + 188 PostgreSQL/Redis host-install 全漏 # 用 Prometheus /api/v1/targets 自動發現全節點服務 @@ -533,21 +824,39 @@ async def _collect_runtime_assets() -> tuple[ logger.warning("collect_prometheus_targets_failed", error=str(e)) collector_errors.append("prometheus_targets") - # 7. PostgreSQL catalog — database/table/view metadata only, never row data. + # 10. PostgreSQL catalog — database/table/view metadata only, never row data. try: - database_assets, database_relationships = await _collect_database_catalog_assets() + ( + database_assets, + database_relationships, + ) = await _collect_database_catalog_assets() assets.extend(database_assets) relationships.extend(database_relationships) if database_assets: - reconciled_prefixes.update( - ("postgres/database/", "postgres/relation/") - ) + reconciled_prefixes.update(("postgres/database/", "postgres/relation/")) except Exception as e: logger.warning("collect_database_catalog_failed", error=str(e)) collector_errors.append("database_catalog") + # 11. AI catalog and model registry — aggregate/runtime config metadata only. + try: + ai_assets, ai_relationships = await _collect_ai_catalog_assets() + assets.extend(ai_assets) + relationships.extend(ai_relationships) + reconciled_prefixes.update( + ( + "ai_catalog/agent/", + "ai_catalog/km/", + "model_registry/model/", + "model_registry/provider/", + ) + ) + except Exception as e: + logger.warning("collect_ai_catalog_failed", error=str(e)) + collector_errors.append("ai_catalog") + return ( - assets, + list({asset["asset_key"]: asset for asset in assets}.values()), relationships, tuple(sorted(reconciled_prefixes)), tuple(collector_errors), @@ -576,7 +885,9 @@ def _is_valid_ipv4(s: str) -> bool: return True -async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict[str, str]]]: +async def _collect_prometheus_targets() -> ( + tuple[list[dict[str, Any]], list[dict[str, str]]] +): """ 從 Prometheus /api/v1/targets 發現所有被監控的 host-install service + 主機. @@ -615,12 +926,36 @@ async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict if not host_ip: # target instance 不是 IP 形式 → 建 third_party_service asset 但 host 留空 asset_key = f"prometheus_target/{job}/{instance}" - assets.append({ + assets.append( + { + "asset_key": asset_key, + "asset_type": "third_party_service", + "host": None, + "namespace": None, + "name": f"{job}:{instance}", + "metadata": { + "job": job, + "instance": instance, + "scrape_url": t.get("scrapeUrl"), + "health": t.get("health"), + "labels": labels, + }, + "tags": [f"job:{job}", "source:prometheus_target"], + } + ) + continue + + # IP 形式 target — 用 'monitoring_target' (asset_inventory CHECK 允許列表) + # host_service 不在 ADR-090 asset_type CHECK 內,之前 1 筆 125:32334 scan 拋 + # CheckViolationError (constraint asset_inventory_type_valid) + asset_key = f"prometheus_target/{job}/{instance}" + assets.append( + { "asset_key": asset_key, - "asset_type": "third_party_service", - "host": None, + "asset_type": "monitoring_target", + "host": host_ip, "namespace": None, - "name": f"{job}:{instance}", + "name": f"{job}@{host_ip}", "metadata": { "job": job, "instance": instance, @@ -628,61 +963,47 @@ async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict "health": t.get("health"), "labels": labels, }, - "tags": [f"job:{job}", "source:prometheus_target"], - }) - continue - - # IP 形式 target — 用 'monitoring_target' (asset_inventory CHECK 允許列表) - # host_service 不在 ADR-090 asset_type CHECK 內,之前 1 筆 125:32334 scan 拋 - # CheckViolationError (constraint asset_inventory_type_valid) - asset_key = f"prometheus_target/{job}/{instance}" - assets.append({ - "asset_key": asset_key, - "asset_type": "monitoring_target", - "host": host_ip, - "namespace": None, - "name": f"{job}@{host_ip}", - "metadata": { - "job": job, - "instance": instance, - "scrape_url": t.get("scrapeUrl"), - "health": t.get("health"), - "labels": labels, - }, - "tags": [f"job:{job}", f"host:{host_ip}", "source:prometheus_target"], - }) + "tags": [f"job:{job}", f"host:{host_ip}", "source:prometheus_target"], + } + ) # 對每個 IP 建 host asset (若尚未) if host_ip not in seen_hosts: seen_hosts.add(host_ip) host_key = f"host/{host_ip}" - assets.append({ - "asset_key": host_key, - "asset_type": "host", - "host": host_ip, - "namespace": None, - "name": host_ip, - "metadata": { - "discovered_by": "prometheus_targets", - "source": "blackbox_icmp_or_node_exporter", - }, - "tags": [f"ip:{host_ip}", "source:prometheus"], - }) + assets.append( + { + "asset_key": host_key, + "asset_type": "host", + "host": host_ip, + "namespace": None, + "name": host_ip, + "metadata": { + "discovered_by": "prometheus_targets", + "source": "blackbox_icmp_or_node_exporter", + }, + "tags": [f"ip:{host_ip}", "source:prometheus"], + } + ) # 建 target → host 的 depends_on relationship - relationships.append({ - "from_key": asset_key, - "to_key": f"host/{host_ip}", - "relationship_type": "depends_on", - }) + relationships.append( + { + "from_key": asset_key, + "to_key": f"host/{host_ip}", + "relationship_type": "depends_on", + } + ) - logger.info("prometheus_targets_collected", count=len(assets), hosts=len(seen_hosts)) + logger.info( + "prometheus_targets_collected", count=len(assets), hosts=len(seen_hosts) + ) return assets, relationships -async def _collect_database_catalog_assets() -> tuple[ - list[dict[str, Any]], list[dict[str, str]] -]: +async def _collect_database_catalog_assets() -> ( + tuple[list[dict[str, Any]], list[dict[str, str]]] +): """Discover PostgreSQL schema metadata without reading table contents.""" from sqlalchemy import text as _sql @@ -744,7 +1065,9 @@ async def _collect_database_catalog_assets() -> tuple[ values = row._mapping if hasattr(row, "_mapping") else row schema_name = str(values["schema_name"]) relation_name = str(values["relation_name"]) - relation_key = f"postgres/relation/{database_name}/{schema_name}/{relation_name}" + relation_key = ( + f"postgres/relation/{database_name}/{schema_name}/{relation_name}" + ) assets.append( { "asset_key": relation_key, @@ -782,10 +1105,179 @@ async def _collect_database_catalog_assets() -> tuple[ return assets, relationships +def _timestamp_text(value: Any) -> str | None: + if value is None: + return None + isoformat = getattr(value, "isoformat", None) + if callable(isoformat): + return str(isoformat()) + return str(value) + + +async def _collect_ai_catalog_assets( + model_registry_path: Path | None = None, +) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + """Inventory AI roles, KM categories and configured models without content.""" + from sqlalchemy import text as _sql + + from src.db.base import get_db_context + + path = model_registry_path or Path(__file__).resolve().parents[2] / "models.json" + registry = _json.loads(path.read_text(encoding="utf-8")) + providers = registry.get("providers", {}) or {} + if not isinstance(providers, dict): + raise RuntimeError("model_registry_providers_invalid") + + assets: list[dict[str, Any]] = [] + relationships: list[dict[str, str]] = [] + + for provider, provider_config in sorted(providers.items()): + provider_name = str(provider) + provider_key = f"model_registry/provider/{provider_name}" + assets.append( + { + "asset_key": provider_key, + "asset_type": "third_party_service", + "host": None, + "namespace": None, + "name": provider_name, + "metadata": { + "kind": "AIModelProvider", + "configured": True, + "credential_or_endpoint_read": False, + }, + "tags": ["source:model_registry", f"provider:{provider_name}"], + } + ) + model_purposes: dict[str, list[str]] = {} + if isinstance(provider_config, dict): + models = provider_config.get("models", {}) or {} + if isinstance(models, dict): + for purpose, model_name in models.items(): + if model_name: + model_purposes.setdefault(str(model_name), []).append( + str(purpose) + ) + + for model_name, purposes in sorted(model_purposes.items()): + model_key = f"model_registry/model/{provider_name}/{model_name}" + assets.append( + { + "asset_key": model_key, + "asset_type": "llm_model", + "host": None, + "namespace": None, + "name": model_name, + "metadata": { + "provider": provider_name, + "purposes": sorted(purposes), + "configured": True, + "credential_or_prompt_read": False, + }, + "tags": [ + "source:model_registry", + f"provider:{provider_name}", + ], + } + ) + relationships.append( + { + "from_key": model_key, + "to_key": provider_key, + "relationship_type": "depends_on", + } + ) + + async with get_db_context(_PROJECT_ID) as db: + agent_rows = ( + await db.execute( + _sql( + """ + SELECT + agent_role::text AS agent_role, + COUNT(*) AS session_count, + MAX(created_at) AS latest_seen_at + FROM agent_sessions + WHERE agent_role IS NOT NULL + AND BTRIM(agent_role::text) <> '' + GROUP BY agent_role::text + ORDER BY agent_role::text + """ + ) + ) + ).fetchall() + knowledge_rows = ( + await db.execute( + _sql( + """ + SELECT + entry_type::text AS entry_type, + COUNT(*) AS entry_count, + MAX(updated_at) AS latest_updated_at + FROM knowledge_entries + WHERE project_id = :project_id + GROUP BY entry_type::text + ORDER BY entry_type::text + """ + ), + {"project_id": _PROJECT_ID}, + ) + ).fetchall() + + for row in agent_rows: + values = row._mapping if hasattr(row, "_mapping") else row + role = str(values["agent_role"]) + assets.append( + { + "asset_key": f"ai_catalog/agent/{role}", + "asset_type": "ai_agent", + "host": None, + "namespace": None, + "name": role, + "metadata": { + "kind": "ObservedAgentRole", + "session_count": int(values["session_count"] or 0), + "latest_seen_at": _timestamp_text(values["latest_seen_at"]), + "session_input_or_output_read": False, + }, + "tags": ["source:agent_sessions_aggregate"], + } + ) + + for row in knowledge_rows: + values = row._mapping if hasattr(row, "_mapping") else row + entry_type = str(values["entry_type"]) + assets.append( + { + "asset_key": f"ai_catalog/km/{entry_type}", + "asset_type": "km_entry", + "host": None, + "namespace": None, + "name": entry_type, + "metadata": { + "kind": "KnowledgeEntryTypeAggregate", + "entry_count": int(values["entry_count"] or 0), + "latest_updated_at": _timestamp_text(values["latest_updated_at"]), + "title_or_content_read": False, + }, + "tags": ["source:knowledge_entries_aggregate"], + } + ) + + logger.info( + "ai_catalog_assets_collected", + provider_count=len(providers), + agent_role_count=len(agent_rows), + knowledge_type_count=len(knowledge_rows), + ) + return assets, relationships + + # ============================================================================ # DB 寫入 # ============================================================================ + async def _start_discovery_run( triggered_by: str, scope: list[str], @@ -803,7 +1295,8 @@ async def _start_discovery_run( async with get_db_context(_PROJECT_ID) as db: row = await db.execute( - _sql(""" + _sql( + """ INSERT INTO asset_discovery_run ( triggered_by, scope, scan_depth, status, new_assets, modified_assets, disappeared_assets, @@ -814,7 +1307,8 @@ async def _start_discovery_run( CAST(:tools AS jsonb) ) RETURNING run_id - """), + """ + ), { "tb": triggered_by, "scope": scope, @@ -871,7 +1365,8 @@ async def _finalize_discovery_run( async with get_db_context(_PROJECT_ID) as db: await db.execute( - _sql(""" + _sql( + """ UPDATE asset_discovery_run SET status = :st, ended_at = NOW(), @@ -882,7 +1377,8 @@ async def _finalize_discovery_run( duration_ms = :dur, error = :err WHERE run_id = CAST(:rid AS uuid) - """), + """ + ), { "st": status, "total": total_assets, @@ -895,7 +1391,8 @@ async def _finalize_discovery_run( }, ) await db.execute( - _sql(""" + _sql( + """ INSERT INTO automation_operation_log ( operation_type, actor, status, input, output, run_id, duration_ms, error, tags @@ -909,7 +1406,8 @@ async def _finalize_discovery_run( :dur, :err, :tags ) - """), + """ + ), { "st": "success" if status == "success" else "failed", "input": _json.dumps(input_payload, ensure_ascii=False), @@ -919,7 +1417,7 @@ async def _finalize_discovery_run( "err": (error or "")[:2000] if error else None, "tags": ["asset_scanner", "discovery", *scope], }, - ) + ) async def _assert_asset_key_integrity() -> None: @@ -1001,7 +1499,8 @@ async def _upsert_assets( for a in assets: # xmax = 0 表示 INSERT (新),否則表示 UPDATE (修改) row = await db.execute( - _sql(""" + _sql( + """ INSERT INTO asset_inventory ( asset_key, asset_type, host, namespace, name, metadata, tags, environment, lifecycle_state, @@ -1020,7 +1519,8 @@ async def _upsert_assets( updated_at = NOW(), decommissioned_at = NULL RETURNING asset_id, (xmax = 0) AS inserted - """), + """ + ), { "ak": a["asset_key"], "at": a["asset_type"], @@ -1115,7 +1615,8 @@ async def _upsert_relationships(relationships: list[dict[str, str]]) -> int: try: # 用 asset_key → asset_id 解析,同時 UPSERT await db.execute( - _sql(""" + _sql( + """ INSERT INTO asset_relationship ( from_asset_id, to_asset_id, relationship_type, first_detected_at, last_verified_at, is_active @@ -1127,7 +1628,8 @@ async def _upsert_relationships(relationships: list[dict[str, str]]) -> int: ON CONFLICT (from_asset_id, to_asset_id, relationship_type) DO UPDATE SET last_verified_at = NOW(), is_active = true - """), + """ + ), { "from_key": rel["from_key"], "to_key": rel["to_key"], @@ -1136,8 +1638,12 @@ async def _upsert_relationships(relationships: list[dict[str, str]]) -> int: ) written += 1 except Exception as e: - logger.debug("relationship_upsert_skipped", - from_key=rel["from_key"], to_key=rel["to_key"], error=str(e)) + logger.debug( + "relationship_upsert_skipped", + from_key=rel["from_key"], + to_key=rel["to_key"], + error=str(e), + ) raise except Exception as e: logger.warning("relationship_upsert_failed", error=str(e)) @@ -1158,7 +1664,8 @@ async def _write_coverage_snapshots(run_id: str) -> None: async with get_db_context(_PROJECT_ID) as db: # 一次性 INSERT: 取所有 active asset × 7 dimensions await db.execute( - _sql(""" + _sql( + """ INSERT INTO asset_coverage_snapshot ( run_id, asset_id, dimension, coverage_status, evidence, detected_by @@ -1178,6 +1685,7 @@ async def _write_coverage_snapshots(run_id: str) -> None: ) AS d(dimension) WHERE ai.lifecycle_state = 'active' ON CONFLICT (run_id, asset_id, dimension) DO NOTHING - """), + """ + ), {"rid": run_id}, ) diff --git a/apps/api/src/services/iwooos_security_asset_control_plane.py b/apps/api/src/services/iwooos_security_asset_control_plane.py index 004c69f7c..482981685 100644 --- a/apps/api/src/services/iwooos_security_asset_control_plane.py +++ b/apps/api/src/services/iwooos_security_asset_control_plane.py @@ -745,6 +745,9 @@ def build_iwooos_security_asset_control_plane( for row in automation_source if str(_value(row, "status", "")) == "failed" ) + # Aggregate stage receipts cannot prove that every stage belongs to one + # trace. Keep strict runtime completion closed until a same-run join exists. + same_run_closed_loop_proven = False relationship_count = int(_value(relationship_row, "relationship_count", 0) or 0) orphan_asset_count = int( @@ -972,6 +975,39 @@ def build_iwooos_security_asset_control_plane( sum(domain["controlled_percent"] for domain in security_program_domains) / len(security_program_domains) ) + strict_runtime_closure_percent = 100 if same_run_closed_loop_proven else 0 + completion_dimensions = [ + { + "dimension_id": "asset_scope", + "completion_percent": type_coverage_percent, + }, + { + "dimension_id": "nist_control", + "completion_percent": overall_control_percent, + }, + { + "dimension_id": "security_program", + "completion_percent": security_program_percent, + }, + { + "dimension_id": "siem_pipeline", + "completion_percent": _percent(siem_observed_stages, len(_SIEM_STAGES)), + }, + { + "dimension_id": "ai_loop_stages", + "completion_percent": _percent( + observed_automation_stages, len(_AI_LOOP_STAGES) + ), + }, + { + "dimension_id": "strict_same_run_runtime_closure", + "completion_percent": strict_runtime_closure_percent, + }, + ] + overall_security_completion_percent = round( + sum(row["completion_percent"] for row in completion_dimensions) + / len(completion_dimensions) + ) return { "schema_version": _SCHEMA_VERSION, @@ -1012,6 +1048,9 @@ def build_iwooos_security_asset_control_plane( if domain["status"] == "controlled" ), "security_program_controlled_percent": security_program_percent, + "overall_security_completion_percent": ( + overall_security_completion_percent + ), "siem_stage_count": len(_SIEM_STAGES), "siem_observed_stage_count": siem_observed_stages, "siem_stage_coverage_percent": _percent( @@ -1031,6 +1070,13 @@ def build_iwooos_security_asset_control_plane( "compliance_dimensions": compliance, "nist_csf_functions": control_functions, "security_program_domains": security_program_domains, + "completion": { + "method": "equal_weighted_runtime_conservative_v1", + "dimension_count": len(completion_dimensions), + "overall_percent": overall_security_completion_percent, + "strict_runtime_closure_percent": strict_runtime_closure_percent, + "dimensions": completion_dimensions, + }, "siem": {**siem, "pipeline": siem_pipeline}, "ai_automation": { "window_hours": _WINDOW_HOURS, @@ -1042,7 +1088,7 @@ def build_iwooos_security_asset_control_plane( "verified_remediation_receipt_count": verified_receipts, "rollback_receipt_count": rollback_receipts, "failed_operation_count": failed_operations, - "same_run_closed_loop_proven": False, + "same_run_closed_loop_proven": same_run_closed_loop_proven, "same_run_closed_loop_count": 0, }, "security_audit": audit, @@ -1093,6 +1139,7 @@ def build_unavailable_iwooos_security_asset_control_plane( "security_program_domain_count": 16, "security_program_controlled_domain_count": 0, "security_program_controlled_percent": 0, + "overall_security_completion_percent": 0, "siem_stage_count": len(_SIEM_STAGES), "siem_observed_stage_count": 0, "siem_stage_coverage_percent": 0, @@ -1175,6 +1222,23 @@ def build_unavailable_iwooos_security_asset_control_plane( ("compliance_audit", "合規、稽核與證據保存"), ) ], + "completion": { + "method": "equal_weighted_runtime_conservative_v1", + "dimension_count": 6, + "overall_percent": 0, + "strict_runtime_closure_percent": 0, + "dimensions": [ + {"dimension_id": dimension_id, "completion_percent": 0} + for dimension_id in ( + "asset_scope", + "nist_control", + "security_program", + "siem_pipeline", + "ai_loop_stages", + "strict_same_run_runtime_closure", + ) + ], + }, "siem": { "pipeline": [ { diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index c3ae72430..90bc09d02 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/apps/api/tests/test_asset_scanner_job.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import json from src.jobs import asset_scanner_job @@ -175,9 +176,7 @@ def test_asset_key_integrity_preflight_accepts_unique_inventory(monkeypatch) -> assert any("enable_indexonlyscan" in statement for statement in statements) assert any("enable_bitmapscan" in statement for statement in statements) assert any("HAVING COUNT(*) > 1" in statement for statement in statements) - assert any( - "asset_inventory_asset_key_key" in statement for statement in statements - ) + assert any("asset_inventory_asset_key_key" in statement for statement in statements) def test_asset_key_integrity_preflight_fails_closed_on_duplicate_key( @@ -217,7 +216,9 @@ def test_asset_key_integrity_preflight_fails_closed_on_duplicate_key( asyncio.run(exercise()) -def test_deprecate_missing_assets_is_bounded_to_successful_prefixes(monkeypatch) -> None: +def test_deprecate_missing_assets_is_bounded_to_successful_prefixes( + monkeypatch, +) -> None: captured: dict[str, object] = {} class CapturingDatabase: @@ -272,6 +273,9 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them( async def fake_database_catalog(): return [], [] + async def fake_ai_catalog(): + return [], [] + monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl) monkeypatch.setattr( asset_scanner_job, @@ -283,6 +287,11 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them( "_collect_database_catalog_assets", fake_database_catalog, ) + monkeypatch.setattr( + asset_scanner_job, + "_collect_ai_catalog_assets", + fake_ai_catalog, + ) assets, relationships, prefixes, errors = asyncio.run( asset_scanner_job._collect_runtime_assets() @@ -296,6 +305,372 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them( assert errors == ("k8s_nodes",) +def test_ingress_collector_builds_website_api_and_certificate_without_key_data() -> ( + None +): + assets, relationships = asset_scanner_job._collect_ingress_assets( + { + "items": [ + { + "metadata": {"namespace": "awoooi-prod", "name": "awoooi"}, + "spec": { + "rules": [ + { + "host": "awoooi.example.test", + "http": { + "paths": [ + { + "path": "/api", + "pathType": "Prefix", + "backend": { + "service": {"name": "awoooi-api"} + }, + } + ] + }, + } + ], + "tls": [ + { + "hosts": ["awoooi.example.test"], + "secretName": "awoooi-tls", + } + ], + }, + } + ] + } + ) + + by_type = {asset["asset_type"]: asset for asset in assets} + assert set(by_type) == {"website", "api_endpoint", "certificate"} + assert by_type["api_endpoint"]["metadata"]["request_or_response_body_read"] is False + assert by_type["certificate"]["metadata"]["certificate_material_read"] is False + assert by_type["certificate"]["metadata"]["secret_object_queried"] is False + route_key = by_type["api_endpoint"]["asset_key"] + assert relationships == [ + { + "from_key": "k8s/website/awoooi.example.test", + "to_key": route_key, + "relationship_type": "routes_to", + }, + { + "from_key": route_key, + "to_key": "k8s/service/awoooi-prod/awoooi-api", + "relationship_type": "routes_to", + }, + { + "from_key": "k8s/website/awoooi.example.test", + "to_key": "k8s/certificate-ref/awoooi-prod/awoooi-tls", + "relationship_type": "authenticates_via", + }, + ] + + +def test_ingress_route_identity_is_stable_when_manifest_order_changes() -> None: + def payload(paths: list[dict[str, object]]) -> dict[str, object]: + return { + "items": [ + { + "metadata": {"namespace": "prod", "name": "public"}, + "spec": { + "rules": [ + { + "host": "service.example.test", + "http": {"paths": paths}, + } + ] + }, + } + ] + } + + api_path = { + "path": "/api", + "backend": {"service": {"name": "api"}}, + } + health_path = { + "path": "/health", + "backend": {"service": {"name": "api"}}, + } + forward_assets, _ = asset_scanner_job._collect_ingress_assets( + payload([api_path, health_path]) + ) + reversed_assets, _ = asset_scanner_job._collect_ingress_assets( + payload([health_path, api_path]) + ) + + forward_keys = { + asset["name"]: asset["asset_key"] + for asset in forward_assets + if asset["asset_type"] == "api_endpoint" + } + reversed_keys = { + asset["name"]: asset["asset_key"] + for asset in reversed_assets + if asset["asset_type"] == "api_endpoint" + } + assert forward_keys == reversed_keys + assert len(set(forward_keys.values())) == 2 + + +def test_k8s_metadata_assets_never_read_volume_secret_or_command_data() -> None: + pvc = asset_scanner_job._build_pvc_asset( + { + "metadata": {"namespace": "prod", "name": "data"}, + "spec": { + "storageClassName": "local-path", + "accessModes": ["ReadWriteOnce"], + "resources": {"requests": {"storage": "10Gi"}}, + }, + "status": {"phase": "Bound"}, + } + ) + cronjob = asset_scanner_job._build_cronjob_asset( + { + "metadata": {"namespace": "prod", "name": "scan"}, + "spec": {"schedule": "0 * * * *", "suspend": False}, + "status": {"active": []}, + } + ) + secret = asset_scanner_job._build_secret_reference_asset( + "prod", + "database-ref", + reference_kind="PodVolume", + ) + + assert pvc["asset_type"] == "volume" + assert pvc["metadata"]["data_read"] is False + assert cronjob["asset_type"] == "scheduled_job" + assert cronjob["metadata"]["command_or_env_read"] is False + assert secret["asset_type"] == "secret" + assert secret["metadata"]["secret_object_queried"] is False + assert secret["metadata"]["secret_value_read"] is False + + +def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefixes( + monkeypatch, +) -> None: + async def fake_kubectl(resource: str, all_namespaces: bool = True): + if resource == "pods": + return { + "items": [ + { + "metadata": {"namespace": "prod", "name": "api"}, + "spec": { + "volumes": [ + { + "name": "credential", + "secret": {"secretName": "database-ref"}, + } + ] + }, + } + ] + } + if resource == "persistentvolumeclaims": + return { + "items": [ + { + "metadata": {"namespace": "prod", "name": "data"}, + "spec": {}, + "status": {}, + } + ] + } + if resource == "ingresses": + return { + "items": [ + { + "metadata": {"namespace": "prod", "name": "public"}, + "spec": { + "rules": [ + { + "host": "service.example.test", + "http": { + "paths": [ + { + "path": "/api", + "backend": {"service": {"name": "api"}}, + } + ] + }, + } + ], + "tls": [ + { + "hosts": ["service.example.test"], + "secretName": "public-tls", + } + ], + }, + } + ] + } + if resource == "cronjobs": + return { + "items": [ + { + "metadata": {"namespace": "prod", "name": "scan"}, + "spec": {"schedule": "0 * * * *"}, + "status": {}, + } + ] + } + return {"items": []} + + async def empty_collector(): + return [], [] + + monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl) + monkeypatch.setattr( + asset_scanner_job, + "_collect_prometheus_targets", + empty_collector, + ) + monkeypatch.setattr( + asset_scanner_job, + "_collect_database_catalog_assets", + empty_collector, + ) + monkeypatch.setattr( + asset_scanner_job, + "_collect_ai_catalog_assets", + empty_collector, + ) + + assets, relationships, prefixes, errors = asyncio.run( + asset_scanner_job._collect_runtime_assets() + ) + + assert errors == () + assert { + "container", + "secret", + "volume", + "website", + "api_endpoint", + "certificate", + "scheduled_job", + } <= {asset["asset_type"] for asset in assets} + assert { + "k8s/secret-ref/", + "k8s/pvc/", + "k8s/website/", + "k8s/ingress-route/", + "k8s/certificate-ref/", + "k8s/cronjob/", + } <= set(prefixes) + assert any( + relationship["to_key"] == "k8s/secret-ref/prod/database-ref" + for relationship in relationships + ) + + +def test_ai_catalog_collector_reads_aggregates_and_model_names_only( + monkeypatch, + tmp_path, +) -> None: + class CatalogResult: + def __init__(self, rows): + self._rows = rows + + def fetchall(self): + return self._rows + + class CatalogDatabase: + async def execute(self, statement, parameters=None): + sql = str(statement) + if "FROM agent_sessions" in sql: + return CatalogResult( + [ + type( + "AgentRow", + (), + { + "_mapping": { + "agent_role": "security_reviewer", + "session_count": 4, + "latest_seen_at": "2026-07-14T00:00:00+00:00", + } + }, + )() + ] + ) + assert parameters == {"project_id": "awoooi"} + return CatalogResult( + [ + type( + "KnowledgeRow", + (), + { + "_mapping": { + "entry_type": "runbook", + "entry_count": 9, + "latest_updated_at": "2026-07-14T00:00:00+00:00", + } + }, + )() + ] + ) + + class CatalogContext: + async def __aenter__(self): + return CatalogDatabase() + + async def __aexit__(self, exc_type, exc, traceback): + return False + + monkeypatch.setattr( + "src.db.base.get_db_context", + lambda project_id=None: CatalogContext(), + ) + registry = tmp_path / "models.json" + registry.write_text( + json.dumps( + { + "providers": { + "ollama": { + "models": { + "default": "model-a", + "summary": "model-a", + } + } + } + } + ), + encoding="utf-8", + ) + + assets, relationships = asyncio.run( + asset_scanner_job._collect_ai_catalog_assets(registry) + ) + by_type: dict[str, list[dict]] = {} + for asset in assets: + by_type.setdefault(asset["asset_type"], []).append(asset) + + assert set(by_type) == { + "third_party_service", + "llm_model", + "ai_agent", + "km_entry", + } + assert len(by_type["llm_model"]) == 1 + assert by_type["llm_model"][0]["metadata"]["purposes"] == [ + "default", + "summary", + ] + assert by_type["ai_agent"][0]["metadata"]["session_input_or_output_read"] is False + assert by_type["km_entry"][0]["metadata"]["title_or_content_read"] is False + assert relationships == [ + { + "from_key": "model_registry/model/ollama/model-a", + "to_key": "model_registry/provider/ollama", + "relationship_type": "depends_on", + } + ] + + def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -> None: terminal: dict[str, object] = {} diff --git a/apps/api/tests/test_iwooos_security_asset_control_plane.py b/apps/api/tests/test_iwooos_security_asset_control_plane.py index 8f5f4fb5b..f5a3ebf76 100644 --- a/apps/api/tests/test_iwooos_security_asset_control_plane.py +++ b/apps/api/tests/test_iwooos_security_asset_control_plane.py @@ -130,6 +130,25 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure assert payload["summary"]["siem_stage_coverage_percent"] == 100 assert payload["summary"]["ai_loop_stage_coverage_percent"] == 100 assert payload["summary"]["verified_remediation_receipt_count_24h"] == 1 + expected_completion = round( + sum( + payload["summary"][field] + for field in ( + "asset_type_coverage_percent", + "nist_controlled_percent", + "security_program_controlled_percent", + "siem_stage_coverage_percent", + "ai_loop_stage_coverage_percent", + ) + ) + / 6 + ) + assert payload["summary"]["overall_security_completion_percent"] == ( + expected_completion + ) + assert payload["completion"]["overall_percent"] == expected_completion + assert payload["completion"]["strict_runtime_closure_percent"] == 0 + assert payload["completion"]["dimension_count"] == 6 assert payload["discovery"]["fresh"] is True assert payload["ai_automation"]["same_run_closed_loop_proven"] is False assert payload["ai_automation"]["same_run_closed_loop_count"] == 0 @@ -191,6 +210,7 @@ def test_unknown_coverage_and_compliance_create_p0_work_items() -> None: assert payload["summary"]["compliance_unknown_count"] == 1 assert payload["summary"]["siem_observed_stage_count"] == 0 assert payload["summary"]["ai_loop_observed_stage_count"] == 0 + assert payload["completion"]["strict_runtime_closure_percent"] == 0 work_item_ids = {item["work_item_id"] for item in payload["work_items"]} assert { "AIA-P0-006-02", @@ -271,6 +291,9 @@ def test_unavailable_payload_is_fixed_degraded_contract_without_raw_error() -> N assert payload["source_status"] == "live_database_unavailable" assert payload["reason_code"] == "database_query_failed" assert payload["summary"]["managed_asset_count"] == 0 + assert payload["summary"]["overall_security_completion_percent"] == 0 + assert payload["completion"]["overall_percent"] == 0 + assert payload["completion"]["strict_runtime_closure_percent"] == 0 assert payload["discovery"]["fresh"] is False assert payload["boundaries"]["raw_asset_identity_returned"] is False assert payload["boundaries"]["raw_event_payload_returned"] is False diff --git a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx index f40cf35aa..8d7e3a011 100644 --- a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx +++ b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx @@ -28,7 +28,7 @@ type Copy = { title: string; live: string; degraded: string; - assets: string; + overall: string; assetTypes: string; nist: string; siem: string; @@ -60,7 +60,7 @@ const COPY: Record<"zh" | "en", Copy> = { title: "AI 資安控制平面", live: "LIVE", degraded: "DEGRADED", - assets: "納管資產", + overall: "整體完成度", assetTypes: "資產類型", nist: "NIST 控制", siem: "SIEM 閉環", @@ -90,7 +90,7 @@ const COPY: Record<"zh" | "en", Copy> = { title: "AI Security Control Plane", live: "LIVE", degraded: "DEGRADED", - assets: "Managed assets", + overall: "Overall completion", assetTypes: "Asset types", nist: "NIST control", siem: "SIEM closure", @@ -327,8 +327,12 @@ export function SecurityAssetControlPlaneCockpit({ -
- +
+ ; + completion: { + method: "equal_weighted_runtime_conservative_v1"; + dimension_count: number; + overall_percent: number; + strict_runtime_closure_percent: number; + dimensions: Array<{ + dimension_id: string; + completion_percent: number; + }>; + }; siem: { pipeline: SecurityControlPlaneStage[]; incident_total_24h?: number; @@ -172,6 +183,10 @@ export function isSecurityAssetControlPlanePayload( typeof value.summary.source_count === "number" && Array.isArray(value.source_health) && typeof value.summary.nist_controlled_percent === "number" && + typeof value.summary.overall_security_completion_percent === "number" && + isRecord(value.completion) && + value.completion.method === "equal_weighted_runtime_conservative_v1" && + Array.isArray(value.completion.dimensions) && value.boundaries.aggregate_only === true && value.boundaries.raw_asset_identity_returned === false && value.boundaries.raw_event_payload_returned === false &&