Add streaming field limits to FormParser

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+* Add opt-in field-count and field-size limits for URL-encoded forms in `FormParser` [#332](https://github.com/Kludex/python-multipart/issues/332).
 * Speed up querystring callback dispatch [#316](https://github.com/Kludex/python-multipart/pull/316).
 
 ## 0.0.32 (2026-06-04)

docs/api.md

@@ -1,3 +1,46 @@
+## URL-encoded field limits
+
+You can bound field count and size while you stream a URL-encoded body.
+
+```python
+from python_multipart import FormParser
+from python_multipart.multipart import Field
+
+fields: list[Field] = []
+parser = FormParser(
+    "application/x-www-form-urlencoded",
+    on_field=fields.append,
+    on_file=None,
+    config={"MAX_FIELDS": 2, "MAX_FIELD_SIZE": 9},
+)
+
+for chunk in [b"first=one", b"", b"&second=two"]:
+    parser.write(chunk)
+parser.finalize()
+
+assert [(field.field_name, field.value) for field in fields] == [
+    (b"first", b"one"),
+    (b"second", b"two"),
+]
+```
+
+`MAX_FIELDS` limits the number of fields, including duplicate names and fields with empty values.
+`MAX_FIELD_SIZE` limits each field's combined raw name/value bytes. It excludes the `=` and `&` separators.
+Percent-encoded bytes count before decoding. Both limits accept non-negative numbers.
+
+`FormParser` raises `FormParserError` when a limit is exceeded. It checks field count before assembling a new field
+and field size before buffering incoming bytes. A check in `on_field` would be too late to bound memory use.
+
+Both limits default to `float("inf")` to preserve existing behavior. A zero field-count limit rejects every field.
+A zero size limit accepts only fields with an empty name and value. These limits apply to
+`application/x-www-form-urlencoded` and `application/x-url-encoded`, not multipart forms or file uploads.
+
+Call `finalize()` after the stream ends. Empty chunks do not signal the end of the body.
+
+## API reference
+
+The following reference lists the public classes, functions, and exceptions.
+
 ::: python_multipart
 
 ::: python_multipart.exceptions

python_multipart/multipart.py

@@ -53,6 +53,8 @@ class FileConfig(TypedDict, total=False):
     class FormParserConfig(FileConfig):
         UPLOAD_ERROR_ON_BAD_CTE: bool
         MAX_BODY_SIZE: float
+        MAX_FIELDS: int | float
+        MAX_FIELD_SIZE: int | float
         MAX_HEADER_COUNT: int
         MAX_HEADER_SIZE: int
 
@@ -1606,12 +1608,17 @@ class FormParser:
             manually.
         config: Configuration to use for this FormParser.  The default values are taken from the DEFAULT_CONFIG value,
             and then any keys present in this dictionary will overwrite the default values.
+            For URL-encoded forms, `MAX_FIELDS` limits the number of fields and `MAX_FIELD_SIZE` limits the combined
+            raw name/value bytes per field. Both default to infinity and accept non-negative numbers.
+            Exceeding either limit raises `FormParserError` before buffering the offending field or bytes.
     """
 
     #: This is the default configuration for our form parser.
     #: Note: all file sizes should be in bytes.
     DEFAULT_CONFIG: FormParserConfig = {
         "MAX_BODY_SIZE": float("inf"),
+        "MAX_FIELDS": float("inf"),
+        "MAX_FIELD_SIZE": float("inf"),
         "MAX_HEADER_COUNT": DEFAULT_MAX_HEADER_COUNT,
         "MAX_HEADER_SIZE": DEFAULT_MAX_HEADER_SIZE,
         "MAX_MEMORY_FILE_SIZE": 1 * 1024 * 1024,
@@ -1686,18 +1693,36 @@ def _on_end() -> None:
             )
 
         elif content_type == "application/x-www-form-urlencoded" or content_type == "application/x-url-encoded":
-            name_buffer: list[bytes] = []
+            max_fields = self.config["MAX_FIELDS"]
+            max_field_size = self.config["MAX_FIELD_SIZE"]
+            for key, limit in (("MAX_FIELDS", max_fields), ("MAX_FIELD_SIZE", max_field_size)):
+                if not isinstance(limit, (int, float)) or not limit >= 0:
+                    raise ValueError(f"{key} must be a non-negative number, not {limit!r}")
 
+            name_buffer: list[bytes] = []
+            field_count = 0
+            field_size = 0
             f: Field | None = None
 
             def on_field_start() -> None:
-                pass
+                nonlocal field_count, field_size
+                field_count += 1
+                if field_count > max_fields:
+                    raise FormParserError(f"Too many fields. Maximum number of fields is {max_fields}.")
+                field_size = 0
 
             def on_field_name(data: bytes, start: int, end: int) -> None:
+                nonlocal field_size
+                field_size += end - start
+                if field_size > max_field_size:
+                    raise FormParserError(f"Field exceeded maximum size of {max_field_size} bytes.")
                 name_buffer.append(data[start:end])
 
             def on_field_data(data: bytes, start: int, end: int) -> None:
-                nonlocal f
+                nonlocal f, field_size
+                field_size += end - start
+                if field_size > max_field_size:
+                    raise FormParserError(f"Field exceeded maximum size of {max_field_size} bytes.")
                 if f is None:
                     f = Field(b"".join(name_buffer))
                     del name_buffer[:]

tests/test_formparser_limits.py

@@ -0,0 +1,118 @@
+from __future__ import annotations
+
+import tracemalloc
+
+import pytest
+
+from python_multipart import FormParser, create_form_parser
+from python_multipart.exceptions import FormParserError
+from python_multipart.multipart import Field
+
+
+@pytest.mark.parametrize("content_type", ["application/x-www-form-urlencoded", "application/x-url-encoded"])
+@pytest.mark.parametrize(
+    "chunks",
+    [
+        [b"first=one&second=two"],
+        [b"first=one&sec", b"ond=two"],
+        [b"first=one&second=two", b""],
+        [b"first=one", b"", b"&second=two", b"", b""],
+        [bytes([byte]) for byte in b"first=one&second=two"],
+    ],
+)
+def test_urlencoded_field_limits_accept_exact_boundaries(content_type: str, chunks: list[bytes]) -> None:
+    fields: list[Field] = []
+    parser = FormParser(content_type, fields.append, None, config={"MAX_FIELDS": 2, "MAX_FIELD_SIZE": 9})
+    for chunk in chunks:
+        parser.write(chunk)
+    parser.finalize()
+    assert [(field.field_name, field.value) for field in fields] == [(b"first", b"one"), (b"second", b"two")]
+
+
+@pytest.mark.parametrize("chunk_size", [1, 2, 1024])
+@pytest.mark.parametrize(
+    "body,config,message,expected",
+    [
+        (b"longname", {"MAX_FIELD_SIZE": 3}, "Field exceeded maximum size of 3 bytes", []),
+        (b"a=longvalue", {"MAX_FIELD_SIZE": 3}, "Field exceeded maximum size of 3 bytes", []),
+        (b"name=value", {"MAX_FIELD_SIZE": 8}, "Field exceeded maximum size of 8 bytes", []),
+        (b"x=%41", {"MAX_FIELD_SIZE": 3}, "Field exceeded maximum size of 3 bytes", []),
+        (b"a=1&b=12", {"MAX_FIELDS": 1}, "Too many fields", [(b"a", b"1")]),
+        (b"a&b", {"MAX_FIELDS": 1}, "Too many fields", [(b"a", None)]),
+        (b"=&=", {"MAX_FIELDS": 1}, "Too many fields", [(b"", None)]),
+        (b"a=1&a=2", {"MAX_FIELDS": 1}, "Too many fields", [(b"a", b"1")]),
+        (b"=", {"MAX_FIELDS": 0}, "Too many fields", []),
+        (b"a", {"MAX_FIELD_SIZE": 0}, "Field exceeded maximum size of 0 bytes", []),
+        (b"=x", {"MAX_FIELD_SIZE": 0}, "Field exceeded maximum size of 0 bytes", []),
+        (b"a=1&b=12", {"MAX_FIELD_SIZE": 2}, "Field exceeded maximum size of 2 bytes", [(b"a", b"1")]),
+    ],
+)
+def test_urlencoded_field_limits_reject_during_write(
+    body: bytes, config: dict[str, int], message: str, expected: list[tuple[bytes, bytes | None]], chunk_size: int
+) -> None:
+    fields: list[Field] = []
+    parser = FormParser("application/x-www-form-urlencoded", fields.append, None, config=config)
+    with pytest.raises(FormParserError, match=message):
+        for offset in range(0, len(body), chunk_size):
+            parser.write(body[offset : offset + chunk_size])
+    assert [(field.field_name, field.value) for field in fields] == expected
+
+
+@pytest.mark.parametrize(
+    "body,max_fields,expected",
+    [(b"", 0, []), (b"&&", 0, []), (b"=", 1, [(b"", None)]), (b"=&=", 2, [(b"", None), (b"", None)])],
+)
+def test_urlencoded_zero_size_limit(body: bytes, max_fields: int, expected: list[tuple[bytes, None]]) -> None:
+    fields: list[Field] = []
+    parser = FormParser(
+        "application/x-www-form-urlencoded", fields.append, None, config={"MAX_FIELDS": max_fields, "MAX_FIELD_SIZE": 0}
+    )
+    parser.write(body)
+    parser.finalize()
+    assert [(field.field_name, field.value) for field in fields] == expected
+
+
+@pytest.mark.parametrize("key", ["MAX_FIELDS", "MAX_FIELD_SIZE"])
+@pytest.mark.parametrize("limit", [-1, float("-inf"), float("nan"), None, "1"])
+def test_urlencoded_invalid_field_limit(key: str, limit: int | float | str | None) -> None:
+    with pytest.raises(ValueError, match=f"{key} must be a non-negative number"):
+        FormParser("application/x-www-form-urlencoded", None, None, config={key: limit})
+
+
+def test_urlencoded_field_limits_disabled_by_default() -> None:
+    fields: list[Field] = []
+    parser = FormParser("application/x-www-form-urlencoded", fields.append, None)
+    parser.write(b"a=1&" * 1001 + b"large=" + b"x" * (1024 * 1024 + 1))
+    parser.finalize()
+    assert len(fields) == 1002
+    assert fields[-1].field_name == b"large"
+    assert fields[-1].value == b"x" * (1024 * 1024 + 1)
+
+
+@pytest.mark.parametrize("config", [{"MAX_FIELDS": 1}, {"MAX_FIELD_SIZE": 1}])
+def test_urlencoded_field_limits_without_callback(config: dict[str, int]) -> None:
+    parser = FormParser("application/x-www-form-urlencoded", None, None, config=config)
+    with pytest.raises(FormParserError):
+        parser.write(b"a=1&b=2")
+
+
+@pytest.mark.parametrize("body", [b"x" * (2 * 1024 * 1024), b"a=" + b"x" * (2 * 1024 * 1024)])
+@pytest.mark.parametrize("config", [{"MAX_FIELDS": 0}, {"MAX_FIELD_SIZE": 2}])
+def test_urlencoded_field_limits_reject_before_buffering(body: bytes, config: dict[str, int]) -> None:
+    parser = FormParser("application/x-www-form-urlencoded", None, None, config=config)
+    tracemalloc.start()
+    try:
+        with pytest.raises(FormParserError):
+            parser.write(body)
+        _, peak = tracemalloc.get_traced_memory()
+    finally:
+        tracemalloc.stop()
+    assert peak < len(body) // 2
+
+
+def test_create_form_parser_forwards_field_limits() -> None:
+    parser = create_form_parser(
+        {"Content-Type": b"application/x-www-form-urlencoded"}, None, None, config={"MAX_FIELDS": 0}
+    )
+    with pytest.raises(FormParserError, match="Too many fields"):
+        parser.write(b"a=1")