Spaces:
Runtime error
Runtime error
File size: 3,919 Bytes
3674b4b | 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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | from fastapi import APIRouter, HTTPException, status, Request
from backend.core.dependencies import DbSession, CurrentUser
from backend.schemas.auth import LoginRequest, Token, RefreshRequest
from backend.schemas.user import UserResponse
from backend.services.auth import authenticate_user, create_token_for_user
from backend.core.security import verify_refresh_token
from backend.models import User, AuditLog
router = APIRouter()
@router.post("/login", response_model=Token)
def login(request: LoginRequest, db: DbSession, req: Request):
user = authenticate_user(db, request.email, request.password)
if not user:
# Log failed login attempt
log = AuditLog(
action="login_failed",
details={"email": request.email},
ip_address=req.client.host if req.client else None
)
db.add(log)
db.commit()
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User account is disabled",
)
# Log successful login
log = AuditLog(
user_id=user.id,
action="login_success",
details={"email": user.email},
ip_address=req.client.host if req.client else None
)
db.add(log)
db.commit()
return create_token_for_user(db, user)
@router.get("/me", response_model=UserResponse)
def get_current_user_info(current_user: CurrentUser, db: DbSession):
# Get group names for response
group_names = [ug.group.name for ug in current_user.groups]
return UserResponse(
id=current_user.id,
email=current_user.email,
full_name=current_user.full_name,
is_admin=current_user.is_admin,
is_active=current_user.is_active,
created_at=current_user.created_at,
last_login=current_user.last_login,
groups=group_names
)
@router.post("/logout")
def logout(current_user: CurrentUser, db: DbSession):
"""Logout user by invalidating all their tokens."""
current_user.token_version += 1
db.commit()
return {"message": "Successfully logged out"}
@router.post("/refresh", response_model=Token)
def refresh_token(request: RefreshRequest, db: DbSession):
"""Exchange a valid refresh token for new access + refresh tokens."""
payload = verify_refresh_token(request.refresh_token)
if payload is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired refresh token",
headers={"WWW-Authenticate": "Bearer"},
)
user_id = payload.get("sub")
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token payload",
)
user = db.query(User).filter(User.id == user_id).first()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found",
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User account is disabled",
)
# Check token version - ensures refresh tokens are invalidated on password change
token_version = payload.get("token_version")
if token_version is None or token_version != user.token_version:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has been revoked",
headers={"WWW-Authenticate": "Bearer"},
)
# Generate new tokens
return create_token_for_user(db, user)
|