from typing import List from fastapi import Depends, HTTPException from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from jose import jwt, JWTError from common.config.app_settings import app_settings from common.constants.jwt_constants import USER_ROLE_NAMES, USER_PERMISSIONS security = HTTPBearer() class CurrentUser: def __init__(self, user_id: str, user_role_names: List[str], user_permission_keys: List[str]): self.user_id = user_id self.user_role_names = user_role_names self.user_permission_keys = user_permission_keys def has_all_permissions(self, permissions: List[str]) -> bool: """Check if the user has all the specified permissions""" if not permissions: return True return all(p in self.user_permission_keys for p in permissions) def has_any_permissions(self, permissions: List[str]) -> bool: """Check if the user has at least one of the specified permissions""" if not permissions: return True return any(p in self.user_permission_keys for p in permissions) def decode_jwt_token(token: str): payload = jwt.decode( token, app_settings.JWT_SECRET_KEY, algorithms=[app_settings.JWT_ALGORITHM], ) return payload async def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), ) -> CurrentUser: try: payload = decode_jwt_token(credentials.credentials) user = payload.get("subject") if not user or "id" not in user: raise HTTPException(status_code=401, detail="Invalid authentication token") return CurrentUser(user.get("id"), user.get(USER_ROLE_NAMES), user.get(USER_PERMISSIONS)) except JWTError: raise HTTPException(status_code=401, detail="Invalid authentication token") def has_all_permissions( permissions: List[str], ): """Check if the user has all the specified permissions""" def inner_dependency(current_user: CurrentUser = Depends(get_current_user)): if not current_user.has_all_permissions(permissions): raise HTTPException(status_code=403, detail="Not allowed") return True return inner_dependency def has_any_permissions( permissions: List[str] ): """Check if the user has at least one of the specified permissions""" def inner_dependency(current_user: CurrentUser = Depends(get_current_user)): if not current_user.has_any_permissions(permissions): raise HTTPException(status_code=403, detail="Not allowed") return True return inner_dependency