127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""FastAPI dependencies: pagination, current user extraction, RBAC role enforcement."""
|
|
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from fastapi import Depends, HTTPException, Query, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.models.user import User, UserRole
|
|
from app.services.auth_service import get_user_by_id
|
|
from app.utils.jwt import verify_access_token
|
|
|
|
security = HTTPBearer(auto_error=False)
|
|
|
|
|
|
def get_pagination(
|
|
page: int = Query(1, ge=1, description="Page number (1-indexed)"),
|
|
page_size: int = Query(20, ge=1, le=100, description="Items per page"),
|
|
) -> dict:
|
|
"""Common pagination parameters."""
|
|
return {"page": page, "page_size": page_size}
|
|
|
|
|
|
async def get_current_user(
|
|
credentials: Optional[HTTPAuthorizationCredentials] = Depends(security),
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> User:
|
|
"""Extract and verify the JWT bearer token, returning the current user.
|
|
|
|
Raises 401 if token is missing, invalid, or the user is not found / inactive.
|
|
"""
|
|
if credentials is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"error": {
|
|
"code": "UNAUTHORIZED",
|
|
"message": "Missing authentication token",
|
|
}
|
|
},
|
|
)
|
|
|
|
token = credentials.credentials
|
|
payload = verify_access_token(token)
|
|
if payload is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"error": {
|
|
"code": "INVALID_TOKEN",
|
|
"message": "Invalid or expired access token",
|
|
}
|
|
},
|
|
)
|
|
|
|
user_id_str = payload.get("sub")
|
|
if not user_id_str:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"error": {"code": "INVALID_TOKEN", "message": "Token missing subject"}
|
|
},
|
|
)
|
|
|
|
try:
|
|
user_uuid = uuid.UUID(user_id_str)
|
|
except (ValueError, TypeError):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"error": {
|
|
"code": "INVALID_TOKEN",
|
|
"message": "Invalid user ID in token",
|
|
}
|
|
},
|
|
)
|
|
|
|
user = await get_user_by_id(db, user_uuid)
|
|
if user is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={"error": {"code": "USER_NOT_FOUND", "message": "User not found"}},
|
|
)
|
|
|
|
if not user.is_active:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail={
|
|
"error": {
|
|
"code": "ACCOUNT_DISABLED",
|
|
"message": "Account is deactivated",
|
|
}
|
|
},
|
|
)
|
|
|
|
return user
|
|
|
|
|
|
def require_role(allowed_roles: list[str]):
|
|
"""RBAC dependency factory: enforce that the current user has one of the allowed roles.
|
|
|
|
Usage:
|
|
@router.get("/", dependencies=[Depends(require_role(["admin"]))])
|
|
or:
|
|
async def handler(user: User = Depends(require_role(["admin"]))):
|
|
"""
|
|
|
|
async def role_checker(user: User = Depends(get_current_user)) -> User:
|
|
user_role = (
|
|
user.role.value if isinstance(user.role, UserRole) else str(user.role)
|
|
)
|
|
if user_role not in allowed_roles:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail={
|
|
"error": {
|
|
"code": "FORBIDDEN",
|
|
"message": f"Role '{user_role}' is not permitted. Required: {allowed_roles}",
|
|
}
|
|
},
|
|
)
|
|
return user
|
|
|
|
return role_checker
|