Job template migration and migration tests

awx/main/migrations/_rbac.py

@@ -1,4 +1,5 @@
 from collections import defaultdict
+import _old_access as old_access
 
 def migrate_users(apps, schema_editor):
     migrations = list()
@@ -52,7 +53,7 @@ def migrate_inventory(apps, schema_editor):
 
     for inventory in Inventory.objects.all():
         teams, users = [], []
-        for perm in Permission.objects.filter(inventory=inventory):
+        for perm in Permission.objects.filter(inventory=inventory, active=True):
             role = None
             execrole = None
             if perm.permission_type == 'admin':
@@ -64,6 +65,10 @@ def migrate_inventory(apps, schema_editor):
             elif perm.permission_type == 'write':
                 role = inventory.updater_role
                 pass
+            elif perm.permission_type == 'check':
+                pass
+            elif perm.permission_type == 'run':
+                pass
             else:
                 raise Exception('Unhandled permission type for inventory: %s' % perm.permission_type)
             if perm.run_ad_hoc_commands:
@@ -122,14 +127,92 @@ def migrate_projects(apps, schema_editor):
                     project.member_role.members.add(user)
                     migrations[project.name]['users'].add(user)
 
-        for perm in Permission.objects.filter(project=project):
+        for perm in Permission.objects.filter(project=project, active=True):
             # All perms at this level just imply a user or team can read
             if perm.team:
-                team.member_role.children.add(project.member_role)
-                migrations[project.name]['teams'].add(team)
+                perm.team.member_role.children.add(project.member_role)
+                migrations[project.name]['teams'].add(perm.team)
 
             if perm.user:
                 project.member_role.members.add(perm.user)
                 migrations[project.name]['users'].add(perm.user)
 
     return migrations
+
+
+
+def migrate_job_templates(apps, schema_editor):
+    '''
+    NOTE: This must be run after orgs, inventory, projects, credential, and
+    users have been migrated
+    '''
+
+
+    '''
+    I can see job templates when:
+     X I am a superuser.
+     - I can read the inventory, project and credential (which means I am an
+       org admin or member of a team with access to all of the above).
+     - I have permission explicitly granted to check/deploy with the inventory
+       and project.
+
+
+    #This does not mean I would be able to launch a job from the template or
+    #edit the template.
+     - access.py can_read for JobTemplate enforces that you can only
+       see it if you can launch it, so the above imply launch too
+    '''
+
+
+    '''
+    Tower administrators, organization administrators, and project
+    administrators, within a project under their purview, may create and modify
+    new job templates for that project.
+
+    When editing a job template, they may select among the inventory groups and
+    credentials in the organization for which they have usage permissions, or
+    they may leave either blank to be selected at runtime.
+
+    Additionally, they may specify one or more users/teams that have execution
+    permission for that job template, among the users/teams that are a member
+    of that project.
+
+    That execution permission is valid irrespective of any explicit permissions
+    the user has or has not been granted to the inventory group or credential
+    specified in the job template.
+
+    '''
+
+    migrations = defaultdict(lambda: defaultdict(set))
+
+    User = apps.get_model('auth', 'User')
+    JobTemplate = apps.get_model('main', 'JobTemplate')
+    Team = apps.get_model('main', 'Team')
+    Permission = apps.get_model('main', 'Permission')
+
+    for jt in JobTemplate.objects.all():
+        for team in Team.objects.all():
+            if Permission.objects.filter(
+                    team=team,
+                    inventory=jt.inventory,
+                    project=jt.project,
+                    active=True,
+                    permission_type__in=['create', 'check', 'run'] if jt.job_type == 'check' else ['create', 'run']
+                ):
+                team.member_role.children.add(jt.executor_role);
+                migrations[jt.name]['teams'].add(team)
+
+
+        for user in User.objects.all():
+            if jt.accessible_by(user, {'execute': True}):
+                # If the job template is already accessible by the user, because they
+                # are a sytem, organization, or project admin, then don't add an explicit
+                # role entry for them
+                continue
+
+            if old_access.check_user_access(user, jt.__class__, 'start', jt, False):
+                jt.executor_role.members.add(user)
+                migrations[jt.name]['users'].add(user)
+
+
+    return migrations

awx/main/models/jobs.py

@@ -195,9 +195,8 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
     )
     executor_role = ImplicitRoleField(
         role_name='Job Template Executor',
-        parent_role='project.auditor_role',
         resource_field='resource',
-        permissions = {'execute': True}
+        permissions = {'read': True, 'execute': True}
     )
 
     @classmethod

awx/main/tests/functional/conftest.py

@@ -6,6 +6,7 @@ from awx.main.models.inventory import (
     Group,
 )
 from awx.main.models.projects import Project
+from awx.main.models.jobs import JobTemplate
 from awx.main.models.organization import (
     Organization,
     Team,
@@ -23,6 +24,28 @@ def user():
         return user
     return u
 
+@pytest.fixture
+def check_jobtemplate(project, inventory, credential):
+    return \
+        JobTemplate.objects.create(
+            job_type='check',
+            project=project,
+            inventory=inventory,
+            credential=credential,
+            name='check-job-template'
+        )
+
+@pytest.fixture
+def deploy_jobtemplate(project, inventory, credential):
+    return \
+        JobTemplate.objects.create(
+            job_type='run',
+            project=project,
+            inventory=inventory,
+            credential=credential,
+            name='deploy-job-template'
+        )
+
 @pytest.fixture
 def team(organization):
     return Team.objects.create(organization=organization, name='test-team')

awx/main/tests/functional/test_rbac_job_templates.py

@@ -0,0 +1,133 @@
+import pytest
+
+from awx.main.migrations import _rbac as rbac
+from awx.main.models import Permission
+from django.apps import apps
+
+@pytest.mark.django_db
+def test_job_template_migration_check(deploy_jobtemplate, check_jobtemplate, user):
+    admin = user('admin', is_superuser=True)
+    joe = user('joe')
+
+
+    check_jobtemplate.project.organizations.all()[0].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 check_jobtemplate.project.accessible_by(joe, {'read': True})
+    assert check_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert check_jobtemplate.accessible_by(joe, {'execute': True}) is False
+
+    migrations = rbac.migrate_job_templates(apps, None)
+
+    assert len(migrations[check_jobtemplate.name]['users']) == 1
+    assert check_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert check_jobtemplate.accessible_by(joe, {'execute': True}) is True
+    assert deploy_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert deploy_jobtemplate.accessible_by(joe, {'execute': True}) is False
+
+@pytest.mark.django_db
+def test_job_template_migration_deploy(deploy_jobtemplate, check_jobtemplate, user):
+    admin = user('admin', is_superuser=True)
+    joe = user('joe')
+
+
+    deploy_jobtemplate.project.organizations.all()[0].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 deploy_jobtemplate.project.accessible_by(joe, {'read': True})
+    assert deploy_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert deploy_jobtemplate.accessible_by(joe, {'execute': True}) is False
+
+    migrations = rbac.migrate_job_templates(apps, None)
+
+    assert len(migrations[deploy_jobtemplate.name]['users']) == 1
+    assert deploy_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert deploy_jobtemplate.accessible_by(joe, {'execute': True}) is True
+    assert check_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert check_jobtemplate.accessible_by(joe, {'execute': True}) is True
+
+
+@pytest.mark.django_db
+def test_job_template_team_migration_check(deploy_jobtemplate, check_jobtemplate, organization, team, user):
+    admin = user('admin', is_superuser=True)
+    joe = user('joe')
+    team.users.add(joe)
+    team.organization = organization
+    team.save()
+
+    check_jobtemplate.project.organizations.all()[0].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 check_jobtemplate.project.accessible_by(joe, {'read': True})
+    assert check_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert check_jobtemplate.accessible_by(joe, {'execute': True}) is False
+
+    migrations = rbac.migrate_job_templates(apps, None)
+
+    assert len(migrations[check_jobtemplate.name]['users']) == 0
+    assert len(migrations[check_jobtemplate.name]['teams']) == 1
+    assert check_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert check_jobtemplate.accessible_by(joe, {'execute': True}) is True
+
+    assert deploy_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert deploy_jobtemplate.accessible_by(joe, {'execute': True}) is False
+
+
+@pytest.mark.django_db
+def test_job_template_team_deploy_migration(deploy_jobtemplate, check_jobtemplate, organization, team, user):
+    admin = user('admin', is_superuser=True)
+    joe = user('joe')
+    team.users.add(joe)
+    team.organization = organization
+    team.save()
+
+    deploy_jobtemplate.project.organizations.all()[0].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 deploy_jobtemplate.project.accessible_by(joe, {'read': True})
+    assert deploy_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert deploy_jobtemplate.accessible_by(joe, {'execute': True}) is False
+
+    migrations = rbac.migrate_job_templates(apps, None)
+
+    assert len(migrations[deploy_jobtemplate.name]['users']) == 0
+    assert len(migrations[deploy_jobtemplate.name]['teams']) == 1
+    assert deploy_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert deploy_jobtemplate.accessible_by(joe, {'execute': True}) is True
+
+    assert check_jobtemplate.accessible_by(admin, {'execute': True}) is True
+    assert check_jobtemplate.accessible_by(joe, {'execute': True}) is True