ccproxy.auth.oauth

ccproxy.auth.oauth

Public router shim for OAuth flows.

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
    """
    ...

get_oauth_flow_result

get_oauth_flow_result(state)

Get and remove OAuth flow result.

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

def get_oauth_flow_result(state: str) -> dict[str, Any] | None:
    """Get and remove OAuth flow result."""
    return _pending_flows.pop(state, None)

register_oauth_flow

register_oauth_flow(
    state, code_verifier, custom_paths=None
)

Register a pending OAuth flow.

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

def register_oauth_flow(
    state: str, code_verifier: str, custom_paths: list[Path] | None = None
) -> None:
    """Register a pending OAuth flow."""
    _pending_flows[state] = {
        "code_verifier": code_verifier,
        "custom_paths": custom_paths,
        "completed": False,
        "success": False,
        "error": None,
    }
    logger.debug(
        "Registered OAuth flow",
        state=state,
        operation="register_oauth_flow",
        category="auth",
    )