Merge pull request #1705 from AlanCoding/1696_cred_start

Change access.py to disallow launching without credential

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

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):