Enforce workspace admin role on collaboration mutations

app/projects/api.py

@@ -39,7 +39,8 @@ router = APIRouter(prefix="/v1/projects", tags=["projects"])
 
 @router.post("/workspace/teams", response_model=TeamResponse)
 async def create_team(request: Request, payload: TeamBase) -> TeamResponse:
-    workspace_id, _, _, _ = _identity(request)
+    workspace_id, user_id, _, _ = _identity(request)
+    await _require_workspace_admin(request, workspace_id, user_id)
     return await request.app.state.container.collaboration.create_team(workspace_id, payload.name)
 
 @router.get("/workspace/teams", response_model=list[TeamResponse])
@@ -71,12 +72,14 @@ async def invite_member(request: Request, payload: InvitationCreate) -> Invitati
 
 @router.patch("/workspace/teams/{team_id}", response_model=TeamResponse)
 async def update_team(request: Request, team_id: str, payload: TeamBase) -> TeamResponse:
-    workspace_id, _, _, _ = _identity(request)
+    workspace_id, user_id, _, _ = _identity(request)
+    await _require_workspace_admin(request, workspace_id, user_id)
     return await request.app.state.container.collaboration.update_team(workspace_id, team_id, payload.name)
 
 @router.delete("/workspace/teams/{team_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def archive_team(request: Request, team_id: str) -> Response:
-    workspace_id, _, _, _ = _identity(request)
+    workspace_id, user_id, _, _ = _identity(request)
+    await _require_workspace_admin(request, workspace_id, user_id)
     await request.app.state.container.collaboration.archive_team(workspace_id, team_id)
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
@@ -128,6 +131,17 @@ async def update_member_role(request: Request, user_id: str, new_role: str) -> M
     return MemberResponse(id=membership.id, workspace_id=membership.workspace_id, user_id=membership.user_id, role=membership.role, created_at=membership.created_at.isoformat())
 
 
+async def _require_workspace_admin(request: Request, workspace_id: str, user_id: str) -> None:
+    """Authorize collaboration mutations against the resolved active membership."""
+    membership = await request.app.state.container.collaboration.get_membership(workspace_id, user_id)
+    if (
+        membership is None
+        or membership.status != "active"
+        or membership.role not in {"owner", "admin"}
+    ):
+        raise ForbiddenError
+
+
 def _identity(request: Request) -> tuple[str, str, str, str]:
     context = request.state.auth
     if not context.workspace_id or not context.user_id:
@@ -338,7 +352,8 @@ async def add_project_collaborator(
     user_id: str,
     role: str
 ) -> MemberResponse:
-    workspace_id, _, _, _ = _identity(request)
+    workspace_id, actor_user_id, _, _ = _identity(request)
+    await _require_workspace_admin(request, workspace_id, actor_user_id)
     return await request.app.state.container.collaboration.add_project_collaborator(
         workspace_id, str(project_id), user_id, role
     )
@@ -349,7 +364,8 @@ async def remove_project_collaborator(
     project_id: UUID,
     user_id: str
 ) -> Response:
-    workspace_id, _, _, _ = _identity(request)
+    workspace_id, actor_user_id, _, _ = _identity(request)
+    await _require_workspace_admin(request, workspace_id, actor_user_id)
     await request.app.state.container.collaboration.remove_project_collaborator(workspace_id, str(project_id), user_id)
     return Response(status_code=status.HTTP_204_NO_CONTENT)