diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index 7370d0fff..995e735c2 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -405,6 +405,10 @@ jobs: ;; apps/api/migrations/adr090f_log_controlled_writeback_dispatch_operation_type_down.sql) ;; + apps/api/migrations/asset_inventory_database_types_2026-07-14.sql) + ;; + apps/api/migrations/asset_inventory_database_types_2026-07-14_down.sql) + ;; apps/api/src/services/auto_approve.py) ;; apps/api/src/services/decision_fusion.py) @@ -1814,6 +1818,144 @@ jobs: echo "⚡ no Docker build lock to release" fi + # Production drift proved the original CREATE TABLE IF NOT EXISTS + # migration did not update an older CHECK constraint. Reconcile this + # one additive allowlist before rolling out the scanner and verify it + # independently without exposing connection details. + - name: Reconcile Asset Inventory Type Constraint + env: + DATABASE_URL: ${{ secrets.DATABASE_URL }} + MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }} + run: | + set -euo pipefail + if [ -z "${MIGRATION_DATABASE_URL:-}" ]; then + echo "::error::MIGRATION_DATABASE_URL secret not set in Gitea" + exit 1 + fi + docker run --rm --network host -i \ + -e DATABASE_URL \ + -e MIGRATION_DATABASE_URL \ + -v "$PWD/apps/api/migrations/asset_inventory_database_types_2026-07-14.sql:/tmp/asset_inventory_type.sql:ro" \ + "${HARBOR}/awoooi/api:${{ github.sha }}" python - <<'PY' + import asyncio + import os + import re + from pathlib import Path + + import asyncpg + + MIGRATION = Path("/tmp/asset_inventory_type.sql").read_text(encoding="utf-8") + REQUIRED_TYPES = { + "host", "container", "k8s_workload", "k8s_resource", "database", "table", + "website", "api_endpoint", "package", "log_stream", "km_entry", "frontend", + "backend", "ci_pipeline", "gitea_repo", "monitoring_target", "secret", "volume", + "network", "certificate", "scheduled_job", "message_queue", "cache", "dashboard", + "ai_agent", "llm_model", "third_party_service", "backup_target", + } + + + def normalize_url(value: str) -> str: + return value.replace("postgresql+asyncpg://", "postgresql://", 1) + + + async def constraint_status( + connection: asyncpg.Connection, + ) -> tuple[bool, str, bool]: + row = await connection.fetchrow( + """ + SELECT + constraint_row.convalidated, + pg_get_constraintdef(constraint_row.oid) AS definition, + EXISTS ( + SELECT 1 + FROM automation_operation_log + WHERE actor = 'gitea_cd_asset_schema_reconciler' + AND operation_type = 'asset_discovered' + AND status = 'success' + AND input ->> 'migration' = + 'asset_inventory_database_types_2026-07-14' + ) AS durable_receipt_present + FROM pg_constraint constraint_row + WHERE constraint_row.conrelid = 'asset_inventory'::regclass + AND constraint_row.conname = 'asset_inventory_type_valid' + """ + ) + if row is None: + return False, "", False + return ( + bool(row["convalidated"]), + str(row["definition"] or ""), + bool(row["durable_receipt_present"]), + ) + + + def allowed_types(definition: str) -> set[str]: + return set(re.findall(r"'([^']+)'(?:::text)?", definition)) + + + async def reconcile(url: str) -> tuple[bool, str, bool, bool]: + connection = await asyncpg.connect(normalize_url(url), timeout=10) + applied = False + try: + validated, definition, receipt_present = await constraint_status(connection) + if ( + not validated + or allowed_types(definition) != REQUIRED_TYPES + or not receipt_present + ): + async with connection.transaction(): + await connection.execute(MIGRATION) + applied = True + validated, definition, receipt_present = await constraint_status(connection) + return validated, definition, receipt_present, applied + finally: + await connection.close() + + + async def main() -> None: + candidates = [ + ("migration_role", os.environ["MIGRATION_DATABASE_URL"]), + ("owner_role", os.environ.get("DATABASE_URL", "")), + ] + for index, (role, url) in enumerate(candidates): + if not url: + continue + try: + validated, definition, receipt_present, applied = await reconcile(url) + type_set = allowed_types(definition) + database_allowed = "database" in type_set + table_allowed = "table" in type_set + type_set_exact = type_set == REQUIRED_TYPES + print(f"asset_inventory_type_migration_role={role}") + print(f"asset_inventory_type_migration_applied={str(applied).lower()}") + print(f"database_type_allowed={str(database_allowed).lower()}") + print(f"table_type_allowed={str(table_allowed).lower()}") + print(f"constraint_type_set_exact={str(type_set_exact).lower()}") + print(f"constraint_validated={str(validated).lower()}") + print(f"durable_receipt_present={str(receipt_present).lower()}") + if not ( + validated + and database_allowed + and table_allowed + and type_set_exact + and receipt_present + ): + raise RuntimeError("asset_inventory_type_constraint_verifier_failed") + return + except asyncpg.PostgresError as exc: + permission_denied = getattr(exc, "sqlstate", "") == "42501" + has_owner_fallback = index == 0 and bool(candidates[1][1]) + if permission_denied and has_owner_fallback: + print("asset_inventory_type_migration_owner_fallback=true") + continue + print(f"asset_inventory_type_migration_error_class={type(exc).__name__}") + raise + raise RuntimeError("asset_inventory_type_migration_no_usable_connection") + + + asyncio.run(main()) + PY + # 2026-03-31 ogt: 移除中間通知 # 2026-03-31 ogt: P0-1 Secrets 自動注入 (ADR-035 強制) diff --git a/apps/api/migrations/asset_inventory_database_types_2026-07-14.sql b/apps/api/migrations/asset_inventory_database_types_2026-07-14.sql new file mode 100644 index 000000000..b78dce9d2 --- /dev/null +++ b/apps/api/migrations/asset_inventory_database_types_2026-07-14.sql @@ -0,0 +1,40 @@ +-- Reconcile the production asset_inventory type constraint with source truth. +-- Safety: allowlist-only DDL, bounded locks, no row data reads or deletes. + +SET lock_timeout = '5s'; +SET statement_timeout = '30s'; + +ALTER TABLE asset_inventory + DROP CONSTRAINT IF EXISTS asset_inventory_type_valid; + +ALTER TABLE asset_inventory + ADD CONSTRAINT asset_inventory_type_valid CHECK (asset_type IN ( + 'host','container','k8s_workload','k8s_resource','database','table', + 'website','api_endpoint','package','log_stream','km_entry', + 'frontend','backend','ci_pipeline','gitea_repo','monitoring_target', + 'secret','volume','network','certificate','scheduled_job', + 'message_queue','cache','dashboard','ai_agent','llm_model', + 'third_party_service','backup_target' + )) NOT VALID; + +ALTER TABLE asset_inventory + VALIDATE CONSTRAINT asset_inventory_type_valid; + +INSERT INTO automation_operation_log ( + operation_type, + actor, + status, + input, + output, + tags +) VALUES ( + 'asset_discovered', + 'gitea_cd_asset_schema_reconciler', + 'success', + '{"work_item_id":"AIA-P0-006","migration":"asset_inventory_database_types_2026-07-14"}'::jsonb, + '{"constraint_validated":true,"constraint_type_set_exact":true,"database_type_allowed":true,"table_type_allowed":true}'::jsonb, + ARRAY['asset_control_plane','schema_migration','controlled_apply','post_verified'] +); + +COMMENT ON CONSTRAINT asset_inventory_type_valid ON asset_inventory IS + 'Asset control-plane allowlist reconciled by Gitea CD; includes database and table metadata assets.'; diff --git a/apps/api/migrations/asset_inventory_database_types_2026-07-14_down.sql b/apps/api/migrations/asset_inventory_database_types_2026-07-14_down.sql new file mode 100644 index 000000000..fc3689623 --- /dev/null +++ b/apps/api/migrations/asset_inventory_database_types_2026-07-14_down.sql @@ -0,0 +1,57 @@ +-- Roll back database/table asset types only when no such inventory rows exist. +-- This script never deletes rows to make rollback possible. + +SET lock_timeout = '5s'; +SET statement_timeout = '30s'; + +DO $migration$ +DECLARE + dependent_rows BIGINT; +BEGIN + SELECT COUNT(*) + INTO dependent_rows + FROM asset_inventory + WHERE asset_type IN ('database', 'table'); + + IF dependent_rows > 0 THEN + RAISE EXCEPTION + 'rollback blocked: % database/table asset rows must be reconciled first', + dependent_rows; + END IF; + + ALTER TABLE asset_inventory + DROP CONSTRAINT IF EXISTS asset_inventory_type_valid; + + ALTER TABLE asset_inventory + ADD CONSTRAINT asset_inventory_type_valid CHECK (asset_type IN ( + 'host','container','k8s_workload','k8s_resource', + 'website','api_endpoint','package','log_stream','km_entry', + 'frontend','backend','ci_pipeline','gitea_repo','monitoring_target', + 'secret','volume','network','certificate','scheduled_job', + 'message_queue','cache','dashboard','ai_agent','llm_model', + 'third_party_service','backup_target' + )) NOT VALID; + + ALTER TABLE asset_inventory + VALIDATE CONSTRAINT asset_inventory_type_valid; + + INSERT INTO automation_operation_log ( + operation_type, + actor, + status, + input, + output, + tags + ) VALUES ( + 'asset_discovered', + 'gitea_cd_asset_schema_reconciler', + 'rolled_back', + '{"work_item_id":"AIA-P0-006","migration":"asset_inventory_database_types_2026-07-14_down"}'::jsonb, + '{"database_type_allowed":false,"table_type_allowed":false,"dependent_rows":0}'::jsonb, + ARRAY['asset_control_plane','schema_migration','controlled_rollback'] + ); +END +$migration$; + +COMMENT ON CONSTRAINT asset_inventory_type_valid ON asset_inventory IS + 'Rollback allowlist without database/table types; blocked while dependent asset rows exist.'; diff --git a/apps/api/src/jobs/asset_scanner_job.py b/apps/api/src/jobs/asset_scanner_job.py index 655077d76..7f15c2c61 100644 --- a/apps/api/src/jobs/asset_scanner_job.py +++ b/apps/api/src/jobs/asset_scanner_job.py @@ -92,7 +92,7 @@ async def scan_once( triggered_by: str = "cron", scope: tuple[str, ...] = ("k8s", "prometheus", "database_catalog"), scan_depth: str = "shallow", -) -> str | None: +) -> str: """ 執行一次資產盤點。 @@ -102,15 +102,14 @@ async def scan_once( scan_depth: shallow (僅 list) / deep (含 describe) / full Returns: - run_id (UUID 字串) 或 None (寫 header 失敗時) + run_id (UUID 字串)。無法建立 durable header 時會拋出例外。 """ started_ms = _time.time() run_id = await _start_discovery_run(triggered_by, list(scope), scan_depth) - if not run_id: - return None new_count = 0 modified_count = 0 + disappeared_count = 0 total_count = 0 error_msg: str | None = None @@ -119,11 +118,29 @@ async def scan_once( # 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload), # services (k8s_resource), nodes (host), configmaps (k8s_resource) # 跳過: secrets (awoooi-executor RBAC 不允許 list) - runtime_assets, relationships = await _collect_runtime_assets() + ( + runtime_assets, + relationships, + reconciled_prefixes, + collector_errors, + ) = await _collect_runtime_assets() total_count = len(runtime_assets) # UPSERT inventory new_count, modified_count = await _upsert_assets(runtime_assets, run_id) + if new_count + modified_count != total_count: + raise RuntimeError( + "asset_upsert_count_mismatch " + f"expected={total_count} persisted={new_count + modified_count}" + ) + + # Only reconcile sources that completed collection in this run. This + # prevents a failed/empty collector from deprecating healthy assets. + disappeared_count = await _deprecate_missing_assets( + runtime_assets, + reconciled_prefixes, + run_id, + ) # 建立 asset_relationship (OwnerReference + Service selector + Pod volumes) rel_written = await _upsert_relationships(relationships) @@ -132,6 +149,10 @@ async def scan_once( await _write_coverage_snapshots(run_id) logger.info("asset_scan_relationships_written", run_id=run_id, relationships=rel_written) + if collector_errors: + raise RuntimeError( + "asset_collector_failures=" + ",".join(collector_errors) + ) except Exception as e: error_msg = f"{type(e).__name__}: {e}"[:1000] @@ -140,26 +161,18 @@ async def scan_once( duration_ms = int((_time.time() - started_ms) * 1000) final_status = "failed" if error_msg else "success" - await _finish_discovery_run( + # The discovery terminal and AOL receipt share one DB transaction. A run + # cannot be marked successful without its durable audit receipt. + await _finalize_discovery_run( run_id=run_id, + triggered_by=triggered_by, status=final_status, total_assets=total_count, new_assets=new_count, modified_assets=modified_count, + disappeared_assets=disappeared_count, duration_ms=duration_ms, error=error_msg, - ) - - # ADR-090 § aol 留痕 — asset_discovered 是合法 op_type - await _log_aol_asset_discovered( - run_id=run_id, - triggered_by=triggered_by, - total=total_count, - new=new_count, - modified=modified_count, - duration_ms=duration_ms, - status=final_status, - error=error_msg, scope=list(scope), scan_depth=scan_depth, ) @@ -171,6 +184,7 @@ async def scan_once( total=total_count, new=new_count, modified=modified_count, + disappeared=disappeared_count, duration_ms=duration_ms, ) return run_id @@ -345,7 +359,12 @@ def _build_configmap_asset(item: dict[str, Any]) -> dict[str, Any]: } -async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str, str]]]: +async def _collect_runtime_assets() -> tuple[ + list[dict[str, Any]], + list[dict[str, str]], + tuple[str, ...], + tuple[str, ...], +]: """ 掃多種 K8s 資源類型 + 提取 relationship. @@ -360,10 +379,16 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str 原 code 跳過 ReplicaSet → Pod→Deployment 全部漏掉. 修: 先掃 ReplicaSet 建 rs_to_deployment map,Pod 用 rs_name 反查 Deployment. - 回傳 (assets, relationships) tuple. + 回傳 (assets, relationships, reconciled_asset_key_prefixes, collector_errors) tuple. + + Only prefixes whose source query completed successfully are returned. The + caller may safely deprecate missing assets within those prefixes without + treating collector failures as asset deletion. """ assets: list[dict[str, Any]] = [] relationships: list[dict[str, str]] = [] + reconciled_prefixes: set[str] = set() + collector_errors: list[str] = [] # 0. ReplicaSets — 僅作為 Pod→Deployment 橋樑,不寫入 asset_inventory rs_to_deployment: dict[str, str] = {} # "ns/rs_name" -> "deployment_name" @@ -378,14 +403,17 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str rs_to_deployment[f"{rs_ns}/{rs_name}"] = ref.get("name", "") except Exception as e: logger.warning("collect_replicasets_failed", error=str(e)) + collector_errors.append("k8s_replicasets") # 1. Nodes (不帶 ns) try: payload = await _fetch_kubectl_json("nodes", all_namespaces=False) for item in payload.get("items", []) or []: assets.append(_build_node_asset(item)) + reconciled_prefixes.add("k8s/node/") except Exception as e: logger.warning("collect_nodes_failed", error=str(e)) + collector_errors.append("k8s_nodes") # 2. Pods — 主體 + 從 ownerReferences 建 relationship pod_by_key: dict[str, dict[str, Any]] = {} @@ -436,8 +464,10 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str "to_key": f"k8s/configmap/{ns}/{cm}", "relationship_type": "depends_on", }) + reconciled_prefixes.add("k8s/pod/") except Exception as e: logger.warning("collect_pods_failed", error=str(e)) + collector_errors.append("k8s_pods") # 3. Deployments / StatefulSets / DaemonSets for kind, resource in (("Deployment", "deployments"), ("StatefulSet", "statefulsets"), ("DaemonSet", "daemonsets")): @@ -445,8 +475,10 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str payload = await _fetch_kubectl_json(resource) for item in payload.get("items", []) or []: assets.append(_build_workload_asset(item, kind)) + reconciled_prefixes.add(f"k8s/{kind.lower()}/") except Exception as e: logger.warning(f"collect_{resource}_failed", error=str(e)) + collector_errors.append(f"k8s_{resource}") # 4. Services + routes_to Pod (via selector match) try: @@ -471,16 +503,20 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str "to_key": pod_key, "relationship_type": "routes_to", }) + reconciled_prefixes.add("k8s/service/") except Exception as e: logger.warning("collect_services_failed", error=str(e)) + collector_errors.append("k8s_services") # 5. ConfigMaps try: payload = await _fetch_kubectl_json("configmaps") for item in payload.get("items", []) or []: assets.append(_build_configmap_asset(item)) + reconciled_prefixes.add("k8s/configmap/") except Exception as e: logger.warning("collect_configmaps_failed", error=str(e)) + collector_errors.append("k8s_configmaps") # 6. Prometheus targets — 補齊 host-install services (110/112/188/125 等非 K8s) # Gap 1 修補 (2026-04-19 audit): 原本 asset_inventory 只涵蓋 K8s, @@ -490,18 +526,30 @@ async def _collect_runtime_assets() -> tuple[list[dict[str, Any]], list[dict[str prom_assets, host_relationships = await _collect_prometheus_targets() assets.extend(prom_assets) relationships.extend(host_relationships) + reconciled_prefixes.update(("host/", "prometheus_target/")) except Exception as e: logger.warning("collect_prometheus_targets_failed", error=str(e)) + collector_errors.append("prometheus_targets") # 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) + if database_assets: + reconciled_prefixes.update( + ("postgres/database/", "postgres/relation/") + ) except Exception as e: logger.warning("collect_database_catalog_failed", error=str(e)) + collector_errors.append("database_catalog") - return assets, relationships + return ( + assets, + relationships, + tuple(sorted(reconciled_prefixes)), + tuple(collector_errors), + ) def _is_valid_ipv4(s: str) -> bool: @@ -543,14 +591,10 @@ async def _collect_prometheus_targets() -> tuple[list[dict[str, Any]], list[dict seen_hosts: set[str] = set() url = f"{settings.PROMETHEUS_URL.rstrip('/')}/api/v1/targets" - try: - async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client: - resp = await client.get(url, params={"state": "active"}) - resp.raise_for_status() - data = resp.json() - except Exception as e: - logger.warning("prometheus_targets_fetch_failed", error=str(e)) - return assets, relationships + async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client: + resp = await client.get(url, params={"state": "active"}) + resp.raise_for_status() + data = resp.json() for t in (data.get("data", {}) or {}).get("activeTargets", []) or []: labels = t.get("labels", {}) or {} @@ -744,11 +788,11 @@ async def _start_discovery_run( triggered_by: str, scope: list[str], scan_depth: str, -) -> str | None: +) -> str: """ 寫 asset_discovery_run header (status='running'), 回傳 run_id。 - 失敗 → log warning + 回 None,主流程靜默跳過本次 scan。 + 失敗會向排程器傳播並進入 bounded backoff，不可靜默跳過整輪掃描。 """ try: from sqlalchemy import text as _sql @@ -783,51 +827,97 @@ async def _start_discovery_run( }, ) run_id = row.scalar() - return str(run_id) if run_id else None + if not run_id: + raise RuntimeError("asset_discovery_run_missing_run_id") + return str(run_id) except Exception as e: logger.warning("asset_discovery_run_start_failed", error=str(e)) - return None + raise -async def _finish_discovery_run( +async def _finalize_discovery_run( run_id: str, + triggered_by: str, status: str, total_assets: int, new_assets: int, modified_assets: int, + disappeared_assets: int, duration_ms: int, error: str | None, + scope: list[str], + scan_depth: str, ) -> None: - try: - from sqlalchemy import text as _sql + """Atomically persist the discovery terminal and its AOL receipt.""" + from sqlalchemy import text as _sql - from src.db.base import get_db_context + from src.db.base import get_db_context - async with get_db_context(_PROJECT_ID) as db: - await db.execute( - _sql(""" - UPDATE asset_discovery_run - SET status = :st, - ended_at = NOW(), - total_assets = :total, - new_assets = :new, - modified_assets = :mod, - duration_ms = :dur, - error = :err - WHERE run_id = CAST(:rid AS uuid) - """), - { - "st": status, - "total": total_assets, - "new": new_assets, - "mod": modified_assets, - "dur": duration_ms, - "err": error, - "rid": run_id, - }, - ) - except Exception as e: - logger.warning("asset_discovery_run_finish_failed", run_id=run_id, error=str(e)) + input_payload = { + "run_id": run_id, + "triggered_by": triggered_by, + "scope": scope, + "scan_depth": scan_depth, + } + output_payload = { + "run_id": run_id, + "total_assets": total_assets, + "new_assets": new_assets, + "modified_assets": modified_assets, + "disappeared_assets": disappeared_assets, + } + + async with get_db_context(_PROJECT_ID) as db: + await db.execute( + _sql(""" + UPDATE asset_discovery_run + SET status = :st, + ended_at = NOW(), + total_assets = :total, + new_assets = :new, + modified_assets = :mod, + disappeared_assets = :gone, + duration_ms = :dur, + error = :err + WHERE run_id = CAST(:rid AS uuid) + """), + { + "st": status, + "total": total_assets, + "new": new_assets, + "mod": modified_assets, + "gone": disappeared_assets, + "dur": duration_ms, + "err": error, + "rid": run_id, + }, + ) + await db.execute( + _sql(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, + input, output, run_id, duration_ms, error, tags + ) VALUES ( + 'asset_discovered', + 'asset_scanner', + :st, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:rid AS uuid), + :dur, :err, + :tags + ) + """), + { + "st": "success" if status == "success" else "failed", + "input": _json.dumps(input_payload, ensure_ascii=False), + "output": _json.dumps(output_payload, ensure_ascii=False), + "rid": run_id, + "dur": duration_ms, + "err": (error or "")[:2000] if error else None, + "tags": ["asset_scanner", "discovery", *scope], + }, + ) async def _upsert_assets( @@ -891,10 +981,62 @@ async def _upsert_assets( modified_count += 1 except Exception as e: logger.exception("asset_upsert_failed", run_id=run_id, error=str(e)) + raise return new_count, modified_count +async def _deprecate_missing_assets( + assets: list[dict[str, Any]], + reconciled_prefixes: tuple[str, ...], + run_id: str, +) -> int: + """Deprecate assets missing from collectors that succeeded in this run.""" + prefixes = sorted({prefix for prefix in reconciled_prefixes if prefix}) + if not prefixes: + return 0 + + seen_keys = sorted( + { + str(asset["asset_key"]) + for asset in assets + if any(str(asset["asset_key"]).startswith(prefix) for prefix in prefixes) + } + ) + + 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( + """ + UPDATE asset_inventory + SET lifecycle_state = 'deprecated', + decommissioned_at = NOW(), + updated_at = NOW() + WHERE lifecycle_state = 'active' + AND asset_key LIKE ANY(CAST(:patterns AS text[])) + AND last_seen_at < ( + SELECT started_at + FROM asset_discovery_run + WHERE run_id = CAST(:rid AS uuid) + ) + AND NOT (asset_key = ANY(CAST(:seen_keys AS text[]))) + """ + ), + { + "patterns": [f"{prefix}%" for prefix in prefixes], + "seen_keys": seen_keys, + "rid": run_id, + }, + ) + + rowcount = getattr(result, "rowcount", 0) or 0 + return max(int(rowcount), 0) + + async def _upsert_relationships(relationships: list[dict[str, str]]) -> int: """ UPSERT asset_relationship (from_asset_id/to_asset_id/relationship_type 為 UNIQUE). @@ -939,8 +1081,10 @@ async def _upsert_relationships(relationships: list[dict[str, str]]) -> int: except Exception as e: logger.debug("relationship_upsert_skipped", from_key=rel["from_key"], to_key=rel["to_key"], error=str(e)) + raise except Exception as e: logger.warning("relationship_upsert_failed", error=str(e)) + raise return written @@ -950,99 +1094,33 @@ async def _write_coverage_snapshots(run_id: str) -> None: 後續 service (rule_catalog / playbook_extractor / km_writer) 會 UPDATE 對應維度。 """ - try: - from sqlalchemy import text as _sql + from sqlalchemy import text as _sql - from src.db.base import get_db_context + from src.db.base import get_db_context - async with get_db_context(_PROJECT_ID) as db: - # 一次性 INSERT: 取所有 active asset × 7 dimensions - await db.execute( - _sql(""" - INSERT INTO asset_coverage_snapshot ( - run_id, asset_id, dimension, coverage_status, - evidence, detected_by - ) - SELECT - CAST(:rid AS uuid), - ai.asset_id, - d.dimension, - 'unknown' AS coverage_status, - '{}'::jsonb, - 'asset_scanner' AS detected_by - FROM asset_inventory ai - CROSS JOIN ( - VALUES ('auto_monitoring'),('auto_alerting'),('auto_rule_creation'), - ('auto_rule_matching'),('auto_playbook'),('auto_remediation'), - ('auto_km_creation') - ) AS d(dimension) - WHERE ai.lifecycle_state = 'active' - ON CONFLICT (run_id, asset_id, dimension) DO NOTHING - """), - {"rid": run_id}, - ) - except Exception as e: - logger.warning("asset_coverage_write_failed", run_id=run_id, error=str(e)) - - -async def _log_aol_asset_discovered( - run_id: str, - triggered_by: str, - total: int, - new: int, - modified: int, - duration_ms: int, - status: str, - error: str | None, - scope: list[str], - scan_depth: str, -) -> None: - """寫 automation_operation_log(asset_discovered)。失敗只 log 不阻塞。""" - try: - from sqlalchemy import text as _sql - - from src.db.base import get_db_context - - aol_status = "success" if status == "success" else "failed" - input_payload = { - "run_id": run_id, - "triggered_by": triggered_by, - "scope": scope, - "scan_depth": scan_depth, - } - output_payload = { - "run_id": run_id, - "total_assets": total, - "new_assets": new, - "modified_assets": modified, - } - - async with get_db_context(_PROJECT_ID) as db: - await db.execute( - _sql(""" - INSERT INTO automation_operation_log ( - operation_type, actor, status, - input, output, run_id, duration_ms, error, tags - ) VALUES ( - 'asset_discovered', - 'asset_scanner', - :st, - CAST(:input AS jsonb), - CAST(:output AS jsonb), - CAST(:rid AS uuid), - :dur, :err, - :tags - ) - """), - { - "st": aol_status, - "input": _json.dumps(input_payload, ensure_ascii=False), - "output": _json.dumps(output_payload, ensure_ascii=False), - "rid": run_id, - "dur": duration_ms, - "err": (error or "")[:2000] if error else None, - "tags": ["asset_scanner", "discovery", *scope], - }, - ) - except Exception as e: - logger.warning("asset_scanner_aol_write_failed", run_id=run_id, error=str(e)) + async with get_db_context(_PROJECT_ID) as db: + # 一次性 INSERT: 取所有 active asset × 7 dimensions + await db.execute( + _sql(""" + INSERT INTO asset_coverage_snapshot ( + run_id, asset_id, dimension, coverage_status, + evidence, detected_by + ) + SELECT + CAST(:rid AS uuid), + ai.asset_id, + d.dimension, + 'unknown' AS coverage_status, + '{}'::jsonb, + 'asset_scanner' AS detected_by + FROM asset_inventory ai + CROSS JOIN ( + VALUES ('auto_monitoring'),('auto_alerting'),('auto_rule_creation'), + ('auto_rule_matching'),('auto_playbook'),('auto_remediation'), + ('auto_km_creation') + ) AS d(dimension) + WHERE ai.lifecycle_state = 'active' + ON CONFLICT (run_id, asset_id, dimension) DO NOTHING + """), + {"rid": run_id}, + ) diff --git a/apps/api/tests/test_asset_inventory_database_type_migration.py b/apps/api/tests/test_asset_inventory_database_type_migration.py new file mode 100644 index 000000000..662999986 --- /dev/null +++ b/apps/api/tests/test_asset_inventory_database_type_migration.py @@ -0,0 +1,75 @@ +import textwrap +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[3] +FORWARD = ( + ROOT + / "apps" + / "api" + / "migrations" + / "asset_inventory_database_types_2026-07-14.sql" +) +ROLLBACK = ( + ROOT + / "apps" + / "api" + / "migrations" + / "asset_inventory_database_types_2026-07-14_down.sql" +) +CD_WORKFLOW = ROOT / ".gitea" / "workflows" / "cd.yaml" + + +def test_forward_migration_repairs_type_constraint_without_row_deletion() -> None: + sql = FORWARD.read_text(encoding="utf-8") + upper = sql.upper() + + assert "SET lock_timeout = '5s'" in sql + assert "SET statement_timeout = '30s'" in sql + assert "'database','table'" in sql + assert "DROP CONSTRAINT IF EXISTS asset_inventory_type_valid" in sql + assert "NOT VALID" in sql + assert "VALIDATE CONSTRAINT asset_inventory_type_valid" in sql + assert "gitea_cd_asset_schema_reconciler" in sql + assert '"work_item_id":"AIA-P0-006"' in sql + assert "automation_operation_log" in sql + assert "DELETE FROM" not in upper + assert "DROP TABLE" not in upper + assert "TRUNCATE" not in upper + + +def test_rollback_refuses_to_remove_types_while_assets_depend_on_them() -> None: + sql = ROLLBACK.read_text(encoding="utf-8") + upper = sql.upper() + + assert "asset_type IN ('database', 'table')" in sql + assert "IF dependent_rows > 0" in sql + assert "RAISE EXCEPTION" in sql + assert "DELETE FROM" not in upper + assert "TRUNCATE" not in upper + + +def test_cd_applies_and_independently_verifies_the_bounded_migration() -> None: + workflow = CD_WORKFLOW.read_text(encoding="utf-8") + step = workflow.split( + "- name: Reconcile Asset Inventory Type Constraint", 1 + )[1].split("# 2026-03-31 ogt: 移除中間通知", 1)[0] + script = step.split("run: |", 1)[1] + + assert "MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}" in step + assert "DATABASE_URL: ${{ secrets.DATABASE_URL }}" in step + assert "${{ secrets." not in script + assert "asset_inventory_database_types_2026-07-14.sql" in script + assert "asset_inventory_type_migration_applied=" in script + assert "database_type_allowed=" in script + assert "table_type_allowed=" in script + assert "constraint_type_set_exact=" in script + assert "constraint_validated=" in script + assert "durable_receipt_present=" in script + assert "asset_inventory_type_constraint_verifier_failed" in script + + +def test_cd_embedded_migration_python_compiles() -> None: + workflow = CD_WORKFLOW.read_text(encoding="utf-8") + source = workflow.split("python - <<'PY'", 1)[1].split(" PY", 1)[0] + + compile(textwrap.dedent(source), "", "exec") diff --git a/apps/api/tests/test_asset_scanner_job.py b/apps/api/tests/test_asset_scanner_job.py index be1367fd5..7182b859c 100644 --- a/apps/api/tests/test_asset_scanner_job.py +++ b/apps/api/tests/test_asset_scanner_job.py @@ -6,6 +6,8 @@ from src.jobs import asset_scanner_job class _Result: + rowcount = 1 + def scalar(self): return "00000000-0000-0000-0000-000000000001" @@ -41,15 +43,6 @@ def test_background_scanner_scopes_every_database_operation(monkeypatch) -> None ) assert run_id == "00000000-0000-0000-0000-000000000001" - await asset_scanner_job._finish_discovery_run( - run_id=run_id, - status="success", - total_assets=1, - new_assets=1, - modified_assets=0, - duration_ms=10, - error=None, - ) inserted, modified = await asset_scanner_job._upsert_assets( [ { @@ -66,6 +59,17 @@ def test_background_scanner_scopes_every_database_operation(monkeypatch) -> None ) assert (inserted, modified) == (1, 0) + disappeared = await asset_scanner_job._deprecate_missing_assets( + [ + { + "asset_key": "host/test", + } + ], + ("host/",), + run_id, + ) + assert disappeared == 1 + written = await asset_scanner_job._upsert_relationships( [ { @@ -78,14 +82,15 @@ def test_background_scanner_scopes_every_database_operation(monkeypatch) -> None assert written == 1 await asset_scanner_job._write_coverage_snapshots(run_id) - await asset_scanner_job._log_aol_asset_discovered( + await asset_scanner_job._finalize_discovery_run( run_id=run_id, triggered_by="cron", - total=1, - new=1, - modified=0, - duration_ms=10, status="success", + total_assets=1, + new_assets=1, + modified_assets=0, + disappeared_assets=1, + duration_ms=10, error=None, scope=["k8s", "prometheus", "database_catalog"], scan_depth="shallow", @@ -96,6 +101,183 @@ def test_background_scanner_scopes_every_database_operation(monkeypatch) -> None assert requested_projects == ["awoooi"] * 6 +def test_asset_upsert_propagates_database_failure(monkeypatch) -> None: + class FailingDatabase: + async def execute(self, statement, parameters=None): + raise RuntimeError("constraint rejected") + + class FailingContext: + async def __aenter__(self): + return FailingDatabase() + + async def __aexit__(self, exc_type, exc, traceback): + return False + + monkeypatch.setattr( + "src.db.base.get_db_context", + lambda project_id=None: FailingContext(), + ) + + async def exercise() -> None: + try: + await asset_scanner_job._upsert_assets( + [ + { + "asset_key": "postgres/database/awoooi", + "asset_type": "database", + "host": None, + "namespace": None, + "name": "awoooi", + "metadata": {}, + "tags": ["source:test"], + } + ], + "00000000-0000-0000-0000-000000000001", + ) + except RuntimeError as exc: + assert str(exc) == "constraint rejected" + else: + raise AssertionError("asset upsert failure must propagate") + + asyncio.run(exercise()) + + +def test_deprecate_missing_assets_is_bounded_to_successful_prefixes(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(), + ) + + disappeared = asyncio.run( + asset_scanner_job._deprecate_missing_assets( + [ + {"asset_key": "k8s/pod/prod/current"}, + {"asset_key": "prometheus_target/node/192.168.0.110:9100"}, + ], + ("k8s/pod/",), + "00000000-0000-0000-0000-000000000001", + ) + ) + + assert disappeared == 1 + assert "last_seen_at <" in str(captured["sql"]) + assert captured["parameters"] == { + "patterns": ["k8s/pod/%"], + "seen_keys": ["k8s/pod/prod/current"], + "rid": "00000000-0000-0000-0000-000000000001", + } + + +def test_runtime_collection_reports_failed_sources_without_reconciling_them( + monkeypatch, +) -> None: + async def fake_kubectl(resource: str, all_namespaces: bool = True): + if resource == "nodes": + raise RuntimeError("nodes unavailable") + return {"items": []} + + async def fake_prometheus(): + return [], [] + + async def fake_database_catalog(): + return [], [] + + monkeypatch.setattr(asset_scanner_job, "_fetch_kubectl_json", fake_kubectl) + monkeypatch.setattr( + asset_scanner_job, + "_collect_prometheus_targets", + fake_prometheus, + ) + monkeypatch.setattr( + asset_scanner_job, + "_collect_database_catalog_assets", + fake_database_catalog, + ) + + assets, relationships, prefixes, errors = asyncio.run( + asset_scanner_job._collect_runtime_assets() + ) + + assert assets == [] + assert relationships == [] + assert "k8s/node/" not in prefixes + assert "k8s/pod/" in prefixes + assert "prometheus_target/" in prefixes + assert errors == ("k8s_nodes",) + + +def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -> None: + terminal: dict[str, object] = {} + + async def fake_start(*args, **kwargs): + return "00000000-0000-0000-0000-000000000001" + + async def fake_collect(): + return [], [], (), ("prometheus_targets",) + + async def fake_upsert(*args, **kwargs): + return 0, 0 + + async def fake_deprecate(*args, **kwargs): + return 0 + + async def fake_relationships(*args, **kwargs): + return 0 + + async def fake_coverage(*args, **kwargs): + return None + + async def fake_finalize(**kwargs): + terminal.update(kwargs) + + monkeypatch.setattr(asset_scanner_job, "_start_discovery_run", fake_start) + monkeypatch.setattr(asset_scanner_job, "_collect_runtime_assets", fake_collect) + monkeypatch.setattr(asset_scanner_job, "_upsert_assets", fake_upsert) + monkeypatch.setattr( + asset_scanner_job, + "_deprecate_missing_assets", + fake_deprecate, + ) + monkeypatch.setattr( + asset_scanner_job, + "_upsert_relationships", + fake_relationships, + ) + monkeypatch.setattr( + asset_scanner_job, + "_write_coverage_snapshots", + fake_coverage, + ) + monkeypatch.setattr( + asset_scanner_job, + "_finalize_discovery_run", + fake_finalize, + ) + + run_id = asyncio.run(asset_scanner_job.scan_once()) + + assert run_id == "00000000-0000-0000-0000-000000000001" + assert terminal["status"] == "failed" + assert terminal["error"] == ( + "RuntimeError: asset_collector_failures=prometheus_targets" + ) + + def test_database_catalog_collector_reads_metadata_only(monkeypatch) -> None: requested_projects: list[str | None] = [] diff --git a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx index 82b528967..f40cf35aa 100644 --- a/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx +++ b/apps/web/src/components/iwooos/security-asset-control-plane-cockpit.tsx @@ -43,6 +43,13 @@ type Copy = { audit: string; aiTrace: string; sources: string; + latestScan: string; + freshness: string; + runAssets: string; + retired: string; + stale: string; + success: string; + failed: string; sameRunMissing: string; unavailable: string; refresh: string; @@ -68,6 +75,13 @@ const COPY: Record<"zh" | "en", Copy> = { audit: "稽核軌跡", aiTrace: "AI 軌跡", sources: "資料源", + latestScan: "最新盤點", + freshness: "新鮮度", + runAssets: "本輪資產", + retired: "本輪退役", + stale: "逾期資產", + success: "成功", + failed: "失敗", sameRunMissing: "尚無同 run 閉環證據", unavailable: "Live readback 無法取得", refresh: "重新整理資安控制平面", @@ -91,6 +105,13 @@ const COPY: Record<"zh" | "en", Copy> = { audit: "Audit trail", aiTrace: "AI trace", sources: "Sources", + latestScan: "Latest scan", + freshness: "Freshness", + runAssets: "Run assets", + retired: "Retired", + stale: "Stale assets", + success: "Success", + failed: "Failed", sameRunMissing: "No same-run closure evidence", unavailable: "Live readback unavailable", refresh: "Refresh security control plane", @@ -107,6 +128,12 @@ function stageCount(stage: SecurityControlPlaneStage) { return stage.evidence_count_24h ?? stage.receipt_count_24h ?? 0; } +function freshnessLabel(ageSeconds: number | null) { + if (ageSeconds === null) return "--"; + if (ageSeconds < 60) return `${Math.max(0, Math.round(ageSeconds))}s`; + return `${Math.round(ageSeconds / 60)}m`; +} + async function fetchControlPlane(): Promise { const base = getRuntimeApiBaseUrl(); const path = "/api/v1/iwooos/security-asset-control-plane"; @@ -146,6 +173,49 @@ function MetricCell({ ); } +function TruthCell({ + label, + value, + icon: Icon, + tone = "neutral", +}: { + label: string; + value: string | number; + icon: typeof ShieldCheck; + tone?: "neutral" | "healthy" | "critical"; +}) { + return ( +
+
+ ); +} + function StageStrip({ title, stages, @@ -286,6 +356,47 @@ export function SecurityAssetControlPlaneCockpit({ /> +
+ + + + 0 ? "healthy" : "neutral"} + /> + 0 ? "critical" : "healthy"} + /> +
+ {payload ? (
diff --git a/apps/web/src/lib/__tests__/security-asset-control-plane.test.ts b/apps/web/src/lib/__tests__/security-asset-control-plane.test.ts index 12782cf42..c2dfcf491 100644 --- a/apps/web/src/lib/__tests__/security-asset-control-plane.test.ts +++ b/apps/web/src/lib/__tests__/security-asset-control-plane.test.ts @@ -46,10 +46,17 @@ function payload(): SecurityAssetControlPlanePayload { }, discovery: { latest_status: "success", + latest_scan_depth: "shallow", + latest_started_at: "2026-07-11T03:49:00+00:00", + latest_ended_at: "2026-07-11T03:50:00+00:00", latest_success_at: "2026-07-11T03:50:00+00:00", age_seconds: 600, freshness_slo_seconds: 7200, fresh: true, + latest_total_assets: 24, + new_assets: 2, + modified_assets: 22, + disappeared_assets: 1, }, asset_scopes: [], nist_csf_functions: [], diff --git a/apps/web/src/lib/security-asset-control-plane.ts b/apps/web/src/lib/security-asset-control-plane.ts index 4a13304a4..dbaada8fe 100644 --- a/apps/web/src/lib/security-asset-control-plane.ts +++ b/apps/web/src/lib/security-asset-control-plane.ts @@ -77,10 +77,17 @@ export type SecurityAssetControlPlanePayload = { summary: SecurityControlPlaneSummary; discovery: { latest_status: string; + latest_scan_depth: string; + latest_started_at: string | null; + latest_ended_at: string | null; latest_success_at: string | null; age_seconds: number | null; freshness_slo_seconds: number; fresh: boolean; + latest_total_assets: number; + new_assets: number; + modified_assets: number; + disappeared_assets: number; }; asset_scopes: SecurityControlPlaneScope[]; nist_csf_functions: SecurityControlPlaneFunction[];