Merge pull request #805 from wwitzel3/rbac

Added Credential migration tests and extended organization tests.

awx/main/fields.py

@@ -2,7 +2,7 @@
 # All Rights Reserved.
 
 # Django
-from django.db.models.signals import post_save, post_init
+from django.db.models.signals import post_save
 from django.db import models
 from django.db.models.fields.related import SingleRelatedObjectDescriptor
 from django.db.models.fields.related import ReverseSingleRelatedObjectDescriptor

awx/main/models/credential.py

@@ -157,18 +157,15 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
     )
     owner_role = ImplicitRoleField(
         role_name='Credential Owner',
-        parent_role=[
+        parent_role='team.admin_role',
-            'user.user_role',
-            'team.admin_role'
-        ],
         resource_field='resource',
-        permissions = { 'all': True }
+        permissions = {'all': True}
     )
     usage_role = ImplicitRoleField(
         role_name='Credential User',
         resource_field='resource',
         parent_role= 'team.member_role',
-        permissions = { 'usage': True }
+        permissions = {'use': True}
     )
 
     @property
@@ -366,6 +363,15 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique, ResourceMixin):
                 update_fields.append('cloud')
         super(Credential, self).save(*args, **kwargs)
 
+    def migrate_to_rbac(self):
+        if self.user:
+            self.owner_role.members.add(self.user)
+            return [self.user]
+        elif self.team:
+            self.owner_role.parents.add(self.team.admin_role)
+            self.usage_role.parents.add(self.team.member_role)
+            return [self.team]
+
 def validate_ssh_private_key(data):
     """Validate that the given SSH private key or certificate is,
     in fact, valid.

awx/main/models/inventory.py

@@ -97,13 +97,13 @@ class Inventory(CommonModel, ResourceMixin):
         role_name='Inventory Administrator',
         parent_role='organization.admin_role',
         resource_field='resource',
-        permissions = { 'all': True }
+        permissions = {'all': True}
     )
     auditor_role = ImplicitRoleField(
         role_name='Inventory Auditor',
         parent_role='organization.auditor_role',
         resource_field='resource',
-        permissions = { 'read': True }
+        permissions = {'read': True}
     )
     updater_role = ImplicitRoleField(
         role_name='Inventory Updater',
@@ -546,13 +546,13 @@ class Group(CommonModelNameNotUnique, ResourceMixin):
         role_name='Inventory Group Administrator',
         parent_role='inventory.admin_role',
         resource_field='resource',
-        permissions = { 'all': True }
+        permissions = {'all': True}
     )
     auditor_role = ImplicitRoleField(
         role_name='Inventory Group Auditor',
         parent_role='inventory.auditor_role',
         resource_field='resource',
-        permissions = { 'read': True }
+        permissions = {'read': True}
     )
     updater_role = ImplicitRoleField(
         role_name='Inventory Group Updater',

awx/main/models/jobs.py

@@ -185,19 +185,19 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
         role_name='Job Template Administrator',
         parent_role='project.admin_role',
         resource_field='resource',
-        permissions = { 'all': True }
+        permissions = {'all': True}
     )
     auditor_role = ImplicitRoleField(
         role_name='Job Template Auditor',
         parent_role='project.auditor_role',
         resource_field='resource',
-        permissions = { 'read': True }
+        permissions = {'read': True}
     )
     executor_role = ImplicitRoleField(
         role_name='Job Template Executor',
         parent_role='project.auditor_role',
         resource_field='resource',
-        permissions = { 'execute': True }
+        permissions = {'execute': True}
     )
 
     @classmethod

awx/main/models/organization.py

@@ -55,12 +55,12 @@ class Organization(CommonModel, ResourceMixin):
     admin_role = ImplicitRoleField(
         role_name='Organization Administrator',
         resource_field='resource',
-        permissions = { 'all': True }
+        permissions = {'all': True}
     )
     auditor_role = ImplicitRoleField(
         role_name='Organization Auditor',
         resource_field='resource',
-        permissions = { 'read': True }
+        permissions = {'read': True}
     )
 
 
@@ -118,17 +118,19 @@ class Team(CommonModelNameNotUnique, ResourceMixin):
         role_name='Team Administrator',
         parent_role='organization.admin_role',
         resource_field='resource',
-        permissions = { 'all': True }
+        permissions = {'all': True}
     )
     auditor_role = ImplicitRoleField(
         role_name='Team Auditor',
         parent_role='organization.auditor_role',
         resource_field='resource',
-        permissions = { 'read': True }
+        permissions = {'read': True}
     )
     member_role = ImplicitRoleField(
         role_name='Team Member',
         parent_role='admin_role',
+        resource_field='resource',
+        permissions = {'read':True},
     )
 
     def get_absolute_url(self):
@@ -142,6 +144,13 @@ class Team(CommonModelNameNotUnique, ResourceMixin):
             cred.mark_inactive()
         super(Team, self).mark_inactive(save=save)
 
+    def migrate_to_rbac(self):
+        migrated = []
+        for user in self.users.all():
+            self.member_role.members.add(user)
+            migrated.append(user)
+        return migrated
+
 class Permission(CommonModelNameNotUnique):
     '''
     A permission allows a user, project, or team to be able to use an inventory source.

awx/main/models/projects.py

@@ -219,25 +219,25 @@ class Project(UnifiedJobTemplate, ProjectOptions, ResourceMixin):
         role_name='Project Administrator',
         parent_role='organization.admin_role',
         resource_field='resource',
-        permissions = { 'all': True }
+        permissions = {'all': True}
     )
     auditor_role = ImplicitRoleField(
         role_name='Project Auditor',
         parent_role='organization.auditor_role',
         resource_field='resource',
-        permissions = { 'read': True }
+        permissions = {'read': True}
     )
     member_role = ImplicitRoleField(
         role_name='Project Member',
         parent_role='admin',
         resource_field='resource',
-        permissions = { 'usage': True }
+        permissions = {'usage': True}
     )
     scm_update_role = ImplicitRoleField(
         role_name='Project Updater',
         parent_role='admin',
         resource_field='resource',
-        permissions = { 'scm_update': True }
+        permissions = {'scm_update': True}
     )
 
     @classmethod

awx/main/models/rbac.py

@@ -82,7 +82,7 @@ class Role(CommonModelNameNotUnique):
         except Role.DoesNotExist:
             ret = Role(singleton_name=name)
             ret.save()
-            return ret;
+            return ret
 
 
 

awx/main/tests/functional/conftest.py

@@ -1,11 +1,35 @@
 import pytest
 
-from awx.main.models.organization import Organization
+from awx.main.models.credential import Credential
+from awx.main.models.organization import (
+    Organization,
+    Team,
+)
+from django.contrib.auth.models import User
+
+@pytest.fixture
+def user():
+    def u(name, is_superuser=False):
+        try:
+            user = User.objects.get(username=name)
+        except User.DoesNotExist:
+            user = User(username=name, is_superuser=is_superuser, password=name)
+            user.save()
+        return user
+    return u
+
+@pytest.fixture
+def team(organization):
+    return Team.objects.create(organization=organization, name='test-team')
 
 @pytest.fixture
 def organization():
     return Organization.objects.create(name="test-org", description="test-org-desc")
 
+@pytest.fixture
+def credential():
+    return Credential.objects.create(kind='aws', name='test-cred')
+
 @pytest.fixture
 def permissions():
     return {
@@ -14,5 +38,8 @@ def permissions():
 
         'auditor':{'read':True, 'create':False, 'write':False,
                    'update':False, 'delete':False, 'scm_update':False, 'execute':False, 'use':False,},
+
+        'usage':{'read':False, 'create':False, 'write':False,
+                 'update':False, 'delete':False, 'scm_update':False, 'execute':False, 'use':True,},
     }
 

awx/main/tests/functional/test_rbac_credential.py

@@ -0,0 +1,44 @@
+import pytest
+
+@pytest.mark.django_db
+def test_credential_migration_user(credential, user, permissions):
+    u = user('user', False)
+    credential.user = u
+    migrated = credential.migrate_to_rbac()
+    assert len(migrated) == 1
+    assert credential.accessible_by(u, permissions['admin'])
+
+@pytest.mark.django_db
+def test_credential_usage_role(credential, user, permissions):
+    u = user('user', False)
+    credential.usage_role.members.add(u)
+    assert credential.accessible_by(u, permissions['usage'])
+
+@pytest.mark.django_db
+def test_credential_migration_team_member(credential, team, user, permissions):
+    u = user('user', False)
+    team.admin_role.members.add(u)
+    credential.team = team
+
+    # No permissions pre-migration
+    assert not credential.accessible_by(u, permissions['admin'])
+
+    migrated = credential.migrate_to_rbac()
+    # Admin permissions post migration
+    assert len(migrated) == 1
+    assert credential.accessible_by(u, permissions['admin'])
+
+@pytest.mark.django_db
+def test_credential_migration_team_admin(credential, team, user, permissions):
+    u = user('user', False)
+    team.member_role.members.add(u)
+    credential.team = team
+
+    # No permissions pre-migration
+    assert not credential.accessible_by(u, permissions['usage'])
+
+    # Usage permissions post migration
+    migrated = credential.migrate_to_rbac()
+    assert len(migrated) == 1
+    assert credential.accessible_by(u, permissions['usage'])
+

awx/main/tests/functional/test_rbac_organization.py

@@ -1,51 +1,43 @@
 import pytest
 
 from awx.main.access import OrganizationAccess
-from django.contrib.auth.models import User
-
-def make_user(name, admin=False):
-    try:
-        user = User.objects.get(username=name)
-    except User.DoesNotExist:
-        user = User(username=name, is_superuser=admin, password=name)
-        user.save()
-    return user
 
 @pytest.mark.django_db
-@pytest.mark.parametrize("username,admin", [
+def test_organization_migration_admin(organization, permissions, user):
-    ("admin", True),
+    u = user('admin', True)
-    ("user", False),
+    organization.admins.add(u)
-])
+
-def test_organization_migration(organization, permissions, username, admin):
+    assert not organization.accessible_by(u, permissions['admin'])
-    user = make_user(username, admin)
-    if admin:
-        organization.admins.add(user)
-    else:
-        organization.users.add(user)
 
     migrated_users = organization.migrate_to_rbac()
     assert len(migrated_users) == 1
-    assert migrated_users[0] == user
+    assert organization.accessible_by(u, permissions['admin'])
-
-    if admin:
-        assert organization.accessible_by(user, permissions['admin']) == True
-    else:
-        assert organization.accessible_by(user, permissions['auditor']) == True
 
 @pytest.mark.django_db
-@pytest.mark.parametrize("username,admin", [
+def test_organization_migration_user(organization, permissions, user):
-    ("admin", True),
+    u = user('user', False)
-    ("user-admin", False),
+    organization.users.add(u)
-    ("user", False)
-])
-def test_organization_access(organization, username, admin):
-    user = make_user(username, admin)
-    access = OrganizationAccess(user)
-    if admin:
-        assert access.can_change(organization, None) == True
-    elif username == "user-admin":
-        organization.admins.add(user)
-        assert access.can_change(organization, None) == True
-    else:
-        assert access.can_change(organization, None) == False
 
+    assert not organization.accessible_by(u, permissions['auditor'])
+
+    migrated_users = organization.migrate_to_rbac()
+    assert len(migrated_users) == 1
+    assert organization.accessible_by(u, permissions['auditor'])
+
+@pytest.mark.django_db
+def test_organization_access_superuser(organization, user):
+    access = OrganizationAccess(user('admin', True))
+    assert access.can_change(organization, None)
+
+@pytest.mark.django_db
+def test_organization_access_admin(organization, user):
+    u = user('admin', False)
+    organization.admins.add(u)
+
+    access = OrganizationAccess(u)
+    assert access.can_change(organization, None)
+
+@pytest.mark.django_db
+def test_organization_access_user(organization, user):
+    access = OrganizationAccess(user('user', False))
+    assert not access.can_change(organization, None)

awx/main/tests/functional/test_rbac_team.py

@@ -0,0 +1,13 @@
+import pytest
+
+@pytest.mark.django_db
+def test_team_migration_user(team, user, permissions):
+    u = user('user', False)
+    team.users.add(u)
+
+    assert not team.accessible_by(u, permissions['auditor'])
+
+    migrated = team.migrate_to_rbac()
+    assert len(migrated) == 1
+    assert team.accessible_by(u, permissions['auditor'])
+