Replace cgi-legacy with multipart package

setup.py

@@ -54,7 +54,7 @@
     package_dir={"": "src"},
     python_requires=">=3.9.0",
     install_requires=[
-        "legacy-cgi>=2.6; python_version>='3.13'",
+        "multipart~=2.0",
     ],
     zip_safe=True,
     extras_require={"testing": testing_extras, "docs": docs_extras},

src/webob/multidict.py

@@ -7,10 +7,13 @@
 
 import binascii
 from collections.abc import MutableMapping
-from urllib.parse import urlencode as url_encode
+from functools import cached_property
+from urllib.parse import parse_qsl, urlencode as url_encode
 import warnings
 
-__all__ = ["MultiDict", "NestedMultiDict", "NoVars", "GetDict"]
+import multipart
+
+__all__ = ["MultiDict", "MultiDictFile", "NestedMultiDict", "NoVars", "GetDict"]
 
 
 class MultiDict(MutableMapping):
@@ -22,9 +25,7 @@ class MultiDict(MutableMapping):
 
     def __init__(self, *args, **kw):
         if len(args) > 1:
-            raise TypeError(
-                "MultiDict can only be called with one positional " "argument"
-            )
+            raise TypeError("MultiDict can only be called with one positional argument")
 
         if args:
             if hasattr(args[0], "items"):
@@ -55,9 +56,18 @@ def view_list(cls, lst):
         return obj
 
     @classmethod
-    def from_fieldstorage(cls, fs):
+    def from_fieldstorage(cls, fs):  # pragma: no cover
         """
         Create a multidict from a cgi.FieldStorage instance
+
+        .. deprecated:: 2.0
+
+            This method will not function in Python 3.13 or greater because the
+            `cgi` module has been removed.  Consider using the `multipart`_
+            library with :meth:`from_multipart` instead.
+
+        .. _multipart: https://pypi.org/project/multipart/
+
         """
         obj = cls()
         # fs.list can be None when there's nothing to parse
@@ -97,6 +107,30 @@ def decode(b):
 
         return obj
 
+    @classmethod
+    def from_multipart(cls, mp):
+        """
+        Create a multidict from a `MultipartParser`_ object.
+
+        .. _MultipartParser: https://multipart.readthedocs.io/en/latest/api.html#multipart.MultipartParser
+
+        """
+        obj = cls()
+
+        for part in mp:
+            if part.filename or not part.is_buffered():
+                obj.add(part.name, MultiDictFile(part))
+            else:
+                obj.add(part.name, part.value)
+        return obj
+
+    @classmethod
+    def from_qs(cls, data, charset="utf-8"):
+        data = parse_qsl(data, keep_blank_values=True)
+        return cls(
+            (key.decode(charset), value.decode(charset)) for (key, value) in data
+        )
+
     def __getitem__(self, key):
         for k, v in reversed(self._items):
             if k == key:
@@ -287,6 +321,48 @@ def values(self):
 _dummy = object()
 
 
+class MultiDictFile:
+    """
+    This object represents file uploads in a ``multipart/form-data`` request.
+
+    This is a thin wrapper around :cls:`multipart.MultipartPart` to mimic the
+    shape of Python's deprecated ``cgi.FieldStorage`` object to maintain backwards
+    compatibility with previous versions of webob.
+    """
+
+    def __init__(self, part: multipart.MultipartPart):
+        self._part = part
+        self.name = part.name
+        self.filename = part.filename
+        self.file = part.file
+        self.type = part.content_type
+        # self.type_options is a property
+        self.disposition = part.disposition
+        # self.disposition_options is a property
+        self.headers = part.headers
+        self.close = part.close
+
+    @cached_property
+    def type_options(self):
+        raw_ctype = self.headers.get("Content-Type", "")
+        return multipart.parse_options_header(raw_ctype)[1]
+
+    @cached_property
+    def disposition_options(self):
+        return multipart.parse_options_header(
+            self.disposition, unquote=multipart.content_disposition_unquote
+        )[1]
+
+    @cached_property
+    def value(self):
+        assert self.file
+        pos = self.file.tell()
+        self.file.seek(0)
+        val = self.file.read()
+        self.file.seek(pos)
+        return val
+
+
 class GetDict(MultiDict):
     #     def __init__(self, data, tracker, encoding, errors):
     #         d = lambda b: b.decode(encoding, errors)

src/webob/request.py

@@ -6,8 +6,9 @@
 import sys
 import tempfile
 from urllib import parse as urlparse
-from urllib.parse import quote as url_quote, quote_plus, urlencode as url_encode
-import warnings
+from urllib.parse import quote as url_quote, urlencode as url_encode
+
+from multipart import MultipartParser
 
 from webob.acceptparse import (
     accept_charset_property,
@@ -16,7 +17,6 @@
     accept_property,
 )
 from webob.cachecontrol import CacheControl, serialize_cache_control
-from webob.compat import cgi_FieldStorage
 from webob.cookies import RequestCookies
 from webob.descriptors import (
     CHARSET_RE,
@@ -168,18 +168,7 @@ def decode(self, charset=None, errors="strict"):
         elif content_type != "multipart/form-data":
             return r
 
-        fs_environ = self.environ.copy()
-        fs_environ.setdefault("CONTENT_LENGTH", "0")
-        fs_environ["QUERY_STRING"] = ""
-        fs = cgi_FieldStorage(
-            fp=self.body_file,
-            environ=fs_environ,
-            keep_blank_values=True,
-            encoding=charset,
-            errors=errors,
-        )
-
-        fout = t.transcode_fs(fs, r._content_type_raw)
+        fout = t.transcode_multipart(self.body_file, r._content_type_raw)
 
         # this order is important, because setting body_file
         # resets content_length
@@ -796,27 +785,22 @@ def POST(self):
             return NoVars(
                 "Not an HTML form submission (Content-Type: %s)" % content_type
             )
-        self._check_charset()
-
-        self.make_body_seekable()
-        self.body_file_raw.seek(0)
 
-        fs_environ = env.copy()
-        # FieldStorage assumes a missing CONTENT_LENGTH, but a
-        # default of 0 is better:
-        fs_environ.setdefault("CONTENT_LENGTH", "0")
-        fs_environ["QUERY_STRING"] = ""
-        fs = cgi_FieldStorage(
-            fp=self.body_file,
-            environ=fs_environ,
-            keep_blank_values=True,
-            encoding="utf8",
-        )
-
-        self.body_file_raw.seek(0)
-        vars = MultiDict.from_fieldstorage(fs)
+        self._check_charset()
+        if content_type == "multipart/form-data":
+            self.make_body_seekable()
+            self.body_file_raw.seek(0)
+            boundary = _get_multipart_boundary(self._content_type_raw)
+            parser = MultipartParser(
+                self.body_file,
+                boundary,
+                charset="utf8",
+            )
+            vars = MultiDict.from_multipart(parser)
+            self.body_file_raw.seek(0)
+        else:
+            vars = MultiDict.from_qs(self.body)
         env["webob._parsed_post_vars"] = (vars, self.body_file_raw)
-
         return vars
 
     @property
@@ -1752,23 +1736,14 @@ def transcode_query(self, q):
 
         return url_encode(q)
 
-    def transcode_fs(self, fs, content_type):
-        # transcode FieldStorage
-        def decode(b):
-            return b
-
-        data = []
-
-        for field in fs.list or ():
-            field.name = decode(field.name)
-
-            if field.filename:
-                field.filename = decode(field.filename)
-                data.append((field.name, field))
-            else:
-                data.append((field.name, decode(field.value)))
-
-        # TODO: transcode big requests to temp file
-        content_type, fout = _encode_multipart(data, content_type, fout=io.BytesIO())
-
+    def transcode_multipart(self, body, content_type):
+        # Transcode multipart
+        boundary = _get_multipart_boundary(content_type)
+        parser = MultipartParser(body, boundary, charset=self.charset)
+        data = MultiDict.from_multipart(parser)
+        content_type, fout = _encode_multipart(
+            data.items(),
+            content_type,
+            fout=io.BytesIO(),
+        )
         return fout

src/webob/util.py

@@ -1,6 +1,6 @@
+from html import escape
 import warnings
 
-from webob.compat import escape
 from webob.headers import _trans_key