External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-23 23:41:23 -03:30
Code Issues Packages Projects Releases Wiki Activity

AC-686 Fix permissions for inventory source to allow org admins and users with write permissions to update.

Browse Source
This commit is contained in:
Chris Church 2013-11-20 13:54:51 -05:00
parent 7fa9e3289b
commit 39ad6b0ad1
3 changed files with 103 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
awx/api/views.py
View File

@ -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:

14
awx/main/access.py
View File

@ -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:

87
awx/main/tests/inventory.py
View File

@ -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', '')