89 lines
3.2 KiB
Python
89 lines
3.2 KiB
Python
# auth.py
|
|
from fastapi import APIRouter, Depends
|
|
from fastapi.responses import RedirectResponse
|
|
from sqlalchemy.orm import Session
|
|
import httpx
|
|
from database import get_db, User
|
|
from itsdangerous import URLSafeTimedSerializer
|
|
|
|
# IMPORTANT: These must be replaced with your actual Twitch application credentials
|
|
TWITCH_CLIENT_ID = "YOUR_TWITCH_CLIENT_ID"
|
|
TWITCH_CLIENT_SECRET = "YOUR_TWITCH_CLIENT_SECRET"
|
|
REDIRECT_URI = "http://localhost:8000/auth/twitch/callback"
|
|
|
|
# IMPORTANT: This should be a long, random string kept secret in a production environment
|
|
SECRET_KEY = "YOUR_SECRET_KEY"
|
|
serializer = URLSafeTimedSerializer(SECRET_KEY)
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/login/twitch")
|
|
async def login_with_twitch():
|
|
# Scopes required to get user's email and channel info
|
|
scopes = "user:read:email"
|
|
auth_url = (
|
|
f"https://id.twitch.tv/oauth2/authorize"
|
|
f"?client_id={TWITCH_CLIENT_ID}"
|
|
f"&redirect_uri={REDIRECT_URI}"
|
|
f"&response_type=code"
|
|
f"&scope={scopes}"
|
|
)
|
|
return RedirectResponse(url=auth_url)
|
|
|
|
@router.get("/auth/twitch/callback")
|
|
async def auth_twitch_callback(code: str, db: Session = Depends(get_db)):
|
|
# Exchange the authorization code for an access token
|
|
token_url = "https://id.twitch.tv/oauth2/token"
|
|
token_data = {
|
|
"client_id": TWITCH_CLIENT_ID,
|
|
"client_secret": TWITCH_CLIENT_SECRET,
|
|
"code": code,
|
|
"grant_type": "authorization_code",
|
|
"redirect_uri": REDIRECT_URI,
|
|
}
|
|
async with httpx.AsyncClient() as client:
|
|
token_response = await client.post(token_url, data=token_data)
|
|
token_json = token_response.json()
|
|
access_token = token_json.get("access_token")
|
|
refresh_token = token_json.get("refresh_token")
|
|
|
|
if not access_token:
|
|
return {"error": "Could not fetch access token"}
|
|
|
|
# Get user info from Twitch API
|
|
user_info_url = "https://api.twitch.tv/helix/users"
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Client-Id": TWITCH_CLIENT_ID,
|
|
}
|
|
user_response = await client.get(user_info_url, headers=headers)
|
|
user_data = user_response.json()["data"][0]
|
|
|
|
twitch_id = user_data["id"]
|
|
twitch_username = user_data["login"]
|
|
|
|
# Check if user exists in the database, otherwise create them
|
|
user = db.query(User).filter(User.twitch_id == twitch_id).first()
|
|
if not user:
|
|
user = User(
|
|
twitch_id=twitch_id,
|
|
twitch_username=twitch_username,
|
|
twitch_access_token=access_token, # TODO: Encrypt this
|
|
twitch_refresh_token=refresh_token, # TODO: Encrypt this
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
else:
|
|
# Update tokens for existing user
|
|
user.twitch_access_token = access_token # TODO: Encrypt this
|
|
user.twitch_refresh_token = refresh_token # TODO: Encrypt this
|
|
db.commit()
|
|
|
|
# Create a session cookie
|
|
response = RedirectResponse(url="/dashboard")
|
|
session_data = {"user_id": user.id}
|
|
session_cookie = serializer.dumps(session_data)
|
|
response.set_cookie(key="session", value=session_cookie)
|
|
return response
|