diff --git a/src/anthropic/lib/credentials/_auth.py b/src/anthropic/lib/credentials/_auth.py index d2019666c..8a5a2729d 100644 --- a/src/anthropic/lib/credentials/_auth.py +++ b/src/anthropic/lib/credentials/_auth.py @@ -9,6 +9,7 @@ from ._cache import TokenCache from ..._utils import asyncify +from ..._exceptions import AnthropicError from ._constants import OAUTH_API_BETA_HEADER __all__ = ["AccessTokenAuth"] @@ -86,6 +87,12 @@ def _has_static_credential(request: httpx.Request) -> bool: return bool(request.headers.get("X-Api-Key") or request.headers.get("Authorization")) def _apply(self, request: httpx.Request, token: str) -> None: + if not isinstance(token, str) or not token or token != token.strip(): + raise AnthropicError( + "Credentials provider returned an invalid access token; expected a non-empty string " + "without surrounding whitespace." + ) + request.headers["Authorization"] = f"Bearer {token}" existing_beta = request.headers.get("anthropic-beta", "") # Tokenize the comma-separated header so dedupe matches whole flag diff --git a/tests/lib/test_access_token_auth_validation.py b/tests/lib/test_access_token_auth_validation.py new file mode 100644 index 000000000..8d0d0e316 --- /dev/null +++ b/tests/lib/test_access_token_auth_validation.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from typing import Any, cast + +import httpx +import pytest + +from anthropic import AnthropicError +from anthropic.lib.credentials._auth import AccessTokenAuth +from anthropic.lib.credentials._cache import TokenCache + + +class _TokenCacheStub: + def __init__(self, token: Any) -> None: + self.token = token + self.calls = 0 + + def get_token(self) -> str: + self.calls += 1 + return cast(str, self.token) + + +def _auth(token: Any) -> tuple[AccessTokenAuth, _TokenCacheStub]: + cache = _TokenCacheStub(token) + return AccessTokenAuth(cast(TokenCache, cache)), cache + + +@pytest.mark.parametrize("token", ["", " token", "token ", "\ttoken"]) +def test_sync_auth_rejects_invalid_provider_tokens(token: str) -> None: + auth, cache = _auth(token) + request = httpx.Request("GET", "https://api.anthropic.com/v1/models") + + with pytest.raises(AnthropicError, match="invalid access token"): + list(auth.sync_auth_flow(request)) + + assert cache.calls == 1 + assert "Authorization" not in request.headers + + +@pytest.mark.asyncio +@pytest.mark.parametrize("token", ["", " token", "token ", "\ttoken"]) +async def test_async_auth_rejects_invalid_provider_tokens(token: str) -> None: + auth, cache = _auth(token) + request = httpx.Request("GET", "https://api.anthropic.com/v1/models") + + with pytest.raises(AnthropicError, match="invalid access token"): + [item async for item in auth.async_auth_flow(request)] + + assert cache.calls == 1 + assert "Authorization" not in request.headers + + +def test_valid_provider_token_is_applied() -> None: + auth, cache = _auth("access-token") + request = httpx.Request("GET", "https://api.anthropic.com/v1/models") + + yielded = list(auth.sync_auth_flow(request)) + + assert yielded == [request] + assert cache.calls == 1 + assert request.headers["Authorization"] == "Bearer access-token" + assert "oauth-2025-04-20" in request.headers["anthropic-beta"] + + +def test_static_authorization_still_bypasses_provider_validation() -> None: + auth, cache = _auth("") + request = httpx.Request( + "GET", + "https://api.anthropic.com/v1/models", + headers={"Authorization": "Bearer explicit-token"}, + ) + + yielded = list(auth.sync_auth_flow(request)) + + assert yielded == [request] + assert cache.calls == 0 + assert request.headers["Authorization"] == "Bearer explicit-token"