from __future__ import annotations from dataclasses import replace from datetime import UTC, datetime, timedelta import unittest from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from govoplan_core.core.encryption import ( CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX, ContentProtectionEnvelope, KeyLifecycleRequest, KeyMaterialDescriptor, KeyRotationRequest, KeyVaultCreateRequest, ProtectionMigrationRequest, ProtectionRegistrationRequest, RecoveryApprovalRequest, RecoveryRequest, ) from govoplan_core.core.identity_trust import ( CAPABILITY_IDENTITY_TRUST_ASSURANCE, AssuranceDecision, ) from govoplan_core.db.base import utcnow from govoplan_encryption.backend.db.models import ( ContentProtectionRecord, EncryptionKeyOperation, EncryptionKeyVersion, EncryptionVault, ProtectionMigration, RecoveryApproval, RecoveryCeremony, ) from govoplan_encryption.backend.service import EncryptionError, SqlEncryptionService NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC) class Principal: def __init__(self, account_id: str, tenant_id: str = "tenant-1") -> None: self.account_id = account_id self.tenant_id = tenant_id class FakeAssurance: def verify_assurance(self, _session, _principal, *, request): return AssuranceDecision( allowed=True, reason="test evidence accepted", assurance_level="high", evidence_ref=request.evidence_ref, verified_at=request.evaluated_at, expires_at=request.evaluated_at + timedelta(minutes=5), provenance={"provider_id": "test-assurance"}, ) class FakeKeyProvider: def __init__(self) -> None: self.algorithms: dict[str, str] = {} self.fail_once: set[str] = set() self.calls: list[str] = [] def provision_key(self, _session, _principal, *, request): self.calls.append(request.idempotency_key) if request.idempotency_key in self.fail_once: self.fail_once.remove(request.idempotency_key) raise RuntimeError("provider diagnostics must not be persisted") reference = f"provider://{request.vault_id}/v{request.key_version}" self.algorithms[reference] = request.algorithm_suite return self._descriptor(reference, "active") def revoke_key( self, _session, _principal, *, provider_key_ref, reason, idempotency_key, ): del reason self.calls.append(idempotency_key) return self._descriptor(provider_key_ref, "revoked") def schedule_key_destruction( self, _session, _principal, *, provider_key_ref, effective_at, idempotency_key, ): del effective_at self.calls.append(idempotency_key) return self._descriptor(provider_key_ref, "destruction_scheduled") def key_status(self, _session, *, provider_key_ref): if provider_key_ref not in self.algorithms: return None return self._descriptor(provider_key_ref, "active") def _descriptor(self, reference: str, state: str) -> KeyMaterialDescriptor: return KeyMaterialDescriptor( provider_id="test", provider_key_ref=reference, algorithm_suite=self.algorithms.get(reference, "AES-256-GCM"), state=state, # type: ignore[arg-type] created_at=NOW, exportable=False, provenance={"operation": "test-vector"}, ) class Registry: def __init__(self, provider: FakeKeyProvider) -> None: self.values = { f"{CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX}test": provider, CAPABILITY_IDENTITY_TRUST_ASSURANCE: FakeAssurance(), } def has_capability(self, name: str) -> bool: return name in self.values def capability(self, name: str): return self.values[name] class EncryptionTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") self.tables = [ EncryptionVault.__table__, EncryptionKeyVersion.__table__, EncryptionKeyOperation.__table__, ContentProtectionRecord.__table__, ProtectionMigration.__table__, RecoveryCeremony.__table__, RecoveryApproval.__table__, ] for table in self.tables: table.create(self.engine) self.session = Session(self.engine) self.provider = FakeKeyProvider() self.service = SqlEncryptionService(Registry(self.provider)) self.principal = Principal("account-1") def tearDown(self) -> None: self.session.close() self.engine.dispose() def create_vault(self, **changes): values = { "tenant_id": "tenant-1", "vault_id": "vault-1", "name": "Records", "provider_id": "test", "purpose": "feature-content", "algorithm_suite": "AES-256-GCM", "scope_type": "tenant", "scope_id": "tenant-1", "policy_ref": "policy:v1", "idempotency_key": "vault-create-1", "recovery_quorum": 2, } values.update(changes) return self.service.create_vault( self.session, self.principal, request=KeyVaultCreateRequest(**values), ) def envelope(self, owner_module: str, suffix: str) -> ContentProtectionEnvelope: return ContentProtectionEnvelope( envelope_id=f"envelope-{suffix}", tenant_id="tenant-1", owner_module=owner_module, resource_type=f"{owner_module}.object", resource_id=f"resource-{suffix}", profile_kind="server_envelope", profile_id="profile-1", provider_id="test", vault_id="vault-1", key_version=1, algorithm_suite="AES-256-GCM", ciphertext_ref=f"object://ciphertext/{suffix}", ciphertext_digest=f"sha256:cipher-{suffix}", authenticated_context_digest=f"sha256:context-{suffix}", state="active", created_at=NOW, metadata={"content_type": "application/octet-stream"}, ) def test_vault_replay_rotation_revocation_and_tenant_isolation(self) -> None: first = self.create_vault() replay = self.create_vault() self.assertEqual("active", first.state) self.assertEqual( first.current_key.provider_key_ref, replay.current_key.provider_key_ref ) self.assertEqual(["vault-create-1"], self.provider.calls) rotated = self.service.rotate_key( self.session, self.principal, request=KeyRotationRequest( tenant_id="tenant-1", vault_id="vault-1", expected_revision=1, reason="scheduled rotation", policy_decision_ref="policy:rotate", assurance_evidence_ref="assurance:rotate", idempotency_key="rotate-1", ), ) self.assertEqual(2, rotated.current_key.version) self.assertEqual(2, rotated.revision) revoked = self.service.revoke_key( self.session, self.principal, request=KeyLifecycleRequest( tenant_id="tenant-1", vault_id="vault-1", key_version=2, expected_revision=2, reason="compromise response", policy_decision_ref="policy:revoke", assurance_evidence_ref="assurance:revoke", idempotency_key="revoke-1", ), ) self.assertEqual("revoked", revoked.current_key.state) self.assertEqual("unavailable", revoked.state) with self.assertRaisesRegex(EncryptionError, "Cross-tenant"): self.service.get_envelope( self.session, Principal("account-1", "tenant-2"), tenant_id="tenant-1", envelope_id="missing", ) def test_provider_failure_is_persisted_and_idempotently_reconciled(self) -> None: self.provider.fail_once.add("vault-failure") value = self.create_vault( vault_id="vault-failure", idempotency_key="vault-failure", ) self.assertEqual("outcome_unknown", value.state) operation = self.session.scalar( select(EncryptionKeyOperation).where( EncryptionKeyOperation.idempotency_key == "vault-failure" ) ) self.assertEqual("provider_runtimeerror", operation.error_code) self.assertNotIn("diagnostics", operation.error_code) reconciled = self.service.reconcile_vault( self.session, self.principal, tenant_id="tenant-1", vault_id="vault-failure", ) self.assertEqual("active", reconciled.state) self.assertEqual(2, self.provider.calls.count("vault-failure")) def test_files_and_postbox_envelopes_migrate_before_disable(self) -> None: self.create_vault() for suffix, owner in (("files", "files"), ("postbox", "postbox")): value = self.service.register_envelope( self.session, self.principal, request=ProtectionRegistrationRequest( envelope=self.envelope(owner, suffix), idempotency_key=f"register-{suffix}", policy_decision_ref="policy:protect", ), ) self.assertEqual(owner, value.owner_module) blocked = self.service.assess_disable(self.session, tenant_id="tenant-1") self.assertFalse(blocked.allowed) self.assertEqual(2, blocked.unresolved_count) for suffix in ("files", "postbox"): migration = self.service.request_migration( self.session, self.principal, request=ProtectionMigrationRequest( tenant_id="tenant-1", envelope_id=f"envelope-{suffix}", target_provider_id="test", target_vault_id="vault-1", target_key_version=1, target_algorithm_suite="AES-256-GCM", mode="destroy", policy_decision_ref="policy:destroy", assurance_evidence_ref="assurance:destroy", idempotency_key=f"destroy-{suffix}", ), ) result = self.service.record_migration_outcome( self.session, self.principal, migration_id=migration.migration_id, state="succeeded", evidence_refs=(f"provider-evidence:{suffix}",), ) self.assertEqual("succeeded", result.state) allowed = self.service.assess_disable(self.session, tenant_id="tenant-1") self.assertTrue(allowed.allowed) self.assertEqual(0, allowed.unresolved_count) def test_recovery_requires_distinct_high_assurance_quorum(self) -> None: self.create_vault() recovery = self.service.request_recovery( self.session, self.principal, request=RecoveryRequest( tenant_id="tenant-1", vault_id="vault-1", reason="custody recovery", requested_scope="vault-status-and-rewrap", policy_decision_ref="policy:recovery", assurance_evidence_ref="assurance:requester", idempotency_key="recovery-1", expires_at=utcnow() + timedelta(hours=1), ), ) with self.assertRaisesRegex(EncryptionError, "own recovery"): self.service.decide_recovery( self.session, self.principal, request=RecoveryApprovalRequest( tenant_id="tenant-1", recovery_id=recovery.recovery_id, decision="approve", reason="approved", assurance_evidence_ref="assurance:requester", expected_revision=1, idempotency_key="requester-approval", ), ) first = self.service.decide_recovery( self.session, Principal("custodian-1"), request=RecoveryApprovalRequest( tenant_id="tenant-1", recovery_id=recovery.recovery_id, decision="approve", reason="evidence reviewed", assurance_evidence_ref="assurance:custodian-1", expected_revision=1, idempotency_key="approval-1", ), ) self.assertEqual("pending", first.state) second = self.service.decide_recovery( self.session, Principal("custodian-2"), request=RecoveryApprovalRequest( tenant_id="tenant-1", recovery_id=recovery.recovery_id, decision="approve", reason="evidence reviewed", assurance_evidence_ref="assurance:custodian-2", expected_revision=2, idempotency_key="approval-2", ), ) self.assertEqual("approved", second.state) self.assertEqual(2, second.approvals) self.assertFalse(second.provenance["key_material_released"]) def test_contract_rejects_secret_bearing_envelope_metadata(self) -> None: with self.assertRaisesRegex(ValueError, "secret material"): replace( self.envelope("files", "bad"), metadata={"password": "must-not-cross-boundary"}, ) if __name__ == "__main__": unittest.main()