feat(security): expand canonical asset coverage

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] = {}
 

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