diff --git a/python_multipart/decoders.py b/python_multipart/decoders.py index c4cdfb6..e79452a 100644 --- a/python_multipart/decoders.py +++ b/python_multipart/decoders.py @@ -142,8 +142,9 @@ def write(self, data: bytes) -> int: # If the last 2 characters have an '=' sign in it, then we won't be # able to decode the encoded value and we'll need to save it for the # next decoding step. - if data[-2:].find(b"=") != -1: - enc, rest = data[:-2], data[-2:] + escape_pos = data.find(b"=", max(0, len(data) - 2)) + if escape_pos != -1: + enc, rest = data[:escape_pos], data[escape_pos:] else: enc = data rest = b"" diff --git a/tests/test_multipart.py b/tests/test_multipart.py index de220eb..9d73c2b 100644 --- a/tests/test_multipart.py +++ b/tests/test_multipart.py @@ -754,6 +754,22 @@ def test_with_split_newline_escape_2(self) -> None: self.d.write(b"\r\nbar") self.assert_data(b"foobar") + def test_adjacent_escapes_across_chunk_splits(self) -> None: + for encoded, expected in ( + (b"=3D=3D", b"=="), + (b"foo=3D=3Dbar", b"foo==bar"), + (b"foo=3D=\r\nbar", b"foo=bar"), + (b"foo=3D=\nbar", b"foo=bar"), + ): + for split in range(len(encoded) + 1): + with self.subTest(encoded=encoded, split=split): + output = BytesIO() + decoder = QuotedPrintableDecoder(output) + decoder.write(encoded[:split]) + decoder.write(encoded[split:]) + decoder.finalize() + self.assertEqual(output.getvalue(), expected) + def test_close_and_finalize(self) -> None: parser = Mock() f = QuotedPrintableDecoder(parser) @@ -1456,6 +1472,32 @@ def test_bad_content_disposition(self) -> None: f.write(data) f.finalize() + def test_quoted_printable_adjacent_escapes_across_chunk_splits(self) -> None: + cases = ((b"=3D=3D", b"=="), (b"=3D=\r\n", b"="), (b"=3D=\n", b"=")) + for is_file in (False, True): + filename = b'; filename="test.txt"' if is_file else b"" + head = ( + b'--boundary\r\nContent-Disposition: form-data; name="upload"' + + filename + + b"\r\nContent-Transfer-Encoding: quoted-printable\r\n\r\n" + ) + for encoded, expected in cases: + for split in range(len(encoded) + 1): + with self.subTest(is_file=is_file, encoded=encoded, split=split): + self.make("boundary") + try: + self.f.write(head + encoded[:split]) + self.f.write(encoded[split:] + b"\r\n--boundary--\r\n") + self.f.finalize() + if is_file: + self.assert_file(b"upload", b"test.txt", None, expected) + else: + self.assert_field(b"upload", expected) + self.assertTrue(self.ended) + finally: + for file in self.files: + file.close() + def test_handles_None_fields(self) -> None: fields: list[Field] = []