Match async proxy policy and preserve multipart filenames

Author: kvz

CHANGELOG.md

@@ -9,6 +9,7 @@
 * Async cancellation drains active multipart reads and resumable workers, including repeated cancellation, before releasing caller-owned files. Isolated public, token, SSE and raw TUS transports preserve proxy/TLS configuration without borrowing session or netrc account credentials. High-level TUS helpers validate metadata and preserve Assembly creation errors.
 * Explicit bearer authentication wins over ambient netrc or session auth. Assembly capability requests isolate account credentials while preserving custom deployment worker hosts. Async multipart uploads retain text-mode file support, and polling continues through `ASSEMBLY_REPLAYING`.
 * Preserved legacy HTTP Assembly URLs by upgrading known API hosts to HTTPS, and aligned regional host checks with the pinned contract. Async transports retain explicit session proxy settings. Text uploads use chunked transfer when encoding/newline conversion changes their byte length; worker cleanup also survives global event-loop shutdown.
+* Async multipart requests preserve spaces and Unicode in filenames with safe header quoting. Isolated async calls use aiohttp's environment proxy policy, including proxy-only netrc authentication.
 * Hardened upload and response edge cases: invalid service URLs and empty template IDs now fail fast, external absolute API URLs are no longer signed, sync TUS uploads now handle nameless streams and submit rate limits before uploading, sync non-resumable retries rewind seekable files and reject non-seekable retry streams, sync/async polling rate-limit retries reset after successful polls and cap repeated rate limits, async form fields match sync boolean serialization, async TUS cancellation waits for worker cleanup, async rate-limit backoff honors server `retryIn`, Smart CDN signing rejects invalid workspace slugs/reserved query keys, and sync non-JSON responses fall back to response text.
 * Hardened sync and async request handling by preserving custom `auth` constraints, quoting path IDs, and keeping explicit/custom service URLs compatible with local, CI, and [Transloadit Gateway](https://github.com/transloadit/gateway) deployments.
 * Fixed sync and async template creation to send the current API `template` payload shape.

tests/test_contract_endpoints.py

@@ -355,6 +355,68 @@ async def scenario():
     asyncio.run(scenario())
 
 
+def test_async_proxy_selection_matches_signed_calls_with_all_proxy(monkeypatch):
+    for name in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "no_proxy"):
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("ALL_PROXY", "socks5://127.0.0.2:1")
+    monkeypatch.setenv("NO_PROXY", "")
+    async def scenario():
+        async with wire_client("async") as (client, server):
+            await client.list_templates()
+            await client.get_assembly(assembly_id=ASSEMBLY_ID)
+            client.bearer_token = "selected-token"
+            await client.list_templates()
+            assert len(server.requests) == 3
+    asyncio.run(scenario())
+
+
+def test_async_environment_proxy_preserves_proxy_netrc_only(monkeypatch, tmp_path):
+    netrc = tmp_path / "netrc"
+    netrc.write_text("machine 127.0.0.1 login proxy-user password proxy-password\nmachine 127.0.0.2 login origin-user password must-not-leak\n")
+    monkeypatch.setenv("NETRC", str(netrc))
+    monkeypatch.delenv("http_proxy", raising=False)
+    monkeypatch.delenv("no_proxy", raising=False)
+    monkeypatch.setenv("NO_PROXY", "")
+    async def scenario():
+        async with wire_client("async") as (client, server):
+            monkeypatch.setenv("HTTP_PROXY", server.url)
+            client.service = "http://127.0.0.2:1"
+            await client.get_assembly(assembly_id=ASSEMBLY_ID)
+            client.bearer_token = "selected-token"
+            await client.list_templates()
+            for request in server.requests:
+                assert request["headers"]["Proxy-Authorization"] == aiohttp.BasicAuth("proxy-user", "proxy-password").encode()
+            assert "Authorization" not in server.requests[0]["headers"]
+            assert server.requests[1]["headers"]["Authorization"] == "Bearer selected-token"
+    asyncio.run(scenario())
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.parametrize("filename", ["my photo.jpg", "caf\u00e9.jpg", "quote\"name.txt", "newline\r\nname.txt"])
+def test_multipart_preserves_original_filename(mode, filename):
+    async def scenario():
+        async with wire_client(mode) as (client, server):
+            upload = io.BytesIO(b"data")
+            upload.name = filename
+            await call(client, "create_assembly", files={"file": upload})
+            # Quoting must prevent header injection while preserving spaces/UTF-8.
+            expected = filename.replace("\r", "%0D").replace("\n", "%0A").replace('"', '%22')
+            assert server.requests[-1]["filenames"]["file"] == expected
+            assert not upload.closed
+    asyncio.run(scenario())
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+def test_multipart_field_names_cannot_inject_headers(mode):
+    async def scenario():
+        async with wire_client(mode) as (client, server):
+            await call(client, "create_assembly", files={"file\r\nname": io.BytesIO(b"data")},
+                       extra_data={"quoted\"field": "value"})
+            assert server.requests[-1]["fields"]["file%0D%0Aname"] == b"data"
+            assert server.requests[-1]["fields"]["quoted%22field"] == "value"
+    asyncio.run(scenario())
+
+
 ASSEMBLY_ID = "a" * 32
 
 
@@ -369,14 +431,17 @@ def __init__(self):
     async def handle(self, request):
         data = b""
         fields = {}
+        filenames = {}
         if request.content_type == "multipart/form-data":
             for key, value in (await request.post()).items():
                 fields[key] = value.file.read() if isinstance(value, web.FileField) else value
+                if isinstance(value, web.FileField):
+                    filenames[key] = value.filename
         else:
             data = await request.read()
             if request.content_type == "application/x-www-form-urlencoded":
                 fields = {key: values[0] for key, values in parse_qs(data.decode()).items()}
-        record = {"method": request.method, "path": request.path, "body": data,
+        record = {"method": request.method, "path": request.path, "body": data, "filenames": filenames,
                   "fields": fields, "query": dict(request.query), "headers": CaseInsensitiveDict(request.headers)}
         self.requests.append(record)
         if request.method == "GET" and request.path == "/assemblies/" + ASSEMBLY_ID:

transloadit/async_request.py

@@ -24,6 +24,10 @@
 TIMEOUT = 60
 
 
+def _multipart_parameter(value):
+    return str(value).replace("\r", "%0D").replace("\n", "%0A").replace('"', "%22")
+
+
 class _NonClosingUploadStream(io.IOBase):
     def __init__(self, file_stream):
         self._file_stream = file_stream
@@ -129,11 +133,11 @@ async def _api_session(self, url):
         else:
             yield await self._ensure_session()
 
-    def _request_options(self, url):
+    async def _request_options(self, url):
         headers = self._headers(url)
         options = {"headers": headers, "timeout": self._timeout(), "allow_redirects": False}
         if "Authorization" in headers:
-            options.update(async_network_options(url, self._session))
+            options.update(await async_network_options(url, self._session))
         return options
 
     def _get_session_lock(self):
@@ -199,7 +203,7 @@ async def get(self, path, params=None):
         async with self._api_session(url) as session, session.get(
             url,
             params=self._to_request_payload(url, params),
-            **self._request_options(url),
+            **(await self._request_options(url)),
         ) as response:
             return Response(
                 data=await self._read_response_data(response),
@@ -221,9 +225,11 @@ async def _form_request(self, method, path, data=None, extra_data=None, files=No
 
         uploads = []
         if files:
-            form = aiohttp.FormData()
+            # API2 preserves percent escapes in filename parameters. Match
+            # Requests' RFC 7578 quoting while retaining spaces and UTF-8.
+            form = aiohttp.FormData(quote_fields=False)
             for key, value in self._normalize_payload(data):
-                form.add_field(key, value)
+                form.add_field(_multipart_parameter(key), value)
 
             for key, file_stream in files.items():
                 wrapper = _NonClosingTextUploadStream if isinstance(file_stream, io.TextIOBase) else _NonClosingUploadStream
@@ -232,9 +238,9 @@ async def _form_request(self, method, path, data=None, extra_data=None, files=No
                 filename = get_upload_filename(file_stream, key)
                 content_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
                 form.add_field(
-                    key,
+                    _multipart_parameter(key),
                     upload,
-                    filename=filename,
+                    filename=_multipart_parameter(filename),
                     content_type=content_type,
                 )
             payload = form
@@ -245,7 +251,7 @@ async def _form_request(self, method, path, data=None, extra_data=None, files=No
             async with self._api_session(url) as session, getattr(session, method)(
                 url,
                 data=payload,
-                **self._request_options(url),
+                **(await self._request_options(url)),
             ) as response:
                 return Response(
                     data=await self._read_response_data(response),
@@ -279,7 +285,7 @@ async def delete(self, path, data=None):
         async with self._api_session(url) as session, session.delete(
             url,
             data=data,
-            **self._request_options(url),
+            **(await self._request_options(url)),
         ) as response:
             return Response(
                 data=await self._read_response_data(response),
@@ -336,7 +342,7 @@ async def public(self, method, path, params=None, data=None):
         async with self._isolated_session() as session:
             async with session.request(method, url, params=params, data=data,
                                        headers=self._headers(), timeout=self._timeout(), allow_redirects=False,
-                                       **async_network_options(url, self._session)) as response:
+                                       **(await async_network_options(url, self._session))) as response:
                 return Response(data=await self._read_response_data(response), status_code=response.status,
                                 headers=CaseInsensitiveDict(response.headers))
 
@@ -345,7 +351,7 @@ async def issue_bearer_token(self, data=None):
         headers.update(self._headers())
         async with self._isolated_session() as session:
             async with session.post(url, data=self._normalize_payload(data), headers=headers,
-                                    timeout=self._timeout(), allow_redirects=False, **async_network_options(url, self._session)) as response:
+                                    timeout=self._timeout(), allow_redirects=False, **(await async_network_options(url, self._session))) as response:
                 return Response(data=await self._read_response_data(response), status_code=response.status,
                                 headers=CaseInsensitiveDict(response.headers))
 
@@ -354,7 +360,7 @@ async def tus(self, operation_id, location, headers=None, body=None):
         headers.update(self._headers())
         async with self._isolated_session() as session:
             async with session.request(spec["method"], url, data=body, headers=headers,
-                                       timeout=self._timeout(), allow_redirects=False, **async_network_options(url, self._session)) as raw:
+                                       timeout=self._timeout(), allow_redirects=False, **(await async_network_options(url, self._session))) as raw:
                 data = await raw.read() if spec["binary_response"] else await self._read_response_data(raw)
                 response = Response(data=data, status_code=raw.status, headers=CaseInsensitiveDict(raw.headers))
         if spec["method"] == "POST":
@@ -368,7 +374,7 @@ async def stream_status(self, url, *, read_timeout=120, max_event_size=1048576):
         timeout = aiohttp.ClientTimeout(total=None, sock_connect=TIMEOUT, sock_read=read_timeout)
         async with self._isolated_session() as session:
             async with session.get(url, headers={**self._headers(), "Accept": "text/event-stream"},
-                                   timeout=timeout, allow_redirects=False, **async_network_options(url, self._session)) as response:
+                                   timeout=timeout, allow_redirects=False, **(await async_network_options(url, self._session))) as response:
                 if response.status != 200:
                     raise StreamError(response.status)
                 if response.headers.get("Content-Type", "").split(";", 1)[0].lower() != "text/event-stream":

transloadit/endpoint_transport.py

@@ -1,5 +1,6 @@
 """Shared wire preparation for OAuth, SSE and raw TUS endpoints."""
 import base64
+import asyncio
 import copy
 import os
 from ipaddress import ip_address
@@ -8,6 +9,8 @@
 
 from requests.structures import CaseInsensitiveDict
 from requests.utils import get_environ_proxies
+from aiohttp.helpers import get_env_proxy_for_url
+from yarl import URL
 
 from ._contract import CONTRACT
 from .response import Response
@@ -19,16 +22,21 @@ def sync_network_options(url):
             "verify": os.environ.get("REQUESTS_CA_BUNDLE") or os.environ.get("CURL_CA_BUNDLE") or True}
 
 
-def async_network_options(url, session=None):
+async def async_network_options(url, session=None):
     # aiohttp exposes the connector publicly, but its session-level proxy defaults
     # only through these attributes. Keep explicit proxy/auth ahead of environment.
     proxy = getattr(session, "_default_proxy", None)
     if proxy is not None:
         return {"proxy": proxy, "proxy_auth": getattr(session, "_default_proxy_auth", None)}
     if session is not None and not getattr(session, "trust_env", True):
         return {"proxy": None}
-    proxies = get_environ_proxies(url)
-    return {"proxy": proxies.get(urlsplit(url).scheme) or proxies.get("all")}
+    try:
+        # Use aiohttp's exact policy (including proxy-only netrc credentials),
+        # and keep netrc file reads off the event loop like ClientSession does.
+        proxy, proxy_auth = await asyncio.to_thread(get_env_proxy_for_url, URL(url))
+    except LookupError:
+        return {"proxy": None}
+    return {"proxy": proxy, "proxy_auth": proxy_auth}
 
 
 def token_request(client, data):