From ca8771a210d8ffc4f4aeab24480f4d0b96dd87c4 Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 14:34:38 +0800 Subject: [PATCH] feat(security): inventory runtime supply chain --- apps/api/src/jobs/asset_scanner_job.py | 219 ++++++++++++++++++ .../iwooos_security_asset_control_plane.py | 116 +++++++++- apps/api/tests/test_asset_scanner_job.py | 152 +++++++++++- ...est_iwooos_security_asset_control_plane.py | 78 +++++++ 4 files changed, 560 insertions(+), 5 deletions(-) diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 3441b04a4..ca3f3f452 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -211,6 +211,7 @@ async def scan_once( "ai_catalog", "model_registry", "gitea_bundle_backup", + "gitea_ci", ), scan_depth: str = "shallow", ) -> str: @@ -516,6 +517,108 @@ def _build_workload_component_assets( return assets, relationships +def _build_workload_package_assets( + item: dict[str, Any], + kind: str, +) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + """Inventory deployed OCI image references without pulling image content.""" + meta = item.get("metadata", {}) or {} + labels = meta.get("labels", {}) or {} + namespace = meta.get("namespace") or "default" + workload_name = meta.get("name") or "unknown" + pod_spec = ( + ((item.get("spec", {}) or {}).get("template", {}) or {}).get("spec", {}) + or {} + ) + owner = ( + _normalize_owner_team(labels.get("system")) + or _owner_from_namespace(namespace) + or "platform_security" + ) + workload_key = f"k8s/{kind.lower()}/{namespace}/{workload_name}" + assets: list[dict[str, Any]] = [] + relationships: list[dict[str, str]] = [] + + for container_kind, containers in ( + ("container", pod_spec.get("containers", []) or []), + ("init_container", pod_spec.get("initContainers", []) or []), + ): + for container in containers: + if not isinstance(container, dict): + continue + container_name = _normalize_owner_team(container.get("name")) + image_ref = str(container.get("image") or "").strip() + if ( + not container_name + or not image_ref + or len(image_ref) > 2048 + or any(ord(character) < 32 for character in image_ref) + ): + continue + + image_lower = image_ref.lower() + forbidden_domain = next( + ( + domain + for domain in ( + "github.com", + "api.github.com", + "raw.githubusercontent.com", + "codeload.github.com", + "ghcr.io", + ) + if image_lower == domain + or image_lower.startswith(f"{domain}/") + ), + None, + ) + digest = image_ref.rsplit("@", 1)[1] if "@" in image_ref else None + last_segment = image_ref.rsplit("/", 1)[-1].split("@", 1)[0] + tag = last_segment.rsplit(":", 1)[1] if ":" in last_segment else None + package_key = ( + f"k8s/package/{kind.lower()}/{namespace}/" + f"{workload_name}/{container_kind}/{container_name}" + ) + assets.append( + { + "asset_key": package_key, + "asset_type": "package", + "host": None, + "namespace": namespace, + "name": f"{workload_name}/{container_name}", + "metadata": { + "kind": "DeployedOCIImage", + "workload_asset_key": workload_key, + "container_kind": container_kind, + "container_name": container_name, + "image_ref": image_ref, + "image_tag": tag, + "image_digest": digest, + "digest_pinned": bool( + digest and digest.lower().startswith("sha256:") + ), + "forbidden_source_domain_detected": forbidden_domain, + "image_content_pulled_by_scanner": False, + "registry_api_called_by_scanner": False, + "secret_value_read": False, + }, + "tags": [ + "source:k8s_workload_spec", + "ecosystem:oci", + f"owner:{owner}", + ], + } + ) + relationships.append( + { + "from_key": workload_key, + "to_key": package_key, + "relationship_type": "depends_on", + } + ) + return assets, relationships + + def _build_service_asset(item: dict[str, Any]) -> dict[str, Any]: """Service → asset_inventory row (asset_type='k8s_resource').""" meta = item.get("metadata", {}) or {} @@ -1080,9 +1183,15 @@ async def _collect_runtime_assets() -> ( component_assets, component_relationships = ( _build_workload_component_assets(item, kind) ) + package_assets, package_relationships = ( + _build_workload_package_assets(item, kind) + ) assets.extend(component_assets) + assets.extend(package_assets) relationships.extend(component_relationships) + relationships.extend(package_relationships) reconciled_prefixes.add(f"k8s/{kind.lower()}/") + reconciled_prefixes.add(f"k8s/package/{kind.lower()}/") for component_type in ( "frontend", "backend", @@ -1254,6 +1363,17 @@ async def _collect_runtime_assets() -> ( logger.warning("collect_gitea_bundle_failed", error=str(e)) collector_errors.append("gitea_bundle_backup") + # 13. Gitea workflow identities joined with live CI receipt-chain status. + try: + ci_assets, ci_relationships = await _collect_gitea_ci_assets() + assets.extend(ci_assets) + relationships.extend(ci_relationships) + if ci_assets: + reconciled_prefixes.add("gitea/workflow/") + except Exception as e: + logger.warning("collect_gitea_ci_failed", error=str(e)) + collector_errors.append("gitea_ci") + return ( list({asset["asset_key"]: asset for asset in assets}.values()), relationships, @@ -1383,6 +1503,105 @@ async def _collect_gitea_bundle_assets() -> ( return _build_gitea_bundle_assets(payload) +def _build_gitea_ci_assets( + workflow_health: dict[str, Any], + receipt_chain: dict[str, Any], +) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + """Project Gitea workflow identities with live receipt-chain status.""" + if receipt_chain.get("source_control_authority") != "gitea": + raise RuntimeError("ci_source_control_authority_not_gitea") + boundaries = receipt_chain.get("operation_boundaries") or {} + if boundaries.get("github_api_allowed") is not False: + raise RuntimeError("ci_github_freeze_not_evidenced") + if workflow_health.get("schema_version") != "gitea_workflow_runner_health_v1": + raise RuntimeError("ci_workflow_health_schema_invalid") + + workflow_rows = workflow_health.get("workflow_records") or [] + if not isinstance(workflow_rows, list) or not workflow_rows: + raise RuntimeError("ci_workflow_records_missing") + summary = receipt_chain.get("summary") or {} + chain_status = str(receipt_chain.get("status") or "unknown") + chain_ready = receipt_chain.get("ready") is True + assets: list[dict[str, Any]] = [] + relationships: list[dict[str, str]] = [] + seen: set[str] = set() + for row in workflow_rows: + if not isinstance(row, dict): + continue + workflow_id = _normalize_owner_team(row.get("workflow_id")) + if not workflow_id or workflow_id in seen: + continue + seen.add(workflow_id) + pipeline_key = f"gitea/workflow/{workflow_id}" + assets.append( + { + "asset_key": pipeline_key, + "asset_type": "ci_pipeline", + "host": None, + "namespace": "wooo/awoooi", + "name": workflow_id, + "metadata": { + "kind": "GiteaWorkflow", + "display_name": str(row.get("display_name") or workflow_id), + "file_ref": str(row.get("file_ref") or ""), + "status": str(row.get("status") or "unknown"), + "risk_level": str(row.get("risk_level") or "unknown"), + "triggers": sorted(str(value) for value in row.get("triggers") or []), + "runner_labels": sorted( + str(value) for value in row.get("runner_labels") or [] + ), + "runner_evidence_status": str( + row.get("runner_evidence_status") or "unknown" + ), + "job_count": int(row.get("job_count") or 0), + "notify_bridge_calls": int(row.get("notify_bridge_calls") or 0), + "receipt_chain_status": chain_status, + "receipt_chain_ready": chain_ready, + "workflow_notify_bridge_count": int( + summary.get("workflow_notify_bridge_count") or 0 + ), + "workflow_source_content_read": False, + "workflow_triggered_by_scanner": False, + "gitea_api_write_performed": False, + "github_api_used": False, + "secret_value_read": False, + }, + "tags": [ + "source:gitea_workflow_health", + "source:gitea_cicd_receipt_chain", + "owner:awoooi", + ], + } + ) + relationships.append( + { + "from_key": pipeline_key, + "to_key": "gitea/repo/wooo/awoooi", + "relationship_type": "depends_on", + } + ) + if not assets: + raise RuntimeError("ci_workflow_identities_invalid") + return assets, relationships + + +async def _collect_gitea_ci_assets() -> ( + tuple[list[dict[str, Any]], list[dict[str, str]]] +): + from src.services.gitea_cicd_alert_receipt_chain_readback import ( + load_latest_gitea_cicd_alert_receipt_chain_readback, + ) + from src.services.gitea_workflow_runner_health import ( + load_latest_gitea_workflow_runner_health, + ) + + workflow_health, receipt_chain = await asyncio.gather( + asyncio.to_thread(load_latest_gitea_workflow_runner_health), + load_latest_gitea_cicd_alert_receipt_chain_readback(project_id=_PROJECT_ID), + ) + return _build_gitea_ci_assets(workflow_health, receipt_chain) + + def _is_valid_ipv4(s: str) -> bool: """嚴格 IPv4 判斷: 4 段 + 每段 0-255 整數. 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 482981685..7174f9a84 100644 --- a/apps/api/src/services/iwooos_security_asset_control_plane.py +++ b/apps/api/src/services/iwooos_security_asset_control_plane.py @@ -364,6 +364,10 @@ def _build_security_program_domains( ai_stage_percent: int, ai_evidence_count: int, audit_evidence_count: int, + package_digest_pinned_percent: int, + github_supply_chain_compliant_percent: int, + forbidden_github_supply_chain_asset_count: int, + package_digest_unpinned_count: int, ) -> list[dict[str, Any]]: scope_by_id = {row["scope_id"]: row for row in asset_scopes} function_by_id = {row["function_id"]: row for row in control_functions} @@ -387,6 +391,13 @@ def _build_security_program_domains( def average(*values: int) -> int: return round(sum(values) / len(values)) if values else 0 + if forbidden_github_supply_chain_asset_count > 0: + supply_chain_gap_code = "forbidden_github_supply_chain_source_detected" + elif package_digest_unpinned_count > 0: + supply_chain_gap_code = "runtime_image_digest_pin_missing" + else: + supply_chain_gap_code = "sbom_signature_dependency_evidence_missing" + domain_values = ( ( "governance_risk", @@ -473,9 +484,11 @@ def _build_security_program_domains( average( scope_percent("supply_chain"), compliance_percent("cve_scan"), + package_digest_pinned_percent, + github_supply_chain_compliant_percent, ), scope_assets("supply_chain") + compliance_total("cve_scan"), - "sbom_signature_dependency_evidence_missing", + supply_chain_gap_code, ), ( "telemetry_detection", @@ -654,6 +667,14 @@ def build_iwooos_security_asset_control_plane( stale_total = sum( int(_value(row, "stale_count", 0) or 0) for row in inventory_source ) + forbidden_github_supply_chain_asset_count = sum( + int(_value(row, "forbidden_github_source_count", 0) or 0) + for row in inventory_source + ) + package_digest_unpinned_count = sum( + int(_value(row, "unpinned_package_count", 0) or 0) + for row in inventory_source + ) present_types = { asset_type for asset_type in _ASSET_TYPES if by_type.get(asset_type, 0) > 0 } @@ -849,6 +870,22 @@ def build_iwooos_security_asset_control_plane( + audit["mcp_tool_audit_count_24h"] + audit["automation_operation_count_24h"] ), + package_digest_pinned_percent=_percent( + max(0, by_type.get("package", 0) - package_digest_unpinned_count), + by_type.get("package", 0), + ), + github_supply_chain_compliant_percent=_percent( + max( + 0, + by_type.get("package", 0) + - forbidden_github_supply_chain_asset_count, + ), + by_type.get("package", 0), + ), + forbidden_github_supply_chain_asset_count=( + forbidden_github_supply_chain_asset_count + ), + package_digest_unpinned_count=package_digest_unpinned_count, ) work_items: list[dict[str, Any]] = [] @@ -886,6 +923,24 @@ def build_iwooos_security_asset_control_plane( "依 scope connector 接入網站、資料、供應鏈、安全工具與 AI 資產。", ) ) + supply_chain_policy_gap_count = ( + forbidden_github_supply_chain_asset_count + + package_digest_unpinned_count + ) + if supply_chain_policy_gap_count > 0: + work_items.append( + _work_item( + "AIA-P0-006-02A", + "P0", + "supply_chain_policy", + "封閉 runtime image 供應鏈", + supply_chain_policy_gap_count, + ( + "以內部 registry 或 repo-owned immutable artifact 替換禁用來源，" + "補 digest pin、canary rollout 與獨立 verifier；不得連線 GitHub。" + ), + ) + ) if coverage_total == 0 or coverage_unknown > 0: work_items.append( _work_item( @@ -975,6 +1030,7 @@ def build_iwooos_security_asset_control_plane( sum(domain["controlled_percent"] for domain in security_program_domains) / len(security_program_domains) ) + package_inventory_evidenced = by_type.get("package", 0) > 0 strict_runtime_closure_percent = 100 if same_run_closed_loop_proven else 0 completion_dimensions = [ { @@ -1034,6 +1090,10 @@ def build_iwooos_security_asset_control_plane( "stale_asset_count": stale_total, "orphan_asset_count": orphan_asset_count, "relationship_count": relationship_count, + "package_digest_unpinned_count": package_digest_unpinned_count, + "forbidden_github_supply_chain_asset_count": ( + forbidden_github_supply_chain_asset_count + ), "automation_coverage_green_percent": _percent( coverage_green, coverage_total ), @@ -1092,6 +1152,30 @@ def build_iwooos_security_asset_control_plane( "same_run_closed_loop_count": 0, }, "security_audit": audit, + "supply_chain": { + "package_asset_count": by_type.get("package", 0), + "package_inventory_evidenced": package_inventory_evidenced, + "package_digest_pinned_count": max( + 0, + by_type.get("package", 0) - package_digest_unpinned_count, + ), + "package_digest_unpinned_count": package_digest_unpinned_count, + "forbidden_github_supply_chain_asset_count": ( + forbidden_github_supply_chain_asset_count + ), + "github_freeze_compliant": ( + package_inventory_evidenced + and forbidden_github_supply_chain_asset_count == 0 + ), + "runtime_image_policy_compliant": ( + package_inventory_evidenced + and forbidden_github_supply_chain_asset_count == 0 + and package_digest_unpinned_count == 0 + ), + "raw_image_identity_returned": False, + "registry_api_called": False, + "github_api_used": False, + }, "source_health": source_health_rows, "work_items": work_items, "boundaries": { @@ -1131,6 +1215,8 @@ def build_unavailable_iwooos_security_asset_control_plane( "stale_asset_count": 0, "orphan_asset_count": 0, "relationship_count": 0, + "package_digest_unpinned_count": 0, + "forbidden_github_supply_chain_asset_count": 0, "automation_coverage_green_percent": 0, "automation_coverage_unknown_count": 0, "compliance_violation_count": 0, @@ -1282,6 +1368,18 @@ def build_unavailable_iwooos_security_asset_control_plane( "accepted_ai_trace_count_24h": 0, "immutable_external_audit_store_evidenced": False, }, + "supply_chain": { + "package_asset_count": 0, + "package_inventory_evidenced": False, + "package_digest_pinned_count": 0, + "package_digest_unpinned_count": 0, + "forbidden_github_supply_chain_asset_count": 0, + "github_freeze_compliant": False, + "runtime_image_policy_compliant": False, + "raw_image_identity_returned": False, + "registry_api_called": False, + "github_api_used": False, + }, "source_health": [ { "source_id": source_id, @@ -1489,7 +1587,21 @@ class IwoooSSecurityAssetControlPlaneService: ) AS unowned_count, count(*) FILTER ( WHERE last_seen_at < NOW() - INTERVAL '2 hours' - ) AS stale_count + ) AS stale_count, + count(*) FILTER ( + WHERE asset_type = 'package' + AND COALESCE( + metadata->>'forbidden_source_domain_detected', + '' + ) <> '' + ) AS forbidden_github_source_count, + count(*) FILTER ( + WHERE asset_type = 'package' + AND COALESCE( + metadata->>'digest_pinned', + 'false' + ) <> 'true' + ) AS unpinned_package_count FROM asset_inventory WHERE lifecycle_state = 'active' GROUP BY asset_type diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index f1b5683e9..a7026209f 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/apps/api/tests/test_asset_scanner_job.py @@ -377,6 +377,9 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them( async def fake_gitea_bundle(): return [], [] + async def fake_gitea_ci(): + return [], [] + monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl) monkeypatch.setattr( asset_scanner_job, @@ -398,6 +401,11 @@ def test_runtime_collection_reports_failed_sources_without_reconciling_them( "_collect_gitea_bundle_assets", fake_gitea_bundle, ) + monkeypatch.setattr( + asset_scanner_job, + "_collect_gitea_ci_assets", + fake_gitea_ci, + ) assets, relationships, prefixes, errors = asyncio.run( asset_scanner_job._collect_runtime_assets() @@ -654,6 +662,103 @@ services: assert len(registry_relationships) == len(registry_assets) +def test_workload_package_projection_never_pulls_images_or_uses_github() -> None: + assets, relationships = asset_scanner_job._build_workload_package_assets( + { + "metadata": { + "namespace": "awoooi-prod", + "name": "awoooi-api", + "labels": {"system": "awoooi"}, + }, + "spec": { + "template": { + "spec": { + "containers": [ + { + "name": "api", + "image": "192.168.0.110:5000/library/api:abc123", + } + ], + "initContainers": [ + { + "name": "blocked-source", + "image": "ghcr.io/example/tool@sha256:abcd", + } + ], + } + } + }, + }, + "Deployment", + ) + + assert len(assets) == 2 + assert all(asset["asset_type"] == "package" for asset in assets) + assert assets[0]["metadata"]["image_content_pulled_by_scanner"] is False + assert assets[0]["metadata"]["registry_api_called_by_scanner"] is False + assert assets[0]["metadata"]["forbidden_source_domain_detected"] is None + assert assets[1]["metadata"]["digest_pinned"] is True + assert assets[1]["metadata"]["forbidden_source_domain_detected"] == "ghcr.io" + assert len(relationships) == 2 + + +def test_gitea_ci_projection_requires_gitea_authority_and_github_freeze() -> None: + assets, relationships = asset_scanner_job._build_gitea_ci_assets( + { + "schema_version": "gitea_workflow_runner_health_v1", + "workflow_records": [ + { + "workflow_id": "cd", + "display_name": "CD", + "file_ref": ".gitea/workflows/cd.yaml", + "status": "manifest_mapped", + "risk_level": "high", + "triggers": ["push", "workflow_dispatch"], + "runner_labels": ["awoooi-non110-ubuntu"], + "runner_evidence_status": "non110_runner_mapped", + "job_count": 2, + "notify_bridge_calls": 1, + } + ], + }, + { + "source_control_authority": "gitea", + "status": "gitea_cicd_alert_receipt_chain_ready", + "ready": True, + "operation_boundaries": {"github_api_allowed": False}, + "summary": {"workflow_notify_bridge_count": 6}, + }, + ) + + assert [asset["asset_type"] for asset in assets] == ["ci_pipeline"] + assert assets[0]["metadata"]["receipt_chain_ready"] is True + assert assets[0]["metadata"]["workflow_triggered_by_scanner"] is False + assert assets[0]["metadata"]["github_api_used"] is False + assert relationships == [ + { + "from_key": "gitea/workflow/cd", + "to_key": "gitea/repo/wooo/awoooi", + "relationship_type": "depends_on", + } + ] + + try: + asset_scanner_job._build_gitea_ci_assets( + { + "schema_version": "gitea_workflow_runner_health_v1", + "workflow_records": [{"workflow_id": "cd"}], + }, + { + "source_control_authority": "gitea", + "operation_boundaries": {"github_api_allowed": True}, + }, + ) + except RuntimeError as exc: + assert str(exc) == "ci_github_freeze_not_evidenced" + else: + raise AssertionError("CI projection must fail closed without GitHub freeze") + + def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefixes( monkeypatch, ) -> None: @@ -696,7 +801,18 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix "system": "awoooi", }, }, - "spec": {}, + "spec": { + "template": { + "spec": { + "containers": [ + { + "name": "web", + "image": "192.168.0.110:5000/library/web:test", + } + ] + } + } + }, "status": {}, }, { @@ -708,7 +824,18 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix "system": "awoooi", }, }, - "spec": {}, + "spec": { + "template": { + "spec": { + "containers": [ + { + "name": "api", + "image": "192.168.0.110:5000/library/api:test", + } + ] + } + } + }, "status": {}, }, ] @@ -724,7 +851,18 @@ def test_runtime_collection_integrates_new_asset_types_and_reconciliation_prefix "app.kubernetes.io/component": "log-collector" }, }, - "spec": {}, + "spec": { + "template": { + "spec": { + "containers": [ + { + "name": "otel-collector", + "image": "otel/collector:0.96.0", + } + ] + } + } + }, "status": {}, } ] @@ -837,6 +975,11 @@ services: "_collect_gitea_bundle_assets", empty_collector, ) + monkeypatch.setattr( + asset_scanner_job, + "_collect_gitea_ci_assets", + empty_collector, + ) assets, relationships, prefixes, errors = asyncio.run( asset_scanner_job._collect_runtime_assets() @@ -858,6 +1001,7 @@ services: "dashboard", "cache", "message_queue", + "package", } <= {asset["asset_type"] for asset in assets} assert { "k8s/secret-ref/", @@ -873,6 +1017,8 @@ services: "service-registry/dashboard/", "service-registry/cache/", "service-registry/message_queue/", + "k8s/package/deployment/", + "k8s/package/daemonset/", } <= set(prefixes) assert any( relationship["to_key"] == "k8s/secret-ref/prod/database-ref" 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 f5a3ebf76..428405ec4 100644 --- a/apps/api/tests/test_iwooos_security_asset_control_plane.py +++ b/apps/api/tests/test_iwooos_security_asset_control_plane.py @@ -162,6 +162,14 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure assert domain_by_id["siem_soc_threat_intel"]["controlled_percent"] == 50 assert len(payload["siem"]["pipeline"]) == 8 assert payload["security_audit"]["mcp_tool_audit_count_24h"] == 11 + assert payload["summary"]["package_digest_unpinned_count"] == 0 + assert payload["summary"]["forbidden_github_supply_chain_asset_count"] == 0 + assert payload["supply_chain"]["package_inventory_evidenced"] is False + assert payload["supply_chain"]["github_freeze_compliant"] is False + assert payload["supply_chain"]["runtime_image_policy_compliant"] is False + assert payload["supply_chain"]["raw_image_identity_returned"] is False + assert payload["supply_chain"]["registry_api_called"] is False + assert payload["supply_chain"]["github_api_used"] is False assert any( item["work_item_id"] == "AIA-P0-006-02" for item in payload["work_items"] ) @@ -177,6 +185,71 @@ def test_builds_live_asset_siem_audit_and_ai_control_plane_without_false_closure assert 'secret_value_collection_allowed": true' not in public_text.lower() +def test_runtime_image_policy_gaps_are_visible_and_never_false_green() -> None: + now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC) + payload = build_iwooos_security_asset_control_plane( + discovery_row=_row( + latest_status="success", + latest_scan_depth="full", + latest_started_at=now - timedelta(minutes=10), + latest_ended_at=now - timedelta(minutes=9), + latest_success_ended_at=now - timedelta(minutes=9), + latest_total_assets=31, + ), + inventory_rows=[ + _row( + asset_type="package", + cnt=31, + unowned_count=0, + stale_count=0, + forbidden_github_source_count=3, + unpinned_package_count=31, + ), + ], + coverage_rows=[], + relationship_row=_row(relationship_count=31, orphan_asset_count=0), + compliance_rows=[], + automation_rows=[], + ai_trace_row=_row(trace_count=0, accepted_trace_count=0), + siem_row=_row(), + audit_row=_row(), + generated_at=now, + ) + + assert payload["summary"]["package_digest_unpinned_count"] == 31 + assert payload["summary"]["forbidden_github_supply_chain_asset_count"] == 3 + assert payload["supply_chain"] == { + "package_asset_count": 31, + "package_inventory_evidenced": True, + "package_digest_pinned_count": 0, + "package_digest_unpinned_count": 31, + "forbidden_github_supply_chain_asset_count": 3, + "github_freeze_compliant": False, + "runtime_image_policy_compliant": False, + "raw_image_identity_returned": False, + "registry_api_called": False, + "github_api_used": False, + } + work_item = next( + item + for item in payload["work_items"] + if item["work_item_id"] == "AIA-P0-006-02A" + ) + assert work_item["gap_count"] == 34 + assert work_item["status"] == "open" + domains = { + domain["domain_id"]: domain for domain in payload["security_program_domains"] + } + assert domains["software_supply_chain"]["gap_code"] == ( + "forbidden_github_supply_chain_source_detected" + ) + assert domains["software_supply_chain"]["status"] != "controlled" + + public_text = json.dumps(payload, ensure_ascii=False).lower() + assert "ghcr.io/" not in public_text + assert "raw.githubusercontent.com/" not in public_text + + def test_unknown_coverage_and_compliance_create_p0_work_items() -> None: now = datetime(2026, 7, 11, 4, 0, tzinfo=UTC) payload = build_iwooos_security_asset_control_plane( @@ -413,6 +486,11 @@ def test_service_bounds_source_concurrency_for_database_budget(monkeypatch) -> N ) assert "ORDER BY snapshot_id DESC" in compliance_query assert "LIMIT 100000" in compliance_query + inventory_query = next( + statement for statement in statements if "FROM asset_inventory" in statement + ) + assert "forbidden_source_domain_detected" in inventory_query + assert "digest_pinned" in inventory_query def test_public_api_returns_live_aggregate(monkeypatch) -> None: