ccproxy.plugins.oauth_claude.provider

ccproxy.plugins.oauth_claude.provider

Claude OAuth provider for plugin registration.

ClaudeOAuthProvider

ClaudeOAuthProvider(
    config=None,
    storage=None,
    http_client=None,
    hook_manager=None,
    detection_service=None,
    settings=None,
)

Bases: ProfileLoggingMixin

Claude OAuth provider implementation for registry.

Parameters:

Name	Type	Description	Default
config	ClaudeOAuthConfig | None	OAuth configuration	None
storage	ClaudeOAuthStorage | None	Token storage	None
http_client	AsyncClient | None	Optional HTTP client (for request tracing support)	None
hook_manager	Any | None	Optional hook manager for emitting events	None
detection_service	CLIDetectionService | None	Optional CLI detection service for headers	None
settings	Settings | None	Optional settings for HTTP client configuration	None

Source code in ccproxy/plugins/oauth_claude/provider.py

def __init__(
    self,
    config: ClaudeOAuthConfig | None = None,
    storage: ClaudeOAuthStorage | None = None,
    http_client: httpx.AsyncClient | None = None,
    hook_manager: Any | None = None,
    detection_service: "CLIDetectionService | None" = None,
    settings: Settings | None = None,
):
    """Initialize Claude OAuth provider.

    Args:
        config: OAuth configuration
        storage: Token storage
        http_client: Optional HTTP client (for request tracing support)
        hook_manager: Optional hook manager for emitting events
        detection_service: Optional CLI detection service for headers
        settings: Optional settings for HTTP client configuration
    """
    self.config = config or ClaudeOAuthConfig()
    self.storage = storage or ClaudeOAuthStorage()
    self.hook_manager = hook_manager
    self.detection_service = detection_service
    self.http_client = http_client
    self.settings = settings
    self._cached_profile: ClaudeProfileInfo | None = (
        None  # Cache enhanced profile data for UI display
    )

    self.client = ClaudeOAuthClient(
        self.config,
        self.storage,
        http_client,
        hook_manager=hook_manager,
        detection_service=detection_service,
        settings=settings,
    )

provider_name property

provider_name

Internal provider name.

provider_display_name property

provider_display_name

Display name for UI.

supports_pkce property

supports_pkce

Whether this provider supports PKCE.

supports_refresh property

supports_refresh

Whether this provider supports token refresh.

requires_client_secret property

requires_client_secret

Whether this provider requires a client secret.

cli property

cli

Get CLI authentication configuration for this provider.

get_authorization_url async

get_authorization_url(
    state, code_verifier=None, redirect_uri=None
)

Get the authorization URL for OAuth flow.

Parameters:

Name	Type	Description	Default
state	str	OAuth state parameter for CSRF protection	required
code_verifier	str | None	PKCE code verifier (if PKCE is supported)	None

Returns:

Type	Description
str	Authorization URL to redirect user to

Source code in ccproxy/plugins/oauth_claude/provider.py

async def get_authorization_url(
    self,
    state: str,
    code_verifier: str | None = None,
    redirect_uri: str | None = None,
) -> str:
    """Get the authorization URL for OAuth flow.

    Args:
        state: OAuth state parameter for CSRF protection
        code_verifier: PKCE code verifier (if PKCE is supported)

    Returns:
        Authorization URL to redirect user to
    """
    # Use provided redirect URI or fall back to config default
    if redirect_uri is None:
        redirect_uri = self.config.get_redirect_uri()

    params = {
        "code": "true",  # Required by Claude OAuth
        "client_id": self.config.client_id,
        "redirect_uri": redirect_uri,
        "response_type": "code",
        "scope": " ".join(self.config.scopes),
        "state": state,
    }

    # Add PKCE challenge if supported and verifier provided
    if self.config.use_pkce and code_verifier:
        code_challenge = (
            urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest())
            .decode()
            .rstrip("=")
        )
        params["code_challenge"] = code_challenge
        params["code_challenge_method"] = "S256"

    auth_url = f"{self.config.authorize_url}?{urlencode(params)}"

    logger.info(
        "claude_oauth_auth_url_generated",
        state=state,
        has_pkce=bool(code_verifier and self.config.use_pkce),
        category="auth",
    )

    return auth_url

handle_callback async

handle_callback(
    code, state, code_verifier=None, redirect_uri=None
)

Handle OAuth callback and exchange code for tokens.

Parameters:

Name	Type	Description	Default
code	str	Authorization code from OAuth callback	required
state	str	State parameter for validation	required
code_verifier	str | None	PKCE code verifier (if PKCE is used)	None
redirect_uri	str | None	Redirect URI used in authorization (optional)	None

Returns:

Type	Description
Any	Claude credentials object

Source code in ccproxy/plugins/oauth_claude/provider.py

async def handle_callback(
    self,
    code: str,
    state: str,
    code_verifier: str | None = None,
    redirect_uri: str | None = None,
) -> Any:
    """Handle OAuth callback and exchange code for tokens.

    Args:
        code: Authorization code from OAuth callback
        state: State parameter for validation
        code_verifier: PKCE code verifier (if PKCE is used)
        redirect_uri: Redirect URI used in authorization (optional)

    Returns:
        Claude credentials object
    """
    # Use the client's handle_callback method which includes code exchange
    # If a specific redirect_uri was provided, create a temporary client with that URI
    if redirect_uri and redirect_uri != self.client.redirect_uri:
        # Create temporary config with the specific redirect URI
        temp_config = ClaudeOAuthConfig(
            client_id=self.config.client_id,
            redirect_uri=redirect_uri,
            scopes=self.config.scopes,
            base_url=self.config.base_url,
            authorize_url=self.config.authorize_url,
            token_url=self.config.token_url,
            use_pkce=self.config.use_pkce,
        )

        # Create temporary client with the correct redirect URI
        temp_client = ClaudeOAuthClient(
            temp_config,
            self.storage,
            self.http_client,
            hook_manager=self.hook_manager,
            detection_service=self.detection_service,
            settings=self.settings,
        )

        credentials = await temp_client.handle_callback(
            code, state, code_verifier or ""
        )
    else:
        # Use the regular client
        credentials = await self.client.handle_callback(
            code, state, code_verifier or ""
        )

    # The client already saves to storage if available, but we can save again
    # to our specific storage if needed
    if self.storage:
        await self.storage.save(credentials)

    logger.info(
        "claude_oauth_callback_handled",
        state=state,
        has_credentials=bool(credentials),
        category="auth",
    )

    return credentials

refresh_access_token async

refresh_access_token(refresh_token)

Refresh access token using refresh token.

Parameters:

Name	Type	Description	Default
refresh_token	str	Refresh token from previous auth	required

Returns:

Type	Description
Any	New token response

Source code in ccproxy/plugins/oauth_claude/provider.py

async def refresh_access_token(self, refresh_token: str) -> Any:
    """Refresh access token using refresh token.

    Args:
        refresh_token: Refresh token from previous auth

    Returns:
        New token response
    """
    credentials = await self.client.refresh_token(refresh_token)

    # Store updated credentials
    if self.storage:
        await self.storage.save(credentials)

    logger.info("claude_oauth_token_refreshed", category="auth")

    return credentials

revoke_token async

revoke_token(token)

Revoke an access or refresh token.

Parameters:

Name	Type	Description	Default
token	str	Token to revoke	required

Source code in ccproxy/plugins/oauth_claude/provider.py

async def revoke_token(self, token: str) -> None:
    """Revoke an access or refresh token.

    Args:
        token: Token to revoke
    """
    # Claude doesn't have a revoke endpoint, so we just delete stored credentials
    if self.storage:
        await self.storage.delete()

    logger.info("claude_oauth_token_revoked_locally", category="auth")

get_provider_info

get_provider_info()

Get provider information for discovery.

Returns:

Type	Description
OAuthProviderInfo	Provider information

Source code in ccproxy/plugins/oauth_claude/provider.py

def get_provider_info(self) -> OAuthProviderInfo:
    """Get provider information for discovery.

    Returns:
        Provider information
    """
    return OAuthProviderInfo(
        name=self.provider_name,
        display_name=self.provider_display_name,
        description="OAuth authentication for Claude AI",
        supports_pkce=self.supports_pkce,
        scopes=self.config.scopes,
        is_available=True,
        plugin_name="oauth_claude",
    )

validate_token async

validate_token(access_token)

Validate an access token.

Parameters:

Name	Type	Description	Default
access_token	str	Token to validate	required

Returns:

Type	Description
bool	True if token is valid

Source code in ccproxy/plugins/oauth_claude/provider.py

async def validate_token(self, access_token: str) -> bool:
    """Validate an access token.

    Args:
        access_token: Token to validate

    Returns:
        True if token is valid
    """
    # Claude doesn't have a validation endpoint, so we check if stored token matches
    if self.storage:
        credentials = await self.storage.load()
        if credentials and credentials.claude_ai_oauth:
            stored_token = (
                credentials.claude_ai_oauth.access_token.get_secret_value()
            )
            return stored_token == access_token
    return False

get_user_info async

get_user_info(access_token)

Get user information using access token.

Parameters:

Name	Type	Description	Default
access_token	str	Valid access token	required

Returns:

Type	Description
dict[str, Any] | None	User information or None

Source code in ccproxy/plugins/oauth_claude/provider.py

async def get_user_info(self, access_token: str) -> dict[str, Any] | None:
    """Get user information using access token.

    Args:
        access_token: Valid access token

    Returns:
        User information or None
    """
    # Load stored credentials which contain user info
    if self.storage:
        credentials = await self.storage.load()
        if credentials and credentials.claude_ai_oauth:
            return {
                "subscription_type": credentials.claude_ai_oauth.subscription_type,
                "scopes": credentials.claude_ai_oauth.scopes,
            }
    return None

get_storage

get_storage()

Get storage implementation for this provider.

Returns:

Type	Description
Any	Storage implementation

Source code in ccproxy/plugins/oauth_claude/provider.py

def get_storage(self) -> Any:
    """Get storage implementation for this provider.

    Returns:
        Storage implementation
    """
    return self.storage

get_config

get_config()

Get configuration for this provider.

Returns:

Type	Description
Any	Configuration implementation

Source code in ccproxy/plugins/oauth_claude/provider.py

def get_config(self) -> Any:
    """Get configuration for this provider.

    Returns:
        Configuration implementation
    """
    return self.config

save_credentials async

save_credentials(credentials, custom_path=None)

Save credentials using provider's storage mechanism.

Parameters:

Name	Type	Description	Default
credentials	Any	Claude credentials object	required
custom_path	Any | None	Optional custom storage path (Path object)	None

Returns:

Type	Description
bool	True if saved successfully, False otherwise

Source code in ccproxy/plugins/oauth_claude/provider.py

async def save_credentials(
    self, credentials: Any, custom_path: Any | None = None
) -> bool:
    """Save credentials using provider's storage mechanism.

    Args:
        credentials: Claude credentials object
        custom_path: Optional custom storage path (Path object)

    Returns:
        True if saved successfully, False otherwise
    """
    try:
        if custom_path:
            # Use custom path for storage
            storage = GenericJsonStorage(Path(custom_path), ClaudeCredentials)
            manager = await self.create_token_manager(storage=storage)
        else:
            # Use default storage
            manager = await self.create_token_manager()

        return await manager.save_credentials(credentials)
    except Exception as e:
        logger.error(
            "Failed to save Claude credentials",
            error=str(e),
            exc_info=e,
            has_custom_path=bool(custom_path),
        )
        return False

load_credentials async

load_credentials(custom_path=None)

Load credentials from provider's storage.

Parameters:

Name	Type	Description	Default
custom_path	Any | None	Optional custom storage path (Path object)	None

Returns:

Type	Description
Any | None	Credentials if found, None otherwise

Source code in ccproxy/plugins/oauth_claude/provider.py

async def load_credentials(self, custom_path: Any | None = None) -> Any | None:
    """Load credentials from provider's storage.

    Args:
        custom_path: Optional custom storage path (Path object)

    Returns:
        Credentials if found, None otherwise
    """
    try:
        if custom_path:
            # Load from custom path
            storage = GenericJsonStorage(Path(custom_path), ClaudeCredentials)
            manager = await self.create_token_manager(storage=storage)
        else:
            # Load from default storage
            manager = await self.create_token_manager()

        credentials = await manager.load_credentials()

        # Use standardized profile logging with rich Claude profile data
        if credentials:
            profile = await manager.get_profile()
            if profile:
                # Cache profile for UI display
                self._cached_profile = profile
                # Create enhanced standardized profile with rich Claude data
                standard_profile = self._create_enhanced_profile(
                    credentials, profile
                )
                self._log_profile_dump("claude", standard_profile)

        return credentials
    except Exception as e:
        logger.error(
            "Failed to load Claude credentials",
            error=str(e),
            exc_info=e,
            has_custom_path=bool(custom_path),
        )
        return None

create_token_manager async

create_token_manager(storage=None)

Create token manager with proper dependency injection.

Provided to allow core/CLI code to obtain a manager without importing plugin classes directly.

Source code in ccproxy/plugins/oauth_claude/provider.py

async def create_token_manager(
    self, storage: Any | None = None
) -> "ClaudeApiTokenManager":
    """Create token manager with proper dependency injection.

    Provided to allow core/CLI code to obtain a manager without
    importing plugin classes directly.
    """
    from .manager import ClaudeApiTokenManager

    return await ClaudeApiTokenManager.create(
        storage=storage,
        http_client=self.http_client,
        oauth_provider=self,  # Inject self as protocol
    )

exchange_manual_code async

exchange_manual_code(code)

Exchange manual authorization code for tokens.

Parameters:

Name	Type	Description	Default
code	str	Authorization code from manual entry	required

Returns:

Type	Description
Any	Claude credentials object

Source code in ccproxy/plugins/oauth_claude/provider.py

async def exchange_manual_code(self, code: str) -> Any:
    """Exchange manual authorization code for tokens.

    Args:
        code: Authorization code from manual entry

    Returns:
        Claude credentials object
    """
    # For manual code flow, use OOB redirect URI and no state validation
    credentials: ClaudeCredentials = await self.client.handle_callback(
        code, "manual", ""
    )

    if self.storage:
        await self.storage.save(credentials)

    logger.info(
        "claude_oauth_manual_code_exchanged",
        has_credentials=bool(credentials),
        category="auth",
    )

    return credentials

cleanup async

cleanup()

Cleanup resources.

Source code in ccproxy/plugins/oauth_claude/provider.py

async def cleanup(self) -> None:
    """Cleanup resources."""
    if self.client:
        await self.client.close()