Timmy_Foundation/hermes-agent
15
0
Fork 0
Code Issues 248 Pull Requests 14 Actions 1 Packages Projects Releases Wiki Activity

fix: Anthropic OAuth — beta header, token refresh, config contamination, reauthentication (#1132)

Browse Source 
Fixes Anthropic OAuth/subscription authentication end-to-end:

Auth failures (401 errors):
- Add missing 'claude-code-20250219' beta header for OAuth tokens. Both
  clawdbot and OpenCode include this alongside 'oauth-2025-04-20' — without
  it, Anthropic's API rejects OAuth tokens with 401 authentication errors.
- Fix _fetch_anthropic_models() to use canonical beta headers from
  _COMMON_BETAS + _OAUTH_ONLY_BETAS instead of hardcoding.

Token refresh:
- Add _refresh_oauth_token() — when Claude Code credentials from
  ~/.claude/.credentials.json are expired but have a refresh token,
  automatically POST to console.anthropic.com/v1/oauth/token to get
  a new access token. Uses the same client_id as Claude Code / OpenCode.
- Add _write_claude_code_credentials() — writes refreshed tokens back
  to ~/.claude/.credentials.json, preserving other fields.
- resolve_anthropic_token() now auto-refreshes expired tokens before
  returning None.

Config contamination:
- Anthropic's _model_flow_anthropic() no longer saves base_url to config.
  Since resolve_runtime_provider() always hardcodes Anthropic's URL, the
  stale base_url was contaminating other providers when users switched
  without re-running 'hermes model' (e.g., Codex hitting api.anthropic.com).
- _update_config_for_provider() now pops base_url when passed empty string.
- Same fix in setup.py.

Flow/UX (hermes model command):
- CLAUDE_CODE_OAUTH_TOKEN env var now checked in credential detection
- Reauthentication option when existing credentials found
- run_oauth_setup_token() runs 'claude setup-token' as interactive
  subprocess, then auto-detects saved credentials
- Clean has_creds/needs_auth flow in both main.py and setup.py

Tests (14 new):
- Beta header assertions for claude-code-20250219
- Token refresh: successful refresh with credential writeback, failed
  refresh returns None, no refresh token returns None
- Credential writeback: new file creation, preserving existing fields
- Auto-refresh integration in resolve_anthropic_token()
- CLAUDE_CODE_OAUTH_TOKEN fallback, credential file auto-discovery
- run_oauth_setup_token() (5 scenarios)
This commit is contained in:
Teknium
2026-03-12 20:45:50 -07:00
committed by GitHub
parent 6ceae61a56
commit d24bcad90b
6 changed files with 500 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
agent/anthropic_adapter.py
View File

@@ -46,7 +46,10 @@ _COMMON_BETAS = [
]
# Additional beta headers required for OAuth/subscription auth
# Both clawdbot and OpenCode include claude-code-20250219 alongside oauth-2025-04-20.
# Without claude-code-20250219, Anthropic's API rejects OAuth tokens with 401.
_OAUTH_ONLY_BETAS = [
"claude-code-20250219",
"oauth-2025-04-20",
]
@@ -157,13 +160,91 @@ def is_claude_code_token_valid(creds: Dict[str, Any]) -> bool:
return now_ms < (expires_at - 60_000)
def _refresh_oauth_token(creds: Dict[str, Any]) -> Optional[str]:
"""Attempt to refresh an expired Claude Code OAuth token.
Uses the same token endpoint and client_id as Claude Code / OpenCode.
Only works for credentials that have a refresh token (from claude /login
or claude setup-token with OAuth flow).
Returns the new access token, or None if refresh fails.
"""
import urllib.parse
import urllib.request
refresh_token = creds.get("refreshToken", "")
if not refresh_token:
logger.debug("No refresh token available — cannot refresh")
return None
# Client ID used by Claude Code's OAuth flow
CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e"
data = urllib.parse.urlencode({
"grant_type": "refresh_token",
"refresh_token": refresh_token,
"client_id": CLIENT_ID,
}).encode()
req = urllib.request.Request(
"https://console.anthropic.com/v1/oauth/token",
data=data,
headers={"Content-Type": "application/x-www-form-urlencoded"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
result = json.loads(resp.read().decode())
new_access = result.get("access_token", "")
new_refresh = result.get("refresh_token", refresh_token)
expires_in = result.get("expires_in", 3600) # seconds
if new_access:
import time
new_expires_ms = int(time.time() * 1000) + (expires_in * 1000)
# Write refreshed credentials back to ~/.claude/.credentials.json
_write_claude_code_credentials(new_access, new_refresh, new_expires_ms)
logger.debug("Successfully refreshed Claude Code OAuth token")
return new_access
except Exception as e:
logger.debug("Failed to refresh Claude Code token: %s", e)
return None
def _write_claude_code_credentials(access_token: str, refresh_token: str, expires_at_ms: int) -> None:
"""Write refreshed credentials back to ~/.claude/.credentials.json."""
cred_path = Path.home() / ".claude" / ".credentials.json"
try:
# Read existing file to preserve other fields
existing = {}
if cred_path.exists():
existing = json.loads(cred_path.read_text(encoding="utf-8"))
existing["claudeAiOauth"] = {
"accessToken": access_token,
"refreshToken": refresh_token,
"expiresAt": expires_at_ms,
}
cred_path.parent.mkdir(parents=True, exist_ok=True)
cred_path.write_text(json.dumps(existing, indent=2), encoding="utf-8")
# Restrict permissions (credentials file)
cred_path.chmod(0o600)
except (OSError, IOError) as e:
logger.debug("Failed to write refreshed credentials: %s", e)
def resolve_anthropic_token() -> Optional[str]:
"""Resolve an Anthropic token from all available sources.
Priority:
1. ANTHROPIC_API_KEY env var (regular API key)
2. ANTHROPIC_TOKEN env var (OAuth/setup token)
3. Claude Code credentials (~/.claude.json or ~/.claude/.credentials.json)
3. CLAUDE_CODE_OAUTH_TOKEN env var
4. Claude Code credentials (~/.claude.json or ~/.claude/.credentials.json)
— with automatic refresh if expired and a refresh token is available
Returns the token string or None.
"""
@@ -177,18 +258,63 @@ def resolve_anthropic_token() -> Optional[str]:
if token:
return token
# Also check CLAUDE_CODE_OAUTH_TOKEN (used by Claude Code for setup-tokens)
# 3. CLAUDE_CODE_OAUTH_TOKEN (used by Claude Code for setup-tokens)
cc_token = os.getenv("CLAUDE_CODE_OAUTH_TOKEN", "").strip()
if cc_token:
return cc_token
# 3. Claude Code credential file
# 4. Claude Code credential file
creds = read_claude_code_credentials()
if creds and is_claude_code_token_valid(creds):
logger.debug("Using Claude Code credentials (auto-detected)")
return creds["accessToken"]
elif creds:
logger.debug("Claude Code credentials expired — run 'claude' to refresh")
# Token expired — attempt to refresh
logger.debug("Claude Code credentials expired — attempting refresh")
refreshed = _refresh_oauth_token(creds)
if refreshed:
return refreshed
logger.debug("Token refresh failed — re-run 'claude setup-token' to reauthenticate")
return None
def run_oauth_setup_token() -> Optional[str]:
"""Run 'claude setup-token' interactively and return the resulting token.
Checks multiple sources after the subprocess completes:
1. Claude Code credential files (may be written by the subprocess)
2. CLAUDE_CODE_OAUTH_TOKEN / ANTHROPIC_TOKEN env vars
Returns the token string, or None if no credentials were obtained.
Raises FileNotFoundError if the 'claude' CLI is not installed.
"""
import shutil
import subprocess
claude_path = shutil.which("claude")
if not claude_path:
raise FileNotFoundError(
"The 'claude' CLI is not installed. "
"Install it with: npm install -g @anthropic-ai/claude-code"
)
# Run interactively — stdin/stdout/stderr inherited so user can interact
try:
subprocess.run([claude_path, "setup-token"])
except (KeyboardInterrupt, EOFError):
return None
# Check if credentials were saved to Claude Code's config files
creds = read_claude_code_credentials()
if creds and is_claude_code_token_valid(creds):
return creds["accessToken"]
# Check env vars that may have been set
for env_var in ("CLAUDE_CODE_OAUTH_TOKEN", "ANTHROPIC_TOKEN"):
val = os.getenv(env_var, "").strip()
if val:
return val
return None

6
hermes_cli/auth.py
View File

@@ -1571,7 +1571,11 @@ def _update_config_for_provider(provider_id: str, inference_base_url: str) -> Pa
model_cfg = {}
model_cfg["provider"] = provider_id
model_cfg["base_url"] = inference_base_url.rstrip("/")
if inference_base_url and inference_base_url.strip():
model_cfg["base_url"] = inference_base_url.rstrip("/")
else:
# Clear stale base_url to prevent contamination when switching providers
model_cfg.pop("base_url", None)
config["model"] = model_cfg
config_path.write_text(yaml.safe_dump(config, sort_keys=False))

133
hermes_cli/main.py
View File

@@ -1590,8 +1590,67 @@ def _model_flow_api_key_provider(config, provider_id, current_model=""):
print("No change.")
def _run_anthropic_oauth_flow(save_env_value):
"""Run the Claude OAuth setup-token flow. Returns True if credentials were saved."""
from agent.anthropic_adapter import run_oauth_setup_token
try:
print()
print(" Running 'claude setup-token' — follow the prompts below.")
print(" A browser window will open for you to authorize access.")
print()
token = run_oauth_setup_token()
if token:
save_env_value("ANTHROPIC_API_KEY", token)
print(" ✓ OAuth credentials saved.")
return True
# Subprocess completed but no token auto-detected — ask user to paste
print()
print(" If the setup-token was displayed above, paste it here:")
print()
try:
manual_token = input(" Paste setup-token (or Enter to cancel): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return False
if manual_token:
save_env_value("ANTHROPIC_API_KEY", manual_token)
print(" ✓ Setup-token saved.")
return True
print(" ⚠ Could not detect saved credentials.")
return False
except FileNotFoundError:
# Claude CLI not installed — guide user through manual setup
print()
print(" The 'claude' CLI is required for OAuth login.")
print()
print(" To install and authenticate:")
print()
print(" 1. Install Claude Code: npm install -g @anthropic-ai/claude-code")
print(" 2. Run: claude setup-token")
print(" 3. Follow the browser prompts to authorize")
print(" 4. Re-run: hermes model")
print()
print(" Or paste an existing setup-token now (sk-ant-oat-...):")
print()
try:
token = input(" Setup-token (or Enter to cancel): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return False
if token:
save_env_value("ANTHROPIC_API_KEY", token)
print(" ✓ Setup-token saved.")
return True
print(" Cancelled — install Claude Code and try again.")
return False
def _model_flow_anthropic(config, current_model=""):
"""Flow for Anthropic provider — setup-token, API key, or Claude Code creds."""
"""Flow for Anthropic provider — OAuth subscription, API key, or Claude Code creds."""
import os
from hermes_cli.auth import (
PROVIDER_REGISTRY, _prompt_model_selection, _save_model_choice,
@@ -1602,12 +1661,13 @@ def _model_flow_anthropic(config, current_model=""):
pconfig = PROVIDER_REGISTRY["anthropic"]
# Check for existing credentials
# Check ALL credential sources
existing_key = (
get_env_value("ANTHROPIC_API_KEY")
or os.getenv("ANTHROPIC_API_KEY", "")
or get_env_value("ANTHROPIC_TOKEN")
or os.getenv("ANTHROPIC_TOKEN", "")
or os.getenv("CLAUDE_CODE_OAUTH_TOKEN", "")
)
cc_available = False
try:
@@ -1618,27 +1678,37 @@ def _model_flow_anthropic(config, current_model=""):
except Exception:
pass
if existing_key:
print(f" Anthropic credentials: {existing_key[:12]}... ✓")
has_creds = bool(existing_key) or cc_available
needs_auth = not has_creds
if has_creds:
# Show what we found
if existing_key:
print(f" Anthropic credentials: {existing_key[:12]}... ✓")
elif cc_available:
print(" Claude Code credentials: ✓ (auto-detected)")
print()
print(" 1. Use existing credentials")
print(" 2. Reauthenticate (new OAuth login)")
print(" 3. Cancel")
print()
try:
update = input("Update credentials? [y/N]: ").strip().lower()
choice = input(" Choice [1/2/3]: ").strip()
except (KeyboardInterrupt, EOFError):
update = ""
if update != "y":
pass # skip to model selection
else:
existing_key = "" # fall through to auth choice below
elif cc_available:
print(" Claude Code credentials: ✓ (auto-detected)")
print()
if not existing_key and not cc_available:
# No credentials — show auth method choice
choice = "1"
if choice == "2":
needs_auth = True
elif choice == "3":
return
# choice == "1" or default: use existing, proceed to model selection
if needs_auth:
# Show auth method choice
print()
print(" Choose authentication method:")
print()
print(" 1. Claude Pro/Max subscription (setup-token)")
print(" 1. Claude Pro/Max subscription (OAuth login)")
print(" 2. Anthropic API key (pay-per-token)")
print(" 3. Cancel")
print()
@@ -1649,33 +1719,15 @@ def _model_flow_anthropic(config, current_model=""):
return
if choice == "1":
print()
print(" To get a setup-token from your Claude subscription:")
print()
print(" 1. Install Claude Code: npm install -g @anthropic-ai/claude-code")
print(" 2. Run: claude setup-token")
print(" 3. Open the URL it prints in your browser")
print(" 4. Log in and click \"Authorize\"")
print(" 5. Paste the auth code back into Claude Code")
print(" 6. Copy the resulting sk-ant-oat01-... token")
print()
try:
token = input(" Paste setup-token here: ").strip()
except (KeyboardInterrupt, EOFError):
print()
if not _run_anthropic_oauth_flow(save_env_value):
return
if not token:
print(" Cancelled.")
return
save_env_value("ANTHROPIC_API_KEY", token)
print(" ✓ Setup-token saved.")
elif choice == "2":
print()
print(" Get an API key at: https://console.anthropic.com/settings/keys")
print()
try:
api_key = input(" API key (sk-ant-api03-...): ").strip()
api_key = input(" API key (sk-ant-...): ").strip()
except (KeyboardInterrupt, EOFError):
print()
return
@@ -1708,14 +1760,17 @@ def _model_flow_anthropic(config, current_model=""):
_save_model_choice(selected)
# Update config with provider
# Update config with provider — clear base_url since
# resolve_runtime_provider() always hardcodes Anthropic's URL.
# Leaving a stale base_url in config can contaminate other
# providers if the user switches without running 'hermes model'.
cfg = load_config()
model = cfg.get("model")
if not isinstance(model, dict):
model = {"default": model} if model else {}
cfg["model"] = model
model["provider"] = "anthropic"
model["base_url"] = pconfig.inference_base_url
model.pop("base_url", None)
save_config(cfg)
deactivate_provider()

3
hermes_cli/models.py
View File

@@ -271,7 +271,8 @@ def _fetch_anthropic_models(timeout: float = 5.0) -> Optional[list[str]]:
headers: dict[str, str] = {"anthropic-version": "2023-06-01"}
if _is_oauth_token(token):
headers["Authorization"] = f"Bearer {token}"
headers["anthropic-beta"] = "oauth-2025-04-20"
from agent.anthropic_adapter import _COMMON_BETAS, _OAUTH_ONLY_BETAS
headers["anthropic-beta"] = ",".join(_COMMON_BETAS + _OAUTH_ONLY_BETAS)
else:
headers["x-api-key"] = token

116
hermes_cli/setup.py
View File

@@ -1076,65 +1076,101 @@ def setup_model_provider(config: dict):
from hermes_cli.auth import PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY["anthropic"]
# Check for Claude Code credential auto-discovery
from agent.anthropic_adapter import read_claude_code_credentials, is_claude_code_token_valid
# Check ALL credential sources
import os as _os
from agent.anthropic_adapter import (
read_claude_code_credentials, is_claude_code_token_valid,
run_oauth_setup_token,
)
cc_creds = read_claude_code_credentials()
if cc_creds and is_claude_code_token_valid(cc_creds):
print_success("Found valid Claude Code credentials (~/.claude/.credentials.json)")
if prompt_yes_no("Use these credentials?", True):
print_success("Using Claude Code subscription credentials")
else:
cc_creds = None
cc_valid = bool(cc_creds and is_claude_code_token_valid(cc_creds))
existing_key = get_env_value("ANTHROPIC_API_KEY") or get_env_value("ANTHROPIC_TOKEN")
existing_key = (
get_env_value("ANTHROPIC_API_KEY")
or get_env_value("ANTHROPIC_TOKEN")
or _os.getenv("CLAUDE_CODE_OAUTH_TOKEN", "")
)
if not (cc_creds and is_claude_code_token_valid(cc_creds)):
has_creds = bool(existing_key) or cc_valid
needs_auth = not has_creds
if has_creds:
if existing_key:
print_info(f"Current credentials: {existing_key[:12]}...")
if not prompt_yes_no("Update credentials?", False):
# User wants to keep existing — skip auth prompt entirely
existing_key = "KEEP" # truthy sentinel to skip auth choice
elif cc_valid:
print_success("Found valid Claude Code credentials (auto-detected)")
if not existing_key and not (cc_creds and is_claude_code_token_valid(cc_creds)):
auth_choices = [
"Claude Pro/Max subscription (setup-token)",
"Anthropic API key (pay-per-token)",
]
auth_idx = prompt_choice("Choose authentication method:", auth_choices, 0)
auth_choices = [
"Use existing credentials",
"Reauthenticate (new OAuth login)",
"Cancel",
]
choice_idx = prompt_choice("What would you like to do?", auth_choices, 0)
if choice_idx == 1:
needs_auth = True
elif choice_idx == 2:
pass # fall through to provider config
if auth_idx == 0:
if needs_auth:
auth_choices = [
"Claude Pro/Max subscription (OAuth login)",
"Anthropic API key (pay-per-token)",
]
auth_idx = prompt_choice("Choose authentication method:", auth_choices, 0)
if auth_idx == 0:
# OAuth setup-token flow
try:
print()
print_info("To get a setup-token from your Claude subscription:")
print_info(" 1. Install Claude Code: npm install -g @anthropic-ai/claude-code")
print_info(" 2. Run: claude setup-token")
print_info(" 3. Open the URL it prints in your browser")
print_info(" 4. Log in and click \"Authorize\"")
print_info(" 5. Paste the auth code back into Claude Code")
print_info(" 6. Copy the resulting sk-ant-oat01-... token")
print_info("Running 'claude setup-token' — follow the prompts below.")
print_info("A browser window will open for you to authorize access.")
print()
token = prompt("Paste setup-token here", password=True)
token = run_oauth_setup_token()
if token:
save_env_value("ANTHROPIC_API_KEY", token)
print_success("OAuth credentials saved")
else:
# Subprocess completed but no token auto-detected
print()
token = prompt("Paste setup-token here (if displayed above)", password=True)
if token:
save_env_value("ANTHROPIC_API_KEY", token)
print_success("Setup-token saved")
else:
print_warning("Skipped — agent won't work without credentials")
except FileNotFoundError:
print()
print_info("The 'claude' CLI is required for OAuth login.")
print()
print_info("To install: npm install -g @anthropic-ai/claude-code")
print_info("Then run: claude setup-token")
print_info("Or paste an existing setup-token below:")
print()
token = prompt("Setup-token (sk-ant-oat-...)", password=True)
if token:
save_env_value("ANTHROPIC_API_KEY", token)
print_success("Setup-token saved")
else:
print_warning("Skipped — agent won't work without credentials")
print_warning("Skipped — install Claude Code and re-run setup")
else:
print()
print_info("Get an API key at: https://console.anthropic.com/settings/keys")
print()
api_key = prompt("API key (sk-ant-...)", password=True)
if api_key:
save_env_value("ANTHROPIC_API_KEY", api_key)
print_success("API key saved")
else:
print()
print_info("Get an API key at: https://console.anthropic.com/settings/keys")
print()
api_key = prompt("API key (sk-ant-api03-...)", password=True)
if api_key:
save_env_value("ANTHROPIC_API_KEY", api_key)
print_success("API key saved")
else:
print_warning("Skipped — agent won't work without credentials")
print_warning("Skipped — agent won't work without credentials")
# Clear custom endpoint vars if switching
if existing_custom:
save_env_value("OPENAI_BASE_URL", "")
save_env_value("OPENAI_API_KEY", "")
_update_config_for_provider("anthropic", pconfig.inference_base_url)
_set_model_provider(config, "anthropic", pconfig.inference_base_url)
# Don't save base_url for Anthropic — resolve_runtime_provider()
# always hardcodes it. Stale base_urls contaminate other providers.
_update_config_for_provider("anthropic", "")
_set_model_provider(config, "anthropic")
# else: provider_idx == 9 (Keep current) — only shown when a provider already exists

193
tests/test_anthropic_adapter.py
View File

@@ -9,6 +9,8 @@ import pytest
from agent.anthropic_adapter import (
_is_oauth_token,
_refresh_oauth_token,
_write_claude_code_credentials,
build_anthropic_client,
build_anthropic_kwargs,
convert_messages_to_anthropic,
@@ -18,6 +20,7 @@ from agent.anthropic_adapter import (
normalize_model_name,
read_claude_code_credentials,
resolve_anthropic_token,
run_oauth_setup_token,
)
@@ -53,6 +56,7 @@ class TestBuildAnthropicClient:
assert "auth_token" in kwargs
betas = kwargs["default_headers"]["anthropic-beta"]
assert "oauth-2025-04-20" in betas
assert "claude-code-20250219" in betas
assert "interleaved-thinking-2025-05-14" in betas
assert "fine-grained-tool-streaming-2025-05-14" in betas
assert "api_key" not in kwargs
@@ -67,6 +71,7 @@ class TestBuildAnthropicClient:
betas = kwargs["default_headers"]["anthropic-beta"]
assert "interleaved-thinking-2025-05-14" in betas
assert "oauth-2025-04-20" not in betas # OAuth-only beta NOT present
assert "claude-code-20250219" not in betas # OAuth-only beta NOT present
def test_custom_base_url(self):
with patch("agent.anthropic_adapter._anthropic_sdk") as mock_sdk:
@@ -145,6 +150,194 @@ class TestResolveAnthropicToken:
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
assert resolve_anthropic_token() is None
def test_falls_back_to_claude_code_oauth_token(self, monkeypatch, tmp_path):
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False)
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "sk-ant-oat01-test-token")
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
assert resolve_anthropic_token() == "sk-ant-oat01-test-token"
def test_falls_back_to_claude_code_credentials(self, monkeypatch, tmp_path):
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False)
monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False)
cred_file = tmp_path / ".claude" / ".credentials.json"
cred_file.parent.mkdir(parents=True)
cred_file.write_text(json.dumps({
"claudeAiOauth": {
"accessToken": "cc-auto-token",
"refreshToken": "refresh",
"expiresAt": int(time.time() * 1000) + 3600_000,
}
}))
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
assert resolve_anthropic_token() == "cc-auto-token"
class TestRefreshOauthToken:
def test_returns_none_without_refresh_token(self):
creds = {"accessToken": "expired", "refreshToken": "", "expiresAt": 0}
assert _refresh_oauth_token(creds) is None
def test_successful_refresh(self, tmp_path, monkeypatch):
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
creds = {
"accessToken": "old-token",
"refreshToken": "refresh-123",
"expiresAt": int(time.time() * 1000) - 3600_000,
}
mock_response = json.dumps({
"access_token": "new-token-abc",
"refresh_token": "new-refresh-456",
"expires_in": 7200,
}).encode()
with patch("urllib.request.urlopen") as mock_urlopen:
mock_ctx = MagicMock()
mock_ctx.__enter__ = MagicMock(return_value=MagicMock(
read=MagicMock(return_value=mock_response)
))
mock_ctx.__exit__ = MagicMock(return_value=False)
mock_urlopen.return_value = mock_ctx
result = _refresh_oauth_token(creds)
assert result == "new-token-abc"
# Verify credentials were written back
cred_file = tmp_path / ".claude" / ".credentials.json"
assert cred_file.exists()
written = json.loads(cred_file.read_text())
assert written["claudeAiOauth"]["accessToken"] == "new-token-abc"
assert written["claudeAiOauth"]["refreshToken"] == "new-refresh-456"
def test_failed_refresh_returns_none(self):
creds = {
"accessToken": "old",
"refreshToken": "refresh-123",
"expiresAt": 0,
}
with patch("urllib.request.urlopen", side_effect=Exception("network error")):
assert _refresh_oauth_token(creds) is None
class TestWriteClaudeCodeCredentials:
def test_writes_new_file(self, tmp_path, monkeypatch):
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
_write_claude_code_credentials("tok", "ref", 12345)
cred_file = tmp_path / ".claude" / ".credentials.json"
assert cred_file.exists()
data = json.loads(cred_file.read_text())
assert data["claudeAiOauth"]["accessToken"] == "tok"
assert data["claudeAiOauth"]["refreshToken"] == "ref"
assert data["claudeAiOauth"]["expiresAt"] == 12345
def test_preserves_existing_fields(self, tmp_path, monkeypatch):
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
cred_dir = tmp_path / ".claude"
cred_dir.mkdir()
cred_file = cred_dir / ".credentials.json"
cred_file.write_text(json.dumps({"otherField": "keep-me"}))
_write_claude_code_credentials("new-tok", "new-ref", 99999)
data = json.loads(cred_file.read_text())
assert data["otherField"] == "keep-me"
assert data["claudeAiOauth"]["accessToken"] == "new-tok"
class TestResolveWithRefresh:
def test_auto_refresh_on_expired_creds(self, monkeypatch, tmp_path):
"""When cred file has expired token + refresh token, auto-refresh is attempted."""
monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False)
monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False)
# Set up expired creds with a refresh token
cred_file = tmp_path / ".claude" / ".credentials.json"
cred_file.parent.mkdir(parents=True)
cred_file.write_text(json.dumps({
"claudeAiOauth": {
"accessToken": "expired-tok",
"refreshToken": "valid-refresh",
"expiresAt": int(time.time() * 1000) - 3600_000,
}
}))
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
# Mock refresh to succeed
with patch("agent.anthropic_adapter._refresh_oauth_token", return_value="refreshed-token"):
result = resolve_anthropic_token()
assert result == "refreshed-token"
class TestRunOauthSetupToken:
def test_raises_when_claude_not_installed(self, monkeypatch):
monkeypatch.setattr("shutil.which", lambda _: None)
with pytest.raises(FileNotFoundError, match="claude.*CLI.*not installed"):
run_oauth_setup_token()
def test_returns_token_from_credential_files(self, monkeypatch, tmp_path):
"""After subprocess completes, reads credentials from Claude Code files."""
monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False)
monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False)
# Pre-create credential files that will be found after subprocess
cred_file = tmp_path / ".claude" / ".credentials.json"
cred_file.parent.mkdir(parents=True)
cred_file.write_text(json.dumps({
"claudeAiOauth": {
"accessToken": "from-cred-file",
"refreshToken": "refresh",
"expiresAt": int(time.time() * 1000) + 3600_000,
}
}))
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
token = run_oauth_setup_token()
assert token == "from-cred-file"
mock_run.assert_called_once()
def test_returns_token_from_env_var(self, monkeypatch, tmp_path):
"""Falls back to CLAUDE_CODE_OAUTH_TOKEN env var when no cred files."""
monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "from-env-var")
monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False)
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
token = run_oauth_setup_token()
assert token == "from-env-var"
def test_returns_none_when_no_creds_found(self, monkeypatch, tmp_path):
"""Returns None when subprocess completes but no credentials are found."""
monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False)
monkeypatch.delenv("ANTHROPIC_TOKEN", raising=False)
monkeypatch.setattr("agent.anthropic_adapter.Path.home", lambda: tmp_path)
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0)
token = run_oauth_setup_token()
assert token is None
def test_returns_none_on_keyboard_interrupt(self, monkeypatch):
"""Returns None gracefully when user interrupts the flow."""
monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/claude")
with patch("subprocess.run", side_effect=KeyboardInterrupt):
token = run_oauth_setup_token()
assert token is None
# ---------------------------------------------------------------------------
# Model name normalization