def generate_header()

in src/aws_encryption_sdk/streaming_client.py [0:0]

• 25 lines of code
• 4 McCabe index (conditional complexity)

    def generate_header(self, message_id):
        """Generates the header object.

        :param message_id: The randomly generated id for the message
        :type message_id: bytes
        """
        version = VERSION
        if self._encryption_materials.algorithm.message_format_version == 0x02:
            version = SerializationVersion.V2

        kwargs = dict(
            version=version,
            algorithm=self._encryption_materials.algorithm,
            message_id=message_id,
            encryption_context=self._encryption_materials.encryption_context,
            encrypted_data_keys=self._encryption_materials.encrypted_data_keys,
            content_type=self.content_type,
            frame_length=self.config.frame_length,
        )

        if self._encryption_materials.algorithm.is_committing():
            commitment_key = calculate_commitment_key(
                source_key=self._encryption_materials.data_encryption_key.data_key,
                algorithm=self._encryption_materials.algorithm,
                message_id=message_id,
            )
            kwargs["commitment_key"] = commitment_key

        if version == SerializationVersion.V1:
            kwargs["type"] = TYPE
            kwargs["content_aad_length"] = 0
            kwargs["header_iv_length"] = self._encryption_materials.algorithm.iv_len

        return MessageHeader(**kwargs)