diff --git a/docs/DEPLOYMENT_OPERATOR_GUIDE.md b/docs/DEPLOYMENT_OPERATOR_GUIDE.md index 6310843..10d02d0 100644 --- a/docs/DEPLOYMENT_OPERATOR_GUIDE.md +++ b/docs/DEPLOYMENT_OPERATOR_GUIDE.md @@ -36,7 +36,7 @@ set +a | Setting | Required outside dev | Purpose | | --- | --- | --- | | `APP_ENV` | yes | Runtime profile. Use `prod`, `staging`, or a deployment-specific value outside local development. | -| `MASTER_KEY_B64` | yes | Fernet key or base64 encoded 32-byte key used for encrypted module secrets. Rotate through an explicit operator plan. | +| `MASTER_KEY_B64` | yes | Fernet key or base64 encoded 32-byte deployment root used for encrypted module secrets and, when enabled, the Encryption module's local server-envelope provider. Rotate only through an explicit provider-aware migration plan. | | `DATABASE_URL` | yes | SQLAlchemy database URL for core and installed modules. SQLite is supported for dev/small installs; PostgreSQL is the preferred production target. | | `ENABLED_MODULES` | yes | Comma-separated startup module set. Keep `tenancy,access` enabled; keep `admin` enabled for operator UI. | @@ -485,7 +485,7 @@ checks, catalog trust, signing, keyring, replay, and license operation. ## Operator Checklist - Runtime secrets are injected outside git. -- `MASTER_KEY_B64` is set and backed up securely. +- `MASTER_KEY_B64` is set and backed up securely; restores of locally encrypted content fail closed without the exact matching key. - Database backup and restore commands are tested. - File/object storage is durable and backed up. - `CORS_ORIGINS` and cookie settings match the deployed WebUI origin. diff --git a/src/govoplan_core/core/encryption.py b/src/govoplan_core/core/encryption.py index 9d68c29..a3a6035 100644 --- a/src/govoplan_core/core/encryption.py +++ b/src/govoplan_core/core/encryption.py @@ -6,11 +6,15 @@ from datetime import datetime from typing import Literal, Protocol, runtime_checkable -CAPABILITY_ENCRYPTION_KEY_VAULT = "encryption.keyVault" -CAPABILITY_ENCRYPTION_CONTENT_PROTECTION = "encryption.contentProtection" -CAPABILITY_ENCRYPTION_RECOVERY = "encryption.recoveryCeremony" -CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT = "encryption.disablePreflight" -CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX = "encryption.keyMaterialProvider." +CAPABILITY_ENCRYPTION_KEY_VAULT = "encryption.key_vault" +CAPABILITY_ENCRYPTION_CONTENT_PROTECTION = "encryption.content_protection" +CAPABILITY_ENCRYPTION_CONTENT_CIPHER = "encryption.content_cipher" +CAPABILITY_ENCRYPTION_RECOVERY = "encryption.recovery_ceremony" +CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT = "encryption.disable_preflight" +CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX = "encryption.key_material_provider." +CAPABILITY_ENCRYPTION_CONTENT_CIPHER_PROVIDER_PREFIX = ( + "encryption.content_cipher_provider." +) ENCRYPTION_CONTRACT_VERSION = "1" ProtectionProfileKind = Literal[ @@ -271,6 +275,220 @@ class ContentProtectionEnvelope: _reject_secret_mapping(self.metadata, "Protection metadata") +@dataclass(frozen=True, slots=True) +class ContentCipherEncryptRequest: + """Provider request whose key material remains behind the provider boundary.""" + + tenant_id: str + vault_id: str + key_version: int + provider_key_ref: str + algorithm_suite: str + plaintext: bytes = field(repr=False) + authenticated_context: bytes = field(repr=False) + idempotency_key: str + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + if self.key_version < 1: + raise ValueError("Content cipher key version must be positive") + for label, value in ( + ("tenant id", self.tenant_id), + ("vault id", self.vault_id), + ("provider key reference", self.provider_key_ref), + ("algorithm suite", self.algorithm_suite), + ("idempotency key", self.idempotency_key), + ): + _require_text(value, label) + if not self.authenticated_context: + raise ValueError("Authenticated context is required") + + +@dataclass(frozen=True, slots=True) +class ContentCipherEncryptResult: + provider_id: str + provider_key_ref: str + wrapped_key_ref: str + algorithm_suite: str + ciphertext: bytes = field(repr=False) + ciphertext_digest: str + authenticated_context_digest: str + created_at: datetime + provenance: Mapping[str, object] = field(default_factory=dict) + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + for label, value in ( + ("provider id", self.provider_id), + ("provider key reference", self.provider_key_ref), + ("wrapped key reference", self.wrapped_key_ref), + ("algorithm suite", self.algorithm_suite), + ("ciphertext digest", self.ciphertext_digest), + ("authenticated context digest", self.authenticated_context_digest), + ): + _require_text(value, label) + _reject_secret_mapping(self.provenance, "Cipher provenance") + + +@dataclass(frozen=True, slots=True) +class ContentCipherDecryptRequest: + tenant_id: str + provider_key_ref: str + wrapped_key_ref: str + algorithm_suite: str + ciphertext: bytes = field(repr=False) + ciphertext_digest: str + authenticated_context: bytes = field(repr=False) + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + for label, value in ( + ("tenant id", self.tenant_id), + ("provider key reference", self.provider_key_ref), + ("wrapped key reference", self.wrapped_key_ref), + ("algorithm suite", self.algorithm_suite), + ("ciphertext digest", self.ciphertext_digest), + ): + _require_text(value, label) + if not self.authenticated_context: + raise ValueError("Authenticated context is required") + + +@dataclass(frozen=True, slots=True) +class ContentCipherDecryptResult: + plaintext: bytes = field(repr=False) + provider_id: str + provider_key_ref: str + wrapped_key_ref: str + verified_at: datetime + provenance: Mapping[str, object] = field(default_factory=dict) + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + for label, value in ( + ("provider id", self.provider_id), + ("provider key reference", self.provider_key_ref), + ("wrapped key reference", self.wrapped_key_ref), + ): + _require_text(value, label) + _reject_secret_mapping(self.provenance, "Cipher provenance") + + +@dataclass(frozen=True, slots=True) +class ContentKeyRewrapRequest: + tenant_id: str + source_provider_key_ref: str + source_wrapped_key_ref: str + target_provider_key_ref: str + algorithm_suite: str + idempotency_key: str + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + for label, value in ( + ("tenant id", self.tenant_id), + ("source provider key reference", self.source_provider_key_ref), + ("source wrapped key reference", self.source_wrapped_key_ref), + ("target provider key reference", self.target_provider_key_ref), + ("algorithm suite", self.algorithm_suite), + ("idempotency key", self.idempotency_key), + ): + _require_text(value, label) + + +@dataclass(frozen=True, slots=True) +class ContentKeyRewrapResult: + provider_id: str + source_wrapped_key_ref: str + target_provider_key_ref: str + target_wrapped_key_ref: str + algorithm_suite: str + completed_at: datetime + provenance: Mapping[str, object] = field(default_factory=dict) + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + for label, value in ( + ("provider id", self.provider_id), + ("source wrapped key reference", self.source_wrapped_key_ref), + ("target provider key reference", self.target_provider_key_ref), + ("target wrapped key reference", self.target_wrapped_key_ref), + ("algorithm suite", self.algorithm_suite), + ): + _require_text(value, label) + _reject_secret_mapping(self.provenance, "Rewrap provenance") + + +@dataclass(frozen=True, slots=True) +class ContentProtectionRequest: + tenant_id: str + owner_module: str + resource_type: str + resource_id: str + profile_id: str + vault_id: str + ciphertext_ref: str + plaintext: bytes = field(repr=False) + policy_decision_ref: str + idempotency_key: str + actor_id: str | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + for label, value in ( + ("tenant id", self.tenant_id), + ("owner module", self.owner_module), + ("resource type", self.resource_type), + ("resource id", self.resource_id), + ("profile id", self.profile_id), + ("vault id", self.vault_id), + ("ciphertext reference", self.ciphertext_ref), + ("policy decision reference", self.policy_decision_ref), + ("idempotency key", self.idempotency_key), + ): + _require_text(value, label) + _reject_secret_mapping(self.metadata, "Protection metadata") + + +@dataclass(frozen=True, slots=True) +class ProtectedContent: + envelope: ContentProtectionEnvelope + ciphertext: bytes = field(repr=False) + replayed: bool = False + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + +@dataclass(frozen=True, slots=True) +class ContentUnprotectionRequest: + tenant_id: str + owner_module: str + resource_type: str + resource_id: str + envelope_id: str + ciphertext: bytes = field(repr=False) + actor_id: str | None = None + contract_version: str = ENCRYPTION_CONTRACT_VERSION + + def __post_init__(self) -> None: + _validate_contract(self.contract_version) + for label, value in ( + ("tenant id", self.tenant_id), + ("owner module", self.owner_module), + ("resource type", self.resource_type), + ("resource id", self.resource_id), + ("envelope id", self.envelope_id), + ): + _require_text(value, label) + + @dataclass(frozen=True, slots=True) class ProtectionRegistrationRequest: envelope: ContentProtectionEnvelope @@ -453,6 +671,41 @@ class EncryptionKeyMaterialProvider(Protocol): ) -> KeyMaterialDescriptor | None: ... +@runtime_checkable +class EncryptionContentCipherProvider(Protocol): + """Cryptographic provider that never exports plaintext content keys.""" + + def encrypt_content( + self, + session: object, + *, + request: ContentCipherEncryptRequest, + ) -> ContentCipherEncryptResult: ... + + def decrypt_content( + self, + session: object, + *, + request: ContentCipherDecryptRequest, + ) -> ContentCipherDecryptResult: ... + + def rewrap_content_key( + self, + session: object, + *, + request: ContentKeyRewrapRequest, + ) -> ContentKeyRewrapResult: ... + + def destroy_wrapped_content_key( + self, + session: object, + *, + tenant_id: str, + wrapped_key_ref: str, + idempotency_key: str, + ) -> None: ... + + @runtime_checkable class EncryptionKeyVault(Protocol): def create_vault( @@ -553,6 +806,43 @@ class ContentProtectionRegistry(Protocol): ) -> ProtectionMigrationResult: ... +@runtime_checkable +class EncryptionContentCipher(Protocol): + """Owner-facing content encryption and decryption boundary.""" + + def protect_content( + self, + session: object, + *, + request: ContentProtectionRequest, + ) -> ProtectedContent: ... + + def unprotect_content( + self, + session: object, + *, + request: ContentUnprotectionRequest, + ) -> bytes: ... + + def execute_rewrap( + self, + session: object, + principal: object, + *, + migration_id: str, + ) -> ProtectionMigrationResult: ... + + def prepare_reencryption( + self, + session: object, + principal: object, + *, + migration_id: str, + source_ciphertext: bytes, + target_ciphertext_ref: str, + ) -> ProtectedContent: ... + + @runtime_checkable class EncryptionRecoveryCeremony(Protocol): def request_recovery( @@ -594,6 +884,19 @@ def key_material_provider( return capability if isinstance(capability, EncryptionKeyMaterialProvider) else None +def content_cipher_provider( + registry: object | None, + provider_id: str, +) -> EncryptionContentCipherProvider | None: + capability = _capability( + registry, + f"{CAPABILITY_ENCRYPTION_CONTENT_CIPHER_PROVIDER_PREFIX}{provider_id}", + ) + return ( + capability if isinstance(capability, EncryptionContentCipherProvider) else None + ) + + def encryption_key_vault(registry: object | None) -> EncryptionKeyVault | None: value = _capability(registry, CAPABILITY_ENCRYPTION_KEY_VAULT) return value if isinstance(value, EncryptionKeyVault) else None @@ -606,6 +909,13 @@ def content_protection_registry( return value if isinstance(value, ContentProtectionRegistry) else None +def encryption_content_cipher( + registry: object | None, +) -> EncryptionContentCipher | None: + value = _capability(registry, CAPABILITY_ENCRYPTION_CONTENT_CIPHER) + return value if isinstance(value, EncryptionContentCipher) else None + + def encryption_recovery_ceremony( registry: object | None, ) -> EncryptionRecoveryCeremony | None: @@ -659,14 +969,26 @@ def _reject_secret_mapping(value: Mapping[str, object], label: str) -> None: __all__ = [ + "CAPABILITY_ENCRYPTION_CONTENT_CIPHER", + "CAPABILITY_ENCRYPTION_CONTENT_CIPHER_PROVIDER_PREFIX", "CAPABILITY_ENCRYPTION_CONTENT_PROTECTION", "CAPABILITY_ENCRYPTION_DISABLE_PREFLIGHT", "CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX", "CAPABILITY_ENCRYPTION_KEY_VAULT", "CAPABILITY_ENCRYPTION_RECOVERY", + "ContentCipherDecryptRequest", + "ContentCipherDecryptResult", + "ContentCipherEncryptRequest", + "ContentCipherEncryptResult", + "ContentKeyRewrapRequest", + "ContentKeyRewrapResult", + "ContentProtectionRequest", "ContentProtectionEnvelope", "ContentProtectionRegistry", + "ContentUnprotectionRequest", "DisablePreflightReport", + "EncryptionContentCipher", + "EncryptionContentCipherProvider", "EncryptionDisablePreflight", "EncryptionKeyMaterialProvider", "EncryptionKeyVault", @@ -681,10 +1003,13 @@ __all__ = [ "ProtectionMigrationRequest", "ProtectionMigrationResult", "ProtectionRegistrationRequest", + "ProtectedContent", "RecoveryApprovalRequest", "RecoveryRef", "RecoveryRequest", + "content_cipher_provider", "content_protection_registry", + "encryption_content_cipher", "encryption_disable_preflight", "encryption_key_vault", "encryption_recovery_ceremony", diff --git a/src/govoplan_core/core/voting.py b/src/govoplan_core/core/voting.py index ebdf1e4..04867b4 100644 --- a/src/govoplan_core/core/voting.py +++ b/src/govoplan_core/core/voting.py @@ -3,7 +3,8 @@ from __future__ import annotations from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime -from typing import Protocol, runtime_checkable +import json +from typing import Literal, Protocol, runtime_checkable CAPABILITY_VOTING_BALLOTS = "voting.ballots" @@ -109,6 +110,9 @@ class VotingResult: result_sha256: str evidence: tuple[Mapping[str, object], ...] = () + def __post_init__(self) -> None: + _validate_provider_evidence(self.evidence) + @dataclass(frozen=True, slots=True) class ExternalVotingFinalizationRequest: @@ -124,6 +128,54 @@ class ExternalVotingFinalizationRequest: idempotency_key: str +@dataclass(frozen=True, slots=True) +class ExternalVotingPreparationRequest: + tenant_id: str + ballot_id: str + requested_provider_ballot_ref: str | None + definition_sha256: str + electorate_sha256: str + assurance_profile: Literal["confidential", "secret", "external_certified"] + method: str + options: tuple[VotingOption, ...] + electorate: tuple[VotingElector, ...] + allow_replacement: bool + quorum_weight: int + threshold_numerator: int + threshold_denominator: int + opens_at: datetime | None + closes_at: datetime | None + requested_at: datetime + idempotency_key: str + + +@dataclass(frozen=True, slots=True) +class ExternalVotingBallotRef: + provider_id: str + provider_ballot_ref: str + state: Literal["prepared", "open", "closed", "outcome_unknown"] + definition_sha256: str + electorate_sha256: str + evidence: tuple[Mapping[str, object], ...] = () + + def __post_init__(self) -> None: + _validate_provider_evidence(self.evidence) + + +@dataclass(frozen=True, slots=True) +class ExternalVotingCastRequest: + tenant_id: str + ballot_id: str + provider_ballot_ref: str + definition_sha256: str + electorate_sha256: str + elector_id: str + selections: tuple[str, ...] + allow_replacement: bool + requested_at: datetime + idempotency_key: str + + @runtime_checkable class ExternalVotingProvider(Protocol): """Provider boundary for confidential, secret, or certified voting. @@ -141,6 +193,36 @@ class ExternalVotingProvider(Protocol): ) -> VotingResult: ... +@runtime_checkable +class InteractiveExternalVotingProvider(ExternalVotingProvider, Protocol): + """Provider that owns external ballot preparation, casting, and tallying.""" + + def prepare_ballot( + self, + session: object, + principal: object, + *, + request: ExternalVotingPreparationRequest, + ) -> ExternalVotingBallotRef: ... + + def cast_ballot( + self, + session: object, + principal: object, + *, + request: ExternalVotingCastRequest, + ) -> VotingReceipt: ... + + def ballot_status( + self, + session: object, + principal: object, + *, + tenant_id: str, + provider_ballot_ref: str, + ) -> ExternalVotingBallotRef | None: ... + + @runtime_checkable class VotingBallotProvider(Protocol): def create_ballot( @@ -201,11 +283,62 @@ class VotingBallotProvider(Protocol): ) -> VotingBallotRef: ... +def _validate_provider_evidence( + evidence: Sequence[Mapping[str, object]], +) -> None: + if len(evidence) > 64: + raise ValueError("Voting provider evidence exceeds the item limit") + try: + encoded = json.dumps( + [dict(item) for item in evidence], + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ).encode("utf-8") + except (TypeError, ValueError) as exc: + raise ValueError("Voting provider evidence must be bounded JSON") from exc + if len(encoded) > 64 * 1024: + raise ValueError("Voting provider evidence exceeds the size limit") + _reject_sensitive_evidence(evidence) + + +def _reject_sensitive_evidence(value: object) -> None: + sensitive_fragments = ( + "access_token", + "authorization", + "credential", + "cookie", + "key_material", + "password", + "passwd", + "plaintext", + "private_key", + "raw_vote", + "refresh_token", + "secret", + "selection", + ) + if isinstance(value, Mapping): + for key, nested in value.items(): + normalized = str(key).strip().lower().replace("-", "_") + if any(fragment in normalized for fragment in sensitive_fragments): + raise ValueError("Voting provider evidence contains a sensitive field") + _reject_sensitive_evidence(nested) + elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)): + for nested in value: + _reject_sensitive_evidence(nested) + + __all__ = [ "CAPABILITY_VOTING_BALLOTS", "CAPABILITY_VOTING_PROVIDER_PREFIX", + "ExternalVotingBallotRef", + "ExternalVotingCastRequest", "ExternalVotingFinalizationRequest", + "ExternalVotingPreparationRequest", "ExternalVotingProvider", + "InteractiveExternalVotingProvider", "VOTING_ASSURANCE_CONFIDENTIAL", "VOTING_ASSURANCE_EXTERNAL_CERTIFIED", "VOTING_ASSURANCE_RECORDED", diff --git a/tests/test_module_system.py b/tests/test_module_system.py index 9d5c3ec..e777760 100644 --- a/tests/test_module_system.py +++ b/tests/test_module_system.py @@ -3412,7 +3412,10 @@ finally: "version_min": "0.1.0", "version_max_exclusive": "0.2.0", }, modules["files"]["requires_interfaces"]) - self.assertEqual(["campaigns"], modules["files"]["optional_dependencies"]) + self.assertEqual( + ["campaigns", "encryption"], + modules["files"]["optional_dependencies"], + ) self.assertIn({"name": "mail.campaign_delivery", "version": "0.2.0"}, modules["mail"]["provides_interfaces"]) self.assertIn({ "name": "campaigns.access", diff --git a/tests/test_voting_contract.py b/tests/test_voting_contract.py new file mode 100644 index 0000000..bf4aaea --- /dev/null +++ b/tests/test_voting_contract.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.voting import VotingResult + + +def result_with_evidence(*evidence): + return VotingResult( + ballot_id="ballot-1", + revision=2, + counts={"yes": 1}, + weighted_counts={"yes": 1}, + cast_count=1, + cast_weight=1, + eligible_count=1, + eligible_weight=1, + quorum_met=True, + threshold_met=True, + winning_options=("yes",), + result_sha256="a" * 64, + evidence=evidence, + ) + + +class VotingContractTests(unittest.TestCase): + def test_accepts_sanitized_provider_evidence(self) -> None: + value = result_with_evidence( + { + "kind": "reference_provider_result", + "provider_id": "local_confidential", + "certified": False, + } + ) + + self.assertEqual("local_confidential", value.evidence[0]["provider_id"]) + + def test_rejects_secret_or_raw_selection_evidence(self) -> None: + with self.assertRaisesRegex(ValueError, "sensitive field"): + result_with_evidence({"access_token": "not-for-projection"}) + with self.assertRaisesRegex(ValueError, "sensitive field"): + result_with_evidence({"nested": {"selections": ["yes"]}}) + + def test_rejects_non_json_or_oversized_evidence(self) -> None: + with self.assertRaisesRegex(ValueError, "bounded JSON"): + result_with_evidence({"value": b"binary"}) + with self.assertRaisesRegex(ValueError, "size limit"): + result_with_evidence({"value": "x" * (64 * 1024)}) + + +if __name__ == "__main__": + unittest.main()