99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from govoplan_core.core.encryption import (
|
|
ContentProtectionRequest,
|
|
ContentUnprotectionRequest,
|
|
ProtectedContent,
|
|
encryption_content_cipher,
|
|
)
|
|
from govoplan_postbox.backend.runtime import get_registry
|
|
|
|
|
|
POSTBOX_PLAINTEXT_PROFILE = "plaintext_v1"
|
|
POSTBOX_SERVER_ENVELOPE_PROFILE = "server_envelope_v1"
|
|
|
|
|
|
class PostboxContentProtectionError(ValueError):
|
|
pass
|
|
|
|
|
|
def protect_message_body(
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
message_id: str,
|
|
vault_id: str,
|
|
plaintext: str,
|
|
actor_id: str,
|
|
operation_ref: str = "v1",
|
|
policy_decision_ref: str = "postbox:configured-server-envelope:v1",
|
|
) -> ProtectedContent:
|
|
capability = encryption_content_cipher(get_registry())
|
|
if capability is None:
|
|
raise PostboxContentProtectionError(
|
|
"Postbox encryption was requested, but Encryption is unavailable."
|
|
)
|
|
try:
|
|
return capability.protect_content(
|
|
session,
|
|
request=ContentProtectionRequest(
|
|
tenant_id=tenant_id,
|
|
owner_module="postbox",
|
|
resource_type="postbox_message_body",
|
|
resource_id=message_id,
|
|
profile_id=POSTBOX_SERVER_ENVELOPE_PROFILE,
|
|
vault_id=vault_id,
|
|
ciphertext_ref=f"postbox-db://messages/{message_id}/body",
|
|
plaintext=plaintext.encode("utf-8"),
|
|
policy_decision_ref=policy_decision_ref,
|
|
idempotency_key=f"postbox-message:{message_id}:body:{operation_ref}",
|
|
actor_id=actor_id,
|
|
metadata={"content_type": "text/plain;charset=utf-8"},
|
|
),
|
|
)
|
|
except Exception as exc:
|
|
raise PostboxContentProtectionError(
|
|
"Postbox message content could not be protected by its configured vault."
|
|
) from exc
|
|
|
|
|
|
def unprotect_message_body(
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
message_id: str,
|
|
envelope_id: str,
|
|
ciphertext: bytes,
|
|
) -> str:
|
|
capability = encryption_content_cipher(get_registry())
|
|
if capability is None:
|
|
raise PostboxContentProtectionError(
|
|
"This Postbox message is encrypted and Encryption is unavailable."
|
|
)
|
|
try:
|
|
plaintext = capability.unprotect_content(
|
|
session,
|
|
request=ContentUnprotectionRequest(
|
|
tenant_id=tenant_id,
|
|
owner_module="postbox",
|
|
resource_type="postbox_message_body",
|
|
resource_id=message_id,
|
|
envelope_id=envelope_id,
|
|
ciphertext=ciphertext,
|
|
),
|
|
)
|
|
return plaintext.decode("utf-8")
|
|
except Exception as exc:
|
|
raise PostboxContentProtectionError(
|
|
"Postbox message content could not be opened with its protection envelope."
|
|
) from exc
|
|
|
|
|
|
__all__ = [
|
|
"POSTBOX_PLAINTEXT_PROFILE",
|
|
"POSTBOX_SERVER_ENVELOPE_PROFILE",
|
|
"PostboxContentProtectionError",
|
|
"protect_message_body",
|
|
"unprotect_message_body",
|
|
]
|