diff --git a/seahub/api2/endpoints/repo_trash.py b/seahub/api2/endpoints/repo_trash.py index 007b908311..b52d587acb 100644 --- a/seahub/api2/endpoints/repo_trash.py +++ b/seahub/api2/endpoints/repo_trash.py @@ -16,6 +16,8 @@ from seahub.signals import clean_up_repo_trash from seahub.utils.timeutils import timestamp_to_isoformat_timestr from seahub.utils.repo import get_repo_owner from seahub.views import check_folder_permission +from seahub.group.utils import is_group_admin +from seahub.api2.endpoints.group_owned_libraries import get_group_id_by_repo_owner from seaserv import seafile_api from pysearpc import SearpcError @@ -129,7 +131,8 @@ class RepoTrash(APIView): """ Clean library's trash. Permission checking: - 1. only repo owner can perform this action. + 1. repo owner can perform this action. + 2. is group admin. """ # argument check @@ -148,10 +151,20 @@ class RepoTrash(APIView): # permission check username = request.user.username repo_owner = get_repo_owner(request, repo_id) - if username != repo_owner or not config.ENABLE_USER_CLEAN_TRASH: + if not config.ENABLE_USER_CLEAN_TRASH: error_msg = 'Permission denied.' return api_error(status.HTTP_403_FORBIDDEN, error_msg) + if '@seafile_group' in repo_owner: + group_id = get_group_id_by_repo_owner(repo_owner) + if not is_group_admin(group_id, username): + error_msg = 'Permission denied.' + return api_error(status.HTTP_403_FORBIDDEN, error_msg) + else: + if username != repo_owner: + error_msg = 'Permission denied.' + return api_error(status.HTTP_403_FORBIDDEN, error_msg) + try: seafile_api.clean_up_repo_history(repo_id, keep_days) org_id = None if not request.user.org else request.user.org.org_id diff --git a/tests/api/endpoints/test_repo_trash.py b/tests/api/endpoints/test_repo_trash.py index ec2a324506..b172a5a070 100644 --- a/tests/api/endpoints/test_repo_trash.py +++ b/tests/api/endpoints/test_repo_trash.py @@ -3,9 +3,10 @@ import json from django.core.urlresolvers import reverse -from seaserv import seafile_api +from seaserv import seafile_api, ccnet_api from seahub.test_utils import BaseTestCase +from seahub.group.utils import is_group_admin from tests.common.utils import randstring class RepoTrashTest(BaseTestCase): @@ -26,9 +27,13 @@ class RepoTrashTest(BaseTestCase): self.url = reverse('api-v2.1-repo-trash', args=[self.repo_id]) + self.tmp_user = self.create_user( + 'user_%s@test.com' % randstring(4), is_staff=False) + def tearDown(self): self.remove_repo() self.remove_group() + self.remove_user(self.tmp_user.username) def test_can_get(self): @@ -83,6 +88,56 @@ class RepoTrashTest(BaseTestCase): json_resp = json.loads(resp.content) assert len(json_resp['data']) == 0 + def test_can_clean_department_repo_trash(self): + # create a department + group_id = ccnet_api.create_group('department_test', 'system admin', parent_group_id=-1) + seafile_api.set_group_quota(group_id, -2) + repo_id = seafile_api.add_group_owned_repo(group_id, 'dep_test', 'rw') + repo_owner = seafile_api.get_repo_owner(repo_id) + assert '@seafile_group' in repo_owner + group_repos = seafile_api.get_repos_by_group(group_id) + assert len(group_repos) == 1 + group = ccnet_api.get_group(group_id) + + # department add user + ccnet_api.group_add_member(group_id, group.creator_name, self.user_name) + ccnet_api.group_add_member(group_id, group.creator_name, self.tmp_user.username) + ccnet_api.group_set_admin(group_id, self.user_name) + ccnet_api.group_unset_admin(group_id, self.tmp_user.username) + assert is_group_admin(group_id, self.user_name) + assert not is_group_admin(group_id, self.tmp_user.username) + + file_name = 'dep_test.txt' + self.create_file( + repo_id=repo_id, parent_dir='/', filename=file_name, username=self.user_name) + + # delete a file first + seafile_api.del_file(repo_id, '/', file_name, self.user_name) + + # get trash item count + self.login_as(self.user) + resp = self.client.get(reverse('api-v2.1-repo-trash', args=[repo_id])) + json_resp = json.loads(resp.content) + assert len(json_resp['data']) > 0 + + # department member can not clean trash + self.logout() + self.login_as(self.tmp_user) + resp = self.client.delete(self.url) + self.assertEqual(403, resp.status_code) + + # department admin can clean library trash + self.logout() + self.login_as(self.user) + ccnet_api.group_set_admin(group_id, self.user_name) + resp = self.client.delete(self.url) + self.assertEqual(200, resp.status_code) + + # get trash item count again + resp = self.client.get(self.url) + json_resp = json.loads(resp.content) + assert len(json_resp['data']) == 0 + def test_can_not_clean_with_invalid_user_permission(self): self.login_as(self.admin)