fix(clinepass): reject realtime credential fallback and isolate provider policy

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Daniel JB Clark 2026-10-04 05:01:28 -04:00
parent 304d7a2e77
commit ace7be673d
No known key found for this signature in database
5 changed files with 143 additions and 15 deletions

View file

@ -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]:

View file

@ -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":

View file

@ -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,

View file

@ -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():

View file

@ -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 == []