463 lines
17 KiB
Python
463 lines
17 KiB
Python
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_operator_queries_are_bounded_tenant_scoped_and_secret_free(self) -> None:
|
|
self.create_vault()
|
|
self.service.register_envelope(
|
|
self.session,
|
|
self.principal,
|
|
request=ProtectionRegistrationRequest(
|
|
envelope=self.envelope("files", "operator-query"),
|
|
idempotency_key="register-operator-query",
|
|
policy_decision_ref="policy:protect",
|
|
),
|
|
)
|
|
migration = self.service.request_migration(
|
|
self.session,
|
|
self.principal,
|
|
request=ProtectionMigrationRequest(
|
|
tenant_id="tenant-1",
|
|
envelope_id="envelope-operator-query",
|
|
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="migration-operator-query",
|
|
),
|
|
)
|
|
recovery = self.service.request_recovery(
|
|
self.session,
|
|
self.principal,
|
|
request=RecoveryRequest(
|
|
tenant_id="tenant-1",
|
|
vault_id="vault-1",
|
|
reason="operator query fixture",
|
|
requested_scope="vault-status",
|
|
policy_decision_ref="policy:recovery",
|
|
assurance_evidence_ref="assurance:requester",
|
|
idempotency_key="recovery-operator-query",
|
|
expires_at=utcnow() + timedelta(hours=1),
|
|
),
|
|
)
|
|
|
|
vaults = self.service.list_vaults(
|
|
self.session, self.principal, tenant_id="tenant-1", limit=1000
|
|
)
|
|
envelopes = self.service.list_envelopes(
|
|
self.session, self.principal, tenant_id="tenant-1", limit=1000
|
|
)
|
|
migrations = self.service.list_migrations(
|
|
self.session, self.principal, tenant_id="tenant-1", limit=1000
|
|
)
|
|
recoveries = self.service.list_recoveries(
|
|
self.session, self.principal, tenant_id="tenant-1", limit=1000
|
|
)
|
|
|
|
self.assertEqual(["vault-1"], [item.vault_id for item in vaults])
|
|
self.assertEqual(
|
|
["envelope-operator-query"],
|
|
[item.envelope_id for item in envelopes],
|
|
)
|
|
self.assertEqual(migration.migration_id, migrations[0].migration_id)
|
|
self.assertEqual(recovery.recovery_id, recoveries[0].recovery_id)
|
|
self.assertEqual("operator query fixture", recoveries[0].reason)
|
|
self.assertFalse(hasattr(vaults[0], "provider_key_ref"))
|
|
self.assertFalse(hasattr(envelopes[0], "wrapped_key_refs"))
|
|
self.assertFalse(hasattr(envelopes[0], "ciphertext_ref"))
|
|
with self.assertRaisesRegex(EncryptionError, "Cross-tenant"):
|
|
self.service.list_vaults(
|
|
self.session,
|
|
Principal("account-1", "tenant-2"),
|
|
tenant_id="tenant-1",
|
|
)
|
|
|
|
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()
|