From 000a511fe5a6af7a88c3079a75c6b213344c2830 Mon Sep 17 00:00:00 2001 From: waheeda bette Date: Fri, 16 Oct 2026 22:45:51 +0530 Subject: [PATCH] Parse chunk size strictly in Response.parse. Also raise ValueError instead of AssertionError for negative values. --- src/websockets/http11.py | 10 +++++++++- tests/test_http11.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/src/websockets/http11.py b/src/websockets/http11.py index 70870074..60686c55 100644 --- a/src/websockets/http11.py +++ b/src/websockets/http11.py @@ -86,6 +86,12 @@ def d(value: bytes | bytearray) -> str: _content_length_re = re.compile(r"[0-9]+") +# chunk-size = 1*HEXDIG, optionally followed by BWS before chunk extensions. + +# See https://datatracker.ietf.org/doc/html/rfc9112#section-7.1 + +_chunk_size_re = re.compile(rb"[0-9a-fA-F]+") + @dataclasses.dataclass class Request: @@ -467,7 +473,9 @@ def read_body( body = b"" while True: chunk_size_line = yield from parse_line(read_line, SecurityError) - raw_chunk_size = chunk_size_line.split(b";", 1)[0] + raw_chunk_size = chunk_size_line.split(b";", 1)[0].rstrip(b" \t") + if not _chunk_size_re.fullmatch(raw_chunk_size): + raise ValueError(f"invalid chunk size: {d(raw_chunk_size)}") # Set a lower limit than default_max_str_digits; 1 EB is plenty. if len(raw_chunk_size) > 15: str_chunk_size = raw_chunk_size.decode(errors="backslashreplace") diff --git a/tests/test_http11.py b/tests/test_http11.py index 4e7da683..58052714 100644 --- a/tests/test_http11.py +++ b/tests/test_http11.py @@ -367,6 +367,37 @@ def test_parse_body_with_chunked_transfer_encoding_and_chunk_way_too_large(self) "chunk too large: 0x1234567890ABCDEF bytes", ) + def test_parse_body_with_chunked_transfer_encoding_and_chunk_extension(self): + self.reader.feed_data( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + b"6 ;foo=bar\r\nHello \r\n7\r\nworld!\n\r\n0\r\n\r\n" + ) + response = self.assertGeneratorReturns(self.parse()) + self.assertEqual(response.body, b"Hello world!\n") + + def test_parse_body_with_chunked_transfer_encoding_and_negative_size(self): + self.reader.feed_data( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n-1\r\n\r\n" + ) + with self.assertRaises(ValueError) as raised: + next(self.parse()) + self.assertEqual( + str(raised.exception), + "invalid chunk size: -1", + ) + + def test_parse_body_with_chunked_transfer_encoding_and_invalid_size(self): + self.reader.feed_data( + b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n" + b"0x6\r\nHello \r\n0\r\n\r\n" + ) + with self.assertRaises(ValueError) as raised: + next(self.parse()) + self.assertEqual( + str(raised.exception), + "invalid chunk size: 0x6", + ) + def test_parse_body_with_unsupported_transfer_encoding(self): self.reader.feed_data(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: compress\r\n\r\n") with self.assertRaises(NotImplementedError) as raised: