Connecting local program to Twitch dev-portal credentials

auth.py

@@ -0,0 +1,91 @@
+import httpx
+import secrets
+from fastapi import APIRouter, Depends, HTTPException, Request, Response
+from fastapi.responses import RedirectResponse
+from sqlalchemy.orm import Session
+
+from config import settings
+from database import SessionLocal
+import models
+import security
+
+router = APIRouter()
+
+# Dependency to get a DB session
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+@router.get("/login/twitch")
+async def login_with_twitch(request: Request):
+    """
+    Step 1 of OAuth flow: Redirect the user to Twitch's authorization page.
+    """
+    # Generate a random state token for CSRF protection
+    state = secrets.token_urlsafe(16)
+    request.session['oauth_state'] = state
+
+    # As per RESEARCH_REPORT.md, these are the minimum required scopes
+    scopes = "chat:read chat:write"
+    
+    # Construct the authorization URL
+    auth_url = (
+        f"https://id.twitch.tv/oauth2/authorize"
+        f"?response_type=code"
+        f"&client_id={settings.TWITCH_CLIENT_ID}"
+        f"&redirect_uri={settings.APP_BASE_URL}/auth/twitch/callback"
+        f"&scope={scopes}"
+        f"&state={state}"
+    )
+    return RedirectResponse(url=auth_url)
+
+@router.get("/auth/twitch/callback")
+async def auth_twitch_callback(code: str, state: str, request: Request, db: Session = Depends(get_db)):
+    """
+    Step 2 of OAuth flow: Handle the callback from Twitch after user authorization.
+    """
+    # CSRF Protection: Validate the state
+    if state != request.session.pop('oauth_state', None):
+        raise HTTPException(status_code=403, detail="Invalid state parameter. CSRF attack suspected.")
+
+    # Step 4: Exchange the authorization code for an access token
+    token_url = "https://id.twitch.tv/oauth2/token"
+    token_data = {
+        "client_id": settings.TWITCH_CLIENT_ID,
+        "client_secret": settings.TWITCH_CLIENT_SECRET,
+        "code": code,
+        "grant_type": "authorization_code",
+        "redirect_uri": f"{settings.APP_BASE_URL}/auth/twitch/callback",
+    }
+    
+    async with httpx.AsyncClient() as client:
+        token_response = await client.post(token_url, data=token_data)
+        if token_response.status_code != 200:
+            raise HTTPException(status_code=400, detail="Failed to retrieve access token from Twitch.")
+        
+        token_json = token_response.json()
+        access_token = token_json["access_token"]
+        refresh_token = token_json["refresh_token"]
+
+        # Step 5: Validate the user and get their details from Twitch API
+        users_url = "https://api.twitch.tv/helix/users"
+        headers = {
+            "Client-ID": settings.TWITCH_CLIENT_ID,
+            "Authorization": f"Bearer {access_token}",
+        }
+        user_response = await client.get(users_url, headers=headers)
+        user_data = user_response.json()["data"][0]
+
+    # TODO: Upsert user into our database
+    # For now, we'll just return the user data as a proof of concept.
+    
+    # Encrypt the tokens for storage
+    encrypted_tokens = security.encrypt_tokens(access_token, refresh_token)
+
+    # Here you would create or update the user in your database
+    # user = db.query(models.User).filter_by(platform_user_id=user_data['id']).first() ... etc.
+
+    return {"message": "Twitch login successful!", "user": user_data, "encrypted_tokens": encrypted_tokens}