diff --git a/python_multipart/decoders.py b/python_multipart/decoders.py index c4cdfb6..9aaabd9 100644 --- a/python_multipart/decoders.py +++ b/python_multipart/decoders.py @@ -60,6 +60,12 @@ def write(self, data: bytes) -> int: :param data: base64 data to decode """ + # Line breaks and spaces may appear anywhere in Base64 (MIME wraps at + # 76 characters). Drop them before counting, otherwise a chunk + # boundary falling inside a quartet is miscounted and the decode fails. + written = len(data) + data = data.translate(None, b"\r\n \t") + # Prepend any cache info to our data. if len(self.cache) > 0: data = bytes(self.cache) + data @@ -85,7 +91,7 @@ def write(self, data: bytes) -> int: self.cache[:] = b"" # Return the length of the data to indicate no error. - return len(data) + return written def close(self) -> None: """Close this decoder. If the underlying object has a `close()` diff --git a/tests/test_multipart.py b/tests/test_multipart.py index de220eb..dd762dc 100644 --- a/tests/test_multipart.py +++ b/tests/test_multipart.py @@ -1,5 +1,6 @@ from __future__ import annotations +import base64 import os import random import sys @@ -692,6 +693,17 @@ def test_long_bad_split(self) -> None: self.d.write(second) self.assert_data(b"foobar") + def test_line_wrapped_input_any_split(self) -> None: + # MIME-style Base64 is wrapped with line breaks; the result must not + # depend on where the chunk boundaries fall. + raw = bytes(range(100)) + wrapped = base64.encodebytes(raw) + for i in range(len(wrapped) + 1): + self.setUp() + self.assertEqual(self.d.write(wrapped[:i]), i) + self.d.write(wrapped[i:]) + self.assert_data(raw) + def test_close_and_finalize(self) -> None: parser = Mock() f = Base64Decoder(parser)