diff --git a/litellm/llms/clinepass/chat/transformation.py b/litellm/llms/clinepass/chat/transformation.py index 301ac936f64..6db0819a51a 100644 --- a/litellm/llms/clinepass/chat/transformation.py +++ b/litellm/llms/clinepass/chat/transformation.py @@ -12,10 +12,13 @@ ClinePass is OpenAI-compatible apart from two quirks, both handled here: prefix before the request is built, so a qualifier has to be restored. Documentation: https://docs.cline.bot/ + +Credentials come only from the request's api_key or CLINEPASS_API_KEY. +Realtime endpoints are unsupported and rejected before HTTP dispatch. """ import json -from typing import TYPE_CHECKING, Any, Final +from typing import TYPE_CHECKING, Any, Final, NoReturn import httpx @@ -121,6 +124,16 @@ class ClinePassConfig(OpenAIGPTConfig): module docstring for the two quirks. """ + @staticmethod + def get_realtime_http_config(model: str) -> NoReturn: + from litellm.exceptions import BadRequestError + + raise BadRequestError( + message="ClinePass does not support realtime endpoints", + model=model, + llm_provider="clinepass", + ) + def _get_openai_compatible_provider_info( self, api_base: str | None, api_key: str | None ) -> tuple[str | None, str | None]: diff --git a/litellm/main.py b/litellm/main.py index e8eb20f2060..83cf6b33d6b 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -2448,10 +2448,10 @@ def _complete_aiohttp_openai( ) -def _complete_clinepass(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: +def _complete_http_provider(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult: acompletion: Final = ctx.acompletion - api_base = ctx.api_base - api_key = ctx.api_key + api_base: Final = ctx.api_base + api_key: Final = ctx.api_key client: Final = _dispatch_client_http(ctx) custom_llm_provider: Final = ctx.custom_llm_provider headers: Final = ctx.headers @@ -2466,11 +2466,6 @@ def _complete_clinepass(ctx: _CompletionDispatchContext) -> _CompletionDispatchR stream: Final = ctx.stream timeout: Final = ctx.timeout - api_key = api_key or get_secret_str("CLINEPASS_API_KEY") - - api_base = api_base or litellm.api_base or get_secret_str("CLINEPASS_API_BASE") or "https://api.cline.bot/api/v1" - - ## COMPLETION CALL response: Final = base_llm_http_handler.completion( model=model, messages=messages, # pyright: ignore[reportUnknownArgumentType] # ctx.messages is list[Unknown] @@ -5966,7 +5961,7 @@ def completion( elif custom_llm_provider == "cometapi": response = _complete_cometapi(_dispatch_ctx) elif custom_llm_provider == "clinepass": - response = _complete_clinepass(_dispatch_ctx) + response = _complete_http_provider(_dispatch_ctx) elif custom_llm_provider == "minimax": response = _complete_minimax(_dispatch_ctx) elif custom_llm_provider == "hosted_vllm": diff --git a/litellm/utils.py b/litellm/utils.py index ab840f4016b..7398a3415ad 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -8537,7 +8537,7 @@ class ProviderConfigManager: LlmProviders.EDENAI: (litellm.EdenAIChatConfig, False), LlmProviders.FAL_AI: (litellm.FalAIChatConfig, False), LlmProviders.COMETAPI: (lambda: litellm.CometAPIConfig(), False), - LlmProviders.CLINEPASS: (lambda: litellm.ClinePassConfig(), False), + LlmProviders.CLINEPASS: (litellm.ClinePassConfig, False), LlmProviders.DATAROBOT: (lambda: litellm.DataRobotConfig(), False), LlmProviders.GEMINI: (lambda: litellm.GoogleAIStudioGeminiConfig(), False), LlmProviders.AI21: (lambda: litellm.AI21ChatConfig(), False), @@ -9744,6 +9744,8 @@ class ProviderConfigManager: (POST /realtime/client_secrets and POST /realtime/calls). """ + if LlmProviders.CLINEPASS == provider: + return litellm.ClinePassConfig.get_realtime_http_config(model=model) if LlmProviders.OPENAI == provider: from litellm.llms.openai.realtime.http_transformation import ( OpenAIRealtimeHTTPConfig, diff --git a/tests/unit/llms/clinepass/chat/test_clinepass_chat_transformation.py b/tests/unit/llms/clinepass/chat/test_clinepass_chat_transformation.py index 2b11898f0bc..e1fc3a2bf3f 100644 --- a/tests/unit/llms/clinepass/chat/test_clinepass_chat_transformation.py +++ b/tests/unit/llms/clinepass/chat/test_clinepass_chat_transformation.py @@ -8,6 +8,7 @@ shipped as dead code. """ import json +from typing import Final from unittest.mock import patch import httpx @@ -58,6 +59,69 @@ def _clinepass_env(monkeypatch): monkeypatch.delenv("CLINEPASS_API_BASE", raising=False) +@pytest.fixture +def unrelated_credentials(monkeypatch): + for env_name in ("OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GROQ_API_KEY", "OPENROUTER_API_KEY"): + monkeypatch.setenv(env_name, f"sk-unrelated-{env_name}") + for attribute in ("api_key", "openai_key", "anthropic_key"): + monkeypatch.setattr(litellm, attribute, f"sk-unrelated-{attribute}") + + +@pytest.mark.parametrize("credential_source", ["missing", "environment", "explicit"]) +@pytest.mark.parametrize("custom_llm_provider", [None, "clinepass"]) +def test_chat_sends_only_clinepass_credentials( + monkeypatch, unrelated_credentials, credential_source, custom_llm_provider +): + if credential_source == "missing": + monkeypatch.delenv("CLINEPASS_API_KEY", raising=False) + explicit_key: Final = "cp-request-key" if credential_source == "explicit" else None + captured = {} + + def fake_post(self, url, *args, **kwargs): + captured["url"] = str(url) + captured["headers"] = httpx.Headers(kwargs["headers"]) + return _response(ENVELOPED_COMPLETION) + + with patch.object(HTTPHandler, "post", fake_post): + response = litellm.completion( + model="deepseek-v4-flash" if custom_llm_provider else "clinepass/deepseek-v4-flash", + custom_llm_provider=custom_llm_provider, + api_key=explicit_key, + messages=[{"role": "user", "content": "ping"}], + ) + + expected_key: Final = explicit_key or (API_KEY if credential_source == "environment" else None) + assert captured["url"] == "https://api.cline.bot/api/v1/chat/completions" + assert captured["headers"].get("authorization") == (f"Bearer {expected_key}" if expected_key else None) + assert response.choices[0].message.content == "pong" + + +@pytest.mark.parametrize("credential_source", ["missing", "environment", "explicit"]) +@pytest.mark.asyncio +async def test_async_chat_sends_only_clinepass_credentials(monkeypatch, unrelated_credentials, credential_source): + if credential_source == "missing": + monkeypatch.delenv("CLINEPASS_API_KEY", raising=False) + explicit_key: Final = "cp-request-key" if credential_source == "explicit" else None + captured = {} + + async def fake_post(self, url, *args, **kwargs): + captured["url"] = str(url) + captured["headers"] = httpx.Headers(kwargs["headers"]) + return _response(ENVELOPED_COMPLETION) + + with patch.object(AsyncHTTPHandler, "post", fake_post): + response = await litellm.acompletion( + model="clinepass/deepseek-v4-flash", + api_key=explicit_key, + messages=[{"role": "user", "content": "ping"}], + ) + + expected_key: Final = explicit_key or (API_KEY if credential_source == "environment" else None) + assert captured["url"] == "https://api.cline.bot/api/v1/chat/completions" + assert captured["headers"].get("authorization") == (f"Bearer {expected_key}" if expected_key else None) + assert response.choices[0].message.content == "pong" + + # -------------------------------------------------------------------------- # Registration / routing # -------------------------------------------------------------------------- @@ -308,8 +372,13 @@ async def test_acompletion_unwraps_envelope_and_prefixes_model(): assert response.choices[0].message.content == "pong" -def test_completion_streaming_is_not_unwrapped(): +@pytest.mark.parametrize("credential_source", ["missing", "environment", "explicit"]) +def test_completion_streaming_is_not_unwrapped(monkeypatch, unrelated_credentials, credential_source): """ClinePass does NOT wrap SSE chunks -- they are already OpenAI-shaped.""" + if credential_source == "missing": + monkeypatch.delenv("CLINEPASS_API_KEY", raising=False) + explicit_key: Final = "cp-stream-key" if credential_source == "explicit" else None + captured = {} chunks = [ { "id": "chatcmpl-test", @@ -332,6 +401,7 @@ def test_completion_streaming_is_not_unwrapped(): body = "".join(f"data: {json.dumps(c)}\n\n" for c in chunks) + "data: [DONE]\n\n" def fake_post(self, url, *args, **kwargs): + captured["authorization"] = httpx.Headers(kwargs["headers"]).get("authorization") return httpx.Response( 200, content=body.encode(), @@ -343,6 +413,7 @@ def test_completion_streaming_is_not_unwrapped(): stream = litellm.completion( model="clinepass/deepseek-v4-flash", messages=[{"role": "user", "content": "count"}], + api_key=explicit_key, max_tokens=4000, stream=True, ) @@ -357,10 +428,17 @@ def test_completion_streaming_is_not_unwrapped(): assert text == "one two three" assert finish_reason == "stop" + expected_key: Final = explicit_key or (API_KEY if credential_source == "environment" else None) + assert captured["authorization"] == (f"Bearer {expected_key}" if expected_key else None) +@pytest.mark.parametrize("credential_source", ["missing", "environment", "explicit"]) @pytest.mark.asyncio -async def test_acompletion_streaming_is_not_unwrapped(): +async def test_acompletion_streaming_is_not_unwrapped(monkeypatch, unrelated_credentials, credential_source): + if credential_source == "missing": + monkeypatch.delenv("CLINEPASS_API_KEY", raising=False) + explicit_key: Final = "cp-stream-key" if credential_source == "explicit" else None + captured = {} chunks = [ { "id": "chatcmpl-test", @@ -383,6 +461,7 @@ async def test_acompletion_streaming_is_not_unwrapped(): body = "".join(f"data: {json.dumps(c)}\n\n" for c in chunks) + "data: [DONE]\n\n" async def fake_post(self, url, *args, **kwargs): + captured["authorization"] = httpx.Headers(kwargs["headers"]).get("authorization") return httpx.Response( 200, content=body.encode(), @@ -394,6 +473,7 @@ async def test_acompletion_streaming_is_not_unwrapped(): stream = await litellm.acompletion( model="clinepass/deepseek-v4-flash", messages=[{"role": "user", "content": "count"}], + api_key=explicit_key, max_tokens=4000, stream=True, ) @@ -408,6 +488,8 @@ async def test_acompletion_streaming_is_not_unwrapped(): assert text == "one two three" assert finish_reason == "stop" + expected_key: Final = explicit_key or (API_KEY if credential_source == "environment" else None) + assert captured["authorization"] == (f"Bearer {expected_key}" if expected_key else None) def test_completion_streaming_tool_call_reassembly(): diff --git a/tests/unit/llms/clinepass/test_clinepass_endpoint_guard.py b/tests/unit/llms/clinepass/test_clinepass_endpoint_guard.py index a961641ce86..35c490db314 100644 --- a/tests/unit/llms/clinepass/test_clinepass_endpoint_guard.py +++ b/tests/unit/llms/clinepass/test_clinepass_endpoint_guard.py @@ -13,6 +13,7 @@ leaves the process at all, and that the OpenAI credential is never transmitted. """ import contextlib +from typing import Final import httpx import pytest @@ -36,10 +37,17 @@ def no_request_allowed(monkeypatch): attempted.append((str(request.url), request.headers.get("authorization", ""))) raise AssertionError(f"outbound request attempted to {request.url}") + def record_handler_and_block(self, url, *args, **kwargs): + attempted.append((str(url), httpx.Headers(kwargs.get("headers") or {}).get("authorization", ""))) + raise AssertionError(f"outbound request attempted to {url}") + + async def record_async_handler_and_block(self, url, *args, **kwargs): + record_handler_and_block(self, url, *args, **kwargs) + monkeypatch.setattr(httpx.Client, "send", record_and_block, raising=True) monkeypatch.setattr(httpx.AsyncClient, "send", record_and_block, raising=True) - for handler in (HTTPHandler, AsyncHTTPHandler): - monkeypatch.setattr(handler, "post", record_and_block, raising=True) + monkeypatch.setattr(HTTPHandler, "post", record_handler_and_block, raising=True) + monkeypatch.setattr(AsyncHTTPHandler, "post", record_async_handler_and_block, raising=True) monkeypatch.setenv("OPENAI_API_KEY", SENTINEL_OPENAI_KEY) monkeypatch.setenv("CLINEPASS_API_KEY", "cp-test-key") @@ -152,3 +160,31 @@ def test_chat_does_not_fall_back_to_the_global_litellm_api_key(monkeypatch): assert sent, "the chat request never reached the transport, so nothing was checked" assert all(SENTINEL_OPENAI_KEY not in authorization for _, authorization in sent) + + +@pytest.mark.parametrize("clinepass_key", [None, "cp-test-key"]) +@pytest.mark.parametrize("explicit_key", [None, "cp-explicit-key"]) +@pytest.mark.parametrize("endpoint", ["client_secret", "transcription_session", "calls"]) +@pytest.mark.asyncio +async def test_realtime_rejects_clinepass_before_credential_fallback( + monkeypatch, no_request_allowed, clinepass_key, explicit_key, endpoint +): + if clinepass_key is None: + monkeypatch.delenv("CLINEPASS_API_KEY", raising=False) + else: + monkeypatch.setenv("CLINEPASS_API_KEY", clinepass_key) + monkeypatch.setattr(litellm, "api_key", SENTINEL_OPENAI_KEY) + monkeypatch.setattr(litellm, "openai_key", SENTINEL_OPENAI_KEY) + kwargs: Final = {"model": "clinepass/deepseek-v4-flash", "api_key": explicit_key} + request: Final = ( + litellm.acreate_realtime_client_secret(**kwargs) + if endpoint == "client_secret" + else litellm.acreate_realtime_transcription_session(**kwargs) + if endpoint == "transcription_session" + else litellm.arealtime_calls(openai_ephemeral_key=SENTINEL_OPENAI_KEY, sdp_body=b"v=0\r\n", **kwargs) + ) + + with pytest.raises(litellm.BadRequestError, match="ClinePass does not support realtime endpoints"): + await request + + assert no_request_allowed == []