def _prep_message()

in src/aws_encryption_sdk/streaming_client.py [0:0]
58 lines of code
10 McCabe index (conditional complexity)

    def _prep_message(self):
        """Performs initial message setup.

        :raises MasterKeyProviderError: if primary master key is not a member of supplied MasterKeyProvider
        :raises MasterKeyProviderError: if no Master Keys are returned from key_provider
        """
        validate_commitment_policy_on_encrypt(self.config.commitment_policy, self.config.algorithm)

        try:
            plaintext_length = self.stream_length
        except NotSupportedError:
            plaintext_length = None
        encryption_materials_request = EncryptionMaterialsRequest(
            algorithm=self.config.algorithm,
            encryption_context=self.config.encryption_context.copy(),
            frame_length=self.config.frame_length,
            plaintext_rostream=aws_encryption_sdk.internal.utils.streams.ROStream(self.source_stream),
            plaintext_length=plaintext_length,
            commitment_policy=self.config.commitment_policy,
        )
        self._encryption_materials = self.config.materials_manager.get_encryption_materials(
            request=encryption_materials_request
        )

        if self.config.algorithm is not None and self._encryption_materials.algorithm != self.config.algorithm:
            raise ActionNotAllowedError(
                (
                    "Cryptographic materials manager provided algorithm suite"
                    " differs from algorithm suite in request.\n"
                    "Required: {requested}\n"
                    "Provided: {provided}"
                ).format(requested=self.config.algorithm, provided=self._encryption_materials.algorithm)
            )

        num_keys = len(self._encryption_materials.encrypted_data_keys)
        if self.config.max_encrypted_data_keys and num_keys > self.config.max_encrypted_data_keys:
            raise MaxEncryptedDataKeysExceeded(num_keys, self.config.max_encrypted_data_keys)

        if self._encryption_materials.signing_key is None:
            self.signer = None
        else:
            # MPL verification key is PEM bytes, not DER bytes.
            # If the underlying CMM is from the MPL, load PEM bytes.
            if (_HAS_MPL
                    and isinstance(self.config.materials_manager, CryptoMaterialsManagerFromMPL)):
                self.signer = Signer.from_key_bytes(
                    algorithm=self._encryption_materials.algorithm, key_bytes=self._encryption_materials.signing_key,
                    encoding=serialization.Encoding.PEM,
                )
            else:
                self.signer = Signer.from_key_bytes(
                    algorithm=self._encryption_materials.algorithm, key_bytes=self._encryption_materials.signing_key
                )
        aws_encryption_sdk.internal.utils.validate_frame_length(
            frame_length=self.config.frame_length, algorithm=self._encryption_materials.algorithm
        )

        message_id = aws_encryption_sdk.internal.utils.message_id(
            self._encryption_materials.algorithm.message_id_length()
        )

        self._derived_data_key = derive_data_encryption_key(
            source_key=self._encryption_materials.data_encryption_key.data_key,
            algorithm=self._encryption_materials.algorithm,
            message_id=message_id,
        )

        self._header = self.generate_header(message_id)

        self._write_header()
        if self.content_type == ContentType.NO_FRAMING:
            self._prep_non_framed()
        self._message_prepped = True