114 lines
4.5 KiB
Python
114 lines
4.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, Header
|
|
from sqlalchemy.orm import Session
|
|
from backend.database import get_db
|
|
from backend.models import User, UserRole
|
|
from backend.schemas import UserOut, GoogleAuthRequest, DemoLoginRequest
|
|
from backend.config import create_access_token, decode_access_token, GOOGLE_CLIENT_ID
|
|
from typing import Optional
|
|
from google.oauth2 import id_token
|
|
from google.auth.transport import requests as google_requests
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
def get_current_user(authorization: Optional[str] = Header(None), db: Session = Depends(get_db)) -> User:
|
|
if not authorization:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing authorization header")
|
|
|
|
parts = authorization.split()
|
|
if len(parts) != 2 or parts[0].lower() != "bearer":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authorization header format")
|
|
|
|
token = parts[1]
|
|
payload = decode_access_token(token)
|
|
if not payload or "sub" not in payload:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
|
|
|
|
user_id = int(payload["sub"])
|
|
user = db.query(User).filter(User.id == user_id, User.is_active == True).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or inactive")
|
|
|
|
return user
|
|
|
|
def get_admin_user(current_user: User = Depends(get_current_user)) -> User:
|
|
if current_user.role != UserRole.ADMIN.value:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin privileges required")
|
|
return current_user
|
|
|
|
@router.post("/google")
|
|
def google_auth(req: GoogleAuthRequest, db: Session = Depends(get_db)):
|
|
email = req.email
|
|
name = req.name or "Google User"
|
|
avatar_url = req.avatar_url or f"https://api.dicebear.com/7.x/bottts/svg?seed={email}"
|
|
google_sub = None
|
|
|
|
# Try verifying real Google token if client ID configured and credential provided
|
|
if req.credential and len(req.credential) > 50:
|
|
try:
|
|
if GOOGLE_CLIENT_ID:
|
|
id_info = id_token.verify_oauth2_token(req.credential, google_requests.Request(), GOOGLE_CLIENT_ID)
|
|
else:
|
|
id_info = id_token.verify_oauth2_token(req.credential, google_requests.Request())
|
|
email = id_info.get("email", email)
|
|
name = id_info.get("name", name)
|
|
avatar_url = id_info.get("picture", avatar_url)
|
|
google_sub = id_info.get("sub")
|
|
except Exception as e:
|
|
# Fallback to provided details if token decode fails in dev
|
|
if not email:
|
|
raise HTTPException(status_code=400, detail=f"Google token verification failed: {str(e)}")
|
|
|
|
if not email:
|
|
raise HTTPException(status_code=400, detail="Email is required")
|
|
|
|
user = db.query(User).filter(User.email == email).first()
|
|
|
|
# Check if this is the first user ever registered
|
|
total_users = db.query(User).count()
|
|
initial_role = UserRole.ADMIN.value if total_users == 0 else UserRole.REGULAR.value
|
|
|
|
if not user:
|
|
user = User(
|
|
email=email,
|
|
name=name,
|
|
avatar_url=avatar_url,
|
|
google_id=google_sub,
|
|
role=initial_role,
|
|
weekly_star_quota=15
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
else:
|
|
# Update existing user google_id or default avatar if missing
|
|
if google_sub and not user.google_id:
|
|
user.google_id = google_sub
|
|
if avatar_url and not user.avatar_url:
|
|
user.avatar_url = avatar_url
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
token = create_access_token({"sub": str(user.id), "email": user.email, "role": user.role})
|
|
return {
|
|
"access_token": token,
|
|
"token_type": "bearer",
|
|
"user": UserOut.model_validate(user)
|
|
}
|
|
|
|
@router.post("/demo-login")
|
|
def demo_login(req: DemoLoginRequest, db: Session = Depends(get_db)):
|
|
user = db.query(User).filter(User.email == req.email).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="Demo user not found. Run backend initialization first.")
|
|
|
|
token = create_access_token({"sub": str(user.id), "email": user.email, "role": user.role})
|
|
return {
|
|
"access_token": token,
|
|
"token_type": "bearer",
|
|
"user": UserOut.model_validate(user)
|
|
}
|
|
|
|
@router.get("/me", response_model=UserOut)
|
|
def get_me(current_user: User = Depends(get_current_user)):
|
|
return current_user
|