ccproxy.plugins.oauth_codex.provider

ccproxy.plugins.oauth_codex.provider

Codex/OpenAI OAuth provider for plugin registration.

CodexOAuthProvider

CodexOAuthProvider(
    config=None,
    storage=None,
    http_client=None,
    hook_manager=None,
    settings=None,
)

Bases: ProfileLoggingMixin

Codex/OpenAI OAuth provider implementation for registry.

Parameters:

Name	Type	Description	Default
config	CodexOAuthConfig | None	OAuth configuration	None
storage	CodexTokenStorage | 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
settings	Settings | None	Optional settings for HTTP client configuration	None

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

def __init__(
    self,
    config: CodexOAuthConfig | None = None,
    storage: CodexTokenStorage | None = None,
    http_client: httpx.AsyncClient | None = None,
    hook_manager: Any | None = None,
    settings: Settings | None = None,
):
    """Initialize Codex 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
        settings: Optional settings for HTTP client configuration
    """
    self.config = config or CodexOAuthConfig()
    self.storage = storage or CodexTokenStorage()
    self.hook_manager = hook_manager
    self.http_client = http_client
    self.settings = settings

    self.client = CodexOAuthClient(
        self.config,
        self.storage,
        http_client,
        hook_manager=hook_manager,
        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_codex/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
    """
    params = {
        "response_type": "code",
        "client_id": self.config.client_id,
        "redirect_uri": redirect_uri or self.config.get_redirect_uri(),
        "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(
        "codex_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	OpenAI credentials object

Source code in ccproxy/plugins/oauth_codex/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:
        OpenAI 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 = CodexOAuthConfig(
            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,
            audience=self.config.audience,
            use_pkce=self.config.use_pkce,
        )

        # Create temporary client with the correct redirect URI
        temp_client = CodexOAuthClient(
            temp_config,
            self.storage,
            self.http_client,
            hook_manager=self.hook_manager,
            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(
        "codex_oauth_callback_handled",
        state=state,
        has_credentials=bool(credentials),
        has_id_token=bool(credentials.id_token),
        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_codex/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("codex_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_codex/provider.py

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

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

    logger.info("codex_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_codex/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 OpenAI Codex",
        supports_pkce=self.supports_pkce,
        scopes=self.config.scopes,
        is_available=True,
        plugin_name="oauth_codex",
    )

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_codex/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
    """
    # OpenAI doesn't have a validation endpoint, so we check if stored token matches
    if self.storage:
        credentials = await self.storage.load()
        if credentials:
            return credentials.access_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_codex/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
    if self.storage:
        credentials = await self.storage.load()
        if credentials:
            info = {
                "account_id": credentials.account_id,
                "active": credentials.active,
                "has_id_token": bool(credentials.id_token),
            }

            # Try to extract info from ID token if present
            if credentials.id_token:
                try:
                    import jwt

                    decoded = jwt.decode(
                        credentials.id_token,
                        options={"verify_signature": False},
                    )
                    info.update(
                        {
                            "email": decoded.get("email"),
                            "name": decoded.get("name"),
                            "sub": decoded.get("sub"),
                        }
                    )
                except Exception:
                    pass

            return info
    return None

get_storage

get_storage()

Get storage implementation for this provider.

Returns:

Type	Description
Any	Storage implementation

Source code in ccproxy/plugins/oauth_codex/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_codex/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	OpenAI 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_codex/provider.py

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

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

    Returns:
        True if saved successfully, False otherwise
    """
    from pathlib import Path

    from ccproxy.auth.storage.generic import GenericJsonStorage

    from .manager import CodexTokenManager
    from .models import OpenAICredentials

    try:
        if custom_path:
            # Use custom path for storage
            storage = GenericJsonStorage(Path(custom_path), OpenAICredentials)
            manager = await CodexTokenManager.create(storage=storage)
        else:
            # Use default storage
            manager = await CodexTokenManager.create()

        return await manager.save_credentials(credentials)
    except Exception as e:
        logger.error(
            "Failed to save OpenAI 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_codex/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
    """
    from pathlib import Path

    from ccproxy.auth.storage.generic import GenericJsonStorage

    from .manager import CodexTokenManager
    from .models import OpenAICredentials

    try:
        if custom_path:
            # Load from custom path
            storage = GenericJsonStorage(Path(custom_path), OpenAICredentials)
            manager = await CodexTokenManager.create(storage=storage)
        else:
            # Load from default storage
            manager = await CodexTokenManager.create()

        credentials = await manager.load_credentials()

        # Use standardized profile logging
        self._log_credentials_loaded("codex", credentials)

        return credentials
    except Exception as e:
        logger.error(
            "Failed to load OpenAI 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 and return the token manager instance.

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

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

async def create_token_manager(self, storage: Any | None = None) -> Any:
    """Create and return the token manager instance.

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

    return await CodexTokenManager.create(storage=storage)

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	OpenAI credentials object

Source code in ccproxy/plugins/oauth_codex/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:
        OpenAI credentials object
    """
    # For manual code flow, use OOB redirect URI and no state validation
    credentials: OpenAICredentials = await self.client.handle_callback(
        code, "manual", ""
    )

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

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

    return credentials

cleanup async

cleanup()

Cleanup resources.

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

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