Parse integers strictly in the HTTP protocol.

This is a follow-up to 0437fbc, covering all similar cases (except in
the legacy implementation) and simplifying checks for invalid values.

Author: aaugustin

src/websockets/http11.py

@@ -80,11 +80,18 @@ def d(value: bytes | bytearray) -> str:
 
 _value_re = re.compile(rb"[\x09\x20-\x7e\x80-\xff]*")
 
-# Content-Length = 1*DIGIT, applied after decoding the header value.
+# Regexes for validating status codes and numeric header values.
+# Set a lower limit than default_max_str_digits; 1 EB is plenty.
 
+# Content-Length = 1*DIGIT
 # See https://datatracker.ietf.org/doc/html/rfc9110#section-8.6
 
-_content_length_re = re.compile(r"[0-9]+")
+_digits_re = re.compile(rb"[0-9]{1,18}")
+
+# chunk-size = 1*HEXDIG
+# See https://datatracker.ietf.org/doc/html/rfc9112#section-7.1
+
+_hexdigits_re = re.compile(rb"[0-9a-fA-F]{1,15}")
 
 
 @dataclasses.dataclass
@@ -304,12 +311,11 @@ def parse(
                 raise ValueError(
                     f"unsupported protocol; expected HTTP/1.1: {d(status_line)}"
                 )
-        try:
-            status_code = int(raw_status_code)
-        except ValueError:  # invalid literal for int() with base 10
+        if not _digits_re.fullmatch(raw_status_code):
             raise ValueError(
                 f"invalid status code; expected integer; got {d(raw_status_code)}"
-            ) from None
+            )
+        status_code = int(raw_status_code)
         if not 100 <= status_code < 600:
             raise ValueError(
                 f"invalid status code; expected 100–599; got {d(raw_status_code)}"
@@ -468,10 +474,8 @@ def read_body(
         while True:
             chunk_size_line = yield from parse_line(read_line, SecurityError)
             raw_chunk_size = chunk_size_line.split(b";", 1)[0]
-            # 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")
-                raise SecurityError(f"chunk too large: 0x{str_chunk_size} bytes")
+            if not _hexdigits_re.fullmatch(raw_chunk_size):
+                raise SecurityError(f"invalid chunk size: {d(raw_chunk_size)}")
             chunk_size = int(raw_chunk_size, 16)
             if chunk_size == 0:
                 break
@@ -487,11 +491,8 @@ def read_body(
         return body
 
     elif (raw_content_length := headers.get("Content-Length")) is not None:
-        if not _content_length_re.fullmatch(raw_content_length):
-            raise ValueError(f"invalid content length: {raw_content_length}")
-        # Set a lower limit than default_max_str_digits; 1 EiB is plenty.
-        if len(raw_content_length) > 18:
-            raise SecurityError(f"body too large: {raw_content_length} bytes")
+        if not _digits_re.fullmatch(raw_content_length.encode("iso-8859-1")):
+            raise SecurityError(f"invalid content length: {raw_content_length}")
         content_length = int(raw_content_length)
         if content_length > MAX_BODY_SIZE:
             raise SecurityError(f"body too large: {content_length} bytes")

tests/test_http11.py

@@ -311,12 +311,12 @@ def test_parse_body_with_content_length_and_body_way_too_large(self):
             next(self.parse())
         self.assertEqual(
             str(raised.exception),
-            "body too large: 1234567890123456789 bytes",
+            "invalid content length: 1234567890123456789",
         )
 
     def test_parse_body_with_negative_content_length(self):
         self.reader.feed_data(b"HTTP/1.1 200 OK\r\nContent-Length: -1\r\n\r\n")
-        with self.assertRaises(ValueError) as raised:
+        with self.assertRaises(SecurityError) as raised:
             next(self.parse())
         self.assertEqual(
             str(raised.exception),
@@ -364,7 +364,7 @@ def test_parse_body_with_chunked_transfer_encoding_and_chunk_way_too_large(self)
             next(self.parse())
         self.assertEqual(
             str(raised.exception),
-            "chunk too large: 0x1234567890ABCDEF bytes",
+            "invalid chunk size: 1234567890ABCDEF",
         )
 
     def test_parse_body_with_unsupported_transfer_encoding(self):