Perform org operations only if the user has permissions

Author: chriscarrollsmith

routers/organization.py

@@ -5,8 +5,9 @@
 from sqlmodel import Session, select
 from utils.db import get_session
 from utils.auth import get_authenticated_user
-from utils.models import Organization, User
+from utils.models import Organization, User, Role, UserOrganizationLink, ValidPermissions, RolePermissionLink, Permission
 from datetime import datetime
+from sqlalchemy import and_
 
 logger = getLogger("uvicorn.error")
 
@@ -45,6 +46,14 @@ def __init__(self):
         )
 
 
+class InsufficientPermissionsError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=403,
+            detail="You don't have permission to perform this action"
+        )
+
+
 router = APIRouter(prefix="/organizations", tags=["organizations"])
 
 
@@ -94,6 +103,49 @@ async def as_form(cls, id: int = Form(...), name: str = Form(...)):
 
 # -- Routes --
 
+def check_user_permission(
+    session: Session,
+    user: User,
+    org_id: int,
+    permission: ValidPermissions
+) -> bool:
+    """
+    Check if user has the specified permission for the organization
+    """
+    # Get user's role in the organization
+    user_org = session.exec(
+        select(UserOrganizationLink).where(
+            and_(
+                UserOrganizationLink.user_id == user.id,
+                UserOrganizationLink.organization_id == org_id
+            )
+        )
+    ).first()
+
+    if not user_org:
+        return False
+
+    # Get permission ID
+    permission_record = session.exec(
+        select(Permission).where(Permission.name == permission)
+    ).first()
+
+    if not permission_record:
+        return False
+
+    # Check if role has the permission
+    role_permission = session.exec(
+        select(RolePermissionLink).where(
+            and_(
+                RolePermissionLink.role_id == user_org.role_id,
+                RolePermissionLink.permission_id == permission_record.id
+            )
+        )
+    ).first()
+
+    return bool(role_permission)
+
+
 @router.post("/", response_class=RedirectResponse)
 def create_organization(
     org: OrganizationCreate = Depends(OrganizationCreate.as_form),
@@ -110,11 +162,54 @@ def create_organization(
     session.commit()
     session.refresh(db_org)
 
+    owner_role = session.exec(
+        select(Role).where(
+            and_(
+                Role.organization_id == db_org.id,
+                Role.name == "Owner"
+            )
+        )
+    ).first()
+
+    if not owner_role:
+        owner_role = Role(
+            name="Owner",
+            organization_id=db_org.id
+        )
+        session.add(owner_role)
+        session.commit()
+        session.refresh(owner_role)
+
+    user_org_link = UserOrganizationLink(
+        user_id=user.id,
+        organization_id=db_org.id,
+        role_id=owner_role.id
+    )
+    session.add(user_org_link)
+    session.commit()
+
     return RedirectResponse(url=f"/organizations/{db_org.id}", status_code=303)
 
 
 @router.get("/{org_id}", response_model=OrganizationRead)
-def read_organization(org_id: int, user: User = Depends(get_authenticated_user), session: Session = Depends(get_session)):
+def read_organization(
+    org_id: int,
+    user: User = Depends(get_authenticated_user),
+    session: Session = Depends(get_session)
+):
+    # First check if user is a member of the organization
+    user_org = session.exec(
+        select(UserOrganizationLink).where(
+            and_(
+                UserOrganizationLink.user_id == user.id,
+                UserOrganizationLink.organization_id == org_id
+            )
+        )
+    ).first()
+
+    if not user_org:
+        raise InsufficientPermissionsError()
+
     db_org = session.get(Organization, org_id)
     if not db_org:
         raise OrganizationNotFoundError()
@@ -127,6 +222,9 @@ def update_organization(
     user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
+    if not check_user_permission(session, user, org.id, ValidPermissions.EDIT_ORGANIZATION):
+        raise InsufficientPermissionsError()
+
     db_org = session.get(Organization, org.id)
     if not db_org:
         raise OrganizationNotFoundError()
@@ -155,6 +253,9 @@ def delete_organization(
     user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
+    if not check_user_permission(session, user, org_id, ValidPermissions.DELETE_ORGANIZATION):
+        raise InsufficientPermissionsError()
+
     db_org = session.get(Organization, org_id)
     if not db_org:
         raise OrganizationNotFoundError()