diff --git a/core/collections/tests/tests.py b/core/collections/tests/tests.py index a19f06d4..ef33b928 100644 --- a/core/collections/tests/tests.py +++ b/core/collections/tests/tests.py @@ -5534,6 +5534,45 @@ def test_verify_scope_no_kwargs_non_get_raises_404(self): response = self.client.post('/collections/', {}, format='json') self.assertEqual(response.status_code, 404) + def test_create_rejects_conflicting_body_owner_collection(self): + user = UserProfileFactory() + org = OrganizationFactory() + org.members.add(user) + other_org = OrganizationFactory() + other_user = UserProfileFactory() + token = f"Token {user.get_token()}" + cases = ( + (f'/users/{user.username}/collections/', {'organization_id': other_org.id}), + (f'/users/{user.username}/collections/', {'organization': org.id}), + (f'/users/{user.username}/collections/', {'user_id': other_user.id}), + (f'/users/{user.username}/collections/', {'user': other_user.id}), + (f'/orgs/{org.mnemonic}/collections/', {'organization_id': other_org.id}), + (f'/orgs/{org.mnemonic}/collections/', {'user_id': user.id}), + ) + for index, (url, owner_data) in enumerate(cases): + response = self.client.post( + url, {'id': f'collection-{index}', 'name': 'n', **owner_data}, HTTP_AUTHORIZATION=token, format='json' + ) + self.assertEqual(response.status_code, 400) + self.assertFalse({'Source': Source, 'Collection': Collection}['Collection'].objects.filter( + mnemonic=f'collection-{index}').exists()) + + def test_create_accepts_omitted_or_matching_body_owner_collection(self): + user = UserProfileFactory() + org = OrganizationFactory() + org.members.add(user) + token = f"Token {user.get_token()}" + for index, (url, owner_data) in enumerate(( + (f'/users/{user.username}/collections/', {}), + (f'/users/{user.username}/collections/', {'user_id': user.id}), + (f'/orgs/{org.mnemonic}/collections/', {'organization': org.id}), + )): + response = self.client.post( + url, {'id': f'collection-ok-{index}', 'name': 'n', **owner_data}, + HTTP_AUTHORIZATION=token, format='json' + ) + self.assertEqual(response.status_code, 201) + def test_verify_scope_no_owner_scope_raises_404(self): response = self.client.get('/collections/some-collection/') self.assertEqual(response.status_code, 404) diff --git a/core/common/mixins.py b/core/common/mixins.py index 993950f2..d15cda8e 100644 --- a/core/common/mixins.py +++ b/core/common/mixins.py @@ -18,6 +18,7 @@ from psycopg2.errorcodes import QUERY_CANCELED from pydash import compact, get from rest_framework import status +from rest_framework.exceptions import ValidationError as DRFValidationError from rest_framework.mixins import ListModelMixin, CreateModelMixin from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response @@ -483,6 +484,18 @@ def set_parent_resource(self): self.kwargs['parent_resource'] = self.parent_resource = parent_resource + def set_owner_from_parent(self, data, key_suffix='', reject_conflict=True): + """The URL parent is the only owner. A body owner naming anyone else raises a 400 (or is dropped + if reject_conflict is False); an omitted or matching one is accepted.""" + parent_type = self.parent_resource.resource_type.lower() + for key in ('organization', 'user', 'organization_id', 'user_id'): + value = data.pop(key, None) + is_parent = key.split('_', maxsplit=1)[0] == parent_type and str(value) == str(self.parent_resource.id) + if reject_conflict and value not in (None, '') and not is_parent: + raise DRFValidationError({key: ['Owner must be the one in the URL.']}) + data[parent_type + key_suffix] = self.parent_resource.id + return data + def create(self, request, **kwargs): # pylint: disable=unused-argument if not self.parent_resource: return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) @@ -497,11 +510,9 @@ def create(self, request, **kwargs): # pylint: disable=unused-argument if isinstance(supported_locales, str): supported_locales = compact(supported_locales.split(',')) - data = { - 'mnemonic': data.get('id'), - 'supported_locales': supported_locales, - 'version': HEAD, **data, **{self.parent_resource.resource_type.lower(): self.parent_resource.id} - } + data = self.set_owner_from_parent({ + 'mnemonic': data.get('id'), 'supported_locales': supported_locales, 'version': HEAD, **data + }) serializer = self.get_serializer(data=data) if serializer.is_valid(): diff --git a/core/map_projects/views.py b/core/map_projects/views.py index 478bb8ad..4937bab6 100644 --- a/core/map_projects/views.py +++ b/core/map_projects/views.py @@ -90,13 +90,9 @@ def create(self, request, **kwargs): # pylint: disable=unused-argument if not self.parent_resource: return Response(status=status.HTTP_405_METHOD_NOT_ALLOWED) - data = MapProject.format_request_data(request.data, self.parent_resource) - # Ownership comes only from the URL parent: a body organization_id posted to a user route would re-parent - # the project past HasMapProjectParentOwnership and CanCreateOrgMapProjects (OpenConceptLab/ocl_issues#2804). - parent_key = self.parent_resource.resource_type.lower() + '_id' - for key in ('organization_id', 'user_id'): - if key != parent_key: - data.pop(key, None) + data = self.set_owner_from_parent( + MapProject.format_request_data(request.data), '_id', reject_conflict=False + ) serializer = self.get_serializer(data=data) if not serializer.is_valid(): return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) diff --git a/core/sources/tests/tests.py b/core/sources/tests/tests.py index 091c77f2..351a06e9 100644 --- a/core/sources/tests/tests.py +++ b/core/sources/tests/tests.py @@ -54,6 +54,45 @@ def test_verify_scope_no_kwargs_non_get_raises_404(self): response = self.client.post('/sources/', {}, format='json') self.assertEqual(response.status_code, 404) + def test_create_rejects_conflicting_body_owner_source(self): + user = UserProfileFactory() + org = OrganizationFactory() + org.members.add(user) + other_org = OrganizationFactory() + other_user = UserProfileFactory() + token = f"Token {user.get_token()}" + cases = ( + (f'/users/{user.username}/sources/', {'organization_id': other_org.id}), + (f'/users/{user.username}/sources/', {'organization': org.id}), + (f'/users/{user.username}/sources/', {'user_id': other_user.id}), + (f'/users/{user.username}/sources/', {'user': other_user.id}), + (f'/orgs/{org.mnemonic}/sources/', {'organization_id': other_org.id}), + (f'/orgs/{org.mnemonic}/sources/', {'user_id': user.id}), + ) + for index, (url, owner_data) in enumerate(cases): + response = self.client.post( + url, {'id': f'source-{index}', 'name': 'n', **owner_data}, HTTP_AUTHORIZATION=token, format='json' + ) + self.assertEqual(response.status_code, 400) + self.assertFalse({'Source': Source, 'Collection': Collection}['Source'].objects.filter( + mnemonic=f'source-{index}').exists()) + + def test_create_accepts_omitted_or_matching_body_owner_source(self): + user = UserProfileFactory() + org = OrganizationFactory() + org.members.add(user) + token = f"Token {user.get_token()}" + for index, (url, owner_data) in enumerate(( + (f'/users/{user.username}/sources/', {}), + (f'/users/{user.username}/sources/', {'user_id': user.id}), + (f'/orgs/{org.mnemonic}/sources/', {'organization': org.id}), + )): + response = self.client.post( + url, {'id': f'source-ok-{index}', 'name': 'n', **owner_data}, + HTTP_AUTHORIZATION=token, format='json' + ) + self.assertEqual(response.status_code, 201) + def test_verify_scope_no_owner_scope_raises_404(self): response = self.client.get('/sources/some-source/') self.assertEqual(response.status_code, 404)