diff --git a/core/common/mixins.py b/core/common/mixins.py index 8b27da890..33682f6ec 100644 --- a/core/common/mixins.py +++ b/core/common/mixins.py @@ -969,6 +969,11 @@ def save_as_new_version(self, user, **kwargs): # pylint: disable=too-many-branc class ConceptContainerExportMixin: permission_classes = (CanViewConceptDictionaryVersion, IsAuthenticated) + @staticmethod + def is_head_export_permitted(user, version): + """Allow HEAD exports for repo admins, not just staff users.""" + return user.is_staff or user.is_superuser or user.is_admin_for(version) + def get_object(self): queryset = self.get_queryset() if 'version' not in self.kwargs: @@ -988,7 +993,7 @@ def get(self, request, *args, **kwargs): # pylint: disable=unused-argument logger.debug( 'Export requested for %s version %s', self.entity.lower(), version.version ) - if version.is_head and not request.user.is_staff: + if version.is_head and not self.is_head_export_permitted(request.user, version): return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) if version.is_exporting: @@ -1014,7 +1019,7 @@ def get(self, request, *args, **kwargs): # pylint: disable=unused-argument def post(self, request, *args, **kwargs): # pylint: disable=unused-argument version = self.get_object() - if version.is_head and not request.user.is_staff: + if version.is_head and not self.is_head_export_permitted(request.user, version): return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) logger.debug('%s Export requested for version %s (post)', self.entity, version.version) @@ -1040,7 +1045,7 @@ def delete(self, request, *args, **kwargs): # pylint: disable=unused-argument user = request.user version = self.get_object() - if version.is_head and not user.is_staff: + if version.is_head and not self.is_head_export_permitted(user, version): return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) permitted = user.is_staff or user.is_superuser or user.is_admin_for(version) diff --git a/core/integration_tests/tests_collections.py b/core/integration_tests/tests_collections.py index 0d8bbe507..bee2282d5 100644 --- a/core/integration_tests/tests_collections.py +++ b/core/integration_tests/tests_collections.py @@ -3,6 +3,7 @@ from celery_once import AlreadyQueued from mock import patch, Mock, ANY +from mock.mock import PropertyMock from rest_framework.exceptions import ErrorDetail from core.collections.models import CollectionReference, Collection @@ -2614,7 +2615,7 @@ def test_get_204_head(self, s3_exists_mock): response = self.client.get( self.collection.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -2662,7 +2663,7 @@ def test_get_302_head(self, s3_exists_mock, s3_url_for_mock): response = self.client.get( self.collection.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -2679,7 +2680,7 @@ def test_get_500_head_when_signed_url_generation_fails(self, s3_exists_mock, s3_ response = self.client.get( self.collection.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -2689,18 +2690,20 @@ def test_get_500_head_when_signed_url_generation_fails(self, s3_exists_mock, s3_ s3_url_for_mock.assert_called_once_with(f"users/username/username_coll_vHEAD.{self.HEAD_updated_at}.zip") def test_get_405(self): + random_user = UserProfileFactory() response = self.client.get( f'/users/{self.collection.parent.mnemonic}/collections/{self.collection.mnemonic}/{"HEAD"}/export/', - HTTP_AUTHORIZATION='Token ' + self.token, + HTTP_AUTHORIZATION='Token ' + random_user.get_token(), format='json' ) self.assertEqual(response.status_code, 405) def test_post_405(self): + random_user = UserProfileFactory() response = self.client.post( f'/users/{self.collection.parent.mnemonic}/collections/{self.collection.mnemonic}/{"HEAD"}/export/', - HTTP_AUTHORIZATION='Token ' + self.token, + HTTP_AUTHORIZATION='Token ' + random_user.get_token(), format='json' ) @@ -2711,7 +2714,7 @@ def test_post_303_head(self, s3_exists_mock): s3_exists_mock.return_value = True response = self.client.post( self.collection.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -2739,7 +2742,7 @@ def test_post_202_head(self, s3_exists_mock, export_collection_mock): export_collection_mock.__name__ = 'export_collection' response = self.client.post( self.collection.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -2773,7 +2776,7 @@ def test_post_409_head(self, s3_exists_mock, export_collection_mock): export_collection_mock.apply_async.side_effect = AlreadyQueued('already-queued') response = self.client.post( self.collection.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -2802,6 +2805,31 @@ def test_post_409_version(self, s3_has_path_mock, export_collection_mock): self.assertEqual( Task.objects.filter(created_by=self.user, state='PENDING', name='export_collection').count(), 0) + def test_delete_405_head(self): + random_user = UserProfileFactory() + response = self.client.delete( + self.collection.uri + 'HEAD/export/', + HTTP_AUTHORIZATION='Token ' + random_user.get_token(), + format='json' + ) + + self.assertEqual(response.status_code, 405) + + @patch('core.collections.models.Collection.version_export_path', new_callable=PropertyMock) + @patch('core.collections.models.Collection.has_export') + @patch('core.services.storages.cloud.aws.S3.remove') + def test_delete_204_head(self, s3_remove_mock, has_export_mock, export_path_mock): + has_export_mock.return_value = True + export_path_mock.return_value = 'head/export/path' + response = self.client.delete( + self.collection.uri + 'HEAD/export/', + HTTP_AUTHORIZATION='Token ' + self.token, + format='json' + ) + + self.assertEqual(response.status_code, 204) + s3_remove_mock.assert_called_once_with('head/export/path') + class CollectionVersionListViewTest(OCLAPITestCase): def setUp(self): diff --git a/core/integration_tests/tests_sources.py b/core/integration_tests/tests_sources.py index d55edbcda..df58d6d9b 100644 --- a/core/integration_tests/tests_sources.py +++ b/core/integration_tests/tests_sources.py @@ -983,7 +983,7 @@ def test_get_204_head(self, s3_exists_mock): response = self.client.get( self.source.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -1011,7 +1011,7 @@ def test_get_302_head(self, s3_exists_mock, s3_url_for_mock): response = self.client.get( self.source.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -1048,7 +1048,7 @@ def test_get_500_head_when_signed_url_generation_fails(self, s3_exists_mock, s3_ response = self.client.get( self.source.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -1064,7 +1064,7 @@ def test_get_208_HEAD(self, s3_exists_mock, is_exporting_mock): response = self.client.get( self.source.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -1086,18 +1086,20 @@ def test_get_208_version(self, s3_has_path_mock, is_exporting_mock): s3_has_path_mock.assert_not_called() def test_get_405(self): + random_user = UserProfileFactory() response = self.client.get( f'/users/{self.source.parent.mnemonic}/sources/{self.source.mnemonic}/{"HEAD"}/export/', - HTTP_AUTHORIZATION='Token ' + self.token, + HTTP_AUTHORIZATION='Token ' + random_user.get_token(), format='json' ) self.assertEqual(response.status_code, 405) def test_post_405(self): + random_user = UserProfileFactory() response = self.client.post( f'/users/{self.source.parent.mnemonic}/sources/{self.source.mnemonic}/{"HEAD"}/export/', - HTTP_AUTHORIZATION='Token ' + self.token, + HTTP_AUTHORIZATION='Token ' + random_user.get_token(), format='json' ) @@ -1108,7 +1110,7 @@ def test_post_303_head(self, s3_exists_mock): s3_exists_mock.return_value = True response = self.client.post( self.source.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -1136,7 +1138,7 @@ def test_post_202_head(self, s3_exists_mock, export_source_mock): s3_exists_mock.return_value = False response = self.client.post( self.source.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -1181,7 +1183,7 @@ def test_post_409_head(self, s3_exists_mock, export_source_mock): s3_exists_mock.return_value = False response = self.client.post( self.source.uri + 'HEAD/export/', - HTTP_AUTHORIZATION='Token ' + self.admin_token, + HTTP_AUTHORIZATION='Token ' + self.token, format='json' ) @@ -1215,6 +1217,21 @@ def test_delete_405(self): self.assertEqual(response.status_code, 405) + @patch('core.sources.models.Source.version_export_path', new_callable=PropertyMock) + @patch('core.sources.models.Source.has_export') + @patch('core.services.storages.cloud.aws.S3.remove') + def test_delete_204_head(self, s3_remove_mock, has_export_mock, export_path_mock): + has_export_mock.return_value = True + export_path_mock.return_value = 'head/export/path' + response = self.client.delete( + self.source.uri + 'HEAD/export/', + HTTP_AUTHORIZATION='Token ' + self.token, + format='json' + ) + + self.assertEqual(response.status_code, 204) + s3_remove_mock.assert_called_once_with('head/export/path') + def test_delete_403(self): random_user = UserProfileFactory() response = self.client.delete( diff --git a/core/users/models.py b/core/users/models.py index c99e49b77..0faee4079 100644 --- a/core/users/models.py +++ b/core/users/models.py @@ -168,8 +168,11 @@ def set_token(self, token): Token.objects.create(user=self, key=token) def is_admin_for(self, concept_container): # pragma: no cover - parent_id = concept_container.parent_id - return parent_id == self.id or self.organizations.filter(id=parent_id).exists() + user_id = concept_container.user_id + if user_id and user_id == self.id: + return True + organization_id = concept_container.organization_id + return self.organizations.filter(id=organization_id).exists() def __create_token(self): return Token.objects.create(user=self)