fix(permissions): restrict deletes and admin member management
This commit is contained in:
@@ -6,6 +6,7 @@ from apps.workspaces.services import (
|
|||||||
CLIENTS_DELETE,
|
CLIENTS_DELETE,
|
||||||
CLIENTS_EDIT,
|
CLIENTS_EDIT,
|
||||||
CLIENTS_VIEW,
|
CLIENTS_VIEW,
|
||||||
|
can_delete_workspace_object,
|
||||||
has_workspace_capability,
|
has_workspace_capability,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -43,4 +44,6 @@ class IsClientWorkspaceMember(permissions.BasePermission):
|
|||||||
"partial_update": CLIENTS_EDIT,
|
"partial_update": CLIENTS_EDIT,
|
||||||
"destroy": CLIENTS_DELETE,
|
"destroy": CLIENTS_DELETE,
|
||||||
}.get(view.action, CLIENTS_VIEW)
|
}.get(view.action, CLIENTS_VIEW)
|
||||||
|
if view.action == "destroy":
|
||||||
|
return can_delete_workspace_object(request.user, obj, CLIENTS_DELETE)
|
||||||
return has_workspace_capability(request.user, obj.workspace, capability)
|
return has_workspace_capability(request.user, obj.workspace, capability)
|
||||||
|
|||||||
@@ -19,11 +19,13 @@ def create_client(user, workspace_id, name, notes=""):
|
|||||||
if Client.objects.filter(workspace_id=workspace_id, name=name, is_deleted=False).exists():
|
if Client.objects.filter(workspace_id=workspace_id, name=name, is_deleted=False).exists():
|
||||||
raise ValidationError({"name": "مشتری با این نام در این فضای کاری وجود دارد."})
|
raise ValidationError({"name": "مشتری با این نام در این فضای کاری وجود دارد."})
|
||||||
|
|
||||||
return Client.objects.create(
|
return Client.objects.create(
|
||||||
workspace_id=workspace_id,
|
workspace_id=workspace_id,
|
||||||
name=name,
|
name=name,
|
||||||
notes=notes
|
notes=notes,
|
||||||
)
|
created_by=user,
|
||||||
|
updated_by=user,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def update_client(client, name=None, notes=None):
|
def update_client(client, name=None, notes=None):
|
||||||
|
|||||||
@@ -37,10 +37,12 @@ from apps.projects.services.memberships import add_project_member, update_projec
|
|||||||
from apps.workspaces.services import (
|
from apps.workspaces.services import (
|
||||||
PROJECTS_ARCHIVE,
|
PROJECTS_ARCHIVE,
|
||||||
PROJECTS_CREATE,
|
PROJECTS_CREATE,
|
||||||
|
PROJECTS_DELETE,
|
||||||
PROJECTS_EDIT,
|
PROJECTS_EDIT,
|
||||||
PROJECT_MEMBERS_ADD,
|
PROJECT_MEMBERS_ADD,
|
||||||
PROJECT_MEMBERS_CHANGE_ROLE,
|
PROJECT_MEMBERS_CHANGE_ROLE,
|
||||||
PROJECT_MEMBERS_REMOVE,
|
PROJECT_MEMBERS_REMOVE,
|
||||||
|
can_delete_workspace_object,
|
||||||
has_project_capability,
|
has_project_capability,
|
||||||
has_workspace_capability,
|
has_workspace_capability,
|
||||||
)
|
)
|
||||||
@@ -78,14 +80,14 @@ class ProjectViewSet(ModelViewSet):
|
|||||||
"""
|
"""
|
||||||
Returns active projects where the current user is an active member.
|
Returns active projects where the current user is an active member.
|
||||||
"""
|
"""
|
||||||
if getattr(self, "swagger_fake_view", False) or not self.request.user.is_authenticated:
|
if getattr(self, "swagger_fake_view", False) or not self.request.user.is_authenticated:
|
||||||
return Project.objects.none()
|
return Project.objects.none()
|
||||||
|
|
||||||
return Project.objects.filter(
|
return Project.objects.filter(
|
||||||
memberships__user=self.request.user,
|
workspace__memberships__user=self.request.user,
|
||||||
memberships__is_active=True,
|
workspace__memberships__is_active=True,
|
||||||
is_deleted=False
|
is_deleted=False
|
||||||
).distinct()
|
).distinct()
|
||||||
|
|
||||||
def get_serializer_class(self):
|
def get_serializer_class(self):
|
||||||
"""
|
"""
|
||||||
@@ -216,14 +218,19 @@ class ProjectViewSet(ModelViewSet):
|
|||||||
output_serializer = ProjectSerializer(updated_project)
|
output_serializer = ProjectSerializer(updated_project)
|
||||||
return Response(output_serializer.data, status=status.HTTP_200_OK)
|
return Response(output_serializer.data, status=status.HTTP_200_OK)
|
||||||
|
|
||||||
def destroy(self, request, *args, **kwargs):
|
def destroy(self, request, *args, **kwargs):
|
||||||
"""
|
"""
|
||||||
Soft deletes a project.
|
Soft deletes a project.
|
||||||
"""
|
"""
|
||||||
project = self.get_object()
|
project = self.get_object()
|
||||||
project.is_deleted = True
|
if not can_delete_workspace_object(request.user, project, PROJECTS_DELETE):
|
||||||
project.save(update_fields=["is_deleted", "updated_at"])
|
return Response(
|
||||||
return Response(status=status.HTTP_204_NO_CONTENT)
|
{"detail": "You do not have permission to delete this project."},
|
||||||
|
status=status.HTTP_403_FORBIDDEN,
|
||||||
|
)
|
||||||
|
project.is_deleted = True
|
||||||
|
project.save(update_fields=["is_deleted", "updated_at"])
|
||||||
|
return Response(status=status.HTTP_204_NO_CONTENT)
|
||||||
|
|
||||||
@action(detail=True, methods=["post"])
|
@action(detail=True, methods=["post"])
|
||||||
def archive(self, request, pk=None):
|
def archive(self, request, pk=None):
|
||||||
|
|||||||
@@ -26,13 +26,15 @@ def create_project(user, workspace, name, client=None, description="", color="")
|
|||||||
if Project.objects.filter(workspace=workspace, name=name, is_deleted=False).exists():
|
if Project.objects.filter(workspace=workspace, name=name, is_deleted=False).exists():
|
||||||
raise ValidationError({"name": "A project with this name already exists in the workspace."})
|
raise ValidationError({"name": "A project with this name already exists in the workspace."})
|
||||||
|
|
||||||
project = Project.objects.create(
|
project = Project.objects.create(
|
||||||
workspace=workspace,
|
workspace=workspace,
|
||||||
name=name,
|
name=name,
|
||||||
client=client,
|
client=client,
|
||||||
description=description,
|
description=description,
|
||||||
color=color
|
color=color,
|
||||||
)
|
created_by=user,
|
||||||
|
updated_by=user,
|
||||||
|
)
|
||||||
|
|
||||||
ProjectMembership.objects.create(
|
ProjectMembership.objects.create(
|
||||||
project=project,
|
project=project,
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ from apps.workspaces.services import (
|
|||||||
TAGS_DELETE,
|
TAGS_DELETE,
|
||||||
TAGS_EDIT,
|
TAGS_EDIT,
|
||||||
TAGS_VIEW,
|
TAGS_VIEW,
|
||||||
|
can_delete_workspace_object,
|
||||||
has_workspace_capability,
|
has_workspace_capability,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -38,4 +39,6 @@ class IsTagWorkspaceAllowed(permissions.BasePermission):
|
|||||||
"partial_update": TAGS_EDIT,
|
"partial_update": TAGS_EDIT,
|
||||||
"destroy": TAGS_DELETE,
|
"destroy": TAGS_DELETE,
|
||||||
}.get(view.action, TAGS_VIEW)
|
}.get(view.action, TAGS_VIEW)
|
||||||
|
if view.action == "destroy":
|
||||||
|
return can_delete_workspace_object(request.user, obj, TAGS_DELETE)
|
||||||
return has_workspace_capability(request.user, obj.workspace, capability)
|
return has_workspace_capability(request.user, obj.workspace, capability)
|
||||||
|
|||||||
@@ -22,11 +22,13 @@ def create_tag(user, workspace_id, name, color=""):
|
|||||||
if Tag.objects.filter(workspace_id=workspace_id, name=name, is_deleted=False).exists():
|
if Tag.objects.filter(workspace_id=workspace_id, name=name, is_deleted=False).exists():
|
||||||
raise ValidationError({"name": "A tag with this name already exists in the workspace."})
|
raise ValidationError({"name": "A tag with this name already exists in the workspace."})
|
||||||
|
|
||||||
return Tag.objects.create(
|
return Tag.objects.create(
|
||||||
workspace_id=workspace_id,
|
workspace_id=workspace_id,
|
||||||
name=name,
|
name=name,
|
||||||
color=color
|
color=color,
|
||||||
)
|
created_by=user,
|
||||||
|
updated_by=user,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def update_tag(tag, **kwargs):
|
def update_tag(tag, **kwargs):
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ from apps.workspaces.services.permissions import (
|
|||||||
WORKSPACE_VIEW,
|
WORKSPACE_VIEW,
|
||||||
can_assign_workspace_role,
|
can_assign_workspace_role,
|
||||||
can_change_workspace_membership,
|
can_change_workspace_membership,
|
||||||
|
can_delete_workspace_object,
|
||||||
can_manage_workspace_members,
|
can_manage_workspace_members,
|
||||||
get_workspace_membership,
|
get_workspace_membership,
|
||||||
get_workspace_role,
|
get_workspace_role,
|
||||||
@@ -72,6 +73,7 @@ __all__ = [
|
|||||||
"can_manage_workspace_members",
|
"can_manage_workspace_members",
|
||||||
"can_assign_workspace_role",
|
"can_assign_workspace_role",
|
||||||
"can_change_workspace_membership",
|
"can_change_workspace_membership",
|
||||||
|
"can_delete_workspace_object",
|
||||||
"upsert_workspace_user_rate",
|
"upsert_workspace_user_rate",
|
||||||
"update_workspace_user_rate",
|
"update_workspace_user_rate",
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -166,6 +166,21 @@ def has_project_capability(user, project, capability: str) -> bool:
|
|||||||
return is_project_manager and capability in PROJECT_MANAGER_CAPABILITIES
|
return is_project_manager and capability in PROJECT_MANAGER_CAPABILITIES
|
||||||
|
|
||||||
|
|
||||||
|
def can_delete_workspace_object(user, obj, capability: str) -> bool:
|
||||||
|
workspace = getattr(obj, "workspace", None)
|
||||||
|
if workspace is None:
|
||||||
|
return False
|
||||||
|
|
||||||
|
if not has_workspace_capability(user, workspace, capability):
|
||||||
|
return False
|
||||||
|
|
||||||
|
actor_role = get_workspace_role(user, workspace)
|
||||||
|
if actor_role == WorkspaceMembership.Role.OWNER:
|
||||||
|
return True
|
||||||
|
|
||||||
|
return getattr(obj, "created_by_id", None) == getattr(user, "id", None)
|
||||||
|
|
||||||
|
|
||||||
def can_manage_workspace_members(user, workspace: Workspace) -> bool:
|
def can_manage_workspace_members(user, workspace: Workspace) -> bool:
|
||||||
return has_workspace_capability(user, workspace, WORKSPACE_MEMBERS_CHANGE_ROLE)
|
return has_workspace_capability(user, workspace, WORKSPACE_MEMBERS_CHANGE_ROLE)
|
||||||
|
|
||||||
@@ -175,7 +190,10 @@ def can_assign_workspace_role(user, workspace: Workspace, role: str) -> bool:
|
|||||||
if actor_role == WorkspaceMembership.Role.OWNER:
|
if actor_role == WorkspaceMembership.Role.OWNER:
|
||||||
return True
|
return True
|
||||||
if actor_role == WorkspaceMembership.Role.ADMIN:
|
if actor_role == WorkspaceMembership.Role.ADMIN:
|
||||||
return role != WorkspaceMembership.Role.OWNER
|
return role not in {
|
||||||
|
WorkspaceMembership.Role.OWNER,
|
||||||
|
WorkspaceMembership.Role.ADMIN,
|
||||||
|
}
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
@@ -193,11 +211,15 @@ def can_change_workspace_membership(user, membership: WorkspaceMembership, *, ne
|
|||||||
|
|
||||||
target_is_canonical_owner = workspace.owner_id == membership.user_id
|
target_is_canonical_owner = workspace.owner_id == membership.user_id
|
||||||
target_is_owner_role = membership.role == WorkspaceMembership.Role.OWNER
|
target_is_owner_role = membership.role == WorkspaceMembership.Role.OWNER
|
||||||
|
target_is_admin_role = membership.role == WorkspaceMembership.Role.ADMIN
|
||||||
|
|
||||||
if actor_role == WorkspaceMembership.Role.ADMIN:
|
if actor_role == WorkspaceMembership.Role.ADMIN:
|
||||||
if target_is_owner_role or target_is_canonical_owner:
|
if target_is_owner_role or target_is_admin_role or target_is_canonical_owner:
|
||||||
return False
|
return False
|
||||||
if new_role == WorkspaceMembership.Role.OWNER:
|
if new_role in {
|
||||||
|
WorkspaceMembership.Role.OWNER,
|
||||||
|
WorkspaceMembership.Role.ADMIN,
|
||||||
|
}:
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|||||||
@@ -215,7 +215,7 @@ def test_guest_is_read_only_for_workspace_resources(api_client, owner, guest, wo
|
|||||||
assert list_projects_response.status_code == 200
|
assert list_projects_response.status_code == 200
|
||||||
assert create_tag_response.status_code == 403
|
assert create_tag_response.status_code == 403
|
||||||
assert create_entry_response.status_code == 403
|
assert create_entry_response.status_code == 403
|
||||||
assert edit_project_response.status_code == 404
|
assert edit_project_response.status_code == 403
|
||||||
|
|
||||||
|
|
||||||
def test_member_project_manager_cannot_edit_project(api_client, member, project):
|
def test_member_project_manager_cannot_edit_project(api_client, member, project):
|
||||||
@@ -256,3 +256,80 @@ def test_admin_cannot_change_owner_membership_but_canonical_owner_can(
|
|||||||
|
|
||||||
assert admin_response.status_code == 403
|
assert admin_response.status_code == 403
|
||||||
assert owner_response.status_code == 200
|
assert owner_response.status_code == 200
|
||||||
|
|
||||||
|
|
||||||
|
def test_admin_cannot_add_or_change_admin_memberships(api_client, owner, admin, member, workspace):
|
||||||
|
admin_membership = WorkspaceMembership.objects.get(workspace=workspace, user=admin, is_deleted=False)
|
||||||
|
|
||||||
|
api_client.force_authenticate(user=admin)
|
||||||
|
create_response = api_client.post(
|
||||||
|
"/api/workspace-memberships/",
|
||||||
|
{
|
||||||
|
"workspace": str(workspace.id),
|
||||||
|
"user": str(member.id),
|
||||||
|
"role": WorkspaceMembership.Role.ADMIN,
|
||||||
|
},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
update_response = api_client.patch(
|
||||||
|
f"/api/workspace-memberships/{admin_membership.id}/",
|
||||||
|
{"role": WorkspaceMembership.Role.MEMBER},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
delete_response = api_client.delete(f"/api/workspace-memberships/{admin_membership.id}/")
|
||||||
|
|
||||||
|
assert create_response.status_code == 403
|
||||||
|
assert update_response.status_code == 403
|
||||||
|
assert delete_response.status_code == 403
|
||||||
|
|
||||||
|
|
||||||
|
def test_admin_can_delete_only_owned_clients_tags_and_projects(api_client, owner, admin, workspace):
|
||||||
|
api_client.force_authenticate(user=owner)
|
||||||
|
owner_client_response = api_client.post(
|
||||||
|
"/api/clients/",
|
||||||
|
{"workspace_id": str(workspace.id), "name": "Owner Client", "notes": ""},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
owner_tag_response = api_client.post(
|
||||||
|
"/api/tags/",
|
||||||
|
{"workspace_id": str(workspace.id), "name": "Owner Tag", "color": "#123456"},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
owner_project_response = api_client.post(
|
||||||
|
"/api/projects/",
|
||||||
|
{"workspace": str(workspace.id), "name": "Owner Project", "description": "", "client": None},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
|
||||||
|
api_client.force_authenticate(user=admin)
|
||||||
|
admin_client_response = api_client.post(
|
||||||
|
"/api/clients/",
|
||||||
|
{"workspace_id": str(workspace.id), "name": "Admin Client", "notes": ""},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
admin_tag_response = api_client.post(
|
||||||
|
"/api/tags/",
|
||||||
|
{"workspace_id": str(workspace.id), "name": "Admin Tag", "color": "#654321"},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
admin_project_response = api_client.post(
|
||||||
|
"/api/projects/",
|
||||||
|
{"workspace": str(workspace.id), "name": "Admin Project", "description": "", "client": None},
|
||||||
|
format="json",
|
||||||
|
)
|
||||||
|
|
||||||
|
delete_owner_client = api_client.delete(f"/api/clients/{owner_client_response.data['id']}/")
|
||||||
|
delete_owner_tag = api_client.delete(f"/api/tags/{owner_tag_response.data['id']}/")
|
||||||
|
delete_owner_project = api_client.delete(f"/api/projects/{owner_project_response.data['id']}/")
|
||||||
|
|
||||||
|
delete_admin_client = api_client.delete(f"/api/clients/{admin_client_response.data['id']}/")
|
||||||
|
delete_admin_tag = api_client.delete(f"/api/tags/{admin_tag_response.data['id']}/")
|
||||||
|
delete_admin_project = api_client.delete(f"/api/projects/{admin_project_response.data['id']}/")
|
||||||
|
|
||||||
|
assert delete_owner_client.status_code == 403
|
||||||
|
assert delete_owner_tag.status_code == 403
|
||||||
|
assert delete_owner_project.status_code in {403, 404}
|
||||||
|
|
||||||
|
assert delete_admin_client.status_code == 204
|
||||||
|
assert delete_admin_tag.status_code == 204
|
||||||
|
assert delete_admin_project.status_code == 204
|
||||||
|
|||||||
Reference in New Issue
Block a user