ccproxy.auth.oauth.registry

ccproxy.auth.oauth.registry

OAuth Provider Registry for dynamic provider management.

This module provides a central registry where plugins can register their OAuth providers at runtime, enabling dynamic discovery and management of OAuth flows.

FlowType

Bases: str, Enum

OAuth flow types for CLI authentication.

CliAuthConfig dataclass

CliAuthConfig(
    preferred_flow=browser,
    callback_port=8080,
    callback_path="/callback",
    fixed_redirect_uri=None,
    manual_redirect_uri=None,
    supports_manual_code=True,
    supports_device_flow=False,
)

CLI authentication configuration for OAuth providers.

OAuthProviderInfo

Bases: BaseModel

Information about a registered OAuth provider.

OAuthProviderProtocol

Bases: Protocol

Protocol for OAuth provider implementations.

provider_name property

provider_name

Internal provider name (e.g., 'claude-api', 'codex').

provider_display_name property

provider_display_name

Display name for UI (e.g., 'Claude API', 'OpenAI Codex').

supports_pkce property

supports_pkce

Whether this provider supports PKCE flow.

supports_refresh property

supports_refresh

Whether this provider supports token refresh.

cli property

cli

CLI authentication configuration for this provider.

Returns:

Type	Description
CliAuthConfig	Configuration object specifying CLI flow preferences and capabilities

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
redirect_uri	str | None	Redirect URI for OAuth callback	None

Returns:

Type	Description
str	Authorization URL to redirect user to

Source code in ccproxy/auth/oauth/registry.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)
        redirect_uri: Redirect URI for OAuth callback

    Returns:
        Authorization URL to redirect user to
    """
    ...

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 the authorization request	None

Returns:

Type	Description
Any	Provider-specific credentials object

Source code in ccproxy/auth/oauth/registry.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 the authorization request

    Returns:
        Provider-specific credentials object
    """
    ...

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/auth/oauth/registry.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
    """
    ...

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/auth/oauth/registry.py

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

    Args:
        token: Token to revoke
    """
    ...

get_provider_info

get_provider_info()

Get provider information for discovery.

Returns:

Type	Description
OAuthProviderInfo	Provider information

Source code in ccproxy/auth/oauth/registry.py

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

    Returns:
        Provider information
    """
    ...

get_storage

get_storage()

Get storage implementation for this provider.

Returns:

Type	Description
Any	Storage implementation or None

Source code in ccproxy/auth/oauth/registry.py

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

    Returns:
        Storage implementation or None
    """
    ...

get_credential_summary

get_credential_summary(credentials)

Get a summary of credentials for display.

Parameters:

Name	Type	Description	Default
credentials	Any	Provider-specific credentials	required

Returns:

Type	Description
dict[str, Any]	Dictionary with display-friendly credential summary

Source code in ccproxy/auth/oauth/registry.py

def get_credential_summary(self, credentials: Any) -> dict[str, Any]:
    """Get a summary of credentials for display.

    Args:
        credentials: Provider-specific credentials

    Returns:
        Dictionary with display-friendly credential summary
    """
    ...

start_device_flow async

start_device_flow()

Start OAuth device code flow.

Returns:

Type	Description
tuple[str, str, str, int]	Tuple of (device_code, user_code, verification_uri, expires_in)

Raises:

Type	Description
NotImplementedError	If device flow is not supported

Source code in ccproxy/auth/oauth/registry.py

async def start_device_flow(self) -> tuple[str, str, str, int]:
    """Start OAuth device code flow.

    Returns:
        Tuple of (device_code, user_code, verification_uri, expires_in)

    Raises:
        NotImplementedError: If device flow is not supported
    """
    raise NotImplementedError("Device flow not supported by this provider")

complete_device_flow async

complete_device_flow(device_code, interval, expires_in)

Complete OAuth device code flow by polling for authorization.

Parameters:

Name	Type	Description	Default
device_code	str	Device code from start_device_flow	required
interval	int	Polling interval in seconds	required
expires_in	int	Code expiration time in seconds	required

Returns:

Type	Description
Any	Provider-specific credentials object

Raises:

Type	Description
NotImplementedError	If device flow is not supported

Source code in ccproxy/auth/oauth/registry.py

async def complete_device_flow(
    self, device_code: str, interval: int, expires_in: int
) -> Any:
    """Complete OAuth device code flow by polling for authorization.

    Args:
        device_code: Device code from start_device_flow
        interval: Polling interval in seconds
        expires_in: Code expiration time in seconds

    Returns:
        Provider-specific credentials object

    Raises:
        NotImplementedError: If device flow is not supported
    """
    raise NotImplementedError("Device flow not supported by this provider")

exchange_manual_code async

exchange_manual_code(code)

Exchange manually entered authorization code for tokens.

This method handles the case where users manually copy/paste authorization codes in restricted environments.

Parameters:

Name	Type	Description	Default
code	str	Authorization code entered manually by user	required

Returns:

Type	Description
Any	Provider-specific credentials object

Raises:

Type	Description
NotImplementedError	If manual code entry is not implemented

Source code in ccproxy/auth/oauth/registry.py

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

    This method handles the case where users manually copy/paste
    authorization codes in restricted environments.

    Args:
        code: Authorization code entered manually by user

    Returns:
        Provider-specific credentials object

    Raises:
        NotImplementedError: If manual code entry is not implemented
    """
    raise NotImplementedError("Manual code entry not implemented by this provider")

save_credentials async

save_credentials(credentials, custom_path=None)

Save credentials using provider's storage mechanism.

Parameters:

Name	Type	Description	Default
credentials	Any	Provider-specific credentials object	required
custom_path	Any | None	Optional custom storage path	None

Returns:

Type	Description
bool	True if saved successfully, False otherwise

Source code in ccproxy/auth/oauth/registry.py

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

    Args:
        credentials: Provider-specific credentials object
        custom_path: Optional custom storage path

    Returns:
        True if saved successfully, False otherwise
    """
    ...

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	None

Returns:

Type	Description
Any | None	Credentials if found, None otherwise

Source code in ccproxy/auth/oauth/registry.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

    Returns:
        Credentials if found, None otherwise
    """
    ...

OAuthRegistry

OAuthRegistry()

Central registry for OAuth providers.

This registry allows plugins to register their OAuth providers at runtime, enabling dynamic discovery and management of OAuth authentication flows.

Source code in ccproxy/auth/oauth/registry.py

def __init__(self) -> None:
    """Initialize the OAuth registry."""
    self._providers: dict[str, OAuthProviderProtocol] = {}
    self._provider_info_cache: dict[str, OAuthProviderInfo] = {}
    logger.debug("oauth_registry_initialized", category="auth")

register

register(provider)

Register an OAuth provider from a plugin.

Parameters:

Name	Type	Description	Default
provider	OAuthProviderProtocol	OAuth provider implementation	required

Raises:

Type	Description
ValueError	If provider with same name already registered

Source code in ccproxy/auth/oauth/registry.py

def register(self, provider: OAuthProviderProtocol) -> None:
    """Register an OAuth provider from a plugin.

    Args:
        provider: OAuth provider implementation

    Raises:
        ValueError: If provider with same name already registered
    """
    provider_name = provider.provider_name

    if provider_name in self._providers:
        raise ValueError(f"OAuth provider '{provider_name}' is already registered")

    self._providers[provider_name] = provider

    # Cache provider info
    try:
        info = provider.get_provider_info()
        self._provider_info_cache[provider_name] = info
        logger.debug(
            "oauth_provider_registered",
            provider=provider_name,
            display_name=info.display_name,
            supports_pkce=info.supports_pkce,
            plugin=info.plugin_name,
            category="auth",
        )
    except Exception as e:
        logger.error(
            "oauth_provider_info_error",
            provider=provider_name,
            error=str(e),
            exc_info=e,
            category="auth",
        )

unregister

unregister(provider_name)

Unregister an OAuth provider.

Parameters:

Name	Type	Description	Default
provider_name	str	Name of provider to unregister	required

Source code in ccproxy/auth/oauth/registry.py

def unregister(self, provider_name: str) -> None:
    """Unregister an OAuth provider.

    Args:
        provider_name: Name of provider to unregister
    """
    if provider_name in self._providers:
        del self._providers[provider_name]
        if provider_name in self._provider_info_cache:
            del self._provider_info_cache[provider_name]
        logger.debug(
            "oauth_provider_unregistered", provider=provider_name, category="auth"
        )

get

get(provider_name)

Get a registered OAuth provider by name.

Parameters:

Name	Type	Description	Default
provider_name	str	Name of the provider	required

Returns:

Type	Description
OAuthProviderProtocol | None	OAuth provider instance or None if not found

Source code in ccproxy/auth/oauth/registry.py

def get(self, provider_name: str) -> OAuthProviderProtocol | None:
    """Get a registered OAuth provider by name.

    Args:
        provider_name: Name of the provider

    Returns:
        OAuth provider instance or None if not found
    """
    return self._providers.get(provider_name)

list

list()

List all registered OAuth providers.

Returns:

Type	Description
dict[str, OAuthProviderInfo]	Dictionary mapping provider names to their info

Source code in ccproxy/auth/oauth/registry.py

def list(self) -> dict[str, OAuthProviderInfo]:
    """List all registered OAuth providers.

    Returns:
        Dictionary mapping provider names to their info
    """
    result = {}
    for name, provider in self._providers.items():
        # Try to get fresh info, fall back to cache
        try:
            info = provider.get_provider_info()
            self._provider_info_cache[name] = info
            result[name] = info
        except Exception as e:
            logger.warning(
                "oauth_provider_info_refresh_error",
                provider=name,
                error=str(e),
                category="auth",
            )
            # Use cached info if available
            if name in self._provider_info_cache:
                result[name] = self._provider_info_cache[name]

    return result

has

has(provider_name)

Check if a provider is registered.

Parameters:

Name	Type	Description	Default
provider_name	str	Name of the provider	required

Returns:

Type	Description
bool	True if provider is registered

Source code in ccproxy/auth/oauth/registry.py

def has(self, provider_name: str) -> bool:
    """Check if a provider is registered.

    Args:
        provider_name: Name of the provider

    Returns:
        True if provider is registered
    """
    return provider_name in self._providers

get_info

get_info(provider_name)

Get information about a specific provider.

Parameters:

Name	Type	Description	Default
provider_name	str	Name of the provider	required

Returns:

Type	Description
OAuthProviderInfo | None	Provider information or None if not found

Source code in ccproxy/auth/oauth/registry.py

def get_info(self, provider_name: str) -> OAuthProviderInfo | None:
    """Get information about a specific provider.

    Args:
        provider_name: Name of the provider

    Returns:
        Provider information or None if not found
    """
    provider = self.get(provider_name)
    if not provider:
        return None

    try:
        info = provider.get_provider_info()
        self._provider_info_cache[provider_name] = info
        return info
    except Exception as e:
        logger.error(
            "oauth_provider_info_error",
            provider=provider_name,
            error=str(e),
            exc_info=e,
            category="auth",
        )
        # Return cached info if available
        return self._provider_info_cache.get(provider_name)

clear

clear()

Clear all registered providers.

This is mainly useful for testing or shutdown scenarios.

Source code in ccproxy/auth/oauth/registry.py

def clear(self) -> None:
    """Clear all registered providers.

    This is mainly useful for testing or shutdown scenarios.
    """
    self._providers.clear()
    self._provider_info_cache.clear()
    logger.info("oauth_registry_cleared", category="auth")