strat-gameplay-webapp/backend/app/services/oauth_state.py

"""
OAuth State Management Service

Stores and validates OAuth state parameters in Redis for CSRF protection.
State is stored with a 10-minute TTL and can only be used once.

Author: Claude (Jarvis)
Date: 2025-11-27
"""

import secrets
import logging
from app.services import redis_client

logger = logging.getLogger(f"{__name__}.OAuthStateService")

OAUTH_STATE_PREFIX = "oauth_state:"
OAUTH_STATE_TTL = 600  # 10 minutes


async def create_oauth_state(return_url: str = "/") -> str:
    """
    Create and store a new OAuth state token.

    Args:
        return_url: URL to redirect user after successful auth

    Returns:
        State token to include in OAuth request
    """
    state = secrets.token_urlsafe(32)
    key = f"{OAUTH_STATE_PREFIX}{state}"

    await redis_client.client.set(key, return_url, ex=OAUTH_STATE_TTL)
    logger.debug(f"Created OAuth state: {state[:10]}... for return_url: {return_url}")

    return state


async def validate_and_consume_state(state: str) -> str | None:
    """
    Validate state token and return stored redirect URL.

    State is deleted after validation (one-time use).

    Args:
        state: State token from OAuth callback

    Returns:
        Stored return_url if valid, None if invalid/expired
    """
    key = f"{OAUTH_STATE_PREFIX}{state}"
    return_url = await redis_client.client.get(key)

    if return_url:
        await redis_client.client.delete(key)
        logger.debug(f"Validated and consumed OAuth state: {state[:10]}...")
        return return_url

    logger.warning(f"Invalid or expired OAuth state: {state[:10]}...")
    return None