ccproxy.auth.openai.oauth_client

ccproxy.auth.openai.oauth_client

OpenAI OAuth PKCE client implementation.

OpenAIOAuthClient

OpenAIOAuthClient(settings, token_manager=None)

OpenAI OAuth PKCE flow client.

Parameters:

Name	Type	Description	Default
settings	CodexSettings	Codex configuration settings	required
token_manager	OpenAITokenManager | None	Token manager for credential storage	None

Source code in ccproxy/auth/openai/oauth_client.py

def __init__(
    self, settings: CodexSettings, token_manager: OpenAITokenManager | None = None
):
    """Initialize OAuth client.

    Args:
        settings: Codex configuration settings
        token_manager: Token manager for credential storage
    """
    self.settings = settings
    self.token_manager = token_manager or OpenAITokenManager()
    self._server_task: asyncio.Task[None] | None = None
    self._auth_complete = asyncio.Event()
    self._auth_result: OpenAICredentials | None = None
    self._auth_error: str | None = None

authenticate async

authenticate(open_browser=True)

Perform OAuth PKCE flow.

Parameters:

Name	Type	Description	Default
open_browser	bool	Whether to automatically open browser	True

Returns:

Type	Description
OpenAICredentials	OpenAI credentials

Raises:

Type	Description
ValueError	If authentication fails

Source code in ccproxy/auth/openai/oauth_client.py

async def authenticate(self, open_browser: bool = True) -> OpenAICredentials:
    """Perform OAuth PKCE flow.

    Args:
        open_browser: Whether to automatically open browser

    Returns:
        OpenAI credentials

    Raises:
        ValueError: If authentication fails
    """
    # Reset state
    self._auth_complete.clear()
    self._auth_result = None
    self._auth_error = None

    # Generate PKCE parameters
    code_verifier, code_challenge = self._generate_pkce_pair()
    state = secrets.token_urlsafe(32)

    # Create callback app
    app = self._create_callback_app(code_verifier, state)

    # Start callback server
    self._server_task = asyncio.create_task(self._run_callback_server(app))

    # Give server time to start
    await asyncio.sleep(1)

    # Build authorization URL
    auth_url = self._build_auth_url(code_challenge, state)

    logger.info("Starting OpenAI OAuth flow")
    print("\nPlease visit this URL to authenticate with OpenAI:")
    print(f"{auth_url}\n")

    if open_browser:
        try:
            webbrowser.open(auth_url)
            print("Opening browser...")
        except Exception as e:
            logger.warning("Failed to open browser automatically", error=str(e))
            print("Please copy and paste the URL above into your browser.")

    print("Waiting for authentication to complete...")

    try:
        # Wait for authentication to complete (with timeout)
        await asyncio.wait_for(self._auth_complete.wait(), timeout=300)  # 5 minutes

        if self._auth_error:
            raise ValueError(self._auth_error)

        if not self._auth_result:
            raise ValueError("Authentication completed but no credentials received")

        logger.info("OpenAI authentication successful")  # type: ignore[unreachable]
        return self._auth_result

    except TimeoutError as e:
        raise ValueError("Authentication timed out (5 minutes)") from e
    finally:
        # Clean up server
        if self._server_task and not self._server_task.done():
            self._server_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._server_task