fix: Hardened header parsing.

Raise if there is more than one Content-Length header per segment. Improve tests a bit. Minor refactoring.

Author: defnull

multipart.py

@@ -671,19 +671,21 @@ def _on_segment_headerline(self, line: Union[bytes, bytearray]):
         if len(self._segment_headerlist) >= self.max_header_count:
             raise ParserLimitReached("Maximum segment header count exceeded")
 
-        # Decode headers into header name and value
+        # Decode headerline into normalized (name, value) pairs
         try:
             name, col, value = line.decode(self.header_charset).partition(":")
-            name = name.strip().title()
-            if not col or not name:
-                raise ParserError("Malformed segment header")
-            if not (name in _KNOWN_HEADERS or _re_hname.fullmatch(name)):
-                raise ParserError("Invalid segment header name")
-            value = value.strip()
         except UnicodeDecodeError as err:
             raise ParserError("Segment header failed to decode", err)
+        if not col:
+            raise ParserError("Malformed segment header")
+        name = name.strip().title()
+        value = value.strip()
+        if not (name in _KNOWN_HEADERS or _re_hname.fullmatch(name)):
+            raise ParserError("Invalid segment header name")
 
         if name == "Content-Length":
+            if self._segment_limit >= 0:
+                raise ParserError("Multiple segment Content-Length headers")
             try:
                 content_length = int(value)
                 if content_length < 0 or str(content_length) != value:

test/test_push_parser.py

@@ -195,21 +195,24 @@ def test_header_continuation_long(self):
         with self.assertParseError("Maximum segment header length exceeded"):
             self.parse(b"\tmoooooooooooooooooooooooooore value\r\n")
 
-    def test_header_bad_name(self):
-        self.reset()
+    def test_header_no_colon(self):
         with self.assertParseError("Malformed segment header"):
             self.parse(b"--boundary\r\nno-colon\r\n\r\ndata\r\n--boundary--")
-        self.reset()
-        with self.assertParseError("Malformed segment header"):
+
+    def test_header_empty_name(self):
+        with self.assertParseError("Invalid segment header name"):
             self.parse(b"--boundary\r\n:empty-name\r\n\r\ndata\r\n--boundary--")
+
+    def test_header_bad_name(self):
         for badchar in (b" ", b"\0", b"\r", b"\n", "ö".encode("utf8")):
             self.reset()
             with self.assertParseError("Invalid segment header name"):
                 self.parse(
                     b"--boundary\r\ninvalid%sname:value\r\n\r\ndata\r\n--boundary--"
                     % badchar
                 )
-        self.reset()
+
+    def test_header_bad_unicode(self):
         with self.assertParseError("Segment header failed to decode"):
             self.parse(
                 b"--boundary\r\ninvalid\xc3\x28:value\r\n\r\ndata\r\n--boundary--"
@@ -312,7 +315,17 @@ def test_segment_clen_bad_value(self):
                 self.parse("x" * 4)
                 self.parse("\r\n--boundary--")
 
-    def test_content_length_limit(self):
+    def test_segment_clen_repeated(self):
+        self.parse("--boundary\r\n")
+        self.parse("Content-Disposition: form-data; name=foo\r\n")
+        with self.assertParseError("Multiple segment Content-Length headers"):
+            self.parse(f"Content-Length: 1024\r\n")
+            self.parse(f"Content-Length: 1024\r\n")
+            self.parse("\r\n")
+            self.parse("x" * 1024)
+            self.parse("\r\n--boundary--")
+
+    def test_segment_clen_limit(self):
         self.reset(max_segment_size = 1024)
         self.parse("--boundary\r\n")
         self.parse("Content-Disposition: form-data; name=foo\r\n")