fix(security): repair canonical asset integrity

apps/api/migrations/asset_inventory_canonical_integrity_2026-07-14.sql

@@ -0,0 +1,215 @@
+-- Repair duplicate canonical asset keys without deleting historical rows.
+-- Safety: bounded quarantine, transactional index rebuild, independent verifier.
+
+SET lock_timeout = '10s';
+SET statement_timeout = '90s';
+SET enable_indexscan = 'off';
+SET enable_indexonlyscan = 'off';
+SET enable_bitmapscan = 'off';
+
+CREATE TEMP TABLE asset_inventory_integrity_repair_map
+ON COMMIT DROP
+AS
+WITH ranked_assets AS (
+    SELECT
+        asset_id,
+        asset_key AS original_asset_key,
+        FIRST_VALUE(asset_id) OVER (
+            PARTITION BY asset_key
+            ORDER BY first_seen_at ASC, asset_id ASC
+        ) AS canonical_asset_id,
+        ROW_NUMBER() OVER (
+            PARTITION BY asset_key
+            ORDER BY first_seen_at ASC, asset_id ASC
+        ) AS duplicate_rank
+    FROM asset_inventory
+)
+SELECT
+    asset_id,
+    original_asset_key,
+    canonical_asset_id
+FROM ranked_assets
+WHERE duplicate_rank > 1;
+
+DO $$
+DECLARE
+    repair_row_count BIGINT;
+BEGIN
+    SELECT COUNT(*) INTO repair_row_count
+    FROM asset_inventory_integrity_repair_map;
+
+    IF repair_row_count > 100 THEN
+        RAISE EXCEPTION
+            'asset_inventory_integrity_repair_bound_exceeded: % rows',
+            repair_row_count;
+    END IF;
+END
+$$;
+
+UPDATE asset_inventory AS duplicate_asset
+SET
+    asset_key = duplicate_asset.asset_key
+        || '/__integrity_quarantine__/'
+        || duplicate_asset.asset_id::TEXT,
+    lifecycle_state = 'deprecated',
+    decommissioned_at = COALESCE(duplicate_asset.decommissioned_at, NOW()),
+    metadata = COALESCE(duplicate_asset.metadata, '{}'::JSONB)
+        || JSONB_BUILD_OBJECT(
+            'integrity_state', 'quarantined_duplicate',
+            'integrity_canonical_asset_id', repair.canonical_asset_id,
+            'integrity_repair', 'asset_inventory_canonical_integrity_2026-07-14'
+        ),
+    tags = CASE
+        WHEN 'integrity:quarantined_duplicate' = ANY(duplicate_asset.tags)
+            THEN duplicate_asset.tags
+        ELSE ARRAY_APPEND(
+            duplicate_asset.tags,
+            'integrity:quarantined_duplicate'
+        )
+    END,
+    updated_at = NOW()
+FROM asset_inventory_integrity_repair_map AS repair
+WHERE duplicate_asset.asset_id = repair.asset_id;
+
+-- Rebuild the unique index after every formerly unindexed duplicate has a
+-- distinct quarantine key. The outer CD transaction rolls back on any error.
+REINDEX INDEX asset_inventory_asset_key_key;
+
+DO $$
+DECLARE
+    duplicate_group_count BIGINT;
+    repaired_row_count BIGINT;
+    expected_repair_count BIGINT;
+    index_healthy BOOLEAN;
+BEGIN
+    SELECT COUNT(*) INTO duplicate_group_count
+    FROM (
+        SELECT asset_key
+        FROM asset_inventory
+        GROUP BY asset_key
+        HAVING COUNT(*) > 1
+    ) AS duplicate_groups;
+
+    SELECT COUNT(*) INTO expected_repair_count
+    FROM asset_inventory_integrity_repair_map;
+
+    SELECT COUNT(*) INTO repaired_row_count
+    FROM asset_inventory AS repaired
+    JOIN asset_inventory_integrity_repair_map AS repair
+      ON repair.asset_id = repaired.asset_id
+    WHERE repaired.lifecycle_state = 'deprecated'
+      AND repaired.asset_key = repair.original_asset_key
+          || '/__integrity_quarantine__/'
+          || repaired.asset_id::TEXT
+      AND repaired.metadata ->> 'integrity_state' = 'quarantined_duplicate';
+
+    SELECT
+        index_row.indisvalid
+        AND index_row.indisready
+        AND index_row.indislive
+        AND index_row.indisunique
+    INTO index_healthy
+    FROM pg_index AS index_row
+    WHERE index_row.indexrelid = 'asset_inventory_asset_key_key'::REGCLASS;
+
+    IF duplicate_group_count <> 0 THEN
+        RAISE EXCEPTION
+            'asset_inventory_duplicate_groups_remain: %',
+            duplicate_group_count;
+    END IF;
+
+    IF repaired_row_count <> expected_repair_count THEN
+        RAISE EXCEPTION
+            'asset_inventory_quarantine_verifier_failed: expected %, got %',
+            expected_repair_count,
+            repaired_row_count;
+    END IF;
+
+    IF index_healthy IS DISTINCT FROM TRUE THEN
+        RAISE EXCEPTION 'asset_inventory_unique_index_not_healthy';
+    END IF;
+END
+$$;
+
+-- Exercise the exact conflict-update path that failed before the repair while
+-- preserving every application value.
+WITH probe AS (
+    SELECT
+        asset_key,
+        asset_type,
+        host,
+        namespace,
+        name,
+        metadata,
+        tags,
+        environment,
+        lifecycle_state,
+        first_seen_at,
+        last_seen_at
+    FROM asset_inventory
+    ORDER BY asset_id
+    LIMIT 1
+)
+INSERT INTO asset_inventory (
+    asset_key,
+    asset_type,
+    host,
+    namespace,
+    name,
+    metadata,
+    tags,
+    environment,
+    lifecycle_state,
+    first_seen_at,
+    last_seen_at
+)
+SELECT
+    asset_key,
+    asset_type,
+    host,
+    namespace,
+    name,
+    metadata,
+    tags,
+    environment,
+    lifecycle_state,
+    first_seen_at,
+    last_seen_at
+FROM probe
+ON CONFLICT (asset_key) DO UPDATE
+SET
+    last_seen_at = asset_inventory.last_seen_at,
+    updated_at = asset_inventory.updated_at;
+
+INSERT INTO automation_operation_log (
+    operation_type,
+    actor,
+    status,
+    input,
+    output,
+    tags
+)
+SELECT
+    'asset_discovered',
+    'gitea_cd_asset_integrity_reconciler',
+    'success',
+    '{"work_item_id":"AIA-P0-006","migration":"asset_inventory_canonical_integrity_2026-07-14"}'::JSONB,
+    JSONB_BUILD_OBJECT(
+        'duplicate_group_count_before', COUNT(DISTINCT original_asset_key),
+        'rows_quarantined', COUNT(*),
+        'duplicate_group_count_after', 0,
+        'index_rebuilt', TRUE,
+        'canonical_upsert_verified', TRUE,
+        'data_rows_deleted', 0
+    ),
+    ARRAY[
+        'asset_control_plane',
+        'integrity_repair',
+        'controlled_apply',
+        'post_verified',
+        'no_delete'
+    ]
+FROM asset_inventory_integrity_repair_map;
+
+COMMENT ON INDEX asset_inventory_asset_key_key IS
+    'Canonical asset identity unique index; transactionally rebuilt and verified by Gitea CD.';

apps/api/src/jobs/asset_scanner_job.py

@@ -114,6 +114,8 @@ async def scan_once(
     error_msg: str | None = None
 
     try:
+        await _assert_asset_key_integrity()
+
         # 2026-04-19 v3 擴充: 多資源類型掃描 + relationship 提取
         # 資源類型: pods (container), deployments/statefulsets/daemonsets (k8s_workload),
         #          services (k8s_resource), nodes (host), configmaps (k8s_resource)
@@ -917,6 +919,61 @@ async def _finalize_discovery_run(
                 "err": (error or "")[:2000] if error else None,
                 "tags": ["asset_scanner", "discovery", *scope],
             },
+    )
+
+
+async def _assert_asset_key_integrity() -> None:
+    """Fail closed before discovery when canonical asset identity is corrupt."""
+    from sqlalchemy import text as _sql
+
+    from src.db.base import get_db_context
+
+    async with get_db_context(_PROJECT_ID) as db:
+        # A corrupt-but-valid unique index can make the planner incorrectly
+        # assume the heap is unique. Inspect the heap before trusting it.
+        await db.execute(_sql("SET LOCAL enable_indexscan = 'off'"))
+        await db.execute(_sql("SET LOCAL enable_indexonlyscan = 'off'"))
+        await db.execute(_sql("SET LOCAL enable_bitmapscan = 'off'"))
+        row = (
+            await db.execute(
+                _sql(
+                    """
+                    SELECT
+                        (
+                            SELECT COUNT(*)
+                            FROM (
+                                SELECT asset_key
+                                FROM asset_inventory
+                                GROUP BY asset_key
+                                HAVING COUNT(*) > 1
+                            ) AS duplicate_groups
+                        ) AS duplicate_group_count,
+                        COALESCE((
+                            SELECT
+                                index_row.indisvalid
+                                AND index_row.indisready
+                                AND index_row.indislive
+                                AND index_row.indisunique
+                            FROM pg_constraint AS constraint_row
+                            JOIN pg_index AS index_row
+                              ON index_row.indexrelid = constraint_row.conindid
+                            WHERE constraint_row.conrelid =
+                                'asset_inventory'::regclass
+                              AND constraint_row.conname =
+                                'asset_inventory_asset_key_key'
+                        ), FALSE) AS unique_index_healthy
+                    """
+                )
+            )
+        ).one()
+
+    duplicate_group_count = int(row[0] or 0)
+    unique_index_healthy = bool(row[1])
+    if duplicate_group_count or not unique_index_healthy:
+        raise RuntimeError(
+            "asset_inventory_key_integrity_failed "
+            f"duplicate_groups={duplicate_group_count} "
+            f"unique_index_healthy={str(unique_index_healthy).lower()}"
         )
 
 

apps/api/tests/test_asset_inventory_database_type_migration.py

@@ -52,7 +52,7 @@ 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]
+    )[1].split("- name: Repair Asset Inventory Canonical Integrity", 1)[0]
     script = step.split("run: |", 1)[1]
 
     assert "MIGRATION_DATABASE_URL: ${{ secrets.MIGRATION_DATABASE_URL }}" in step

apps/api/tests/test_asset_inventory_integrity_migration.py

@@ -0,0 +1,71 @@
+import textwrap
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+MIGRATION = (
+    ROOT
+    / "apps"
+    / "api"
+    / "migrations"
+    / "asset_inventory_canonical_integrity_2026-07-14.sql"
+)
+CD_WORKFLOW = ROOT / ".gitea" / "workflows" / "cd.yaml"
+
+
+def test_migration_quarantines_duplicates_without_deleting_history() -> None:
+    sql = MIGRATION.read_text(encoding="utf-8")
+    upper = sql.upper()
+
+    assert "SET lock_timeout = '10s'" in sql
+    assert "SET statement_timeout = '90s'" in sql
+    assert "SET enable_indexscan = 'off'" in sql
+    assert "SET enable_indexonlyscan = 'off'" in sql
+    assert "SET enable_bitmapscan = 'off'" in sql
+    assert "ROW_NUMBER() OVER" in sql
+    assert "repair_row_count > 100" in sql
+    assert "__integrity_quarantine__" in sql
+    assert "lifecycle_state = 'deprecated'" in sql
+    assert "REINDEX INDEX asset_inventory_asset_key_key" in sql
+    assert "HAVING COUNT(*) > 1" in sql
+    assert "ON CONFLICT (asset_key) DO UPDATE" in sql
+    assert "gitea_cd_asset_integrity_reconciler" in sql
+    assert '"work_item_id":"AIA-P0-006"' in sql
+    assert "'data_rows_deleted', 0" in sql
+    assert "DELETE FROM" not in upper
+    assert "DROP TABLE" not in upper
+    assert "TRUNCATE" not in upper
+
+
+def test_cd_runs_bounded_integrity_repair_and_independent_verifier() -> None:
+    workflow = CD_WORKFLOW.read_text(encoding="utf-8")
+    step = workflow.split(
+        "- name: Repair Asset Inventory Canonical Integrity", 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 "--network bridge" in script
+    assert "--network host" not in script
+    assert "asset_inventory_canonical_integrity_2026-07-14.sql" in script
+    assert "SET enable_indexscan TO off" in script
+    assert "SET enable_indexonlyscan TO off" in script
+    assert "SET enable_bitmapscan TO off" in script
+    assert "asset_inventory_integrity_migration_applied=" in script
+    assert "duplicate_group_count=" in script
+    assert "quarantined_row_count=" in script
+    assert "unique_index_healthy=" in script
+    assert "durable_integrity_receipt_present=" in script
+    assert "canonical_integrity_verified=" in script
+    assert "asset_inventory_canonical_integrity_verifier_failed" in script
+
+
+def test_cd_embedded_integrity_migration_python_compiles() -> None:
+    workflow = CD_WORKFLOW.read_text(encoding="utf-8")
+    step = workflow.split(
+        "- name: Repair Asset Inventory Canonical Integrity", 1
+    )[1]
+    source = step.split("python - <<'PY'", 1)[1].split("          PY", 1)[0]
+
+    compile(textwrap.dedent(source), "<asset-inventory-integrity-migration>", "exec")

apps/api/tests/test_asset_scanner_job.py

@@ -142,6 +142,81 @@ def test_asset_upsert_propagates_database_failure(monkeypatch) -> None:
     asyncio.run(exercise())
 
 
+def test_asset_key_integrity_preflight_accepts_unique_inventory(monkeypatch) -> None:
+    requested_projects: list[str | None] = []
+    statements: list[str] = []
+
+    class IntegrityResult:
+        def one(self):
+            return 0, True
+
+    class IntegrityDatabase:
+        async def execute(self, statement, parameters=None):
+            statements.append(str(statement))
+            return IntegrityResult()
+
+    class IntegrityContext:
+        async def __aenter__(self):
+            return IntegrityDatabase()
+
+        async def __aexit__(self, exc_type, exc, traceback):
+            return False
+
+    def fake_db_context(project_id=None):
+        requested_projects.append(project_id)
+        return IntegrityContext()
+
+    monkeypatch.setattr("src.db.base.get_db_context", fake_db_context)
+
+    asyncio.run(asset_scanner_job._assert_asset_key_integrity())
+
+    assert requested_projects == ["awoooi"]
+    assert any("enable_indexscan" in statement for statement in statements)
+    assert any("enable_indexonlyscan" in statement for statement in statements)
+    assert any("enable_bitmapscan" in statement for statement in statements)
+    assert any("HAVING COUNT(*) > 1" in statement for statement in statements)
+    assert any(
+        "asset_inventory_asset_key_key" in statement for statement in statements
+    )
+
+
+def test_asset_key_integrity_preflight_fails_closed_on_duplicate_key(
+    monkeypatch,
+) -> None:
+    class IntegrityResult:
+        def one(self):
+            return 1, True
+
+    class IntegrityDatabase:
+        async def execute(self, statement, parameters=None):
+            return IntegrityResult()
+
+    class IntegrityContext:
+        async def __aenter__(self):
+            return IntegrityDatabase()
+
+        async def __aexit__(self, exc_type, exc, traceback):
+            return False
+
+    monkeypatch.setattr(
+        "src.db.base.get_db_context",
+        lambda project_id=None: IntegrityContext(),
+    )
+
+    async def exercise() -> None:
+        try:
+            await asset_scanner_job._assert_asset_key_integrity()
+        except RuntimeError as exc:
+            assert str(exc) == (
+                "asset_inventory_key_integrity_failed "
+                "duplicate_groups=1 unique_index_healthy=true"
+            )
+        else:
+            raise AssertionError("duplicate canonical keys must fail closed")
+
+    asyncio.run(exercise())
+
+
 def test_deprecate_missing_assets_is_bounded_to_successful_prefixes(monkeypatch) -> None:
     captured: dict[str, object] = {}
 
@@ -230,6 +305,9 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -
     async def fake_collect():
         return [], [], (), ("prometheus_targets",)
 
+    async def fake_integrity():
+        return None
+
     async def fake_upsert(*args, **kwargs):
         return 0, 0
 
@@ -246,6 +324,11 @@ def test_scan_terminal_is_failed_when_any_collector_is_incomplete(monkeypatch) -
         terminal.update(kwargs)
 
     monkeypatch.setattr(asset_scanner_job, "_start_discovery_run", fake_start)
+    monkeypatch.setattr(
+        asset_scanner_job,
+        "_assert_asset_key_integrity",
+        fake_integrity,
+    )
     monkeypatch.setattr(asset_scanner_job, "_collect_runtime_assets", fake_collect)
     monkeypatch.setattr(asset_scanner_job, "_upsert_assets", fake_upsert)
     monkeypatch.setattr(