feat: Implement role-based access control for OverhaulScope and integrate it into API endpoints.
parent
a728d39fce
commit
c5f7384b2d
@ -0,0 +1,122 @@
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
Loading…
Reference in New Issue