oauth_handler.py

"""
Google OAuth Authentication Handler
Provides OAuth login/logout functionality for Wikipedia API
"""
from __future__ import annotations

from google_auth_oauthlib.flow import Flow
from google.oauth2 import id_token
from google.auth.transport import requests as google_requests
import os
import secrets
from typing import Optional, Tuple
from fastapi import HTTPException
from datetime import datetime, timedelta

# OAuth Configuration
GOOGLE_CLIENT_ID = os.environ.get("GOOGLE_CLIENT_ID", "")
GOOGLE_CLIENT_SECRET = os.environ.get("GOOGLE_CLIENT_SECRET", "")
OAUTH_REDIRECT_URI = os.environ.get("OAUTH_REDIRECT_URI", "https://wiki.built-simple.ai/auth/google/callback")
APP_URL = os.environ.get("APP_URL", "https://wiki.built-simple.ai")

# Session storage (in-memory, replace with Redis for production)
oauth_sessions = {}

class OAuthHandler:
    """Handle Google OAuth authentication"""

    def __init__(self):
        if not GOOGLE_CLIENT_ID or not GOOGLE_CLIENT_SECRET:
            raise ValueError("Google OAuth credentials not configured")

        self.client_config = {
            "web": {
                "client_id": GOOGLE_CLIENT_ID,
                "client_secret": GOOGLE_CLIENT_SECRET,
                "auth_uri": "https://accounts.google.com/o/oauth2/auth",
                "token_uri": "https://oauth2.googleapis.com/token",
                "redirect_uris": [OAUTH_REDIRECT_URI]
            }
        }

        self.scopes = [
            'openid',
            'https://www.googleapis.com/auth/userinfo.email',
            'https://www.googleapis.com/auth/userinfo.profile'
        ]

    def create_authorization_url(self) -> tuple[str, str]:
        """
        Create authorization URL for Google OAuth.
        Returns: (authorization_url, state)
        """
        flow = Flow.from_client_config(
            self.client_config,
            scopes=self.scopes,
            redirect_uri=OAUTH_REDIRECT_URI
        )

        authorization_url, state = flow.authorization_url(
            access_type='offline',
            include_granted_scopes='true',
            prompt='select_account'
        )

        # Store state in session for CSRF protection
        oauth_sessions[state] = {
            'created_at': datetime.now(),
            'flow': flow
        }

        return authorization_url, state

    def handle_callback(self, state: str, code: str, error: Optional[str] = None) -> dict:
        """
        Handle OAuth callback from Google.
        Returns: user_info dict with email, name, picture
        """
        if error:
            raise HTTPException(status_code=400, detail=f"OAuth error: {error}")

        if not state or state not in oauth_sessions:
            raise HTTPException(status_code=400, detail="Invalid or expired state")

        session = oauth_sessions.pop(state)

        # Check session age (expire after 10 minutes)
        if datetime.now() - session['created_at'] > timedelta(minutes=10):
            raise HTTPException(status_code=400, detail="OAuth session expired")

        flow = session['flow']

        try:
            # Exchange authorization code for tokens
            flow.fetch_token(code=code)
            credentials = flow.credentials

            # Verify ID token and extract user info
            idinfo = id_token.verify_oauth2_token(
                credentials.id_token,
                google_requests.Request(),
                GOOGLE_CLIENT_ID
            )

            # Extract user information
            user_info = {
                'email': idinfo.get('email'),
                'name': idinfo.get('name'),
                'picture': idinfo.get('picture'),
                'google_id': idinfo.get('sub'),
                'email_verified': idinfo.get('email_verified', False)
            }

            return user_info

        except ValueError as e:
            raise HTTPException(status_code=400, detail=f"Token verification failed: {str(e)}")
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"OAuth error: {str(e)}")

    def cleanup_expired_sessions(self):
        """Remove expired OAuth sessions (older than 10 minutes)"""
        now = datetime.now()
        expired = [
            state for state, session in oauth_sessions.items()
            if now - session['created_at'] > timedelta(minutes=10)
        ]
        for state in expired:
            del oauth_sessions[state]


# Global OAuth handler instance
oauth_handler = OAuthHandler() if GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET else None