Spaces:
Sleeping
Sleeping
File size: 1,902 Bytes
cfe45d5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | """
Authentication utilities: JWT creation/verification and FastAPI dependency.
"""
from datetime import datetime, timedelta
from typing import Optional
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from app.config import settings
from app.database import get_db
from app.models.user import User
security = HTTPBearer()
def create_access_token(user_id: int) -> str:
"""Create a JWT access token for a user."""
payload = {
"sub": str(user_id),
"exp": datetime.utcnow() + timedelta(hours=settings.JWT_EXPIRATION_HOURS),
"iat": datetime.utcnow(),
}
return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM)
def decode_access_token(token: str) -> Optional[int]:
"""Decode and validate a JWT token. Returns user_id or None."""
try:
payload = jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM])
user_id = int(payload.get("sub", 0))
return user_id if user_id else None
except (jwt.ExpiredSignatureError, jwt.InvalidTokenError, ValueError):
return None
def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: Session = Depends(get_db),
) -> User:
"""FastAPI dependency: extract and validate the current user from JWT."""
user_id = decode_access_token(credentials.credentials)
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
)
user = db.query(User).filter(User.id == user_id).first()
if not user or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found or inactive",
)
return user
|