fix(security): reconcile asset coverage and ownership

apps/api/tests/test_asset_scanner_job.py

@@ -143,6 +143,104 @@ def test_asset_upsert_propagates_database_failure(monkeypatch) -> None:
     asyncio.run(exercise())
 
 
+def test_owner_resolution_prefers_explicit_labels_and_safe_scoped_fallbacks() -> None:
+    assert asset_scanner_job._resolve_owner_team(
+        {
+            "asset_key": "k8s/deployment/awoooi-prod/api",
+            "namespace": "awoooi-prod",
+            "metadata": {
+                "labels": {"wooo.work/owner-team": "Platform Security"}
+            },
+            "tags": [],
+        }
+    ) == ("platform_security", "metadata.labels.wooo.work/owner-team")
+    assert asset_scanner_job._resolve_owner_team(
+        {
+            "asset_key": "k8s/service/awoooi-prod/api",
+            "namespace": "awoooi-prod",
+            "metadata": {},
+            "tags": [],
+        }
+    ) == ("awoooi", "k8s.namespace")
+    assert asset_scanner_job._resolve_owner_team(
+        {
+            "asset_key": "postgres/relation/awoooi/public/incidents",
+            "namespace": "public",
+            "metadata": {},
+            "tags": ["source:pg_catalog"],
+        }
+    ) == ("awoooi", "project_scoped_collector")
+    assert asset_scanner_job._resolve_owner_team(
+        {
+            "asset_key": "model_registry/model/ollama/model-a",
+            "namespace": None,
+            "metadata": {},
+            "tags": ["source:model_registry"],
+        }
+    ) == ("awoooi", "project_scoped_collector")
+    assert asset_scanner_job._resolve_owner_team(
+        {
+            "asset_key": "k8s/node/worker-1",
+            "namespace": None,
+            "metadata": {"labels": {}},
+            "tags": [],
+        }
+    ) == (None, "unresolved")
+
+
+def test_asset_upsert_persists_owner_candidate_without_overwriting_human_owner(
+    monkeypatch,
+) -> None:
+    captured: dict[str, object] = {}
+
+    class CapturingDatabase:
+        async def execute(self, statement, parameters=None):
+            captured["sql"] = str(statement)
+            captured["parameters"] = parameters
+            return _Result()
+
+    class CapturingContext:
+        async def __aenter__(self):
+            return CapturingDatabase()
+
+        async def __aexit__(self, exc_type, exc, traceback):
+            return False
+
+    monkeypatch.setattr(
+        "src.db.base.get_db_context",
+        lambda project_id=None: CapturingContext(),
+    )
+
+    inserted, modified = asyncio.run(
+        asset_scanner_job._upsert_assets(
+            [
+                {
+                    "asset_key": "k8s/deployment/awoooi-prod/api",
+                    "asset_type": "k8s_workload",
+                    "host": None,
+                    "namespace": "awoooi-prod",
+                    "name": "api",
+                    "metadata": {},
+                    "tags": [],
+                }
+            ],
+            "00000000-0000-0000-0000-000000000001",
+        )
+    )
+
+    assert (inserted, modified) == (1, 0)
+    assert captured["parameters"]["owner_team"] == "awoooi"
+    persisted_metadata = json.loads(captured["parameters"]["md"])
+    assert persisted_metadata["owner_inference"] == {
+        "candidate": "awoooi",
+        "source": "k8s.namespace",
+        "policy": "deterministic_v1",
+    }
+    sql = str(captured["sql"])
+    assert "BTRIM(asset_inventory.owner_team) = ''" in sql
+    assert "ELSE asset_inventory.owner_team" in sql
+
+
 def test_asset_key_integrity_preflight_accepts_unique_inventory(monkeypatch) -> None:
     requested_projects: list[str | None] = []
     statements: list[str] = []

apps/api/tests/test_coverage_evaluator_job.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import asyncio
+import inspect
+
+from src.jobs import coverage_evaluator_job
+
+
+class _Result:
+    rowcount = 1
+
+    def scalar(self):
+        return "00000000-0000-0000-0000-000000000001"
+
+    def fetchall(self):
+        return []
+
+    def fetchone(self):
+        return None
+
+
+class _Database:
+    async def execute(self, statement, parameters=None):
+        return _Result()
+
+
+class _Context:
+    async def __aenter__(self):
+        return _Database()
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+
+def test_coverage_evaluator_scopes_every_database_operation(monkeypatch) -> None:
+    requested_projects: list[str | None] = []
+
+    def fake_db_context(project_id: str | None = None):
+        requested_projects.append(project_id)
+        return _Context()
+
+    async def fake_prometheus_targets() -> list[str]:
+        return ["192.0.2.10"]
+
+    monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
+    monkeypatch.setattr(
+        coverage_evaluator_job,
+        "_fetch_prometheus_target_ips",
+        fake_prometheus_targets,
+    )
+
+    async def exercise_evaluator_operations() -> None:
+        run_id = await coverage_evaluator_job._get_latest_run_id()
+        assert run_id == "00000000-0000-0000-0000-000000000001"
+
+        await coverage_evaluator_job._fetch_red_summary()
+        await coverage_evaluator_job._evaluate_monitoring(run_id)
+        await coverage_evaluator_job._evaluate_alerting(run_id)
+        await coverage_evaluator_job._evaluate_km_coverage(run_id)
+        await coverage_evaluator_job._evaluate_playbook_coverage(run_id)
+        await coverage_evaluator_job._evaluate_remediation_coverage(run_id)
+        await coverage_evaluator_job._evaluate_rule_matching_coverage(run_id)
+        await coverage_evaluator_job._evaluate_rule_creation_coverage(run_id)
+        await coverage_evaluator_job._log_aol(
+            stats={"monitoring_updated": 1},
+            duration_ms=10,
+            error=None,
+        )
+        assert (
+            await coverage_evaluator_job._auto_create_rules_for_uncovered_assets(
+                run_id
+            )
+            == 0
+        )
+
+    asyncio.run(exercise_evaluator_operations())
+
+    assert requested_projects == ["awoooi"] * 11
+
+
+def test_coverage_evaluator_has_no_unscoped_database_context() -> None:
+    source = inspect.getsource(coverage_evaluator_job)
+
+    assert "get_db_context()" not in source
+    assert source.count("get_db_context(_PROJECT_ID)") == 12