Decode extended filenames for streamed downloads

AUTHORS.rst

@@ -56,4 +56,6 @@ Patches and Suggestions
 
 - Chris van Marle (https://github.com/qistoph)
 
-- Florence Blanc-Renaud <flo@redhat.com> (@flo-renaud)
+- Florence Blanc-Renaud <flo@redhat.com> (@flo-renaud)
+
+- İbrahim S. Akarçeşme (@saitakarcesme)

HISTORY.rst

@@ -7,6 +7,8 @@ History
 Fixed Bugs
 ~~~~~~~~~~
 
+- Decode extended ``filename*`` parameters when streaming downloads to files
+
 - Save and recover ``base_url`` in ``BaseUrlSession`` during pickle and unpickle
 
 1.0.0 -- 2023-05-01

requests_toolbelt/downloadutils/stream.py

@@ -1,8 +1,14 @@
 # -*- coding: utf-8 -*-
 """Utilities for dealing with streamed requests."""
+import ntpath
 import os.path
 import re
 
+try:
+    from urllib.parse import unquote_to_bytes
+except ImportError:  # Python 2
+    from urllib import unquote as unquote_to_bytes
+
 from .. import exceptions as exc
 
 # Regular expressions stolen from werkzeug/http.py
@@ -16,12 +22,26 @@
 
 
 def _get_filename(content_disposition):
+    filename = None
     for match in _OPTION_HEADER_PIECE_RE.finditer(content_disposition):
         k, v = match.groups()
-        if k == 'filename':
-            # ignore any directory paths in the filename
-            return os.path.split(v)[1]
-    return None
+        if k.lower() == 'filename' and v is not None and filename is None:
+            # Keep the ordinary filename as a fallback for invalid filename*.
+            filename = os.path.split(v)[1]
+        elif k.lower() == 'filename*' and v is not None:
+            try:
+                charset, language, encoded = v.strip().split("'", 2)
+                # Invalid escapes must not silently become literal filenames.
+                if re.search(r'%(?![0-9a-fA-F]{2})', encoded):
+                    continue
+                value = unquote_to_bytes(encoded).decode(charset)
+            except (ValueError, LookupError, UnicodeError):
+                continue
+            # Strip path components after decoding, including Windows paths.
+            value = os.path.basename(ntpath.basename(value))
+            if value and value not in ('.', '..') and '\x00' not in value:
+                return value
+    return filename
 
 
 def get_download_file_path(response, path):
@@ -33,6 +53,9 @@ def get_download_file_path(response, path):
     file as reported by the server, and return a file path in the specified
     directory.
 
+    RFC 5987 encoded ``filename*`` values take precedence over ``filename``.
+    Invalid extended values fall back to the ordinary filename.
+
     If ``path`` is empty or None, this function will return a path relative
     to the process' current working directory.
 

tests/test_downloadutils.py

@@ -221,3 +221,44 @@ def test_tee_to_bytearray(streamed_response):
 def test_tee_to_bytearray_only_accepts_bytearrays():
     with pytest.raises(TypeError):
         tee.tee_to_bytearray(None, object())
+
+
+@pytest.mark.parametrize('header,expected', [
+    ("attachment; filename*=UTF-8''caf%C3%A9.txt", u'caf\u00e9.txt'),
+    ("attachment; filename*=UTF-8'tr'%C4%B0stanbul.txt", u'\u0130stanbul.txt'),
+    ("attachment; filename*=ISO-8859-1'en'%A3rates.txt", u'\u00a3rates.txt'),
+    ("attachment; FILENAME*=UTF-8''report+notes.txt", 'report+notes.txt'),
+    ("attachment; filename=fallback.txt; filename*=UTF-8''caf%C3%A9.txt",
+     u'caf\u00e9.txt'),
+    ("attachment; filename*=UTF-8''caf%C3%A9.txt; filename=fallback.txt",
+     u'caf\u00e9.txt'),
+    ("attachment; filename*=UTF-8''..%2Ffolder%2Freport.txt", 'report.txt'),
+    ("attachment; filename*=UTF-8''..%5Cfolder%5Creport.txt", 'report.txt'),
+    ("attachment; filename*=UTF-8''C%3Areport.txt", 'report.txt'),
+])
+def test_extended_download_filename(header, expected):
+    response = requests.Response()
+    response.headers['Content-Disposition'] = header
+    assert stream.get_download_file_path(response, None) == expected
+
+
+@pytest.mark.parametrize('extended', [
+    "missing-separators", "unknown''file.txt", "UTF-8''%FF.txt",
+    "UTF-8''bad%ZZ.txt", "UTF-8''bad%.txt", "UTF-8''bad%00.txt",
+    "UTF-8''..", "UTF-8''",
+])
+def test_invalid_extended_filename_falls_back(extended):
+    header = 'attachment; filename=fallback.txt; filename*=' + extended
+    assert stream._get_filename(header) == 'fallback.txt'
+
+
+def test_stream_response_to_extended_filename(tmpdir):
+    response = requests.Response()
+    response.headers['Content-Disposition'] = (
+        "attachment; filename*=UTF-8''caf%C3%A9.txt")
+    response._content = b'file contents'
+    response._content_consumed = True
+    filename = stream.stream_response_to_file(response, str(tmpdir))
+    assert filename == os.path.join(str(tmpdir), u'caf\u00e9.txt')
+    with open(filename, 'rb') as downloaded:
+        assert downloaded.read() == b'file contents'