import logging import os import threading from datetime import datetime, timedelta, 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", ] # Refresh proactively: never hand out a token that could die mid-request. _ACCESS_TOKEN_REFRESH_BUFFER_SECONDS = 60 # Module-level access-token cache. Google access tokens live ~1h; previously # every get_credentials() call (sync, unsubscribe, ...) paid for a full token # exchange with Google. The lock keeps concurrent callers sharing one refresh # instead of racing each other to the token endpoint. _credentials_lock = threading.Lock() _cached_credentials: Credentials | None = None class OAuthNotConnected(Exception): pass def _client_config() -> dict: return { "web": { "client_id": settings.google_client_id, "client_secret": settings.google_client_secret, "auth_uri": settings.google_auth_uri, "token_uri": settings.google_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( settings.google_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(settings.google_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: global _cached_credentials encrypted = encrypt_token(credentials.refresh_token) 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 db.commit() # The exchanged credentials carry a fresh access token -- reuse them so # the immediately following syncs don't pay for a second Google request. with _credentials_lock: _cached_credentials = credentials def clear_credentials(db: Session) -> None: global _cached_credentials row = db.get(OAuthCredentials, SINGLETON_ID) if row is not None: db.delete(row) db.commit() with _credentials_lock: _cached_credentials = None 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 _token_missing_or_expiring(credentials: Credentials) -> bool: if not credentials.token: return True expiry = credentials.expiry if expiry is None: # Unknown expiry -- be conservative and refresh. return True if expiry.tzinfo is None: # google-auth reports expiry as a naive UTC datetime. expiry = expiry.replace(tzinfo=timezone.utc) return expiry <= datetime.now(timezone.utc) + timedelta(seconds=_ACCESS_TOKEN_REFRESH_BUFFER_SECONDS) def get_credentials(db: Session) -> Credentials: global _cached_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) with _credentials_lock: credentials = _cached_credentials if credentials is None or credentials.refresh_token != refresh_token: # No cached token yet, or the account was reconnected with a new # refresh token. The refresh below does the initial exchange. credentials = Credentials( token=None, refresh_token=refresh_token, token_uri=settings.google_token_uri, client_id=settings.google_client_id, client_secret=settings.google_client_secret, scopes=SCOPES, ) _cached_credentials = credentials if _token_missing_or_expiring(credentials): try: credentials.refresh(GoogleAuthRequest()) except Exception: # Don't keep a broken cached object behind. _cached_credentials = None raise return credentials