diff --git a/apps/api/src/services/backup_restore_alertmanager_ingress.py b/apps/api/src/services/backup_restore_alertmanager_ingress.py index 3d8c17243..c2880d17f 100644 --- a/apps/api/src/services/backup_restore_alertmanager_ingress.py +++ b/apps/api/src/services/backup_restore_alertmanager_ingress.py @@ -40,6 +40,9 @@ BACKUP_RESTORE_AGENT99_ROUTE_ID = "agent99:backup_health:BackupCheck" BACKUP_RESTORE_DISPATCH_OWNER = "alertmanager_raw_signal" BACKUP_RESTORE_OWNER_SCHEMA_VERSION = "backup_restore_ingress_owner_v1" APPROVAL_FINGERPRINT_MAX_LENGTH = 64 +APPROVAL_FINGERPRINT_HASH_DOMAIN = ( + "awoooi.backup_restore.approval_fingerprint.v2" +) BACKUP_RESTORE_PROHIBITED_ACTIONS = ( "run_backup", "run_restore", @@ -57,12 +60,76 @@ def _stable_digest(*parts: object) -> str: def _approval_storage_fingerprint(source_fingerprint: str) -> str: - """Fit a source identity into approval_records.fingerprint VARCHAR(64).""" + """Return the domain-separated durable identity for every source value. + + Hashing only values longer than the PostgreSQL column made a long source + collide with a raw 64-character source equal to that source's SHA-256. + Hash every value in the lane with a domain tag so the two source domains + remain distinct while still fitting ``approval_records.fingerprint``. + """ normalized = str(source_fingerprint or "").strip() - if len(normalized) <= APPROVAL_FINGERPRINT_MAX_LENGTH: - return normalized - return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + payload = f"{APPROVAL_FINGERPRINT_HASH_DOMAIN}\x1f{normalized}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _approval_fingerprint_lookup_candidates( + source_fingerprint: str, +) -> tuple[str, ...]: + """Return v2 first, followed by the one compatible legacy encoding. + + The v1 implementation stored short values verbatim and unscoped-hashed + long values. Both representations remain read-compatible during the + rollout, but new rows always use the domain-separated v2 fingerprint. + """ + + normalized = str(source_fingerprint or "").strip() + canonical = _approval_storage_fingerprint(normalized) + legacy = ( + normalized + if len(normalized) <= APPROVAL_FINGERPRINT_MAX_LENGTH + else hashlib.sha256(normalized.encode("utf-8")).hexdigest() + ) + return tuple(dict.fromkeys((canonical, legacy))) + + +def _approval_matches_source_fingerprint( + approval: Any, + source_fingerprint: str, +) -> bool: + """Verify a lookup row before reusing its canonical Incident binding.""" + + metadata = getattr(approval, "metadata", None) + if not isinstance(metadata, dict): + return False + persisted_source = str(metadata.get("source_fingerprint") or "").strip() + return persisted_source == str(source_fingerprint or "").strip() + + +async def _find_compatible_source_approval( + approval_service: Any, + *, + source_fingerprint: str, +) -> Any | None: + """Read v2/v1 identities without accepting a cross-domain collision.""" + + for storage_fingerprint in _approval_fingerprint_lookup_candidates( + source_fingerprint + ): + recent = await approval_service.find_by_fingerprint( + fingerprint=storage_fingerprint, + debounce_minutes=30, + ) + if recent is None: + continue + if _approval_matches_source_fingerprint(recent, source_fingerprint): + return recent + logger.warning( + "backup_restore_approval_fingerprint_identity_mismatch", + storage_fingerprint=storage_fingerprint, + approval_id=_value(getattr(recent, "id", "")), + ) + return None def build_backup_restore_ingress_owner( @@ -252,9 +319,9 @@ async def process_backup_restore_alertmanager_signal( created_owner = approval is None legacy_incident_id = "" if approval is None: - recent = await approval_service.find_by_fingerprint( - fingerprint=approval_storage_fingerprint, - debounce_minutes=30, + recent = await _find_compatible_source_approval( + approval_service, + source_fingerprint=source_fingerprint, ) legacy_incident_id = _value(getattr(recent, "incident_id", "")) request = _backup_restore_approval_request( diff --git a/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py b/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py index b356584be..72d83838d 100644 --- a/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py +++ b/apps/api/tests/integration/test_backup_restore_prod_db_regressions.py @@ -147,35 +147,57 @@ async def test_backfill_run_inserts_satisfy_no_default_cost_column() -> None: @pytest.mark.asyncio -async def test_long_occurrence_fingerprint_fits_real_pg_varchar() -> None: +async def test_domain_separated_fingerprints_avoid_cross_domain_pg_collision() -> None: engine = create_async_engine(_explicit_local_test_database_url(), echo=False) source_fingerprint = "legacy-backup-occurrence:" + ("a" * 64) - stored_fingerprint = ( + deliberate_raw_fingerprint = backfill._sha256(source_fingerprint) + stored_long_fingerprint = ( backup_restore_alertmanager_ingress._approval_storage_fingerprint( source_fingerprint ) ) + stored_raw_fingerprint = ( + backup_restore_alertmanager_ingress._approval_storage_fingerprint( + deliberate_raw_fingerprint + ) + ) + assert stored_long_fingerprint != stored_raw_fingerprint try: async with engine.connect() as connection: transaction = await connection.begin() try: await connection.execute(text(""" CREATE TEMP TABLE approval_records ( - fingerprint VARCHAR(64) NOT NULL + source_identity TEXT NOT NULL, + fingerprint VARCHAR(64) NOT NULL UNIQUE ) ON COMMIT DROP """)) await connection.execute( text(""" - INSERT INTO approval_records (fingerprint) - VALUES (:fingerprint) + INSERT INTO approval_records ( + source_identity, fingerprint + ) VALUES + ('long', :stored_long_fingerprint), + ('deliberate_raw', :stored_raw_fingerprint) """), - {"fingerprint": stored_fingerprint}, + { + "stored_long_fingerprint": stored_long_fingerprint, + "stored_raw_fingerprint": stored_raw_fingerprint, + }, ) - persisted = await connection.scalar( - text("SELECT fingerprint FROM approval_records") - ) - assert persisted == stored_fingerprint - assert len(persisted) == 64 + persisted = ( + await connection.execute(text(""" + SELECT source_identity, fingerprint + FROM approval_records + ORDER BY source_identity + """)) + ).mappings().all() + assert len(persisted) == 2 + assert {row["fingerprint"] for row in persisted} == { + stored_long_fingerprint, + stored_raw_fingerprint, + } + assert all(len(row["fingerprint"]) == 64 for row in persisted) finally: await transaction.rollback() finally: diff --git a/apps/api/tests/test_backup_restore_alertmanager_ingress.py b/apps/api/tests/test_backup_restore_alertmanager_ingress.py index c51b03c21..8ea3f1f74 100644 --- a/apps/api/tests/test_backup_restore_alertmanager_ingress.py +++ b/apps/api/tests/test_backup_restore_alertmanager_ingress.py @@ -85,15 +85,85 @@ def test_backup_ingress_owner_is_stable_per_source_occurrence() -> None: assert first["read_only"] is True -def test_backup_approval_fingerprint_fits_durable_varchar_boundary() -> None: +def test_backup_approval_fingerprint_is_domain_separated_and_bounded() -> None: short = "source-fingerprint" long = "legacy-backup-occurrence:" + ("a" * 64) + legacy_long_digest = hashlib.sha256(long.encode("utf-8")).hexdigest() - assert ingress._approval_storage_fingerprint(short) == short - stored = ingress._approval_storage_fingerprint(long) - assert stored == hashlib.sha256(long.encode("utf-8")).hexdigest() - assert len(stored) == ingress.APPROVAL_FINGERPRINT_MAX_LENGTH - assert ingress._approval_storage_fingerprint(long) == stored + stored_short = ingress._approval_storage_fingerprint(short) + stored_long = ingress._approval_storage_fingerprint(long) + stored_deliberate_raw = ingress._approval_storage_fingerprint( + legacy_long_digest + ) + + assert stored_short != short + assert stored_long != legacy_long_digest + assert stored_long != stored_deliberate_raw + assert len(stored_short) == ingress.APPROVAL_FINGERPRINT_MAX_LENGTH + assert len(stored_long) == ingress.APPROVAL_FINGERPRINT_MAX_LENGTH + assert ingress._approval_storage_fingerprint(long) == stored_long + assert ingress._approval_fingerprint_lookup_candidates(short) == ( + stored_short, + short, + ) + assert ingress._approval_fingerprint_lookup_candidates(long) == ( + stored_long, + legacy_long_digest, + ) + + +@pytest.mark.asyncio +async def test_backup_approval_legacy_lookup_rejects_deliberate_collision() -> None: + long = "legacy-backup-occurrence:" + ("a" * 64) + deliberate_raw = hashlib.sha256(long.encode("utf-8")).hexdigest() + canonical, legacy = ingress._approval_fingerprint_lookup_candidates(long) + assert legacy == deliberate_raw + + wrong_source = SimpleNamespace( + id=UUID("11111111-1111-4111-8111-111111111111"), + incident_id="INC-WRONG-SOURCE", + metadata={"source_fingerprint": deliberate_raw}, + ) + calls: list[str] = [] + + async def find_by_fingerprint(*, fingerprint: str, debounce_minutes: int): + calls.append(fingerprint) + assert debounce_minutes == 30 + return wrong_source if fingerprint == legacy else None + + result = await ingress._find_compatible_source_approval( + SimpleNamespace(find_by_fingerprint=find_by_fingerprint), + source_fingerprint=long, + ) + + assert result is None + assert calls == [canonical, legacy] + + +@pytest.mark.asyncio +async def test_backup_approval_legacy_89_char_lookup_remains_compatible() -> None: + source = "legacy-backup-occurrence:" + ("a" * 64) + canonical, legacy = ingress._approval_fingerprint_lookup_candidates(source) + existing = SimpleNamespace( + id=UUID("22222222-2222-4222-8222-222222222222"), + incident_id="INC-LEGACY-89-CHAR", + metadata={"source_fingerprint": source}, + ) + calls: list[str] = [] + + async def find_by_fingerprint(*, fingerprint: str, debounce_minutes: int): + calls.append(fingerprint) + assert debounce_minutes == 30 + return existing if fingerprint == legacy else None + + result = await ingress._find_compatible_source_approval( + SimpleNamespace(find_by_fingerprint=find_by_fingerprint), + source_fingerprint=source, + ) + + assert result is existing + assert result.incident_id == "INC-LEGACY-89-CHAR" + assert calls == [canonical, legacy] @pytest.mark.asyncio @@ -179,10 +249,10 @@ class _FakeApprovalService: async def find_by_fingerprint(self, **_kwargs): self.calls.append("legacy_owner_read") - expected = ingress._approval_storage_fingerprint( + expected = ingress._approval_fingerprint_lookup_candidates( str(_signal_kwargs()["source_fingerprint"]) ) - assert _kwargs["fingerprint"] == expected + assert _kwargs["fingerprint"] in expected return None async def create_approval_with_fingerprint(self, *, request, fingerprint): @@ -280,6 +350,14 @@ async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads **_signal_kwargs() ) + assert approval_service.approval is not None + approval_service.approval.fingerprint = hashlib.sha256( + str(_signal_kwargs()["source_fingerprint"]).encode("utf-8") + ).hexdigest() + replay = await ingress.process_backup_restore_alertmanager_signal( + **_signal_kwargs() + ) + assert calls.index("owner_create") < calls.index("incident_create") assert calls.index("incident_create") < calls.index("incident_bind") assert calls.index("incident_bind") < calls.index("incident_canonical_readback") @@ -297,6 +375,13 @@ async def test_backup_ingress_binds_canonical_incident_then_dispatches_and_reads assert result["runtime_execution_authorized"] is False assert result["runtime_closure_verified"] is False assert result["production_write_performed"] is False + assert replay["owner_created"] is False + assert calls.count("owner_create") == 1 + assert calls.count("incident_create") == 1 + assert calls.count("owner_recurrence") == 1 + assert approval_service.approval.fingerprint == hashlib.sha256( + str(_signal_kwargs()["source_fingerprint"]).encode("utf-8") + ).hexdigest() assert approval_service.created_request.risk_level.value == "low" assert approval_service.created_request.metadata["telegram_role"] == ( "receipt_projection_only"