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

feat(postbox): add governed content protection profiles

Browse Source
This commit is contained in:
zemion
2026-08-20 03:42:58 +02:00
parent 15d93aaa25
commit 174ee97719
16 changed files with 3917 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_postbox/backend/schemas.py
+291 -31
View File
@@ -5,6 +5,12 @@ from typing import Any, Literal
from pydantic import BaseModel, Field, model_validator
from govoplan_postbox.backend.protection_profiles import (
POSTBOX_MANAGED_ENVELOPE_PROFILE,
POSTBOX_PLAINTEXT_PROFILE,
PostboxProtectionProfile,
)
PostboxClassification = Literal[
"public",
@@ -48,6 +54,10 @@ class PostboxDirectoryItem(BaseModel):
template_revision_id: str | None = None
holder_count: int = 0
vacant: bool = True
encryption_profile: str = POSTBOX_PLAINTEXT_PROFILE
key_epoch: int = Field(default=1, ge=1)
encryption_vault_id: str | None = None
protection_policy: dict[str, Any] = Field(default_factory=dict)
access: PostboxAccessDecisionResponse | None = None
resource_revision: int = Field(default=1, ge=1)
etag: str | None = None
@@ -156,11 +166,32 @@ class PostboxMessageAuthoringPayload(BaseModel):
idempotency_key: str = Field(min_length=1, max_length=255)
subject: str = Field(min_length=1, max_length=1000)
body_text: str | None = None
ciphertext_ref: str | None = Field(default=None, max_length=1000)
signed_manifest_ref: str | None = Field(default=None, max_length=1000)
wrapped_keys: list[PostboxWrappedKeyPayload] = Field(default_factory=list)
classification: PostboxClassification = "internal"
participants: list[PostboxParticipantPayload] = Field(default_factory=list)
attachments: list[PostboxAttachmentPayload] = Field(default_factory=list)
metadata: dict[str, Any] = Field(default_factory=dict)
@model_validator(mode="after")
def validate_content_source(self) -> "PostboxMessageAuthoringPayload":
if self.body_text is not None and self.ciphertext_ref:
raise ValueError(
"Provide plaintext or an external ciphertext envelope, not both."
)
if self.ciphertext_ref and (
not self.signed_manifest_ref or not self.wrapped_keys
):
raise ValueError(
"External E2EE content requires a signed manifest and wrapped keys."
)
if not self.ciphertext_ref and (self.signed_manifest_ref or self.wrapped_keys):
raise ValueError(
"A signed manifest and wrapped keys require an external ciphertext reference."
)
return self
class PostboxMessageCreateRequest(PostboxMessageAuthoringPayload):
postbox_id: str = Field(min_length=1, max_length=36)
@@ -178,9 +209,7 @@ class PostboxTargetPayload(BaseModel):
def validate_target(self) -> "PostboxTargetPayload":
direct = bool(self.postbox_id or self.address_key)
templated = bool(
self.template_id
and self.organization_unit_id
and self.function_id
self.template_id and self.organization_unit_id and self.function_id
)
if direct == templated:
raise ValueError(
@@ -210,6 +239,24 @@ class PostboxDeliveryCreateRequest(BaseModel):
)
metadata: dict[str, Any] = Field(default_factory=dict)
@model_validator(mode="after")
def validate_content_source(self) -> "PostboxDeliveryCreateRequest":
if self.body_text is not None and self.ciphertext_ref:
raise ValueError(
"Provide plaintext or an external ciphertext envelope, not both."
)
if self.ciphertext_ref and (
not self.signed_manifest_ref or not self.wrapped_keys
):
raise ValueError(
"External E2EE content requires a signed manifest and wrapped keys."
)
if not self.ciphertext_ref and (self.signed_manifest_ref or self.wrapped_keys):
raise ValueError(
"A signed manifest and wrapped keys require an external ciphertext reference."
)
return self
class PostboxDeliveryResponse(BaseModel):
delivery_id: str
@@ -248,14 +295,10 @@ class PostboxLinkedCopyPolicyPayload(BaseModel):
def validate_enabled_policy(self) -> "PostboxLinkedCopyPolicyPayload":
self.relation_type_ids = list(dict.fromkeys(self.relation_type_ids))
self.allowed_classifications = list(
dict.fromkeys(
value.strip() for value in self.allowed_classifications
)
dict.fromkeys(value.strip() for value in self.allowed_classifications)
)
self.allowed_producer_modules = list(
dict.fromkeys(
value.strip() for value in self.allowed_producer_modules
)
dict.fromkeys(value.strip() for value in self.allowed_producer_modules)
)
if any(not value for value in self.relation_type_ids):
raise ValueError("Relation type IDs must not be empty.")
@@ -316,16 +359,10 @@ class PostboxRoutingPolicyPayload(BaseModel):
@model_validator(mode="after")
def validate_semantics(self) -> "PostboxRoutingPolicyPayload":
if (
self.attention.mode == "vacancy_escalation"
and (
not self.linked_copy.enabled
or self.linked_copy.fanout != "nearest"
)
if self.attention.mode == "vacancy_escalation" and (
not self.linked_copy.enabled or self.linked_copy.fanout != "nearest"
):
raise ValueError(
"Vacancy escalation requires nearest linked-copy routing."
)
raise ValueError("Vacancy escalation requires nearest linked-copy routing.")
return self
@@ -361,6 +398,73 @@ class PostboxRouteDryRunResponse(BaseModel):
diagnostics: list[str] = Field(default_factory=list)
class PostboxProtectionPolicyPayload(BaseModel):
new_incumbent_history: Literal[
"all_retained",
"since_assignment",
"bounded_days",
] = "since_assignment"
history_days: int | None = Field(default=None, ge=1, le=36500)
ordinary_rotation: Literal["rewrap", "reencrypt"] = "rewrap"
compromise_rotation: Literal["rewrap", "reencrypt"] = "reencrypt"
recovery_authority: Literal[
"disabled",
"user_consent",
"institutional_key_holders",
"dual_control",
] = "institutional_key_holders"
recovery_quorum: int = Field(default=2, ge=1, le=20)
handover_authority: Literal[
"user_consent",
"institutional_key_holders",
"dual_control",
] = "dual_control"
handover_quorum: int = Field(default=2, ge=1, le=20)
emergency_access: Literal["disabled", "dual_control"] = "dual_control"
emergency_quorum: int = Field(default=2, ge=1, le=20)
export_authority: Literal[
"user_consent",
"institutional_key_holders",
"dual_control",
] = "dual_control"
export_quorum: int = Field(default=2, ge=1, le=20)
destruction_authority: Literal[
"institutional_key_holders",
"dual_control",
] = "dual_control"
destruction_quorum: int = Field(default=2, ge=1, le=20)
external_recipient_assurance: Literal[
"disabled",
"email_otp",
"strong_identity",
] = "strong_identity"
vacancy_escalation_content_access: Literal["metadata_only"] = "metadata_only"
@model_validator(mode="after")
def validate_history_policy(self) -> "PostboxProtectionPolicyPayload":
if self.new_incumbent_history == "bounded_days" and self.history_days is None:
raise ValueError("Bounded incumbent history requires a day limit.")
if self.new_incumbent_history != "bounded_days":
self.history_days = None
if self.handover_authority == "dual_control" and self.handover_quorum < 2:
raise ValueError(
"Dual-control hand-over requires a quorum of at least two."
)
if self.emergency_access == "dual_control" and self.emergency_quorum < 2:
raise ValueError(
"Emergency dual control requires a quorum of at least two."
)
if self.recovery_authority == "dual_control" and self.recovery_quorum < 2:
raise ValueError("Dual-control recovery requires a quorum of at least two.")
if self.export_authority == "dual_control" and self.export_quorum < 2:
raise ValueError("Dual-control export requires a quorum of at least two.")
if self.destruction_authority == "dual_control" and self.destruction_quorum < 2:
raise ValueError(
"Dual-control destruction requires a quorum of at least two."
)
return self
class PostboxExactCreateRequest(BaseModel):
name: str = Field(min_length=1, max_length=500)
description: str | None = None
@@ -369,25 +473,31 @@ class PostboxExactCreateRequest(BaseModel):
address_key: str | None = Field(default=None, max_length=120)
classification: PostboxClassification = "internal"
portal_visible: bool = False
encryption_profile: Literal["plaintext_v1", "server_envelope_v1"] = (
"plaintext_v1"
)
encryption_profile: PostboxProtectionProfile = POSTBOX_PLAINTEXT_PROFILE
encryption_vault_id: str | None = Field(default=None, max_length=255)
protection_policy: PostboxProtectionPolicyPayload = Field(
default_factory=PostboxProtectionPolicyPayload
)
@model_validator(mode="after")
def validate_encryption(self) -> "PostboxExactCreateRequest":
if self.encryption_profile == "server_envelope_v1":
if self.encryption_profile == POSTBOX_MANAGED_ENVELOPE_PROFILE:
if not str(self.encryption_vault_id or "").strip():
raise ValueError(
"Server-envelope Postboxes require an encryption vault."
)
elif self.encryption_vault_id:
raise ValueError(
"A plaintext Postbox cannot select an encryption vault."
"Only an institution-managed Postbox can select an encryption vault."
)
return self
class PostboxProtectionPolicyUpdateRequest(BaseModel):
base_revision: int = Field(ge=1)
protection_policy: PostboxProtectionPolicyPayload
class PostboxTemplateRevisionPayload(BaseModel):
function_type_id: str | None = Field(default=None, max_length=36)
scope_kind: Literal["tenant", "unit", "subtree", "unit_type"] = "tenant"
@@ -407,10 +517,11 @@ class PostboxTemplateRevisionPayload(BaseModel):
classification: PostboxClassification = "internal"
allow_vacant_delivery: bool = True
portal_visible: bool = False
encryption_profile: Literal["plaintext_v1", "server_envelope_v1"] = (
"plaintext_v1"
)
encryption_profile: PostboxProtectionProfile = POSTBOX_PLAINTEXT_PROFILE
encryption_vault_id: str | None = Field(default=None, max_length=255)
protection_policy: PostboxProtectionPolicyPayload = Field(
default_factory=PostboxProtectionPolicyPayload
)
routing_policy: PostboxRoutingPolicyPayload = Field(
default_factory=PostboxRoutingPolicyPayload
)
@@ -426,18 +537,169 @@ class PostboxTemplateRevisionPayload(BaseModel):
@model_validator(mode="after")
def validate_encryption(self) -> "PostboxTemplateRevisionPayload":
if self.encryption_profile == "server_envelope_v1":
if self.encryption_profile == POSTBOX_MANAGED_ENVELOPE_PROFILE:
if not str(self.encryption_vault_id or "").strip():
raise ValueError(
"Server-envelope Postbox templates require an encryption vault."
)
elif self.encryption_vault_id:
raise ValueError(
"A plaintext Postbox template cannot select an encryption vault."
"Only an institution-managed Postbox template can select an encryption vault."
)
return self
class PostboxProtectionProfileItem(BaseModel):
id: PostboxProtectionProfile
label: str
description: str
server_can_decrypt: bool
requires_encryption_module: bool
requires_external_client: bool
available: bool
standard: bool = False
class PostboxProtectionProfileListResponse(BaseModel):
standard_profile: PostboxProtectionProfile
profiles: list[PostboxProtectionProfileItem]
class PostboxProtectionTransitionCreateRequest(BaseModel):
idempotency_key: str = Field(min_length=1, max_length=255)
base_revision: int = Field(ge=1)
target_profile: PostboxProtectionProfile
target_vault_id: str | None = Field(default=None, max_length=255)
history_mode: Literal["future_only", "migrate_history"] = "future_only"
authority_mode: Literal[
"user_consent",
"institutional_key_holders",
"dual_control",
]
required_quorum: int = Field(default=1, ge=1, le=20)
user_consent_refs: list[str] = Field(default_factory=list, max_length=50)
institutional_authorization_refs: list[str] = Field(
default_factory=list, max_length=50
)
reason: str = Field(min_length=1, max_length=2000)
acknowledge_irreversibility: bool
@model_validator(mode="after")
def validate_transition(self) -> "PostboxProtectionTransitionCreateRequest":
self.user_consent_refs = list(
dict.fromkeys(
item.strip() for item in self.user_consent_refs if item.strip()
)
)
self.institutional_authorization_refs = list(
dict.fromkeys(
item.strip()
for item in self.institutional_authorization_refs
if item.strip()
)
)
evidence_count = len(
set(self.user_consent_refs + self.institutional_authorization_refs)
)
if evidence_count < self.required_quorum:
raise ValueError("The evidence set does not satisfy the selected quorum.")
if self.authority_mode in {"user_consent", "dual_control"} and not (
self.user_consent_refs
):
raise ValueError(
"The selected authority mode requires user consent evidence."
)
if (
self.authority_mode
in {
"institutional_key_holders",
"dual_control",
}
and not self.institutional_authorization_refs
):
raise ValueError(
"The selected authority mode requires institutional authorization evidence."
)
if self.authority_mode == "dual_control" and self.required_quorum < 2:
raise ValueError("Dual control requires a quorum of at least two.")
if not self.acknowledge_irreversibility:
raise ValueError(
"Confirm that previously decrypted, copied, or exported content cannot be recalled."
)
if self.target_profile == POSTBOX_MANAGED_ENVELOPE_PROFILE:
if not str(self.target_vault_id or "").strip():
raise ValueError("Institution-managed envelopes require a vault.")
elif self.target_vault_id:
raise ValueError("Only institution-managed envelopes select a vault.")
return self
class PostboxProtectionTransformRequest(BaseModel):
base_revision: int = Field(ge=1)
message_id: str = Field(min_length=1, max_length=36)
plaintext: str | None = None
ciphertext_ref: str | None = Field(default=None, max_length=1000)
signed_manifest_ref: str | None = Field(default=None, max_length=1000)
wrapped_keys: list[PostboxWrappedKeyPayload] = Field(default_factory=list)
content_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
transformation_evidence_ref: str = Field(min_length=1, max_length=1000)
@model_validator(mode="after")
def validate_target_payload(self) -> "PostboxProtectionTransformRequest":
if self.plaintext is not None and self.ciphertext_ref:
raise ValueError("Provide transformed plaintext or ciphertext, not both.")
if self.ciphertext_ref and (
not self.signed_manifest_ref or not self.wrapped_keys
):
raise ValueError(
"E2EE transformation requires a signed manifest and wrapped keys."
)
return self
class PostboxProtectionTransitionItemResponse(BaseModel):
id: str
message_id: str
source_profile: str
target_profile: str
state: str
source_digest: str | None = None
target_digest: str | None = None
completed_by: str | None = None
completed_at: datetime | None = None
error_code: str | None = None
evidence: dict[str, Any] = Field(default_factory=dict)
class PostboxProtectionTransitionResponse(BaseModel):
id: str
postbox_id: str
source_profile: str
target_profile: str
source_vault_id: str | None = None
target_vault_id: str | None = None
history_mode: str
authority_mode: str
required_quorum: int
evidence_refs: list[str]
reason: str
state: str
message_count: int
completed_count: int
failed_count: int
requested_by: str | None = None
activated_at: datetime | None = None
completed_at: datetime | None = None
resource_revision: int = Field(ge=1)
etag: str
configuration_snapshot: dict[str, Any] = Field(default_factory=dict)
items: list[PostboxProtectionTransitionItemResponse] = Field(default_factory=list)
class PostboxProtectionTransitionListResponse(BaseModel):
transitions: list[PostboxProtectionTransitionResponse]
def _validate_template_write_scope(
payload: PostboxTemplateRevisionPayload,
) -> None:
@@ -586,9 +848,7 @@ class PostboxOrganizationStructureItem(BaseModel):
class PostboxOrganizationTargetsResponse(BaseModel):
units: list[PostboxOrganizationUnitItem]
structures: list[PostboxOrganizationStructureItem] = Field(
default_factory=list
)
structures: list[PostboxOrganizationStructureItem] = Field(default_factory=list)
class PostboxGroupingPayload(BaseModel):