def _serialize_header_v1()

in src/aws_encryption_sdk/internal/formatting/serialize.py [0:0]

• 44 lines of code
• 3 McCabe index (conditional complexity)

def _serialize_header_v1(header, signer=None):
    """Serializes a header object for messages with SerializationVersion.V1.

    :param header: Header to serialize
    :type header: aws_encryption_sdk.structures.MessageHeader
    :param signer: Cryptographic signer object (optional)
    :type signer: aws_encryption_sdk.internal.crypto.Signer
    :returns: Serialized header
    :rtype: bytes
    """
    ec_serialized = aws_encryption_sdk.internal.formatting.encryption_context.serialize_encryption_context(
        header.encryption_context
    )
    header_start_format = (
        ">"  # big endian
        "B"  # version
        "B"  # type
        "H"  # algorithm ID
        "16s"  # message ID
        "H"  # encryption context length
        "{}s"  # serialized encryption context
    ).format(len(ec_serialized))
    header_bytes = bytearray()
    header_bytes.extend(
        struct.pack(
            header_start_format,
            header.version.value,
            header.type.value,
            header.algorithm.algorithm_id,
            header.message_id,
            len(ec_serialized),
            ec_serialized,
        )
    )

    serialized_data_keys = bytearray()
    for data_key in header.encrypted_data_keys:
        serialized_data_keys.extend(serialize_encrypted_data_key(data_key))

    header_bytes.extend(struct.pack(">H", len(header.encrypted_data_keys)))
    header_bytes.extend(serialized_data_keys)

    header_close_format = (
        ">"  # big endian
        "B"  # content type (no framing vs framing)
        "4x"  # reserved (formerly content AAD length)
        "B"  # nonce/IV length, this applies to all IVs in this message
        "I"  # frame length
    )
    header_bytes.extend(
        struct.pack(header_close_format, header.content_type.value, header.algorithm.iv_len, header.frame_length)
    )
    output = bytes(header_bytes)
    if signer is not None:
        signer.update(output)
    return output