Шифрование полей и всего тела#

Письмо проходит две стадии. Сначала Eazy SDK преобразует subject и body внутри модели, затем JSON-кодек создаёт байты и второй алгоритм шифрует их целиком. HMAC из предыдущей главы видит уже шифротекст, поэтому сервер проверяет ровно те байты, которые получил.

Стадия	Вход	Результат
Преобразование полей	`NewMessage.subject`, `NewMessage.body`	строки с префиксом `mail-field:`
JSON-кодек	модель с преобразованными полями	компактный UTF-8 JSON
Преобразование байтов	сериализованный JSON	байты с префиксом `mail-body:`
Подпись	окончательные байты	`X-Mail-Signature` по шифротексту
Входящий ответ	зашифрованные байты	обычный `SendResult` после обратных стадий

→ Request · POST /api/v1/messages/send

Authorization: Bearer session-1
Content-Type: application/vnd.mail.encrypted+json
X-Mail-Signature: cebf73d9b5258813cd412be5173cf14633cf6cf26812c08ddbfc595cb58ddb1a

mail-body:eyJib2R5IjoibWFpbC1maWVsZDowSnJRdTlHTzBZZlF1Q0RRdjlHQTBMalFzdEMxMExmUmc5R0NJTkN5SU5DLzBZL1JndEM5MExqUmh0R0RMZz09IiwicmVjaXBpZW50IjoiZ3JhY2VAbWFpbC5leGFtcGxlIiwic3ViamVjdCI6Im1haWwtZmllbGQ6MEtIUXRkQzYwWURRdGRHQzBMM1FzTkdQSU5DeTBZSFJndEdBMExYUmg5Q3cifQ==

← Response · 200 OK

Content-Type: application/vnd.mail.encrypted+json

mail-body:eyJzdGF0dXMiOiJvayIsImJvZHkiOnsib3V0Y29tZSI6InNlbnQiLCJtZXNzYWdlX2lkIjoxMDB9fQ==

HTTP

Алгоритмы принадлежат приложению. Учебные реализации кодируют значение в Base64 с различимым префиксом; CryptoContext они принимают по тому же протоколу, что и настоящий алгоритм из KMS или хранилища ключей.

# examples/mail/crypto.py
@dataclass(frozen=True, slots=True)
class DemoFieldCipher:
    name: str = "mail-fields-v1"

    def encrypt(self, value: FrozenValue, *, context: CryptoContext) -> FrozenValue:
        _ = context
        if not isinstance(value, str):
            raise TypeError("the teaching field cipher accepts strings")
        return "mail-field:" + base64.b64encode(value.encode()).decode()

    def decrypt(self, value: FrozenValue, *, context: CryptoContext) -> FrozenValue:
        _ = context
        if not isinstance(value, str) or not value.startswith("mail-field:"):
            raise ValueError("unexpected teaching field ciphertext")
        return base64.b64decode(value.removeprefix("mail-field:")).decode()


@dataclass(frozen=True, slots=True)
class DemoBodyCipher:
    name: str = "mail-body-v1"

    def encrypt(self, value: bytes, *, context: CryptoContext) -> bytes:
        _ = context
        return b"mail-body:" + base64.b64encode(value)

    def decrypt(self, value: bytes, *, context: CryptoContext) -> bytes:
        _ = context
        if not value.startswith(b"mail-body:"):
            raise ValueError("unexpected teaching body ciphertext")
        return base64.b64decode(value.removeprefix(b"mail-body:"))


FIELD_CIPHER = DemoFieldCipher()
BODY_CIPHER = DemoBodyCipher()

Профиль задаёт порядок стадий и пределы размера. encrypt_field(...) получает тип модели и путь, который проверяет типизатор. encrypt_encoded(...) работает позже, уже с сериализованными байтами. Обратное преобразование байтов нужно для зашифрованного ответа.

# examples/mail/http/encryption.py
MAIL_CRYPTO = payload_crypto(
    "mail-send-v1",
    outbound=encrypt_outbound(
        encrypt_field(NewMessage, lambda message: message.subject, using=FIELD_CIPHER),
        encrypt_field(NewMessage, lambda message: message.body, using=FIELD_CIPHER),
        encoded=encrypt_encoded(
            using=BODY_CIPHER,
            max_input_bytes=16_000,
            max_output_bytes=24_000,
        ),
    ),
    inbound=decrypt_inbound(
        encoded=decrypt_encoded(
            using=BODY_CIPHER,
            max_input_bytes=24_000,
            max_output_bytes=16_000,
        )
    ),
)

MAIL_WIRE = http_encrypted(
    content_type=MAIL_ENCRYPTED_CONTENT_TYPE,
    clear_content_type="application/json",
    plaintext_statuses=frozenset({401, 403}),
)

Операция связывает профиль с контрактом представления. Внешний Content-Type принадлежит зашифрованному формату, а clear_content_type говорит декодеру, что после расшифровки лежит JSON. plaintext_statuses оставляет ответы 401 и 403 открытыми: сервер возвращает такие отказы до расшифровки тела. Подпись остаётся той же, но исполняется после обеих исходящих стадий.

# examples/mail/http/encryption.py
@dataclass(frozen=True, slots=True, kw_only=True)
class SendEncryptedMessage(HttpOperation[SendResult]):
    __http__ = Http.post(
        "/api/v1/messages/send",
        success={200: Json(SendEnvelope)},
        errors={200: (Json(RejectedEnvelope), RecipientRejected)},
        security=MAIL_BEARER,
        signing=MAIL_HMAC,
        crypto=MAIL_CRYPTO,
        wire=Wire(encrypted=MAIL_WIRE),
    )

    body: Annotated[NewMessage, JsonBody()]


class EncryptedMailApi(SyncApi):
    base_url = "https://mail.example"

    send = op(SendEncryptedMessage)
    sent = op(FindSent)

После отправки обычный GET ищет письмо по открытой теме. Результат подтверждает, что сайт расшифровал поле перед сохранением, а вызывающий получил обычную модель ответа.

# examples/mail/http/encryption.py
def send_encrypted_message() -> tuple[int, str]:
    identity = Identity(
        auth=(MAIL_BEARER.static("session-1"),),
        key_provider=signing_key,
    )
    subject = "Секретная встреча"
    with mail_client() as client:
        api = EncryptedMailApi(client, identity=identity)
        sent = api.send(
            body=NewMessage(
                recipient="grace@mail.example",
                subject=subject,
                body="Ключи привезут в пятницу.",
            )
        )
        found = api.sent(recipient="grace@mail.example", subject=subject)
    return sent.message_id or 0, found.items[0].subject

Файл операции целиком

"""Encrypt selected mail fields and then the serialized HTTP body."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Annotated

from eazy_sdk import Http, HttpOperation, Identity, SyncApi, op
from eazy_sdk.crypto import (
    decrypt_encoded,
    decrypt_inbound,
    encrypt_encoded,
    encrypt_field,
    encrypt_outbound,
    http_encrypted,
    payload_crypto,
)
from eazy_sdk.request import Wire
from eazy_sdk.request.markers import JsonBody
from eazy_sdk.response import Json

from examples.mail.crypto import (
    BODY_CIPHER,
    FIELD_CIPHER,
    MAIL_ENCRYPTED_CONTENT_TYPE,
)
from examples.mail.http._client import mail_client
from examples.mail.http.send import (
    MAIL_BEARER,
    MAIL_HMAC,
    FindSent,
    NewMessage,
    RecipientRejected,
    RejectedEnvelope,
    SendEnvelope,
    SendResult,
    signing_key,
)


# region docs: mail-encryption-profile
# examples/mail/http/encryption.py
MAIL_CRYPTO = payload_crypto(
    "mail-send-v1",
    outbound=encrypt_outbound(
        encrypt_field(NewMessage, lambda message: message.subject, using=FIELD_CIPHER),
        encrypt_field(NewMessage, lambda message: message.body, using=FIELD_CIPHER),
        encoded=encrypt_encoded(
            using=BODY_CIPHER,
            max_input_bytes=16_000,
            max_output_bytes=24_000,
        ),
    ),
    inbound=decrypt_inbound(
        encoded=decrypt_encoded(
            using=BODY_CIPHER,
            max_input_bytes=24_000,
            max_output_bytes=16_000,
        )
    ),
)

MAIL_WIRE = http_encrypted(
    content_type=MAIL_ENCRYPTED_CONTENT_TYPE,
    clear_content_type="application/json",
    plaintext_statuses=frozenset({401, 403}),
)
# endregion docs: mail-encryption-profile


# region docs: encrypted-mail-operation
# examples/mail/http/encryption.py
@dataclass(frozen=True, slots=True, kw_only=True)
class SendEncryptedMessage(HttpOperation[SendResult]):
    __http__ = Http.post(
        "/api/v1/messages/send",
        success={200: Json(SendEnvelope)},
        errors={200: (Json(RejectedEnvelope), RecipientRejected)},
        security=MAIL_BEARER,
        signing=MAIL_HMAC,
        crypto=MAIL_CRYPTO,
        wire=Wire(encrypted=MAIL_WIRE),
    )

    body: Annotated[NewMessage, JsonBody()]


class EncryptedMailApi(SyncApi):
    base_url = "https://mail.example"

    send = op(SendEncryptedMessage)
    sent = op(FindSent)
# endregion docs: encrypted-mail-operation


# region docs: encrypted-mail-flow
# examples/mail/http/encryption.py
def send_encrypted_message() -> tuple[int, str]:
    identity = Identity(
        auth=(MAIL_BEARER.static("session-1"),),
        key_provider=signing_key,
    )
    subject = "Секретная встреча"
    with mail_client() as client:
        api = EncryptedMailApi(client, identity=identity)
        sent = api.send(
            body=NewMessage(
                recipient="grace@mail.example",
                subject=subject,
                body="Ключи привезут в пятницу.",
            )
        )
        found = api.sent(recipient="grace@mail.example", subject=subject)
    return sent.message_id or 0, found.items[0].subject
# endregion docs: encrypted-mail-flow


def main() -> None:
    message_id, subject = send_encrypted_message()
    print(f"encrypted message: {message_id}")
    print(f"found clear subject: {subject}")


if __name__ == "__main__":
    main()

Запуск: uv run python -m examples.mail.http.encryption.

encrypted message: 100
found clear subject: Секретная встреча

Браузер

JavaScript учебной страницы сам шифрует поля и тело перед fetch и расшифровывает ответ; браузерная операция только заполняет форму и распознаёт три исхода композера.

Предупреждение

Base64 не обеспечивает секретность. Эти обратимые алгоритмы показывают места подключения; в приложении передайте алгоритм с проверенной схемой, управлением ключами и уникальным nonce.

Шифрование полей и всего тела#

Дальше#