Parse chunk size strictly in Response.parse

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")

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: