def _read_bytes_from_non_framed_body()

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

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

    def _read_bytes_from_non_framed_body(self, b):
        """Reads the requested number of bytes from a streaming non-framed message body.

        :param int b: Number of bytes to read
        :returns: Decrypted bytes from source stream
        :rtype: bytes
        """
        _LOGGER.debug("starting non-framed body read")
        # Always read the entire message for non-framed message bodies.
        bytes_to_read = self.body_length

        _LOGGER.debug("%d bytes requested; reading %d bytes", b, bytes_to_read)
        ciphertext = self.source_stream.read(bytes_to_read)

        if len(self.output_buffer) + len(ciphertext) < self.body_length:
            raise SerializationError("Total message body contents less than specified in body description")

        if self.verifier is not None:
            self.verifier.update(ciphertext)

        tag = deserialize_tag(stream=self.source_stream, header=self._header, verifier=self.verifier)

        aad_content_string = aws_encryption_sdk.internal.utils.get_aad_content_string(
            content_type=self._header.content_type, is_final_frame=True
        )
        associated_data = assemble_content_aad(
            message_id=self._header.message_id,
            aad_content_string=aad_content_string,
            seq_num=1,
            length=self.body_length,
        )
        self.decryptor = Decryptor(
            algorithm=self._header.algorithm,
            key=self._derived_data_key,
            associated_data=associated_data,
            iv=self._unframed_body_iv,
            tag=tag,
        )

        plaintext = self.decryptor.update(ciphertext)
        plaintext += self.decryptor.finalize()

        self.footer = deserialize_footer(stream=self.source_stream, verifier=self.verifier)
        return plaintext