From 0eb003d150343374ff609edd2c273fa055711fe8 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 20:54:29 +0800 Subject: [PATCH] feat(security): inventory database catalog assets --- apps/api/src/jobs/asset_scanner_job.py | 144 +++++++++++++++++++++-- apps/api/tests/test_asset_scanner_job.py | 63 ++++++++++ 2 files changed, 196 insertions(+), 11 deletions(-) diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 68227b68d..655077d76 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -1,14 +1,16 @@ """ Asset Scanner Job — ADR-090 §4.1 資產盤點 cron ================================================ -每 1 小時掃描 K8s + 寫入 asset_inventory + asset_discovery_run + asset_coverage_snapshot. +每 1 小時掃描 K8s、Prometheus 與 PostgreSQL catalog，寫入資產控制平面。 職責邊界: ✅ K8s API 列出全部 namespace 的 pods/deployments/services (shallow scan) + ✅ Prometheus targets 發現主機與監控目標 + ✅ PostgreSQL catalog 發現 database/table/view metadata（不讀資料列） ✅ UPSERT asset_inventory (asset_key 為 UNIQUE) ✅ 為每個 active asset 寫 7 維 asset_coverage_snapshot (預設 unknown,後續 service 補) ✅ 完成時寫 automation_operation_log(asset_discovered) - ❌ 不掃 Prometheus targets / Gitea repos / Docker compose (留下一階段) + ❌ 不掃 Gitea repos / Docker compose / package SBOM (留下一階段) ❌ 不算 capacity 欄位 (留 capacity_scanner_job) 設計鐵律 (參考 ADR-090 §3.4): @@ -88,7 +90,7 @@ async def run_asset_scanner_loop() -> None: async def scan_once( triggered_by: str = "cron", - scope: tuple[str, ...] = ("k8s",), + scope: tuple[str, ...] = ("k8s", "prometheus", "database_catalog"), scan_depth: str = "shallow", ) -> str | None: """ @@ -117,11 +119,11 @@ async def scan_once( # 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload), # services (k8s_resource), nodes (host), configmaps (k8s_resource) # 跳過: secrets (awoooi-executor RBAC 不允許 list) - k8s_assets, relationships = await _collect_all_k8s_assets() - total_count = len(k8s_assets) + runtime_assets, relationships = await _collect_runtime_assets() + total_count = len(runtime_assets) # UPSERT inventory - new_count, modified_count = await _upsert_assets(k8s_assets, run_id) + new_count, modified_count = await _upsert_assets(runtime_assets, run_id) # 建立 asset_relationship (OwnerReference + Service selector + Pod volumes) rel_written = await _upsert_relationships(relationships) @@ -158,6 +160,8 @@ async def scan_once( duration_ms=duration_ms, status=final_status, error=error_msg, + scope=list(scope), + scan_depth=scan_depth, ) logger.info( @@ -341,7 +345,7 @@ def _build_configmap_asset(item: dict[str, Any]) -> dict[str, Any]: } -async def _collect_all_k8s_assets() -> tuple[list[dict[str, Any]], list[dict[str, str]]]: +async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str, str]]]: """ 掃多種 K8s 資源類型 + 提取 relationship. @@ -489,6 +493,14 @@ async def _collect_all_k8s_assets() -> tuple[list[dict[str, Any]], list[dict[str except Exception as e: logger.warning("collect_prometheus_targets_failed", error=str(e)) + # 7. PostgreSQL catalog — database/table/view metadata only, never row data. + try: + database_assets, database_relationships = await _collect_database_catalog_assets() + assets.extend(database_assets) + relationships.extend(database_relationships) + except Exception as e: + logger.warning("collect_database_catalog_failed", error=str(e)) + return assets, relationships @@ -622,6 +634,108 @@ async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict return assets, relationships +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 + + from src.db.base import get_db_context + + async with get_db_context(_PROJECT_ID) as db: + result = await db.execute( + _sql( + """ + SELECT + current_database() AS database_name, + namespace.nspname AS schema_name, + relation.relname AS relation_name, + CASE relation.relkind + WHEN 'r' THEN 'table' + WHEN 'p' THEN 'partitioned_table' + WHEN 'v' THEN 'view' + WHEN 'm' THEN 'materialized_view' + WHEN 'f' THEN 'foreign_table' + END AS relation_kind, + relation.relrowsecurity AS row_security_enabled + FROM pg_class relation + JOIN pg_namespace namespace + ON namespace.oid = relation.relnamespace + WHERE relation.relkind IN ('r', 'p', 'v', 'm', 'f') + AND namespace.nspname NOT IN ('pg_catalog', 'information_schema') + AND namespace.nspname NOT LIKE 'pg_toast%' + AND namespace.nspname NOT LIKE 'pg_temp_%' + ORDER BY namespace.nspname, relation.relname + """ + ) + ) + rows = result.fetchall() + + if not rows: + return [], [] + + first = rows[0]._mapping if hasattr(rows[0], "_mapping") else rows[0] + database_name = str(first["database_name"]) + database_key = f"postgres/database/{database_name}" + assets: list[dict[str, Any]] = [ + { + "asset_key": database_key, + "asset_type": "database", + "host": None, + "namespace": None, + "name": database_name, + "metadata": { + "engine": "postgresql", + "discovered_by": "pg_catalog", + "row_data_read": False, + }, + "tags": ["engine:postgresql", "source:pg_catalog"], + } + ] + relationships: list[dict[str, str]] = [] + + for row in rows: + 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}" + assets.append( + { + "asset_key": relation_key, + "asset_type": "table", + "host": None, + "namespace": schema_name, + "name": relation_name, + "metadata": { + "database": database_name, + "schema": schema_name, + "relation_kind": str(values["relation_kind"]), + "row_security_enabled": bool(values["row_security_enabled"]), + "row_data_read": False, + }, + "tags": [ + f"database:{database_name}", + f"schema:{schema_name}", + "source:pg_catalog", + ], + } + ) + relationships.append( + { + "from_key": relation_key, + "to_key": database_key, + "relationship_type": "depends_on", + } + ) + + logger.info( + "database_catalog_assets_collected", + database_count=1, + relation_count=len(rows), + ) + return assets, relationships + + # ============================================================================ # DB 寫入 # ============================================================================ @@ -659,7 +773,13 @@ async def _start_discovery_run( "tb": triggered_by, "scope": scope, "sd": scan_depth, - "tools": _json.dumps({"k8s": "kubectl_get pods --all-namespaces"}), + "tools": _json.dumps( + { + "k8s": "kubectl_get metadata", + "prometheus": "active_target_metadata", + "database_catalog": "pg_catalog metadata", + } + ), }, ) run_id = row.scalar() @@ -874,6 +994,8 @@ async def _log_aol_asset_discovered( duration_ms: int, status: str, error: str | None, + scope: list[str], + scan_depth: str, ) -> None: """寫 automation_operation_log(asset_discovered)。失敗只 log 不阻塞。""" try: @@ -885,8 +1007,8 @@ async def _log_aol_asset_discovered( input_payload = { "run_id": run_id, "triggered_by": triggered_by, - "scope": ["k8s"], - "scan_depth": "shallow", + "scope": scope, + "scan_depth": scan_depth, } output_payload = { "run_id": run_id, @@ -919,7 +1041,7 @@ async def _log_aol_asset_discovered( "rid": run_id, "dur": duration_ms, "err": (error or "")[:2000] if error else None, - "tags": ["asset_scanner", "discovery", "k8s"], + "tags": ["asset_scanner", "discovery", *scope], }, ) except Exception as e: diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index 6106ca412..be1367fd5 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/apps/api/tests/test_asset_scanner_job.py @@ -87,8 +87,71 @@ def test_background_scanner_scopes_every_database_operation(monkeypatch) -> None duration_ms=10, status="success", error=None, + scope=["k8s", "prometheus", "database_catalog"], + scan_depth="shallow", ) asyncio.run(exercise_scanner_writes()) assert requested_projects == ["awoooi"] * 6 + + +def test_database_catalog_collector_reads_metadata_only(monkeypatch) -> None: + requested_projects: list[str | None] = [] + + class CatalogResult: + def fetchall(self): + return [ + type( + "CatalogRow", + (), + { + "_mapping": { + "database_name": "awoooi", + "schema_name": "public", + "relation_name": "incidents", + "relation_kind": "table", + "row_security_enabled": True, + } + }, + )() + ] + + class CatalogDatabase: + async def execute(self, statement, parameters=None): + return CatalogResult() + + class CatalogContext: + async def __aenter__(self): + return CatalogDatabase() + + async def __aexit__(self, exc_type, exc, traceback): + return False + + def fake_db_context(project_id: str | None = None): + requested_projects.append(project_id) + return CatalogContext() + + monkeypatch.setattr("src.db.base.get_db_context", fake_db_context) + + assets, relationships = asyncio.run( + asset_scanner_job._collect_database_catalog_assets() + ) + + assert requested_projects == ["awoooi"] + assert [asset["asset_type"] for asset in assets] == ["database", "table"] + assert assets[0]["metadata"]["row_data_read"] is False + assert assets[1]["metadata"] == { + "database": "awoooi", + "schema": "public", + "relation_kind": "table", + "row_security_enabled": True, + "row_data_read": False, + } + assert relationships == [ + { + "from_key": "postgres/relation/awoooi/public/incidents", + "to_key": "postgres/database/awoooi", + "relationship_type": "depends_on", + } + ]