Validate provider access tokens before auth

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

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"