diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 7ba0c5b04..16aaa77c3 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -30,6 +30,7 @@ from __future__ import annotations import asyncio import hashlib import json as _json +import re import time as _time from pathlib import Path from typing import Any @@ -47,6 +48,37 @@ _KUBECTL_TIMEOUT_SEC = 30 _LOOP_BACKOFF_SEC = 300 # 異常時 backoff 5 分鐘 _PROJECT_ID = "awoooi" +_OWNER_SIGNAL_KEYS = ( + "wooo.work/owner-team", + "app.kubernetes.io/owner", + "owner_team", + "owner-team", + "owner", + "team", + "project_id", + "project", + "app.kubernetes.io/part-of", +) +_GENERIC_NAMESPACES = frozenset( + { + "default", + "kube-system", + "kube-public", + "kube-node-lease", + "monitoring", + "observability", + "cert-manager", + "ingress-nginx", + "metallb-system", + "longhorn-system", + } +) +_PROJECT_SCOPED_ASSET_PREFIXES = ( + "postgres/", + "ai_catalog/", + "model_registry/", +) + # 7 個自動化覆蓋維度 (ADR-090 §3.5) _COVERAGE_DIMENSIONS = ( "auto_monitoring", @@ -70,6 +102,80 @@ _K8S_RESOURCE_TO_ASSET_TYPE = { } +def _normalize_owner_team(value: Any) -> str | None: + if value is None: + return None + normalized = re.sub( + r"[^a-z0-9]+", + "_", + str(value).strip().lower(), + ).strip("_") + if not normalized or len(normalized) > 64: + return None + if normalized in {"default", "none", "null", "unknown", "unowned"}: + return None + return normalized + + +def _owner_from_namespace(value: Any) -> str | None: + namespace = _normalize_owner_team(value) + if not namespace or namespace.replace("_", "-") in _GENERIC_NAMESPACES: + return None + + for suffix in ( + "_production", + "_prod", + "_staging", + "_stage", + "_development", + "_dev", + "_test", + ): + if namespace.endswith(suffix): + namespace = namespace[: -len(suffix)].rstrip("_") + break + + if not namespace or namespace.replace("_", "-") in _GENERIC_NAMESPACES: + return None + return namespace + + +def _resolve_owner_team(asset: dict[str, Any]) -> tuple[str | None, str]: + """Resolve an auditable owner candidate without overwriting human ownership.""" + metadata = asset.get("metadata") or {} + labels = metadata.get("labels") if isinstance(metadata, dict) else None + if isinstance(labels, dict): + for key in _OWNER_SIGNAL_KEYS: + owner = _normalize_owner_team(labels.get(key)) + if owner: + return owner, f"metadata.labels.{key}" + + for key in ("namespace", "kubernetes_namespace"): + owner = _owner_from_namespace(labels.get(key)) + if owner: + return owner, f"metadata.labels.{key}" + + tags = asset.get("tags") or [] + if isinstance(tags, (list, tuple)): + for tag in tags: + prefix, separator, value = str(tag).partition(":") + if separator and prefix in {"owner", "team", "project"}: + owner = _normalize_owner_team(value) + if owner: + return owner, f"tag.{prefix}" + + asset_key = str(asset.get("asset_key") or "") + if asset_key.startswith("k8s/"): + owner = _owner_from_namespace(asset.get("namespace")) + if owner: + return owner, "k8s.namespace" + + if asset_key.startswith(_PROJECT_SCOPED_ASSET_PREFIXES): + return _PROJECT_ID, "project_scoped_collector" + + return None, "unresolved" + + # ============================================================================ # Public entry — main.py lifespan 呼叫 # ============================================================================ @@ -1497,17 +1603,24 @@ async def _upsert_assets( try: async with get_db_context(_PROJECT_ID) as db: for a in assets: + owner_team, owner_source = _resolve_owner_team(a) + metadata = dict(a.get("metadata") or {}) + metadata["owner_inference"] = { + "candidate": owner_team, + "source": owner_source, + "policy": "deterministic_v1", + } # xmax = 0 表示 INSERT (新),否則表示 UPDATE (修改) row = await db.execute( _sql( """ INSERT INTO asset_inventory ( asset_key, asset_type, host, namespace, name, - metadata, tags, environment, lifecycle_state, + metadata, tags, owner_team, environment, lifecycle_state, first_seen_at, last_seen_at ) VALUES ( :ak, :at, :host, :ns, :name, - CAST(:md AS jsonb), :tags, 'prod', 'active', + CAST(:md AS jsonb), :tags, :owner_team, 'prod', 'active', NOW(), NOW() ) ON CONFLICT (asset_key) DO UPDATE @@ -1515,6 +1628,12 @@ async def _upsert_assets( host = EXCLUDED.host, metadata = EXCLUDED.metadata, tags = EXCLUDED.tags, + owner_team = CASE + WHEN asset_inventory.owner_team IS NULL + OR BTRIM(asset_inventory.owner_team) = '' + THEN EXCLUDED.owner_team + ELSE asset_inventory.owner_team + END, lifecycle_state = 'active', updated_at = NOW(), decommissioned_at = NULL @@ -1527,8 +1646,9 @@ async def _upsert_assets( "host": a["host"], "ns": a["namespace"], "name": a["name"], - "md": _json.dumps(a["metadata"], ensure_ascii=False), + "md": _json.dumps(metadata, ensure_ascii=False), "tags": a["tags"], + "owner_team": owner_team, }, ) _, inserted = row.one() diff --git a/apps/api/src/jobs/coverage_evaluator_job.py b/apps/api/src/jobs/coverage_evaluator_job.py index 8d9843dbd..1a9a92881 100644 --- a/apps/api/src/jobs/coverage_evaluator_job.py +++ b/apps/api/src/jobs/coverage_evaluator_job.py @@ -43,6 +43,7 @@ _EVAL_INTERVAL_SEC = 3600 _FIRST_DELAY_SEC = 300 _HTTP_TIMEOUT_SEC = 10 _LOOP_BACKOFF_SEC = 600 +_PROJECT_ID = "awoooi" # ============================================================================ @@ -168,7 +169,7 @@ async def _fetch_red_summary() -> dict[str, Any] | None: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: # 總覽: 每維度 red count dim_rows = await db.execute(_sql(""" SELECT dimension, count(*) AS cnt @@ -361,7 +362,7 @@ async def _get_latest_run_id() -> str | None: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: row = await db.execute( _sql("SELECT run_id FROM asset_discovery_run WHERE status='success' ORDER BY ended_at DESC LIMIT 1"), ) @@ -391,7 +392,7 @@ async def _evaluate_monitoring(run_id: str) -> int: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: # host asset: 看 metadata.internal_ip 是否在 targets # 其他 asset type: 留 unknown (Prometheus 不直接 scrape) result = await db.execute( @@ -464,7 +465,7 @@ async def _evaluate_alerting(run_id: str) -> int: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: result = await db.execute( _sql(""" UPDATE asset_coverage_snapshot cs @@ -511,7 +512,7 @@ async def _evaluate_km_coverage(run_id: str) -> int: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: result = await db.execute( _sql(""" UPDATE asset_coverage_snapshot cs @@ -555,7 +556,7 @@ async def _evaluate_playbook_coverage(run_id: str) -> int: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: result = await db.execute( _sql(""" UPDATE asset_coverage_snapshot cs @@ -596,7 +597,7 @@ async def _evaluate_remediation_coverage(run_id: str) -> int: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: result = await db.execute( _sql(""" UPDATE asset_coverage_snapshot cs @@ -638,7 +639,7 @@ async def _evaluate_rule_matching_coverage(run_id: str) -> int: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: result = await db.execute( _sql(""" UPDATE asset_coverage_snapshot cs @@ -685,7 +686,7 @@ async def _evaluate_rule_creation_coverage(run_id: str) -> int: from src.db.base import get_db_context try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: result = await db.execute( _sql(""" UPDATE asset_coverage_snapshot cs @@ -728,7 +729,7 @@ async def _log_aol(stats: dict[str, int], duration_ms: int, error: str | None) - from src.db.base import get_db_context aol_status = "failed" if error else "success" - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: await db.execute( _sql(""" INSERT INTO automation_operation_log ( @@ -788,7 +789,7 @@ async def _auto_create_rules_for_uncovered_assets(run_id: str | None) -> int: created = 0 try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: # 查 auto_alerting=red 的 host 和 k8s_workload asset（最多 5 筆） rows = await db.execute( _sql(""" @@ -873,7 +874,7 @@ async def _auto_create_rules_for_uncovered_assets(run_id: str | None) -> int: # UPSERT 進 alert_rule_catalog（source='ai_generated'） # 用 RETURNING 判斷是否實際插入（ON CONFLICT DO NOTHING 衝突時無 RETURNING row） try: - async with get_db_context() as db: + async with get_db_context(_PROJECT_ID) as db: row = await db.execute( _sql(""" INSERT INTO alert_rule_catalog ( diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index 90bc09d02..b122a743a 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/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] = [] diff --git a/apps/api/tests/test_coverage_evaluator_job.py b/apps/api/tests/test_coverage_evaluator_job.py new file mode 100644 index 000000000..c4d0d1446 --- /dev/null +++ b/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