hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 29 Packages Projects Releases Wiki Activity

test(anthropic): one wire_headers probe; the bearer mirror lives only in the portal wire test

Browse Source 
The SDK-private _build_headers probe was spelled three times across two files; the api-key shard also re-asserted the bearer invariant the portal test owns. dict(headers or {}) → dict(headers): every caller passes _beta_header's dict.
This commit is contained in:
kshitijk4poor
2026-09-11 12:43:47 +05:30
committed by kshitij
parent ed2dd27e85
commit 036a20b3ca
3 changed files with 14 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
agent/anthropic_adapter.py
View File

@@ -351,7 +351,7 @@ def _new_sdk_client(sdk, kwargs: Dict[str, Any], headers: Dict[str, str]):
/ OAuth / Entra / third-party endpoints (#26970, #105774). An ``Omit()`` default header is the
SDK-sanctioned way to drop the other header, and unlike an attribute clear it survives
``with_options()``, which re-runs the constructor and re-reads the environment."""
merged = dict(headers or {})
merged = dict(headers)
if "api_key" in kwargs and "auth_token" not in kwargs:
merged["Authorization"] = sdk.Omit()
elif "auth_token" in kwargs and "api_key" not in kwargs:

24
tests/agent/test_anthropic_client_auth.py
View File

@@ -14,30 +14,28 @@ from agent.anthropic_adapter import build_anthropic_client
SENTINEL = "sentinel-env-token-DO-NOT-SEND"
def wire_headers(client) -> dict:
"""Headers the SDK would put on a /v1/messages POST (the Omit() default is resolved here)."""
from anthropic._models import FinalRequestOptions
return dict(client._build_headers(FinalRequestOptions(method="post", url="/v1/messages", json_data={})))
def test_api_key_client_and_its_copies_never_carry_the_env_bearer(monkeypatch):
"""The guard is a copy-safe Omit() default header: ``with_options()`` re-runs the constructor
and re-reads ANTHROPIC_AUTH_TOKEN, so an attribute clear alone would re-leak on the copy."""
anthropic_sdk = pytest.importorskip("anthropic")
from anthropic._models import FinalRequestOptions
monkeypatch.setenv("ANTHROPIC_AUTH_TOKEN", SENTINEL)
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"}, {})
for wire_client in (client, client.with_options(timeout=30)):
headers = dict(wire_client._build_headers(
FinalRequestOptions(method="post", url="/v1/messages", json_data={})))
headers = wire_headers(wire_client)
assert headers.get("x-api-key") == "provider-key"
assert "authorization" not in headers
# Mirror case: bearer-style construction must not ship an env ANTHROPIC_API_KEY, on copies either.
monkeypatch.setenv("ANTHROPIC_API_KEY", "sentinel-env-key-DO-NOT-SEND")
bearer = _new_sdk_client(anthropic_sdk, {"auth_token": "bearer-secret", "base_url": "http://127.0.0.1:1"}, {})
for wire_client in (bearer, bearer.with_options(timeout=30)):
headers = dict(wire_client._build_headers(
FinalRequestOptions(method="post", url="/v1/messages", json_data={})))
assert headers.get("authorization") == "Bearer bearer-secret"
assert "x-api-key" not in headers
# The bearer mirror (no env x-api-key beside a portal JWT) is owned by
# tests/agent/test_nous_portal_anthropic_wire.py::TestClientShape.
def test_third_party_request_on_the_wire_carries_no_foreign_bearer(monkeypatch):

5
tests/agent/test_nous_portal_anthropic_wire.py
View File

@@ -212,14 +212,13 @@ class TestClientShape:
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-should-not-leak")
client = build_anthropic_client("portal-invoke-jwt", PORTAL_URL)
from anthropic._models import FinalRequestOptions
from tests.agent.test_anthropic_client_auth import wire_headers
assert client.auth_token == "portal-invoke-jwt"
# The guard is a copy-safe Omit() default header, so assert what reaches the wire —
# on the client and on a with_options() copy (which re-reads ANTHROPIC_API_KEY).
for wire_client in (client, client.with_options(timeout=30)):
headers = dict(wire_client._build_headers(
FinalRequestOptions(method="post", url="/v1/messages", json_data={})))
headers = wire_headers(wire_client)
assert "x-api-key" not in headers
assert headers.get("authorization", "").startswith("Bearer portal-invoke-jwt")