def _read_bytes_to_framed_body()

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

• 49 lines of code
• 12 McCabe index (conditional complexity)

    def _read_bytes_to_framed_body(self, b):
        """Reads the requested number of bytes from source to a streaming framed message body.

        :param int b: Number of bytes to read
        :returns: Bytes read from source stream, encrypted, and serialized
        :rtype: bytes
        """
        _LOGGER.debug("collecting %d bytes", b)
        _b = b

        if b > 0:
            _frames_to_read = math.ceil(b / float(self.config.frame_length))
            b = int(_frames_to_read * self.config.frame_length)
        _LOGGER.debug("%d bytes requested; reading %d bytes after normalizing to frame length", _b, b)

        plaintext = self.source_stream.read(b)
        plaintext_length = len(plaintext)
        _LOGGER.debug("%d bytes read from source", plaintext_length)

        finalize = False

        if b < 0 or plaintext_length < b:
            _LOGGER.debug("Final plaintext read from source")
            finalize = True

        output = b""
        final_frame_written = False

        while (
            # If not finalizing on this pass, exit when plaintext is exhausted
            (not finalize and plaintext)
            # If finalizing on this pass, wait until final frame is written
            or (finalize and not final_frame_written)
        ):
            current_plaintext_length = len(plaintext)
            is_final_frame = finalize and current_plaintext_length < self.config.frame_length
            bytes_in_frame = min(current_plaintext_length, self.config.frame_length)
            _LOGGER.debug(
                "Writing %d bytes into%s frame %d",
                bytes_in_frame,
                " final" if is_final_frame else "",
                self.sequence_number,
            )
            self._bytes_encrypted += bytes_in_frame
            ciphertext, plaintext = serialize_frame(
                algorithm=self._encryption_materials.algorithm,
                plaintext=plaintext,
                message_id=self._header.message_id,
                data_encryption_key=self._derived_data_key,
                frame_length=self.config.frame_length,
                sequence_number=self.sequence_number,
                is_final_frame=is_final_frame,
                signer=self.signer,
            )
            final_frame_written = is_final_frame
            output += ciphertext
            self.sequence_number += 1

        if finalize:
            _LOGGER.debug("Writing footer")
            if self.signer is not None:
                output += serialize_footer(self.signer)
            self.__message_complete = True
        return output