ccproxy.plugins.oauth_codex

ccproxy.plugins.oauth_codex

OAuth Codex plugin for standalone OpenAI Codex OAuth authentication.

CodexOAuthClient

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

Bases: BaseOAuthClient[OpenAICredentials]

Codex/OpenAI OAuth implementation for the OAuth Codex plugin.

Parameters:

Name	Type	Description	Default
config	CodexOAuthConfig	OAuth configuration	required
storage	TokenStorage[OpenAICredentials] | None	Token storage backend	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/client.py

def __init__(
    self,
    config: CodexOAuthConfig,
    storage: TokenStorage[OpenAICredentials] | None = None,
    http_client: httpx.AsyncClient | None = None,
    hook_manager: Any | None = None,
    settings: Settings | None = None,
):
    """Initialize Codex OAuth client.

    Args:
        config: OAuth configuration
        storage: Token storage backend
        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.oauth_config = config

    # Resolve effective redirect URI from config
    redirect_uri = config.get_redirect_uri()

    # Initialize base class
    super().__init__(
        client_id=config.client_id,
        redirect_uri=redirect_uri,
        base_url=config.base_url,
        scopes=config.scopes,
        storage=storage,
        http_client=http_client,
        hook_manager=hook_manager,
        settings=settings,
    )

get_custom_auth_params

get_custom_auth_params()

Get OpenAI-specific authorization parameters.

Returns:

Type	Description
dict[str, str]	Dictionary of custom parameters

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

def get_custom_auth_params(self) -> dict[str, str]:
    """Get OpenAI-specific authorization parameters.

    Returns:
        Dictionary of custom parameters
    """
    # OpenAI does not use the audience parameter in authorization requests
    return {}

get_custom_headers

get_custom_headers()

Get OpenAI-specific HTTP headers.

Returns:

Type	Description
dict[str, str]	Dictionary of custom headers

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

def get_custom_headers(self) -> dict[str, str]:
    """Get OpenAI-specific HTTP headers.

    Returns:
        Dictionary of custom headers
    """
    return {
        "User-Agent": self.oauth_config.user_agent,
    }

parse_token_response async

parse_token_response(data)

Parse OpenAI-specific token response.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Raw token response from OpenAI	required

Returns:

Type	Description
OpenAICredentials	OpenAI credentials object

Raises:

Type	Description
OAuthError	If response parsing fails

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

async def parse_token_response(self, data: dict[str, Any]) -> OpenAICredentials:
    """Parse OpenAI-specific token response.

    Args:
        data: Raw token response from OpenAI

    Returns:
        OpenAI credentials object

    Raises:
        OAuthError: If response parsing fails
    """
    try:
        # Extract tokens
        access_token: str = data["access_token"]
        refresh_token: str = data.get("refresh_token", "")
        id_token: str = data.get("id_token", "")

        # Build credentials in the current nested schema; legacy inputs are also accepted
        # by the model's validator if needed.
        tokens = OpenAITokens(
            id_token=SecretStr(id_token),
            access_token=SecretStr(access_token),
            refresh_token=SecretStr(refresh_token or ""),
            account_id="",
        )
        credentials = OpenAICredentials(
            OPENAI_API_KEY=None,
            tokens=tokens,
            last_refresh=datetime.now(UTC).replace(microsecond=0).isoformat(),
            active=True,
        )

        # Try to extract account_id from JWT claims (id_token preferred)
        try:
            token_to_decode = id_token or access_token
            decoded = jwt.decode(
                token_to_decode, options={"verify_signature": False}
            )
            account_id = (
                decoded.get("sub")
                or decoded.get("account_id")
                or decoded.get("org_id")
                or ""
            )
            # Pydantic model has properties mapping; update underlying field
            credentials.tokens.account_id = str(account_id)
            logger.debug(
                "codex_oauth_id_token_decoded",
                sub=decoded.get("sub"),
                email=decoded.get("email"),
                category="auth",
            )
        except Exception as e:
            logger.warning(
                "codex_oauth_id_token_decode_error",
                error=str(e),
                exc_info=e,
                category="auth",
            )

        logger.info(
            "codex_oauth_credentials_parsed",
            has_refresh_token=bool(refresh_token),
            has_id_token=bool(id_token),
            account_id=credentials.account_id,
            category="auth",
        )

        return credentials

    except KeyError as e:
        logger.error(
            "codex_oauth_token_response_missing_field",
            missing_field=str(e),
            response_keys=list(data.keys()),
            category="auth",
        )
        raise OAuthError(f"Missing required field in token response: {e}") from e
    except Exception as e:
        logger.error(
            "codex_oauth_token_response_parse_error",
            error=str(e),
            error_type=type(e).__name__,
            category="auth",
        )
        raise OAuthError(f"Failed to parse OpenAI token response: {e}") from e

refresh_token async

refresh_token(refresh_token)

Refresh OpenAI access token.

Parameters:

Name	Type	Description	Default
refresh_token	str	Refresh token	required

Returns:

Type	Description
OpenAICredentials	New OpenAI credentials

Raises:

Type	Description
OAuthError	If refresh fails

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

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

    Args:
        refresh_token: Refresh token

    Returns:
        New OpenAI credentials

    Raises:
        OAuthError: If refresh fails
    """
    token_endpoint = self._get_token_endpoint()
    data = {
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "client_id": self.client_id,
        "scope": "openid profile email offline_access",
    }
    headers = self.get_custom_headers()
    headers["Content-Type"] = "application/x-www-form-urlencoded"

    try:
        response = await self.http_client.post(
            token_endpoint,
            data=data,  # OpenAI uses form encoding
            headers=headers,
            timeout=30.0,
        )
        response.raise_for_status()

        token_response = response.json()
        return await self.parse_token_response(token_response)

    except Exception as e:
        logger.error(
            "codex_oauth_token_refresh_failed",
            error=str(e),
            exc_info=False,
            category="auth",
        )
        raise OAuthError(f"Failed to refresh OpenAI token: {e}") from e

CodexOAuthConfig

Bases: BaseModel

OAuth-specific configuration for OpenAI Codex.

get_redirect_uri

get_redirect_uri()

Return redirect URI, auto-generated from callback_port when unset.

Uses the standard plugin callback path: /auth/callback.

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

def get_redirect_uri(self) -> str:
    """Return redirect URI, auto-generated from callback_port when unset.

    Uses the standard plugin callback path: `/auth/callback`.
    """
    if self.redirect_uri:
        return self.redirect_uri
    return f"http://localhost:{self.callback_port}/auth/callback"

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()

CodexTokenStorage

CodexTokenStorage(storage_path=None)

Bases: BaseJsonStorage[OpenAICredentials]

Codex/OpenAI OAuth-specific token storage implementation.

Parameters:

Name	Type	Description	Default
storage_path	Path | None	Path to storage file	None

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

def __init__(self, storage_path: Path | None = None):
    """Initialize Codex token storage.

    Args:
        storage_path: Path to storage file
    """
    if storage_path is None:
        # Default to standard OpenAI credentials location
        storage_path = Path.home() / ".codex" / "auth.json"

    super().__init__(storage_path)
    self.provider_name = "codex"

save async

save(credentials)

Save OpenAI credentials.

Parameters:

Name	Type	Description	Default
credentials	OpenAICredentials	OpenAI credentials to save	required

Returns:

Type	Description
bool	True if saved successfully, False otherwise

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

async def save(self, credentials: OpenAICredentials) -> bool:
    """Save OpenAI credentials.

    Args:
        credentials: OpenAI credentials to save

    Returns:
        True if saved successfully, False otherwise
    """
    try:
        # Convert to dict for storage
        data = credentials.model_dump(mode="json", exclude_none=True)

        # Use parent class's atomic write with backup
        await self._write_json(data)

        logger.info(
            "codex_oauth_credentials_saved",
            has_refresh_token=bool(credentials.refresh_token),
            storage_path=str(self.file_path),
            category="auth",
        )
        return True
    except Exception as e:
        logger.error(
            "codex_oauth_save_failed", error=str(e), exc_info=e, category="auth"
        )
        return False

load async

load()

Load OpenAI credentials.

Returns:

Type	Description
OpenAICredentials | None	Stored credentials or None

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

async def load(self) -> OpenAICredentials | None:
    """Load OpenAI credentials.

    Returns:
        Stored credentials or None
    """
    try:
        # Use parent class's read method (avoid redundant exists() checks)
        data = await self._read_json()
        if not data:
            logger.debug(
                "codex_auth_file_empty",
                storage_path=str(self.file_path),
                category="auth",
            )
            return None

        credentials = OpenAICredentials.model_validate(data)
        logger.info(
            "codex_oauth_credentials_loaded",
            has_refresh_token=bool(credentials.refresh_token),
            category="auth",
        )
        return credentials
    except Exception as e:
        logger.error(
            "codex_oauth_credentials_load_error",
            error=str(e),
            exc_info=e,
            category="auth",
        )
        return None