diff --git a/README.md b/README.md index d5d9685..938f024 100644 --- a/README.md +++ b/README.md @@ -4,45 +4,39 @@ **Repository type:** module (platform). -`govoplan-encryption` is the optional platform boundary for managed key vaults, -content encryption, end-to-end encryption profiles, key rotation, recovery, and -cryptographic deletion assurance. +`govoplan-encryption` is the optional provider-neutral boundary for governed key +vaults, versioned content-protection envelopes, key lifecycle operations, +recovery ceremonies, and disable/uninstall assurance. -The module is deliberately scaffolded before cryptographic implementation. It -announces stable integration surfaces and policy boundaries without presenting -placeholder cryptography as production-ready protection. +## Implemented -## Boundary +- opaque provider capability contracts; raw key material never crosses them; +- tenant/scope-bound vault and key-version metadata; +- idempotent provision, rotation, revocation, destruction scheduling, and + outcome-unknown reconciliation; +- versioned ciphertext envelopes with authenticated-context and migration + provenance; +- resumable, evidence-backed rewrap, re-encryption, decrypt, export, and destroy + state transitions; +- recent high-assurance, distinct-custodian quorum recovery authorization; +- typed APIs, audit-safe events, Alembic migration, and uninstall blocking; +- Files and Postbox boundary fixtures. -The module will own: +## Deliberately Not Bundled -- vault and key hierarchy metadata -- envelope-key wrapping and rotation workflows -- content-protection profiles, including E2EE profiles -- device and recipient key registration -- recovery ceremonies, assurance evidence, and key custody -- disable and uninstall preflight for protected content +The module contains no cipher implementation, raw key store, development master +key, KMS/HSM driver, client E2EE protocol, or provider credentials. A deployment +must install a separately reviewed provider capability. Until a concrete profile +passes conformance, backup/restore, key-loss, interoperability, and security +review, GovOPlaN must not claim that content is encrypted or end-to-end +encrypted. -Feature modules continue to own their content, authorization, retention, and -normal ownership records. They integrate through optional encryption -capabilities and store only stable protection references and ciphertext. +Feature modules continue to own content, authorization, retention, and resource +ownership. Access approval, resource ownership, Identity Trust, and key custody +are separate decisions. -## Disable Rule - -Disabling or uninstalling this module must never make protected content silently -unreadable. Every protected object must first be decrypted, rewrapped to another -available provider, exported with explicit key custody, or deliberately -destroyed under an authorized cryptographic-deletion process. The eventual -uninstall guard will block while unresolved protection references remain. - -See [docs/ENCRYPTION_BOUNDARY.md](docs/ENCRYPTION_BOUNDARY.md) for the initial -threat, ownership, recovery, and lifecycle decisions. - -## Current State - -The repository currently contains a discoverable manifest, permission and role -metadata, versioned interface announcements, documentation, and focused tests. -It has no routes, persistence, WebUI, key generation, or encryption runtime yet. +See [docs/ENCRYPTION_BOUNDARY.md](docs/ENCRYPTION_BOUNDARY.md) for the threat +model, profile consequences, algorithms, recovery, and disable semantics. ## Verification diff --git a/docs/ENCRYPTION_BOUNDARY.md b/docs/ENCRYPTION_BOUNDARY.md index d7a2def..3ca2225 100644 --- a/docs/ENCRYPTION_BOUNDARY.md +++ b/docs/ENCRYPTION_BOUNDARY.md @@ -1,50 +1,212 @@ -# Encryption Boundary +# Encryption Boundary and Threat Model -## Purpose +## Status -`govoplan-encryption` is an optional cryptographic service module. It supplies -key and content-protection capabilities to modules that choose to use them. It -does not own feature data, normal authorization, retention, or business -ownership. +`govoplan-encryption` is the optional governance and lifecycle boundary for +cryptographic protection. It persists opaque provider references, versioned +content-protection envelopes, idempotent operation state, recovery decisions, +and disable-preflight evidence. It does not ship a cipher, a raw key store, a +KMS/HSM driver, or a client E2EE protocol. -## Initial Invariants +Enabling the module is therefore not an encryption or E2EE claim. A deployment +may make such a claim only for a named protection profile whose concrete +provider, clients, algorithms, backup procedure, and failure tests have passed +the profile's conformance and security review. -1. Resource ownership and cryptographic key custody are independent. -2. Administrative ownership recovery does not grant encryption keys. -3. Key recovery is a separate, high-assurance, auditable ceremony. -4. Every ciphertext carries a stable protection profile, key version, algorithm - suite, and authenticated context reference. -5. Rotation and rewrapping are resumable and idempotent. -6. Protected content cannot silently fall back to plaintext. -7. Provider loss, disabled modules, unavailable devices, and recovery delay are - visible states, not generic read failures. -8. Disabling or uninstalling is blocked while unresolved protected content - remains. +## Ownership Boundary -## Planned Capability Families +Encryption owns: -- **Key vault:** create and resolve key hierarchies without exposing raw key - material to feature modules. -- **Content protection:** authenticated envelope encryption and decryption for - authorized feature-owned payloads. -- **E2EE profiles:** recipient/device key envelopes where the server cannot - decrypt normal content. -- **Rotation:** rewrap keys and migrate algorithm suites with bounded progress - and evidence. -- **Recovery ceremony:** quorum, delay, recent authentication, custody evidence, - and explicit scope. -- **Disable preflight:** inventory protection references and produce a blocking - migration plan. +- vault and opaque provider-key reference metadata; +- key-version and provider-operation lifecycle state; +- versioned protection envelopes and migration evidence; +- high-assurance, quorum-based recovery authorization; +- disable and uninstall preflight. -## Decisions Deferred Until Threat Modeling +Feature modules still own domain content, access policy, retention, legal-hold +meaning, and normal resource ownership. Identity Trust owns public device keys, +key epochs, and recent-assurance evidence. Access decides whether an account may +perform a resource action. A positive Access decision does not release a key; +an ownership transfer does not transfer cryptographic custody. -- cryptographic libraries and external KMS/HSM providers -- server-managed versus client-managed profile suites -- per-object, per-vault, and per-recipient data-key granularity -- device enrollment and revocation protocol -- external-recipient key discovery and trust -- recovery quorum defaults for each assurance profile -- ciphertext portability and offline export format +No capability in this module accepts or returns plaintext key material. -These decisions require explicit threat models and interoperability fixtures. -They must not be inferred from ordinary resource-ownership workflows. +## Assets and Threat Actors + +Protected assets include content plaintext, data-encryption keys, wrapping keys, +recipient/device private keys, authenticated context, recovery authority, and +the integrity and provenance of lifecycle decisions. + +The threat model includes: + +- an unauthorized user or tenant attempting cross-tenant access; +- a compromised account, API key, worker, or application node; +- a malicious or mistaken tenant/platform administrator; +- a compromised storage service, database backup, queue, log collector, or + observability system; +- a lost or revoked device and a former function holder; +- a malicious or unavailable key provider; +- replay, race, partial failure, rollback, and outcome-unknown provider calls; +- ciphertext substitution, context confusion, algorithm downgrade, and mixed + migration state; +- backup restoration without matching provider keys, or key restoration without + matching metadata. + +The initial model does not claim protection against endpoint compromise while a +legitimate client has plaintext open, coercion of all configured recovery +custodians, traffic analysis, or denial of service by a provider that controls +required keys. + +## Observable Metadata + +Unless an independently reviewed profile says otherwise, the server and its +database can observe tenant, owner module, resource type and stable id, profile, +vault and key version, algorithm suite, ciphertext location and digest, +authenticated-context digest, wrapped-key references, timestamps, migration +state, policy references, recovery scope, and actor/evidence references. + +Titles, recipient identities, content length, access patterns, and relationship +graphs may also leak through the owning module or storage provider even when the +payload is encrypted. Product UI and documentation must not imply that payload +encryption hides this metadata. + +## Deployment Profiles + +### Server-side envelope encryption + +The application or a deployment KMS/HSM can unwrap data keys after normal +authorization. This protects backups and storage media but not plaintext from a +compromised authorized application tier. Provider credentials and key policy are +deployment secrets, not envelope metadata. Legal hold can preserve ciphertext +and key availability under separately authorized policy. + +### Tenant-held key provider + +The tenant controls an external KMS/HSM or equivalent provider. GovOPlaN holds +only references and invokes the provider under tenant policy. The provider may +allow server-side processing, but tenant revocation or provider outage can make +content unavailable. Backup recovery requires both GovOPlaN state and the +tenant's provider recovery procedure. + +### End-to-end encryption + +Only authorized clients possess usable recipient/device private keys; ordinary +servers cannot decrypt payloads. Search, preview, server-side Dataflow, +virus/content inspection, reporting, legal-hold export, and recovery degrade or +become impossible unless the selected profile deliberately adds an authorized +recipient or escrow mechanism. Such a mechanism changes the threat model and +must be displayed as part of the profile. + +The current code can label and track an `end_to_end` envelope, but it does not +implement the client protocol. That label alone is never evidence of E2EE. + +## Profile Consequences + +| Concern | Server envelope | Tenant-held | End-to-end | +| --- | --- | --- | --- | +| Normal server processing | Available after authorization | Provider-policy dependent | Unavailable unless a client performs it | +| Provider outage | Decryption unavailable | Decryption unavailable | Existing enrolled clients may remain usable | +| Platform admin can obtain plaintext | Possible under governed server policy | Tenant/provider dependent | Not from ordinary server custody | +| Legal hold | Preserve ciphertext and governed keys | Requires tenant/provider cooperation | Ciphertext only unless the profile has a hold recipient | +| Export | Governed plaintext or portable encrypted export | Provider cooperation required | Client-produced or recipient-key export | +| Recovery | Provider recovery plus GovOPlaN quorum | Tenant provider recovery plus quorum | Device/recipient recovery defined by the client profile | +| Ownership transfer | Separate reauthorization/rewrap | Separate reauthorization/rewrap | Explicit recipient change and re-encryption | +| Revocation | Stops future provider use; cannot erase prior plaintext | Provider enforced | New key epoch; cannot erase prior client plaintext | + +## Algorithms, Libraries, and Agility + +Concrete providers must declare a versioned suite id and pass conformance tests. +The initial allowlist for reviewed profiles is: + +- AES-256-GCM through a deployment KMS/HSM or a maintained OpenSSL-backed + cryptographic library; +- XChaCha20-Poly1305 through a maintained libsodium implementation where the + provider/profile requires extended nonces; +- RFC 9180 HPKE suites for a future reviewed recipient-envelope protocol. + +Selection from this list is not automatic approval. A provider profile must +also define nonce generation and uniqueness, associated-data construction, key +separation, limits, error handling, and test vectors. Home-grown primitives, +unauthenticated encryption, AES-ECB, bare AES-CBC, static/reused nonces, raw RSA +encryption, SHA-1 integrity, provider-supplied secret values in logs, and silent +algorithm fallback are prohibited. + +Every envelope fixes its profile, algorithm suite, vault, key version, +ciphertext digest, and authenticated-context digest. Changing any of these is a +new, evidence-backed migration. Readers never infer an algorithm from current +defaults. Old readers fail closed on unknown suites or contract versions. + +## Durable Operation Semantics + +Key-provider operations use a stable idempotency key. GovOPlaN records an +operation as `prepared`, `succeeded`, or `outcome_unknown`. If a provider call +raises or a process dies around the external side effect, absence of an error is +never treated as success. Reconciliation repeats the exact idempotent provider +request and records only bounded error class codes. + +Content migrations are explicit records. Rewrap and re-encryption succeed only +when an exact target envelope references the authorized source and migration. +Decrypt, export, and destruction require evidence references. Until an outcome +is known, the source remains `migration_pending`; no implicit plaintext fallback +is allowed. + +All runtime state is in the shared database or an external provider/object +store reference. Application and worker nodes require no local encryption state, +so multiple nodes can retry the same operation safely when the provider honors +the idempotency contract. + +## Recovery Ceremony + +Recovery requires current Identity Trust assurance at the `high` level, a +policy-decision reference, expiry, distinct custodian accounts, and the vault's +configured quorum. The requester cannot approve their own ceremony; one +rejection rejects the current ceremony. Decisions are optimistic-concurrency +protected and replay safe. + +An `approved` ceremony authorizes a later provider-specific recovery action. It +does not return a key, change resource ownership, or prove that recovery has +executed. A production provider must add separately auditable execution and +post-recovery rotation/rewrap evidence. + +## Disable and Uninstall + +Disabling or uninstalling never deletes keys or silently makes ciphertext look +like ordinary unreadable content. Preflight inventories all envelopes. It blocks +while any envelope is active, pending, migrating, unavailable, or otherwise +unresolved. An envelope is locally terminal only after it is: + +- migrated to another registered protection envelope; +- decrypted by an authorized workflow; +- explicitly exported with documented custody; or +- cryptographically destroyed with evidence. + +A migrated target remains an active protected envelope and therefore still +blocks removal of the module/provider that owns it. Destructive table retirement +is permitted only after this proof. + +## Backup, Restore, and Key Loss + +A recoverable backup set consists of the shared database, ciphertext/object +storage, provider configuration references, provider-side key backups or HSM +replication, and audit evidence. These materials must be backed up separately +according to their trust boundaries. Database restore without matching provider +keys produces an explicit unavailable state; it must never create replacement +keys under an old reference. + +Before production use, each concrete provider/profile needs automated restore, +provider-loss, stale-backup, partial-migration, network-failure, and permanent +key-loss exercises. Permanent key loss is data loss for the affected ciphertext +and must be reported as such. + +## Current Integration Contract + +Files, Postbox, Campaign, Calendar, and connector credentials may store an +envelope id and ciphertext reference while retaining their own resource model. +They resolve Encryption through optional capabilities and must fail closed when +an encrypted record is encountered without the capability. Unencrypted module +combinations continue to operate without Encryption. + +Focused fixtures currently validate Files- and Postbox-owned envelope records. +Concrete feature adapters and production provider conformance remain tracked +work; direct imports of Encryption database models from feature modules are not +allowed. diff --git a/src/govoplan_encryption/backend/db/__init__.py b/src/govoplan_encryption/backend/db/__init__.py new file mode 100644 index 0000000..42ad371 --- /dev/null +++ b/src/govoplan_encryption/backend/db/__init__.py @@ -0,0 +1,19 @@ +from govoplan_encryption.backend.db.models import ( + ContentProtectionRecord, + EncryptionKeyOperation, + EncryptionKeyVersion, + EncryptionVault, + ProtectionMigration, + RecoveryApproval, + RecoveryCeremony, +) + +__all__ = [ + "ContentProtectionRecord", + "EncryptionKeyOperation", + "EncryptionKeyVersion", + "EncryptionVault", + "ProtectionMigration", + "RecoveryApproval", + "RecoveryCeremony", +] diff --git a/src/govoplan_encryption/backend/db/models.py b/src/govoplan_encryption/backend/db/models.py new file mode 100644 index 0000000..57e3b0f --- /dev/null +++ b/src/govoplan_encryption/backend/db/models.py @@ -0,0 +1,313 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any +import uuid + +from sqlalchemy import ( + DateTime, + Index, + Integer, + JSON, + String, + Text, + UniqueConstraint, +) +from sqlalchemy.orm import Mapped, mapped_column + +from govoplan_core.db.base import Base, TimestampMixin + + +def new_uuid() -> str: + return str(uuid.uuid4()) + + +class EncryptionVault(Base, TimestampMixin): + __tablename__ = "encryption_vaults" + __table_args__ = ( + UniqueConstraint("tenant_id", "vault_id", name="uq_encryption_vault"), + UniqueConstraint( + "tenant_id", + "create_idempotency_key", + name="uq_encryption_vault_create_idempotency", + ), + Index( + "ix_encryption_vault_scope", + "tenant_id", + "scope_type", + "scope_id", + "state", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + vault_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True) + name: Mapped[str] = mapped_column(String(255), nullable=False) + provider_id: Mapped[str] = mapped_column(String(120), nullable=False, index=True) + purpose: Mapped[str] = mapped_column(String(255), nullable=False) + profile_kind: Mapped[str] = mapped_column(String(40), nullable=False) + scope_type: Mapped[str] = mapped_column(String(80), nullable=False, index=True) + scope_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + policy_ref: Mapped[str] = mapped_column(String(1000), nullable=False) + recovery_quorum: Mapped[int] = mapped_column(Integer, nullable=False) + state: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + current_key_version: Mapped[int | None] = mapped_column(Integer, nullable=True) + create_idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + create_request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + provenance: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + created_by: Mapped[str] = mapped_column(String(255), nullable=False) + updated_by: Mapped[str] = mapped_column(String(255), nullable=False) + + +class EncryptionKeyVersion(Base, TimestampMixin): + __tablename__ = "encryption_key_versions" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "vault_id", + "version", + name="uq_encryption_key_version", + ), + UniqueConstraint( + "tenant_id", + "provider_id", + "provider_key_ref", + name="uq_encryption_provider_key_ref", + ), + Index( + "ix_encryption_key_vault_state", + "tenant_id", + "vault_id", + "state", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + vault_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True) + version: Mapped[int] = mapped_column(Integer, nullable=False) + provider_id: Mapped[str] = mapped_column(String(120), nullable=False, index=True) + provider_key_ref: Mapped[str | None] = mapped_column(String(1000), nullable=True) + algorithm_suite: Mapped[str] = mapped_column(String(120), nullable=False) + state: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + public_key_ref: Mapped[str | None] = mapped_column(String(1000), nullable=True) + imported: Mapped[bool] = mapped_column(default=False, nullable=False) + exportable: Mapped[bool] = mapped_column(default=False, nullable=False) + provider_version: Mapped[str | None] = mapped_column(String(120), nullable=True) + provider_provenance: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + activated_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + revoked_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + destruction_scheduled_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + destroyed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + +class EncryptionKeyOperation(Base, TimestampMixin): + __tablename__ = "encryption_key_operations" + __table_args__ = ( + UniqueConstraint( + "tenant_id", "idempotency_key", name="uq_encryption_key_operation_idem" + ), + Index( + "ix_encryption_key_operation_state", + "tenant_id", + "state", + "updated_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + vault_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True) + key_version: Mapped[int] = mapped_column(Integer, nullable=False) + operation: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + provider_id: Mapped[str] = mapped_column(String(120), nullable=False) + state: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + request_payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + error_code: Mapped[str | None] = mapped_column(String(255), nullable=True) + policy_decision_ref: Mapped[str | None] = mapped_column(String(1000), nullable=True) + assurance_evidence_ref: Mapped[str | None] = mapped_column( + String(1000), nullable=True + ) + requested_by: Mapped[str] = mapped_column(String(255), nullable=False) + completed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + +class ContentProtectionRecord(Base, TimestampMixin): + __tablename__ = "encryption_content_protections" + __table_args__ = ( + UniqueConstraint( + "tenant_id", "envelope_id", name="uq_encryption_content_envelope" + ), + UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_content_idempotency", + ), + Index( + "ix_encryption_content_owner", + "tenant_id", + "owner_module", + "resource_type", + "resource_id", + ), + Index("ix_encryption_content_state", "tenant_id", "state", "updated_at"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + envelope_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + owner_module: Mapped[str] = mapped_column(String(120), nullable=False, index=True) + resource_type: Mapped[str] = mapped_column(String(120), nullable=False) + resource_id: Mapped[str] = mapped_column(String(255), nullable=False) + profile_kind: Mapped[str] = mapped_column(String(40), nullable=False) + profile_id: Mapped[str] = mapped_column(String(255), nullable=False) + provider_id: Mapped[str] = mapped_column(String(120), nullable=False) + vault_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True) + key_version: Mapped[int] = mapped_column(Integer, nullable=False) + algorithm_suite: Mapped[str] = mapped_column(String(120), nullable=False) + ciphertext_ref: Mapped[str] = mapped_column(String(2000), nullable=False) + ciphertext_digest: Mapped[str] = mapped_column(String(255), nullable=False) + authenticated_context_digest: Mapped[str] = mapped_column( + String(255), nullable=False + ) + wrapped_key_refs: Mapped[list[str]] = mapped_column( + JSON, default=list, nullable=False + ) + state: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + source_envelope_id: Mapped[str | None] = mapped_column(String(255), nullable=True) + migration_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + envelope_metadata: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + policy_decision_ref: Mapped[str] = mapped_column(String(1000), nullable=False) + registered_by: Mapped[str] = mapped_column(String(255), nullable=False) + + +class ProtectionMigration(Base, TimestampMixin): + __tablename__ = "encryption_protection_migrations" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_migration_idempotency", + ), + Index("ix_encryption_migration_state", "tenant_id", "state", "updated_at"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + source_envelope_id: Mapped[str] = mapped_column( + String(255), nullable=False, index=True + ) + target_envelope_id: Mapped[str | None] = mapped_column(String(255), nullable=True) + target_provider_id: Mapped[str] = mapped_column(String(120), nullable=False) + target_vault_id: Mapped[str] = mapped_column(String(255), nullable=False) + target_key_version: Mapped[int] = mapped_column(Integer, nullable=False) + target_algorithm_suite: Mapped[str] = mapped_column(String(120), nullable=False) + mode: Mapped[str] = mapped_column(String(40), nullable=False) + state: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + policy_decision_ref: Mapped[str] = mapped_column(String(1000), nullable=False) + assurance_evidence_ref: Mapped[str] = mapped_column(String(1000), nullable=False) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + evidence_refs: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False) + error_code: Mapped[str | None] = mapped_column(String(255), nullable=True) + provenance: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + requested_by: Mapped[str] = mapped_column(String(255), nullable=False) + completed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + +class RecoveryCeremony(Base, TimestampMixin): + __tablename__ = "encryption_recovery_ceremonies" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_recovery_idempotency", + ), + Index("ix_encryption_recovery_state", "tenant_id", "state", "expires_at"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + vault_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True) + state: Mapped[str] = mapped_column(String(40), nullable=False, index=True) + requested_scope: Mapped[str] = mapped_column(String(255), nullable=False) + reason: Mapped[str] = mapped_column(Text, nullable=False) + quorum: Mapped[int] = mapped_column(Integer, nullable=False) + revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + policy_decision_ref: Mapped[str] = mapped_column(String(1000), nullable=False) + requester_assurance_ref: Mapped[str] = mapped_column(String(1000), nullable=False) + requester_account_id: Mapped[str] = mapped_column(String(255), nullable=False) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + expires_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, index=True + ) + execution_ref: Mapped[str | None] = mapped_column(String(1000), nullable=True) + provenance: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + + +class RecoveryApproval(Base, TimestampMixin): + __tablename__ = "encryption_recovery_approvals" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "recovery_id", + "approver_account_id", + name="uq_encryption_recovery_approver", + ), + UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_recovery_approval_idem", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + recovery_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + approver_account_id: Mapped[str] = mapped_column(String(255), nullable=False) + decision: Mapped[str] = mapped_column(String(20), nullable=False) + reason: Mapped[str] = mapped_column(Text, nullable=False) + assurance_evidence_ref: Mapped[str] = mapped_column(String(1000), nullable=False) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + request_digest: Mapped[str] = mapped_column(String(64), nullable=False) + + +__all__ = [ + "ContentProtectionRecord", + "EncryptionKeyOperation", + "EncryptionKeyVersion", + "EncryptionVault", + "ProtectionMigration", + "RecoveryApproval", + "RecoveryCeremony", + "new_uuid", +] diff --git a/src/govoplan_encryption/backend/manifest.py b/src/govoplan_encryption/backend/manifest.py index 8d3ed38..08737bf 100644 --- a/src/govoplan_encryption/backend/manifest.py +++ b/src/govoplan_encryption/backend/manifest.py @@ -1,13 +1,33 @@ from __future__ import annotations +from pathlib import Path + +from govoplan_core.core.encryption import ( + CAPABILITY_ENCRYPTION_CONTENT_PROTECTION, + CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT, + CAPABILITY_ENCRYPTION_KEY_VAULT, + CAPABILITY_ENCRYPTION_RECOVERY, +) +from govoplan_core.core.module_guards import ( + drop_table_retirement_provider, + persistent_table_uninstall_guard, +) from govoplan_core.core.modules import ( + CapabilityDocumentation, + DocumentationLink, DocumentationTopic, + MigrationSpec, + ModuleContext, ModuleInterfaceProvider, ModuleManifest, + ModuleUninstallGuardResult, PermissionDefinition, RoleTemplate, ) from govoplan_core.core.provider_governance import declared_module_architecture +from govoplan_core.db.base import Base +from govoplan_encryption.backend.db import models +from govoplan_encryption.backend.service import SqlEncryptionService MODULE_ID = "encryption" @@ -27,14 +47,11 @@ OPTIONAL_DEPENDENCIES = ( "postbox", "campaigns", "workflow_engine", + "identity_trust", ) -def _permission( - scope: str, - label: str, - description: str, -) -> PermissionDefinition: +def _permission(scope: str, label: str, description: str) -> PermissionDefinition: module_id, resource, action = scope.split(":", 2) return PermissionDefinition( scope=scope, @@ -48,21 +65,64 @@ def _permission( ) +def _router(context: ModuleContext): + from govoplan_encryption.backend.router import create_router + + return create_router(context.registry) + + +def _service(context: ModuleContext) -> SqlEncryptionService: + return SqlEncryptionService(context.registry) + + +def _disable_guard( + session: object | None, + _module_id: str, +) -> tuple[ModuleUninstallGuardResult, ...]: + if session is None: + return ( + ModuleUninstallGuardResult( + "blocker", + "encryption_disable_unverified", + "Encryption cannot be disabled without proving the state of every protection envelope.", + ), + ) + try: + report = SqlEncryptionService().assess_disable(session) + except Exception as exc: + return ( + ModuleUninstallGuardResult( + "blocker", + "encryption_disable_check_failed", + f"Encryption disable preflight failed: {type(exc).__name__}.", + ), + ) + if report.allowed: + return () + return ( + ModuleUninstallGuardResult( + "blocker", + "encryption_protected_content_present", + f"Encryption still protects {report.unresolved_count} unresolved envelope(s). Migrate, decrypt, explicitly export, or cryptographically destroy them before disabling the module.", + ), + ) + + PERMISSIONS = ( _permission( USE_SCOPE, "Use encryption profiles", - "Protect and decrypt authorized content through an available profile.", + "Register and resolve protected content through configured profiles.", ), _permission( ADMIN_SCOPE, "Administer encryption", - "Manage vaults, protection profiles, key rotation, and provider policy.", + "Manage vault metadata, provider operations, rotation, migration, and policy provenance.", ), _permission( RECOVERY_SCOPE, "Approve key recovery", - "Participate in an auditable recovery ceremony without gaining content ownership.", + "Participate in a high-assurance recovery ceremony without gaining content ownership.", ), ) @@ -87,7 +147,10 @@ manifest = ModuleManifest( version=MODULE_VERSION, optional_dependencies=OPTIONAL_DEPENDENCIES, provides_interfaces=( - ModuleInterfaceProvider(name="encryption.key_vault", version="1.0.0"), + ModuleInterfaceProvider( + name="encryption.key_vault", + version="1.0.0", + ), ModuleInterfaceProvider( name="encryption.content_protection", version="1.0.0", @@ -103,6 +166,81 @@ manifest = ModuleManifest( ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, + route_factory=_router, + capability_factories={ + CAPABILITY_ENCRYPTION_KEY_VAULT: _service, + CAPABILITY_ENCRYPTION_CONTENT_PROTECTION: _service, + CAPABILITY_ENCRYPTION_RECOVERY: _service, + CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT: _service, + }, + capability_documentation={ + CAPABILITY_ENCRYPTION_KEY_VAULT: CapabilityDocumentation( + label="Governed encryption key vault", + summary=( + "Orchestrates opaque, idempotent provider key references, " + "lifecycle state, and policy provenance without exposing key material." + ), + contract_version="1.0.0", + audience=("administrator", "security_officer", "auditor"), + ), + CAPABILITY_ENCRYPTION_CONTENT_PROTECTION: CapabilityDocumentation( + label="Content-protection envelope registry", + summary=( + "Registers versioned protection envelopes and fail-closed, " + "evidence-backed migration state for feature-owned content." + ), + contract_version="1.0.0", + ), + CAPABILITY_ENCRYPTION_RECOVERY: CapabilityDocumentation( + label="Encryption recovery ceremony", + summary=( + "Requires recent high assurance, distinct custodians, quorum, " + "expiry, and immutable evidence without changing ownership." + ), + contract_version="1.0.0", + ), + CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT: CapabilityDocumentation( + label="Encryption disable preflight", + summary=( + "Blocks disable or uninstall while any protection envelope " + "remains unresolved." + ), + contract_version="1.0.0", + ), + }, + migration_spec=MigrationSpec( + module_id=MODULE_ID, + metadata=Base.metadata, + script_location=str(Path(__file__).with_name("migrations") / "versions"), + retirement_supported=True, + retirement_provider=drop_table_retirement_provider( + models.RecoveryApproval, + models.RecoveryCeremony, + models.ProtectionMigration, + models.ContentProtectionRecord, + models.EncryptionKeyOperation, + models.EncryptionKeyVersion, + models.EncryptionVault, + label=MODULE_NAME, + ), + retirement_notes=( + "Destructive retirement remains blocked until disable preflight " + "proves that no unresolved protected envelope remains." + ), + ), + uninstall_guard_providers=( + _disable_guard, + persistent_table_uninstall_guard( + models.EncryptionVault, + models.EncryptionKeyVersion, + models.EncryptionKeyOperation, + models.ContentProtectionRecord, + models.ProtectionMigration, + models.RecoveryCeremony, + models.RecoveryApproval, + label=MODULE_NAME, + ), + ), documentation=( DocumentationTopic( id="encryption.boundary", @@ -114,26 +252,57 @@ manifest = ModuleManifest( body=( "Encryption protects feature-owned content without taking over " "its business ownership. Resource ownership recovery never " - "implicitly grants cryptographic keys. Disabling the module is " - "blocked until protected objects are decrypted, rewrapped, " - "explicitly exported, or cryptographically deleted." + "implicitly grants cryptographic keys. High-risk lifecycle " + "actions require recent Identity Trust assurance. Disabling is " + "blocked until each envelope is migrated, decrypted, explicitly " + "exported, or cryptographically destroyed. No bundled provider " + "or E2EE claim is implied by enabling this module." ), layer="available", documentation_types=("admin", "user"), - audience=("user", "administrator", "security_officer", "product_owner"), + audience=( + "user", + "administrator", + "security_officer", + "product_owner", + "auditor", + ), related_modules=OPTIONAL_DEPENDENCIES, order=100, + links=( + DocumentationLink( + label="Encryption boundary and threat model", + href="govoplan-encryption/docs/ENCRYPTION_BOUNDARY.md", + kind="repository", + ), + ), ), ), architecture=declared_module_architecture( layer="institutional_foundation", kind="foundation", - maturity="scaffold", + maturity="vertical_slice", documentation_ref="docs/ENCRYPTION_BOUNDARY.md", - known_limits=("No production key vault, protected-content persistence, rotation worker, or recovery ceremony is implemented yet.",), - owned_concepts=("key vault", "content-protection envelope", "cryptographic recovery ceremony"), - non_owned_concepts=("domain content", "resource ownership", "account authentication"), + test_ref="tests/test_encryption.py", + known_limits=( + "The module orchestrates references and evidence but ships no concrete cryptographic provider, raw key store, cipher implementation, client E2EE protocol, KMS/HSM conformance suite, or production recovery executor.", + "A true E2EE claim remains prohibited until a selected client/provider profile passes its threat model, interoperability fixtures, backup/restore tests, and independent review.", + ), + owned_concepts=( + "key vault", + "content-protection envelope", + "cryptographic recovery ceremony", + ), + non_owned_concepts=( + "domain content", + "resource ownership", + "account authentication", + "provider key material", + ), + migration_docs=("docs/ENCRYPTION_BOUNDARY.md",), + recovery_docs=("docs/ENCRYPTION_BOUNDARY.md",), security_docs=("docs/ENCRYPTION_BOUNDARY.md",), + operations_docs=("README.md",), ), ) diff --git a/src/govoplan_encryption/backend/migrations/__init__.py b/src/govoplan_encryption/backend/migrations/__init__.py new file mode 100644 index 0000000..a654f16 --- /dev/null +++ b/src/govoplan_encryption/backend/migrations/__init__.py @@ -0,0 +1 @@ +"""Encryption database migrations.""" diff --git a/src/govoplan_encryption/backend/migrations/versions/__init__.py b/src/govoplan_encryption/backend/migrations/versions/__init__.py new file mode 100644 index 0000000..7672250 --- /dev/null +++ b/src/govoplan_encryption/backend/migrations/versions/__init__.py @@ -0,0 +1 @@ +"""Encryption Alembic revisions.""" diff --git a/src/govoplan_encryption/backend/migrations/versions/d4a6b8c0e2f3_v0114_encryption_lifecycle.py b/src/govoplan_encryption/backend/migrations/versions/d4a6b8c0e2f3_v0114_encryption_lifecycle.py new file mode 100644 index 0000000..716e451 --- /dev/null +++ b/src/govoplan_encryption/backend/migrations/versions/d4a6b8c0e2f3_v0114_encryption_lifecycle.py @@ -0,0 +1,319 @@ +"""v0.1.14 encryption lifecycle and protection metadata + +Revision ID: d4a6b8c0e2f3 +Revises: None +Create Date: 2026-08-01 00:00:00.000000 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "d4a6b8c0e2f3" +down_revision = None +branch_labels = None +depends_on = None + + +def _timestamps() -> tuple[sa.Column, sa.Column]: + return ( + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + ) + + +def upgrade() -> None: + op.create_table( + "encryption_vaults", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("tenant_id", sa.String(36), nullable=False), + sa.Column("vault_id", sa.String(255), nullable=False), + sa.Column("name", sa.String(255), nullable=False), + sa.Column("provider_id", sa.String(120), nullable=False), + sa.Column("purpose", sa.String(255), nullable=False), + sa.Column("profile_kind", sa.String(40), nullable=False), + sa.Column("scope_type", sa.String(80), nullable=False), + sa.Column("scope_id", sa.String(255), nullable=True), + sa.Column("policy_ref", sa.String(1000), nullable=False), + sa.Column("recovery_quorum", sa.Integer(), nullable=False), + sa.Column("state", sa.String(40), nullable=False), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("current_key_version", sa.Integer(), nullable=True), + sa.Column("create_idempotency_key", sa.String(255), nullable=False), + sa.Column("create_request_digest", sa.String(64), nullable=False), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.Column("created_by", sa.String(255), nullable=False), + sa.Column("updated_by", sa.String(255), nullable=False), + *_timestamps(), + sa.UniqueConstraint("tenant_id", "vault_id", name="uq_encryption_vault"), + sa.UniqueConstraint( + "tenant_id", + "create_idempotency_key", + name="uq_encryption_vault_create_idempotency", + ), + ) + op.create_index( + "ix_encryption_vault_scope", + "encryption_vaults", + ["tenant_id", "scope_type", "scope_id", "state"], + ) + for column in ( + "tenant_id", + "vault_id", + "provider_id", + "scope_type", + "scope_id", + "state", + ): + op.create_index(f"ix_encryption_vaults_{column}", "encryption_vaults", [column]) + + op.create_table( + "encryption_key_versions", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("tenant_id", sa.String(36), nullable=False), + sa.Column("vault_id", sa.String(255), nullable=False), + sa.Column("version", sa.Integer(), nullable=False), + sa.Column("provider_id", sa.String(120), nullable=False), + sa.Column("provider_key_ref", sa.String(1000), nullable=True), + sa.Column("algorithm_suite", sa.String(120), nullable=False), + sa.Column("state", sa.String(40), nullable=False), + sa.Column("public_key_ref", sa.String(1000), nullable=True), + sa.Column("imported", sa.Boolean(), nullable=False), + sa.Column("exportable", sa.Boolean(), nullable=False), + sa.Column("provider_version", sa.String(120), nullable=True), + sa.Column("provider_provenance", sa.JSON(), nullable=False), + sa.Column("activated_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "destruction_scheduled_at", sa.DateTime(timezone=True), nullable=True + ), + sa.Column("destroyed_at", sa.DateTime(timezone=True), nullable=True), + *_timestamps(), + sa.UniqueConstraint( + "tenant_id", "vault_id", "version", name="uq_encryption_key_version" + ), + sa.UniqueConstraint( + "tenant_id", + "provider_id", + "provider_key_ref", + name="uq_encryption_provider_key_ref", + ), + ) + op.create_index( + "ix_encryption_key_vault_state", + "encryption_key_versions", + ["tenant_id", "vault_id", "state"], + ) + for column in ("tenant_id", "vault_id", "provider_id", "state"): + op.create_index( + f"ix_encryption_key_versions_{column}", + "encryption_key_versions", + [column], + ) + + op.create_table( + "encryption_key_operations", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("tenant_id", sa.String(36), nullable=False), + sa.Column("vault_id", sa.String(255), nullable=False), + sa.Column("key_version", sa.Integer(), nullable=False), + sa.Column("operation", sa.String(40), nullable=False), + sa.Column("provider_id", sa.String(120), nullable=False), + sa.Column("state", sa.String(40), nullable=False), + sa.Column("idempotency_key", sa.String(255), nullable=False), + sa.Column("request_digest", sa.String(64), nullable=False), + sa.Column("request_payload", sa.JSON(), nullable=False), + sa.Column("error_code", sa.String(255), nullable=True), + sa.Column("policy_decision_ref", sa.String(1000), nullable=True), + sa.Column("assurance_evidence_ref", sa.String(1000), nullable=True), + sa.Column("requested_by", sa.String(255), nullable=False), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + *_timestamps(), + sa.UniqueConstraint( + "tenant_id", "idempotency_key", name="uq_encryption_key_operation_idem" + ), + ) + op.create_index( + "ix_encryption_key_operation_state", + "encryption_key_operations", + ["tenant_id", "state", "updated_at"], + ) + for column in ("tenant_id", "vault_id", "operation", "state"): + op.create_index( + f"ix_encryption_key_operations_{column}", + "encryption_key_operations", + [column], + ) + + op.create_table( + "encryption_content_protections", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("envelope_id", sa.String(255), nullable=False), + sa.Column("tenant_id", sa.String(36), nullable=False), + sa.Column("owner_module", sa.String(120), nullable=False), + sa.Column("resource_type", sa.String(120), nullable=False), + sa.Column("resource_id", sa.String(255), nullable=False), + sa.Column("profile_kind", sa.String(40), nullable=False), + sa.Column("profile_id", sa.String(255), nullable=False), + sa.Column("provider_id", sa.String(120), nullable=False), + sa.Column("vault_id", sa.String(255), nullable=False), + sa.Column("key_version", sa.Integer(), nullable=False), + sa.Column("algorithm_suite", sa.String(120), nullable=False), + sa.Column("ciphertext_ref", sa.String(2000), nullable=False), + sa.Column("ciphertext_digest", sa.String(255), nullable=False), + sa.Column("authenticated_context_digest", sa.String(255), nullable=False), + sa.Column("wrapped_key_refs", sa.JSON(), nullable=False), + sa.Column("state", sa.String(40), nullable=False), + sa.Column("source_envelope_id", sa.String(255), nullable=True), + sa.Column("migration_id", sa.String(36), nullable=True), + sa.Column("envelope_metadata", sa.JSON(), nullable=False), + sa.Column("idempotency_key", sa.String(255), nullable=False), + sa.Column("request_digest", sa.String(64), nullable=False), + sa.Column("policy_decision_ref", sa.String(1000), nullable=False), + sa.Column("registered_by", sa.String(255), nullable=False), + *_timestamps(), + sa.UniqueConstraint( + "tenant_id", "envelope_id", name="uq_encryption_content_envelope" + ), + sa.UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_content_idempotency", + ), + ) + op.create_index( + "ix_encryption_content_owner", + "encryption_content_protections", + ["tenant_id", "owner_module", "resource_type", "resource_id"], + ) + op.create_index( + "ix_encryption_content_state", + "encryption_content_protections", + ["tenant_id", "state", "updated_at"], + ) + for column in ("envelope_id", "tenant_id", "owner_module", "vault_id", "state"): + op.create_index( + f"ix_encryption_content_protections_{column}", + "encryption_content_protections", + [column], + ) + + op.create_table( + "encryption_protection_migrations", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("tenant_id", sa.String(36), nullable=False), + sa.Column("source_envelope_id", sa.String(255), nullable=False), + sa.Column("target_envelope_id", sa.String(255), nullable=True), + sa.Column("target_provider_id", sa.String(120), nullable=False), + sa.Column("target_vault_id", sa.String(255), nullable=False), + sa.Column("target_key_version", sa.Integer(), nullable=False), + sa.Column("target_algorithm_suite", sa.String(120), nullable=False), + sa.Column("mode", sa.String(40), nullable=False), + sa.Column("state", sa.String(40), nullable=False), + sa.Column("policy_decision_ref", sa.String(1000), nullable=False), + sa.Column("assurance_evidence_ref", sa.String(1000), nullable=False), + sa.Column("idempotency_key", sa.String(255), nullable=False), + sa.Column("request_digest", sa.String(64), nullable=False), + sa.Column("evidence_refs", sa.JSON(), nullable=False), + sa.Column("error_code", sa.String(255), nullable=True), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.Column("requested_by", sa.String(255), nullable=False), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + *_timestamps(), + sa.UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_migration_idempotency", + ), + ) + op.create_index( + "ix_encryption_migration_state", + "encryption_protection_migrations", + ["tenant_id", "state", "updated_at"], + ) + for column in ("tenant_id", "source_envelope_id", "state"): + op.create_index( + f"ix_encryption_protection_migrations_{column}", + "encryption_protection_migrations", + [column], + ) + + op.create_table( + "encryption_recovery_ceremonies", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("tenant_id", sa.String(36), nullable=False), + sa.Column("vault_id", sa.String(255), nullable=False), + sa.Column("state", sa.String(40), nullable=False), + sa.Column("requested_scope", sa.String(255), nullable=False), + sa.Column("reason", sa.Text(), nullable=False), + sa.Column("quorum", sa.Integer(), nullable=False), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("policy_decision_ref", sa.String(1000), nullable=False), + sa.Column("requester_assurance_ref", sa.String(1000), nullable=False), + sa.Column("requester_account_id", sa.String(255), nullable=False), + sa.Column("idempotency_key", sa.String(255), nullable=False), + sa.Column("request_digest", sa.String(64), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("execution_ref", sa.String(1000), nullable=True), + sa.Column("provenance", sa.JSON(), nullable=False), + *_timestamps(), + sa.UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_recovery_idempotency", + ), + ) + op.create_index( + "ix_encryption_recovery_state", + "encryption_recovery_ceremonies", + ["tenant_id", "state", "expires_at"], + ) + for column in ("tenant_id", "vault_id", "state", "expires_at"): + op.create_index( + f"ix_encryption_recovery_ceremonies_{column}", + "encryption_recovery_ceremonies", + [column], + ) + + op.create_table( + "encryption_recovery_approvals", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("tenant_id", sa.String(36), nullable=False), + sa.Column("recovery_id", sa.String(36), nullable=False), + sa.Column("approver_account_id", sa.String(255), nullable=False), + sa.Column("decision", sa.String(20), nullable=False), + sa.Column("reason", sa.Text(), nullable=False), + sa.Column("assurance_evidence_ref", sa.String(1000), nullable=False), + sa.Column("idempotency_key", sa.String(255), nullable=False), + sa.Column("request_digest", sa.String(64), nullable=False), + *_timestamps(), + sa.UniqueConstraint( + "tenant_id", + "recovery_id", + "approver_account_id", + name="uq_encryption_recovery_approver", + ), + sa.UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_encryption_recovery_approval_idem", + ), + ) + for column in ("tenant_id", "recovery_id"): + op.create_index( + f"ix_encryption_recovery_approvals_{column}", + "encryption_recovery_approvals", + [column], + ) + + +def downgrade() -> None: + op.drop_table("encryption_recovery_approvals") + op.drop_table("encryption_recovery_ceremonies") + op.drop_table("encryption_protection_migrations") + op.drop_table("encryption_content_protections") + op.drop_table("encryption_key_operations") + op.drop_table("encryption_key_versions") + op.drop_table("encryption_vaults") diff --git a/src/govoplan_encryption/backend/router.py b/src/govoplan_encryption/backend/router.py new file mode 100644 index 0000000..f65b999 --- /dev/null +++ b/src/govoplan_encryption/backend/router.py @@ -0,0 +1,456 @@ +from __future__ import annotations + +from dataclasses import asdict + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from govoplan_core.audit.logging import audit_event +from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope +from govoplan_core.core.encryption import ( + ContentProtectionEnvelope, + KeyLifecycleRequest, + KeyRotationRequest, + KeyVaultCreateRequest, + ProtectionMigrationRequest, + ProtectionRegistrationRequest, + RecoveryApprovalRequest, + RecoveryRequest, +) +from govoplan_core.db.session import get_session +from govoplan_encryption.backend.schemas import ( + DisablePreflightResponse, + EnvelopePayload, + EnvelopeRegistrationPayload, + EnvelopeResponse, + KeyLifecyclePayload, + KeyRotationPayload, + MigrationOutcomePayload, + MigrationRequestPayload, + MigrationResponse, + RecoveryDecisionPayload, + RecoveryRequestPayload, + RecoveryResponse, + VaultCreatePayload, + VaultResponse, +) +from govoplan_encryption.backend.service import EncryptionError, SqlEncryptionService + + +def create_router(registry: object | None = None) -> APIRouter: + router = APIRouter(prefix="/encryption", tags=["encryption"]) + service = SqlEncryptionService(registry) + + @router.post("/vaults", response_model=VaultResponse) + def create_vault( + payload: VaultCreatePayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> VaultResponse: + _require(principal, "encryption:vault:admin") + value = _call( + lambda: service.create_vault( + session, + principal, + request=KeyVaultCreateRequest( + tenant_id=principal.tenant_id, + **payload.model_dump(), + ), + ) + ) + _audit( + session, + principal, + "encryption.vault.created", + "encryption_vault", + value.vault_id, + {"provider_id": value.provider_id, "state": value.state}, + ) + session.commit() + return _vault_response(value) + + @router.get("/vaults/{vault_id}", response_model=VaultResponse) + def get_vault( + vault_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> VaultResponse: + _require(principal, "encryption:vault:use", "encryption:vault:admin") + value = service.get_vault( + session, + tenant_id=principal.tenant_id, + vault_id=vault_id, + ) + if value is None: + raise HTTPException(status_code=404, detail="Encryption vault not found.") + return _vault_response(value) + + @router.post("/vaults/{vault_id}/rotate", response_model=VaultResponse) + def rotate_key( + vault_id: str, + payload: KeyRotationPayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> VaultResponse: + _require(principal, "encryption:vault:admin") + value = _call( + lambda: service.rotate_key( + session, + principal, + request=KeyRotationRequest( + tenant_id=principal.tenant_id, + vault_id=vault_id, + **payload.model_dump(), + ), + ) + ) + _audit_lifecycle(session, principal, value, "rotated") + session.commit() + return _vault_response(value) + + @router.post("/vaults/{vault_id}/revoke", response_model=VaultResponse) + def revoke_key( + vault_id: str, + payload: KeyLifecyclePayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> VaultResponse: + _require(principal, "encryption:vault:admin") + value = _call( + lambda: service.revoke_key( + session, + principal, + request=KeyLifecycleRequest( + tenant_id=principal.tenant_id, + vault_id=vault_id, + **payload.model_dump(), + ), + ) + ) + _audit_lifecycle(session, principal, value, "revoked") + session.commit() + return _vault_response(value) + + @router.post("/vaults/{vault_id}/destruction", response_model=VaultResponse) + def schedule_destruction( + vault_id: str, + payload: KeyLifecyclePayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> VaultResponse: + _require(principal, "encryption:vault:admin") + value = _call( + lambda: service.schedule_destruction( + session, + principal, + request=KeyLifecycleRequest( + tenant_id=principal.tenant_id, + vault_id=vault_id, + **payload.model_dump(), + ), + ) + ) + _audit_lifecycle(session, principal, value, "destruction_scheduled") + session.commit() + return _vault_response(value) + + @router.post("/vaults/{vault_id}/reconcile", response_model=VaultResponse) + def reconcile_vault( + vault_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> VaultResponse: + _require(principal, "encryption:vault:admin") + value = _call( + lambda: service.reconcile_vault( + session, + principal, + tenant_id=principal.tenant_id, + vault_id=vault_id, + ) + ) + _audit_lifecycle(session, principal, value, "reconciled") + session.commit() + return _vault_response(value) + + @router.post("/envelopes", response_model=EnvelopeResponse) + def register_envelope( + payload: EnvelopeRegistrationPayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> EnvelopeResponse: + _require(principal, "encryption:vault:use", "encryption:vault:admin") + envelope = _envelope_contract(payload.envelope, principal.tenant_id) + value = _call( + lambda: service.register_envelope( + session, + principal, + request=ProtectionRegistrationRequest( + envelope=envelope, + idempotency_key=payload.idempotency_key, + policy_decision_ref=payload.policy_decision_ref, + ), + ) + ) + _audit( + session, + principal, + "encryption.envelope.registered", + "content_protection_envelope", + value.envelope_id, + { + "owner_module": value.owner_module, + "profile_kind": value.profile_kind, + "key_version": value.key_version, + }, + ) + session.commit() + return _envelope_response(value) + + @router.get("/envelopes/{envelope_id}", response_model=EnvelopeResponse) + def get_envelope( + envelope_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> EnvelopeResponse: + _require(principal, "encryption:vault:use", "encryption:vault:admin") + value = service.get_envelope( + session, + principal, + tenant_id=principal.tenant_id, + envelope_id=envelope_id, + ) + if value is None: + raise HTTPException( + status_code=404, detail="Protection envelope not found." + ) + return _envelope_response(value) + + @router.post("/migrations", response_model=MigrationResponse) + def request_migration( + payload: MigrationRequestPayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> MigrationResponse: + _require(principal, "encryption:vault:admin") + value = _call( + lambda: service.request_migration( + session, + principal, + request=ProtectionMigrationRequest( + tenant_id=principal.tenant_id, + **payload.model_dump(), + ), + ) + ) + _audit( + session, + principal, + "encryption.migration.requested", + "protection_migration", + value.migration_id, + {"state": value.state}, + ) + session.commit() + return _migration_response(value) + + @router.post("/migrations/{migration_id}/outcome", response_model=MigrationResponse) + def record_migration_outcome( + migration_id: str, + payload: MigrationOutcomePayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> MigrationResponse: + _require(principal, "encryption:vault:admin") + target = ( + _envelope_contract(payload.target_envelope, principal.tenant_id) + if payload.target_envelope is not None + else None + ) + value = _call( + lambda: service.record_migration_outcome( + session, + principal, + migration_id=migration_id, + state=payload.state, + evidence_refs=tuple(payload.evidence_refs), + target_envelope=target, + error_code=payload.error_code, + ) + ) + _audit( + session, + principal, + "encryption.migration.outcome_recorded", + "protection_migration", + migration_id, + {"state": value.state, "evidence_count": len(value.evidence_refs)}, + ) + session.commit() + return _migration_response(value) + + @router.post("/recoveries", response_model=RecoveryResponse) + def request_recovery( + payload: RecoveryRequestPayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> RecoveryResponse: + _require(principal, "encryption:recovery:approve") + value = _call( + lambda: service.request_recovery( + session, + principal, + request=RecoveryRequest( + tenant_id=principal.tenant_id, + **payload.model_dump(), + ), + ) + ) + _audit( + session, + principal, + "encryption.recovery.requested", + "encryption_recovery", + value.recovery_id, + {"quorum": value.quorum, "requested_scope": value.requested_scope}, + ) + session.commit() + return RecoveryResponse(**_serializable(value)) + + @router.post("/recoveries/{recovery_id}/decision", response_model=RecoveryResponse) + def decide_recovery( + recovery_id: str, + payload: RecoveryDecisionPayload, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> RecoveryResponse: + _require(principal, "encryption:recovery:approve") + value = _call( + lambda: service.decide_recovery( + session, + principal, + request=RecoveryApprovalRequest( + tenant_id=principal.tenant_id, + recovery_id=recovery_id, + **payload.model_dump(), + ), + ) + ) + _audit( + session, + principal, + "encryption.recovery.decided", + "encryption_recovery", + recovery_id, + {"state": value.state, "approvals": value.approvals}, + ) + session.commit() + return RecoveryResponse(**_serializable(value)) + + @router.get("/disable-preflight", response_model=DisablePreflightResponse) + def disable_preflight( + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> DisablePreflightResponse: + _require(principal, "encryption:vault:admin") + value = service.assess_disable(session, tenant_id=principal.tenant_id) + return DisablePreflightResponse(**_serializable(value)) + + return router + + +def _require(principal: ApiPrincipal, *scopes: str) -> None: + if any(has_scope(principal, scope) for scope in scopes): + return + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Missing one of: {', '.join(scopes)}", + ) + + +def _call(callback): + try: + return callback() + except (EncryptionError, ValueError) as exc: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, detail=str(exc) + ) from exc + + +def _envelope_contract( + payload: EnvelopePayload, + tenant_id: str, +) -> ContentProtectionEnvelope: + return ContentProtectionEnvelope( + tenant_id=tenant_id, + **payload.model_dump(), + ) + + +def _serializable(value) -> dict[str, object]: + data = asdict(value) + for key in ("blocking_envelope_refs", "required_actions", "evidence_refs"): + if key in data and isinstance(data[key], tuple): + data[key] = list(data[key]) + return data + + +def _vault_response(value) -> VaultResponse: + return VaultResponse(**_serializable(value)) + + +def _envelope_response(value) -> EnvelopeResponse: + data = _serializable(value) + data["wrapped_key_refs"] = list(value.wrapped_key_refs) + return EnvelopeResponse(**data) + + +def _migration_response(value) -> MigrationResponse: + data = _serializable(value) + data["evidence_refs"] = list(value.evidence_refs) + if value.target_envelope is not None: + data["target_envelope"] = _envelope_response(value.target_envelope) + return MigrationResponse(**data) + + +def _audit_lifecycle( + session: Session, + principal: ApiPrincipal, + value, + action: str, +) -> None: + _audit( + session, + principal, + f"encryption.key.{action}", + "encryption_vault", + value.vault_id, + { + "state": value.state, + "revision": value.revision, + "key_version": value.current_key.version if value.current_key else None, + }, + ) + + +def _audit( + session: Session, + principal: ApiPrincipal, + action: str, + object_type: str, + object_id: str, + details: dict[str, object], +) -> None: + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action=action, + object_type=object_type, + object_id=object_id, + details={**details, "secret_material_present": False}, + ) + + +__all__ = ["create_router"] diff --git a/src/govoplan_encryption/backend/schemas.py b/src/govoplan_encryption/backend/schemas.py new file mode 100644 index 0000000..725861b --- /dev/null +++ b/src/govoplan_encryption/backend/schemas.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +class VaultCreatePayload(BaseModel): + vault_id: str = Field(min_length=1, max_length=255) + name: str = Field(min_length=1, max_length=255) + provider_id: str = Field(min_length=1, max_length=120) + purpose: str = Field(min_length=1, max_length=255) + algorithm_suite: str = Field(min_length=1, max_length=120) + scope_type: str = Field(min_length=1, max_length=80) + scope_id: str | None = Field(default=None, max_length=255) + policy_ref: str = Field(min_length=1, max_length=1000) + idempotency_key: str = Field(min_length=1, max_length=255) + recovery_quorum: int = Field(default=2, ge=1, le=32) + profile_kind: Literal["server_envelope", "tenant_held", "end_to_end"] = ( + "server_envelope" + ) + import_reference: str | None = Field(default=None, max_length=1000) + provider_policy: dict[str, Any] = Field(default_factory=dict) + + +class KeyRotationPayload(BaseModel): + expected_revision: int = Field(ge=1) + reason: str = Field(min_length=1, max_length=2000) + policy_decision_ref: str = Field(min_length=1, max_length=1000) + assurance_evidence_ref: str = Field(min_length=1, max_length=1000) + idempotency_key: str = Field(min_length=1, max_length=255) + algorithm_suite: str | None = Field(default=None, max_length=120) + provider_policy: dict[str, Any] = Field(default_factory=dict) + + +class KeyLifecyclePayload(BaseModel): + key_version: int = Field(ge=1) + expected_revision: int = Field(ge=1) + reason: str = Field(min_length=1, max_length=2000) + policy_decision_ref: str = Field(min_length=1, max_length=1000) + assurance_evidence_ref: str = Field(min_length=1, max_length=1000) + idempotency_key: str = Field(min_length=1, max_length=255) + effective_at: datetime | None = None + + +class KeyVersionResponse(BaseModel): + tenant_id: str + vault_id: str + version: int + provider_id: str + provider_key_ref: str + algorithm_suite: str + state: str + created_at: datetime + activated_at: datetime | None = None + revoked_at: datetime | None = None + destruction_scheduled_at: datetime | None = None + destroyed_at: datetime | None = None + public_key_ref: str | None = None + imported: bool = False + exportable: bool = False + provenance: dict[str, Any] = Field(default_factory=dict) + contract_version: str = "1" + + +class VaultResponse(BaseModel): + tenant_id: str + vault_id: str + name: str + provider_id: str + purpose: str + profile_kind: str + scope_type: str + scope_id: str | None = None + policy_ref: str + recovery_quorum: int + state: str + revision: int + current_key: KeyVersionResponse | None = None + created_at: datetime + updated_at: datetime + provenance: dict[str, Any] = Field(default_factory=dict) + contract_version: str = "1" + + +class EnvelopePayload(BaseModel): + envelope_id: str = Field(min_length=1, max_length=255) + owner_module: str = Field(min_length=1, max_length=120) + resource_type: str = Field(min_length=1, max_length=120) + resource_id: str = Field(min_length=1, max_length=255) + profile_kind: Literal["server_envelope", "tenant_held", "end_to_end"] + profile_id: str = Field(min_length=1, max_length=255) + provider_id: str = Field(min_length=1, max_length=120) + vault_id: str = Field(min_length=1, max_length=255) + key_version: int = Field(ge=1) + algorithm_suite: str = Field(min_length=1, max_length=120) + ciphertext_ref: str = Field(min_length=1, max_length=2000) + ciphertext_digest: str = Field(min_length=1, max_length=255) + authenticated_context_digest: str = Field(min_length=1, max_length=255) + state: Literal[ + "active", + "migration_pending", + "migrating", + "migrated", + "decrypted", + "exported", + "destroyed", + "unavailable", + ] = "active" + created_at: datetime + wrapped_key_refs: list[str] = Field(default_factory=list, max_length=1000) + source_envelope_id: str | None = Field(default=None, max_length=255) + migration_id: str | None = Field(default=None, max_length=36) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class EnvelopeRegistrationPayload(BaseModel): + envelope: EnvelopePayload + idempotency_key: str = Field(min_length=1, max_length=255) + policy_decision_ref: str = Field(min_length=1, max_length=1000) + + +class EnvelopeResponse(EnvelopePayload): + tenant_id: str + contract_version: str = "1" + + +class MigrationRequestPayload(BaseModel): + envelope_id: str = Field(min_length=1, max_length=255) + target_provider_id: str = Field(min_length=1, max_length=120) + target_vault_id: str = Field(min_length=1, max_length=255) + target_key_version: int = Field(ge=1) + target_algorithm_suite: str = Field(min_length=1, max_length=120) + mode: Literal["rewrap", "reencrypt", "decrypt", "export", "destroy"] + policy_decision_ref: str = Field(min_length=1, max_length=1000) + assurance_evidence_ref: str = Field(min_length=1, max_length=1000) + idempotency_key: str = Field(min_length=1, max_length=255) + + +class MigrationOutcomePayload(BaseModel): + state: Literal["succeeded", "rejected", "outcome_unknown"] + evidence_refs: list[str] = Field(default_factory=list, max_length=1000) + target_envelope: EnvelopePayload | None = None + error_code: str | None = Field(default=None, max_length=255) + + +class MigrationResponse(BaseModel): + migration_id: str + state: str + source_envelope_id: str + target_envelope: EnvelopeResponse | None = None + error: str | None = None + evidence_refs: list[str] = Field(default_factory=list) + provenance: dict[str, Any] = Field(default_factory=dict) + contract_version: str = "1" + + +class RecoveryRequestPayload(BaseModel): + vault_id: str = Field(min_length=1, max_length=255) + reason: str = Field(min_length=1, max_length=2000) + requested_scope: str = Field(min_length=1, max_length=255) + policy_decision_ref: str = Field(min_length=1, max_length=1000) + assurance_evidence_ref: str = Field(min_length=1, max_length=1000) + idempotency_key: str = Field(min_length=1, max_length=255) + expires_at: datetime + + +class RecoveryDecisionPayload(BaseModel): + decision: Literal["approve", "reject"] + reason: str = Field(min_length=1, max_length=2000) + assurance_evidence_ref: str = Field(min_length=1, max_length=1000) + expected_revision: int = Field(ge=1) + idempotency_key: str = Field(min_length=1, max_length=255) + + +class RecoveryResponse(BaseModel): + tenant_id: str + recovery_id: str + vault_id: str + state: str + requested_scope: str + quorum: int + approvals: int + rejections: int + revision: int + expires_at: datetime + policy_decision_ref: str + execution_ref: str | None = None + provenance: dict[str, Any] = Field(default_factory=dict) + contract_version: str = "1" + + +class DisablePreflightResponse(BaseModel): + allowed: bool + protected_count: int + unresolved_count: int + state_counts: dict[str, int] + blocking_envelope_refs: list[str] + required_actions: list[str] + generated_at: datetime + contract_version: str = "1" + + +__all__ = [ + "DisablePreflightResponse", + "EnvelopePayload", + "EnvelopeRegistrationPayload", + "EnvelopeResponse", + "KeyLifecyclePayload", + "KeyRotationPayload", + "MigrationOutcomePayload", + "MigrationRequestPayload", + "MigrationResponse", + "RecoveryDecisionPayload", + "RecoveryRequestPayload", + "RecoveryResponse", + "VaultCreatePayload", + "VaultResponse", +] diff --git a/src/govoplan_encryption/backend/service.py b/src/govoplan_encryption/backend/service.py new file mode 100644 index 0000000..959c1ce --- /dev/null +++ b/src/govoplan_encryption/backend/service.py @@ -0,0 +1,1469 @@ +from __future__ import annotations + +from collections import Counter +from collections.abc import Mapping +from datetime import datetime, timezone +import hashlib +import json + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from govoplan_core.core.encryption import ( + ContentProtectionEnvelope, + DisablePreflightReport, + KeyLifecycleRequest, + KeyMaterialDescriptor, + KeyMaterialProvisionRequest, + KeyRotationRequest, + KeyVaultCreateRequest, + KeyVaultRef, + KeyVersionRef, + ProtectionMigrationRequest, + ProtectionMigrationResult, + ProtectionRegistrationRequest, + RecoveryApprovalRequest, + RecoveryRef, + RecoveryRequest, + key_material_provider, +) +from govoplan_core.core.identity_trust import ( + AssuranceCheckRequest, + identity_trust_assurance, +) +from govoplan_core.db.base import utcnow +from govoplan_encryption.backend.db.models import ( + ContentProtectionRecord, + EncryptionKeyOperation, + EncryptionKeyVersion, + EncryptionVault, + ProtectionMigration, + RecoveryApproval, + RecoveryCeremony, +) + + +TERMINAL_PROTECTION_STATES = frozenset( + {"migrated", "decrypted", "exported", "destroyed"} +) + + +class EncryptionError(ValueError): + pass + + +class SqlEncryptionService: + """Provider-neutral encryption metadata and recovery orchestration. + + Concrete providers receive only opaque references and idempotent operation + requests. This service never accepts or returns plaintext key material. + """ + + def __init__(self, registry: object | None = None) -> None: + self.registry = registry + + def create_vault( + self, + session: object, + principal: object, + *, + request: KeyVaultCreateRequest, + ) -> KeyVaultRef: + db = _session(session) + _require_tenant(principal, request.tenant_id) + provider = self._provider(request.provider_id) + payload = _create_payload(request) + digest = _digest(payload) + replay = self._operation_replay( + db, + tenant_id=request.tenant_id, + idempotency_key=request.idempotency_key, + digest=digest, + ) + if replay is not None: + return self._vault_ref( + self._vault( + db, + tenant_id=request.tenant_id, + vault_id=replay.vault_id, + ), + db, + ) + if db.scalar( + select(EncryptionVault.id).where( + EncryptionVault.tenant_id == request.tenant_id, + EncryptionVault.vault_id == request.vault_id, + ) + ): + raise EncryptionError("The vault id already exists.") + actor_id = _account_id(principal) + vault = EncryptionVault( + tenant_id=request.tenant_id, + vault_id=request.vault_id, + name=request.name.strip(), + provider_id=request.provider_id, + purpose=request.purpose, + profile_kind=request.profile_kind, + scope_type=request.scope_type, + scope_id=request.scope_id, + policy_ref=request.policy_ref, + recovery_quorum=request.recovery_quorum, + state="provisioning", + revision=1, + current_key_version=None, + create_idempotency_key=request.idempotency_key, + create_request_digest=digest, + provenance={"contract_version": request.contract_version}, + created_by=actor_id, + updated_by=actor_id, + ) + key = EncryptionKeyVersion( + tenant_id=request.tenant_id, + vault_id=request.vault_id, + version=1, + provider_id=request.provider_id, + provider_key_ref=None, + algorithm_suite=request.algorithm_suite, + state="unavailable", + ) + operation = self._new_operation( + request.tenant_id, + request.vault_id, + 1, + "provision", + request.provider_id, + request.idempotency_key, + digest, + payload, + actor_id, + policy_decision_ref=request.policy_ref, + ) + db.add_all((vault, key, operation)) + db.flush() + self._execute_provision( + db, + principal, + provider=provider, + vault=vault, + key=key, + operation=operation, + ) + return self._vault_ref(vault, db) + + def rotate_key( + self, + session: object, + principal: object, + *, + request: KeyRotationRequest, + ) -> KeyVaultRef: + db = _session(session) + _require_tenant(principal, request.tenant_id) + vault = self._vault( + db, + tenant_id=request.tenant_id, + vault_id=request.vault_id, + lock=True, + ) + if vault.current_key_version is None: + raise EncryptionError("The vault has no active key to rotate.") + current_key = self._key(db, vault, vault.current_key_version) + algorithm_suite = request.algorithm_suite or current_key.algorithm_suite + payload = _rotation_payload(request, vault, algorithm_suite) + digest = _digest(payload) + replay = self._operation_replay( + db, + tenant_id=request.tenant_id, + idempotency_key=request.idempotency_key, + digest=digest, + ) + if replay is not None: + return self._vault_ref(vault, db) + _expected_revision(vault, request.expected_revision) + self._verify_assurance( + db, + principal, + tenant_id=request.tenant_id, + evidence_ref=request.assurance_evidence_ref, + purpose="encryption.key.rotate", + ) + provider = self._provider(vault.provider_id) + version = vault.current_key_version + 1 + actor_id = _account_id(principal) + key = EncryptionKeyVersion( + tenant_id=request.tenant_id, + vault_id=request.vault_id, + version=version, + provider_id=vault.provider_id, + provider_key_ref=None, + algorithm_suite=algorithm_suite, + state="unavailable", + ) + operation = self._new_operation( + request.tenant_id, + request.vault_id, + version, + "rotate", + vault.provider_id, + request.idempotency_key, + digest, + payload, + actor_id, + policy_decision_ref=request.policy_decision_ref, + assurance_evidence_ref=request.assurance_evidence_ref, + ) + vault.state = "rotation_pending" + vault.revision += 1 + vault.updated_by = actor_id + db.add_all((key, operation)) + db.flush() + self._execute_provision( + db, + principal, + provider=provider, + vault=vault, + key=key, + operation=operation, + ) + return self._vault_ref(vault, db) + + def revoke_key( + self, + session: object, + principal: object, + *, + request: KeyLifecycleRequest, + ) -> KeyVaultRef: + return self._change_key_lifecycle( + session, + principal, + request=request, + operation_name="revoke", + ) + + def schedule_destruction( + self, + session: object, + principal: object, + *, + request: KeyLifecycleRequest, + ) -> KeyVaultRef: + if request.effective_at is None: + raise EncryptionError("A destruction schedule requires an effective time.") + return self._change_key_lifecycle( + session, + principal, + request=request, + operation_name="destroy", + ) + + def get_vault( + self, + session: object, + *, + tenant_id: str, + vault_id: str, + ) -> KeyVaultRef | None: + db = _session(session) + vault = db.scalar( + select(EncryptionVault).where( + EncryptionVault.tenant_id == tenant_id, + EncryptionVault.vault_id == vault_id, + ) + ) + return self._vault_ref(vault, db) if vault is not None else None + + def reconcile_vault( + self, + session: object, + principal: object, + *, + tenant_id: str, + vault_id: str, + ) -> KeyVaultRef: + db = _session(session) + _require_tenant(principal, tenant_id) + vault = self._vault(db, tenant_id=tenant_id, vault_id=vault_id, lock=True) + operation = db.scalar( + select(EncryptionKeyOperation) + .where( + EncryptionKeyOperation.tenant_id == tenant_id, + EncryptionKeyOperation.vault_id == vault_id, + EncryptionKeyOperation.state == "outcome_unknown", + ) + .order_by(EncryptionKeyOperation.created_at.desc()) + .with_for_update() + ) + if operation is None: + return self._vault_ref(vault, db) + provider = self._provider(operation.provider_id) + key = self._key(db, vault, operation.key_version) + if operation.operation in {"provision", "rotate"}: + self._execute_provision( + db, + principal, + provider=provider, + vault=vault, + key=key, + operation=operation, + reconciliation=True, + ) + else: + self._execute_lifecycle( + db, + principal, + provider=provider, + vault=vault, + key=key, + operation=operation, + reconciliation=True, + ) + return self._vault_ref(vault, db) + + def register_envelope( + self, + session: object, + principal: object, + *, + request: ProtectionRegistrationRequest, + ) -> ContentProtectionEnvelope: + db = _session(session) + _require_tenant(principal, request.envelope.tenant_id) + return self._register_envelope(db, principal, request=request) + + def get_envelope( + self, + session: object, + principal: object, + *, + tenant_id: str, + envelope_id: str, + ) -> ContentProtectionEnvelope | None: + db = _session(session) + _require_tenant(principal, tenant_id) + item = db.scalar( + select(ContentProtectionRecord).where( + ContentProtectionRecord.tenant_id == tenant_id, + ContentProtectionRecord.envelope_id == envelope_id, + ) + ) + return _envelope_ref(item) if item is not None else None + + def request_migration( + self, + session: object, + principal: object, + *, + request: ProtectionMigrationRequest, + ) -> ProtectionMigrationResult: + db = _session(session) + _require_tenant(principal, request.tenant_id) + source = self._envelope( + db, + tenant_id=request.tenant_id, + envelope_id=request.envelope_id, + lock=True, + ) + payload = _migration_payload(request) + digest = _digest(payload) + replay = db.scalar( + select(ProtectionMigration).where( + ProtectionMigration.tenant_id == request.tenant_id, + ProtectionMigration.idempotency_key == request.idempotency_key, + ) + ) + if replay is not None: + if replay.request_digest != digest: + raise EncryptionError( + "The migration idempotency key was reused for another request." + ) + return self._migration_ref(db, replay) + if source.state != "active": + raise EncryptionError("Only an active envelope can enter migration.") + target_vault = self._vault( + db, + tenant_id=request.tenant_id, + vault_id=request.target_vault_id, + ) + target_key = self._key(db, target_vault, request.target_key_version) + if ( + target_vault.provider_id != request.target_provider_id + or target_key.algorithm_suite != request.target_algorithm_suite + or target_key.state != "active" + ): + raise EncryptionError( + "The target vault, key version, provider, and algorithm do not match." + ) + self._verify_assurance( + db, + principal, + tenant_id=request.tenant_id, + evidence_ref=request.assurance_evidence_ref, + purpose=f"encryption.content.{request.mode}", + ) + item = ProtectionMigration( + tenant_id=request.tenant_id, + source_envelope_id=request.envelope_id, + target_provider_id=request.target_provider_id, + target_vault_id=request.target_vault_id, + target_key_version=request.target_key_version, + target_algorithm_suite=request.target_algorithm_suite, + mode=request.mode, + state="requested", + policy_decision_ref=request.policy_decision_ref, + assurance_evidence_ref=request.assurance_evidence_ref, + idempotency_key=request.idempotency_key, + request_digest=digest, + evidence_refs=[], + provenance={"mixed_state_allowed": False}, + requested_by=_account_id(principal), + ) + db.add(item) + db.flush() + source.state = "migration_pending" + source.migration_id = item.id + db.flush() + return self._migration_ref(db, item) + + def record_migration_outcome( + self, + session: object, + principal: object, + *, + migration_id: str, + state: str, + evidence_refs: tuple[str, ...], + target_envelope: ContentProtectionEnvelope | None = None, + error_code: str | None = None, + ) -> ProtectionMigrationResult: + db = _session(session) + item = db.scalar( + select(ProtectionMigration) + .where(ProtectionMigration.id == migration_id) + .with_for_update() + ) + if item is None: + raise EncryptionError("Protection migration not found.") + _require_tenant(principal, item.tenant_id) + source = self._envelope( + db, + tenant_id=item.tenant_id, + envelope_id=item.source_envelope_id, + lock=True, + ) + normalized_state = state.strip().lower() + if normalized_state not in {"succeeded", "rejected", "outcome_unknown"}: + raise EncryptionError("Unsupported migration outcome.") + clean_evidence = tuple( + value.strip() for value in evidence_refs if value.strip() + ) + if item.state in {"succeeded", "rejected", "outcome_unknown"}: + if ( + item.state != normalized_state + or tuple(item.evidence_refs) != clean_evidence + ): + raise EncryptionError("The migration already has another outcome.") + return self._migration_ref(db, item) + if normalized_state == "succeeded" and not clean_evidence: + raise EncryptionError( + "A successful migration requires evidence references." + ) + if normalized_state == "succeeded": + if item.mode in {"rewrap", "reencrypt"}: + if target_envelope is None: + raise EncryptionError( + "Rewrap and re-encryption require a target envelope." + ) + self._validate_migration_target(item, source, target_envelope) + target = self._register_envelope( + db, + principal, + request=ProtectionRegistrationRequest( + envelope=target_envelope, + idempotency_key=f"migration:{item.id}:target", + policy_decision_ref=item.policy_decision_ref, + ), + ) + item.target_envelope_id = target.envelope_id + source.state = "migrated" + elif item.mode == "decrypt": + source.state = "decrypted" + elif item.mode == "export": + source.state = "exported" + elif item.mode == "destroy": + source.state = "destroyed" + item.completed_at = _as_utc(utcnow()) + elif normalized_state == "rejected": + source.state = "active" + source.migration_id = None + item.completed_at = _as_utc(utcnow()) + else: + source.state = "migration_pending" + item.state = normalized_state + item.evidence_refs = list(clean_evidence) + item.error_code = _safe_error_code(error_code) + item.provenance = { + **dict(item.provenance), + "outcome_recorded_by": _account_id(principal), + "cryptographic_material_present": False, + } + db.flush() + return self._migration_ref(db, item) + + def reconcile_migration( + self, + session: object, + principal: object, + *, + migration_id: str, + ) -> ProtectionMigrationResult: + db = _session(session) + item = db.get(ProtectionMigration, migration_id) + if item is None: + raise EncryptionError("Protection migration not found.") + _require_tenant(principal, item.tenant_id) + return self._migration_ref(db, item) + + def request_recovery( + self, + session: object, + principal: object, + *, + request: RecoveryRequest, + ) -> RecoveryRef: + db = _session(session) + _require_tenant(principal, request.tenant_id) + now = _as_utc(utcnow()) + if _as_utc(request.expires_at) <= now: + raise EncryptionError("Recovery expiry must be in the future.") + vault = self._vault( + db, + tenant_id=request.tenant_id, + vault_id=request.vault_id, + ) + payload = _recovery_payload(request) + digest = _digest(payload) + replay = db.scalar( + select(RecoveryCeremony).where( + RecoveryCeremony.tenant_id == request.tenant_id, + RecoveryCeremony.idempotency_key == request.idempotency_key, + ) + ) + if replay is not None: + if replay.request_digest != digest: + raise EncryptionError( + "The recovery idempotency key was reused for another request." + ) + return self._recovery_ref(db, replay) + self._verify_assurance( + db, + principal, + tenant_id=request.tenant_id, + evidence_ref=request.assurance_evidence_ref, + purpose="encryption.recovery.request", + ) + item = RecoveryCeremony( + tenant_id=request.tenant_id, + vault_id=request.vault_id, + state="pending", + requested_scope=request.requested_scope, + reason=request.reason, + quorum=vault.recovery_quorum, + revision=1, + policy_decision_ref=request.policy_decision_ref, + requester_assurance_ref=request.assurance_evidence_ref, + requester_account_id=_account_id(principal), + idempotency_key=request.idempotency_key, + request_digest=digest, + expires_at=_as_utc(request.expires_at), + provenance={ + "key_material_released": False, + "resource_ownership_changed": False, + }, + ) + db.add(item) + db.flush() + return self._recovery_ref(db, item) + + def decide_recovery( + self, + session: object, + principal: object, + *, + request: RecoveryApprovalRequest, + ) -> RecoveryRef: + db = _session(session) + _require_tenant(principal, request.tenant_id) + ceremony = db.scalar( + select(RecoveryCeremony) + .where( + RecoveryCeremony.tenant_id == request.tenant_id, + RecoveryCeremony.id == request.recovery_id, + ) + .with_for_update() + ) + if ceremony is None: + raise EncryptionError("Recovery ceremony not found.") + self._refresh_recovery_state(ceremony) + if ceremony.state != "pending": + return self._recovery_ref(db, ceremony) + _expected_revision(ceremony, request.expected_revision) + actor_id = _account_id(principal) + if actor_id == ceremony.requester_account_id: + raise EncryptionError("The requester cannot approve their own recovery.") + payload = { + "tenant_id": request.tenant_id, + "recovery_id": request.recovery_id, + "decision": request.decision, + "reason": request.reason, + "assurance_evidence_ref": request.assurance_evidence_ref, + } + digest = _digest(payload) + replay = db.scalar( + select(RecoveryApproval).where( + RecoveryApproval.tenant_id == request.tenant_id, + RecoveryApproval.idempotency_key == request.idempotency_key, + ) + ) + if replay is not None: + if replay.request_digest != digest: + raise EncryptionError( + "The approval idempotency key was reused for another decision." + ) + return self._recovery_ref(db, ceremony) + if db.scalar( + select(RecoveryApproval.id).where( + RecoveryApproval.tenant_id == request.tenant_id, + RecoveryApproval.recovery_id == request.recovery_id, + RecoveryApproval.approver_account_id == actor_id, + ) + ): + raise EncryptionError("Each custodian may decide a recovery only once.") + self._verify_assurance( + db, + principal, + tenant_id=request.tenant_id, + evidence_ref=request.assurance_evidence_ref, + purpose="encryption.recovery.approve", + ) + db.add( + RecoveryApproval( + tenant_id=request.tenant_id, + recovery_id=request.recovery_id, + approver_account_id=actor_id, + decision=request.decision, + reason=request.reason, + assurance_evidence_ref=request.assurance_evidence_ref, + idempotency_key=request.idempotency_key, + request_digest=digest, + ) + ) + db.flush() + decisions = Counter( + db.scalars( + select(RecoveryApproval.decision).where( + RecoveryApproval.tenant_id == request.tenant_id, + RecoveryApproval.recovery_id == request.recovery_id, + ) + ) + ) + if decisions["reject"]: + ceremony.state = "rejected" + elif decisions["approve"] >= ceremony.quorum: + ceremony.state = "approved" + ceremony.revision += 1 + ceremony.provenance = { + **dict(ceremony.provenance), + "approved_by_distinct_accounts": decisions["approve"], + "key_material_released": False, + } + db.flush() + return self._recovery_ref(db, ceremony) + + def assess_disable( + self, + session: object, + *, + tenant_id: str | None = None, + limit: int = 100, + ) -> DisablePreflightReport: + db = _session(session) + statement = select( + ContentProtectionRecord.state, + func.count(ContentProtectionRecord.id), + ).group_by(ContentProtectionRecord.state) + if tenant_id is not None: + statement = statement.where(ContentProtectionRecord.tenant_id == tenant_id) + state_counts = { + str(state): int(count) for state, count in db.execute(statement).all() + } + unresolved = sum( + count + for state, count in state_counts.items() + if state not in TERMINAL_PROTECTION_STATES + ) + blockers = select( + ContentProtectionRecord.tenant_id, + ContentProtectionRecord.envelope_id, + ).where(ContentProtectionRecord.state.not_in(TERMINAL_PROTECTION_STATES)) + if tenant_id is not None: + blockers = blockers.where(ContentProtectionRecord.tenant_id == tenant_id) + blockers = blockers.order_by( + ContentProtectionRecord.tenant_id, + ContentProtectionRecord.envelope_id, + ).limit(max(1, min(limit, 500))) + refs = tuple( + f"{tenant}:{envelope}" for tenant, envelope in db.execute(blockers) + ) + return DisablePreflightReport( + allowed=unresolved == 0, + protected_count=sum(state_counts.values()), + unresolved_count=unresolved, + state_counts=state_counts, + blocking_envelope_refs=refs, + required_actions=( + () + if unresolved == 0 + else ( + "Migrate, decrypt, explicitly export, or cryptographically destroy every unresolved envelope.", + "Reconcile all outcome-unknown operations before disabling Encryption.", + ) + ), + generated_at=_as_utc(utcnow()), + ) + + def _change_key_lifecycle( + self, + session: object, + principal: object, + *, + request: KeyLifecycleRequest, + operation_name: str, + ) -> KeyVaultRef: + db = _session(session) + _require_tenant(principal, request.tenant_id) + vault = self._vault( + db, + tenant_id=request.tenant_id, + vault_id=request.vault_id, + lock=True, + ) + payload = _lifecycle_payload(request, operation_name) + digest = _digest(payload) + replay = self._operation_replay( + db, + tenant_id=request.tenant_id, + idempotency_key=request.idempotency_key, + digest=digest, + ) + if replay is not None: + return self._vault_ref(vault, db) + _expected_revision(vault, request.expected_revision) + key = self._key(db, vault, request.key_version) + if key.provider_key_ref is None: + raise EncryptionError("The key provider reference is unavailable.") + self._verify_assurance( + db, + principal, + tenant_id=request.tenant_id, + evidence_ref=request.assurance_evidence_ref, + purpose=f"encryption.key.{operation_name}", + ) + operation = self._new_operation( + request.tenant_id, + request.vault_id, + request.key_version, + operation_name, + vault.provider_id, + request.idempotency_key, + digest, + payload, + _account_id(principal), + policy_decision_ref=request.policy_decision_ref, + assurance_evidence_ref=request.assurance_evidence_ref, + ) + vault.state = f"{operation_name}_pending" + vault.revision += 1 + vault.updated_by = _account_id(principal) + db.add(operation) + db.flush() + self._execute_lifecycle( + db, + principal, + provider=self._provider(vault.provider_id), + vault=vault, + key=key, + operation=operation, + ) + return self._vault_ref(vault, db) + + def _execute_provision( + self, + db: Session, + principal: object, + *, + provider: object, + vault: EncryptionVault, + key: EncryptionKeyVersion, + operation: EncryptionKeyOperation, + reconciliation: bool = False, + ) -> None: + payload = operation.request_payload + request = KeyMaterialProvisionRequest( + tenant_id=operation.tenant_id, + vault_id=operation.vault_id, + key_version=operation.key_version, + algorithm_suite=str(payload["algorithm_suite"]), + purpose=str(payload["purpose"]), + idempotency_key=operation.idempotency_key, + provider_policy=dict(payload.get("provider_policy") or {}), + import_reference=( + str(payload["import_reference"]) + if payload.get("import_reference") is not None + else None + ), + ) + try: + descriptor = provider.provision_key( # type: ignore[attr-defined] + db, + principal, + request=request, + ) + self._apply_descriptor(vault, key, descriptor) + operation.state = "succeeded" + operation.error_code = None + operation.completed_at = _as_utc(utcnow()) + vault.current_key_version = key.version + vault.state = "active" + if reconciliation: + vault.revision += 1 + except Exception as exc: + operation.state = "outcome_unknown" + operation.error_code = _exception_code(exc) + vault.state = "outcome_unknown" + key.state = "unavailable" + db.flush() + + def _execute_lifecycle( + self, + db: Session, + principal: object, + *, + provider: object, + vault: EncryptionVault, + key: EncryptionKeyVersion, + operation: EncryptionKeyOperation, + reconciliation: bool = False, + ) -> None: + if key.provider_key_ref is None: + raise EncryptionError("The key provider reference is unavailable.") + try: + if operation.operation == "revoke": + descriptor = provider.revoke_key( # type: ignore[attr-defined] + db, + principal, + provider_key_ref=key.provider_key_ref, + reason=str(operation.request_payload["reason"]), + idempotency_key=operation.idempotency_key, + ) + else: + effective_at = datetime.fromisoformat( + str(operation.request_payload["effective_at"]) + ) + descriptor = provider.schedule_key_destruction( # type: ignore[attr-defined] + db, + principal, + provider_key_ref=key.provider_key_ref, + effective_at=effective_at, + idempotency_key=operation.idempotency_key, + ) + self._apply_descriptor(vault, key, descriptor) + operation.state = "succeeded" + operation.error_code = None + operation.completed_at = _as_utc(utcnow()) + vault.state = "active" + if key.version == vault.current_key_version and key.state in { + "revoked", + "destroyed", + }: + vault.state = "unavailable" + if reconciliation: + vault.revision += 1 + except Exception as exc: + operation.state = "outcome_unknown" + operation.error_code = _exception_code(exc) + vault.state = "outcome_unknown" + db.flush() + + @staticmethod + def _apply_descriptor( + vault: EncryptionVault, + key: EncryptionKeyVersion, + descriptor: KeyMaterialDescriptor, + ) -> None: + if descriptor.provider_id != vault.provider_id: + raise EncryptionError("The provider returned a key for another provider.") + if descriptor.algorithm_suite != key.algorithm_suite: + raise EncryptionError("The provider returned another algorithm suite.") + key.provider_key_ref = descriptor.provider_key_ref + key.state = descriptor.state + key.public_key_ref = descriptor.public_key_ref + key.imported = descriptor.imported + key.exportable = descriptor.exportable + key.provider_version = descriptor.provider_version + key.provider_provenance = dict(descriptor.provenance) + if descriptor.state == "active" and key.activated_at is None: + key.activated_at = _as_utc(descriptor.created_at) + if descriptor.state == "revoked": + key.revoked_at = _as_utc(utcnow()) + if descriptor.state == "destruction_scheduled": + key.destruction_scheduled_at = _as_utc(utcnow()) + if descriptor.state == "destroyed": + key.destroyed_at = _as_utc(utcnow()) + + def _register_envelope( + self, + db: Session, + principal: object, + *, + request: ProtectionRegistrationRequest, + ) -> ContentProtectionEnvelope: + envelope = request.envelope + digest = _digest(_envelope_payload(envelope, request.policy_decision_ref)) + replay = db.scalar( + select(ContentProtectionRecord).where( + ContentProtectionRecord.tenant_id == envelope.tenant_id, + ContentProtectionRecord.idempotency_key == request.idempotency_key, + ) + ) + if replay is not None: + if replay.request_digest != digest: + raise EncryptionError( + "The envelope idempotency key was reused for another envelope." + ) + return _envelope_ref(replay) + if db.scalar( + select(ContentProtectionRecord.id).where( + ContentProtectionRecord.tenant_id == envelope.tenant_id, + ContentProtectionRecord.envelope_id == envelope.envelope_id, + ) + ): + raise EncryptionError("The protection envelope id already exists.") + vault = self._vault( + db, + tenant_id=envelope.tenant_id, + vault_id=envelope.vault_id, + ) + key = self._key(db, vault, envelope.key_version) + if ( + vault.provider_id != envelope.provider_id + or vault.profile_kind != envelope.profile_kind + or key.algorithm_suite != envelope.algorithm_suite + or key.state != "active" + ): + raise EncryptionError( + "The envelope does not match an active vault key and protection profile." + ) + item = ContentProtectionRecord( + envelope_id=envelope.envelope_id, + tenant_id=envelope.tenant_id, + owner_module=envelope.owner_module, + resource_type=envelope.resource_type, + resource_id=envelope.resource_id, + profile_kind=envelope.profile_kind, + profile_id=envelope.profile_id, + provider_id=envelope.provider_id, + vault_id=envelope.vault_id, + key_version=envelope.key_version, + algorithm_suite=envelope.algorithm_suite, + ciphertext_ref=envelope.ciphertext_ref, + ciphertext_digest=envelope.ciphertext_digest, + authenticated_context_digest=envelope.authenticated_context_digest, + wrapped_key_refs=list(envelope.wrapped_key_refs), + state=envelope.state, + source_envelope_id=envelope.source_envelope_id, + migration_id=envelope.migration_id, + envelope_metadata=dict(envelope.metadata), + idempotency_key=request.idempotency_key, + request_digest=digest, + policy_decision_ref=request.policy_decision_ref, + registered_by=_account_id(principal), + ) + db.add(item) + db.flush() + return _envelope_ref(item) + + @staticmethod + def _validate_migration_target( + item: ProtectionMigration, + source: ContentProtectionRecord, + target: ContentProtectionEnvelope, + ) -> None: + if ( + target.tenant_id != item.tenant_id + or target.source_envelope_id != source.envelope_id + or target.migration_id != item.id + or target.owner_module != source.owner_module + or target.resource_type != source.resource_type + or target.resource_id != source.resource_id + or target.provider_id != item.target_provider_id + or target.vault_id != item.target_vault_id + or target.key_version != item.target_key_version + or target.algorithm_suite != item.target_algorithm_suite + or target.state != "active" + ): + raise EncryptionError( + "The target envelope does not exactly match the authorized migration." + ) + + def _migration_ref( + self, + db: Session, + item: ProtectionMigration, + ) -> ProtectionMigrationResult: + target = None + if item.target_envelope_id: + target_item = db.scalar( + select(ContentProtectionRecord).where( + ContentProtectionRecord.tenant_id == item.tenant_id, + ContentProtectionRecord.envelope_id == item.target_envelope_id, + ) + ) + target = _envelope_ref(target_item) if target_item is not None else None + return ProtectionMigrationResult( + migration_id=item.id, + state=item.state, # type: ignore[arg-type] + source_envelope_id=item.source_envelope_id, + target_envelope=target, + error=item.error_code, + evidence_refs=tuple(item.evidence_refs), + provenance={ + **dict(item.provenance), + "mode": item.mode, + "policy_decision_ref": item.policy_decision_ref, + }, + ) + + def _recovery_ref(self, db: Session, item: RecoveryCeremony) -> RecoveryRef: + self._refresh_recovery_state(item) + decisions = Counter( + db.scalars( + select(RecoveryApproval.decision).where( + RecoveryApproval.tenant_id == item.tenant_id, + RecoveryApproval.recovery_id == item.id, + ) + ) + ) + return RecoveryRef( + tenant_id=item.tenant_id, + recovery_id=item.id, + vault_id=item.vault_id, + state=item.state, # type: ignore[arg-type] + requested_scope=item.requested_scope, + quorum=item.quorum, + approvals=decisions["approve"], + rejections=decisions["reject"], + revision=item.revision, + expires_at=item.expires_at, + policy_decision_ref=item.policy_decision_ref, + execution_ref=item.execution_ref, + provenance=dict(item.provenance), + ) + + @staticmethod + def _refresh_recovery_state(item: RecoveryCeremony) -> None: + if item.state == "pending" and _as_utc(item.expires_at) <= _as_utc(utcnow()): + item.state = "expired" + item.revision += 1 + + def _verify_assurance( + self, + db: Session, + principal: object, + *, + tenant_id: str, + evidence_ref: str, + purpose: str, + ) -> None: + capability = identity_trust_assurance(self.registry) + if capability is None: + raise EncryptionError( + "Recent high-assurance evidence is required, but Identity Trust is unavailable." + ) + decision = capability.verify_assurance( + db, + principal, + request=AssuranceCheckRequest( + tenant_id=tenant_id, + account_id=_account_id(principal), + purpose=purpose, + minimum_level="high", + evidence_ref=evidence_ref, + evaluated_at=_as_utc(utcnow()), + maximum_age_seconds=300, + ), + ) + if not decision.allowed: + raise EncryptionError( + f"High-assurance verification failed: {decision.reason}" + ) + + def _provider(self, provider_id: str) -> object: + provider = key_material_provider(self.registry, provider_id) + if provider is None: + raise EncryptionError( + f"Encryption key-material provider {provider_id!r} is unavailable." + ) + return provider + + @staticmethod + def _new_operation( + tenant_id: str, + vault_id: str, + key_version: int, + operation: str, + provider_id: str, + idempotency_key: str, + request_digest: str, + request_payload: Mapping[str, object], + requested_by: str, + *, + policy_decision_ref: str | None = None, + assurance_evidence_ref: str | None = None, + ) -> EncryptionKeyOperation: + return EncryptionKeyOperation( + tenant_id=tenant_id, + vault_id=vault_id, + key_version=key_version, + operation=operation, + provider_id=provider_id, + state="prepared", + idempotency_key=idempotency_key, + request_digest=request_digest, + request_payload=dict(request_payload), + policy_decision_ref=policy_decision_ref, + assurance_evidence_ref=assurance_evidence_ref, + requested_by=requested_by, + ) + + @staticmethod + def _operation_replay( + db: Session, + *, + tenant_id: str, + idempotency_key: str, + digest: str, + ) -> EncryptionKeyOperation | None: + item = db.scalar( + select(EncryptionKeyOperation).where( + EncryptionKeyOperation.tenant_id == tenant_id, + EncryptionKeyOperation.idempotency_key == idempotency_key, + ) + ) + if item is not None and item.request_digest != digest: + raise EncryptionError( + "The key-operation idempotency key was reused for another request." + ) + return item + + @staticmethod + def _vault( + db: Session, + *, + tenant_id: str, + vault_id: str, + lock: bool = False, + ) -> EncryptionVault: + statement = select(EncryptionVault).where( + EncryptionVault.tenant_id == tenant_id, + EncryptionVault.vault_id == vault_id, + ) + if lock: + statement = statement.with_for_update() + item = db.scalar(statement) + if item is None: + raise EncryptionError("Encryption vault not found.") + return item + + @staticmethod + def _key( + db: Session, + vault: EncryptionVault, + version: int, + ) -> EncryptionKeyVersion: + item = db.scalar( + select(EncryptionKeyVersion).where( + EncryptionKeyVersion.tenant_id == vault.tenant_id, + EncryptionKeyVersion.vault_id == vault.vault_id, + EncryptionKeyVersion.version == version, + ) + ) + if item is None: + raise EncryptionError("Encryption key version not found.") + return item + + @staticmethod + def _envelope( + db: Session, + *, + tenant_id: str, + envelope_id: str, + lock: bool = False, + ) -> ContentProtectionRecord: + statement = select(ContentProtectionRecord).where( + ContentProtectionRecord.tenant_id == tenant_id, + ContentProtectionRecord.envelope_id == envelope_id, + ) + if lock: + statement = statement.with_for_update() + item = db.scalar(statement) + if item is None: + raise EncryptionError("Protection envelope not found.") + return item + + def _vault_ref(self, item: EncryptionVault, db: Session) -> KeyVaultRef: + key = None + if item.current_key_version is not None: + key = _key_ref(self._key(db, item, item.current_key_version)) + return KeyVaultRef( + tenant_id=item.tenant_id, + vault_id=item.vault_id, + name=item.name, + provider_id=item.provider_id, + purpose=item.purpose, + profile_kind=item.profile_kind, # type: ignore[arg-type] + scope_type=item.scope_type, + scope_id=item.scope_id, + policy_ref=item.policy_ref, + recovery_quorum=item.recovery_quorum, + state=item.state, + revision=item.revision, + current_key=key, + created_at=item.created_at, + updated_at=item.updated_at, + provenance=dict(item.provenance), + ) + + +def _key_ref(item: EncryptionKeyVersion) -> KeyVersionRef: + if item.provider_key_ref is None: + raise EncryptionError("The current key has no provider reference.") + return KeyVersionRef( + tenant_id=item.tenant_id, + vault_id=item.vault_id, + version=item.version, + provider_id=item.provider_id, + provider_key_ref=item.provider_key_ref, + algorithm_suite=item.algorithm_suite, + state=item.state, # type: ignore[arg-type] + created_at=item.created_at, + activated_at=item.activated_at, + revoked_at=item.revoked_at, + destruction_scheduled_at=item.destruction_scheduled_at, + destroyed_at=item.destroyed_at, + public_key_ref=item.public_key_ref, + imported=item.imported, + exportable=item.exportable, + provenance={ + **dict(item.provider_provenance), + "provider_version": item.provider_version, + }, + ) + + +def _envelope_ref(item: ContentProtectionRecord) -> ContentProtectionEnvelope: + return ContentProtectionEnvelope( + envelope_id=item.envelope_id, + tenant_id=item.tenant_id, + owner_module=item.owner_module, + resource_type=item.resource_type, + resource_id=item.resource_id, + profile_kind=item.profile_kind, # type: ignore[arg-type] + profile_id=item.profile_id, + provider_id=item.provider_id, + vault_id=item.vault_id, + key_version=item.key_version, + algorithm_suite=item.algorithm_suite, + ciphertext_ref=item.ciphertext_ref, + ciphertext_digest=item.ciphertext_digest, + authenticated_context_digest=item.authenticated_context_digest, + state=item.state, # type: ignore[arg-type] + created_at=item.created_at, + wrapped_key_refs=tuple(item.wrapped_key_refs), + source_envelope_id=item.source_envelope_id, + migration_id=item.migration_id, + metadata=dict(item.envelope_metadata), + ) + + +def _create_payload(request: KeyVaultCreateRequest) -> dict[str, object]: + return { + "tenant_id": request.tenant_id, + "vault_id": request.vault_id, + "name": request.name, + "provider_id": request.provider_id, + "purpose": request.purpose, + "algorithm_suite": request.algorithm_suite, + "scope_type": request.scope_type, + "scope_id": request.scope_id, + "policy_ref": request.policy_ref, + "recovery_quorum": request.recovery_quorum, + "profile_kind": request.profile_kind, + "import_reference": request.import_reference, + "provider_policy": dict(request.provider_policy), + } + + +def _rotation_payload( + request: KeyRotationRequest, + vault: EncryptionVault, + algorithm_suite: str, +) -> dict[str, object]: + return { + "tenant_id": request.tenant_id, + "vault_id": request.vault_id, + "purpose": vault.purpose, + "algorithm_suite": algorithm_suite, + "reason": request.reason, + "policy_decision_ref": request.policy_decision_ref, + "assurance_evidence_ref": request.assurance_evidence_ref, + "provider_policy": dict(request.provider_policy), + "import_reference": None, + } + + +def _lifecycle_payload( + request: KeyLifecycleRequest, + operation: str, +) -> dict[str, object]: + return { + "tenant_id": request.tenant_id, + "vault_id": request.vault_id, + "key_version": request.key_version, + "operation": operation, + "reason": request.reason, + "policy_decision_ref": request.policy_decision_ref, + "assurance_evidence_ref": request.assurance_evidence_ref, + "effective_at": ( + _as_utc(request.effective_at).isoformat() + if request.effective_at is not None + else None + ), + } + + +def _migration_payload(request: ProtectionMigrationRequest) -> dict[str, object]: + return { + "tenant_id": request.tenant_id, + "envelope_id": request.envelope_id, + "target_provider_id": request.target_provider_id, + "target_vault_id": request.target_vault_id, + "target_key_version": request.target_key_version, + "target_algorithm_suite": request.target_algorithm_suite, + "mode": request.mode, + "policy_decision_ref": request.policy_decision_ref, + "assurance_evidence_ref": request.assurance_evidence_ref, + } + + +def _recovery_payload(request: RecoveryRequest) -> dict[str, object]: + return { + "tenant_id": request.tenant_id, + "vault_id": request.vault_id, + "reason": request.reason, + "requested_scope": request.requested_scope, + "policy_decision_ref": request.policy_decision_ref, + "assurance_evidence_ref": request.assurance_evidence_ref, + "expires_at": _as_utc(request.expires_at).isoformat(), + } + + +def _envelope_payload( + envelope: ContentProtectionEnvelope, + policy_decision_ref: str, +) -> dict[str, object]: + return { + "envelope_id": envelope.envelope_id, + "tenant_id": envelope.tenant_id, + "owner_module": envelope.owner_module, + "resource_type": envelope.resource_type, + "resource_id": envelope.resource_id, + "profile_kind": envelope.profile_kind, + "profile_id": envelope.profile_id, + "provider_id": envelope.provider_id, + "vault_id": envelope.vault_id, + "key_version": envelope.key_version, + "algorithm_suite": envelope.algorithm_suite, + "ciphertext_ref": envelope.ciphertext_ref, + "ciphertext_digest": envelope.ciphertext_digest, + "authenticated_context_digest": envelope.authenticated_context_digest, + "state": envelope.state, + "wrapped_key_refs": list(envelope.wrapped_key_refs), + "source_envelope_id": envelope.source_envelope_id, + "migration_id": envelope.migration_id, + "metadata": dict(envelope.metadata), + "policy_decision_ref": policy_decision_ref, + } + + +def _expected_revision(item: object, expected: int) -> None: + actual = int(getattr(item, "revision", 0)) + if actual != expected: + raise EncryptionError( + f"The resource changed (expected revision {expected}, current {actual})." + ) + + +def _require_tenant(principal: object, tenant_id: str) -> None: + principal_tenant = str(getattr(principal, "tenant_id", "") or "") + if not principal_tenant or principal_tenant != tenant_id: + raise EncryptionError("Cross-tenant encryption access is not allowed.") + + +def _account_id(principal: object) -> str: + user = getattr(principal, "user", None) + value = ( + getattr(user, "id", None) + or getattr(principal, "account_id", None) + or getattr(principal, "api_key_id", None) + ) + if value is None or not str(value).strip(): + raise EncryptionError("An accountable principal is required.") + return str(value) + + +def _session(value: object) -> Session: + if not isinstance(value, Session): + raise TypeError("Encryption requires a SQLAlchemy Session.") + return value + + +def _digest(value: Mapping[str, object]) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _as_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _exception_code(exc: Exception) -> str: + return f"provider_{type(exc).__name__.lower()}"[:255] + + +def _safe_error_code(value: str | None) -> str | None: + if value is None: + return None + clean = "".join( + character for character in value if character.isalnum() or character in "._-" + ) + return clean[:255] or "provider_error" + + +__all__ = [ + "EncryptionError", + "SqlEncryptionService", + "TERMINAL_PROTECTION_STATES", +] diff --git a/tests/test_encryption.py b/tests/test_encryption.py new file mode 100644 index 0000000..cdd9a51 --- /dev/null +++ b/tests/test_encryption.py @@ -0,0 +1,389 @@ +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() diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 5307a71..bbf90dd 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -2,37 +2,34 @@ from __future__ import annotations import unittest -from govoplan_encryption.backend.manifest import ( - ADMIN_SCOPE, - RECOVERY_SCOPE, - USE_SCOPE, - get_manifest, +from govoplan_core.core.encryption import ( + CAPABILITY_ENCRYPTION_CONTENT_PROTECTION, + CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT, + CAPABILITY_ENCRYPTION_KEY_VAULT, + CAPABILITY_ENCRYPTION_RECOVERY, ) +from govoplan_encryption.backend.manifest import get_manifest class EncryptionManifestTests(unittest.TestCase): - def test_manifest_is_an_optional_contract_only_seed(self) -> None: + def test_manifest_exposes_headless_governed_capabilities(self) -> None: manifest = get_manifest() - self.assertEqual(manifest.id, "encryption") - self.assertEqual(manifest.dependencies, ()) - self.assertIn("access", manifest.optional_dependencies) + self.assertEqual("encryption", manifest.id) + self.assertEqual((), manifest.dependencies) self.assertEqual( - {permission.scope for permission in manifest.permissions}, - {USE_SCOPE, ADMIN_SCOPE, RECOVERY_SCOPE}, - ) - self.assertEqual( - {provider.name for provider in manifest.provides_interfaces}, { - "encryption.key_vault", - "encryption.content_protection", - "encryption.recovery_ceremony", - "encryption.disable_preflight", + CAPABILITY_ENCRYPTION_KEY_VAULT, + CAPABILITY_ENCRYPTION_CONTENT_PROTECTION, + CAPABILITY_ENCRYPTION_RECOVERY, + CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT, }, + set(manifest.capability_factories), ) - self.assertIsNone(manifest.route_factory) - self.assertIsNone(manifest.migration_spec) + self.assertIsNotNone(manifest.route_factory) + self.assertIsNotNone(manifest.migration_spec) self.assertIsNone(manifest.frontend) + self.assertEqual("vertical_slice", manifest.architecture.maturity) if __name__ == "__main__": diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..1b4f413 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from alembic.runtime.migration import MigrationContext +from sqlalchemy import create_engine, inspect + +from govoplan_core.db.migrations import migrate_database +from govoplan_encryption.backend.manifest import get_manifest + + +class EncryptionMigrationTests(unittest.TestCase): + def test_migration_creates_lifecycle_tables(self) -> None: + with tempfile.TemporaryDirectory(prefix="govoplan-encryption-") as directory: + url = f"sqlite:///{Path(directory) / 'encryption.db'}" + migrate_database( + database_url=url, + enabled_modules=("encryption",), + manifest_factories=(get_manifest,), + ) + engine = create_engine(url) + try: + tables = set(inspect(engine).get_table_names()) + self.assertTrue( + { + "encryption_vaults", + "encryption_key_versions", + "encryption_key_operations", + "encryption_content_protections", + "encryption_protection_migrations", + "encryption_recovery_ceremonies", + "encryption_recovery_approvals", + }.issubset(tables) + ) + with engine.connect() as connection: + self.assertIn( + "d4a6b8c0e2f3", + set(MigrationContext.configure(connection).get_current_heads()), + ) + finally: + engine.dispose() + + +if __name__ == "__main__": + unittest.main()