From 39ad6b0ad1cf5e8aea5d20fa25a3275b93f21572 Mon Sep 17 00:00:00 2001 From: Chris Church Date: Wed, 20 Nov 2013 13:54:51 -0500 Subject: [PATCH] AC-686 Fix permissions for inventory source to allow org admins and users with write permissions to update. --- awx/api/views.py | 3 +- awx/main/access.py | 14 ++++++ awx/main/tests/inventory.py | 87 +++++++++++++++++++++++++++++++++++++ 3 files changed, 103 insertions(+), 1 deletion(-) diff --git a/awx/api/views.py b/awx/api/views.py index 0ab654b7cb..9a7c95dc5f 100644 --- a/awx/api/views.py +++ b/awx/api/views.py @@ -826,6 +826,7 @@ class InventorySourceUpdatesList(SubListAPIView): class InventorySourceUpdateView(GenericAPIView): model = InventorySource + is_job_start = True new_in_14 = True def get(self, request, *args, **kwargs): @@ -838,7 +839,7 @@ class InventorySourceUpdateView(GenericAPIView): def post(self, request, *args, **kwargs): obj = self.get_object() if obj.can_update: - inventory_update = obj.update(**request.DATA) + inventory_update = obj.update() if not inventory_update: return Response({}, status=status.HTTP_400_BAD_REQUEST) else: diff --git a/awx/main/access.py b/awx/main/access.py index 5af04b4b2c..ca69acc389 100644 --- a/awx/main/access.py +++ b/awx/main/access.py @@ -81,6 +81,8 @@ def check_user_access(user, model_class, action, *args, **kwargs): access_instance = access_class(user) access_method = getattr(access_instance, 'can_%s' % action, None) if not access_method: + logger.debug('%s.%s not found', access_instance.__class__.__name__, + 'can_%s' % action) continue result = access_method(*args, **kwargs) logger.debug('%s.%s %r returned %r', access_instance.__class__.__name__, @@ -464,6 +466,9 @@ class InventorySourceAccess(BaseAccess): else: return False + def can_start(self, obj): + return self.can_change(obj, {}) and obj.can_update + class InventoryUpdateAccess(BaseAccess): ''' I can see inventory updates when I can see the inventory source. @@ -478,6 +483,9 @@ class InventoryUpdateAccess(BaseAccess): inventory_sources_qs = self.user.get_queryset(InventorySource) return qs.filter(inventory_source__in=inventory_sources_qs) + def can_cancel(self, obj): + return self.can_change(obj, {}) and obj.can_cancel + class CredentialAccess(BaseAccess): ''' I can see credentials when: @@ -646,6 +654,9 @@ class ProjectAccess(BaseAccess): def can_delete(self, obj): return self.can_change(obj, None) + def can_start(self, obj): + return self.can_change(obj, {}) and obj.can_update + class ProjectUpdateAccess(BaseAccess): ''' I can see project updates when I can see the project. @@ -660,6 +671,9 @@ class ProjectUpdateAccess(BaseAccess): projects_qs = self.user.get_queryset(Project) return qs.filter(project__in=projects_qs) + def can_cancel(self, obj): + return self.can_change(obj, {}) and obj.can_cancel + class PermissionAccess(BaseAccess): ''' I can see a permission when: diff --git a/awx/main/tests/inventory.py b/awx/main/tests/inventory.py index 4d49eb107f..9872a35633 100644 --- a/awx/main/tests/inventory.py +++ b/awx/main/tests/inventory.py @@ -1030,6 +1030,93 @@ class InventoryUpdatesTest(BaseTransactionTest): self.assertFalse(re.match(r'^i-[0-9a-f]{8}$', group.name, re.I), group.name) + def test_post_inventory_source_update(self): + creds_url = reverse('api:credential_list') + inv_src_url = reverse('api:inventory_source_detail', + args=(self.group.inventory_source.pk,)) + inv_src_update_url = reverse('api:inventory_source_update_view', + args=(self.group.inventory_source.pk,)) + # Create a credential to use for this inventory source. + aws_cred_data = { + 'name': 'AWS key that does not need to have valid info because we ' + 'do not care if the update actually succeeds', + 'kind': 'aws', + 'user': self.super_django_user.pk, + 'username': 'aws access key id goes here', + 'password': 'aws secret access key goes here', + } + with self.current_user(self.super_django_user): + aws_cred_response = self.post(creds_url, aws_cred_data, expect=201) + aws_cred_id = aws_cred_response['id'] + # Updaate the inventory source to use EC2. + inv_src_data = { + 'source': 'ec2', + 'credential': aws_cred_id, + } + with self.current_user(self.super_django_user): + self.put(inv_src_url, inv_src_data, expect=200) + # Read the inventory source, verify the update URL returns can_update. + with self.current_user(self.super_django_user): + self.get(inv_src_url, expect=200) + response = self.get(inv_src_update_url, expect=200) + self.assertTrue(response['can_update']) + # Now do the update. + with self.current_user(self.super_django_user): + self.post(inv_src_update_url, {}, expect=202) + # Normal user should be allowed as an org admin. + with self.current_user(self.normal_django_user): + self.get(inv_src_url, expect=200) + response = self.get(inv_src_update_url, expect=200) + self.assertTrue(response['can_update']) + with self.current_user(self.normal_django_user): + self.post(inv_src_update_url, {}, expect=202) + # Other user should be denied as only an org user. + with self.current_user(self.other_django_user): + self.get(inv_src_url, expect=403) + response = self.get(inv_src_update_url, expect=403) + with self.current_user(self.other_django_user): + self.post(inv_src_update_url, {}, expect=403) + # If given read permission to the inventory, other user should be able + # to see the inventory source and update view, but not start an update. + other_perms_url = reverse('api:user_permissions_list', + args=(self.other_django_user.pk,)) + other_perms_data = { + 'name': 'read only inventory permission for other', + 'user': self.other_django_user.pk, + 'inventory': self.inventory.pk, + 'permission_type': 'read', + } + with self.current_user(self.super_django_user): + self.post(other_perms_url, other_perms_data, expect=201) + with self.current_user(self.other_django_user): + self.get(inv_src_url, expect=200) + response = self.get(inv_src_update_url, expect=200) + with self.current_user(self.other_django_user): + self.post(inv_src_update_url, {}, expect=403) + # Once given write permission, the normal user is able to update the + # inventory source. + other_perms_data = { + 'name': 'read-write inventory permission for other', + 'user': self.other_django_user.pk, + 'inventory': self.inventory.pk, + 'permission_type': 'write', + } + with self.current_user(self.super_django_user): + self.post(other_perms_url, other_perms_data, expect=201) + with self.current_user(self.other_django_user): + self.get(inv_src_url, expect=200) + response = self.get(inv_src_update_url, expect=200) + # FIXME: This is misleading, as an update would fail... + self.assertTrue(response['can_update']) + with self.current_user(self.other_django_user): + self.post(inv_src_update_url, {}, expect=202) + # Nobody user should be denied as well. + with self.current_user(self.nobody_django_user): + self.get(inv_src_url, expect=403) + response = self.get(inv_src_update_url, expect=403) + with self.current_user(self.nobody_django_user): + self.post(inv_src_update_url, {}, expect=403) + def test_update_from_ec2(self): source_username = getattr(settings, 'TEST_AWS_ACCESS_KEY_ID', '') source_password = getattr(settings, 'TEST_AWS_SECRET_ACCESS_KEY', '')