110 lines
4.0 KiB
Python
110 lines
4.0 KiB
Python
"""Auth helpers, JWT token creation/verification, and FastAPI dependencies."""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from jose import JWTError, jwt
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import get_session
|
|
from app.user_models import User
|
|
|
|
|
|
class AuthBearer(HTTPBearer):
|
|
"""HTTP Bearer scheme that doesn't auto-fail on missing token."""
|
|
def __init__(self, auto_error: bool = False):
|
|
super().__init__(auto_error=auto_error)
|
|
|
|
|
|
bearer = AuthBearer(auto_error=False)
|
|
|
|
|
|
# ── Token helpers ──────────────────────────────────────────────
|
|
|
|
def create_access_token(user_id: int, is_admin: bool) -> str:
|
|
"""Create a JWT access token for the given user."""
|
|
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.JWT_EXPIRE_MINUTES)
|
|
payload = {
|
|
"sub": str(user_id),
|
|
"is_admin": is_admin,
|
|
"exp": expire,
|
|
}
|
|
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
|
|
|
|
|
|
def decode_token(token: str) -> dict:
|
|
"""Decode and verify a JWT token. Raises HTTPException 401 on failure."""
|
|
try:
|
|
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
|
|
user_id = payload.get("sub")
|
|
if user_id is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
|
|
return payload
|
|
except JWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
|
|
|
|
|
|
# ── FastAPI dependencies ──────────────────────────────────────
|
|
|
|
async def get_current_user(
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(bearer),
|
|
session: AsyncSession = Depends(get_session),
|
|
) -> User:
|
|
"""Validate JWT and return the current User."""
|
|
if credentials is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
|
|
|
|
payload = decode_token(credentials.credentials)
|
|
user_id = int(payload["sub"])
|
|
|
|
result = await session.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
if user is None:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
|
|
return user
|
|
|
|
|
|
async def get_current_admin_user(
|
|
current_user: User = Depends(get_current_user),
|
|
) -> User:
|
|
"""Raise 403 if the current user is not an admin."""
|
|
if not current_user.is_admin:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required")
|
|
return current_user
|
|
|
|
|
|
# ── Google OIDC verification ─────────────────────────────────
|
|
|
|
_GOOGLE_CERTS_URL = "https://www.googleapis.com/oauth2/v3/certs"
|
|
_GOOGLE_TOKEN_INFO_URL = "https://oauth2.googleapis.com/tokeninfo"
|
|
|
|
|
|
async def verify_google_id_token(id_token: str) -> dict:
|
|
"""Verify a Google ID token and return (email, name, picture)."""
|
|
async with httpx.AsyncClient() as client:
|
|
# Use tokeninfo endpoint for simplicity (no signature parsing needed)
|
|
resp = await client.get(_GOOGLE_TOKEN_INFO_URL, params={"idtoken": id_token})
|
|
|
|
if resp.status_code != 200:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Invalid Google ID token",
|
|
)
|
|
|
|
info = resp.json()
|
|
email = info.get("email")
|
|
if not email:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No email in Google token")
|
|
|
|
return {
|
|
"email": email,
|
|
"name": info.get("name", email.split("@")[0]),
|
|
"picture": info.get("picture", ""),
|
|
}
|