scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-21 00:21:49 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 74

fix(claude_code_gateway): mint the bearer before consuming the device code so a signing failure never spends the login

Browse source
This commit is contained in:
mateo-berri 2026-09-18 18:25:55 -07:00
parent ca8062e506
commit 06a5594bb6
2 changed files with 16 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
litellm/proxy/anthropic_endpoints/gateway_endpoints.py
View file

@ -315,13 +315,12 @@ async def _handle_device_code_grant(device_code: str | None) -> JSONResponse:
if isinstance(login, _OAuthError):
return _oauth_error_response(login)
access_token: Final = _mint_access_token(login)
if not await _claim_device_code(login_id, cli_sso_session_cache):
return _oauth_error_response(_OAuthError(status_code=400, error="expired_token"))
await cli_sso_session_cache.async_delete_cache(key=_get_cli_sso_flow_cache_key(login_id))
body: Final = _AccessTokenBody(
access_token=_mint_access_token(login), expires_in=CLI_JWT_EXPIRATION_HOURS * _SECONDS_PER_HOUR
)
body: Final = _AccessTokenBody(access_token=access_token, expires_in=CLI_JWT_EXPIRATION_HOURS * _SECONDS_PER_HOUR)
return JSONResponse(content=body.model_dump())

17
tests/test_litellm/proxy/anthropic_endpoints/test_gateway_endpoints.py
View file

@ -320,6 +320,18 @@ def test_token_malformed_session_is_invalid_grant_and_does_not_consume_the_login
mint.assert_not_called()
def test_token_mint_failure_leaves_the_login_unconsumed():
with _gateway_env() as (client, cache):
device_code = _start_device_flow(client)
_complete_flow(cache, device_code)
with patch(_MINT, side_effect=RuntimeError("signing key unavailable")), pytest.raises(RuntimeError):
_request_token(client, device_code)
with patch(_MINT, return_value="sk-session"):
retry = _request_token(client, device_code)
assert retry.status_code == 200
assert retry.json()["access_token"] == "sk-session"
def test_token_unknown_team_grants_is_invalid_grant():
with _gateway_env() as (client, cache):
device_code = _start_device_flow(client)
@ -349,11 +361,10 @@ def test_token_refuses_a_device_code_another_replica_already_claimed():
_set_cli_sso_flow(login_id=_SHARED_LOGIN_ID, cache=replica_a, flow=_completed_flow())
assert asyncio.run(gateway_endpoints._claim_device_code(_SHARED_LOGIN_ID, replica_a)) is True
with _gateway_env(cache=_replica(redis)) as (client, _), patch(_MINT) as mint:
with _gateway_env(cache=_replica(redis)) as (client, _), patch(_MINT, return_value="sk-session"):
resp = _request_token(client, _SHARED_DEVICE_CODE)
assert resp.status_code == 400
assert resp.json()["error"] == "expired_token"
mint.assert_not_called()
assert resp.json() == {"error": "expired_token"}
def test_token_unknown_device_code_is_expired_token():