diff --git a/awx/api/views.py b/awx/api/views.py index d31b46e916..9414794199 100644 --- a/awx/api/views.py +++ b/awx/api/views.py @@ -3711,6 +3711,13 @@ class JobRelaunch(RetrieveAPIView, GenericAPIView): def dispatch(self, *args, **kwargs): return super(JobRelaunch, self).dispatch(*args, **kwargs) + def check_object_permissions(self, request, obj): + if request.method == 'POST' and obj: + relaunch_perm, messages = request.user.can_access_with_errors(self.model, 'start', obj) + if not relaunch_perm and 'detail' in messages: + self.permission_denied(request, message=messages['detail']) + return super(JobRelaunch, self).check_object_permissions(request, obj) + def post(self, request, *args, **kwargs): obj = self.get_object() diff --git a/awx/main/access.py b/awx/main/access.py index 01165191f0..7a29f5aacc 100644 --- a/awx/main/access.py +++ b/awx/main/access.py @@ -1426,7 +1426,7 @@ class JobAccess(BaseAccess): for fd in ignored_fields: if fd == 'extra_credentials': if set(job_fields[fd].all()) != set(getattr(obj.job_template, fd).all()): - # Job has field that is not promptable + # Job has extra_credentials that are not promptable prompts_access = False elif fd != 'extra_vars' and job_fields[fd] != getattr(obj.job_template, fd): # Job has field that is not promptable @@ -1443,7 +1443,28 @@ class JobAccess(BaseAccess): project_access = obj.project is None or self.user in obj.project.admin_role # job can be relaunched if user could make an equivalent JT - return inventory_access and credential_access and (org_access or project_access) + ret = inventory_access and credential_access and (org_access or project_access) + if not ret and self.save_messages: + if not obj.job_template: + pretext = _('Job has been orphaned from its job template.') + elif prompts_access: + self.messages['detail'] = _('You do not have execute permission to related job template.') + return False + else: + pretext = _('Job was launched with prompted fields.') + if inventory_access and credential_access: + self.messages['detail'] = '{} {}'.format(pretext, _(' Organization level permissions required.')) + else: + self.messages['detail'] = '{} {}'.format(pretext, _(' You do not have permission to related resources.')) + return ret + + def get_method_capability(self, method, obj, parent_obj): + if method == 'start': + # Return simplistic permission, will perform detailed check on POST + if not obj.job_template: + return True + return self.user in obj.job_template.execute_role + return super(JobAccess, self).get_method_capability(method, obj, parent_obj) def can_cancel(self, obj): if not obj.can_cancel: @@ -1968,6 +1989,7 @@ class UnifiedJobTemplateAccess(BaseAccess): # 'project', # 'inventory', # 'credential', + # 'credential__credential_type', #) return qs.all() @@ -2015,6 +2037,7 @@ class UnifiedJobAccess(BaseAccess): # 'project', # 'inventory', # 'credential', + # 'credential__credential_type', # 'job_template', # 'inventory_source', # 'project___credential', diff --git a/awx/main/tests/functional/api/test_job.py b/awx/main/tests/functional/api/test_job.py index 828ac89c54..e10156ce62 100644 --- a/awx/main/tests/functional/api/test_job.py +++ b/awx/main/tests/functional/api/test_job.py @@ -2,8 +2,9 @@ import pytest from awx.api.versioning import reverse +from awx.main.models import JobTemplate, User + -# TODO: test this with RBAC and lower-priveleged users @pytest.mark.django_db def test_extra_credentials(get, organization_factory, job_template_factory, credential): objs = organization_factory("org", superusers=['admin']) @@ -16,3 +17,23 @@ def test_extra_credentials(get, organization_factory, job_template_factory, cred url = reverse('api:job_extra_credentials_list', kwargs={'version': 'v2', 'pk': job.pk}) response = get(url, user=objs.superusers.admin) assert response.data.get('count') == 1 + + +@pytest.mark.django_db +def test_job_relaunch_permission_denied_response( + post, get, inventory, project, credential, net_credential, machine_credential): + jt = JobTemplate.objects.create(name='testjt', inventory=inventory, project=project, + credential=machine_credential) + jt_user = User.objects.create(username='jobtemplateuser') + jt.execute_role.members.add(jt_user) + job = jt.create_unified_job() + + # User capability is shown for this + r = get(job.get_absolute_url(), jt_user, expect=200) + assert r.data['summary_fields']['user_capabilities']['start'] + + # Job has prompted extra_credential, launch denied w/ message + job.extra_credentials.add(net_credential) + r = post(reverse('api:job_relaunch', kwargs={'pk':job.pk}), {}, jt_user, expect=403) + assert 'launched with prompted fields' in r.data['detail'] + assert 'do not have permission' in r.data['detail'] diff --git a/awx/main/tests/functional/api/test_job_runtime_params.py b/awx/main/tests/functional/api/test_job_runtime_params.py index d26967dd1b..5ffdf9fc34 100644 --- a/awx/main/tests/functional/api/test_job_runtime_params.py +++ b/awx/main/tests/functional/api/test_job_runtime_params.py @@ -271,43 +271,6 @@ def test_job_block_scan_job_inv_change(mocker, bad_scan_JT, runtime_data, post, assert 'inventory' in response.data -@pytest.mark.django_db -@pytest.mark.job_runtime_vars -def test_job_relaunch_copy_vars(job_with_links, machine_credential, inventory, - deploy_jobtemplate, post, mocker): - job_with_links.job_template = deploy_jobtemplate - job_with_links.limit = "my_server" - with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate._get_unified_job_field_names', - return_value=['inventory', 'credential', 'limit']): - second_job = job_with_links.copy_unified_job() - - # Check that job data matches the original variables - assert second_job.credential == job_with_links.credential - assert second_job.inventory == job_with_links.inventory - assert second_job.limit == 'my_server' - - -@pytest.mark.django_db -@pytest.mark.job_runtime_vars -def test_job_relaunch_resource_access(job_with_links, user): - inventory_user = user('user1', False) - credential_user = user('user2', False) - both_user = user('user3', False) - - # Confirm that a user with inventory & credential access can launch - job_with_links.credential.use_role.members.add(both_user) - job_with_links.inventory.use_role.members.add(both_user) - assert both_user.can_access(Job, 'start', job_with_links) - - # Confirm that a user with credential access alone cannot launch - job_with_links.credential.use_role.members.add(credential_user) - assert not credential_user.can_access(Job, 'start', job_with_links) - - # Confirm that a user with inventory access alone cannot launch - job_with_links.inventory.use_role.members.add(inventory_user) - assert not inventory_user.can_access(Job, 'start', job_with_links) - - @pytest.mark.django_db def test_job_launch_JT_with_validation(machine_credential, deploy_jobtemplate): deploy_jobtemplate.extra_vars = '{"job_template_var": 3}' diff --git a/awx/main/tests/functional/models/test_unified_job.py b/awx/main/tests/functional/models/test_unified_job.py index 4d19e4191e..5b89644457 100644 --- a/awx/main/tests/functional/models/test_unified_job.py +++ b/awx/main/tests/functional/models/test_unified_job.py @@ -4,7 +4,7 @@ import pytest from django.contrib.contenttypes.models import ContentType # AWX -from awx.main.models import UnifiedJobTemplate, JobTemplate, WorkflowJobTemplate, Project +from awx.main.models import UnifiedJobTemplate, Job, JobTemplate, WorkflowJobTemplate, Project @pytest.mark.django_db @@ -16,11 +16,11 @@ def test_subclass_types(rando): ]) +@pytest.mark.django_db class TestCreateUnifiedJob: ''' Ensure that copying a job template to a job handles many to many field copy ''' - @pytest.mark.django_db def test_many_to_many(self, mocker, job_template_labels): jt = job_template_labels _get_unified_job_field_names = mocker.patch('awx.main.models.jobs.JobTemplate._get_unified_job_field_names', return_value=['labels']) @@ -34,7 +34,6 @@ class TestCreateUnifiedJob: ''' Ensure that data is looked for in parameter list before looking at the object ''' - @pytest.mark.django_db def test_many_to_many_kwargs(self, mocker, job_template_labels): jt = job_template_labels mocked = mocker.MagicMock() @@ -47,3 +46,22 @@ class TestCreateUnifiedJob: _get_unified_job_field_names.assert_called_with() mocked.all.assert_called_with() + + ''' + Ensure that extra_credentials m2m field is copied to new relaunched job + ''' + def test_job_relaunch_copy_vars(self, machine_credential, inventory, + deploy_jobtemplate, post, mocker, net_credential): + job_with_links = Job.objects.create(name='existing-job', credential=machine_credential, inventory=inventory) + job_with_links.job_template = deploy_jobtemplate + job_with_links.limit = "my_server" + job_with_links.extra_credentials.add(net_credential) + with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate._get_unified_job_field_names', + return_value=['inventory', 'credential', 'limit']): + second_job = job_with_links.copy_unified_job() + + # Check that job data matches the original variables + assert second_job.credential == job_with_links.credential + assert second_job.inventory == job_with_links.inventory + assert second_job.limit == 'my_server' + assert net_credential in second_job.extra_credentials.all() diff --git a/awx/main/tests/functional/test_rbac_job.py b/awx/main/tests/functional/test_rbac_job.py index de1ccf6817..7ad79adaac 100644 --- a/awx/main/tests/functional/test_rbac_job.py +++ b/awx/main/tests/functional/test_rbac_job.py @@ -8,10 +8,12 @@ from awx.main.access import ( ) from awx.main.models import ( Job, + JobTemplate, AdHocCommand, InventoryUpdate, InventorySource, - ProjectUpdate + ProjectUpdate, + User ) @@ -124,6 +126,45 @@ def test_project_org_admin_delete_allowed(normal_job, org_admin): assert access.can_delete(normal_job) +@pytest.mark.django_db +class TestJobRelaunchAccess: + + def test_job_relaunch_normal_resource_access(self, user, inventory, machine_credential): + job_with_links = Job.objects.create(name='existing-job', credential=machine_credential, inventory=inventory) + inventory_user = user('user1', False) + credential_user = user('user2', False) + both_user = user('user3', False) + + # Confirm that a user with inventory & credential access can launch + job_with_links.credential.use_role.members.add(both_user) + job_with_links.inventory.use_role.members.add(both_user) + assert both_user.can_access(Job, 'start', job_with_links, validate_license=False) + + # Confirm that a user with credential access alone cannot launch + job_with_links.credential.use_role.members.add(credential_user) + assert not credential_user.can_access(Job, 'start', job_with_links, validate_license=False) + + # Confirm that a user with inventory access alone cannot launch + job_with_links.inventory.use_role.members.add(inventory_user) + assert not inventory_user.can_access(Job, 'start', job_with_links, validate_license=False) + + def test_job_relaunch_extra_credential_access( + self, post, inventory, project, credential, net_credential): + jt = JobTemplate.objects.create(name='testjt', inventory=inventory, project=project) + jt.extra_credentials.add(credential) + job = jt.create_unified_job() + + # Job is unchanged from JT, user has ability to launch + jt_user = User.objects.create(username='jobtemplateuser') + jt.execute_role.members.add(jt_user) + assert jt_user in job.job_template.execute_role + assert jt_user.can_access(Job, 'start', job, validate_license=False) + + # Job has prompted extra_credential, launch denied w/ message + job.extra_credentials.add(net_credential) + assert not jt_user.can_access(Job, 'start', job, validate_license=False) + + @pytest.mark.django_db class TestJobAndUpdateCancels: