fix: block path traversal SSRF in BitBucket, Arize Phoenix, and AssemblyAI clients (#26943)

* fix: sanitize BitBucket file path to block path traversal SSRF

* fix: sanitize Arize Phoenix prompt_version_id to block SSRF

* fix: sanitize AssemblyAI transcript_id to block SSRF

* test: add path traversal SSRF security tests for BitBucket client

* test: add SSRF security tests for Arize Phoenix client

* style: black format arize_phoenix_client.py

* style: black format assembly_passthrough_logging_handler.py

* test: add SSRF security tests for AssemblyAI transcript_id validation

* fix: move AssemblyAI transcript_id validation before try/except so ValueError propagates

litellm/integrations/arize/arize_phoenix_client.py

@@ -2,11 +2,23 @@
 Arize Phoenix API client for fetching prompt versions from Arize Phoenix.
 """
 
+import urllib.parse
 from typing import Any, Dict, Optional
 
 from litellm.llms.custom_httpx.http_handler import HTTPHandler
 
 
+def _sanitize_id(identifier: str) -> str:
+    """Reject path traversal characters and URL-encode the identifier."""
+    if any(c in identifier for c in ("/", "\\", "#", "?")):
+        raise ValueError(
+            f"Invalid identifier {identifier!r}: contains disallowed characters"
+        )
+    if ".." in identifier:
+        raise ValueError(f"Invalid identifier {identifier!r}: path traversal detected")
+    return urllib.parse.quote(identifier, safe="")
+
+
 class ArizePhoenixClient:
     """
     Client for interacting with Arize Phoenix API to fetch prompt versions.
@@ -53,7 +65,8 @@ class ArizePhoenixClient:
         Returns:
             Dictionary containing prompt version data, or None if not found
         """
-        url = f"{self.api_base}/v1/prompt_versions/{prompt_version_id}"
+        safe_id = _sanitize_id(prompt_version_id)
+        url = f"{self.api_base}/v1/prompt_versions/{safe_id}"
 
         try:
             # Use the underlying httpx client directly to avoid query param extraction

litellm/integrations/bitbucket/bitbucket_client.py

@@ -3,11 +3,27 @@ BitBucket API client for fetching .prompt files from BitBucket repositories.
 """
 
 import base64
+import urllib.parse
 from typing import Any, Dict, List, Optional
 
 from litellm.llms.custom_httpx.http_handler import HTTPHandler
 
 
+def _sanitize_file_path(file_path: str) -> str:
+    """Reject path traversal and URL-encode each path segment."""
+    if "#" in file_path or "?" in file_path:
+        raise ValueError(
+            f"Invalid file path {file_path!r}: contains URL special characters"
+        )
+    parts = file_path.split("/")
+    for part in parts:
+        if part == "..":
+            raise ValueError(
+                f"Invalid file path {file_path!r}: path traversal detected"
+            )
+    return "/".join(urllib.parse.quote(part, safe="") for part in parts)
+
+
 class BitBucketClient:
     """
     Client for interacting with BitBucket API to fetch .prompt files.
@@ -72,7 +88,8 @@ class BitBucketClient:
         Returns:
             File content as string, or None if file not found
         """
-        url = f"{self.base_url}/repositories/{self.workspace}/{self.repository}/src/{self.branch}/{file_path}"
+        safe_path = _sanitize_file_path(file_path)
+        url = f"{self.base_url}/repositories/{self.workspace}/{self.repository}/src/{self.branch}/{safe_path}"
 
         try:
             response = self.http_handler.get(url, headers=self.headers)
@@ -119,7 +136,8 @@ class BitBucketClient:
         Returns:
             List of file paths
         """
-        url = f"{self.base_url}/repositories/{self.workspace}/{self.repository}/src/{self.branch}/{directory_path}"
+        safe_dir = _sanitize_file_path(directory_path) if directory_path else ""
+        url = f"{self.base_url}/repositories/{self.workspace}/{self.repository}/src/{self.branch}/{safe_dir}"
 
         try:
             response = self.http_handler.get(url, headers=self.headers)
@@ -211,7 +229,8 @@ class BitBucketClient:
         Returns:
             Dictionary containing file metadata, or None if file not found
         """
-        url = f"{self.base_url}/repositories/{self.workspace}/{self.repository}/src/{self.branch}/{file_path}"
+        safe_path = _sanitize_file_path(file_path)
+        url = f"{self.base_url}/repositories/{self.workspace}/{self.repository}/src/{self.branch}/{safe_path}"
 
         try:
             # Use GET with Range header to get just the headers (HEAD equivalent)

litellm/proxy/pass_through_endpoints/llm_provider_handlers/assembly_passthrough_logging_handler.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import time
+import urllib.parse
 from datetime import datetime
 from typing import Literal, Optional
 from urllib.parse import urlparse
@@ -203,8 +204,16 @@ class AssemblyAIPassthroughLoggingHandler:
         )
         if _api_key is None:
             raise ValueError("AssemblyAI API key not found")
+        if (
+            any(c in transcript_id for c in ("/", "\\", "#", "?"))
+            or ".." in transcript_id
+        ):
+            raise ValueError(
+                f"Invalid transcript_id {transcript_id!r}: contains disallowed characters"
+            )
+        safe_transcript_id = urllib.parse.quote(transcript_id, safe="")
         try:
-            url = f"{_base_url}/v2/transcript/{transcript_id}"
+            url = f"{_base_url}/v2/transcript/{safe_transcript_id}"
             headers = {
                 "Authorization": f"Bearer {_api_key}",
                 "Content-Type": "application/json",

tests/pass_through_unit_tests/test_assemblyai_unit_tests_passthrough.py

@@ -134,3 +134,62 @@ def test_is_assemblyai_route():
         == False
     )
     assert handler.is_assemblyai_route("") == False
+
+
+# --- Security: SSRF via transcript_id path traversal ---
+
+
+def test_get_assembly_transcript_rejects_slash_in_id(assembly_handler):
+    with patch(
+        "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
+        return_value="test-key",
+    ):
+        with pytest.raises(ValueError, match="disallowed characters"):
+            assembly_handler._get_assembly_transcript("../../admin/credentials")
+
+
+def test_get_assembly_transcript_rejects_dotdot_in_id(assembly_handler):
+    with patch(
+        "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
+        return_value="test-key",
+    ):
+        with pytest.raises(ValueError, match="disallowed characters"):
+            assembly_handler._get_assembly_transcript("..evil")
+
+
+def test_get_assembly_transcript_rejects_fragment_in_id(assembly_handler):
+    with patch(
+        "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
+        return_value="test-key",
+    ):
+        with pytest.raises(ValueError, match="disallowed characters"):
+            assembly_handler._get_assembly_transcript("abc#suffix")
+
+
+def test_get_assembly_transcript_rejects_query_in_id(assembly_handler):
+    with patch(
+        "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
+        return_value="test-key",
+    ):
+        with pytest.raises(ValueError, match="disallowed characters"):
+            assembly_handler._get_assembly_transcript("abc?x=1")
+
+
+def test_get_assembly_transcript_allows_valid_id(
+    assembly_handler, mock_transcript_response
+):
+    with patch(
+        "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints.passthrough_endpoint_router.get_credentials",
+        return_value="test-key",
+    ):
+        with patch("httpx.get") as mock_get:
+            mock_get.return_value.json.return_value = mock_transcript_response
+            mock_get.return_value.raise_for_status.return_value = None
+
+            transcript = assembly_handler._get_assembly_transcript(
+                "abc123-valid-id_xyz"
+            )
+            assert transcript == mock_transcript_response
+            called_url = mock_get.call_args[0][0]
+            assert "abc123-valid-id_xyz" in called_url
+            assert ".." not in called_url

tests/test_litellm/integrations/arize/test_arize_phoenix.py

@@ -280,3 +280,55 @@ class TestDynamicProjectNameOnSpan:
 
 if __name__ == "__main__":
     unittest.main()
+
+
+# --- Security: SSRF via prompt_version_id path traversal ---
+
+
+def test_arize_phoenix_client_sanitize_id_rejects_traversal():
+    from litellm.integrations.arize.arize_phoenix_client import _sanitize_id
+
+    # dotdot without slashes
+    with pytest.raises(ValueError, match="path traversal"):
+        _sanitize_id("..something")
+    # full traversal (slash caught first)
+    with pytest.raises(ValueError, match="disallowed characters"):
+        _sanitize_id("../../projects")
+
+
+def test_arize_phoenix_client_sanitize_id_rejects_slash():
+    from litellm.integrations.arize.arize_phoenix_client import _sanitize_id
+
+    with pytest.raises(ValueError, match="disallowed characters"):
+        _sanitize_id("valid/extra")
+
+
+def test_arize_phoenix_client_sanitize_id_rejects_fragment():
+    from litellm.integrations.arize.arize_phoenix_client import _sanitize_id
+
+    with pytest.raises(ValueError, match="disallowed characters"):
+        _sanitize_id("abc#suffix")
+
+
+def test_arize_phoenix_client_sanitize_id_rejects_query():
+    from litellm.integrations.arize.arize_phoenix_client import _sanitize_id
+
+    with pytest.raises(ValueError, match="disallowed characters"):
+        _sanitize_id("abc?x=1")
+
+
+def test_arize_phoenix_client_sanitize_id_allows_uuid():
+    from litellm.integrations.arize.arize_phoenix_client import _sanitize_id
+
+    uid = "550e8400-e29b-41d4-a716-446655440000"
+    assert _sanitize_id(uid) == uid
+
+
+def test_arize_phoenix_client_get_prompt_version_rejects_traversal():
+    from litellm.integrations.arize.arize_phoenix_client import ArizePhoenixClient
+
+    client = ArizePhoenixClient(
+        api_key="test-key", api_base="https://app.phoenix.arize.com"
+    )
+    with pytest.raises(ValueError, match="disallowed characters"):
+        client.get_prompt_version("../../projects")

tests/test_litellm/integrations/bitbucket/test_bitbucket_integration.py

@@ -11,6 +11,7 @@ sys.path.insert(
 
 import litellm
 from litellm.integrations.bitbucket import BitBucketPromptManager
+from litellm.integrations.bitbucket.bitbucket_client import _sanitize_file_path
 
 
 @patch("litellm.integrations.bitbucket.bitbucket_prompt_manager.BitBucketClient")
@@ -370,3 +371,45 @@ def test_bitbucket_prompt_manager_list_templates(mock_client_class):
     templates = manager.prompt_manager.list_templates()
     assert isinstance(templates, list)
     assert "test_prompt" in templates
+
+
+# --- Security: path traversal / SSRF ---
+
+
+def test_sanitize_file_path_rejects_traversal():
+    with pytest.raises(ValueError, match="path traversal"):
+        _sanitize_file_path("../../etc/passwd")
+
+
+def test_sanitize_file_path_rejects_fragment():
+    with pytest.raises(ValueError, match="URL special characters"):
+        _sanitize_file_path("secret#.prompt")
+
+
+def test_sanitize_file_path_rejects_query():
+    with pytest.raises(ValueError, match="URL special characters"):
+        _sanitize_file_path("secret?.prompt")
+
+
+def test_sanitize_file_path_encodes_special_chars():
+    result = _sanitize_file_path("prompts/my prompt.prompt")
+    assert result == "prompts/my%20prompt.prompt"
+
+
+def test_sanitize_file_path_allows_normal_paths():
+    assert _sanitize_file_path("prompts/my-prompt") == "prompts/my-prompt"
+    assert _sanitize_file_path("simple") == "simple"
+
+
+def test_bitbucket_client_rejects_traversal_in_get_file_content():
+    from litellm.integrations.bitbucket.bitbucket_client import BitBucketClient
+
+    client = BitBucketClient(
+        {
+            "workspace": "ws",
+            "repository": "repo",
+            "access_token": "tok",
+        }
+    )
+    with pytest.raises(ValueError, match="path traversal"):
+        client.get_file_content("../../admin/credentials")