[pull] master from aio-libs:master

CHANGES/13499.bugfix.rst

@@ -0,0 +1,5 @@
+Fixed ``content_disposition_filename()`` truncating filenames split into 10 or
+more RFC 2231 continuation sections by sorting the sections numerically instead
+of lexicographically, and decoding each section according to its own
+encoded/quoted marker. A section sequence with a gap or a leading zero is now
+rejected instead of yielding a silently truncated name -- by :user:`2sumtech`.

CHANGES/13500.bugfix.rst

@@ -0,0 +1 @@
+13499.bugfix.rst

CONTRIBUTORS.txt

@@ -1,5 +1,6 @@
 - Contributors -
 ----------------
+2sumtech
 A. Jesse Jiryu Davis
 Abdur Rehman Ali
 Adam Bannister

aiohttp/multipart.py

@@ -202,33 +202,48 @@ def content_disposition_filename(
     elif name in params:
         return params[name]
     else:
-        parts = []
-        fnparams = sorted(
-            (key, value) for key, value in params.items() if key.startswith(name_suf)
+        # The index is capped at six digits so a header cannot push int() past
+        # CPython's int-to-str limit; a longer run of digits simply never matches.
+        section_re = re.compile(re.escape(name) + r"\*([0-9]{1,6})(\*)?")
+        matches = (
+            (m, value)
+            for key, value in params.items()
+            if (m := section_re.fullmatch(key)) is not None
         )
-        for num, (key, value) in enumerate(fnparams):
-            _, tail = key.split("*", 1)
-            if tail.endswith("*"):
-                tail = tail[:-1]
-            if tail == str(num):
-                parts.append(value)
-            else:
-                break
-        if not parts:
-            return None
-        value = "".join(parts)
-        if "'" in value:
-            encoding, _, value = value.split("'", 2)
-            encoding = encoding or "utf-8"
+        # https://www.rfc-editor.org/info/rfc2231/#section-3
+        # Order numerically.
+        fnparams = sorted(matches, key=lambda mv: int(mv[0].group(1)))
+        parts: list[str] = []
+        # Consecutive encoded sections are decoded as one unit, because a
+        # single multibyte character may be split across a section boundary.
+        pending: list[str] = []
+        encoding = "utf-8"
+        for num, (m, value) in enumerate(fnparams):
+            if m.group(1) != str(num):  # Missing section or leading zero.
+                return None
+            if m.group(2) is not None:  # encoded parameter
+                # https://www.rfc-editor.org/info/rfc2231/#section-4.1
+                if num == 0 and value.count("'") >= 2:
+                    encoding, _, value = value.split("'", 2)
+                    encoding = encoding or "utf-8"
+                pending.append(value)
+                continue
+            if pending:
+                # Current value is not encoded, so process encoding of previous parts
+                try:
+                    parts.append(unquote("".join(pending), encoding, "strict"))
+                except (builtins.LookupError, UnicodeDecodeError):
+                    return None
+                pending.clear()
+            parts.append(value)
+        if pending:
             try:
-                return unquote(value, encoding, "strict").lstrip("\\/")
+                parts.append(unquote("".join(pending), encoding, "strict"))
             except (builtins.LookupError, UnicodeDecodeError):
-                # Both the charset name and the octets are attacker-controlled
-                # here; an unknown encoding raises the builtin LookupError
-                # (shadowed in this module by payload.LookupError) and
-                # undecodable bytes raise UnicodeDecodeError.
                 return None
-        return value.lstrip("\\/")
+        if not parts:
+            return None
+        return "".join(parts).lstrip("\\/")
 
 
 class MultipartResponseWrapper:

tests/test_multipart_helpers.py

@@ -753,6 +753,76 @@ def test_attfncontqs(self) -> None:
         params = {"filename*0": "foo", "filename*1": "bar.html"}
         assert "foobar.html" == content_disposition_filename(params)
 
+    def test_attfncont_single_section(self) -> None:
+        params = {"filename*0": "foo.html"}
+        assert "foo.html" == content_disposition_filename(params)
+
+    def test_attfncont_many_sections(self) -> None:
+        # https://www.rfc-editor.org/info/rfc2231/#section-3
+        # Must be numeric ordering; "filename*10" must sort after "filename*2".
+        params = {f"filename*{i}": f"seg{i}-" for i in range(11)}
+        expected = "".join(f"seg{i}-" for i in range(11))
+        assert expected == content_disposition_filename(params)
+
+    def test_attfncont_many_sections_enc(self) -> None:
+        # https://www.rfc-editor.org/info/rfc2231/#section-4.1
+        params: dict[str, str] = {"filename*0*": "UTF-8''foo-"}
+        params.update({f"filename*{i}": f"s{i}-" for i in range(1, 10)})
+        params["filename*10*"] = "%c3%a4.html"
+        expected = "foo-" + "".join(f"s{i}-" for i in range(1, 10)) + "ä.html"
+        assert expected == content_disposition_filename(params)
+
+    def test_attfncont_encoded_after_quoted(self) -> None:
+        # https://www.rfc-editor.org/info/rfc2231/#section-4.1
+        params = {"filename*0": "foo", "filename*1*": "%20bar.html"}
+        assert "foo bar.html" == content_disposition_filename(params)
+
+    def test_attfncont_split_multibyte(self) -> None:
+        # A multibyte character may straddle a section boundary, so the
+        # octets have to be joined before they are decoded.
+        params = {"filename*0*": "UTF-8''%c3", "filename*1*": "%a4.html"}
+        assert "ä.html" == content_disposition_filename(params)
+
+    def test_attfncont_split_multibyte_over_three_sections(self) -> None:
+        # The three octets of a single character may even land in three
+        # different sections.
+        params = {
+            "filename*0*": "UTF-8''%e2",
+            "filename*1*": "%82",
+            "filename*2*": "%ac.html",
+        }
+        assert "€.html" == content_disposition_filename(params)
+
+    def test_attfncont_non_numeric_section(self) -> None:
+        # Only numbered sections take part in the continuation; a section
+        # with a non-numeric index is not a continuation at all.
+        params = {"filename*0": "foo", "filename*x": "bar", "filename*1": ".html"}
+        assert "foo.html" == content_disposition_filename(params)
+
+    def test_attfncont_absurd_section_index(self) -> None:
+        # A section index long enough to trip CPython's int-to-str limit must
+        # not raise; the over-long key is simply ignored as a non-continuation.
+        params = {"filename*0": "foo.html", "filename*" + "0" * 4400: "x"}
+        assert "foo.html" == content_disposition_filename(params)
+
+    def test_attfncont_undecodable_octets(self) -> None:
+        # Octets that are still invalid once the sections are joined are
+        # rejected rather than decoded with replacement characters.
+        params = {"filename*0*": "UTF-8''%ff", "filename*1*": "%fe.html"}
+        assert content_disposition_filename(params) is None
+
+    def test_attfncont_unknown_charset(self) -> None:
+        params = {"filename*0*": "bogus-charset''foo", "filename*1*": "%c3%a4"}
+        assert content_disposition_filename(params) is None
+
+    def test_attfncontqs_apostrophe(self) -> None:
+        # Apostrophes in quoted sections are plain characters, not an
+        # RFC 5987 charset'language' prefix.
+        params = {"filename*0": "it's", "filename*1": ".html"}
+        assert "it's.html" == content_disposition_filename(params)
+        params = {"filename*0": "a'b'c.html"}
+        assert "a'b'c.html" == content_disposition_filename(params)
+
     def test_attfncontenc(self) -> None:
         params = {"filename*0*": "UTF-8''foo-%c3%a4", "filename*1": ".html"}
         assert "foo-ä.html" == content_disposition_filename(params)
@@ -780,12 +850,13 @@ def test_attfncontenc_baddecode(self, params: dict[str, str]) -> None:
         assert content_disposition_filename(params) is None
 
     def test_attfncontlz(self) -> None:
+        # A malformed section sequence is rejected rather than truncated.
         params = {"filename*0": "foo", "filename*01": "bar"}
-        assert "foo" == content_disposition_filename(params)
+        assert content_disposition_filename(params) is None
 
     def test_attfncontnc(self) -> None:
         params = {"filename*0": "foo", "filename*2": "bar"}
-        assert "foo" == content_disposition_filename(params)
+        assert content_disposition_filename(params) is None
 
     def test_attfnconts1(self) -> None:
         params = {"filename*1": "foo", "filename*2": "bar"}