ccproxy.auth.oauth.session

ccproxy.auth.oauth.session

OAuth session management for handling OAuth state and PKCE.

This module provides session management for OAuth flows, storing state, PKCE verifiers, and other session data during the OAuth process.

OAuthSessionManager

OAuthSessionManager(ttl_seconds=600)

Manages OAuth session data during authentication flows.

This is a simple in-memory implementation. In production, consider using Redis or another persistent store.

Parameters:

Name	Type	Description	Default
ttl_seconds	int	Time-to-live for sessions in seconds (default: 10 minutes)	600

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

def __init__(self, ttl_seconds: int = 600) -> None:
    """Initialize the session manager.

    Args:
        ttl_seconds: Time-to-live for sessions in seconds (default: 10 minutes)
    """
    self._sessions: dict[str, dict[str, Any]] = {}
    self._ttl_seconds = ttl_seconds
    logger.info(
        "oauth_session_manager_initialized",
        ttl_seconds=ttl_seconds,
        category="auth",
    )

create_session async

create_session(state, data)

Create a new OAuth session.

Parameters:

Name	Type	Description	Default
state	str	OAuth state parameter (session key)	required
data	dict[str, Any]	Session data to store	required

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

async def create_session(self, state: str, data: dict[str, Any]) -> None:
    """Create a new OAuth session.

    Args:
        state: OAuth state parameter (session key)
        data: Session data to store
    """
    self._sessions[state] = {
        **data,
        "created_at": time.time(),
    }
    logger.debug(
        "oauth_session_created",
        state=state,
        provider=data.get("provider"),
        has_pkce=bool(data.get("code_verifier")),
        category="auth",
    )

    # Clean up expired sessions
    await self._cleanup_expired()

get_session async

get_session(state)

Retrieve session data by state.

Parameters:

Name	Type	Description	Default
state	str	OAuth state parameter	required

Returns:

Type	Description
dict[str, Any] | None	Session data or None if not found/expired

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

async def get_session(self, state: str) -> dict[str, Any] | None:
    """Retrieve session data by state.

    Args:
        state: OAuth state parameter

    Returns:
        Session data or None if not found/expired
    """
    session = self._sessions.get(state)

    if not session:
        logger.debug("oauth_session_not_found", state=state, category="auth")
        return None

    # Check if session expired
    created_at = session.get("created_at", 0)
    if time.time() - created_at > self._ttl_seconds:
        logger.debug("oauth_session_expired", state=state, category="auth")
        await self.delete_session(state)
        return None

    logger.debug(
        "oauth_session_retrieved",
        state=state,
        provider=session.get("provider"),
        category="auth",
    )
    return session

delete_session async

delete_session(state)

Delete a session.

Parameters:

Name	Type	Description	Default
state	str	OAuth state parameter	required

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

async def delete_session(self, state: str) -> None:
    """Delete a session.

    Args:
        state: OAuth state parameter
    """
    if state in self._sessions:
        provider = self._sessions[state].get("provider")
        del self._sessions[state]
        logger.debug(
            "oauth_session_deleted", state=state, provider=provider, category="auth"
        )

clear_all

clear_all()

Clear all sessions (mainly for testing).

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

def clear_all(self) -> None:
    """Clear all sessions (mainly for testing)."""
    count = len(self._sessions)
    self._sessions.clear()
    logger.info("oauth_sessions_cleared", count=count, category="auth")

get_oauth_session_manager

get_oauth_session_manager()

Get the global OAuth session manager instance.

Returns:

Type	Description
OAuthSessionManager	Global OAuth session manager

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

def get_oauth_session_manager() -> OAuthSessionManager:
    """Get the global OAuth session manager instance.

    Returns:
        Global OAuth session manager
    """
    global _session_manager
    if _session_manager is None:
        _session_manager = OAuthSessionManager()
    return _session_manager

reset_oauth_session_manager

reset_oauth_session_manager()

Reset the global OAuth session manager.

This clears all sessions and creates a new manager. Mainly useful for testing.

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

def reset_oauth_session_manager() -> None:
    """Reset the global OAuth session manager.

    This clears all sessions and creates a new manager.
    Mainly useful for testing.
    """
    global _session_manager
    if _session_manager:
        _session_manager.clear_all()
    _session_manager = OAuthSessionManager()