wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

feat(security): inventory database catalog assets

Browse Source
This commit is contained in:
ogt
2026-07-11 20:54:29 +08:00
parent ad3219c6cf
commit 0eb003d150
2 changed files with 196 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

144
apps/api/src/jobs/asset_scanner_job.py
View File

@@ -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:

63
apps/api/tests/test_asset_scanner_job.py
View File

@@ -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",
}
]