GovOPlaN/govoplan-core
1
0
Fork 0
Code Issues 15 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: finalize encryption and voting provider contracts

Browse Source
This commit is contained in:
zemion
2026-08-02 03:40:44 +02:00
parent 2b4eb0151f
commit 972c681650
5 changed files with 522 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/DEPLOYMENT_OPERATOR_GUIDE.md
+2 -2
View File
@@ -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.
src/govoplan_core/core/encryption.py
+330 -5
View File
@@ -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",
src/govoplan_core/core/voting.py
+134 -1
View File
@@ -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",
tests/test_module_system.py
+4 -1
View File
@@ -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",
tests/test_voting_contract.py
+52
View File
@@ -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()