Harden request URL and upload handling

Author: kvz

tests/test_async_client.py

@@ -365,6 +365,27 @@ async def test_async_client_normalizes_service_and_rejects_missing_ids(self):
         with self.assertRaises(ValueError):
             await client.cancel_assembly()
 
+        with self.assertRaises(ValueError):
+            await client.get_assembly(assembly_url="https://example.com/assemblies/abc123")
+
+        with self.assertRaises(ValueError):
+            await client.cancel_assembly(assembly_url="https://example.com/assemblies/abc123")
+
+        transloadit_session = _RecordingSession({"ok": "ASSEMBLY_COMPLETED"})
+        transloadit_client = AsyncTransloadit(
+            "key",
+            "secret",
+            service="https://api2.transloadit.com",
+            session=transloadit_session,
+        )
+        await transloadit_client.get_assembly(
+            assembly_url="https://api2-region.transloadit.com/assemblies/abc123"
+        )
+        self.assertEqual(
+            transloadit_session.calls[0][0],
+            "https://api2-region.transloadit.com/assemblies/abc123",
+        )
+
         await client.close()
 
         self.assertFalse(session.closed)
@@ -564,7 +585,7 @@ async def test_async_assembly_wait_raises_on_plain_text_poll_response(self):
         )
         sleep_mock.assert_awaited_once_with(0)
 
-    async def test_async_assembly_wait_returns_plain_text_poll_response(self):
+    async def test_async_assembly_wait_raises_on_plain_text_success_poll_response(self):
         initial_response = Response(
             data={
                 "ok": "ASSEMBLY_PROCESSING",
@@ -586,9 +607,9 @@ async def test_async_assembly_wait_returns_plain_text_poll_response(self):
             with mock.patch.object(client.request, "post", new=mock.AsyncMock(return_value=initial_response)) as post_mock:
                 with mock.patch.object(client, "get_assembly", new=mock.AsyncMock(return_value=plain_response)) as get_mock:
                     with mock.patch("asyncio.sleep", new_callable=mock.AsyncMock) as sleep_mock:
-                        response = await assembly.create(wait=True, resumable=False)
+                        with self.assertRaises(RuntimeError):
+                            await assembly.create(wait=True, resumable=False)
 
-        self.assertIs(response, plain_response)
         post_mock.assert_awaited_once()
         get_mock.assert_awaited_once_with(
             assembly_url=f"{self.server.base_url}/assemblies/assembly-123"
@@ -1313,7 +1334,7 @@ async def test_async_assembly_rate_limit_ignores_malformed_error_values(self):
         self.assertFalse(assembly._rate_limit_reached({"error": ["RATE_LIMIT_REACHED"]}))
         self.assertFalse(assembly._rate_limit_reached({"error": {"code": "RATE_LIMIT_REACHED"}}))
 
-    async def test_async_tus_upload_cancellation_waits_for_thread_to_finish(self):
+    async def test_async_tus_upload_cancellation_returns_before_thread_finishes(self):
         client = AsyncTransloadit("key", "secret", service=self.server.base_url)
         assembly = client.new_assembly()
         started = threading.Event()
@@ -1338,13 +1359,14 @@ def blocking_upload(assembly_url, tus_url, retries):
         upload_task.cancel()
         await asyncio.sleep(0.05)
 
-        self.assertFalse(upload_task.done())
+        self.assertTrue(upload_task.done())
         self.assertFalse(finished.is_set())
 
-        release.set()
         with self.assertRaises(asyncio.CancelledError):
             await upload_task
 
+        release.set()
+        await asyncio.to_thread(finished.wait, 5)
         self.assertTrue(finished.is_set())
 
     async def test_async_request_uses_connect_and_read_timeouts_for_uploads(self):
@@ -1406,7 +1428,7 @@ async def test_async_request_filters_none_and_lowercases_booleans_in_extra_data(
         response = await client.request.post(
             "/assemblies",
             data={"foo": "bar"},
-            extra_data={"enabled": True, "skip": None},
+            extra_data={"enabled": True, "skip": None, "tags": ["a", "b"]},
             files={"file": upload},
         )
 
@@ -1415,6 +1437,21 @@ async def test_async_request_filters_none_and_lowercases_booleans_in_extra_data(
         self.assertIn("enabled", fields)
         self.assertNotIn("skip", fields)
         self.assertEqual(fields["enabled"][2], "true")
+        tag_values = [field[2] for field in session.calls[0][1]["data"]._fields if field[0]["name"] == "tags"]
+        self.assertEqual(tag_values, ["a", "b"])
+
+    def test_non_closing_upload_stream_reflects_seekability(self):
+        class _NonSeekableUpload(io.BytesIO):
+            def seekable(self):
+                return False
+
+        class _BrokenSeekableUpload(io.BytesIO):
+            def seekable(self):
+                raise OSError("seekable failed")
+
+        self.assertTrue(_NonClosingUploadStream(io.BytesIO(b"payload")).seekable())
+        self.assertFalse(_NonClosingUploadStream(_NonSeekableUpload(b"payload")).seekable())
+        self.assertFalse(_NonClosingUploadStream(_BrokenSeekableUpload(b"payload")).seekable())
 
     async def test_async_request_uses_filename_fallback_for_trailing_slash_stream_name(self):
         session = _RecordingSession({"ok": "ASSEMBLY_COMPLETED"})

tests/test_client.py

@@ -72,6 +72,19 @@ def test_get_assembly(self, mock):
         self.assertEqual(response.data["ok"], "ASSEMBLY_COMPLETED")
         self.assertEqual(response.data["assembly_id"], "abcdef12345")
 
+    def test_quotes_path_ids(self):
+        with mock.patch.object(self.transloadit.request, 'get') as get_mock:
+            self.transloadit.get_assembly(assembly_id='assembly/with?chars')
+            self.transloadit.get_template('template/with?chars')
+
+        self.assertEqual(
+            get_mock.call_args_list,
+            [
+                mock.call('/assemblies/assembly%2Fwith%3Fchars'),
+                mock.call('/templates/template%2Fwith%3Fchars'),
+            ],
+        )
+
     @requests_mock.Mocker()
     def test_list_assemblies(self, mock):
         url = f"{self.transloadit.service}/assemblies"

tests/test_request.py

@@ -56,6 +56,18 @@ def test_payload_preserves_custom_auth_constraints(self):
         self.assertEqual(params["auth"]["max_size"], 1024)
         self.assertEqual(params["auth"]["referer"], "https://example.com")
 
+    def test_full_url_rejects_external_absolute_urls(self):
+        self.assertEqual(
+            self.request._get_full_url(f"{self.transloadit.service}/foo"),
+            f"{self.transloadit.service}/foo",
+        )
+        self.assertEqual(
+            self.request._get_full_url("https://api2-region.transloadit.com/foo"),
+            "https://api2-region.transloadit.com/foo",
+        )
+        with self.assertRaises(ValueError):
+            self.request._get_full_url("https://example.com/foo")
+
     @requests_mock.Mocker()
     def test_put(self, mock):
         url = f"{self.transloadit.service}/foo"

transloadit/async_assembly.py

@@ -78,16 +78,7 @@ def _do_tus_upload(self, assembly_url, tus_url, retries):
             ).upload()
 
     async def _do_tus_upload_async(self, assembly_url, tus_url, retries):
-        upload_task = asyncio.create_task(
-            asyncio.to_thread(self._do_tus_upload, assembly_url, tus_url, retries)
-        )
-        try:
-            await asyncio.shield(upload_task)
-        except asyncio.CancelledError:
-            try:
-                await asyncio.shield(upload_task)
-            finally:
-                raise
+        await asyncio.to_thread(self._do_tus_upload, assembly_url, tus_url, retries)
 
     async def create(self, wait=False, resumable=True, retries=3):
         """
@@ -111,10 +102,8 @@ async def create(self, wait=False, resumable=True, retries=3):
 
             response_data = self._response_data(response)
             if response_data is None:
-                if response.status_code >= 400:
+                if response.status_code >= 400 or wait or (resumable and self.files):
                     raise RuntimeError(f"Unexpected non-JSON response ({response.status_code}).")
-                if resumable and self.files:
-                    raise RuntimeError("Resumable assembly response is missing upload URLs.")
                 return response
 
             if self._rate_limit_reached(response_data):
@@ -163,9 +152,7 @@ async def create(self, wait=False, resumable=True, retries=3):
                     )
                     poll_data = self._response_data(poll_response)
                     if poll_data is None:
-                        if poll_response.status_code >= 400:
-                            raise RuntimeError(f"Unexpected non-JSON response ({poll_response.status_code}).")
-                        return poll_response
+                        raise RuntimeError(f"Unexpected non-JSON response ({poll_response.status_code}).")
 
                 return poll_response
 

transloadit/async_request.py

@@ -8,6 +8,7 @@
 import json
 from types import MappingProxyType
 from datetime import datetime, timedelta, timezone
+from urllib.parse import urlparse
 
 import aiohttp
 from requests.structures import CaseInsensitiveDict
@@ -18,6 +19,10 @@
 TIMEOUT = 60
 
 
+def _is_transloadit_host(hostname):
+    return hostname == "transloadit.com" or hostname.endswith(".transloadit.com")
+
+
 def _get_upload_filename(file_stream, fallback):
     name = getattr(file_stream, "name", None)
     if isinstance(name, (bytes, os.PathLike)):
@@ -60,6 +65,12 @@ def seek(self, *args):
         return self._file_stream.seek(*args)
 
     def seekable(self):
+        seekable = getattr(self._file_stream, "seekable", None)
+        if callable(seekable):
+            try:
+                return seekable()
+            except (OSError, ValueError):
+                return False
         return hasattr(self._file_stream, "seek")
 
     def tell(self):
@@ -121,14 +132,18 @@ def _timeout(self, files=False):
         )
 
     def _normalize_payload(self, data):
-        normalized = {}
+        normalized = []
         for key, value in data.items():
             if value is None:
                 continue
-            if isinstance(value, bool):
-                normalized[key] = "true" if value else "false"
-            else:
-                normalized[key] = str(value)
+            values = value if isinstance(value, (list, tuple)) else [value]
+            for item in values:
+                if item is None:
+                    continue
+                if isinstance(item, bool):
+                    normalized.append((key, "true" if item else "false"))
+                else:
+                    normalized.append((key, str(item)))
         return normalized
 
     async def _read_response_data(self, response):
@@ -144,9 +159,10 @@ async def get(self, path, params=None):
         """
         Makes an asynchronous HTTP GET request.
         """
+        url = self._get_full_url(path)
         session = await self._ensure_session()
         async with session.get(
-            self._get_full_url(path),
+            url,
             params=self._to_payload(params),
             headers=self._headers(),
             timeout=self._timeout(),
@@ -161,14 +177,15 @@ async def post(self, path, data=None, extra_data=None, files=None):
         """
         Makes an asynchronous HTTP POST request.
         """
+        url = self._get_full_url(path)
         session = await self._ensure_session()
         data = self._to_payload(data)
         if extra_data:
             data.update(extra_data)
 
         if files:
             form = aiohttp.FormData()
-            for key, value in self._normalize_payload(data).items():
+            for key, value in self._normalize_payload(data):
                 form.add_field(key, value)
 
             for key, file_stream in files.items():
@@ -185,7 +202,7 @@ async def post(self, path, data=None, extra_data=None, files=None):
             payload = self._normalize_payload(data)
 
         async with session.post(
-            self._get_full_url(path),
+            url,
             data=payload,
             headers=self._headers(),
             timeout=self._timeout(files=bool(files)),
@@ -200,10 +217,11 @@ async def put(self, path, data=None):
         """
         Makes an asynchronous HTTP PUT request.
         """
+        url = self._get_full_url(path)
         session = await self._ensure_session()
         data = self._normalize_payload(self._to_payload(data))
         async with session.put(
-            self._get_full_url(path),
+            url,
             data=data,
             headers=self._headers(),
             timeout=self._timeout(),
@@ -218,10 +236,11 @@ async def delete(self, path, data=None):
         """
         Makes an asynchronous HTTP DELETE request.
         """
+        url = self._get_full_url(path)
         session = await self._ensure_session()
         data = self._normalize_payload(self._to_payload(data))
         async with session.delete(
-            self._get_full_url(path),
+            url,
             data=data,
             headers=self._headers(),
             timeout=self._timeout(),
@@ -252,5 +271,15 @@ def _sign_data(self, message):
 
     def _get_full_url(self, url):
         if url.startswith(("http://", "https://")):
+            service = urlparse(self.transloadit.service)
+            target = urlparse(url)
+            same_origin = (target.scheme, target.netloc) == (service.scheme, service.netloc)
+            transloadit_origin = (
+                target.scheme == service.scheme
+                and _is_transloadit_host(service.hostname or "")
+                and _is_transloadit_host(target.hostname or "")
+            )
+            if not (same_origin or transloadit_origin):
+                raise ValueError("Absolute API URLs must use the configured Transloadit service origin.")
             return url
         return self.transloadit.service + url

transloadit/client.py

@@ -2,7 +2,7 @@
 import hmac
 import hashlib
 import time
-from urllib.parse import urlencode, quote_plus
+from urllib.parse import quote, quote_plus, urlencode
 
 from typing import Optional, Union, List
 
@@ -18,6 +18,10 @@ def _stringify_url_param(value: Union[str, int, float, bool]) -> str:
     return str(value)
 
 
+def _quote_path_segment(value: str) -> str:
+    return quote(str(value), safe="")
+
+
 class Transloadit:
     """
     This class serves as a client interface to the Transloadit API.
@@ -77,7 +81,7 @@ def get_assembly(self, assembly_id: str = None, assembly_url: str = None):
         if not (assembly_id or assembly_url):
             raise ValueError("Either 'assembly_id' or 'assembly_url' cannot be None.")
 
-        url = assembly_url if assembly_url else f"/assemblies/{assembly_id}"
+        url = assembly_url if assembly_url else f"/assemblies/{_quote_path_segment(assembly_id)}"
         return self.request.get(url)
 
     def list_assemblies(self, params: dict = None):
@@ -107,7 +111,7 @@ def cancel_assembly(self, assembly_id: str = None, assembly_url: str = None):
         if not (assembly_id or assembly_url):
             raise ValueError("Either 'assembly_id' or 'assembly_url' cannot be None.")
 
-        url = assembly_url if assembly_url else f"/assemblies/{assembly_id}"
+        url = assembly_url if assembly_url else f"/assemblies/{_quote_path_segment(assembly_id)}"
         return self.request.delete(url)
 
     def get_template(self, template_id: str):
@@ -119,7 +123,7 @@ def get_template(self, template_id: str):
 
         Return an instance of <transloadit.response.Response>
         """
-        return self.request.get(f"/templates/{template_id}")
+        return self.request.get(f"/templates/{_quote_path_segment(template_id)}")
 
     def list_templates(self, params: Optional[dict] = None):
         """
@@ -154,7 +158,7 @@ def update_template(self, template_id: str, data: dict):
 
         Return an instance of <transloadit.response.Response>
         """
-        return self.request.put(f"/templates/{template_id}", data=data)
+        return self.request.put(f"/templates/{_quote_path_segment(template_id)}", data=data)
 
     def delete_template(self, template_id: str):
         """
@@ -165,7 +169,7 @@ def delete_template(self, template_id: str):
 
         Return an instance of <transloadit.response.Response>
         """
-        return self.request.delete(f"/templates/{template_id}")
+        return self.request.delete(f"/templates/{_quote_path_segment(template_id)}")
 
     def get_bill(self, month: int, year: int):
         """