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

Merge pull request #6163 from wwitzel3/issue-5922

Browse Source 
Remove deprecated fields / old migration tests.
This commit is contained in:
Wayne Witzel III
2017-05-03 10:42:13 -04:00
committed by GitHub
parent 1b9bd8d9fb 9f28c57521
commit 465d620629
18 changed files with 103 additions and 1174 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
awx/main/access.py
View File

@@ -28,29 +28,6 @@ __all__ = ['get_user_queryset', 'check_user_access', 'check_user_access_with_err
'user_accessible_objects', 'consumer_access', 'user_accessible_objects', 'consumer_access',
'user_admin_role', 'StateConflict',] 'user_admin_role', 'StateConflict',]
PERMISSION_TYPES = [
PERM_INVENTORY_ADMIN,
PERM_INVENTORY_READ,
PERM_INVENTORY_WRITE,
PERM_INVENTORY_DEPLOY,
PERM_INVENTORY_CHECK,
]
PERMISSION_TYPES_ALLOWING_INVENTORY_READ = [
PERM_INVENTORY_ADMIN,
PERM_INVENTORY_WRITE,
PERM_INVENTORY_READ,
]
PERMISSION_TYPES_ALLOWING_INVENTORY_WRITE = [
PERM_INVENTORY_ADMIN,
PERM_INVENTORY_WRITE,
]
PERMISSION_TYPES_ALLOWING_INVENTORY_ADMIN = [
PERM_INVENTORY_ADMIN,
]
logger = logging.getLogger('awx.main.access') logger = logging.getLogger('awx.main.access')
access_registry = { access_registry = {

65
awx/main/migrations/0038_v320_release.py
View File

@@ -159,4 +159,69 @@ class Migration(migrations.Migration):
name='allow_simultaneous', name='allow_simultaneous',
field=models.BooleanField(default=False), field=models.BooleanField(default=False),
), ),
# Permission and Deprecated Field Removal
migrations.RemoveField(
model_name='permission',
name='created_by',
),
migrations.RemoveField(
model_name='permission',
name='inventory',
),
migrations.RemoveField(
model_name='permission',
name='modified_by',
),
migrations.RemoveField(
model_name='permission',
name='project',
),
migrations.RemoveField(
model_name='permission',
name='tags',
),
migrations.RemoveField(
model_name='permission',
name='team',
),
migrations.RemoveField(
model_name='permission',
name='user',
),
migrations.RemoveField(
model_name='activitystream',
name='permission',
),
migrations.RemoveField(
model_name='credential',
name='deprecated_team',
),
migrations.RemoveField(
model_name='credential',
name='deprecated_user',
),
migrations.RemoveField(
model_name='organization',
name='deprecated_admins',
),
migrations.RemoveField(
model_name='organization',
name='deprecated_projects',
),
migrations.RemoveField(
model_name='organization',
name='deprecated_users',
),
migrations.RemoveField(
model_name='team',
name='deprecated_projects',
),
migrations.RemoveField(
model_name='team',
name='deprecated_users',
),
migrations.DeleteModel(
name='Permission',
),
] ]

31
awx/main/migrations/_old_access.py
View File

@@ -27,6 +27,11 @@ from awx.conf.license import LicenseForbids
__all__ = ['get_user_queryset', 'check_user_access'] __all__ = ['get_user_queryset', 'check_user_access']
PERM_INVENTORY_ADMIN = 'admin'
PERM_INVENTORY_READ = 'read'
PERM_INVENTORY_WRITE = 'write'
PERM_JOBTEMPLATE_CREATE = 'create'
PERMISSION_TYPES = [ PERMISSION_TYPES = [
PERM_INVENTORY_ADMIN, PERM_INVENTORY_ADMIN,
PERM_INVENTORY_READ, PERM_INVENTORY_READ,
@@ -57,10 +62,12 @@ access_registry = {
# ... # ...
} }
def register_access(model_class, access_class): def register_access(model_class, access_class):
access_classes = access_registry.setdefault(model_class, []) access_classes = access_registry.setdefault(model_class, [])
access_classes.append(access_class) access_classes.append(access_class)
def get_user_queryset(user, model_class): def get_user_queryset(user, model_class):
''' '''
Return a queryset for the given model_class containing only the instances Return a queryset for the given model_class containing only the instances
@@ -80,6 +87,7 @@ def get_user_queryset(user, model_class):
queryset = queryset.filter(pk__in=qs.values_list('pk', flat=True)) queryset = queryset.filter(pk__in=qs.values_list('pk', flat=True))
return queryset return queryset
def check_user_access(user, model_class, action, *args, **kwargs): def check_user_access(user, model_class, action, *args, **kwargs):
''' '''
Return True if user can perform action against model_class with the Return True if user can perform action against model_class with the
@@ -243,6 +251,7 @@ class UserAccess(BaseAccess):
return bool(self.user.is_superuser or return bool(self.user.is_superuser or
obj.deprecated_organizations.filter(deprecated_admins__in=[self.user]).exists()) obj.deprecated_organizations.filter(deprecated_admins__in=[self.user]).exists())
class OrganizationAccess(BaseAccess): class OrganizationAccess(BaseAccess):
''' '''
I can see organizations when: I can see organizations when:
@@ -270,6 +279,7 @@ class OrganizationAccess(BaseAccess):
self.check_license(feature='multiple_organizations', check_expiration=False) self.check_license(feature='multiple_organizations', check_expiration=False)
return self.can_change(obj, None) return self.can_change(obj, None)
class InventoryAccess(BaseAccess): class InventoryAccess(BaseAccess):
''' '''
I can see inventory when: I can see inventory when:
@@ -365,6 +375,7 @@ class InventoryAccess(BaseAccess):
def can_run_ad_hoc_commands(self, obj): def can_run_ad_hoc_commands(self, obj):
return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_READ, True) return self.has_permission_types(obj, PERMISSION_TYPES_ALLOWING_INVENTORY_READ, True)
class HostAccess(BaseAccess): class HostAccess(BaseAccess):
''' '''
I can see hosts whenever I can see their inventory. I can see hosts whenever I can see their inventory.
@@ -421,6 +432,7 @@ class HostAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
return obj and check_user_access(self.user, Inventory, 'delete', obj.inventory) return obj and check_user_access(self.user, Inventory, 'delete', obj.inventory)
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.
@@ -517,6 +529,7 @@ class InventorySourceAccess(BaseAccess):
def can_start(self, obj): def can_start(self, obj):
return self.can_change(obj, {}) and obj.can_update return self.can_change(obj, {}) and obj.can_update
class InventoryUpdateAccess(BaseAccess): class InventoryUpdateAccess(BaseAccess):
''' '''
I can see inventory updates when I can see the inventory source. I can see inventory updates when I can see the inventory source.
@@ -536,6 +549,7 @@ class InventoryUpdateAccess(BaseAccess):
def can_cancel(self, obj): def can_cancel(self, obj):
return self.can_change(obj, {}) and obj.can_cancel return self.can_change(obj, {}) and obj.can_cancel
class CredentialAccess(BaseAccess): class CredentialAccess(BaseAccess):
''' '''
I can see credentials when: I can see credentials when:
@@ -615,6 +629,7 @@ class CredentialAccess(BaseAccess):
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 can see a team when:
@@ -662,6 +677,7 @@ class TeamAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
return self.can_change(obj, None) return self.can_change(obj, None)
class ProjectAccess(BaseAccess): class ProjectAccess(BaseAccess):
''' '''
I can see projects when: I can see projects when:
@@ -728,6 +744,7 @@ class ProjectAccess(BaseAccess):
def can_start(self, obj): def can_start(self, obj):
return self.can_change(obj, {}) and obj.can_update return self.can_change(obj, {}) and obj.can_update
class ProjectUpdateAccess(BaseAccess): class ProjectUpdateAccess(BaseAccess):
''' '''
I can see project updates when I can see the project. I can see project updates when I can see the project.
@@ -749,6 +766,7 @@ class ProjectUpdateAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
return obj and check_user_access(self.user, Project, 'delete', obj.project) return obj and check_user_access(self.user, Project, 'delete', obj.project)
class PermissionAccess(BaseAccess): class PermissionAccess(BaseAccess):
''' '''
I can see a permission when: I can see a permission when:
@@ -842,6 +860,7 @@ class PermissionAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
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 can see job templates when:
@@ -1068,6 +1087,7 @@ class JobTemplateAccess(BaseAccess):
job_type=obj.job_type) job_type=obj.job_type)
return self.can_add(add_obj) return self.can_add(add_obj)
class JobAccess(BaseAccess): class JobAccess(BaseAccess):
model = Job model = Job
@@ -1168,6 +1188,7 @@ class JobAccess(BaseAccess):
def can_cancel(self, obj): def can_cancel(self, obj):
return self.can_read(obj) and obj.can_cancel return self.can_read(obj) and obj.can_cancel
class SystemJobTemplateAccess(BaseAccess): class SystemJobTemplateAccess(BaseAccess):
''' '''
I can only see/manage System Job Templates if I'm a super user I can only see/manage System Job Templates if I'm a super user
@@ -1178,12 +1199,14 @@ class SystemJobTemplateAccess(BaseAccess):
def can_start(self, obj): def can_start(self, obj):
return self.can_read(obj) return self.can_read(obj)
class SystemJobAccess(BaseAccess): class SystemJobAccess(BaseAccess):
''' '''
I can only see manage System Jobs if I'm a super user I can only see manage System Jobs if I'm a super user
''' '''
model = SystemJob model = SystemJob
class AdHocCommandAccess(BaseAccess): class AdHocCommandAccess(BaseAccess):
''' '''
I can only see/run ad hoc commands when: I can only see/run ad hoc commands when:
@@ -1259,6 +1282,7 @@ class AdHocCommandAccess(BaseAccess):
def can_cancel(self, obj): def can_cancel(self, obj):
return self.can_read(obj) and obj.can_cancel return self.can_read(obj) and obj.can_cancel
class AdHocCommandEventAccess(BaseAccess): class AdHocCommandEventAccess(BaseAccess):
''' '''
I can see ad hoc command event records whenever I can read both ad hoc I can see ad hoc command event records whenever I can read both ad hoc
@@ -1288,6 +1312,7 @@ class AdHocCommandEventAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
return False return False
class JobHostSummaryAccess(BaseAccess): class JobHostSummaryAccess(BaseAccess):
''' '''
I can see job/host summary records whenever I can read both job and host. I can see job/host summary records whenever I can read both job and host.
@@ -1313,6 +1338,7 @@ class JobHostSummaryAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
return False return False
class JobEventAccess(BaseAccess): class JobEventAccess(BaseAccess):
''' '''
I can see job event records whenever I can read both job and host. I can see job event records whenever I can read both job and host.
@@ -1347,6 +1373,7 @@ class JobEventAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
return False return False
class UnifiedJobTemplateAccess(BaseAccess): class UnifiedJobTemplateAccess(BaseAccess):
''' '''
I can see a unified job template whenever I can see the same project, I can see a unified job template whenever I can see the same project,
@@ -1379,6 +1406,7 @@ class UnifiedJobTemplateAccess(BaseAccess):
# FIXME: Figure out how to do select/prefetch on related project/inventory/credential/cloud_credential. # FIXME: Figure out how to do select/prefetch on related project/inventory/credential/cloud_credential.
return qs return qs
class UnifiedJobAccess(BaseAccess): class UnifiedJobAccess(BaseAccess):
''' '''
I can see a unified job whenever I can see the same project update, I can see a unified job whenever I can see the same project update,
@@ -1417,6 +1445,7 @@ class UnifiedJobAccess(BaseAccess):
# FIXME: Figure out how to do select/prefetch on related project/inventory/credential/cloud_credential. # FIXME: Figure out how to do select/prefetch on related project/inventory/credential/cloud_credential.
return qs return qs
class ScheduleAccess(BaseAccess): class ScheduleAccess(BaseAccess):
''' '''
I can see a schedule if I can see it's related unified job, I can create them or update them if I have write access I can see a schedule if I can see it's related unified job, I can create them or update them if I have write access
@@ -1475,6 +1504,7 @@ class ScheduleAccess(BaseAccess):
else: else:
return False return False
class ActivityStreamAccess(BaseAccess): class ActivityStreamAccess(BaseAccess):
''' '''
I can see activity stream events only when I have permission on all objects included in the event I can see activity stream events only when I have permission on all objects included in the event
@@ -1592,6 +1622,7 @@ class ActivityStreamAccess(BaseAccess):
def can_delete(self, obj): def can_delete(self, obj):
return False return False
class CustomInventoryScriptAccess(BaseAccess): class CustomInventoryScriptAccess(BaseAccess):
model = CustomInventoryScript model = CustomInventoryScript

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

@@ -112,7 +112,6 @@ activity_stream_registrar.connect(Credential)
activity_stream_registrar.connect(Team) activity_stream_registrar.connect(Team)
activity_stream_registrar.connect(Project) activity_stream_registrar.connect(Project)
#activity_stream_registrar.connect(ProjectUpdate) #activity_stream_registrar.connect(ProjectUpdate)
activity_stream_registrar.connect(Permission)
activity_stream_registrar.connect(JobTemplate) activity_stream_registrar.connect(JobTemplate)
activity_stream_registrar.connect(Job) activity_stream_registrar.connect(Job)
activity_stream_registrar.connect(AdHocCommand) activity_stream_registrar.connect(AdHocCommand)

1
awx/main/models/activity_stream.py
View File

@@ -48,7 +48,6 @@ class ActivityStream(models.Model):
team = models.ManyToManyField("Team", blank=True) team = models.ManyToManyField("Team", blank=True)
project = models.ManyToManyField("Project", blank=True) project = models.ManyToManyField("Project", blank=True)
project_update = models.ManyToManyField("ProjectUpdate", blank=True) project_update = models.ManyToManyField("ProjectUpdate", blank=True)
permission = models.ManyToManyField("Permission", blank=True)
job_template = models.ManyToManyField("JobTemplate", blank=True) job_template = models.ManyToManyField("JobTemplate", blank=True)
job = models.ManyToManyField("Job", blank=True) job = models.ManyToManyField("Job", blank=True)
workflow_job_template_node = models.ManyToManyField("WorkflowJobTemplateNode", blank=True) workflow_job_template_node = models.ManyToManyField("WorkflowJobTemplateNode", blank=True)

21
awx/main/models/base.py
View File

@@ -26,20 +26,15 @@ from awx.main.utils import encrypt_field
__all__ = ['prevent_search', 'VarsDictProperty', 'BaseModel', 'CreatedModifiedModel', __all__ = ['prevent_search', 'VarsDictProperty', 'BaseModel', 'CreatedModifiedModel',
'PasswordFieldsModel', 'PrimordialModel', 'CommonModel', 'PasswordFieldsModel', 'PrimordialModel', 'CommonModel',
'CommonModelNameNotUnique', 'NotificationFieldsModel', 'CommonModelNameNotUnique', 'NotificationFieldsModel',
'PERM_INVENTORY_ADMIN', 'PERM_INVENTORY_READ', 'PERM_INVENTORY_DEPLOY', 'PERM_INVENTORY_SCAN',
'PERM_INVENTORY_WRITE', 'PERM_INVENTORY_DEPLOY', 'PERM_INVENTORY_SCAN', 'PERM_INVENTORY_CHECK', 'JOB_TYPE_CHOICES',
'PERM_INVENTORY_CHECK', 'PERM_JOBTEMPLATE_CREATE', 'JOB_TYPE_CHOICES',
'AD_HOC_JOB_TYPE_CHOICES', 'PROJECT_UPDATE_JOB_TYPE_CHOICES', 'AD_HOC_JOB_TYPE_CHOICES', 'PROJECT_UPDATE_JOB_TYPE_CHOICES',
'PERMISSION_TYPE_CHOICES', 'CLOUD_INVENTORY_SOURCES', 'CLOUD_INVENTORY_SOURCES',
'VERBOSITY_CHOICES'] 'VERBOSITY_CHOICES']
PERM_INVENTORY_ADMIN = 'admin'
PERM_INVENTORY_READ = 'read'
PERM_INVENTORY_WRITE = 'write'
PERM_INVENTORY_DEPLOY = 'run' PERM_INVENTORY_DEPLOY = 'run'
PERM_INVENTORY_CHECK = 'check' PERM_INVENTORY_CHECK = 'check'
PERM_INVENTORY_SCAN = 'scan' PERM_INVENTORY_SCAN = 'scan'
PERM_JOBTEMPLATE_CREATE = 'create'
JOB_TYPE_CHOICES = [ JOB_TYPE_CHOICES = [
(PERM_INVENTORY_DEPLOY, _('Run')), (PERM_INVENTORY_DEPLOY, _('Run')),
@@ -57,16 +52,6 @@ PROJECT_UPDATE_JOB_TYPE_CHOICES = [
(PERM_INVENTORY_CHECK, _('Check')), (PERM_INVENTORY_CHECK, _('Check')),
] ]
PERMISSION_TYPE_CHOICES = [
(PERM_INVENTORY_READ, _('Read Inventory')),
(PERM_INVENTORY_WRITE, _('Edit Inventory')),
(PERM_INVENTORY_ADMIN, _('Administrate Inventory')),
(PERM_INVENTORY_DEPLOY, _('Deploy To Inventory')),
(PERM_INVENTORY_CHECK, _('Deploy To Inventory (Dry Run)')),
(PERM_INVENTORY_SCAN, _('Scan an Inventory')),
(PERM_JOBTEMPLATE_CREATE, _('Create a Job Template')),
]
CLOUD_INVENTORY_SOURCES = ['ec2', 'rax', 'vmware', 'gce', 'azure', 'azure_rm', 'openstack', 'custom', 'satellite6', 'cloudforms'] CLOUD_INVENTORY_SOURCES = ['ec2', 'rax', 'vmware', 'gce', 'azure', 'azure_rm', 'openstack', 'custom', 'satellite6', 'cloudforms']
VERBOSITY_CHOICES = [ VERBOSITY_CHOICES = [

45
awx/main/models/credential.py
View File

@@ -215,22 +215,6 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
PASSWORD_FIELDS = ['inputs'] PASSWORD_FIELDS = ['inputs']
deprecated_user = models.ForeignKey(
'auth.User',
null=True,
default=None,
blank=True,
on_delete=models.CASCADE,
related_name='deprecated_credentials',
)
deprecated_team = models.ForeignKey(
'Team',
null=True,
default=None,
blank=True,
on_delete=models.CASCADE,
related_name='deprecated_credentials',
)
credential_type = models.ForeignKey( credential_type = models.ForeignKey(
'CredentialType', 'CredentialType',
related_name='credentials', related_name='credentials',
@@ -391,40 +375,15 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
'SSH key is not encrypted.')) 'SSH key is not encrypted.'))
return self.ssh_key_unlock return self.ssh_key_unlock
def clean(self):
if self.deprecated_user and self.deprecated_team:
raise ValidationError(_('Credential cannot be assigned to both a user and team.'))
def _password_field_allows_ask(self, field): def _password_field_allows_ask(self, field):
return field in self.credential_type.askable_fields return field in self.credential_type.askable_fields
def save(self, *args, **kwargs): def save(self, *args, **kwargs):
inputs_before = {}
# If update_fields has been specified, add our field names to it,
# if hit hasn't been specified, then we're just doing a normal save.
update_fields = kwargs.get('update_fields', [])
# If updating a credential, make sure that we only allow user OR team
# to be set, and clear out the other field based on which one has
# changed.
if self.pk:
cred_before = Credential.objects.get(pk=self.pk)
if self.deprecated_user and self.deprecated_team:
# If the user changed, remove the previously assigned team.
if cred_before.user != self.user:
self.deprecated_team = None
if 'deprecated_team' not in update_fields:
update_fields.append('deprecated_team')
# If the team changed, remove the previously assigned user.
elif cred_before.deprecated_team != self.deprecated_team:
self.deprecated_user = None
if 'deprecated_user' not in update_fields:
update_fields.append('deprecated_user')
inputs_before = cred_before.inputs
self.PASSWORD_FIELDS = self.credential_type.secret_fields self.PASSWORD_FIELDS = self.credential_type.secret_fields
if self.pk: if self.pk:
cred_before = Credential.objects.get(pk=self.pk)
inputs_before = cred_before.inputs
# Look up the currently persisted value so that we can replace # Look up the currently persisted value so that we can replace
# $encrypted$ with the actual DB-backed value # $encrypted$ with the actual DB-backed value
for field in self.PASSWORD_FIELDS: for field in self.PASSWORD_FIELDS:

85
awx/main/models/organization.py
View File

@@ -24,7 +24,7 @@ from awx.main.models.rbac import (
) )
from awx.main.models.mixins import ResourceMixin from awx.main.models.mixins import ResourceMixin
__all__ = ['Organization', 'Team', 'Permission', 'Profile', 'AuthToken'] __all__ = ['Organization', 'Team', 'Profile', 'AuthToken']
class Organization(CommonModel, NotificationFieldsModel, ResourceMixin): class Organization(CommonModel, NotificationFieldsModel, ResourceMixin):
@@ -36,21 +36,6 @@ class Organization(CommonModel, NotificationFieldsModel, ResourceMixin):
app_label = 'main' app_label = 'main'
ordering = ('name',) ordering = ('name',)
deprecated_users = models.ManyToManyField(
'auth.User',
blank=True,
related_name='deprecated_organizations',
)
deprecated_admins = models.ManyToManyField(
'auth.User',
blank=True,
related_name='deprecated_admin_of_organizations',
)
deprecated_projects = models.ManyToManyField(
'Project',
blank=True,
related_name='deprecated_organizations',
)
admin_role = ImplicitRoleField( admin_role = ImplicitRoleField(
parent_role='singleton:' + ROLE_SINGLETON_SYSTEM_ADMINISTRATOR, parent_role='singleton:' + ROLE_SINGLETON_SYSTEM_ADMINISTRATOR,
) )
@@ -82,11 +67,6 @@ class Team(CommonModelNameNotUnique, ResourceMixin):
unique_together = [('organization', 'name')] unique_together = [('organization', 'name')]
ordering = ('organization__name', 'name') ordering = ('organization__name', 'name')
deprecated_users = models.ManyToManyField(
'auth.User',
blank=True,
related_name='deprecated_teams',
)
organization = models.ForeignKey( organization = models.ForeignKey(
'Organization', 'Organization',
blank=False, blank=False,
@@ -94,11 +74,6 @@ class Team(CommonModelNameNotUnique, ResourceMixin):
on_delete=models.CASCADE, on_delete=models.CASCADE,
related_name='teams', related_name='teams',
) )
deprecated_projects = models.ManyToManyField(
'Project',
blank=True,
related_name='deprecated_teams',
)
admin_role = ImplicitRoleField( admin_role = ImplicitRoleField(
parent_role='organization.admin_role', parent_role='organization.admin_role',
) )
@@ -113,64 +88,6 @@ class Team(CommonModelNameNotUnique, ResourceMixin):
return reverse('api:team_detail', kwargs={'pk': self.pk}, request=request) return reverse('api:team_detail', kwargs={'pk': self.pk}, request=request)
class Permission(CommonModelNameNotUnique):
'''
A permission allows a user, project, or team to be able to use an inventory source.
NOTE: This class is deprecated, permissions and access is to be handled by
our new RBAC system. This class should be able to be safely removed after a 3.0.0
migration. - anoek 2016-01-28
'''
class Meta:
app_label = 'main'
# permissions are granted to either a user or a team:
user = models.ForeignKey('auth.User', null=True, on_delete=models.SET_NULL, blank=True, related_name='permissions')
team = models.ForeignKey('Team', null=True, on_delete=models.SET_NULL, blank=True, related_name='permissions')
# to be used against a project or inventory (or a project and inventory in conjunction):
project = models.ForeignKey(
'Project',
blank=True,
null=True,
on_delete=models.SET_NULL,
related_name='permissions',
)
inventory = models.ForeignKey('Inventory', null=True, on_delete=models.SET_NULL, related_name='permissions')
# permission system explanation:
#
# for example, user A on inventory X has write permissions (PERM_INVENTORY_WRITE)
# team C on inventory X has read permissions (PERM_INVENTORY_READ)
# user A can create job templates (PERM_JOBTEMPLATE_CREATE)
# team C on inventory X and project Y has launch permissions (PERM_INVENTORY_DEPLOY)
# team C on inventory X and project Z has dry run permissions (PERM_INVENTORY_CHECK)
#
# basically for launching, permissions can be awarded to the whole inventory source or just the inventory source
# in context of a given project.
#
# the project parameter is not used when dealing with READ, WRITE, or ADMIN permissions.
permission_type = models.CharField(max_length=64, choices=PERMISSION_TYPE_CHOICES)
run_ad_hoc_commands = models.BooleanField(default=False,
help_text=_('Execute Commands on the Inventory'))
def __unicode__(self):
return unicode("Permission(name=%s,ON(user=%s,team=%s),FOR(project=%s,inventory=%s,type=%s%s))" % (
self.name,
self.user,
self.team,
self.project,
self.inventory,
self.permission_type,
'+adhoc' if self.run_ad_hoc_commands else '',
))
def get_absolute_url(self, request=None):
return reverse('api:permission_detail', kwargs={'pk': self.pk}, request=request)
class Profile(CreatedModifiedModel): class Profile(CreatedModifiedModel):
''' '''
Profile model related to User object. Currently stores LDAP DN for users Profile model related to User object. Currently stores LDAP DN for users

7
awx/main/tests/functional/conftest.py
View File

@@ -19,7 +19,6 @@ from jsonbfield.fields import JSONField
# AWX # AWX
from awx.main.models.projects import Project from awx.main.models.projects import Project
from awx.main.models.base import PERM_INVENTORY_READ
from awx.main.models.ha import Instance from awx.main.models.ha import Instance
from awx.main.models.fact import Fact from awx.main.models.fact import Fact
@@ -38,7 +37,6 @@ from awx.main.models.inventory import (
) )
from awx.main.models.organization import ( from awx.main.models.organization import (
Organization, Organization,
Permission,
Team, Team,
) )
from awx.main.models.rbac import Role from awx.main.models.rbac import Role
@@ -528,11 +526,6 @@ def fact_services_json():
return _fact_json('services') return _fact_json('services')
@pytest.fixture
def permission_inv_read(organization, inventory, team):
return Permission.objects.create(inventory=inventory, team=team, permission_type=PERM_INVENTORY_READ)
@pytest.fixture @pytest.fixture
def job_template(organization): def job_template(organization):
jt = JobTemplate(name='test-job_template') jt = JobTemplate(name='test-job_template')

18
awx/main/tests/functional/test_inventory_source_migration.py
View File

@@ -25,21 +25,3 @@ def test_inv_src_rename(inventory_source_factory):
inv_src01.refresh_from_db() inv_src01.refresh_from_db()
# inv-is-t1 is generated in the inventory_source_factory # inv-is-t1 is generated in the inventory_source_factory
assert inv_src01.name == 't1 - inv-is-t1 - 0' assert inv_src01.name == 't1 - inv-is-t1 - 0'
@pytest.mark.django_db
def test_inv_src_nolink_removal(inventory_source_factory):
inventory_source_factory('t1')
inv_src02 = inventory_source_factory('t2')
inv_src02.inventory = None
inv_src02.deprecated_group = None
inv_src02.save()
assert InventorySource.objects.count() == 2
invsrc.remove_inventory_source_with_no_inventory_link(apps, None)
objs = InventorySource.objects.all()
assert len(objs) == 1
assert 't1' in objs[0].name

259
awx/main/tests/functional/test_rbac_credential.py
View File

@@ -2,40 +2,9 @@ import pytest
from awx.main.access import CredentialAccess from awx.main.access import CredentialAccess
from awx.main.models.credential import Credential from awx.main.models.credential import Credential
from awx.main.models.jobs import JobTemplate
from awx.main.models.inventory import InventorySource
from awx.main.migrations import _rbac as rbac
from django.apps import apps
from django.contrib.auth.models import User from django.contrib.auth.models import User
@pytest.mark.django_db
def test_credential_migration_user(credential, user, permissions):
u = user('user', False)
credential.deprecated_user = u
credential.save()
rbac.migrate_credential(apps, None)
assert u in credential.admin_role
@pytest.mark.django_db
def test_two_teams_same_cred_name(organization_factory, credentialtype_net):
objects = organization_factory("test",
teams=["team1", "team2"])
cred1 = Credential.objects.create(name="test", credential_type=credentialtype_net, deprecated_team=objects.teams.team1)
cred2 = Credential.objects.create(name="test", credential_type=credentialtype_net, deprecated_team=objects.teams.team2)
rbac.migrate_credential(apps, None)
assert objects.teams.team1.admin_role in cred1.admin_role.parents.all()
assert objects.teams.team2.admin_role in cred2.admin_role.parents.all()
assert objects.teams.team1.member_role in cred1.use_role.parents.all()
assert objects.teams.team2.member_role in cred2.use_role.parents.all()
@pytest.mark.django_db @pytest.mark.django_db
def test_credential_use_role(credential, user, permissions): def test_credential_use_role(credential, user, permissions):
u = user('user', False) u = user('user', False)
@@ -43,59 +12,6 @@ def test_credential_use_role(credential, user, permissions):
assert u in credential.use_role assert u in credential.use_role
@pytest.mark.django_db
def test_credential_migration_team_member(credential, team, user, permissions):
u = user('user', False)
team.member_role.members.add(u)
credential.deprecated_team = team
credential.save()
# No permissions pre-migration (this happens automatically so we patch this)
team.admin_role.children.remove(credential.admin_role)
team.member_role.children.remove(credential.use_role)
assert u not in credential.admin_role
rbac.migrate_credential(apps, None)
# User permissions post migration
assert u in credential.use_role
assert u not in credential.admin_role
@pytest.mark.django_db
def test_credential_migration_team_admin(credential, team, user, permissions):
u = user('user', False)
team.admin_role.members.add(u)
credential.deprecated_team = team
credential.save()
assert u not in credential.use_role
# Admin permissions post migration
rbac.migrate_credential(apps, None)
assert u in credential.admin_role
@pytest.mark.django_db
def test_credential_migration_org_auditor(credential, team, org_auditor):
# Team's organization is the org_auditor's org
credential.deprecated_team = team
credential.save()
# No permissions pre-migration (this happens automatically so we patch this)
team.admin_role.children.remove(credential.admin_role)
team.member_role.children.remove(credential.use_role)
assert org_auditor not in credential.read_role
rbac.migrate_credential(apps, None)
rbac.infer_credential_org_from_team(apps, None)
# Read permissions post migration
assert org_auditor not in credential.use_role
assert org_auditor in credential.read_role
def test_credential_access_superuser(): def test_credential_access_superuser():
u = User(username='admin', is_superuser=True) u = User(username='admin', is_superuser=True)
access = CredentialAccess(u) access = CredentialAccess(u)
@@ -118,33 +34,6 @@ def test_credential_access_auditor(credential, organization_factory):
assert access.can_read(credential) assert access.can_read(credential)
@pytest.mark.django_db
def test_credential_access_admin(user, team, credential, credentialtype_aws):
u = user('org-admin', False)
team.organization.admin_role.members.add(u)
access = CredentialAccess(u)
assert access.can_add({'user': u.pk})
assert not access.can_change(credential, {'user': u.pk})
# unowned credential is superuser only
assert not access.can_delete(credential)
# credential is now part of a team
# that is part of an organization
# that I am an admin for
credential.admin_role.parents.add(team.admin_role)
credential.save()
cred = Credential.objects.create(credential_type=credentialtype_aws, name='test-cred')
cred.deprecated_team = team
cred.save()
# should have can_change access as org-admin
assert access.can_change(credential, {'description': 'New description.'})
@pytest.mark.django_db @pytest.mark.django_db
def test_org_credential_access_member(alice, org_credential, credential): def test_org_credential_access_member(alice, org_credential, credential):
org_credential.admin_role.members.add(alice) org_credential.admin_role.members.add(alice)
@@ -163,156 +52,8 @@ def test_org_credential_access_member(alice, org_credential, credential):
'organization': None}) 'organization': None})
@pytest.mark.django_db
def test_cred_job_template_xfail(user, deploy_jobtemplate):
' Personal credential migration '
a = user('admin', False)
org = deploy_jobtemplate.project.organization
org.admin_role.members.add(a)
cred = deploy_jobtemplate.credential
cred.deprecated_user = user('john', False)
cred.save()
access = CredentialAccess(a)
rbac.migrate_credential(apps, None)
assert not access.can_change(cred, {'organization': org.pk})
@pytest.mark.django_db
def test_cred_job_template(user, team, deploy_jobtemplate):
' Team credential migration => org credential '
a = user('admin', False)
org = deploy_jobtemplate.project.organization
org.admin_role.members.add(a)
cred = deploy_jobtemplate.credential
cred.deprecated_team = team
cred.save()
access = CredentialAccess(a)
rbac.migrate_credential(apps, None)
cred.refresh_from_db()
assert access.can_change(cred, {'organization': org.pk})
org.admin_role.members.remove(a)
assert not access.can_change(cred, {'organization': org.pk})
@pytest.mark.django_db
def test_cred_multi_job_template_single_org_xfail(user, deploy_jobtemplate):
a = user('admin', False)
org = deploy_jobtemplate.project.organization
org.admin_role.members.add(a)
cred = deploy_jobtemplate.credential
cred.deprecated_user = user('john', False)
cred.save()
access = CredentialAccess(a)
rbac.migrate_credential(apps, None)
cred.refresh_from_db()
assert not access.can_change(cred, {'organization': org.pk})
@pytest.mark.django_db
def test_cred_multi_job_template_single_org(user, team, deploy_jobtemplate):
a = user('admin', False)
org = deploy_jobtemplate.project.organization
org.admin_role.members.add(a)
cred = deploy_jobtemplate.credential
cred.deprecated_team = team
cred.save()
access = CredentialAccess(a)
rbac.migrate_credential(apps, None)
cred.refresh_from_db()
assert access.can_change(cred, {'organization': org.pk})
org.admin_role.members.remove(a)
assert not access.can_change(cred, {'organization': org.pk})
@pytest.mark.django_db
def test_single_cred_multi_job_template_multi_org(user, organizations, credential, team):
orgs = organizations(2)
credential.deprecated_team = team
credential.save()
jts = []
for org in orgs:
inv = org.inventories.create(name="inv-%d" % org.pk)
jt = JobTemplate.objects.create(
inventory=inv,
credential=credential,
name="test-jt-org-%d" % org.pk,
job_type='check',
)
jts.append(jt)
a = user('admin', False)
orgs[0].admin_role.members.add(a)
orgs[1].admin_role.members.add(a)
rbac.migrate_credential(apps, None)
for jt in jts:
jt.refresh_from_db()
credential.refresh_from_db()
assert jts[0].credential != jts[1].credential
@pytest.mark.django_db
def test_cred_inventory_source(user, inventory, credential):
u = user('member', False)
inventory.organization.member_role.members.add(u)
InventorySource.objects.create(
name="test-inv-src",
credential=credential,
inventory=inventory,
)
assert u not in credential.use_role
rbac.migrate_credential(apps, None)
assert u not in credential.use_role
@pytest.mark.django_db
def test_cred_project(user, credential, project):
u = user('member', False)
project.organization.member_role.members.add(u)
project.credential = credential
project.save()
assert u not in credential.use_role
rbac.migrate_credential(apps, None)
assert u not in credential.use_role
@pytest.mark.django_db @pytest.mark.django_db
def test_cred_no_org(user, credential): def test_cred_no_org(user, credential):
su = user('su', True) su = user('su', True)
access = CredentialAccess(su) access = CredentialAccess(su)
assert access.can_change(credential, {'user': su.pk}) assert access.can_change(credential, {'user': su.pk})
@pytest.mark.django_db
def test_cred_team(user, team, credential):
u = user('a', False)
team.member_role.members.add(u)
credential.deprecated_team = team
credential.save()
assert u not in credential.use_role
rbac.migrate_credential(apps, None)
assert u in credential.use_role

155
awx/main/tests/functional/test_rbac_inventory.py
View File

@@ -1,8 +1,6 @@
import pytest import pytest
from awx.main.migrations import _rbac as rbac
from awx.main.models import ( from awx.main.models import (
Permission,
Host, Host,
CustomInventoryScript, CustomInventoryScript,
Schedule Schedule
@@ -15,7 +13,6 @@ from awx.main.access import (
CustomInventoryScriptAccess, CustomInventoryScriptAccess,
ScheduleAccess ScheduleAccess
) )
from django.apps import apps
@pytest.mark.django_db @pytest.mark.django_db
@@ -54,158 +51,6 @@ def test_org_member_inventory_script_permissions(org_member, organization):
assert not access.can_change(custom_inv, {'name': 'ed-test'}) assert not access.can_change(custom_inv, {'name': 'ed-test'})
@pytest.mark.django_db
def test_inventory_admin_user(inventory, permissions, user):
u = user('admin', False)
perm = Permission(user=u, inventory=inventory, permission_type='admin')
perm.save()
assert u not in inventory.admin_role
rbac.migrate_inventory(apps, None)
assert u in inventory.admin_role
assert inventory.use_role.members.filter(id=u.id).exists() is False
assert inventory.update_role.members.filter(id=u.id).exists() is False
@pytest.mark.django_db
def test_inventory_auditor_user(inventory, permissions, user):
u = user('auditor', False)
perm = Permission(user=u, inventory=inventory, permission_type='read')
perm.save()
assert u not in inventory.admin_role
assert u not in inventory.read_role
rbac.migrate_inventory(apps, None)
assert u not in inventory.admin_role
assert u in inventory.read_role
assert inventory.use_role.members.filter(id=u.id).exists() is False
assert inventory.update_role.members.filter(id=u.id).exists() is False
@pytest.mark.django_db
def test_inventory_updater_user(inventory, permissions, user):
u = user('updater', False)
perm = Permission(user=u, inventory=inventory, permission_type='write')
perm.save()
assert u not in inventory.admin_role
assert u not in inventory.read_role
rbac.migrate_inventory(apps, None)
assert u not in inventory.admin_role
assert inventory.use_role.members.filter(id=u.id).exists() is False
assert inventory.update_role.members.filter(id=u.id).exists()
@pytest.mark.django_db
def test_inventory_executor_user(inventory, permissions, user):
u = user('executor', False)
perm = Permission(user=u, inventory=inventory, permission_type='read', run_ad_hoc_commands=True)
perm.save()
assert u not in inventory.admin_role
assert u not in inventory.read_role
rbac.migrate_inventory(apps, None)
assert u not in inventory.admin_role
assert u in inventory.read_role
assert inventory.use_role.members.filter(id=u.id).exists()
assert inventory.update_role.members.filter(id=u.id).exists() is False
@pytest.mark.django_db
def test_inventory_admin_team(inventory, permissions, user, team):
u = user('admin', False)
perm = Permission(team=team, inventory=inventory, permission_type='admin')
perm.save()
team.deprecated_users.add(u)
assert u not in inventory.admin_role
rbac.migrate_team(apps, None)
rbac.migrate_inventory(apps, None)
assert team.member_role.members.count() == 1
assert inventory.admin_role.members.filter(id=u.id).exists() is False
assert inventory.read_role.members.filter(id=u.id).exists() is False
assert inventory.use_role.members.filter(id=u.id).exists() is False
assert inventory.update_role.members.filter(id=u.id).exists() is False
assert u in inventory.read_role
assert u in inventory.admin_role
@pytest.mark.django_db
def test_inventory_auditor(inventory, permissions, user, team):
u = user('auditor', False)
perm = Permission(team=team, inventory=inventory, permission_type='read')
perm.save()
team.deprecated_users.add(u)
assert u not in inventory.admin_role
assert u not in inventory.read_role
rbac.migrate_team(apps,None)
rbac.migrate_inventory(apps, None)
assert team.member_role.members.count() == 1
assert inventory.admin_role.members.filter(id=u.id).exists() is False
assert inventory.read_role.members.filter(id=u.id).exists() is False
assert inventory.use_role.members.filter(id=u.id).exists() is False
assert inventory.update_role.members.filter(id=u.id).exists() is False
assert u in inventory.read_role
assert u not in inventory.admin_role
@pytest.mark.django_db
def test_inventory_updater(inventory, permissions, user, team):
u = user('updater', False)
perm = Permission(team=team, inventory=inventory, permission_type='write')
perm.save()
team.deprecated_users.add(u)
assert u not in inventory.admin_role
assert u not in inventory.read_role
rbac.migrate_team(apps,None)
rbac.migrate_inventory(apps, None)
assert team.member_role.members.count() == 1
assert inventory.admin_role.members.filter(id=u.id).exists() is False
assert inventory.read_role.members.filter(id=u.id).exists() is False
assert inventory.use_role.members.filter(id=u.id).exists() is False
assert inventory.update_role.members.filter(id=u.id).exists() is False
assert team.member_role.is_ancestor_of(inventory.update_role)
assert team.member_role.is_ancestor_of(inventory.use_role) is False
@pytest.mark.django_db
def test_inventory_executor(inventory, permissions, user, team):
u = user('executor', False)
perm = Permission(team=team, inventory=inventory, permission_type='read', run_ad_hoc_commands=True)
perm.save()
team.deprecated_users.add(u)
assert u not in inventory.admin_role
assert u not in inventory.read_role
rbac.migrate_team(apps, None)
rbac.migrate_inventory(apps, None)
assert team.member_role.members.count() == 1
assert inventory.admin_role.members.filter(id=u.id).exists() is False
assert inventory.read_role.members.filter(id=u.id).exists() is False
assert inventory.use_role.members.filter(id=u.id).exists() is False
assert inventory.update_role.members.filter(id=u.id).exists() is False
assert team.member_role.is_ancestor_of(inventory.update_role) is False
assert team.member_role.is_ancestor_of(inventory.use_role)
@pytest.mark.django_db @pytest.mark.django_db
def test_access_admin(organization, inventory, user): def test_access_admin(organization, inventory, user):
a = user('admin', False) a = user('admin', False)

140
awx/main/tests/functional/test_rbac_job_templates.py
View File

@@ -7,12 +7,8 @@ from awx.main.access import (
JobTemplateAccess, JobTemplateAccess,
ScheduleAccess ScheduleAccess
) )
from awx.main.migrations import _rbac as rbac
from awx.main.models import Permission
from awx.main.models.jobs import JobTemplate from awx.main.models.jobs import JobTemplate
from awx.main.models.schedules import Schedule from awx.main.models.schedules import Schedule
from django.apps import apps
@pytest.fixture @pytest.fixture
@@ -23,142 +19,6 @@ def jt_objects(job_template_factory):
return objects return objects
@pytest.mark.django_db
def test_job_template_migration_check(credential, deploy_jobtemplate, check_jobtemplate, user):
admin = user('admin', is_superuser=True)
joe = user('joe')
credential.deprecated_user = joe
credential.save()
check_jobtemplate.project.organization.deprecated_users.add(joe)
Permission(user=joe, inventory=check_jobtemplate.inventory, permission_type='read').save()
Permission(user=joe, inventory=check_jobtemplate.inventory,
project=check_jobtemplate.project, permission_type='check').save()
rbac.migrate_users(apps, None)
rbac.migrate_organization(apps, None)
rbac.migrate_projects(apps, None)
rbac.migrate_inventory(apps, None)
assert joe in check_jobtemplate.project.read_role
assert admin in check_jobtemplate.execute_role
assert joe not in check_jobtemplate.execute_role
rbac.migrate_job_templates(apps, None)
assert admin in check_jobtemplate.execute_role
assert joe in check_jobtemplate.execute_role
assert admin in deploy_jobtemplate.execute_role
assert joe not in deploy_jobtemplate.execute_role
@pytest.mark.django_db
def test_job_template_migration_deploy(credential, deploy_jobtemplate, check_jobtemplate, user):
admin = user('admin', is_superuser=True)
joe = user('joe')
credential.deprecated_user = joe
credential.save()
deploy_jobtemplate.project.organization.deprecated_users.add(joe)
Permission(user=joe, inventory=deploy_jobtemplate.inventory, permission_type='read').save()
Permission(user=joe, inventory=deploy_jobtemplate.inventory,
project=deploy_jobtemplate.project, permission_type='run').save()
rbac.migrate_users(apps, None)
rbac.migrate_organization(apps, None)
rbac.migrate_projects(apps, None)
rbac.migrate_inventory(apps, None)
assert joe in deploy_jobtemplate.project.read_role
assert admin in deploy_jobtemplate.execute_role
assert joe not in deploy_jobtemplate.execute_role
rbac.migrate_job_templates(apps, None)
assert admin in deploy_jobtemplate.execute_role
assert joe in deploy_jobtemplate.execute_role
assert admin in check_jobtemplate.execute_role
assert joe in check_jobtemplate.execute_role
@pytest.mark.django_db
def test_job_template_team_migration_check(credential, deploy_jobtemplate, check_jobtemplate, organization, team, user):
admin = user('admin', is_superuser=True)
joe = user('joe')
team.deprecated_users.add(joe)
team.organization = organization
team.save()
credential.deprecated_team = team
credential.save()
check_jobtemplate.project.organization.deprecated_users.add(joe)
Permission(team=team, inventory=check_jobtemplate.inventory, permission_type='read').save()
Permission(team=team, inventory=check_jobtemplate.inventory,
project=check_jobtemplate.project, permission_type='check').save()
rbac.migrate_users(apps, None)
rbac.migrate_team(apps, None)
rbac.migrate_organization(apps, None)
rbac.migrate_projects(apps, None)
rbac.migrate_inventory(apps, None)
assert joe not in check_jobtemplate.read_role
assert admin in check_jobtemplate.execute_role
assert joe not in check_jobtemplate.execute_role
rbac.migrate_job_templates(apps, None)
assert admin in check_jobtemplate.execute_role
assert joe in check_jobtemplate.execute_role
assert admin in deploy_jobtemplate.execute_role
assert joe not in deploy_jobtemplate.execute_role
@pytest.mark.django_db
def test_job_template_team_deploy_migration(credential, deploy_jobtemplate, check_jobtemplate, organization, team, user):
admin = user('admin', is_superuser=True)
joe = user('joe')
team.deprecated_users.add(joe)
team.organization = organization
team.save()
credential.deprecated_team = team
credential.save()
deploy_jobtemplate.project.organization.deprecated_users.add(joe)
Permission(team=team, inventory=deploy_jobtemplate.inventory, permission_type='read').save()
Permission(team=team, inventory=deploy_jobtemplate.inventory,
project=deploy_jobtemplate.project, permission_type='run').save()
rbac.migrate_users(apps, None)
rbac.migrate_team(apps, None)
rbac.migrate_organization(apps, None)
rbac.migrate_projects(apps, None)
rbac.migrate_inventory(apps, None)
assert joe not in deploy_jobtemplate.read_role
assert admin in deploy_jobtemplate.execute_role
assert joe not in deploy_jobtemplate.execute_role
rbac.migrate_job_templates(apps, None)
assert joe in deploy_jobtemplate.read_role
assert admin in deploy_jobtemplate.execute_role
assert joe in deploy_jobtemplate.execute_role
assert admin in check_jobtemplate.execute_role
assert joe in check_jobtemplate.execute_role
@mock.patch.object(BaseAccess, 'check_license', return_value=None) @mock.patch.object(BaseAccess, 'check_license', return_value=None)
@pytest.mark.django_db @pytest.mark.django_db
def test_job_template_access_superuser(check_license, user, deploy_jobtemplate): def test_job_template_access_superuser(check_license, user, deploy_jobtemplate):

44
awx/main/tests/functional/test_rbac_organization.py
View File

@@ -1,54 +1,10 @@
import mock import mock
import pytest import pytest
from awx.main.migrations import _rbac as rbac
from awx.main.access import ( from awx.main.access import (
BaseAccess, BaseAccess,
OrganizationAccess, OrganizationAccess,
) )
from django.apps import apps
@pytest.mark.django_db
def test_organization_migration_admin(organization, permissions, user):
u = user('admin', False)
organization.deprecated_admins.add(u)
# Undo some automatic work that we're supposed to be testing with our migration
organization.admin_role.members.remove(u)
assert u not in organization.admin_role
rbac.migrate_organization(apps, None)
assert u in organization.admin_role
@pytest.mark.django_db
def test_organization_migration_user(organization, permissions, user):
u = user('user', False)
organization.deprecated_users.add(u)
# Undo some automatic work that we're supposed to be testing with our migration
organization.member_role.members.remove(u)
assert u not in organization.read_role
rbac.migrate_organization(apps, None)
assert u in organization.read_role
@mock.patch.object(BaseAccess, 'check_license', return_value=None)
@pytest.mark.django_db
def test_organization_access_superuser(cl, organization, user):
access = OrganizationAccess(user('admin', True))
organization.deprecated_users.add(user('user', False))
assert access.can_change(organization, None)
assert access.can_delete(organization)
org = access.get_queryset()[0]
assert len(org.deprecated_admins.all()) == 0
assert len(org.deprecated_users.all()) == 1
@mock.patch.object(BaseAccess, 'check_license', return_value=None) @mock.patch.object(BaseAccess, 'check_license', return_value=None)

235
awx/main/tests/functional/test_rbac_project.py
View File

@@ -1,235 +0,0 @@
import pytest
from awx.main.migrations import _rbac as rbac
from awx.main.models import Role, Permission, Project, Organization, Credential, JobTemplate, Inventory
from awx.main.access import ProjectAccess
from django.apps import apps
from awx.main.migrations import _old_access as old_access
@pytest.mark.django_db
def test_project_migration(credentialtype_ssh):
'''
o1 o2 o3 with o1 -- i1 o2 -- i2
\ | /
\ | /
c1 ---- p1
/ | \
/ | \
jt1 jt2 jt3
| | |
i1 i2 i1
goes to
o1
|
|
c1 ---- p1
/ |
/ |
jt1 jt3
| |
i1 i1
o2
|
|
c1 ---- p2
|
|
jt2
|
i2
o3
|
|
c1 ---- p3
'''
o1 = Organization.objects.create(name='o1')
o2 = Organization.objects.create(name='o2')
o3 = Organization.objects.create(name='o3')
c1 = Credential.objects.create(name='c1', credential_type=credentialtype_ssh)
project_name = unicode("\xc3\xb4", "utf-8")
p1 = Project.objects.create(name=project_name, credential=c1)
p1.deprecated_organizations.add(o1, o2, o3)
i1 = Inventory.objects.create(name='i1', organization=o1)
i2 = Inventory.objects.create(name='i2', organization=o2)
jt1 = JobTemplate.objects.create(name='jt1', project=p1, inventory=i1)
jt2 = JobTemplate.objects.create(name='jt2', project=p1, inventory=i2)
jt3 = JobTemplate.objects.create(name='jt3', project=p1, inventory=i1)
assert o1.projects.count() == 0
assert o2.projects.count() == 0
assert o3.projects.count() == 0
rbac.migrate_projects(apps, None)
jt1 = JobTemplate.objects.get(pk=jt1.pk)
jt2 = JobTemplate.objects.get(pk=jt2.pk)
jt3 = JobTemplate.objects.get(pk=jt3.pk)
assert jt1.project == jt3.project
assert jt1.project != jt2.project
assert o1.projects.count() == 1
assert o2.projects.count() == 1
assert o3.projects.count() == 1
assert o1.projects.all()[0].jobtemplates.count() == 2
assert o2.projects.all()[0].jobtemplates.count() == 1
assert o3.projects.all()[0].jobtemplates.count() == 0
@pytest.mark.django_db
def test_single_org_project_migration(organization):
project = Project.objects.create(name='my project',
description="description",
organization=None)
organization.deprecated_projects.add(project)
assert project.organization is None
rbac.migrate_projects(apps, None)
project = Project.objects.get(id=project.id)
assert project.organization.id == organization.id
@pytest.mark.django_db
def test_no_org_project_migration(organization):
project = Project.objects.create(name='my project',
description="description",
organization=None)
assert project.organization is None
rbac.migrate_projects(apps, None)
assert project.organization is None
@pytest.mark.django_db
def test_multi_org_project_migration():
org1 = Organization.objects.create(name="org1", description="org1 desc")
org2 = Organization.objects.create(name="org2", description="org2 desc")
project = Project.objects.create(name='my project',
description="description",
organization=None)
assert Project.objects.all().count() == 1
assert Project.objects.filter(organization=org1).count() == 0
assert Project.objects.filter(organization=org2).count() == 0
project.deprecated_organizations.add(org1)
project.deprecated_organizations.add(org2)
assert project.organization is None
rbac.migrate_projects(apps, None)
assert Project.objects.filter(organization=org1).count() == 1
assert Project.objects.filter(organization=org2).count() == 1
@pytest.mark.django_db
def test_project_user_project(user_project, project, user):
u = user('owner')
assert old_access.check_user_access(u, user_project.__class__, 'read', user_project)
assert old_access.check_user_access(u, project.__class__, 'read', project) is False
assert u not in user_project.read_role
assert u not in project.read_role
rbac.migrate_projects(apps, None)
assert u in user_project.read_role
assert u not in project.read_role
@pytest.mark.django_db
def test_project_accessible_by_sa(user, project):
u = user('systemadmin', is_superuser=True)
# This gets setup by a signal, but we want to test the migration which will set this up too, so remove it
Role.singleton('system_administrator').members.remove(u)
assert u not in project.read_role
rbac.migrate_organization(apps, None)
rbac.migrate_users(apps, None)
rbac.migrate_projects(apps, None)
print(project.admin_role.ancestors.all())
print(project.admin_role.ancestors.all())
assert u in project.admin_role
@pytest.mark.django_db
def test_project_org_members(user, organization, project):
admin = user('orgadmin')
member = user('orgmember')
assert admin not in project.read_role
assert member not in project.read_role
organization.deprecated_admins.add(admin)
organization.deprecated_users.add(member)
rbac.migrate_organization(apps, None)
rbac.migrate_projects(apps, None)
assert admin in project.admin_role
assert member in project.read_role
@pytest.mark.django_db
def test_project_team(user, team, project):
nonmember = user('nonmember')
member = user('member')
team.deprecated_users.add(member)
project.deprecated_teams.add(team)
assert nonmember not in project.read_role
assert member not in project.read_role
rbac.migrate_team(apps, None)
rbac.migrate_organization(apps, None)
rbac.migrate_projects(apps, None)
assert member in project.read_role
assert nonmember not in project.read_role
@pytest.mark.django_db
def test_project_explicit_permission(user, team, project, organization):
u = user('prjuser')
assert old_access.check_user_access(u, project.__class__, 'read', project) is False
organization.deprecated_users.add(u)
p = Permission(user=u, project=project, permission_type='create', name='Perm name')
p.save()
assert u not in project.read_role
rbac.migrate_organization(apps, None)
rbac.migrate_projects(apps, None)
assert u in project.read_role
@pytest.mark.django_db
def test_create_project_foreign_org_admin(org_admin, organization, organization_factory):
"""Org admins can only create projects in their own org."""
other_org = organization_factory('not-my-org').organization
access = ProjectAccess(org_admin)
assert not access.can_add({'organization': other_org.pk, 'name': 'new-project'})
@pytest.mark.django_db
def test_modify_project_foreign_org_admin(org_admin, organization, organization_factory, project):
"""Org admins can only modify projects in their own org."""
other_org = organization_factory('not-my-org').organization
access = ProjectAccess(org_admin)
assert not access.can_change(project, {'organization': other_org.pk, 'name': 'new-project'})

26
awx/main/tests/functional/test_rbac_user.py
View File

@@ -1,11 +1,9 @@
import pytest import pytest
from django.apps import apps
from django.test import TransactionTestCase from django.test import TransactionTestCase
from awx.main.migrations import _rbac as rbac
from awx.main.access import UserAccess from awx.main.access import UserAccess
from awx.main.models import Role, User, Organization, Inventory from awx.main.models import User, Organization, Inventory
@pytest.mark.django_db @pytest.mark.django_db
@@ -46,28 +44,6 @@ def test_system_auditor_is_system_auditor(system_auditor):
assert system_auditor.is_system_auditor assert system_auditor.is_system_auditor
@pytest.mark.django_db
def test_user_admin(user_project, project, user):
username = unicode("\xc3\xb4", "utf-8")
joe = user(username, is_superuser = False)
admin = user('admin', is_superuser = True)
sa = Role.singleton('system_administrator')
# this should happen automatically with our signal
assert sa.members.filter(id=admin.id).exists() is True
sa.members.remove(admin)
assert sa.members.filter(id=joe.id).exists() is False
assert sa.members.filter(id=admin.id).exists() is False
rbac.migrate_users(apps, None)
# The migration should add the admin back in
assert sa.members.filter(id=joe.id).exists() is False
assert sa.members.filter(id=admin.id).exists() is True
@pytest.mark.django_db @pytest.mark.django_db
def test_user_queryset(user): def test_user_queryset(user):
u = user('pete', False) u = user('pete', False)

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

@@ -394,93 +394,6 @@ class BaseJobTestMixin(BaseTestMixin):
) )
self.team_ops_testers.member_role.children.add(self.cred_ops_test.use_role) self.team_ops_testers.member_role.children.add(self.cred_ops_test.use_role)
self.ops_east_permission = Permission.objects.create(
inventory = self.inv_ops_east,
project = self.proj_prod,
team = self.team_ops_east,
permission_type = PERM_JOBTEMPLATE_CREATE,
created_by = self.user_sue
)
self.ops_east_permission_prod_east = Permission.objects.create(
inventory = self.inv_ops_east,
project = self.proj_prod_east,
team = self.team_ops_east,
permission_type = PERM_JOBTEMPLATE_CREATE,
created_by = self.user_sue
)
self.ops_east_permission_inv_admin = Permission.objects.create(
inventory = self.inv_ops_east,
team = self.team_ops_east,
permission_type = PERM_INVENTORY_ADMIN,
created_by = self.user_sue
)
self.ops_testers_permission = Permission.objects.create(
inventory = self.inv_ops_west,
project = self.proj_prod,
team = self.team_ops_testers,
permission_type = PERM_INVENTORY_CHECK,
created_by = self.user_sue
)
self.ops_testers_permission_inv_read = Permission.objects.create(
inventory = self.inv_ops_west,
team = self.team_ops_testers,
permission_type = PERM_INVENTORY_READ,
created_by = self.user_sue
)
self.doug_check_permission = Permission.objects.create(
inventory = self.inv_eng,
project = self.proj_dev,
user = self.user_doug,
permission_type = PERM_INVENTORY_CHECK,
created_by = self.user_sue
)
self.doug_inv_read_permission = Permission.objects.create(
inventory = self.inv_eng,
user = self.user_doug,
permission_type = PERM_INVENTORY_READ,
created_by = self.user_sue
)
self.juan_deploy_permission = Permission.objects.create(
inventory = self.inv_eng,
project = self.proj_dev,
user = self.user_juan,
permission_type = PERM_INVENTORY_DEPLOY,
created_by = self.user_sue
)
self.hannibal_create_permission = Permission.objects.create(
inventory = self.inv_eng,
project = self.proj_dev,
user = self.user_hannibal,
permission_type = PERM_JOBTEMPLATE_CREATE,
created_by = self.user_sue
)
# FIXME: Define explicit permissions for tests.
# other django user is on the project team and can deploy
#self.permission1 = Permission.objects.create(
# inventory = self.inventory,
# project = self.project,
# team = self.team,
# permission_type = PERM_INVENTORY_DEPLOY,
# created_by = self.normal_django_user
#)
# individual permission granted to other2 user, can run check mode
#self.permission2 = Permission.objects.create(
# inventory = self.inventory,
# project = self.project,
# user = self.other2_django_user,
# permission_type = PERM_INVENTORY_CHECK,
# created_by = self.normal_django_user
#)
# Engineering has job templates to check/run the dev project onto # Engineering has job templates to check/run the dev project onto
# their own inventory. # their own inventory.
self.jt_eng_check = JobTemplate.objects.create( self.jt_eng_check = JobTemplate.objects.create(

34
awx/main/tests/old/jobs/jobs_monolithic.py
View File

@@ -244,40 +244,6 @@ class JobTemplateTest(BaseJobTestMixin, django.test.TransactionTestCase):
#print [x['name'] for x in resp['results']] #print [x['name'] for x in resp['results']]
self.assertEquals(resp['count'], 0) self.assertEquals(resp['count'], 0)
# We give Juan inventory permission and he can see both Job Templates because he already has deploy permission
# Now he can see both job templates
Permission.objects.create(
inventory = self.inv_eng,
user = self.user_juan,
permission_type = PERM_INVENTORY_READ,
created_by = self.user_sue
)
with self.current_user(self.user_juan):
resp = self.get(url, expect=200)
#print [x['name'] for x in resp['results']]
self.assertEquals(resp['count'], 2)
# Randall is on the ops testers team that has permission to run a single check playbook on ops west
with self.current_user(self.user_randall):
resp = self.get(url, expect=200)
#print [x['name'] for x in resp['results']]
self.assertEquals(resp['count'], 1)
# Holly is on the ops east team and can see all of that team's job templates
with self.current_user(self.user_holly):
resp = self.get(url, expect=200)
#print [x['name'] for x in resp['results']]
self.assertEquals(resp['count'], 3)
# Chuck is temporarily assigned to ops east team to help them running some playbooks
# even though he's in a different group and org entirely he'll now see their job templates
self.team_ops_east.deprecated_users.add(self.user_chuck)
with self.current_user(self.user_chuck):
resp = self.get(url, expect=200)
#print [x['name'] for x in resp['results']]
self.assertEquals(resp['count'], 6)
def test_credentials_list(self): def test_credentials_list(self):
url = reverse('api:credential_list') url = reverse('api:credential_list')
# Greg can't see the 'south' credential because the 'southerns' team is inactive # Greg can't see the 'south' credential because the 'southerns' team is inactive