912 lines
33 KiB
Python
912 lines
33 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any, Literal
|
|
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
from govoplan_postbox.backend.grouping_policies import PostboxGroupingPolicyMode
|
|
from govoplan_postbox.backend.protection_profiles import (
|
|
POSTBOX_MANAGED_ENVELOPE_PROFILE,
|
|
POSTBOX_PLAINTEXT_PROFILE,
|
|
PostboxProtectionProfile,
|
|
)
|
|
|
|
|
|
PostboxClassification = Literal[
|
|
"public",
|
|
"internal",
|
|
"confidential",
|
|
"restricted",
|
|
]
|
|
|
|
|
|
class PostboxAccessDecisionResponse(BaseModel):
|
|
allowed: bool
|
|
action: str
|
|
postbox_id: str
|
|
reason_code: str
|
|
explanation: str
|
|
organization_unit_id: str | None = None
|
|
function_id: str | None = None
|
|
assignment_ids: list[str] = Field(default_factory=list)
|
|
assignment_sources: list[str] = Field(default_factory=list)
|
|
selected_assignment_id: str | None = None
|
|
holder_count: int = 0
|
|
vacant: bool = True
|
|
classification: str = "internal"
|
|
classification_allowed: bool = True
|
|
binding_status: str = "active"
|
|
|
|
|
|
class PostboxDirectoryItem(BaseModel):
|
|
id: str
|
|
tenant_id: str
|
|
address: str
|
|
address_key: str
|
|
name: str
|
|
status: str
|
|
classification: str
|
|
organization_unit_id: str | None = None
|
|
organization_unit_name: str | None = None
|
|
function_id: str | None = None
|
|
function_name: str | None = None
|
|
context_key: str | None = None
|
|
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)
|
|
grouping_policy: dict[str, Any] = Field(default_factory=dict)
|
|
access: PostboxAccessDecisionResponse | None = None
|
|
resource_revision: int = Field(default=1, ge=1)
|
|
etag: str | None = None
|
|
|
|
|
|
class PostboxDirectoryResponse(BaseModel):
|
|
postboxes: list[PostboxDirectoryItem]
|
|
|
|
|
|
class PostboxParticipantPayload(BaseModel):
|
|
kind: str = Field(min_length=1, max_length=30)
|
|
reference_type: str = Field(min_length=1, max_length=50)
|
|
reference_id: str | None = Field(default=None, max_length=255)
|
|
label: str | None = Field(default=None, max_length=500)
|
|
address: str | None = Field(default=None, max_length=500)
|
|
|
|
|
|
class PostboxAttachmentPayload(BaseModel):
|
|
reference_type: str = Field(min_length=1, max_length=50)
|
|
reference_id: str = Field(min_length=1, max_length=255)
|
|
name: str | None = Field(default=None, max_length=1000)
|
|
media_type: str | None = Field(default=None, max_length=255)
|
|
size_bytes: int | None = Field(default=None, ge=0)
|
|
digest: str | None = Field(default=None, max_length=255)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PostboxAttachmentResolutionItem(PostboxAttachmentPayload):
|
|
available: bool = False
|
|
reason_code: str
|
|
file_asset_id: str | None = None
|
|
file_version_id: str | None = None
|
|
download_path: str | None = None
|
|
provenance: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PostboxAttachmentResolutionResponse(BaseModel):
|
|
attachments: list[PostboxAttachmentResolutionItem] = Field(default_factory=list)
|
|
|
|
|
|
class PostboxWrappedKeyPayload(BaseModel):
|
|
recipient_type: str = Field(min_length=1, max_length=50)
|
|
recipient_id: str = Field(min_length=1, max_length=255)
|
|
key_epoch: int = Field(ge=1)
|
|
wrapped_key_ref: str = Field(min_length=1, max_length=2000)
|
|
algorithm: str | None = Field(default=None, max_length=100)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PostboxExternalRecipientTokenPayload(BaseModel):
|
|
token_id: str = Field(min_length=1, max_length=255)
|
|
state: Literal["pending", "available", "fetched", "expired", "revoked"]
|
|
expires_at: datetime | None = None
|
|
one_time: bool = False
|
|
key_fetched_at: datetime | None = None
|
|
revoked_at: datetime | None = None
|
|
assurance_profile: str | None = Field(default=None, max_length=100)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PostboxMessageItem(BaseModel):
|
|
id: str
|
|
tenant_id: str
|
|
postbox_id: str
|
|
subject: str
|
|
body_text: str | None = None
|
|
status: str
|
|
availability: Literal["available", "withdrawn", "expired"]
|
|
classification: str
|
|
sender_label: str | None = None
|
|
delivered_at: datetime
|
|
read_at: datetime | None = None
|
|
acknowledged_at: datetime | None = None
|
|
expires_at: datetime | None = None
|
|
withdrawn_at: datetime | None = None
|
|
producer_module: str | None = None
|
|
producer_resource_type: str | None = None
|
|
producer_resource_id: str | None = None
|
|
in_reply_to_message_id: str | None = None
|
|
replaces_message_id: str | None = None
|
|
encryption_profile: str
|
|
key_epoch: int
|
|
ciphertext_ref: str | None = None
|
|
signed_manifest_ref: str | None = None
|
|
wrapped_keys: list[PostboxWrappedKeyPayload] = Field(default_factory=list)
|
|
external_recipient_tokens: list[PostboxExternalRecipientTokenPayload] = Field(
|
|
default_factory=list
|
|
)
|
|
participants: list[PostboxParticipantPayload] = Field(default_factory=list)
|
|
attachments: list[PostboxAttachmentPayload] = Field(default_factory=list)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PostboxMessageListResponse(BaseModel):
|
|
messages: list[PostboxMessageItem]
|
|
total: int
|
|
limit: int
|
|
offset: int
|
|
|
|
|
|
class PostboxMessageStateRequest(BaseModel):
|
|
state: Literal["read", "acknowledged"]
|
|
|
|
|
|
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)
|
|
|
|
|
|
class PostboxTargetPayload(BaseModel):
|
|
postbox_id: str | None = Field(default=None, max_length=36)
|
|
address_key: str | None = Field(default=None, max_length=500)
|
|
template_id: str | None = Field(default=None, max_length=36)
|
|
organization_unit_id: str | None = Field(default=None, max_length=36)
|
|
function_id: str | None = Field(default=None, max_length=36)
|
|
context_key: str | None = Field(default=None, max_length=255)
|
|
|
|
@model_validator(mode="after")
|
|
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
|
|
)
|
|
if direct == templated:
|
|
raise ValueError(
|
|
"Specify one direct Postbox target or one complete template target."
|
|
)
|
|
return self
|
|
|
|
|
|
class PostboxDeliveryCreateRequest(BaseModel):
|
|
target: PostboxTargetPayload
|
|
producer_module: str = Field(min_length=1, max_length=100)
|
|
producer_resource_type: str = Field(min_length=1, max_length=100)
|
|
producer_resource_id: str | None = Field(default=None, max_length=255)
|
|
idempotency_key: str = Field(min_length=1, max_length=255)
|
|
subject: str = Field(min_length=1, max_length=1000)
|
|
body_text: str | None = None
|
|
sender_label: str | None = Field(default=None, max_length=500)
|
|
classification: PostboxClassification = "internal"
|
|
action_required: bool = False
|
|
participants: list[PostboxParticipantPayload] = Field(default_factory=list)
|
|
attachments: list[PostboxAttachmentPayload] = Field(default_factory=list)
|
|
expires_at: datetime | 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)
|
|
external_recipient_tokens: list[PostboxExternalRecipientTokenPayload] = Field(
|
|
default_factory=list
|
|
)
|
|
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
|
|
postbox_id: str
|
|
message_id: str
|
|
address: str
|
|
status: str
|
|
vacant: bool
|
|
holder_count: int
|
|
duplicate: bool = False
|
|
evidence: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class PostboxLinkedCopyPolicyPayload(BaseModel):
|
|
enabled: bool = False
|
|
structure_id: str | None = Field(default=None, max_length=36)
|
|
relation_type_ids: list[str] = Field(default_factory=list, max_length=20)
|
|
max_depth: int = Field(default=1, ge=1, le=20)
|
|
stop_unit_id: str | None = Field(default=None, max_length=36)
|
|
stop_unit_type_id: str | None = Field(default=None, max_length=36)
|
|
target_function_type_id: str | None = Field(default=None, max_length=36)
|
|
target_template_id: str | None = Field(default=None, max_length=36)
|
|
fanout: Literal["nearest", "all"] = "nearest"
|
|
allowed_classifications: list[PostboxClassification] = Field(
|
|
default_factory=lambda: ["internal"],
|
|
max_length=20,
|
|
)
|
|
allowed_producer_modules: list[str] = Field(
|
|
default_factory=list,
|
|
max_length=50,
|
|
)
|
|
require_expiry: bool = False
|
|
max_retention_days: int | None = Field(default=None, ge=1, le=36500)
|
|
|
|
@model_validator(mode="after")
|
|
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)
|
|
)
|
|
self.allowed_producer_modules = list(
|
|
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.")
|
|
if any(not value for value in self.allowed_classifications):
|
|
raise ValueError("Allowed classifications must not be empty.")
|
|
if any(not value for value in self.allowed_producer_modules):
|
|
raise ValueError("Allowed producer modules must not be empty.")
|
|
if self.enabled and not all(
|
|
(
|
|
self.structure_id,
|
|
self.target_function_type_id,
|
|
self.target_template_id,
|
|
self.allowed_classifications,
|
|
self.allowed_producer_modules,
|
|
)
|
|
):
|
|
raise ValueError(
|
|
"Enabled hierarchy copy requires a structure, target function "
|
|
"type, target template, classification gate, and producer allowlist."
|
|
)
|
|
return self
|
|
|
|
|
|
class PostboxAttentionPolicyPayload(BaseModel):
|
|
mode: Literal["none", "vacancy_escalation"] = "none"
|
|
delay_minutes: int | None = Field(default=None, ge=1, le=43200)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_delay(self) -> "PostboxAttentionPolicyPayload":
|
|
if self.mode == "vacancy_escalation" and self.delay_minutes is None:
|
|
raise ValueError("Vacancy escalation requires a delay.")
|
|
if self.mode == "none":
|
|
self.delay_minutes = None
|
|
return self
|
|
|
|
|
|
class PostboxSharedVisibilityPolicyPayload(BaseModel):
|
|
mode: Literal["none"] = "none"
|
|
|
|
|
|
class PostboxRoutingPolicyPayload(BaseModel):
|
|
linked_copy: PostboxLinkedCopyPolicyPayload = Field(
|
|
default_factory=PostboxLinkedCopyPolicyPayload
|
|
)
|
|
attention: PostboxAttentionPolicyPayload = Field(
|
|
default_factory=PostboxAttentionPolicyPayload
|
|
)
|
|
shared_visibility: PostboxSharedVisibilityPolicyPayload = Field(
|
|
default_factory=PostboxSharedVisibilityPolicyPayload
|
|
)
|
|
|
|
@model_validator(mode="before")
|
|
@classmethod
|
|
def normalize_legacy_policy(cls, value: Any) -> Any:
|
|
if value in (None, {}, {"mode": "none"}):
|
|
return {}
|
|
return value
|
|
|
|
@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"
|
|
):
|
|
raise ValueError("Vacancy escalation requires nearest linked-copy routing.")
|
|
return self
|
|
|
|
|
|
class PostboxRoutePreviewTarget(BaseModel):
|
|
depth: int
|
|
organization_unit_id: str
|
|
organization_unit_name: str
|
|
function_id: str | None = None
|
|
function_name: str | None = None
|
|
target_postbox_id: str | None = None
|
|
target_address: str | None = None
|
|
status: str
|
|
vacant: bool = True
|
|
holder_count: int = 0
|
|
path: list[dict[str, Any]] = Field(default_factory=list)
|
|
diagnostics: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class PostboxRouteDryRunRequest(BaseModel):
|
|
target: PostboxTargetPayload
|
|
producer_module: str = Field(min_length=1, max_length=100)
|
|
classification: PostboxClassification = "internal"
|
|
expires_at: datetime | None = None
|
|
|
|
|
|
class PostboxRouteDryRunResponse(BaseModel):
|
|
status: str
|
|
source_postbox_id: str | None = None
|
|
policy: PostboxRoutingPolicyPayload = Field(
|
|
default_factory=PostboxRoutingPolicyPayload
|
|
)
|
|
routes: list[PostboxRoutePreviewTarget] = Field(default_factory=list)
|
|
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 PostboxGroupingPolicyPayload(BaseModel):
|
|
mode: PostboxGroupingPolicyMode = "allow"
|
|
reason: str | None = Field(default=None, max_length=1000)
|
|
|
|
@model_validator(mode="after")
|
|
def normalize_reason(self) -> "PostboxGroupingPolicyPayload":
|
|
self.reason = self.reason.strip() if self.reason else None
|
|
return self
|
|
|
|
|
|
class PostboxExactCreateRequest(BaseModel):
|
|
name: str = Field(min_length=1, max_length=500)
|
|
description: str | None = None
|
|
organization_unit_id: str = Field(min_length=1, max_length=36)
|
|
function_id: str = Field(min_length=1, max_length=36)
|
|
address_key: str | None = Field(default=None, max_length=120)
|
|
classification: PostboxClassification = "internal"
|
|
portal_visible: bool = False
|
|
encryption_profile: PostboxProtectionProfile = POSTBOX_PLAINTEXT_PROFILE
|
|
encryption_vault_id: str | None = Field(default=None, max_length=255)
|
|
protection_policy: PostboxProtectionPolicyPayload = Field(
|
|
default_factory=PostboxProtectionPolicyPayload
|
|
)
|
|
grouping_policy: PostboxGroupingPolicyPayload = Field(
|
|
default_factory=PostboxGroupingPolicyPayload
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_encryption(self) -> "PostboxExactCreateRequest":
|
|
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(
|
|
"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 PostboxGroupingPolicyUpdateRequest(BaseModel):
|
|
base_revision: int = Field(ge=1)
|
|
grouping_policy: PostboxGroupingPolicyPayload
|
|
|
|
|
|
class PostboxTemplateRevisionPayload(BaseModel):
|
|
function_type_id: str | None = Field(default=None, max_length=36)
|
|
scope_kind: Literal["tenant", "unit", "subtree", "unit_type"] = "tenant"
|
|
scope_id: str | None = Field(default=None, max_length=255)
|
|
scope_structure_id: str | None = Field(default=None, max_length=36)
|
|
scope_relation_type_ids: list[str] = Field(default_factory=list, max_length=20)
|
|
name_pattern: str = Field(
|
|
default="{unit_name} / {function_name}",
|
|
min_length=1,
|
|
max_length=500,
|
|
)
|
|
address_pattern: str = Field(
|
|
default="{template_slug}.{unit_slug}.{function_slug}",
|
|
min_length=1,
|
|
max_length=500,
|
|
)
|
|
classification: PostboxClassification = "internal"
|
|
allow_vacant_delivery: bool = True
|
|
portal_visible: bool = False
|
|
encryption_profile: PostboxProtectionProfile = POSTBOX_PLAINTEXT_PROFILE
|
|
encryption_vault_id: str | None = Field(default=None, max_length=255)
|
|
protection_policy: PostboxProtectionPolicyPayload = Field(
|
|
default_factory=PostboxProtectionPolicyPayload
|
|
)
|
|
grouping_policy: PostboxGroupingPolicyPayload = Field(
|
|
default_factory=PostboxGroupingPolicyPayload
|
|
)
|
|
routing_policy: PostboxRoutingPolicyPayload = Field(
|
|
default_factory=PostboxRoutingPolicyPayload
|
|
)
|
|
|
|
@model_validator(mode="after")
|
|
def normalize_scope(self) -> "PostboxTemplateRevisionPayload":
|
|
self.scope_relation_type_ids = list(
|
|
dict.fromkeys(
|
|
value.strip() for value in self.scope_relation_type_ids if value.strip()
|
|
)
|
|
)
|
|
return self
|
|
|
|
@model_validator(mode="after")
|
|
def validate_encryption(self) -> "PostboxTemplateRevisionPayload":
|
|
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(
|
|
"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:
|
|
if payload.scope_kind == "subtree" and not payload.scope_structure_id:
|
|
raise ValueError("A subtree scope requires an organization structure.")
|
|
if payload.scope_kind != "subtree" and (
|
|
payload.scope_structure_id or payload.scope_relation_type_ids
|
|
):
|
|
raise ValueError(
|
|
"Hierarchy structure and relation filters apply only to subtree scopes."
|
|
)
|
|
|
|
|
|
class PostboxTemplateCreateRequest(PostboxTemplateRevisionPayload):
|
|
slug: str = Field(min_length=1, max_length=120)
|
|
name: str = Field(min_length=1, max_length=250)
|
|
description: str | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def validate_write_scope(self) -> "PostboxTemplateCreateRequest":
|
|
_validate_template_write_scope(self)
|
|
return self
|
|
|
|
|
|
class PostboxTemplatePreviewRequest(PostboxTemplateCreateRequest):
|
|
template_id: str | None = Field(default=None, max_length=36)
|
|
context_key: str | None = Field(default=None, max_length=255)
|
|
limit: int = Field(default=200, ge=1, le=500)
|
|
|
|
|
|
class PostboxTemplatePreviewTarget(BaseModel):
|
|
organization_unit_id: str
|
|
organization_unit_name: str
|
|
function_id: str
|
|
function_name: str
|
|
address: str
|
|
name: str
|
|
holder_count: int = Field(ge=0)
|
|
vacant: bool
|
|
status: str
|
|
existing_postbox_id: str | None = None
|
|
diagnostics: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class PostboxTemplatePreviewResponse(BaseModel):
|
|
targets: list[PostboxTemplatePreviewTarget] = Field(default_factory=list)
|
|
total: int = Field(ge=0)
|
|
ready_count: int = Field(ge=0)
|
|
existing_count: int = Field(ge=0)
|
|
vacant_count: int = Field(ge=0)
|
|
blocked_count: int = Field(ge=0)
|
|
truncated: bool = False
|
|
diagnostics: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class PostboxTemplateReviseRequest(PostboxTemplateRevisionPayload):
|
|
base_revision: int = Field(ge=1)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_write_scope(self) -> "PostboxTemplateReviseRequest":
|
|
_validate_template_write_scope(self)
|
|
return self
|
|
|
|
|
|
class PostboxTemplateRevisionItem(PostboxTemplateRevisionPayload):
|
|
id: str
|
|
revision: int
|
|
history_policy: dict[str, Any] = Field(default_factory=dict)
|
|
retention_policy: dict[str, Any] = Field(default_factory=dict)
|
|
published_at: datetime | None = None
|
|
created_at: datetime
|
|
|
|
|
|
class PostboxTemplateItem(BaseModel):
|
|
id: str
|
|
tenant_id: str
|
|
slug: str
|
|
name: str
|
|
description: str | None = None
|
|
status: str
|
|
current_revision: int
|
|
resource_revision: int = Field(ge=1)
|
|
etag: str
|
|
published_revision_id: str | None = None
|
|
revisions: list[PostboxTemplateRevisionItem] = Field(default_factory=list)
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class PostboxTemplateListResponse(BaseModel):
|
|
templates: list[PostboxTemplateItem]
|
|
|
|
|
|
class PostboxTemplatePublishRequest(BaseModel):
|
|
revision: int | None = Field(default=None, ge=1)
|
|
base_revision: int = Field(ge=1)
|
|
|
|
|
|
class PostboxMutationRequest(BaseModel):
|
|
base_revision: int = Field(ge=1)
|
|
|
|
|
|
class PostboxMaterializeRequest(BaseModel):
|
|
organization_unit_id: str = Field(min_length=1, max_length=36)
|
|
function_id: str = Field(min_length=1, max_length=36)
|
|
context_key: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class PostboxOrganizationFunctionItem(BaseModel):
|
|
id: str
|
|
slug: str
|
|
name: str
|
|
function_type_id: str | None = None
|
|
delegable: bool = False
|
|
act_in_place_allowed: bool = False
|
|
|
|
|
|
class PostboxOrganizationUnitItem(BaseModel):
|
|
id: str
|
|
slug: str
|
|
name: str
|
|
unit_type_id: str | None = None
|
|
parent_id: str | None = None
|
|
functions: list[PostboxOrganizationFunctionItem] = Field(default_factory=list)
|
|
|
|
|
|
class PostboxOrganizationRelationTypeItem(BaseModel):
|
|
id: str
|
|
slug: str
|
|
name: str
|
|
structure_id: str | None = None
|
|
is_hierarchical: bool = True
|
|
status: str = "active"
|
|
|
|
|
|
class PostboxOrganizationStructureItem(BaseModel):
|
|
id: str
|
|
slug: str
|
|
name: str
|
|
structure_kind: str
|
|
status: str = "active"
|
|
relation_types: list[PostboxOrganizationRelationTypeItem] = Field(
|
|
default_factory=list
|
|
)
|
|
|
|
|
|
class PostboxOrganizationTargetsResponse(BaseModel):
|
|
units: list[PostboxOrganizationUnitItem]
|
|
structures: list[PostboxOrganizationStructureItem] = Field(default_factory=list)
|
|
|
|
|
|
class PostboxGroupingPayload(BaseModel):
|
|
name: str = Field(min_length=1, max_length=250)
|
|
is_default: bool = False
|
|
postbox_ids: list[str] = Field(default_factory=list, max_length=250)
|
|
|
|
|
|
class PostboxGroupingUpdateRequest(PostboxGroupingPayload):
|
|
base_revision: int = Field(ge=1)
|
|
|
|
|
|
class PostboxGroupingConstraintItem(BaseModel):
|
|
code: Literal[
|
|
"source_requires_separation",
|
|
"classification_separation_required",
|
|
]
|
|
mode: PostboxGroupingPolicyMode
|
|
postbox_id: str
|
|
reason: str | None = None
|
|
enforced_by: Literal["postbox_configuration"] = "postbox_configuration"
|
|
|
|
|
|
class PostboxGroupingItem(PostboxGroupingPayload):
|
|
id: str
|
|
total_count: int = Field(default=0, ge=0)
|
|
unread_count: int = Field(default=0, ge=0)
|
|
constraints: list[PostboxGroupingConstraintItem] = Field(default_factory=list)
|
|
resource_revision: int = Field(ge=1)
|
|
etag: str
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
class PostboxGroupingListResponse(BaseModel):
|
|
groupings: list[PostboxGroupingItem]
|