fix(anthropic): clear SDK env-derived auth_token on api-key-style clients

Anthropic SDK fills an unset auth_token from ANTHROPIC_AUTH_TOKEN and sends
it as a Bearer credential alongside x-api-key to third-party
Anthropic-compatible endpoints. Mirror the existing bearer-side guard in
_new_sdk_client so api-key-style construction clears auth_token after
construction, and cover it with regression tests (client attributes, header
shape, and an end-to-end local-server capture).
This commit is contained in:
Konstantin Khlopkov
2026-09-08 12:21:22 +00:00
committed by kshitij
parent 3a76fe8e67
commit 174eff3f4c
2 changed files with 97 additions and 2 deletions

View File

@@ -345,13 +345,18 @@ def _build_anthropic_client_with_bearer_hook(
def _new_sdk_client(sdk, kwargs: Dict[str, Any], headers: Dict[str, str]):
"""``sdk.Anthropic(**kwargs)`` with ``headers`` attached. Bearer-only construction leaves
``api_key`` unset, so the SDK fills it from ANTHROPIC_API_KEY (loaded from ~/.hermes/.env) and
sends dual auth — X-Api-Key *and* Authorization: Bearer — on every Portal/MiniMax/OAuth/Entra
request; clear it whenever we intentionally authenticated via auth_token."""
sends dual auth — X-Api-Key *and* Authorization: Bearer *** on every Portal/MiniMax/OAuth/Entra
request; clear it whenever we intentionally authenticated via auth_token. Api-key-only
construction has the mirror problem: the SDK fills ``auth_token`` from ANTHROPIC_AUTH_TOKEN in
the environment and ships that Bearer credential to third-party Anthropic-compatible endpoints
alongside x-api-key — clear it whenever we intentionally authenticated via api_key."""
if headers:
kwargs["default_headers"] = headers
client = sdk.Anthropic(**kwargs)
if "auth_token" in kwargs and "api_key" not in kwargs:
client.api_key = None
if "api_key" in kwargs and "auth_token" not in kwargs:
client.auth_token = None
return client

View File

@@ -1896,3 +1896,93 @@ class TestFinalPayloadHasNoBlankTextBlocks:
)
image_blocks = [b for b in tool_result_block["content"] if b.get("type") == "image"]
assert len(image_blocks) == 1
class TestApiKeyConstructionClearsEnvBearerToken:
"""Api-key-style clients must not inherit ANTHROPIC_AUTH_TOKEN from the environment.
The Anthropic SDK fills an unset ``auth_token`` from ANTHROPIC_AUTH_TOKEN and then sends
``Authorization: Bearer ***`` alongside ``x-api-key`` on every request, shipping a
foreign shell credential to third-party Anthropic-compatible endpoints.
"""
def test_api_key_style_client_drops_env_derived_auth_token(self, monkeypatch):
anthropic_sdk = pytest.importorskip("anthropic")
monkeypatch.setenv("ANTHROPIC_AUTH_TOKEN", "sentinel-env-token-DO-NOT-SEND")
from agent.anthropic_adapter import _new_sdk_client
client = _new_sdk_client(
anthropic_sdk,
{"api_key": "provider-key", "base_url": "http://127.0.0.1:1"},
{},
)
assert client.api_key == "provider-key"
assert client.auth_token is None
assert "Authorization" not in client.auth_headers
assert client.auth_headers == {"X-Api-Key": "provider-key"}
def test_bearer_style_client_keeps_its_auth_token(self, monkeypatch):
anthropic_sdk = pytest.importorskip("anthropic")
monkeypatch.setenv("ANTHROPIC_API_KEY", "sentinel-env-key-DO-NOT-SEND")
from agent.anthropic_adapter import _new_sdk_client
client = _new_sdk_client(
anthropic_sdk,
{"auth_token": "bearer-secret", "base_url": "http://127.0.0.1:1"},
{},
)
assert client.auth_token == "bearer-secret"
assert client.api_key is None
assert client.auth_headers == {"Authorization": "Bearer bearer-secret"}
def test_third_party_request_carries_no_foreign_bearer(self, monkeypatch):
"""End-to-end over a local header-capturing server: x-api-key intact, sentinel absent."""
anthropic_sdk = pytest.importorskip("anthropic")
http_server = pytest.importorskip("http.server")
monkeypatch.setenv("ANTHROPIC_AUTH_TOKEN", "sentinel-env-token-DO-NOT-SEND")
import json as _json
import threading
captured = {}
class _Handler(http_server.BaseHTTPRequestHandler):
def do_POST(self):
captured["headers"] = {k.lower(): v for k, v in self.headers.items()}
self.rfile.read(int(self.headers.get("content-length", 0)))
body = _json.dumps({
"id": "msg_test",
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "ok"}],
"model": "test",
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
}).encode()
self.send_response(200)
self.send_header("content-type", "application/json")
self.send_header("content-length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *args):
pass
server = http_server.HTTPServer(("127.0.0.1", 0), _Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
try:
client = build_anthropic_client(
"third-party-provider-key",
base_url=f"http://127.0.0.1:{server.server_port}",
)
client.messages.create(
model="test-model",
max_tokens=8,
messages=[{"role": "user", "content": "hi"}],
)
finally:
server.shutdown()
headers = captured["headers"]
assert headers.get("x-api-key") == "third-party-provider-key"
assert "sentinel-env-token-DO-NOT-SEND" not in headers.get("authorization", "")