From 498592e0ce607014673f722ddf8f25e5ad734a4b Mon Sep 17 00:00:00 2001 From: derhornspieler <15236687+derhornspieler@users.noreply.github.com> Date: Sun, 23 Aug 2026 12:58:54 -0400 Subject: [PATCH] test(anthropic): cover the token exchange and Keycloak default posters Both posters built their HTTPHandler inline, so nothing exercised the guarantees that matter there: redirects stay disabled, an HTTPStatusError comes back as its response rather than escaping, and a None response raises instead of being dereferenced. Each now takes a handler factory that defaults to the real one, which is the same injection seam the rest of this package already uses for its poster and secret reader Coverage for litellm/llms/base_llm/auth goes from 90% to 97%: client_credentials from 79% to 98%, with only its assert_never arm left, and token_exchange from 90% to 96% --- .../llms/base_llm/auth/client_credentials.py | 17 +-- litellm/llms/base_llm/auth/token_exchange.py | 17 +-- .../base_llm/auth/test_client_credentials.py | 81 ++++++++++++++ .../llms/base_llm/auth/test_token_exchange.py | 102 ++++++++++++++++++ 4 files changed, 205 insertions(+), 12 deletions(-) diff --git a/litellm/llms/base_llm/auth/client_credentials.py b/litellm/llms/base_llm/auth/client_credentials.py index e68c34ad1bb..b06f69bff01 100644 --- a/litellm/llms/base_llm/auth/client_credentials.py +++ b/litellm/llms/base_llm/auth/client_credentials.py @@ -49,24 +49,29 @@ def _default_secret_reader(ref: str) -> str | None: return get_secret_str(ref) +def _new_keycloak_handler() -> "HTTPHandler": + from litellm.llms.custom_httpx.http_handler import HTTPHandler + + handler: Final = HTTPHandler(timeout=httpx.Timeout(timeout=30.0, connect=5.0)) + handler.client.follow_redirects = False + return handler + + class _HttpxSyncKeycloakPoster: """Dedicated HTTPHandler for the Keycloak token POST: no ``logging_obj`` (so litellm's request/response logging never sees the client secret or the fetched token), redirects disabled. A separate instance from the outer engine's own poster, since this is a genuinely new HTTP call site whose no-logging guarantee must be built here, not assumed inherited.""" - def __init__(self) -> None: + def __init__(self, handler_factory: Callable[[], "HTTPHandler"] = _new_keycloak_handler) -> None: self._lock: Final = threading.Lock() + self._handler_factory: Final = handler_factory self._handler: HTTPHandler | None = None def _handler_instance(self) -> "HTTPHandler": - from litellm.llms.custom_httpx.http_handler import HTTPHandler - with self._lock: if self._handler is None: - handler: Final = HTTPHandler(timeout=httpx.Timeout(timeout=30.0, connect=5.0)) - handler.client.follow_redirects = False - self._handler = handler + self._handler = self._handler_factory() return self._handler def post(self, url: str, *, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response: diff --git a/litellm/llms/base_llm/auth/token_exchange.py b/litellm/llms/base_llm/auth/token_exchange.py index 4e3b0f41281..30615a70469 100644 --- a/litellm/llms/base_llm/auth/token_exchange.py +++ b/litellm/llms/base_llm/auth/token_exchange.py @@ -233,23 +233,28 @@ def _default_assertion_reader(ref: str) -> str | None: return get_secret_str(ref) +def _new_exchange_handler() -> "HTTPHandler": + from litellm.llms.custom_httpx.http_handler import HTTPHandler + + handler: Final = HTTPHandler(timeout=httpx.Timeout(timeout=30.0, connect=5.0)) + handler.client.follow_redirects = False + return handler + + class _HttpxSyncTokenPoster: """Default poster: a dedicated HTTPHandler (no logging_obj, so litellm's pre/post-call body logging never sees the exchange POST); returns the response for any status.""" - def __init__(self) -> None: + def __init__(self, handler_factory: Callable[[], "HTTPHandler"] = _new_exchange_handler) -> None: self._lock: Final = threading.Lock() + self._handler_factory: Final = handler_factory self._handler: HTTPHandler | None = None def _handler_instance(self) -> "HTTPHandler": - from litellm.llms.custom_httpx.http_handler import HTTPHandler - with self._lock: if self._handler is None: - handler: Final = HTTPHandler(timeout=httpx.Timeout(timeout=30.0, connect=5.0)) - handler.client.follow_redirects = False - self._handler = handler + self._handler = self._handler_factory() return self._handler def post(self, url: str, *, content: bytes, headers: Mapping[str, str], timeout: float) -> httpx.Response: diff --git a/tests/test_litellm/llms/base_llm/auth/test_client_credentials.py b/tests/test_litellm/llms/base_llm/auth/test_client_credentials.py index 553087d393a..d58c2673c45 100644 --- a/tests/test_litellm/llms/base_llm/auth/test_client_credentials.py +++ b/tests/test_litellm/llms/base_llm/auth/test_client_credentials.py @@ -8,10 +8,14 @@ import httpx import pytest from litellm.llms.base_llm.auth.client_credentials import ( + _HttpxSyncKeycloakPoster, + _default_secret_reader, + _new_keycloak_handler, fetch_keycloak_assertion, keycloak_assertion_source, ) from litellm.llms.base_llm.auth.identity_source import KeycloakSource, identity_source_ref +from litellm.llms.base_llm.auth.token_exchange import MAX_RESPONSE_BYTES TOKEN_URL: Final = "https://keycloak.example/realms/litellm/protocol/openid-connect/token" CLIENT_ID: Final = "litellm" @@ -331,3 +335,80 @@ class TestClientSecretNeverLeaks: ) assert CLIENT_SECRET not in caplog.text + + +class StubHandler: + """Stands in for the HTTPHandler the default poster builds, so the poster's own contract + (redirects off, error responses returned rather than raised, no-response guarded) is testable + without a socket.""" + + def __init__(self, result: httpx.Response | Exception | None) -> None: + self.calls: list[dict[str, object]] = [] + self._result = result + + def post(self, url: str, *, content: bytes, headers: dict[str, str], timeout: float) -> httpx.Response | None: + self.calls.append({"url": url, "content": content, "headers": headers, "timeout": timeout}) + if isinstance(self._result, Exception): + raise self._result + return self._result + + +class TestDefaultKeycloakPoster: + def test_builds_its_handler_once_with_redirects_disabled(self): + built: list[StubHandler] = [] + + def factory() -> StubHandler: + handler = StubHandler(httpx.Response(200, json={"access_token": "kc-token"})) + built.append(handler) + return handler + + poster: Final = _HttpxSyncKeycloakPoster(handler_factory=factory) # pyright: ignore[reportArgumentType] # StubHandler stands in for the legacy-untyped HTTPHandler + for _ in range(3): + poster.post(TOKEN_URL, content=b"grant_type=client_credentials", headers={}, timeout=1.0) + + assert len(built) == 1, "the handler is built once and reused" + assert len(built[0].calls) == 3 + + def test_the_real_handler_refuses_to_follow_redirects(self): + handler: Final = _new_keycloak_handler() + assert handler.client.follow_redirects is False, ( + "a redirected token POST would replay the client secret to whatever host the redirect names" + ) + + def test_an_http_status_error_becomes_its_response_rather_than_an_exception(self): + response: Final = httpx.Response( + 401, json={"error": "invalid_client"}, request=httpx.Request("POST", TOKEN_URL) + ) + poster: Final = _HttpxSyncKeycloakPoster( + handler_factory=lambda: StubHandler( + httpx.HTTPStatusError("boom", request=response.request, response=response) + ) # pyright: ignore[reportArgumentType] # StubHandler stands in for the legacy-untyped HTTPHandler + ) + + assert poster.post(TOKEN_URL, content=b"", headers={}, timeout=1.0).status_code == 401 + + def test_a_missing_response_is_a_transport_error_not_a_none_deref(self): + poster: Final = _HttpxSyncKeycloakPoster(handler_factory=lambda: StubHandler(None)) # pyright: ignore[reportArgumentType] # StubHandler stands in for the legacy-untyped HTTPHandler + + with pytest.raises(httpx.TransportError): + poster.post(TOKEN_URL, content=b"", headers={}, timeout=1.0) + + +class TestDefaultSecretReader: + def test_reads_through_litellm_secret_resolution(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("KEYCLOAK_CLIENT_SECRET_FOR_DEFAULT_READER", CLIENT_SECRET) + + assert _default_secret_reader("os.environ/KEYCLOAK_CLIENT_SECRET_FOR_DEFAULT_READER") == CLIENT_SECRET + + def test_an_unset_reference_reads_as_none_so_the_caller_raises(self): + assert _default_secret_reader("os.environ/DEFINITELY_NOT_SET_KEYCLOAK_SECRET_REF") is None + + +class TestOversizedSuccessBody: + def test_a_success_body_over_the_cap_is_refused_before_it_is_parsed(self): + oversized: Final = httpx.Response(200, content=b'{"access_token": "' + b"x" * MAX_RESPONSE_BYTES + b'"}') + + with pytest.raises(ValueError, match="exceeded the size cap"): + fetch_keycloak_assertion( + make_config(), poster=ScriptedPoster([oversized]), secret_reader=DEFAULT_SECRET_READER + ) diff --git a/tests/test_litellm/llms/base_llm/auth/test_token_exchange.py b/tests/test_litellm/llms/base_llm/auth/test_token_exchange.py index 5bfcc05fcad..d25a530c38d 100644 --- a/tests/test_litellm/llms/base_llm/auth/test_token_exchange.py +++ b/tests/test_litellm/llms/base_llm/auth/test_token_exchange.py @@ -19,6 +19,10 @@ from litellm.llms.base_llm.auth.token_exchange import ( MAX_ASSERTION_BYTES, MAX_RESPONSE_BYTES, JwtBearerTokenExchangeEngine, + _default_assertion_reader, + _error_summary, + _HttpxSyncTokenPoster, + _new_exchange_handler, redact_oauth_error_body, ) from litellm.llms.base_llm.auth.types import ( @@ -35,6 +39,7 @@ from litellm.secret_managers.main import OidcPathNotAllowedError, _resolve_oidc_ DEFAULT_REF: Final = "oidc/env/TEST_ASSERTION" DEFAULT_ASSERTION: Final = "test-jwt-assertion" +EXCHANGE_URL: Final = "https://token.example/v1/oauth/token" class FakeClock: @@ -1118,3 +1123,100 @@ def test_invalidate_bypasses_lead_backoff(): assert isinstance(second, MintedToken) assert second.access_token.get_secret_value() == "post-invalidate" assert len(poster.requests) == 2 + + +class StubExchangeHandler: + """Stands in for the HTTPHandler the default poster builds, so the poster's own contract is + testable without a socket.""" + + def __init__(self, result: httpx.Response | Exception | None) -> None: + self.calls = 0 + self._result = result + + def post(self, url: str, *, content: bytes, headers: dict[str, str], timeout: float) -> httpx.Response | None: + self.calls += 1 + if isinstance(self._result, Exception): + raise self._result + return self._result + + +class TestDefaultTokenPoster: + def test_builds_its_handler_once_and_reuses_it(self): + built: list[StubExchangeHandler] = [] + + def factory() -> StubExchangeHandler: + handler = StubExchangeHandler(httpx.Response(200, json={"access_token": "t"})) + built.append(handler) + return handler + + poster: Final = _HttpxSyncTokenPoster(handler_factory=factory) # pyright: ignore[reportArgumentType] # StubExchangeHandler stands in for the legacy-untyped HTTPHandler + for _ in range(3): + poster.post(EXCHANGE_URL, content=b"", headers={}, timeout=1.0) + + assert len(built) == 1 + assert built[0].calls == 3 + + def test_the_real_handler_refuses_to_follow_redirects(self): + assert _new_exchange_handler().client.follow_redirects is False, ( + "a redirected exchange POST would replay the workload assertion to the redirect target" + ) + + def test_an_http_status_error_becomes_its_response(self): + response: Final = httpx.Response( + 401, json={"error": "invalid_grant"}, request=httpx.Request("POST", EXCHANGE_URL) + ) + poster: Final = _HttpxSyncTokenPoster( + handler_factory=lambda: StubExchangeHandler( # pyright: ignore[reportArgumentType] # StubExchangeHandler stands in for the legacy-untyped HTTPHandler + httpx.HTTPStatusError("boom", request=response.request, response=response) + ) + ) + + assert poster.post(EXCHANGE_URL, content=b"", headers={}, timeout=1.0).status_code == 401 + + def test_a_missing_response_is_a_transport_error(self): + poster: Final = _HttpxSyncTokenPoster(handler_factory=lambda: StubExchangeHandler(None)) # pyright: ignore[reportArgumentType] # StubExchangeHandler stands in for the legacy-untyped HTTPHandler + + with pytest.raises(httpx.TransportError): + poster.post(EXCHANGE_URL, content=b"", headers={}, timeout=1.0) + + +class TestDefaultAssertionReader: + def test_reads_through_litellm_secret_resolution(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("WIF_ASSERTION_FOR_DEFAULT_READER", "header.payload.signature") + + assert _default_assertion_reader("os.environ/WIF_ASSERTION_FOR_DEFAULT_READER") == "header.payload.signature" + + def test_an_unset_reference_reads_as_none(self): + assert _default_assertion_reader("os.environ/DEFINITELY_NOT_SET_WIF_ASSERTION_REF") is None + + +class TestErrorSummary: + def test_every_error_variant_summarises_without_carrying_a_secret(self): + summaries: Final = { + _error_summary(AssertionSourceError(kind="unreadable", source_ref="oidc/file/x")), + _error_summary(InsecureTokenUrl(host="token.internal")), + _error_summary(TokenEndpointError(status_code=401, redacted_body="invalid_grant")), + _error_summary(TokenTransportError(detail="ConnectError: refused")), + _error_summary(MalformedTokenResponse(detail="empty access_token")), + } + + assert {s.split(":")[0] for s in summaries} == { + "AssertionSourceError", + "InsecureTokenUrl", + "TokenEndpointError", + "TokenTransportError", + "MalformedTokenResponse", + }, "each variant names itself so a log line says which stage failed" + + +class TestNonBearerTokenType: + def test_a_non_bearer_token_type_is_refused(self): + poster: Final = ScriptedPoster( + [httpx.Response(200, json={"access_token": "tok", "token_type": "mac", "expires_in": 300})] + ) + engine: Final = JwtBearerTokenExchangeEngine(poster=poster, assertion_reader=lambda _ref: DEFAULT_ASSERTION) + + result: Final = engine.get_token(make_spec()) + + assert isinstance(result, MalformedTokenResponse) + assert "non-bearer" in result.detail