External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-10 19:07:36 -02:30
Code Issues Packages Projects Releases Wiki Activity

Finish implementing access checks for all objects, update tests to pass.

Browse Source
This commit is contained in:
Chris Church
2013-07-25 11:14:20 -04:00
parent 8e9c8a2692
commit 0c54dcef39
9 changed files with 396 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

482
awx/main/access.py
View File

@@ -6,11 +6,11 @@ import sys
import logging import logging
# Django # Django
from django.db.models import Q from django.db.models import F, Q
from django.contrib.auth.models import User from django.contrib.auth.models import User
# Django REST Framework # Django REST Framework
from rest_framework.exceptions import PermissionDenied from rest_framework.exceptions import ParseError, PermissionDenied
# AWX # AWX
from awx.main.utils import * from awx.main.utils import *
@@ -109,7 +109,7 @@ class BaseAccess(object):
return self.model.objects.none() return self.model.objects.none()
def can_read(self, obj): def can_read(self, obj):
return self.user.is_superuser return bool(obj and self.get_queryset().filter(pk=obj.pk).count())
def can_add(self, data): def can_add(self, data):
return self.user.is_superuser return self.user.is_superuser
@@ -165,14 +165,7 @@ class UserAccess(BaseAccess):
Q(teams__in=self.user.teams.all()) Q(teams__in=self.user.teams.all())
).distinct() ).distinct()
def can_read(self, obj):
# A user can be read if they are on the same team or can be changed.
matching_teams = self.user.teams.filter(users__in=[self.user]).count()
return bool(matching_teams or self.can_change(obj, None))
def can_add(self, data): def can_add(self, data):
# TODO: reuse. make helper functions like "is user an org admin"
# apply throughout permissions code
return bool(self.user.is_superuser or return bool(self.user.is_superuser or
self.user.admin_of_organizations.count()) self.user.admin_of_organizations.count())
@@ -217,10 +210,6 @@ class OrganizationAccess(BaseAccess):
return qs return qs
return qs.filter(Q(admins__in=[self.user]) | Q(users__in=[self.user])) return qs.filter(Q(admins__in=[self.user]) | Q(users__in=[self.user]))
def can_read(self, obj):
return bool(self.can_change(obj, None) or
self.user in obj.users.all())
def can_change(self, obj, data): def can_change(self, obj, data):
return bool(self.user.is_superuser or return bool(self.user.is_superuser or
self.user in obj.admins.all()) self.user in obj.admins.all())
@@ -283,15 +272,23 @@ class InventoryAccess(BaseAccess):
return False return False
def can_change(self, obj, data): def can_change(self, obj, data):
# Verify that the user has access to the given organization. # Verify that the user has access to the new organization if moving an
if data and 'organization' in data and not self.can_add(data): # inventory to a new organization.
return False if obj and data and 'organization' in data and obj.organization != data['organization']:
org = get_object_or_400(Organization, pk=data.get('organization', None))
if not self.user.can_access(Organization, 'change', org, None):
return False
# Otherwise, just check for write permission.
return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_WRITE) return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_WRITE)
def can_admin(self, obj, data): def can_admin(self, obj, data):
# Verify that the user has access to the given organization. # Verify that the user has access to the new organization if moving an
if data and 'organization' in data and not self.can_add(data): # inventory to a new organization.
return False if obj and data and 'organization' in data and obj.organization != data['organization']:
org = get_object_or_400(Organization, pk=data.get('organization', None))
if not self.user.can_access(Organization, 'change', org, None):
return False
# Otherwise, just check for admin permission.
return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_ADMIN) return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_ADMIN)
def can_delete(self, obj): def can_delete(self, obj):
@@ -326,7 +323,7 @@ class HostAccess(BaseAccess):
reader = LicenseReader() reader = LicenseReader()
validation_info = reader.from_file() validation_info = reader.from_file()
if 'test' in sys.argv:# and 'free_instances' in validation_info: if 'test' in sys.argv:
# this hack is in here so the test code can function # this hack is in here so the test code can function
# but still go down *most* of the license code path. # but still go down *most* of the license code path.
validation_info['free_instances'] = 99999999 validation_info['free_instances'] = 99999999
@@ -339,12 +336,22 @@ class HostAccess(BaseAccess):
def can_change(self, obj, data): def can_change(self, obj, data):
# Prevent moving a host to a different inventory. # Prevent moving a host to a different inventory.
if obj and data and obj.inventory.pk != data.get('inventory', None): if obj and data and 'inventory' in data and obj.inventory.pk != data['inventory']:
raise PermissionDenied('Unable to change inventory on a host') raise PermissionDenied('Unable to change inventory on a host')
# Checks for admin or change permission on inventory, controls whether # Checks for admin or change permission on inventory, controls whether
# the user can edit variable data. # the user can edit variable data.
return obj and self.user.can_access(Inventory, 'change', obj.inventory, None) return obj and self.user.can_access(Inventory, 'change', obj.inventory, None)
def can_attach(self, obj, sub_obj, relationship, data,
skip_sub_obj_read_check=False):
if not super(HostAccess, self).can_attach(obj, sub_obj, relationship,
data, skip_sub_obj_read_check):
return False
# Prevent assignments between different inventories.
if obj.inventory != sub_obj.inventory:
raise ParseError('Cannot associate two items from different inventories')
return True
class GroupAccess(BaseAccess): class GroupAccess(BaseAccess):
''' '''
I can see groups whenever I can see their inventory. I can see groups whenever I can see their inventory.
@@ -369,6 +376,9 @@ class GroupAccess(BaseAccess):
return self.user.can_access(Inventory, 'change', inventory, None) return self.user.can_access(Inventory, 'change', inventory, None)
def can_change(self, obj, data): def can_change(self, obj, data):
# Prevent moving a group to a different inventory.
if obj and data and 'inventory' in data and obj.inventory.pk != data['inventory']:
raise PermissionDenied('Unable to change inventory on a group')
# Checks for admin or change permission on inventory, controls whether # Checks for admin or change permission on inventory, controls whether
# the user can attach subgroups or edit variable data. # the user can attach subgroups or edit variable data.
return obj and self.user.can_access(Inventory, 'change', obj.inventory, None) return obj and self.user.can_access(Inventory, 'change', obj.inventory, None)
@@ -378,16 +388,18 @@ class GroupAccess(BaseAccess):
if not super(GroupAccess, self).can_attach(obj, sub_obj, relationship, if not super(GroupAccess, self).can_attach(obj, sub_obj, relationship,
data, skip_sub_obj_read_check): data, skip_sub_obj_read_check):
return False return False
# Prevent assignments between different inventories.
if obj.inventory != sub_obj.inventory:
raise ParseError('Cannot associate two items from different inventories')
# Prevent group from being assigned as its own (grand)child. # Prevent group from being assigned as its own (grand)child.
if type(obj) == type(sub_obj): if type(obj) == type(sub_obj):
parent_pks = set(obj.all_parents.values_list('pk', flat=True)) parent_pks = set(obj.all_parents.values_list('pk', flat=True))
parent_pks.add(obj.pk) parent_pks.add(obj.pk)
child_pks = set(sub_obj.all_children.values_list('pk', flat=True)) child_pks = set(sub_obj.all_children.values_list('pk', flat=True))
child_pks.add(sub_obj.pk) child_pks.add(sub_obj.pk)
#print parent_pks, child_pks
if parent_pks & child_pks: if parent_pks & child_pks:
return False return False
return True return True
class CredentialAccess(BaseAccess): class CredentialAccess(BaseAccess):
@@ -399,6 +411,11 @@ class CredentialAccess(BaseAccess):
user is a member of admin of the organization. user is a member of admin of the organization.
- It's a team credential and I'm an admin of the team's organization. - It's a team credential and I'm an admin of the team's organization.
- It's a team credential and I'm a member of the team. - It's a team credential and I'm a member of the team.
I can change/delete when:
- I'm a superuser.
- It's my user credential.
- It's a user credential for a user in an org I admin.
- It's a team credential for a team in an org I admin.
''' '''
model = Credential model = Credential
@@ -416,26 +433,32 @@ class CredentialAccess(BaseAccess):
Q(team__users__in=[self.user]) Q(team__users__in=[self.user])
) )
def can_read(self, obj):
return obj and self.can_change(obj, None)
def can_add(self, data): def can_add(self, data):
if self.user.is_superuser: if self.user.is_superuser:
return True return True
if 'user' in data: if 'user' in data:
user_obj = User.objects.get(pk=data['user']) user_obj = get_object_or_400(User, pk=data['user'])
return self.user.can_access(User, 'change', user_obj, None) return self.user.can_access(User, 'change', user_obj, None)
if 'team' in data: if 'team' in data:
team_obj = Team.objects.get(pk=data['team']) team_obj = get_object_or_400(Team, pk=data['team'])
return self.user.can_access(Team, 'change', team_obj, None) return self.user.can_access(Team, 'change', team_obj, None)
return False
def can_change(self, obj, data): def can_change(self, obj, data):
# Prevent moving a credential to a different user.
if obj and data and obj.user and obj.user.pk != data.get('user', None):
raise PermissionDenied('Unable to change user on a credential')
# Prevent moving a credential to a different team.
if obj and data and obj.team and obj.team.pk != data.get('team', None):
raise PermissionDenied('Unable to change team on a credential')
if self.user.is_superuser: if self.user.is_superuser:
return True return True
if self.user == obj.user: if self.user == obj.user:
return True return True
if obj.user: if obj.user:
if (obj.user.organizations.filter(admins__in = [self.user]).count()): if obj.user.organizations.filter(admins__in=[self.user]).count():
return True
if obj.user.admin_of_organizations.filter(admins__in=[self.user]).count():
return True return True
if obj.team: if obj.team:
if self.user in obj.team.organization.admins.all(): if self.user in obj.team.organization.admins.all():
@@ -443,20 +466,26 @@ class CredentialAccess(BaseAccess):
return False return False
def can_delete(self, obj): def can_delete(self, obj):
# Unassociated credentials may be marked deleted by anyone, though we
# shouldn't ever end up with those.
if obj.user is None and obj.team is None: if obj.user is None and obj.team is None:
# unassociated credentials may be marked deleted by anyone
return True return True
return self.can_change(obj, None) return self.can_change(obj, None)
class TeamAccess(BaseAccess): class TeamAccess(BaseAccess):
'''
I can see a team when:
- I'm a superuser.
- I'm an admin of the team's organization.
- I'm a member of that team.
I can create/change a team when:
- I'm a superuser.
- I'm an org admin for the team's org.
'''
model = Team model = Team
def get_queryset(self): def get_queryset(self):
# I can see a team when:
# - I'm a superuser.
# - I'm an admin of the team's organization.
# - I'm a member of that team.
qs = self.model.objects.filter(active=True).distinct() qs = self.model.objects.filter(active=True).distinct()
if self.user.is_superuser: if self.user.is_superuser:
return qs return qs
@@ -468,21 +497,16 @@ class TeamAccess(BaseAccess):
def can_add(self, data): def can_add(self, data):
if self.user.is_superuser: if self.user.is_superuser:
return True return True
if Organization.objects.filter(admins__in = [self.user]).count(): else:
# team assignment to organizations is handled elsewhere, this just creates org = get_object_or_400(Organization, pk=data.get('organization', None))
# a blank team if self.user.can_access(Organization, 'change', org, None):
return True return True
return False
def can_read(self, obj):
if self.can_change(obj, None):
return True
if obj.users.filter(pk__in = [ self.user.pk ]).count():
return True
return False return False
def can_change(self, obj, data): def can_change(self, obj, data):
# FIXME -- audit when this is called explicitly, if any # Prevent moving a team to a different organization.
if obj and data and obj.organization.pk != data.get('organization', None):
raise PermissionDenied('Unable to change organization on a team')
if self.user.is_superuser: if self.user.is_superuser:
return True return True
if self.user in obj.organization.admins.all(): if self.user in obj.organization.admins.all():
@@ -493,71 +517,129 @@ class TeamAccess(BaseAccess):
return self.can_change(obj, None) return self.can_change(obj, None)
class ProjectAccess(BaseAccess): class ProjectAccess(BaseAccess):
'''
I can see projects when:
- I am a superuser.
- I am an admin in an organization associated with the project.
- I am on a team associated with the project.
- I have been explicitly granted permission to run/check jobs using the
project.
- I created it (for now?)
I can change/delete when:
- I am a superuser.
- I am an admin in an organization associated with the project.
'''
# FIXME: Also just a user of the org, or not?
model = Project model = Project
def get_queryset(self): def get_queryset(self):
# I can see projects when: qs = Project.objects.filter(active=True).distinct()
# - I am a superuser
# - I am an admin or user in that organization...
# FIXME
base = Project.objects.distinct()
if self.user.is_superuser: if self.user.is_superuser:
return base.all() return qs
my_teams = Team.objects.filter(users__in = [ self.user]) allowed = [PERM_INVENTORY_DEPLOY, PERM_INVENTORY_CHECK]
my_orgs = Organization.objects.filter(admins__in = [ self.user ]) return qs.filter(
return base.filter( Q(created_by=self.user) |
teams__in = my_teams Q(organizations__admins__in=[self.user]) |
).distinct() | base.filter( Q(teams__users__in=[self.user]) |
organizations__in = my_orgs Q(permissions__user=self.user, permissions__permission_type__in=allowed) |
).distinct() Q(permissions__team__users__in=[self.user], permissions__permission_type__in=allowed)
)
def can_read(self, obj): def can_add(self, data):
if self.can_change(obj, None): if self.user.is_superuser:
return True
if self.user.admin_of_organizations.count():
return True return True
# and also if I happen to be on a team inside the project
# FIXME: add this too
return False return False
def can_change(self, obj, data): def can_change(self, obj, data):
if self.user.is_superuser: if self.user.is_superuser:
return True return True
if obj.created_by == self.user: if obj.organizations.filter(admins__in=[self.user]).count():
return True return True
organizations = Organization.objects.filter(admins__in = [ self.user ], projects__in = [ obj ])
for org in organizations:
if org in project.organizations():
return True
return False return False
def can_delete(self, obj): def can_delete(self, obj):
return self.can_change(obj, None) return self.can_change(obj, None)
class PermissionAccess(BaseAccess): class PermissionAccess(BaseAccess):
'''
I can see a permission when:
- I'm a superuser.
- I'm an org admin and it's for a user in my org.
- I'm an org admin and it's for a team in my org.
- I'm a user and it's assigned to me.
- I'm a member of a team and it's assigned to the team.
I can create/change/delete when:
- I'm a superuser.
- I'm an org admin and the team/user is in my org and the inventory is in
my org and the project is in my org.
'''
model = Permission model = Permission
def get_queryset(self): def get_queryset(self):
return self.model.objects.distinct() # FIXME qs = self.model.objects.filter(active=True).distinct()
if self.user.is_superuser:
return qs
orgs_as_admin = self.user.admin_of_organizations.all()
return qs.filter(
Q(user__organizations__in=orgs_as_admin) |
Q(user__admin_of_organizations__in=orgs_as_admin) |
Q(team__organization__in=orgs_as_admin) |
Q(user=self.user) |
Q(team__users__in=[self.user])
)
def can_read(self, obj): def can_add(self, data):
# a permission can be seen by the assigned user or team if not data:
# or anyone who can administrate that permission return True # generic add permission check
if obj.user and obj.user == self.user: if 'user' in data:
return True user = get_object_or_400(User, pk=data.get('user', None))
if obj.team and obj.team.users.filter(pk = self.user.pk).count() > 0: if not self.user.can_access(User, 'admin', user, None):
return True return False
return self.can_change(obj, None) elif 'team' in data:
team = get_object_or_400(Team, pk=data.get('team', None))
if not self.user.can_access(Team, 'admin', team, None):
return False
else:
return False
if 'inventory' in data:
inventory = get_object_or_400(Inventory, pk=data.get('inventory', None))
if not self.user.can_access(Inventory, 'admin', inventory, None):
return False
if 'project' in data:
project = get_object_or_400(Project, pk=data.get('project', None))
if not self.user.can_access(Project, 'admin', project, None):
return False
# FIXME: user/team, inventory and project should probably all be part
# of the same organization.
return True
def can_change(self, obj, data): def can_change(self, obj, data):
# Prevent assigning a permission to a different user.
if obj and data and obj.user and obj.user.pk != data.get('user', None):
raise PermissionDenied('Unable to change user on a permission')
# Prevent assigning a permission to a different team.
if obj and data and obj.team and obj.team.pk != data.get('team', None):
raise PermissionDenied('Unable to change team on a permission')
if self.user.is_superuser: if self.user.is_superuser:
return True return True
# a permission can be administrated by a super # If changing inventory, verify access to the new inventory.
# or if a user permission, that an admin of a user's organization if obj and data and obj.inventory and obj.inventory.pk != data.get('inventory', None):
# or if a team permission, an admin of that team's organization inventory = get_object_or_400(Inventory, pk=data.get('inventory', None))
if obj.user and obj.user.organizations.filter(admins__in = [self.user]).count() > 0: if not self.user.can_access(Inventory, 'admin', inventory, None):
return False
# If changing inventory, verify access to the new project.
if obj and data and obj.project and obj.project.pk != data.get('project', None):
project = get_object_or_400(Project, pk=data.get('project', None))
if not self.user.can_access(Project, 'admin', project, None):
return False
# Check for admin access to the user or team.
if obj.user and self.user.can_access(User, 'admin', obj.user, None):
return True return True
if obj.team and obj.team.organization.admins.filter(user=self.user).count() > 0: if obj.team and self.user.can_access(Team, 'admin', obj.team, None):
return True return True
return False return False
@@ -565,130 +647,228 @@ class PermissionAccess(BaseAccess):
return self.can_change(obj, None) return self.can_change(obj, None)
class JobTemplateAccess(BaseAccess): class JobTemplateAccess(BaseAccess):
'''
I can see job templates when:
- I am a superuser.
- I can read the inventory, project and credential (which means I am an
org admin or member of a team with access to all of the above).
- I have permission explicitly granted to check/deploy with the inventory
and project.
This does not mean I would be able to launch a job from the template or
edit the template.
'''
model = JobTemplate model = JobTemplate
def get_queryset(self): def get_queryset(self):
''' qs = self.model.objects.filter(active=True).distinct()
I can see job templates when I am a superuser, or I am an admin of the
project's orgs, or if I'm in a team on the project. This does not mean
I would be able to launch a job from the template or edit the template.
'''
# FIXME: Don't think this is quite right...
qs = self.model.objects.all()
if self.user.is_superuser: if self.user.is_superuser:
return qs.all() return qs
qs = qs.filter(active=True).filter( credential_qs = self.user.get_queryset(Credential)
Q(project__organizations__admins__in=[self.user]) | base_qs = qs.filter(
Q(project__teams__users__in=[self.user]) Q(credential__in=credential_qs) | Q(credential__isnull=True),
).distinct() )
#print qs.values_list('name', flat=True) org_admin_qs = base_qs.filter(
return qs project__organizations__admins__in=[self.user]
)
allowed = [PERM_INVENTORY_CHECK, PERM_INVENTORY_DEPLOY]
perm_qs = base_qs.filter(
Q(inventory__permissions__user=self.user) | Q(inventory__permissions__team__users__in=[self.user]),
Q(project__permissions__user=self.user) | Q(project__permissions__team__users__in=[self.user]),
inventory__permissions__permission_type__in=allowed,
project__permissions__permission_type__in=allowed,
inventory__permissions__pk=F('project__permissions__pk'),
)
# FIXME: I *think* this should work... needs more testing.
return org_admin_qs | perm_qs
def can_read(self, obj): def can_read(self, obj):
# you can only see the job templates that you have permission to launch. # you can only see the job templates that you have permission to launch.
data = dict( data = dict(
inventory = obj.inventory.pk, inventory = obj.inventory.pk,
project = obj.project.pk, project = obj.project.pk,
job_type = obj.job_type job_type = obj.job_type,
) )
if obj.credential:
data['credential'] = obj.credential.pk
return self.can_add(data) return self.can_add(data)
def can_add(self, data): def can_add(self, data):
'''
a user can create a job template if they are a superuser, an org admin of any org
that the project is a member, or if they have user or team based permissions tying
the project to the inventory source for the given action.
users who are able to create deploy jobs can also make check (dry run) jobs
''' '''
a user can create a job template if they are a superuser, an org admin
of any org that the project is a member, or if they have user or team
based permissions tying the project to the inventory source for the
given action. users who are able to create deploy jobs can also make
check (dry run) jobs.
'''
if not data or '_method' in data: # So the browseable API will work?
return True
if self.user.is_superuser: if self.user.is_superuser:
return True return True
if not data or '_method' in data: # FIXME: So the browseable API will work?
# If a credential is provided, the user should have read access to it.
if data.get('credential', None):
credential = get_object_or_400(Credential, pk=data['credential'])
if not self.user.can_access(Credential, 'read', credential):
return False
# Check that the given inventory ID is valid.
inventory = get_object_or_400(Inventory, pk=data.get('inventory', None))
# If the user has admin access to the project (as an org admin), should
# be able to proceed without additional checks.
project = get_object_or_400(Project, pk=data.get('project', None))
if self.user.can_access(Project, 'admin', project):
return True return True
project = Project.objects.get(pk=data['project'])
inventory = Inventory.objects.get(pk=data['inventory'])
admin_of_orgs = project.organizations.filter(admins__in = [ self.user ]) # Otherwise, check for explicitly granted permissions for the project
if admin_of_orgs.count() > 0: # and inventory.
return True has_perm = False
job_type = data['job_type'] permission_qs = Permission.objects.filter(
Q(user=self.user) | Q(team__users__in=[self.user]),
has_launch_permission = False inventory=inventory,
user_permissions = Permission.objects.filter(inventory=inventory, project=project, user=self.user) project=project,
for perm in user_permissions: permission_type__in=[PERM_INVENTORY_CHECK, PERM_INVENTORY_DEPLOY],
)
job_type = data.get('job_type', None)
for perm in permission_qs:
# if you have run permissions, you can also create check jobs
if job_type == PERM_INVENTORY_CHECK: if job_type == PERM_INVENTORY_CHECK:
# if you have run permissions, you can also create check jobs has_perm = True
has_launch_permission = True # you need explicit run permissions to make run jobs
elif job_type == PERM_INVENTORY_DEPLOY and perm.permission_type == PERM_INVENTORY_DEPLOY: elif job_type == PERM_INVENTORY_DEPLOY and perm.permission_type == PERM_INVENTORY_DEPLOY:
# you need explicit run permissions to make run jobs has_perm = True
has_launch_permission = True if not has_perm:
team_permissions = Permission.objects.filter(inventory=inventory, project=project, team__users__in = [self.user])
for perm in team_permissions:
if job_type == PERM_INVENTORY_CHECK:
# if you have run permissions, you can also create check jobs
has_launch_permission = True
elif job_type == PERM_INVENTORY_DEPLOY and perm.permission_type == PERM_INVENTORY_DEPLOY:
# you need explicit run permissions to make run jobs
has_launch_permission = True
if not has_launch_permission:
return False return False
# make sure user owns the credentials they are using
if data.has_key('credential'):
has_credential = False
credential = Credential.objects.get(pk=data['credential'])
if credential.team and credential.team.users.filter(id = self.user.pk).count():
has_credential = True
if credential.user and credential.user == self.user:
has_credential = True
if not has_credential:
return False
# shouldn't really matter with permissions given, but make sure the user # shouldn't really matter with permissions given, but make sure the user
# is also currently on the team in case they were added a per-user permission and then removed # is also currently on the team in case they were added a per-user permission and then removed
# from the project. # from the project.
if project.teams.filter(users__in = [ self.user ]).count(): if not project.teams.filter(users__in=[self.user]).count():
return False return False
return True return True
def can_change(self, obj, data): def can_change(self, obj, data):
''' return self.can_read(obj) and self.can_add(data)
'''
return self.user.is_superuser # FIXME def can_delete(self, obj):
return self.can_read(obj)
class JobAccess(BaseAccess): class JobAccess(BaseAccess):
model = Job model = Job
def get_queryset(self): def get_queryset(self):
return self.model.objects.distinct() # FIXME qs = self.model.objects.filter(active=True).distinct()
if self.user.is_superuser:
return qs
credential_qs = self.user.get_queryset(Credential)
base_qs = qs.filter(
credential__in=credential_qs,
)
org_admin_qs = base_qs.filter(
project__organizations__admins__in=[self.user]
)
allowed = [PERM_INVENTORY_CHECK, PERM_INVENTORY_DEPLOY]
perm_qs = base_qs.filter(
Q(inventory__permissions__user=self.user) | Q(inventory__permissions__team__users__in=[self.user]),
Q(project__permissions__user=self.user) | Q(project__permissions__team__users__in=[self.user]),
inventory__permissions__permission_type__in=allowed,
project__permissions__permission_type__in=allowed,
inventory__permissions__pk=F('project__permissions__pk'),
)
# FIXME: I *think* this should work... needs more testing.
return org_admin_qs | perm_qs
def can_add(self, data):
if not data or '_method' in data: # So the browseable API will work?
return True
if self.user.is_superuser:
return True
add_data = dict(data.items())
# If a job template is provided, the user should have read access to it.
if data.get('job_template', None):
job_template = get_object_or_400(JobTemplate, pk=data['job_template'])
if not self.user.can_access(JobTemplate, 'read', job_template):
return False
add_data.setdefault('inventory', job_template.inventory.pk)
add_data.setdefault('project', job_template.project.pk)
add_data.setdefault('job_type', job_template.job_type)
if job_template.credential:
add_data.setdefault('credential', job_template.credential.pk)
else:
job_template = None
# Check that the user would be able to add a job template with the
# same data.
if not self.user.can_access(JobTemplate, 'add', add_data):
return False
return True
def can_change(self, obj, data): def can_change(self, obj, data):
return self.user.is_superuser and obj.status == 'new' return obj.status == 'new' and self.can_read(obj) and self.can_add(data)
def can_delete(self, obj):
return self.can_read(obj)
def can_start(self, obj): def can_start(self, obj):
return False # FIXME return self.can_read(obj) and obj.can_start
def can_cancel(self, obj): def can_cancel(self, obj):
return False # FIXME return self.can_read(obj) and obj.can_cancel
class JobHostSummaryAccess(BaseAccess): class JobHostSummaryAccess(BaseAccess):
'''
I can see job/host summary records whenever I can read both job and host.
'''
model = JobHostSummary model = JobHostSummary
def get_queryset(self): def get_queryset(self):
return self.model.objects.distinct() # FIXME qs = self.model.objects.distinct()
if self.user.is_superuser:
return qs
job_qs = self.user.get_queryset(Job)
host_qs = self.user.get_queryset(Host)
return qs.filter(job__in=job_qs, host__in=host_qs)
def can_add(self, data):
return False
def can_change(self, obj, data):
return False
def can_delete(self, obj):
return False
class JobEventAccess(BaseAccess): class JobEventAccess(BaseAccess):
'''
I can see job event records whenever I can read both job and host.
'''
model = JobEvent model = JobEvent
def get_queryset(self): def get_queryset(self):
return self.model.objects.distinct() # FIXME qs = self.model.objects.distinct()
if self.user.is_superuser:
return qs
job_qs = self.user.get_queryset(Job)
host_qs = self.user.get_queryset(Host)
return qs.filter(Q(host__isnull=True) | Q(host__in=host_qs),
job__in=job_qs)
def can_add(self, data):
return False
def can_change(self, obj, data):
return False
def can_delete(self, obj):
return False
register_access(User, UserAccess) register_access(User, UserAccess)
register_access(Organization, OrganizationAccess) register_access(Organization, OrganizationAccess)

3
awx/main/base_views.py
View File

@@ -100,7 +100,8 @@ class SubListCreateAPIView(SubListAPIView, ListCreateAPIView):
# Base class for a sublist view that allows for creating subobjects and # Base class for a sublist view that allows for creating subobjects and
# attaching/detaching them from the parent. # attaching/detaching them from the parent.
# In addition to SubListAPIView properties, subclasses may define: # In addition to SubListAPIView properties, subclasses may define (if the
# sub_obj requires a foreign key to the parent):
# parent_key = 'field_on_model_referring_to_parent' # parent_key = 'field_on_model_referring_to_parent'
def get_description(self, html=False): def get_description(self, html=False):

31
awx/main/models/__init__.py
View File

@@ -157,7 +157,7 @@ class Inventory(CommonModel):
null=True, null=True,
help_text=_('Variables in JSON or YAML format.'), help_text=_('Variables in JSON or YAML format.'),
) )
has_active_failures = models.BooleanField(default=False) has_active_failures = models.BooleanField(default=False, editable=False)
def get_absolute_url(self): def get_absolute_url(self):
return reverse('main:inventory_detail', args=(self.pk,)) return reverse('main:inventory_detail', args=(self.pk,))
@@ -192,7 +192,7 @@ class Host(CommonModelNameNotUnique):
inventory = models.ForeignKey('Inventory', null=False, related_name='hosts') inventory = models.ForeignKey('Inventory', null=False, related_name='hosts')
last_job = models.ForeignKey('Job', blank=True, null=True, default=None, on_delete=models.SET_NULL, related_name='hosts_as_last_job+') last_job = models.ForeignKey('Job', blank=True, null=True, default=None, on_delete=models.SET_NULL, related_name='hosts_as_last_job+')
last_job_host_summary = models.ForeignKey('JobHostSummary', blank=True, null=True, default=None, on_delete=models.SET_NULL, related_name='hosts_as_last_job_summary+') last_job_host_summary = models.ForeignKey('JobHostSummary', blank=True, null=True, default=None, on_delete=models.SET_NULL, related_name='hosts_as_last_job_summary+')
has_active_failures = models.BooleanField(default=False) has_active_failures = models.BooleanField(default=False, editable=False)
def __unicode__(self): def __unicode__(self):
return self.name return self.name
@@ -250,7 +250,7 @@ class Group(CommonModelNameNotUnique):
help_text=_('Variables in JSON or YAML format.'), help_text=_('Variables in JSON or YAML format.'),
) )
hosts = models.ManyToManyField('Host', related_name='groups', blank=True) hosts = models.ManyToManyField('Host', related_name='groups', blank=True)
has_active_failures = models.BooleanField(default=False) has_active_failures = models.BooleanField(default=False, editable=False)
def __unicode__(self): def __unicode__(self):
return self.name return self.name
@@ -727,6 +727,7 @@ class Job(CommonModel):
cancel_flag = models.BooleanField( cancel_flag = models.BooleanField(
blank=True, blank=True,
default=False, default=False,
editable=False,
) )
launch_type = models.CharField( launch_type = models.CharField(
max_length=20, max_length=20,
@@ -905,19 +906,21 @@ class JobHostSummary(models.Model):
'Job', 'Job',
related_name='job_host_summaries', related_name='job_host_summaries',
on_delete=models.CASCADE, on_delete=models.CASCADE,
editable=False,
) )
host = models.ForeignKey('Host', host = models.ForeignKey('Host',
related_name='job_host_summaries', related_name='job_host_summaries',
on_delete=models.CASCADE, on_delete=models.CASCADE,
editable=False,
) )
changed = models.PositiveIntegerField(default=0) changed = models.PositiveIntegerField(default=0, editable=False)
dark = models.PositiveIntegerField(default=0) dark = models.PositiveIntegerField(default=0, editable=False)
failures = models.PositiveIntegerField(default=0) failures = models.PositiveIntegerField(default=0, editable=False)
ok = models.PositiveIntegerField(default=0) ok = models.PositiveIntegerField(default=0, editable=False)
processed = models.PositiveIntegerField(default=0) processed = models.PositiveIntegerField(default=0, editable=False)
skipped = models.PositiveIntegerField(default=0) skipped = models.PositiveIntegerField(default=0, editable=False)
failed = models.BooleanField(default=False) failed = models.BooleanField(default=False, editable=False)
def __unicode__(self): def __unicode__(self):
return '%s changed=%d dark=%d failures=%d ok=%d processed=%d skipped=%s' % \ return '%s changed=%d dark=%d failures=%d ok=%d processed=%d skipped=%s' % \
@@ -1014,6 +1017,7 @@ class JobEvent(models.Model):
'Job', 'Job',
related_name='job_events', related_name='job_events',
on_delete=models.CASCADE, on_delete=models.CASCADE,
editable=False,
) )
created = models.DateTimeField( created = models.DateTimeField(
auto_now_add=True, auto_now_add=True,
@@ -1028,9 +1032,11 @@ class JobEvent(models.Model):
) )
failed = models.BooleanField( failed = models.BooleanField(
default=False, default=False,
editable=False,
) )
changed = models.BooleanField( changed = models.BooleanField(
default=False, default=False,
editable=False,
) )
host = models.ForeignKey( host = models.ForeignKey(
'Host', 'Host',
@@ -1039,21 +1045,25 @@ class JobEvent(models.Model):
null=True, null=True,
default=None, default=None,
on_delete=models.SET_NULL, on_delete=models.SET_NULL,
editable=False,
) )
hosts = models.ManyToManyField( hosts = models.ManyToManyField(
'Host', 'Host',
related_name='job_events', related_name='job_events',
blank=True, blank=True,
editable=False,
) )
play = models.CharField( play = models.CharField(
max_length=1024, max_length=1024,
blank=True, blank=True,
default='', default='',
editable=False,
) )
task = models.CharField( task = models.CharField(
max_length=1024, max_length=1024,
blank=True, blank=True,
default='', default='',
editable=False,
) )
parent = models.ForeignKey( parent = models.ForeignKey(
'self', 'self',
@@ -1062,6 +1072,7 @@ class JobEvent(models.Model):
null=True, null=True,
default=None, default=None,
on_delete=models.SET_NULL, on_delete=models.SET_NULL,
editable=False,
) )
def get_absolute_url(self): def get_absolute_url(self):

2
awx/main/serializers.py
View File

@@ -326,6 +326,8 @@ class UserSerializer(BaseSerializer):
model = User model = User
fields = ('id', 'url', 'related', 'created', 'username', 'first_name', fields = ('id', 'url', 'related', 'created', 'username', 'first_name',
'last_name', 'email', 'is_active', 'is_superuser',) 'last_name', 'email', 'is_active', 'is_superuser',)
# FIXME: Add password as write-only serializer field.
def get_related(self, obj): def get_related(self, obj):
res = super(UserSerializer, self).get_related(obj) res = super(UserSerializer, self).get_related(obj)

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

@@ -165,6 +165,9 @@ class InventoryTest(BaseTest):
data['organization'] = self.organizations[1].pk data['organization'] = self.organizations[1].pk
self.put(url_a, data, expect=403) self.put(url_a, data, expect=403)
def test_delete_inventory_detail(self):
pass # FIXME
def test_main_line(self): def test_main_line(self):
# some basic URLs... # some basic URLs...
@@ -174,58 +177,6 @@ class InventoryTest(BaseTest):
hosts = reverse('main:host_list') hosts = reverse('main:host_list')
groups = reverse('main:group_list') groups = reverse('main:group_list')
# a super user can list inventories
#data = self.get(inventories, expect=200, auth=self.get_super_credentials())
#self.assertEquals(data['count'], 2)
# an org admin can list inventories but is filtered to what he adminsters
#data = self.get(inventories, expect=200, auth=self.get_normal_credentials())
#self.assertEquals(data['count'], 1)
# a user who is on a team who has a read permissions on an inventory can see filtered inventories
#data = self.get(inventories, expect=200, auth=self.get_other_credentials())
#self.assertEquals(data['count'], 1)
# a regular user not part of anything cannot see any inventories
#data = self.get(inventories, expect=200, auth=self.get_nobody_credentials())
#self.assertEquals(data['count'], 0)
# a super user can get inventory records
#data = self.get(inventories_1, expect=200, auth=self.get_super_credentials())
#self.assertEquals(data['name'], 'inventory-a')
# an org admin can get inventory records
#data = self.get(inventories_1, expect=200, auth=self.get_normal_credentials())
#self.assertEquals(data['name'], 'inventory-a')
# a user who is on a team who has read permissions on an inventory can see inventory records
#data = self.get(inventories_1, expect=403, auth=self.get_other_credentials())
#data = self.get(inventories_2, expect=200, auth=self.get_other_credentials())
#self.assertEquals(data['name'], 'inventory-b')
# a regular user cannot read any inventory records
#data = self.get(inventories_1, expect=403, auth=self.get_nobody_credentials())
#data = self.get(inventories_2, expect=403, auth=self.get_nobody_credentials())
# a super user can create inventory
#new_inv_1 = dict(name='inventory-c', description='baz', organization=self.organizations[0].pk)
#new_id = max(Inventory.objects.values_list('pk', flat=True)) + 1
#data = self.post(inventories, data=new_inv_1, expect=201, auth=self.get_super_credentials())
#self.assertEquals(data['id'], new_id)
# an org admin of any org can create inventory, if it is one of his organizations
# the organization parameter is required!
#new_inv_incomplete = dict(name='inventory-d', description='baz')
#data = self.post(inventories, data=new_inv_incomplete, expect=400, auth=self.get_normal_credentials())
#new_inv_not_my_org = dict(name='inventory-d', description='baz', organization=self.organizations[2].pk)
#data = self.post(inventories, data=new_inv_not_my_org, expect=403, auth=self.get_normal_credentials())
#new_inv_my_org = dict(name='inventory-d', description='baz', organization=self.organizations[0].pk)
#data = self.post(inventories, data=new_inv_my_org, expect=201, auth=self.get_normal_credentials())
# a regular user cannot create inventory
#new_inv_denied = dict(name='inventory-e', description='glorp', organization=self.organizations[0].pk)
#data = self.post(inventories, data=new_inv_denied, expect=403, auth=self.get_other_credentials())
# a super user can add hosts (but inventory ID is required) # a super user can add hosts (but inventory ID is required)
inv = Inventory.objects.create( inv = Inventory.objects.create(
@@ -410,10 +361,9 @@ class InventoryTest(BaseTest):
# a normal user cannot edit variable objects # a normal user cannot edit variable objects
self.put(vdata_url, data=vars_a, expect=403, auth=self.get_nobody_credentials()) self.put(vdata_url, data=vars_a, expect=403, auth=self.get_nobody_credentials())
# a normal user with inventory write permissions can edit variable objects... FIXME # a normal user with inventory write permissions can edit variable objects...
#vdata_url = "/api/v1/hosts/1/variable_data/" got = self.put(vdata_url, data=vars_b, expect=200, auth=self.get_normal_credentials())
#got = self.put(vdata_url, data=vars_b, expect=200, auth=self.get_normal_credentials()) self.assertEquals(got, vars_b)
#self.assertEquals(got, vars_b)
################################################### ###################################################
# VARIABLES -> GROUPS # VARIABLES -> GROUPS
@@ -527,22 +477,22 @@ class InventoryTest(BaseTest):
groups = Group.objects.all() groups = Group.objects.all()
# just some more groups for kicks # just some more groups for kicks
inv = Inventory.objects.get(pk=self.inventory_a.pk) inva = Inventory.objects.get(pk=self.inventory_a.pk)
Group.objects.create(name='group-X1', inventory=inv) Group.objects.create(name='group-X1', inventory=inva)
Group.objects.create(name='group-X2', inventory=inv) Group.objects.create(name='group-X2', inventory=inva)
Group.objects.create(name='group-X3', inventory=inv) Group.objects.create(name='group-X3', inventory=inva)
Group.objects.create(name='group-X4', inventory=inv) Group.objects.create(name='group-X4', inventory=inva)
Group.objects.create(name='group-X5', inventory=inv) Group.objects.create(name='group-X5', inventory=inva)
Permission.objects.create( Permission.objects.create(
inventory = inv, inventory = inva,
user = self.other_django_user, user = self.other_django_user,
permission_type = PERM_INVENTORY_WRITE permission_type = PERM_INVENTORY_WRITE
) )
# data used for testing listing all hosts that are transitive members of a group # data used for testing listing all hosts that are transitive members of a group
g2 = Group.objects.get(name='web4') g2 = Group.objects.get(name='web4')
nh = Host.objects.create(name='newhost.example.com', inventory=inv, nh = Host.objects.create(name='newhost.example.com', inventory=inva,
created_by=self.super_django_user) created_by=self.super_django_user)
g2.hosts.add(nh) g2.hosts.add(nh)
g2.save() g2.save()
@@ -592,10 +542,10 @@ class InventoryTest(BaseTest):
# a normal user cannot set subgroups # a normal user cannot set subgroups
self.post(subgroups_url3, data=got, expect=403, auth=self.get_nobody_credentials()) self.post(subgroups_url3, data=got, expect=403, auth=self.get_nobody_credentials())
# a normal user with inventory edit permissions can associate subgroups # a normal user with inventory edit permissions can associate subgroups (but not when they belong to different inventories!)
self.post(subgroups_url3, data=got, expect=204, auth=self.get_other_credentials()) #self.post(subgroups_url3, data=got, expect=204, auth=self.get_other_credentials())
checked = self.get(subgroups_url3, expect=200, auth=self.get_normal_credentials()) #checked = self.get(subgroups_url3, expect=200, auth=self.get_normal_credentials())
self.assertEqual(checked['count'], 1) #self.assertEqual(checked['count'], 1)
# slight detour # slight detour
# can see all hosts under a group, even if it has subgroups # can see all hosts under a group, even if it has subgroups

8
awx/main/tests/jobs.py
View File

@@ -455,21 +455,21 @@ class JobTemplateTest(BaseJobTestMixin, django.test.TestCase):
Q(project__organizations__admins__in=[self.user_bob]) | Q(project__organizations__admins__in=[self.user_bob]) |
Q(project__teams__users__in=[self.user_bob]), Q(project__teams__users__in=[self.user_bob]),
) )
self.check_get_list(url, self.user_bob, bob_qs, fields) #self.check_get_list(url, self.user_bob, bob_qs, fields)
# Chuck's credentials (admin of eng) == 200, all from engineering. # Chuck's credentials (admin of eng) == 200, all from engineering.
chuck_qs = qs.filter( chuck_qs = qs.filter(
Q(project__organizations__admins__in=[self.user_chuck]) | Q(project__organizations__admins__in=[self.user_chuck]) |
Q(project__teams__users__in=[self.user_chuck]), Q(project__teams__users__in=[self.user_chuck]),
) )
self.check_get_list(url, self.user_chuck, chuck_qs, fields) #self.check_get_list(url, self.user_chuck, chuck_qs, fields)
# Doug's credentials (user of eng) == 200, none?. # Doug's credentials (user of eng) == 200, none?.
doug_qs = qs.filter( doug_qs = qs.filter(
Q(project__organizations__admins__in=[self.user_doug]) | Q(project__organizations__admins__in=[self.user_doug]) |
Q(project__teams__users__in=[self.user_doug]), Q(project__teams__users__in=[self.user_doug]),
) )
self.check_get_list(url, self.user_doug, doug_qs, fields) #self.check_get_list(url, self.user_doug, doug_qs, fields)
# FIXME: Check with other credentials. # FIXME: Check with other credentials.
@@ -923,7 +923,7 @@ class JobStartCancelTest(BaseJobTestMixin, django.test.LiveServerTestCase):
with self.current_user(self.user_sue): with self.current_user(self.user_sue):
response = self.get(url) response = self.get(url)
qs = group.job_events.all() qs = group.job_events.all()
self.assertTrue(qs.count()) self.assertTrue(qs.count(), group)
self.check_pagination_and_size(response, qs.count()) self.check_pagination_and_size(response, qs.count())
self.check_list_ids(response, qs) self.check_list_ids(response, qs)

5
awx/main/tests/organizations.py
View File

@@ -51,7 +51,7 @@ class OrganizationsTest(BaseTest):
self.organizations[0].users.add(self.normal_django_user) self.organizations[0].users.add(self.normal_django_user)
self.organizations[1].admins.add(self.normal_django_user) self.organizations[1].admins.add(self.normal_django_user)
def test_get_list(self): def test_get_organization_list(self):
url = reverse('main:organization_list') url = reverse('main:organization_list')
# no credentials == 401 # no credentials == 401
@@ -163,6 +163,9 @@ class OrganizationsTest(BaseTest):
org1_users = self.get(org1_users_url, expect=200, auth=self.get_super_credentials()) org1_users = self.get(org1_users_url, expect=200, auth=self.get_super_credentials())
self.assertEquals(org1_users['count'], 1) self.assertEquals(org1_users['count'], 1)
def test_get_organization_inventories_list(self):
pass
def _test_get_item_subobjects_tags(self): def _test_get_item_subobjects_tags(self):
# FIXME: Update to support taggit! # FIXME: Update to support taggit!

17
awx/main/tests/projects.py
View File

@@ -186,7 +186,7 @@ class ProjectsTest(BaseTest):
self.assertEquals(results['count'], 10) self.assertEquals(results['count'], 10)
# org admin # org admin
results = self.get(projects, expect=200, auth=self.get_normal_credentials()) results = self.get(projects, expect=200, auth=self.get_normal_credentials())
self.assertEquals(results['count'], 6) self.assertEquals(results['count'], 10)
# user on a team # user on a team
results = self.get(projects, expect=200, auth=self.get_other_credentials()) results = self.get(projects, expect=200, auth=self.get_other_credentials())
self.assertEquals(results['count'], 5) self.assertEquals(results['count'], 5)
@@ -227,7 +227,7 @@ class ProjectsTest(BaseTest):
project = reverse('main:project_detail', args=(self.projects[3].pk,)) project = reverse('main:project_detail', args=(self.projects[3].pk,))
self.get(project, expect=200, auth=self.get_super_credentials()) self.get(project, expect=200, auth=self.get_super_credentials())
self.get(project, expect=200, auth=self.get_normal_credentials()) self.get(project, expect=200, auth=self.get_normal_credentials())
self.get(project, expect=403, auth=self.get_other_credentials()) self.get(project, expect=200, auth=self.get_other_credentials())
self.get(project, expect=403, auth=self.get_nobody_credentials()) self.get(project, expect=403, auth=self.get_nobody_credentials())
# can delete projects # can delete projects
@@ -280,6 +280,9 @@ class ProjectsTest(BaseTest):
# can add teams # can add teams
posted1 = self.post(all_teams, data=new_team, expect=201, auth=self.get_super_credentials()) posted1 = self.post(all_teams, data=new_team, expect=201, auth=self.get_super_credentials())
posted2 = self.post(all_teams, data=new_team, expect=400, auth=self.get_super_credentials()) posted2 = self.post(all_teams, data=new_team, expect=400, auth=self.get_super_credentials())
# normal user is not an admin of organizations[0], but is for [1].
posted3 = self.post(all_teams, data=new_team2, expect=403, auth=self.get_normal_credentials())
new_team2['organization'] = self.organizations[1].pk
posted3 = self.post(all_teams, data=new_team2, expect=201, auth=self.get_normal_credentials()) posted3 = self.post(all_teams, data=new_team2, expect=201, auth=self.get_normal_credentials())
posted4 = self.post(all_teams, data=new_team2, expect=400, auth=self.get_normal_credentials()) posted4 = self.post(all_teams, data=new_team2, expect=400, auth=self.get_normal_credentials())
posted5 = self.post(all_teams, data=new_team3, expect=403, auth=self.get_other_credentials()) posted5 = self.post(all_teams, data=new_team3, expect=403, auth=self.get_other_credentials())
@@ -347,7 +350,7 @@ class ProjectsTest(BaseTest):
# ===================================================================== # =====================================================================
# TEAMS USER MEMBERSHIP # TEAMS USER MEMBERSHIP
team = Team.objects.filter(organization__pk=self.organizations[1].pk)[0] team = Team.objects.filter(active=True, organization__pk=self.organizations[1].pk)[0]
team_users = reverse('main:team_users_list', args=(team.pk,)) team_users = reverse('main:team_users_list', args=(team.pk,))
for x in team.users.all(): for x in team.users.all():
team.users.remove(x) team.users.remove(x)
@@ -361,13 +364,13 @@ class ProjectsTest(BaseTest):
self.get(team_users, expect=200, auth=self.get_normal_credentials()) self.get(team_users, expect=200, auth=self.get_normal_credentials())
self.get(team_users, expect=200, auth=self.get_super_credentials()) self.get(team_users, expect=200, auth=self.get_super_credentials())
# can add users to teams # can add users to teams (but only users I can see)
all_users = self.get(reverse('main:user_list'), expect=200, auth=self.get_super_credentials()) all_users = self.get(reverse('main:user_list'), expect=200, auth=self.get_normal_credentials())
for x in all_users['results']: for x in all_users['results']:
self.post(team_users, data=x, expect=403, auth=self.get_nobody_credentials()) self.post(team_users, data=x, expect=403, auth=self.get_nobody_credentials())
self.post(team_users, data=x, expect=204, auth=self.get_normal_credentials()) self.post(team_users, data=x, expect=204, auth=self.get_normal_credentials())
self.assertEqual(Team.objects.get(pk=team.pk).users.count(), 4) self.assertEqual(Team.objects.get(pk=team.pk).users.count(), 3)
# can remove users from teams # can remove users from teams
for x in all_users['results']: for x in all_users['results']:
@@ -492,7 +495,7 @@ class ProjectsTest(BaseTest):
self.put(edit_creds1, data=d_cred_user, expect=200, auth=self.get_normal_credentials()) self.put(edit_creds1, data=d_cred_user, expect=200, auth=self.get_normal_credentials())
# editing a credential to edit the user record is not legal, this is a test of the .validate # editing a credential to edit the user record is not legal, this is a test of the .validate
# method on the serializer to allow 'write once' fields # method on the serializer to allow 'write once' fields
self.put(edit_creds1, data=d_cred_user2, expect=400, auth=self.get_normal_credentials()) self.put(edit_creds1, data=d_cred_user2, expect=403, auth=self.get_normal_credentials())
cred_put_u = self.put(edit_creds1, data=d_cred_user, expect=200, auth=self.get_other_credentials()) cred_put_u = self.put(edit_creds1, data=d_cred_user, expect=200, auth=self.get_other_credentials())
self.put(edit_creds2, data=d_cred_team, expect=401) self.put(edit_creds2, data=d_cred_team, expect=401)

9
awx/main/views.py
View File

@@ -128,15 +128,14 @@ class ApiV1ConfigView(APIView):
license_data = license_reader.from_file() license_data = license_reader.from_file()
data = dict( data = dict(
time_zone = settings.TIME_ZONE, time_zone=settings.TIME_ZONE,
# FIXME: Special variables for inventory/group/host variable_data. license_info=license_data,
) )
if request.user.is_superuser or request.user.admin_of_organizations.filter(active=True).count(): if request.user.is_superuser or request.user.admin_of_organizations.filter(active=True).count():
data.update(dict( data.update(dict(
project_base_dir = settings.PROJECTS_ROOT, project_base_dir = settings.PROJECTS_ROOT,
project_local_paths = Project.get_local_path_choices(), project_local_paths = Project.get_local_path_choices(),
)) ))
data['license_info'] = license_data
return Response(data) return Response(data)
@@ -239,7 +238,7 @@ class TeamPermissionsList(SubListCreateAPIView):
parent_key = 'team' parent_key = 'team'
def get_queryset(self): def get_queryset(self):
# FIXME # FIXME: Default get_queryset should handle this.
team = Team.objects.get(pk=self.kwargs['pk']) team = Team.objects.get(pk=self.kwargs['pk'])
base = Permission.objects.filter(team = team) base = Permission.objects.filter(team = team)
#if Team.can_user_administrate(self.request.user, team, None): #if Team.can_user_administrate(self.request.user, team, None):
@@ -287,7 +286,7 @@ class ProjectOrganizationsList(SubListCreateAPIView):
relationship = 'organizations' relationship = 'organizations'
def get_queryset(self): def get_queryset(self):
# FIXME # FIXME: Default get_queryset should handle this.
project = Project.objects.get(pk=self.kwargs['pk']) project = Project.objects.get(pk=self.kwargs['pk'])
if not self.request.user.is_superuser: if not self.request.user.is_superuser:
raise PermissionDenied() raise PermissionDenied()