From 167fc4871a69893c296312e910a54a39c7a144b3 Mon Sep 17 00:00:00 2001
From: Ad1th <90973387+Ad1th@users.noreply.github.com>
Date: Thu, 1 Oct 2026 02:02:45 +0530
Subject: [PATCH] fix(media): detect GCS upload URLs by hostname instead of
 substring

The upload URL was checked with `"storage.googleapis.com" in upload_url`,
so any URL containing that string (e.g. in a query parameter or as a
subdomain prefix of another domain) was treated as a GCS bucket and sent
without the x-ms-blob-type / x-amz-checksum-sha256 headers. Parse the URL
and compare the hostname instead.
---
 langfuse/_task_manager/media_manager.py |  7 ++++-
 tests/unit/test_media_manager.py        | 35 +++++++++++++++++++++++++
 2 files changed, 41 insertions(+), 1 deletion(-)

diff --git a/langfuse/_task_manager/media_manager.py b/langfuse/_task_manager/media_manager.py
index 9a66ecd63..600c81bad 100644
--- a/langfuse/_task_manager/media_manager.py
+++ b/langfuse/_task_manager/media_manager.py
@@ -2,6 +2,7 @@
 import time
 from queue import Empty, Full, Queue
 from typing import Any, Callable, Optional, TypeVar, cast
+from urllib.parse import urlparse
 
 import backoff
 import httpx
@@ -448,7 +449,11 @@ def _process_upload_media_job(
         headers = {"Content-Type": data["content_type"]}
 
         # In self-hosted setups with GCP, do not add unsupported headers that fail the upload
-        is_self_hosted_gcs_bucket = "storage.googleapis.com" in upload_url
+        upload_host = (urlparse(upload_url).hostname or "").lower()
+        is_self_hosted_gcs_bucket = (
+            upload_host == "storage.googleapis.com"
+            or upload_host.endswith(".storage.googleapis.com")
+        )
 
         if not is_self_hosted_gcs_bucket:
             headers["x-ms-blob-type"] = "BlockBlob"
diff --git a/tests/unit/test_media_manager.py b/tests/unit/test_media_manager.py
index 3ab4e3226..0238363e8 100644
--- a/tests/unit/test_media_manager.py
+++ b/tests/unit/test_media_manager.py
@@ -277,3 +277,38 @@ def test_find_and_process_media_gemini_inline_data_non_string_data_passes_throug
 
     assert result == data
     assert queue.empty()
+
+
+@pytest.mark.parametrize(
+    ("upload_url", "is_gcs"),
+    [
+        ("https://storage.googleapis.com/bucket/obj?X-Goog-Signature=abc", True),
+        ("https://bucket.storage.googleapis.com/obj?X-Goog-Signature=abc", True),
+        ("https://evil.example/upload?next=storage.googleapis.com", False),
+        ("https://storage.googleapis.com.evil.example/upload", False),
+        ("https://s3.amazonaws.com/bucket/obj", False),
+    ],
+)
+def test_media_upload_gcs_detection_uses_url_host(upload_url, is_gcs):
+    media_api = Mock()
+    media_api.get_upload_url.return_value = SimpleNamespace(
+        upload_url=upload_url,
+        media_id="media-id",
+    )
+    media_api.patch.return_value = None
+
+    httpx_client = Mock()
+    httpx_client.put.return_value = _upload_response(200, "ok")
+
+    manager = MediaManager(
+        api_client=SimpleNamespace(media=media_api),
+        httpx_client=httpx_client,
+        media_upload_queue=Queue(),
+        max_retries=1,
+    )
+
+    manager._process_upload_media_job(data=_upload_job())
+
+    headers = httpx_client.put.call_args.kwargs["headers"]
+    assert ("x-ms-blob-type" not in headers) is is_gcs
+    assert ("x-amz-checksum-sha256" not in headers) is is_gcs