mirror of
https://github.com/ansible/awx.git
synced 2026-02-28 08:18:43 -03:30
Initial pass of removing RBAC deprecated fields and Permission
This commit is contained in:
@@ -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)
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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,10 +375,6 @@ 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
|
||||||
|
|
||||||
@@ -402,26 +382,6 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
|
|||||||
inputs_before = {}
|
inputs_before = {}
|
||||||
# If update_fields has been specified, add our field names to it,
|
# 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.
|
# 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:
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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')
|
||||||
|
|||||||
@@ -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
|
|
||||||
|
|||||||
@@ -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
|
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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):
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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'})
|
|
||||||
@@ -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)
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user