ccproxy.auth.oauth.protocol

ccproxy.auth.oauth.protocol

OAuth protocol definitions for plugin OAuth implementations.

This module defines the protocols and interfaces that plugins must implement to provide OAuth authentication capabilities.

StandardProfileFields

Bases: BaseModel

Standardized profile fields for consistent UI display across OAuth providers.

ProfileLoggingMixin

Mixin to provide standardized profile dump logging for OAuth providers.

get_standard_profile async

get_standard_profile(credentials=None)

Return standardized profile fields for UI display.

If credentials are not provided, attempts to load them via a provider's load_credentials() method when available. This method intentionally avoids network calls and relies on locally available information or cached profile data inside provider implementations.

Parameters:

Name	Type	Description	Default
credentials	Any | None	Optional provider-specific credentials	None

Returns:

Type	Description
StandardProfileFields | None	StandardProfileFields or None if unavailable

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

async def get_standard_profile(
    self, credentials: Any | None = None
) -> StandardProfileFields | None:
    """Return standardized profile fields for UI display.

    If credentials are not provided, attempts to load them via a
    provider's `load_credentials()` method when available. This method
    intentionally avoids network calls and relies on locally available
    information or cached profile data inside provider implementations.

    Args:
        credentials: Optional provider-specific credentials

    Returns:
        StandardProfileFields or None if unavailable
    """
    try:
        creds = credentials
        if creds is None and hasattr(self, "load_credentials"):
            # Best-effort local load (provider-specific, may use storage)
            load_fn = self.load_credentials
            if callable(load_fn):
                creds = await cast(Callable[[], Awaitable[Any]], load_fn)()

        if not creds:
            return None

        return self._extract_standard_profile(creds)
    except Exception as e:
        logger.debug(
            "standard_profile_generation_failed",
            provider=getattr(self, "provider_name", type(self).__name__),
            error=str(e),
        )
        return None

OAuthConfig

Bases: BaseModel

Base configuration for OAuth providers.

OAuthStorageProtocol

Bases: Protocol

Protocol for OAuth token storage implementations.

save_tokens async

save_tokens(
    provider,
    access_token,
    refresh_token=None,
    expires_in=None,
    **kwargs,
)

Save OAuth tokens.

Parameters:

Name	Type	Description	Default
provider	str	Provider name	required
access_token	str	Access token	required
refresh_token	str | None	Optional refresh token	None
expires_in	int | None	Token expiration in seconds	None
**kwargs	Any	Additional provider-specific data	{}

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

async def save_tokens(
    self,
    provider: str,
    access_token: str,
    refresh_token: str | None = None,
    expires_in: int | None = None,
    **kwargs: Any,
) -> None:
    """Save OAuth tokens.

    Args:
        provider: Provider name
        access_token: Access token
        refresh_token: Optional refresh token
        expires_in: Token expiration in seconds
        **kwargs: Additional provider-specific data
    """
    ...

get_tokens async

get_tokens(provider)

Retrieve stored tokens for a provider.

Parameters:

Name	Type	Description	Default
provider	str	Provider name	required

Returns:

Type	Description
dict[str, Any] | None	Token data or None if not found

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

async def get_tokens(self, provider: str) -> dict[str, Any] | None:
    """Retrieve stored tokens for a provider.

    Args:
        provider: Provider name

    Returns:
        Token data or None if not found
    """
    ...

delete_tokens async

delete_tokens(provider)

Delete stored tokens for a provider.

Parameters:

Name	Type	Description	Default
provider	str	Provider name	required

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

async def delete_tokens(self, provider: str) -> None:
    """Delete stored tokens for a provider.

    Args:
        provider: Provider name
    """
    ...

has_valid_tokens async

has_valid_tokens(provider)

Check if valid tokens exist for a provider.

Parameters:

Name	Type	Description	Default
provider	str	Provider name	required

Returns:

Type	Description
bool	True if valid tokens exist

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

async def has_valid_tokens(self, provider: str) -> bool:
    """Check if valid tokens exist for a provider.

    Args:
        provider: Provider name

    Returns:
        True if valid tokens exist
    """
    ...

OAuthConfigProtocol

Bases: Protocol

Protocol for OAuth configuration providers.

get_client_id

get_client_id()

Get OAuth client ID.

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

def get_client_id(self) -> str:
    """Get OAuth client ID."""
    ...

get_client_secret

get_client_secret()

Get OAuth client secret (if applicable).

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

def get_client_secret(self) -> str | None:
    """Get OAuth client secret (if applicable)."""
    ...

get_redirect_uri

get_redirect_uri()

Get OAuth redirect URI.

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

def get_redirect_uri(self) -> str:
    """Get OAuth redirect URI."""
    ...

get_authorize_url

get_authorize_url()

Get authorization endpoint URL.

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

def get_authorize_url(self) -> str:
    """Get authorization endpoint URL."""
    ...

get_token_url

get_token_url()

Get token endpoint URL.

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

def get_token_url(self) -> str:
    """Get token endpoint URL."""
    ...

get_scopes

get_scopes()

Get requested OAuth scopes.

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

def get_scopes(self) -> list[str]:
    """Get requested OAuth scopes."""
    ...

uses_pkce

uses_pkce()

Check if PKCE should be used.

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

def uses_pkce(self) -> bool:
    """Check if PKCE should be used."""
    ...

TokenResponse

Bases: BaseModel

Standard OAuth token response.

OAuthProviderBase

Bases: Protocol

Extended protocol for OAuth providers with additional capabilities.

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.

get_authorization_url async

get_authorization_url(state, code_verifier=None)

Get authorization URL.

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

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

handle_callback async

handle_callback(code, state, code_verifier=None)

Handle OAuth callback.

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

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

refresh_access_token async

refresh_access_token(refresh_token)

Refresh access token.

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

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

revoke_token async

revoke_token(token)

Revoke a token.

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

async def revoke_token(self, token: str) -> None:
    """Revoke a token."""
    ...

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

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

get_storage

get_storage()

Get storage implementation for this provider.

Returns:

Type	Description
OAuthStorageProtocol | None	Storage implementation or None if provider handles storage

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

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

    Returns:
        Storage implementation or None if provider handles storage
    """
    ...

get_config

get_config()

Get configuration for this provider.

Returns:

Type	Description
OAuthConfigProtocol | None	Configuration implementation or None

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

def get_config(self) -> OAuthConfigProtocol | None:
    """Get configuration for this provider.

    Returns:
        Configuration implementation or None
    """
    ...