37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import Depends
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
|
|
from app.core.constants import TokenType
|
|
from app.core.exceptions import ForbiddenException, UnauthorizedException
|
|
from app.core.security import decode_token
|
|
|
|
bearer_scheme = HTTPBearer()
|
|
|
|
|
|
async def get_current_user_payload(
|
|
credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
|
|
) -> dict:
|
|
payload = decode_token(credentials.credentials)
|
|
if payload is None:
|
|
raise UnauthorizedException("Invalid or expired token")
|
|
if payload.get("type") != TokenType.ACCESS:
|
|
raise UnauthorizedException("Invalid token type")
|
|
return payload
|
|
|
|
|
|
async def get_current_user_id(
|
|
payload: dict = Depends(get_current_user_payload),
|
|
) -> int:
|
|
return int(payload["sub"])
|
|
|
|
|
|
def require_role(*allowed_roles: str):
|
|
async def _check(payload: dict = Depends(get_current_user_payload)) -> dict:
|
|
if payload.get("role") not in allowed_roles:
|
|
raise ForbiddenException("Insufficient permissions")
|
|
return payload
|
|
|
|
return _check
|