Files
Aperant/apps/backend/core/auth.py
T
Andy bfc232825b feat(pr-review): evidence-based validation and trigger-driven exploration (#1593)
* gsd update

* docs: define v1 requirements with holistic PR understanding

29 requirements across 6 categories:
- Holistic PR Understanding (5) - context synthesis and passing
- Validation Pipeline (3) - finding-validator for all reviews
- Schema Enforcement (5) - VerificationEvidence required
- Prompt Improvements (6) - understand intent, evidence requirements
- Code Simplification (6) - remove programmatic filters
- Measurement (4) - 5 PRs to validate

Key addition: Pass gathered context (related files, import graph) to specialists.
Currently gathered but unused.

* feat(01-01): add Phase 0 synthesis instruction to orchestrator prompt

- Add 'Phase 0: Understand the PR Holistically' section before Phase 1
- Include PR UNDERSTANDING output format (intent, critical changes, risk areas, files to verify)
- Add explicit gate: 'Only AFTER completing Phase 0, proceed to Phase 1'
- Add 'Understand First' principle to Key Principles section

Covers: CONTEXT-01, CONTEXT-05

* feat(01-01): add related files and import graph to orchestrator prompt

- Add related files section categorizing tests vs dependencies/callers
- Add import graph section showing what files import/are imported by changed files
- Limit to 30 related files (15 tests, 15 deps) and 20 import entries
- Include actionable guidance for using the context

Covers: CONTEXT-02, CONTEXT-03

* feat(01-02): add investigation context to specialist agent descriptions

- security-reviewer: check related files for affected callers, verify tests
- quality-reviewer: check related files for pattern consistency
- logic-reviewer: check callers/dependents for broken assumptions
- codebase-fit-reviewer: use related files to understand existing patterns
- finding-validator: check related files for missed mitigations
- ai-triage-reviewer: unchanged (doesn't need related file guidance)

CONTEXT-04: Specialists now know which files to investigate beyond the diff

* feat(01-02): add specialist-specific delegation guidance to related files section

- Updated header: "Pass relevant files to specialists when delegating"
- Added per-specialist guidance for security, logic, quality, codebase-fit
- Added example delegation showing how to include related files in task

Orchestrator now knows HOW to pass investigation context to each specialist type

* feat(02-01): add VerificationEvidence class and update finding models

- Add VerificationEvidence class with required code_examined, line_range_examined, verification_method fields
- Add required verification field to BaseFinding
- Add required verification field to ParallelOrchestratorFinding
- Add is_impact_finding boolean field to ParallelOrchestratorFinding (default False)
- Add checked_for_handling_elsewhere boolean field to ParallelOrchestratorFinding (default False)
- Mark old evidence field as DEPRECATED in both BaseFinding and ParallelOrchestratorFinding

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* test(02-01): add tests for schema enforcement and verification evidence

- Add TestVerificationEvidence class with 5 tests for VerificationEvidence model
- Add TestParallelOrchestratorFindingVerification class with 6 tests for verification requirement
- Add TestVerificationSchemaGeneration class with 2 tests for JSON schema generation
- Update existing TestSecurityFinding and TestDeepAnalysisFinding to include verification field
- Import VerificationEvidence, ParallelOrchestratorFinding, BaseFinding in test imports

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat(03-02): add 'What the Diff Is For' section to orchestrator

- Reframe diff as question to investigate, not document to nitpick
- Add 3 questions to answer before delegation
- Include 'Delegate with Context' guidance
- Position after Phase 0, before Phase 1

* feat(03-01): add Understand Intent phase to all specialist prompts

- Add Phase 1: Understand the PR Intent to security, logic, quality, codebase_fit agents
- Force AI to understand PR purpose before searching for issues
- Prevents flagging intentional design decisions as bugs

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat(03-02): enhance delegation guidance with context requirements

- Add Context-Rich Delegation section with 3 requirements
- Include PR intent summary, specific concerns, files of interest
- Show anti-pattern vs good pattern comparison
- Update example delegation with specific verification items

* feat(03-01): add Evidence Requirements and Valid Outputs sections

- Add Evidence Requirements section documenting VerificationEvidence schema
- Document code_examined, line_range_examined, verification_method fields
- Document is_impact_finding and checked_for_handling_elsewhere fields
- Add Valid Outputs section allowing no-issues as valid output
- Document invalid outputs (forced issues, theoretical edge cases)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat(03-01): update output format examples with verification object

- Add verification object with code_examined, line_range_examined, verification_method
- Add is_impact_finding and checked_for_handling_elsewhere fields
- Use domain-appropriate verification_method values per agent
- Security: direct_code_inspection for injection examples
- Logic: direct_code_inspection for off-by-one and race conditions
- Quality: direct_code_inspection + cross_file_trace for duplication
- Codebase fit: cross_file_trace for reinvention, direct_code_inspection for naming

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat(04-01): add _verify_line_numbers() method

- Pre-filter findings with invalid line numbers before AI validation
- Cache file line counts to avoid re-reading same file
- Reject findings where line > file length
- Log each rejection with finding ID and reason
- Conservative: allow findings if file read fails

* feat(04-02): add hypothesis-validation structure to finding validator

- Add "Hypothesis-Validation Structure (MANDATORY)" section with 4 steps
- Define TRUE/FALSE conditions for hypothesis testing
- Include worked example showing confirmed_valid conclusion path
- Include counter-example showing dismissed_false_positive path
- Reference structure from Investigation Process section

* feat(04-01): add _validate_findings() method

- Import FindingValidationResponse from pydantic_models
- Create finding-validator agent client with pr_finding_validator type
- Build validation prompt with findings JSON and changed files
- Filter findings by validation_status:
  - confirmed_valid: keep with validation evidence
  - dismissed_false_positive: exclude from results
  - needs_human_review: keep with [NEEDS REVIEW] prefix
- Fail-safe: return original findings on any error
- Log validation statistics

* feat(04-01): wire validation pipeline into review() method

- Stage 1: Line verification after cross-validation (cheap pre-filter)
- Stage 2: AI validation for findings that pass line check
- Update programmatic filter loop to use validated_by_ai
- Log validation statistics at each stage
- Uses project_root (worktree or fallback) for file access

* refactor(05-01): remove evidence filter and confidence routing from review()

- Remove _validate_finding_evidence() call from loop
- Remove _apply_confidence_routing() call
- Simplify loop to only check scope
- Replace routed_findings with direct validated_findings assignment

* feat(05-02): remove false positive patterns from validator

- Remove VAGUE_PATTERNS constant (10 patterns)
- Remove GENERIC_PATTERNS constant (6 patterns)
- Remove _is_false_positive() method (44 lines)
- Remove _is_false_positive call from _is_valid()
- Remove TestFalsePositiveDetection class (4 tests)
- Update test_low_severity_higher_threshold to use actionability score

REMOVE-04: VAGUE_PATTERNS, GENERIC_PATTERNS deleted
REMOVE-05: _is_false_positive() method deleted

* refactor(05-01): remove redundant functions and simplify scope check

- Remove ConfidenceTier enum (no longer used)
- Remove _validate_finding_evidence function (schema enforces evidence)
- Remove _apply_confidence_routing method (validation is binary)
- Remove 'from enum import Enum' import
- Simplify _is_finding_in_scope to use schema field is_impact_finding
  instead of keyword detection

* fix(pr-review): add Task tool to orchestrator configs for SDK subagents

The pr_orchestrator_parallel and pr_followup_parallel agents need the
Task tool in their tools list to invoke SDK subagents (security-reviewer,
logic-reviewer, etc.). Without Task, the SDK cannot spawn subagents,
resulting in "Agent type not found" errors.

Also fixes test_integration_phase4.py to set is_impact_finding as an
attribute rather than constructor arg, since PRReviewFinding doesn't
have this field (it's on ParallelOrchestratorFinding Pydantic model).

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(pr-review): add explicit Task tool invocation syntax for specialist agents

The orchestrator was using the built-in general-purpose agent instead of
our custom specialist agents (security-reviewer, logic-reviewer, etc.)
because the prompt described agents but didn't show explicit Task tool
invocation syntax.

Changes:
- Add "CRITICAL: How to Invoke Specialist Agents" section with exact
  subagent_type values in a reference table
- Add Task tool invocation format with example syntax
- Add example showing parallel invocation of multiple specialists
- Add explicit "DO NOT USE" section warning against general-purpose
- Update example delegation to use Task tool syntax instead of prose
- Add example validation invocation for finding-validator

This ensures Claude uses our custom specialists instead of defaulting
to the built-in general-purpose agent.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat(pr-review): implement evidence-based validation and trigger-driven exploration

Major enhancements to the PR review system:

**Evidence-Based Validation:**
- Shift from confidence-based to evidence-based finding validation
- All findings now require VerificationEvidence with code_examined, line_range_examined
- finding-validator validates ALL findings (CRITICAL through LOW) before output
- Add dismissed_findings array for transparency - users see what was investigated

**Trigger-Driven Exploration (6 Semantic Triggers):**
- OUTPUT CONTRACT CHANGED - function returns different value/type/structure
- INPUT CONTRACT CHANGED - parameters added/removed/reordered
- BEHAVIORAL CONTRACT CHANGED - same I/O but different internal behavior
- SIDE EFFECT CONTRACT CHANGED - observable effects added/removed
- FAILURE CONTRACT CHANGED - error handling changed
- NULL/UNDEFINED CONTRACT CHANGED - null handling changed

Orchestrator detects triggers in Phase 1 and passes them to specialists
with explicit "TRIGGER:", "EXPLORATION REQUIRED:", "Stop when:" instructions.

**Implementation Changes:**
- Add _PRDebugLogger for comprehensive agent communication logging
- Add CI status integration to verdict logic (failing CI blocks merge)
- Extract with_working_dir() to shared agent_utils.py module
- Inject working directory into all subagent prompts
- Bump SDK requirement to >=0.1.22 for custom subagent support

**Frontend:**
- Tighten AUTH_FAILURE_PATTERNS to avoid false positives on AI auth discussion
- Update tests for new pattern requirements

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(pr-review): wait for both queued AND in_progress CI checks

Previously, the CI wait logic only blocked on "in_progress" checks,
but not "queued" checks. This meant if a CI check (like CodeRabbit)
was queued but not yet running, the review would start immediately
and report "CI is pending" - which would be stale by the time the
contributor sees it.

Now we wait for ALL checks to reach "completed" status before
starting the review, ensuring the CI status in our review is accurate.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* chore: remove duplicate .planning entries from .gitignore

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* chore: remove docs/ from git tracking (already in .gitignore)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(pr-review): propagate is_impact_finding field to allow impact findings

The is_impact_finding field was defined in ParallelOrchestratorFinding
but never propagated to PRReviewFinding, causing ALL impact findings
(findings about callers/affected files outside the PR's changed files)
to be incorrectly filtered out as "not in scope".

Changes:
- Add is_impact_finding field to PRReviewFinding dataclass
- Extract and pass is_impact_finding in _create_finding_from_structured()
- Add to to_dict() and from_dict() for serialization

This enables the trigger-driven exploration feature to actually work,
allowing the review to report issues in files affected by contract changes.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(pr-review): add Task tool invocation syntax to followup orchestrator

The follow-up review orchestrator was missing explicit Task tool
invocation syntax and examples. The AI didn't know HOW to invoke
the specialist agents (resolution-verifier, finding-validator, etc.),
causing resolution checking to never happen.

Added:
- Exact agent names table (subagent_type values)
- Task tool invocation format with examples
- Complete follow-up review workflow with Task calls
- DO NOT USE section (avoid general-purpose, Explore, Plan)
- Decision matrix for when to invoke each agent
- Explicit Task tool calls in Phase 2 workflow

This matches the main orchestrator prompt which has extensive
Task tool examples and works correctly.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(pr-review): propagate is_impact_finding in follow-up reviewer

Applied the same is_impact_finding propagation fix to the follow-up
reviewer that was already applied to the main orchestrator reviewer.

Fixes:
1. Add is_impact_finding field to ParallelFollowupFinding Pydantic model
2. Propagate is_impact_finding when creating PRReviewFinding for new findings
3. Copy is_impact_finding from original finding for unresolved findings

Without this fix, impact findings (about callers/affected files outside
the PR's changed files) would be incorrectly filtered as "not in scope"
during follow-up reviews.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* feat(auth): enhance keychain service integration with config directory support

Added functionality to support profile-specific credentials by introducing a hash-based service name for macOS Keychain and updating Windows credential retrieval to utilize a provided config directory. This ensures that tokens are fetched from the correct profile-specific storage locations, improving credential management across different environments.

Changes include:
- New functions for calculating config directory hashes and generating keychain service names.
- Updated `get_token_from_keychain` and related functions to accept an optional config directory argument.
- Enhanced logging for better debugging when no token is found.

This aligns the backend credential handling with the frontend's expectations for profile-specific storage.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-29 14:34:08 +01:00

1142 lines
40 KiB
Python

"""
Authentication helpers for Auto Claude.
Provides centralized authentication token resolution with fallback support
for multiple environment variables, and SDK environment variable passthrough
for custom API endpoints.
"""
import hashlib
import json
import logging
import os
import shutil
import subprocess
from typing import TYPE_CHECKING
from core.platform import (
is_linux,
is_macos,
is_windows,
)
logger = logging.getLogger(__name__)
# Optional import for Linux secret-service support
# secretstorage provides access to the Freedesktop.org Secret Service API via DBus
if TYPE_CHECKING:
import secretstorage
else:
try:
import secretstorage # type: ignore[import-untyped]
except ImportError:
secretstorage = None # type: ignore[assignment]
# Priority order for auth token resolution
# NOTE: We intentionally do NOT fall back to ANTHROPIC_API_KEY.
# Auto Claude is designed to use Claude Code OAuth tokens only.
# This prevents silent billing to user's API credits when OAuth fails.
AUTH_TOKEN_ENV_VARS = [
"CLAUDE_CODE_OAUTH_TOKEN", # OAuth token from Claude Code CLI
"ANTHROPIC_AUTH_TOKEN", # CCR/proxy token (for enterprise setups)
]
# Environment variables to pass through to SDK subprocess
# NOTE: ANTHROPIC_API_KEY is intentionally excluded to prevent silent API billing
SDK_ENV_VARS = [
# API endpoint configuration
"ANTHROPIC_BASE_URL",
"ANTHROPIC_AUTH_TOKEN",
# Model overrides (from API Profile custom model mappings)
"ANTHROPIC_MODEL",
"ANTHROPIC_DEFAULT_HAIKU_MODEL",
"ANTHROPIC_DEFAULT_SONNET_MODEL",
"ANTHROPIC_DEFAULT_OPUS_MODEL",
# SDK behavior configuration
"NO_PROXY",
"DISABLE_TELEMETRY",
"DISABLE_COST_WARNINGS",
"API_TIMEOUT_MS",
# Windows-specific: Git Bash path for Claude Code CLI
"CLAUDE_CODE_GIT_BASH_PATH",
# Claude CLI path override (allows frontend to pass detected CLI path to SDK)
"CLAUDE_CLI_PATH",
# Profile's custom config directory (for multi-profile token storage)
"CLAUDE_CONFIG_DIR",
]
def _calculate_config_dir_hash(config_dir: str) -> str:
"""
Calculate hash of config directory path for Keychain service name.
This MUST match the frontend's calculateConfigDirHash() in credential-utils.ts.
The frontend uses SHA256 hash of the config dir path, taking first 8 hex chars.
Args:
config_dir: Path to the config directory (should be absolute/expanded)
Returns:
8-character hex hash string (e.g., "d74c9506")
"""
return hashlib.sha256(config_dir.encode()).hexdigest()[:8]
def _get_keychain_service_name(config_dir: str | None = None) -> str:
"""
Get the Keychain service name for credential storage.
This MUST match the frontend's getKeychainServiceName() in credential-utils.ts.
All profiles use hash-based keychain entries for isolation:
- Profile with configDir: "Claude Code-credentials-{hash}"
- No configDir (legacy/default): "Claude Code-credentials"
Args:
config_dir: Optional CLAUDE_CONFIG_DIR path. If provided, uses hash-based name.
Returns:
Keychain service name (e.g., "Claude Code-credentials-d74c9506")
"""
if not config_dir:
return "Claude Code-credentials"
# Expand ~ to home directory (matching frontend normalization)
expanded_dir = os.path.expanduser(config_dir)
# Calculate hash and return hash-based service name
hash_suffix = _calculate_config_dir_hash(expanded_dir)
return f"Claude Code-credentials-{hash_suffix}"
def is_encrypted_token(token: str | None) -> bool:
"""
Check if a token is encrypted (has "enc:" prefix).
Args:
token: Token string to check (can be None)
Returns:
True if token starts with "enc:", False otherwise
"""
return bool(token and token.startswith("enc:"))
def validate_token_not_encrypted(token: str) -> None:
"""
Validate that a token is not in encrypted format.
This function should be called before passing a token to the Claude Agent SDK
to ensure proper error messages when decryption has failed.
Args:
token: Token string to validate
Raises:
ValueError: If token is in encrypted format (enc:...)
"""
if is_encrypted_token(token):
raise ValueError(
"Authentication token is in encrypted format and cannot be used.\n\n"
"The token decryption process failed or was not attempted.\n\n"
"To fix this issue:\n"
" 1. Re-authenticate with Claude Code CLI: claude setup-token\n"
" 2. Or set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in your .env file\n\n"
"Note: Encrypted tokens require the Claude Code CLI to be installed\n"
"and properly configured with system keychain access."
)
def decrypt_token(encrypted_token: str) -> str:
"""
Decrypt Claude Code encrypted token.
NOTE: This implementation currently relies on the system keychain (macOS Keychain,
Linux Secret Service, Windows Credential Manager) to provide already-decrypted tokens.
Encrypted tokens in the CLAUDE_CODE_OAUTH_TOKEN environment variable are NOT supported
and will fail with NotImplementedError.
For encrypted token support, users should:
1. Run: claude setup-token (stores decrypted token in system keychain)
2. Or set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in .env file
Claude Code CLI stores OAuth tokens in encrypted format with "enc:" prefix.
This function attempts to decrypt the token using platform-specific methods.
Cross-platform token decryption approaches:
- macOS: Token stored in Keychain with encryption key
- Linux: Token stored in Secret Service API with encryption key
- Windows: Token stored in Credential Manager or .credentials.json
Args:
encrypted_token: Token with 'enc:' prefix from Claude Code CLI
Returns:
Decrypted token in format 'sk-ant-oat01-...'
Raises:
ValueError: If token format is invalid or decryption fails
"""
# Validate encrypted token format
if not isinstance(encrypted_token, str):
raise ValueError(
f"Invalid token type. Expected string, got: {type(encrypted_token).__name__}"
)
if not encrypted_token.startswith("enc:"):
raise ValueError(
"Invalid encrypted token format. Token must start with 'enc:' prefix."
)
# Remove 'enc:' prefix to get encrypted data
encrypted_data = encrypted_token[4:]
if not encrypted_data:
raise ValueError("Empty encrypted token data after 'enc:' prefix")
# Basic validation of encrypted data format
# Encrypted data should be a reasonable length (at least 10 chars)
if len(encrypted_data) < 10:
raise ValueError(
"Encrypted token data is too short. The token may be corrupted."
)
# Check for obviously invalid characters that suggest corruption
# Accepts both standard base64 (+/) and URL-safe base64 (-_) to be permissive
if not all(c.isalnum() or c in "+-_/=" for c in encrypted_data):
raise ValueError(
"Encrypted token contains invalid characters. "
"Expected base64-encoded data. The token may be corrupted."
)
# Attempt platform-specific decryption
try:
if is_macos():
return _decrypt_token_macos(encrypted_data)
elif is_linux():
return _decrypt_token_linux(encrypted_data)
elif is_windows():
return _decrypt_token_windows(encrypted_data)
else:
raise ValueError("Unsupported platform for token decryption")
except NotImplementedError as e:
# Decryption not implemented - log warning and provide guidance
logger.warning(
"Token decryption failed: %s. Users must use plaintext tokens.", str(e)
)
raise ValueError(
f"Encrypted token decryption is not yet implemented: {str(e)}\n\n"
"To fix this issue:\n"
" 1. Set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token (without 'enc:' prefix)\n"
" 2. Or re-authenticate with: claude setup-token"
)
except ValueError:
# Re-raise ValueError as-is (already has good error message)
raise
except FileNotFoundError as e:
# File-related errors (missing credentials file, missing binary)
raise ValueError(
f"Failed to decrypt token - required file not found: {str(e)}\n\n"
"To fix this issue:\n"
" 1. Re-authenticate with Claude Code CLI: claude setup-token\n"
" 2. Or set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in your .env file"
)
except PermissionError as e:
# Permission errors (can't access keychain, credential manager, etc.)
raise ValueError(
f"Failed to decrypt token - permission denied: {str(e)}\n\n"
"To fix this issue:\n"
" 1. Grant keychain/credential manager access to this application\n"
" 2. Or set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in your .env file"
)
except subprocess.TimeoutExpired:
# Timeout during decryption process
raise ValueError(
"Failed to decrypt token - operation timed out.\n\n"
"This may indicate a problem with system keychain access.\n\n"
"To fix this issue:\n"
" 1. Re-authenticate with Claude Code CLI: claude setup-token\n"
" 2. Or set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in your .env file"
)
except Exception as e:
# Catch-all for other errors - provide helpful error message
error_type = type(e).__name__
raise ValueError(
f"Failed to decrypt token ({error_type}): {str(e)}\n\n"
"To fix this issue:\n"
" 1. Re-authenticate with Claude Code CLI: claude setup-token\n"
" 2. Or set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in your .env file\n\n"
"Note: Encrypted tokens (enc:...) require the Claude Code CLI to be installed\n"
"and properly configured with system keychain access."
)
def _decrypt_token_macos(encrypted_data: str) -> str:
"""
Decrypt token on macOS using Keychain.
Args:
encrypted_data: Encrypted token data (without 'enc:' prefix)
Returns:
Decrypted token
Raises:
ValueError: If decryption fails or Claude CLI not available
"""
# Verify Claude CLI is installed (required for future decryption implementation)
if not shutil.which("claude"):
raise ValueError(
"Claude Code CLI not found. Please install it from https://code.claude.com"
)
# The Claude Code CLI handles token decryption internally when it runs
# We can trigger this by running a simple command that requires authentication
# and capturing the decrypted token from the environment it sets up
#
# However, there's no direct CLI command to decrypt tokens.
# The SDK should handle this automatically when it receives encrypted tokens.
raise NotImplementedError(
"Encrypted tokens in environment variables are not supported. "
"Please use one of these options:\n"
" 1. Run 'claude setup-token' to store token in system keychain\n"
" 2. Set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in .env file\n\n"
"Note: This requires Claude Agent SDK >= 0.1.19"
)
def _decrypt_token_linux(encrypted_data: str) -> str:
"""
Decrypt token on Linux using Secret Service API.
Args:
encrypted_data: Encrypted token data (without 'enc:' prefix)
Returns:
Decrypted token
Raises:
ValueError: If decryption fails or dependencies not available
"""
# Linux token decryption requires secretstorage library
if secretstorage is None:
raise ValueError(
"secretstorage library not found. Install it with: pip install secretstorage"
)
# Similar to macOS, the actual decryption mechanism isn't publicly documented
# The Claude Agent SDK should handle this automatically
raise NotImplementedError(
"Encrypted tokens in environment variables are not supported. "
"Please use one of these options:\n"
" 1. Run 'claude setup-token' to store token in system keychain\n"
" 2. Set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in .env file\n\n"
"Note: This requires Claude Agent SDK >= 0.1.19"
)
def _decrypt_token_windows(encrypted_data: str) -> str:
"""
Decrypt token on Windows using Credential Manager.
Args:
encrypted_data: Encrypted token data (without 'enc:' prefix)
Returns:
Decrypted token
Raises:
ValueError: If decryption fails
"""
# Windows token decryption from Credential Manager or .credentials.json
# The Claude Agent SDK should handle this automatically
raise NotImplementedError(
"Encrypted tokens in environment variables are not supported. "
"Please use one of these options:\n"
" 1. Run 'claude setup-token' to store token in system keychain\n"
" 2. Set CLAUDE_CODE_OAUTH_TOKEN to a plaintext token in .env file\n\n"
"Note: This requires Claude Agent SDK >= 0.1.19"
)
def _try_decrypt_token(token: str | None) -> str | None:
"""
Attempt to decrypt an encrypted token, returning original if decryption fails.
This helper centralizes the decrypt-or-return-as-is logic used when resolving
tokens from various sources (env vars, config dir, keychain).
Args:
token: Token string (may be encrypted with "enc:" prefix, plaintext, or None)
Returns:
- Decrypted token if successfully decrypted
- Original token if decryption fails (allows client validation to report error)
- Original token if not encrypted
- None if token is None
"""
if not token:
return None
if is_encrypted_token(token):
try:
return decrypt_token(token)
except ValueError:
# Decryption failed - return encrypted token so client validation
# (validate_token_not_encrypted) can provide specific error message.
return token
return token
def get_token_from_keychain(config_dir: str | None = None) -> str | None:
"""
Get authentication token from system credential store.
Reads Claude Code credentials from:
- macOS: Keychain (uses hash-based service name if config_dir provided)
- Windows: Credential Manager
- Linux: Secret Service API (via dbus/secretstorage)
Args:
config_dir: Optional CLAUDE_CONFIG_DIR path for profile-specific credentials.
When provided, reads from hash-based keychain entry matching
the frontend's storage location.
Returns:
Token string if found, None otherwise
"""
if is_macos():
return _get_token_from_macos_keychain(config_dir)
elif is_windows():
return _get_token_from_windows_credential_files(config_dir)
else:
# Linux: use secret-service API via DBus
return _get_token_from_linux_secret_service(config_dir)
def _get_token_from_macos_keychain(config_dir: str | None = None) -> str | None:
"""Get token from macOS Keychain.
Args:
config_dir: Optional CLAUDE_CONFIG_DIR path. When provided, uses hash-based
service name (e.g., "Claude Code-credentials-d74c9506") matching
the frontend's credential storage location.
"""
# Get the correct service name (hash-based if config_dir provided)
service_name = _get_keychain_service_name(config_dir)
try:
result = subprocess.run(
[
"/usr/bin/security",
"find-generic-password",
"-s",
service_name,
"-w",
],
capture_output=True,
text=True,
timeout=5,
)
if result.returncode != 0:
# If hash-based lookup fails and we have a config_dir, DON'T fall back
# to default service name - that would return the wrong profile's token.
# The config_dir was provided explicitly, so we should only use that.
if config_dir:
logger.debug(
f"No keychain entry found for service '{service_name}' "
f"(config_dir: {config_dir})"
)
return None
credentials_json = result.stdout.strip()
if not credentials_json:
return None
data = json.loads(credentials_json)
token = data.get("claudeAiOauth", {}).get("accessToken")
if not token:
return None
# Validate token format (Claude OAuth tokens start with sk-ant-oat01-)
# Also accept encrypted tokens (enc:) which will be decrypted later
if not (token.startswith("sk-ant-oat01-") or token.startswith("enc:")):
return None
logger.debug(f"Found token in keychain service '{service_name}'")
return token
except (subprocess.TimeoutExpired, json.JSONDecodeError, KeyError, Exception):
return None
def _get_token_from_windows_credential_files(
config_dir: str | None = None,
) -> str | None:
"""Get token from Windows credential files.
Claude Code on Windows stores credentials in ~/.claude/.credentials.json
For custom profiles, uses the config_dir's .credentials.json file.
Args:
config_dir: Optional CLAUDE_CONFIG_DIR path for profile-specific credentials.
"""
try:
# If config_dir is provided, read from that directory first
if config_dir:
expanded_dir = os.path.expanduser(config_dir)
profile_cred_paths = [
os.path.join(expanded_dir, ".credentials.json"),
os.path.join(expanded_dir, "credentials.json"),
]
for cred_path in profile_cred_paths:
if os.path.exists(cred_path):
with open(cred_path, encoding="utf-8") as f:
data = json.load(f)
token = data.get("claudeAiOauth", {}).get("accessToken")
if token and (
token.startswith("sk-ant-oat01-")
or token.startswith("enc:")
):
logger.debug(f"Found token in {cred_path}")
return token
# If config_dir provided but no token found, don't fall back to default
return None
# Default Claude Code credential paths (no profile specified)
cred_paths = [
os.path.expandvars(r"%USERPROFILE%\.claude\.credentials.json"),
os.path.expandvars(r"%USERPROFILE%\.claude\credentials.json"),
os.path.expandvars(r"%LOCALAPPDATA%\Claude\credentials.json"),
os.path.expandvars(r"%APPDATA%\Claude\credentials.json"),
]
for cred_path in cred_paths:
if os.path.exists(cred_path):
with open(cred_path, encoding="utf-8") as f:
data = json.load(f)
token = data.get("claudeAiOauth", {}).get("accessToken")
if token and (
token.startswith("sk-ant-oat01-") or token.startswith("enc:")
):
return token
return None
except (json.JSONDecodeError, KeyError, FileNotFoundError, Exception):
return None
def _get_token_from_linux_secret_service(config_dir: str | None = None) -> str | None:
"""Get token from Linux Secret Service API via DBus.
Claude Code on Linux stores credentials in the Secret Service API
using the 'org.freedesktop.secrets' collection. This implementation
uses the secretstorage library which communicates via DBus.
The credential is stored with:
- Label: "Claude Code-credentials" or "Claude Code-credentials-{hash}" for profiles
- Attributes: {application: "claude-code"}
Args:
config_dir: Optional CLAUDE_CONFIG_DIR path for profile-specific credentials.
Returns:
Token string if found, None otherwise
"""
if secretstorage is None:
# secretstorage not installed, fall back to env var
return None
# Get the correct service name (hash-based if config_dir provided)
target_label = _get_keychain_service_name(config_dir)
try:
# Get the default collection (typically "login" keyring)
# secretstorage handles DBus communication internally
try:
collection = secretstorage.get_default_collection(None)
except (
AttributeError,
secretstorage.exceptions.SecretServiceNotAvailableException,
):
# DBus not available or secret-service not running
return None
if collection.is_locked():
# Try to unlock the collection (may prompt user for password)
try:
collection.unlock()
except secretstorage.exceptions.SecretStorageException:
# User cancelled or unlock failed
return None
# Search for items with our application attribute
items = collection.search_items({"application": "claude-code"})
for item in items:
# Check if this is the correct Claude Code credentials item
label = item.get_label()
# Use exact match for target label (profile-specific or default)
if label == target_label:
# Get the secret (stored as JSON string)
secret = item.get_secret()
if not secret:
continue
try:
# Explicitly decode bytes to string if needed
if isinstance(secret, bytes):
secret = secret.decode("utf-8")
data = json.loads(secret)
token = data.get("claudeAiOauth", {}).get("accessToken")
if token and (
token.startswith("sk-ant-oat01-") or token.startswith("enc:")
):
logger.debug(
f"Found token in secret service with label '{target_label}'"
)
return token
except json.JSONDecodeError:
continue
# If config_dir was provided but no token found, don't fall back
if config_dir:
logger.debug(
f"No secret service entry found with label '{target_label}' "
f"(config_dir: {config_dir})"
)
return None
except (
secretstorage.exceptions.SecretStorageException,
json.JSONDecodeError,
KeyError,
AttributeError,
TypeError,
):
# Any error with secret-service, fall back to env var
return None
def _get_token_from_config_dir(config_dir: str) -> str | None:
"""
Read token from a custom config directory's credentials file.
Claude Code stores credentials in .credentials.json within the config directory.
This function reads from a profile's custom configDir instead of the default location.
Args:
config_dir: Path to the config directory (e.g., ~/.auto-claude/profiles/work)
Returns:
Token string if found, None otherwise
"""
# Expand ~ if present
expanded_dir = os.path.expanduser(config_dir)
# Claude stores credentials in these files within the config dir
cred_files = [
os.path.join(expanded_dir, ".credentials.json"),
os.path.join(expanded_dir, "credentials.json"),
]
for cred_path in cred_files:
if os.path.exists(cred_path):
try:
with open(cred_path, encoding="utf-8") as f:
data = json.load(f)
# Try both credential structures
oauth_data = data.get("claudeAiOauth") or data.get("oauthAccount") or {}
token = oauth_data.get("accessToken")
# Accept both plaintext tokens (sk-ant-oat01-) and encrypted tokens (enc:)
if token and (
token.startswith("sk-ant-oat01-") or token.startswith("enc:")
):
logger.debug(f"Found token in {cred_path}")
return token
except (json.JSONDecodeError, KeyError, Exception) as e:
logger.debug(f"Failed to read {cred_path}: {e}")
continue
return None
def get_auth_token(config_dir: str | None = None) -> str | None:
"""
Get authentication token from environment variables or credential store.
Args:
config_dir: Optional custom config directory (profile's configDir).
If provided, reads credentials from this directory.
If None, checks CLAUDE_CONFIG_DIR env var, then uses default locations.
Checks multiple sources in priority order:
1. CLAUDE_CODE_OAUTH_TOKEN (env var)
2. ANTHROPIC_AUTH_TOKEN (CCR/proxy env var for enterprise setups)
3. Custom config directory (config_dir param or CLAUDE_CONFIG_DIR env var)
4. System credential store (macOS Keychain, Windows Credential Manager, Linux Secret Service)
NOTE: ANTHROPIC_API_KEY is intentionally NOT supported to prevent
silent billing to user's API credits when OAuth is misconfigured.
If the token has an "enc:" prefix (encrypted format), it will be automatically
decrypted before being returned.
Returns:
Token string if found, None otherwise
"""
# First check environment variables (highest priority)
for var in AUTH_TOKEN_ENV_VARS:
token = os.environ.get(var)
if token:
return _try_decrypt_token(token)
# Check CLAUDE_CONFIG_DIR environment variable (profile's custom config directory)
env_config_dir = os.environ.get("CLAUDE_CONFIG_DIR")
effective_config_dir = config_dir or env_config_dir
# Debug: Log which config_dir is being used for credential resolution
debug = os.environ.get("DEBUG", "").lower() in ("true", "1")
if debug and effective_config_dir:
service_name = _get_keychain_service_name(effective_config_dir)
logger.info(
f"[Auth] Resolving credentials for profile config_dir: {effective_config_dir} "
f"(Keychain service: {service_name})"
)
# If a custom config directory is specified, read from there first
if effective_config_dir:
# Try reading from .credentials.json file in the config directory
token = _get_token_from_config_dir(effective_config_dir)
if token:
return _try_decrypt_token(token)
# Also try the system credential store with hash-based service name
# This is needed because macOS stores credentials in Keychain, not files
token = get_token_from_keychain(effective_config_dir)
if token:
return _try_decrypt_token(token)
# If config_dir was explicitly provided, DON'T fall back to default keychain
# - that would return the wrong profile's token
logger.debug(
f"No credentials found for config_dir '{effective_config_dir}' "
"in file or keychain"
)
return None
# No config_dir specified - use default system credential store
return _try_decrypt_token(get_token_from_keychain())
def get_auth_token_source(config_dir: str | None = None) -> str | None:
"""
Get the name of the source that provided the auth token.
Args:
config_dir: Optional custom config directory (profile's configDir).
If provided, checks this directory for credentials.
If None, checks CLAUDE_CONFIG_DIR env var.
"""
# Check environment variables first
for var in AUTH_TOKEN_ENV_VARS:
if os.environ.get(var):
return var
# Check if token came from custom config directory (profile's configDir)
env_config_dir = os.environ.get("CLAUDE_CONFIG_DIR")
effective_config_dir = config_dir or env_config_dir
if effective_config_dir:
# Check file-based storage
if _get_token_from_config_dir(effective_config_dir):
return "CLAUDE_CONFIG_DIR"
# Check hash-based keychain entry for this profile
if get_token_from_keychain(effective_config_dir):
if is_macos():
return "macOS Keychain (profile)"
elif is_windows():
return "Windows Credential Files (profile)"
else:
return "Linux Secret Service (profile)"
# Check if token came from default system credential store
if get_token_from_keychain():
if is_macos():
return "macOS Keychain"
elif is_windows():
return "Windows Credential Files"
else:
return "Linux Secret Service"
return None
def require_auth_token(config_dir: str | None = None) -> str:
"""
Get authentication token or raise ValueError.
Args:
config_dir: Optional custom config directory (profile's configDir).
If provided, reads credentials from this directory.
If None, checks CLAUDE_CONFIG_DIR env var, then uses default locations.
Raises:
ValueError: If no auth token is found in any supported source
"""
token = get_auth_token(config_dir)
if not token:
error_msg = (
"No OAuth token found.\n\n"
"Auto Claude requires Claude Code OAuth authentication.\n"
"Direct API keys (ANTHROPIC_API_KEY) are not supported.\n\n"
)
# Provide platform-specific guidance
if is_macos():
error_msg += (
"To authenticate:\n"
" 1. Run: claude\n"
" 2. Type: /login\n"
" 3. Press Enter to open browser\n"
" 4. Complete OAuth login in browser\n\n"
"The token will be saved to macOS Keychain automatically."
)
elif is_windows():
error_msg += (
"To authenticate:\n"
" 1. Run: claude\n"
" 2. Type: /login\n"
" 3. Press Enter to open browser\n"
" 4. Complete OAuth login in browser\n\n"
"The token will be saved to Windows Credential Manager."
)
else:
# Linux
error_msg += (
"To authenticate:\n"
" 1. Run: claude\n"
" 2. Type: /login\n"
" 3. Press Enter to open browser\n"
" 4. Complete OAuth login in browser\n\n"
"Or set CLAUDE_CODE_OAUTH_TOKEN in your .env file."
)
raise ValueError(error_msg)
return token
def _find_git_bash_path() -> str | None:
"""
Find git-bash (bash.exe) path on Windows.
Uses 'where git' to find git.exe, then derives bash.exe location from it.
Git for Windows installs bash.exe in the 'bin' directory alongside git.exe
or in the parent 'bin' directory when git.exe is in 'cmd'.
Returns:
Full path to bash.exe if found, None otherwise
"""
if not is_windows():
return None
# If already set in environment, use that
existing = os.environ.get("CLAUDE_CODE_GIT_BASH_PATH")
if existing and os.path.exists(existing):
return existing
git_path = None
# Method 1: Use 'where' command to find git.exe
try:
# Use where.exe explicitly for reliability
result = subprocess.run(
["where.exe", "git"],
capture_output=True,
text=True,
timeout=5,
shell=False,
)
if result.returncode == 0 and result.stdout.strip():
git_paths = result.stdout.strip().splitlines()
if git_paths:
git_path = git_paths[0].strip()
except (subprocess.TimeoutExpired, FileNotFoundError, subprocess.SubprocessError):
# Intentionally suppress errors - best-effort detection with fallback to common paths
pass
# Method 2: Check common installation paths if 'where' didn't work
if not git_path:
common_git_paths = [
os.path.expandvars(r"%PROGRAMFILES%\Git\cmd\git.exe"),
os.path.expandvars(r"%PROGRAMFILES%\Git\bin\git.exe"),
os.path.expandvars(r"%PROGRAMFILES(X86)%\Git\cmd\git.exe"),
os.path.expandvars(r"%LOCALAPPDATA%\Programs\Git\cmd\git.exe"),
]
for path in common_git_paths:
if os.path.exists(path):
git_path = path
break
if not git_path:
return None
# Derive bash.exe location from git.exe location
# Git for Windows structure:
# C:\...\Git\cmd\git.exe -> bash.exe is at C:\...\Git\bin\bash.exe
# C:\...\Git\bin\git.exe -> bash.exe is at C:\...\Git\bin\bash.exe
# C:\...\Git\mingw64\bin\git.exe -> bash.exe is at C:\...\Git\bin\bash.exe
git_dir = os.path.dirname(git_path)
git_parent = os.path.dirname(git_dir)
git_grandparent = os.path.dirname(git_parent)
# Check common bash.exe locations relative to git installation
possible_bash_paths = [
os.path.join(git_parent, "bin", "bash.exe"), # cmd -> bin
os.path.join(git_dir, "bash.exe"), # If git.exe is in bin
os.path.join(git_grandparent, "bin", "bash.exe"), # mingw64/bin -> bin
]
for bash_path in possible_bash_paths:
if os.path.exists(bash_path):
return bash_path
return None
def get_sdk_env_vars() -> dict[str, str]:
"""
Get environment variables to pass to SDK.
Collects relevant env vars (ANTHROPIC_BASE_URL, etc.) that should
be passed through to the claude-agent-sdk subprocess.
On Windows, auto-detects CLAUDE_CODE_GIT_BASH_PATH if not already set.
Returns:
Dict of env var name -> value for non-empty vars
"""
env = {}
for var in SDK_ENV_VARS:
value = os.environ.get(var)
if value:
env[var] = value
# On Windows, auto-detect git-bash path if not already set
# Claude Code CLI requires bash.exe to run on Windows
if is_windows() and "CLAUDE_CODE_GIT_BASH_PATH" not in env:
bash_path = _find_git_bash_path()
if bash_path:
env["CLAUDE_CODE_GIT_BASH_PATH"] = bash_path
# Explicitly unset PYTHONPATH in SDK subprocess environment to prevent
# pollution of agent subprocess environments. This fixes ACS-251 where
# external projects with different Python versions would fail due to
# inheriting Auto-Claude's PYTHONPATH (which points to Python 3.12 packages).
#
# The SDK merges os.environ with the env dict we provide, so setting
# PYTHONPATH to an empty string here overrides any inherited value.
# The empty string ensures Python doesn't add any extra paths to sys.path.
env["PYTHONPATH"] = ""
return env
def ensure_claude_code_oauth_token() -> None:
"""
Ensure CLAUDE_CODE_OAUTH_TOKEN is set (for SDK compatibility).
If not set but other auth tokens are available, copies the value
to CLAUDE_CODE_OAUTH_TOKEN so the underlying SDK can use it.
"""
if os.environ.get("CLAUDE_CODE_OAUTH_TOKEN"):
return
token = get_auth_token()
if token:
os.environ["CLAUDE_CODE_OAUTH_TOKEN"] = token
def trigger_login() -> bool:
"""
Trigger Claude Code OAuth login flow.
Opens the Claude Code CLI and sends /login command to initiate
browser-based OAuth authentication. The token is automatically
saved to the system credential store (macOS Keychain, Windows
Credential Manager).
Returns:
True if login was successful, False otherwise
"""
if is_macos():
return _trigger_login_macos()
elif is_windows():
return _trigger_login_windows()
else:
# Linux: fall back to manual instructions
print("\nTo authenticate, run 'claude' and type '/login'")
return False
def _trigger_login_macos() -> bool:
"""Trigger login on macOS using expect."""
import shutil
import tempfile
# Check if expect is available
if not shutil.which("expect"):
print("\nTo authenticate, run 'claude' and type '/login'")
return False
# Create expect script
expect_script = """#!/usr/bin/expect -f
set timeout 120
spawn claude
expect {
-re ".*" {
send "/login\\r"
expect {
"Press Enter" {
send "\\r"
}
-re ".*login.*" {
send "\\r"
}
timeout {
send "\\r"
}
}
}
}
# Keep running until user completes login or exits
interact
"""
# Use TemporaryDirectory context manager for automatic cleanup
# This prevents information leakage about authentication activity
# Directory created with mode 0o700 (owner read/write/execute only)
try:
with tempfile.TemporaryDirectory() as temp_dir:
# Ensure directory has owner-only permissions
os.chmod(temp_dir, 0o700)
# Write expect script to temp file in our private directory
script_path = os.path.join(temp_dir, "login.exp")
with open(script_path, "w", encoding="utf-8") as f:
f.write(expect_script)
# Set script permissions to owner-only (0o700)
os.chmod(script_path, 0o700)
print("\n" + "=" * 60)
print("CLAUDE CODE LOGIN")
print("=" * 60)
print("\nOpening Claude Code for authentication...")
print("A browser window will open for OAuth login.")
print("After completing login in the browser, press Ctrl+C to exit.\n")
# Run expect script
subprocess.run(
["expect", script_path],
timeout=300, # 5 minute timeout
)
# Verify token was saved
token = get_token_from_keychain()
if token:
print("\n✓ Login successful! Token saved to macOS Keychain.")
return True
else:
print(
"\n✗ Login may not have completed. Try running 'claude' and type '/login'"
)
return False
except subprocess.TimeoutExpired:
print("\nLogin timed out. Try running 'claude' manually and type '/login'")
return False
except KeyboardInterrupt:
# User pressed Ctrl+C - check if login completed
token = get_token_from_keychain()
if token:
print("\n✓ Login successful! Token saved to macOS Keychain.")
return True
return False
except Exception as e:
print(f"\nLogin failed: {e}")
print("Try running 'claude' manually and type '/login'")
return False
def _trigger_login_windows() -> bool:
"""Trigger login on Windows."""
# Windows doesn't have expect by default, so we use a simpler approach
# that just launches claude and tells the user what to type
print("\n" + "=" * 60)
print("CLAUDE CODE LOGIN")
print("=" * 60)
print("\nLaunching Claude Code...")
print("Please type '/login' and press Enter.")
print("A browser window will open for OAuth login.\n")
try:
# Launch claude interactively
subprocess.run(["claude"], timeout=300)
# Verify token was saved
token = _get_token_from_windows_credential_files()
if token:
print("\n✓ Login successful!")
return True
else:
print("\n✗ Login may not have completed.")
return False
except Exception as e:
print(f"\nLogin failed: {e}")
return False
def ensure_authenticated() -> str:
"""
Ensure the user is authenticated, prompting for login if needed.
Checks for existing token and triggers login flow if not found.
Returns:
The authentication token
Raises:
ValueError: If authentication fails after login attempt
"""
# First check if already authenticated
token = get_auth_token()
if token:
return token
# No token found - trigger login
print("\nNo OAuth token found. Starting login flow...")
if trigger_login():
# Re-check for token after login
token = get_auth_token()
if token:
return token
# Login failed or was cancelled
raise ValueError(
"Authentication required.\n\n"
"To authenticate:\n"
" 1. Run: claude\n"
" 2. Type: /login\n"
" 3. Press Enter to open browser\n"
" 4. Complete OAuth login in browser"
)