ccproxy.plugins.oauth_codex.models

ccproxy.plugins.oauth_codex.models

OpenAI-specific authentication models.

OpenAITokens

Bases: BaseModel

Nested token structure from OpenAI OAuth.

serialize_secret

serialize_secret(value)

Serialize SecretStr to plain string for JSON output.

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

@field_serializer("id_token", "access_token", "refresh_token")
def serialize_secret(self, value: SecretStr) -> str:
    """Serialize SecretStr to plain string for JSON output."""
    return value.get_secret_value() if value else ""

validate_tokens classmethod

validate_tokens(v)

Convert string values to SecretStr.

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

@field_validator("id_token", "access_token", "refresh_token", mode="before")
@classmethod
def validate_tokens(cls, v: str | SecretStr | None) -> SecretStr | None:
    """Convert string values to SecretStr."""
    if v is None:
        return None
    if isinstance(v, str):
        return SecretStr(v)
    return v

OpenAICredentials

Bases: BaseModel

OpenAI authentication credentials model matching actual auth file schema.

access_token property

access_token

Get access token from nested structure.

refresh_token property

refresh_token

Get refresh token from nested structure.

id_token property

id_token

Get ID token from nested structure.

account_id property

account_id

Get account ID from nested structure.

expires_at property

expires_at

Extract expiration from access token JWT.

is_expired

is_expired()

Check if the access token is expired.

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

def is_expired(self) -> bool:
    """Check if the access token is expired."""
    now = datetime.now(UTC)
    return now >= self.expires_at

expires_in_seconds

expires_in_seconds()

Get seconds until token expires.

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

def expires_in_seconds(self) -> int:
    """Get seconds until token expires."""
    now = datetime.now(UTC)
    delta = self.expires_at - now
    return max(0, int(delta.total_seconds()))

to_dict

to_dict()

Convert to dictionary for storage.

Implements BaseCredentials protocol.

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

def to_dict(self) -> dict[str, Any]:
    """Convert to dictionary for storage.

    Implements BaseCredentials protocol.
    """
    return {
        "OPENAI_API_KEY": self.OPENAI_API_KEY,
        "tokens": {
            "id_token": self.tokens.id_token.get_secret_value(),
            "access_token": self.tokens.access_token.get_secret_value(),
            "refresh_token": self.tokens.refresh_token.get_secret_value(),
            "account_id": self.tokens.account_id,
        },
        "last_refresh": self.last_refresh,
        "active": self.active,
    }

from_dict classmethod

from_dict(data)

Create from dictionary.

Implements BaseCredentials protocol.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> "OpenAICredentials":
    """Create from dictionary.

    Implements BaseCredentials protocol.
    """
    return cls(**data)

OpenAITokenWrapper

Bases: BaseTokenInfo

Wrapper for OpenAI credentials that adds computed properties.

This wrapper maintains the original OpenAICredentials structure while providing a unified interface through BaseTokenInfo.

refresh_token_value property

refresh_token_value

Get refresh token.

expires_at_datetime property

expires_at_datetime

Get expiration (already a datetime in OpenAI).

account_id property

account_id

Get account ID (extracted from JWT by validator).

id_token property

id_token

Get ID token if available.

access_token_value

access_token_value()

Get access token (now SecretStr in OpenAI).

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

@computed_field
def access_token_value(self) -> str:
    """Get access token (now SecretStr in OpenAI)."""
    return self.credentials.access_token

OpenAIProfileInfo

Bases: BaseProfileInfo

OpenAI-specific profile extracted from JWT tokens.

OpenAI embeds profile information in JWT claims rather than providing a separate API endpoint.

chatgpt_account_id property

chatgpt_account_id

Get ChatGPT account ID from JWT claims.

organization_id property

organization_id

Get organization ID from JWT claims.

auth0_subject property

auth0_subject

Get Auth0 subject (sub claim).

from_token classmethod

from_token(credentials)

Extract profile from JWT token claims.

Parameters:

Name	Type	Description	Default
credentials	OpenAICredentials	OpenAI credentials containing JWT tokens	required

Returns:

Type	Description
OpenAIProfileInfo	OpenAIProfileInfo with all JWT claims preserved

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

@classmethod
def from_token(cls, credentials: OpenAICredentials) -> "OpenAIProfileInfo":
    """Extract profile from JWT token claims.

    Args:
        credentials: OpenAI credentials containing JWT tokens

    Returns:
        OpenAIProfileInfo with all JWT claims preserved
    """
    # Prefer id_token as it has more claims, fallback to access_token
    token_to_decode = credentials.id_token or credentials.access_token

    try:
        # Decode without verification to extract claims
        claims = jwt.decode(token_to_decode, options={"verify_signature": False})
        logger.debug(
            "Extracted JWT claims", num_claims=len(claims), category="auth"
        )
    except Exception as e:
        logger.warning("failed_to_decode_jwt_token", error=str(e), category="auth")
        claims = {}

    # Use the account_id already extracted by OpenAICredentials validator
    account_id = credentials.account_id

    # Extract common fields if present in claims
    email = claims.get("email", "")
    display_name = claims.get("name") or claims.get("given_name")

    # Store ALL JWT claims in extras for complete information
    # This includes: sub, aud, iss, exp, iat, org_id, chatgpt_account_id, etc.
    return cls(
        account_id=account_id,
        email=email,
        display_name=display_name,
        extras=claims,  # Preserve all JWT claims
    )