myYouTube/backend/app/services/google_oauth.py

import logging
import os
from datetime import datetime, timezone

# Google frequently echoes back a scope string that's a superset/reordering
# of what we requested (e.g. we ask for "youtube", Google's token response
# also lists "youtube.readonly" since it's implied). oauthlib does an exact
# string comparison by default and raises on any mismatch -- this disables
# that overly strict check. Must be set before requests_oauthlib reads it.
os.environ.setdefault("OAUTHLIB_RELAX_TOKEN_SCOPE", "1")

import httpx
from google.auth.transport.requests import Request as GoogleAuthRequest
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import Flow
from sqlalchemy.orm import Session

from app.config import settings
from app.core.crypto import decrypt_token, encrypt_token
from app.models.oauth_credentials import SINGLETON_ID, OAuthCredentials

logger = logging.getLogger(__name__)

SCOPES = [
    # Full read/write scope, not just youtube.readonly: unsubscribing from a
    # channel (subscriptions.delete) requires write access. Deliberate
    # deviation from the original "minimal scope" TZ recommendation, per
    # explicit user request. Existing stored refresh tokens were granted
    # under the old readonly-only scope and won't cover this -- users must
    # reconnect once for this to take effect.
    "https://www.googleapis.com/auth/youtube",
    "openid",
    "https://www.googleapis.com/auth/userinfo.email",
    "https://www.googleapis.com/auth/userinfo.profile",
]

AUTH_URI = "https://accounts.google.com/o/oauth2/auth"
TOKEN_URI = "https://oauth2.googleapis.com/token"
USERINFO_URI = "https://www.googleapis.com/oauth2/v3/userinfo"
REVOKE_URI = "https://oauth2.googleapis.com/revoke"


class OAuthNotConnected(Exception):
    pass


def _client_config() -> dict:
    return {
        "web": {
            "client_id": settings.google_client_id,
            "client_secret": settings.google_client_secret,
            "auth_uri": AUTH_URI,
            "token_uri": TOKEN_URI,
            "redirect_uris": [settings.google_redirect_uri],
        }
    }


def _build_flow(state: str | None = None) -> Flow:
    return Flow.from_client_config(
        _client_config(),
        scopes=SCOPES,
        state=state,
        redirect_uri=settings.google_redirect_uri,
    )


def build_authorization_url() -> tuple[str, str]:
    flow = _build_flow()
    auth_url, state = flow.authorization_url(
        access_type="offline",
        prompt="consent",
        include_granted_scopes="true",
    )
    return auth_url, state


def exchange_code(code: str, state: str) -> Credentials:
    flow = _build_flow(state=state)
    flow.fetch_token(code=code)
    return flow.credentials


def fetch_userinfo(access_token: str) -> dict:
    response = httpx.get(
        USERINFO_URI,
        headers={"Authorization": f"Bearer {access_token}"},
        timeout=settings.metube_request_timeout_seconds,
    )
    response.raise_for_status()
    return response.json()


def revoke_token(token: str) -> None:
    try:
        httpx.post(REVOKE_URI, params={"token": token}, timeout=10)
    except Exception:
        logger.warning("Failed to revoke Google token", exc_info=True)


def store_credentials(db: Session, google_email: str, credentials: Credentials) -> None:
    encrypted = encrypt_token(credentials.refresh_token)
    expires_at = credentials.expiry
    if expires_at is not None and expires_at.tzinfo is None:
        expires_at = expires_at.replace(tzinfo=timezone.utc)

    row = db.get(OAuthCredentials, SINGLETON_ID)
    if row is None:
        row = OAuthCredentials(id=SINGLETON_ID, google_email=google_email, encrypted_refresh_token=encrypted)
        db.add(row)
    else:
        row.google_email = google_email
        row.encrypted_refresh_token = encrypted
    row.access_token_expires_at = expires_at
    db.commit()


def clear_credentials(db: Session) -> None:
    row = db.get(OAuthCredentials, SINGLETON_ID)
    if row is not None:
        db.delete(row)
        db.commit()


def is_connected(db: Session) -> bool:
    return db.get(OAuthCredentials, SINGLETON_ID) is not None


def get_connected_email(db: Session) -> str | None:
    row = db.get(OAuthCredentials, SINGLETON_ID)
    return row.google_email if row else None


def get_credentials(db: Session) -> Credentials:
    row = db.get(OAuthCredentials, SINGLETON_ID)
    if row is None:
        raise OAuthNotConnected("Google account is not connected")

    refresh_token = decrypt_token(row.encrypted_refresh_token)
    credentials = Credentials(
        token=None,
        refresh_token=refresh_token,
        token_uri=TOKEN_URI,
        client_id=settings.google_client_id,
        client_secret=settings.google_client_secret,
        scopes=SCOPES,
    )
    credentials.refresh(GoogleAuthRequest())

    expires_at = credentials.expiry
    if expires_at is not None and expires_at.tzinfo is None:
        expires_at = expires_at.replace(tzinfo=timezone.utc)
    row.access_token_expires_at = expires_at
    db.commit()

    return credentials