from dataclasses import dataclass from enum import Enum from typing import Any, List, Union from fastapi import Request, HTTPException, status Allow: str = "allow" Deny: str = "deny" @dataclass(frozen=True) class Principal: key: str value: str def __repr__(self) -> str: return f"{self.key}:{self.value}" def __str__(self) -> str: return self.__repr__() @dataclass(frozen=True) class SystemPrincipal(Principal): def __init__(self, value: str): super().__init__(key="system", value=value) @dataclass(frozen=True) class RolePrincipal(Principal): def __init__(self, value: str): super().__init__(key="role", value=value) Everyone = SystemPrincipal(value="everyone") Authenticated = SystemPrincipal(value="authenticated") class OHPermission(Enum): CREATE = "create" READ = "read" EDIT = "edit" DELETE = "delete" class AccessControl: def __init__(self, permission_exception: Any = None): self.permission_exception = permission_exception or HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Permission denied" ) def _acl(self, resource): acl = getattr(resource, "__acl__", []) if callable(acl): return acl() return acl def has_permission( self, principals: List[Principal], required_permissions: List[OHPermission], resource: Any ) -> bool: if not isinstance(resource, list): resource = [resource] permits = [] for res in resource: granted = False acl = self._acl(res) for action, principal, permissions in acl: # Check if any of the required permissions are in the allowed permissions for this principal is_permitted = any(p in permissions for p in required_permissions) if (action == Allow and is_permitted) and ( principal in principals or principal == Everyone ): granted = True break elif (action == Deny and is_permitted) and ( principal in principals or principal == Everyone ): granted = False break permits.append(granted) return all(permits) def assert_access( self, principals: List[Principal], required_permissions: List[OHPermission], resource: Any ): if not self.has_permission(principals, required_permissions, resource): raise self.permission_exception # Roles allowed to access the backend (whitelist). # Management is intentionally excluded. ALLOWED_ROLES = {"Admin", "Super Admin", "Engineer", "Application Administrator"} def require_any_role(*allowed_roles: str): """ FastAPI dependency — whitelist-based role check. Raises 403 if the authenticated user's role is not in the allowed set. Comparison is case-insensitive. Usage (router-level): router = APIRouter(dependencies=[Depends(require_any_role(*ALLOWED_ROLES))]) """ normalized = {r.lower() for r in allowed_roles} async def dependency(request: Request): user = getattr(request.state, "user", None) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated" ) user_role = user.get("role") if isinstance(user, dict) else getattr(user, "role", None) if not user_role or user_role.lower() not in normalized: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Your role does not have permission to access this resource.", ) return True return dependency def required_permission(permissions: Union[OHPermission, List[OHPermission]], resources: Any): """ FastAPI dependency for role-based access control. """ if isinstance(permissions, OHPermission): permissions = [permissions] async def dependency(request: Request): user = getattr(request.state, "user", None) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated" ) user_role = None if isinstance(user, dict): user_role = user.get("role") else: user_role = getattr(user, "role", None) if not user_role: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="No user role found" ) user_principals = [Authenticated, RolePrincipal(user_role)] ac = AccessControl() ac.assert_access(user_principals, permissions, resources) return True return dependency