diff --git a/awx/main/access.py b/awx/main/access.py index ec1ad16317..9f4b3c06a9 100644 --- a/awx/main/access.py +++ b/awx/main/access.py @@ -875,11 +875,12 @@ class JobAccess(BaseAccess): return self.user in obj.job_template.execute_role inventory_access = self.user in obj.inventory.use_role + credential_access = self.user in obj.credential.use_role org_access = self.user in obj.inventory.organization.admin_role project_access = obj.project is None or self.user in obj.project.admin_role - return inventory_access and (org_access or project_access) + return inventory_access and credential_access and (org_access or project_access) def can_cancel(self, obj): return self.can_read(obj) and obj.can_cancel 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 c6b5521966..bd79c7aab3 100644 --- a/awx/main/tests/functional/api/test_job_runtime_params.py +++ b/awx/main/tests/functional/api/test_job_runtime_params.py @@ -22,6 +22,10 @@ def runtime_data(organization): credential=cred_obj.pk, ) +@pytest.fixture +def job_with_links(machine_credential, inventory): + return Job.objects.create(name='existing-job', credential=machine_credential, inventory=inventory) + @pytest.fixture def job_template_prompts(project, inventory, machine_credential): def rf(on_off): @@ -171,47 +175,68 @@ def test_job_launch_fails_without_inventory(deploy_jobtemplate, post, user): @pytest.mark.django_db @pytest.mark.job_runtime_vars -def test_job_launch_fails_without_inventory_access(job_template_prompts, runtime_data, machine_credential, post, user, mocker): +def test_job_launch_fails_without_inventory_access(job_template_prompts, runtime_data, post, user): job_template = job_template_prompts(True) common_user = user('test-user', False) job_template.execute_role.members.add(common_user) - # Assure that the base job template can be launched to begin with - mock_job = mocker.MagicMock(spec=Job, id=968, **runtime_data) - with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job', return_value=mock_job): - with mocker.patch('awx.api.serializers.JobSerializer.to_representation'): - response = post(reverse('api:job_template_launch', - args=[job_template.pk]), {}, common_user) - - assert response.status_code == 201 - # Assure that giving an inventory without access to the inventory blocks the launch - new_inv = job_template.project.organization.inventories.create(name="user-can-not-use") + runtime_inventory = Inventory.objects.get(pk=runtime_data['inventory']) response = post(reverse('api:job_template_launch', args=[job_template.pk]), - dict(inventory=new_inv.pk), common_user) + dict(inventory=runtime_inventory.pk), common_user) assert response.status_code == 403 assert response.data['detail'] == u'You do not have permission to perform this action.' @pytest.mark.django_db @pytest.mark.job_runtime_vars -def test_job_relaunch_copy_vars(runtime_data, job_template_prompts, project, post, mocker): +def test_job_launch_fails_without_credential_access(job_template_prompts, runtime_data, post, user): job_template = job_template_prompts(True) + common_user = user('test-user', False) + job_template.execute_role.members.add(common_user) - # Create a job with the given data that will be relaunched - job_create_kwargs = runtime_data - inv_obj = Inventory.objects.get(pk=job_create_kwargs.pop('inventory')) - cred_obj = Credential.objects.get(pk=job_create_kwargs.pop('credential')) - original_job = Job.objects.create(inventory=inv_obj, credential=cred_obj, job_template=job_template, **job_create_kwargs) - with mocker.patch('awx.main.models.unified_jobs.UnifiedJobTemplate._get_unified_job_field_names', return_value=runtime_data.keys()): - second_job = original_job.copy() + # Assure that giving a credential without access blocks the launch + runtime_credential = Credential.objects.get(pk=runtime_data['credential']) + response = post(reverse('api:job_template_launch', args=[job_template.pk]), + dict(credential=runtime_credential.pk), common_user) + + assert response.status_code == 403 + assert response.data['detail'] == u'You do not have permission to perform this action.' + +@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() # Check that job data matches the original variables - assert 'job_launch_var' in yaml.load(second_job.extra_vars) - assert original_job.limit == second_job.limit - assert original_job.job_type == second_job.job_type - assert original_job.inventory.pk == second_job.inventory.pk - assert original_job.job_tags == second_job.job_tags + 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 can not 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 can not 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):