From 79629c5a2c13547fe30dd102dfca7cf70baf56af Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 20 Aug 2026 12:12:27 +0200 Subject: [PATCH] feat: define campaign archive encryption policy contract --- src/govoplan_core/core/policy.py | 76 ++++++++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/src/govoplan_core/core/policy.py b/src/govoplan_core/core/policy.py index ab1838f..a53aab2 100644 --- a/src/govoplan_core/core/policy.py +++ b/src/govoplan_core/core/policy.py @@ -7,6 +7,14 @@ from urllib.parse import quote, unquote from govoplan_core.core.access import PrincipalRef PolicyScopeType = Literal["system", "tenant", "user", "group", "campaign"] +CampaignArchiveEncryptionMethod = Literal["aes", "zip_standard"] +CampaignArchivePasswordDeliveryChannel = Literal[ + "separate_mail", + "sms", + "letter", + "phone", + "in_person", +] SchedulingParticipantVisibility = Literal["aggregates_only", "names_and_statuses"] DefinitionScopeType = Literal["system", "tenant", "group", "user"] DefinitionKind = Literal["flow", "template"] @@ -45,6 +53,7 @@ CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY = "policy.schedulingParticipant CAPABILITY_POLICY_DEFINITION_GOVERNANCE = "policy.definitionGovernance" CAPABILITY_POLICY_VIEW_GOVERNANCE = "policy.viewGovernance" CAPABILITY_POLICY_FUNCTION_ASSIGNMENT_GOVERNANCE = "policy.functionAssignmentGovernance" +CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION = "policy.campaignArchiveEncryption" POLICY_SCOPE_TYPES: tuple[PolicyScopeType, ...] = ( "system", @@ -182,6 +191,73 @@ class PolicyDecision: } +@dataclass(frozen=True, slots=True) +class CampaignArchiveEncryptionRequest: + """Context required to resolve one Campaign archive-encryption ceiling. + + The owning module supplies the stable Campaign and owner references. Policy + owns hierarchy evaluation; Campaign owns archive configuration and evidence. + """ + + tenant_id: str + campaign_id: str + owner_type: Literal["user", "group"] | None = None + owner_id: str | None = None + + +@dataclass(frozen=True, slots=True) +class CampaignArchiveEncryptionDecision: + allowed_password_encryption_methods: frozenset[CampaignArchiveEncryptionMethod] + allowed_password_delivery_channels: frozenset[ + CampaignArchivePasswordDeliveryChannel + ] + policy_hash: str + source_path: tuple[PolicySourceStep, ...] = () + reason: str | None = None + diagnostics: tuple[Mapping[str, Any], ...] = () + + def to_dict(self) -> dict[str, Any]: + return { + "allowed_password_encryption_methods": sorted( + self.allowed_password_encryption_methods + ), + "allowed_password_delivery_channels": sorted( + self.allowed_password_delivery_channels + ), + "policy_hash": self.policy_hash, + "source_path": [step.to_dict() for step in self.source_path], + "reason": self.reason, + "diagnostics": [dict(item) for item in self.diagnostics], + } + + +@runtime_checkable +class CampaignArchiveEncryptionPolicy(Protocol): + def resolve_campaign_archive_encryption( + self, + session: object | None = None, + *, + request: CampaignArchiveEncryptionRequest, + ) -> CampaignArchiveEncryptionDecision: ... + + +def campaign_archive_encryption_policy( + registry: object | None, +) -> CampaignArchiveEncryptionPolicy | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not registry.has_capability(CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION) + ): + return None + capability = registry.capability(CAPABILITY_POLICY_CAMPAIGN_ARCHIVE_ENCRYPTION) + return ( + capability + if isinstance(capability, CampaignArchiveEncryptionPolicy) + else None + ) + + @dataclass(frozen=True, slots=True) class FunctionAssignmentGovernanceRequest: tenant_id: str