From 5dc17cd72f29b3338230cb57cd2e9fb561f9bb03 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Tue, 26 Apr 2016 17:03:11 -0400 Subject: [PATCH 1/5] New tests for user launching/relaunching with credential/inventory access --- awx/main/access.py | 3 +- .../functional/api/test_job_runtime_params.py | 39 ++++++++++++++++++- 2 files changed, 40 insertions(+), 2 deletions(-) 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..61c8bbea75 100644 --- a/awx/main/tests/functional/api/test_job_runtime_params.py +++ b/awx/main/tests/functional/api/test_job_runtime_params.py @@ -5,6 +5,7 @@ from awx.api.serializers import JobLaunchSerializer from awx.main.models.credential import Credential from awx.main.models.inventory import Inventory from awx.main.models.jobs import Job, JobTemplate +from awx.main.access import JobAccess from django.core.urlresolvers import reverse @@ -171,7 +172,8 @@ 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_or_cred_access( + job_template_prompts, runtime_data, machine_credential, post, user, mocker): job_template = job_template_prompts(True) common_user = user('test-user', False) job_template.execute_role.members.add(common_user) @@ -193,6 +195,14 @@ def test_job_launch_fails_without_inventory_access(job_template_prompts, runtime assert response.status_code == 403 assert response.data['detail'] == u'You do not have permission to perform this action.' + # Assure that giving a credential without access blocks the launch + new_cred = Credential.objects.create(name='machine-cred-you-cant-use', kind='ssh', username='test_user', password='pas4word') + response = post(reverse('api:job_template_launch', args=[job_template.pk]), + dict(credential=new_cred.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): @@ -213,6 +223,33 @@ def test_job_relaunch_copy_vars(runtime_data, job_template_prompts, project, pos assert original_job.inventory.pk == second_job.inventory.pk assert original_job.job_tags == second_job.job_tags +@pytest.mark.django_db +@pytest.mark.job_runtime_vars +def test_job_relaunch_resource_access(runtime_data, project, user): + the_cred = Credential.objects.get(pk=runtime_data['credential']) + the_inv = Inventory.objects.get(pk=runtime_data['inventory']) + + original_job = Job.objects.create( + name='existing-job', credential=the_cred, inventory=the_inv + ) + ordinary_user = user('commoner', False) + 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 + the_cred.use_role.members.add(both_user) + the_inv.use_role.members.add(both_user) + assert both_user.can_access(Job, 'start', original_job) + + # Confirm that a user with credential access alone can not launch + the_cred.use_role.members.add(credential_user) + assert not credential_user.can_access(Job, 'start', original_job) + + # Confirm that a user with inventory access alone can not launch + the_inv.use_role.members.add(inventory_user) + assert not inventory_user.can_access(Job, 'start', original_job) + @pytest.mark.django_db def test_job_launch_JT_with_validation(machine_credential, deploy_jobtemplate): deploy_jobtemplate.extra_vars = '{"job_template_var": 3}' From 271701a9689f87f2f133ad1786c574167a69e3bc Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Wed, 27 Apr 2016 10:43:03 -0400 Subject: [PATCH 2/5] clean up unused test code --- awx/main/tests/functional/api/test_job_runtime_params.py | 2 -- 1 file changed, 2 deletions(-) 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 61c8bbea75..7e0902fcd9 100644 --- a/awx/main/tests/functional/api/test_job_runtime_params.py +++ b/awx/main/tests/functional/api/test_job_runtime_params.py @@ -5,7 +5,6 @@ from awx.api.serializers import JobLaunchSerializer from awx.main.models.credential import Credential from awx.main.models.inventory import Inventory from awx.main.models.jobs import Job, JobTemplate -from awx.main.access import JobAccess from django.core.urlresolvers import reverse @@ -232,7 +231,6 @@ def test_job_relaunch_resource_access(runtime_data, project, user): original_job = Job.objects.create( name='existing-job', credential=the_cred, inventory=the_inv ) - ordinary_user = user('commoner', False) inventory_user = user('user1', False) credential_user = user('user2', False) both_user = user('user3', False) From a7311a69c999b3b48ad5a482b11342f0b4d4f0f2 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Wed, 27 Apr 2016 14:24:54 -0400 Subject: [PATCH 3/5] runtime test fixtures organized better --- .../functional/api/test_job_runtime_params.py | 51 ++++++++----------- 1 file changed, 22 insertions(+), 29 deletions(-) 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 7e0902fcd9..0be655a835 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): @@ -204,49 +208,38 @@ def test_job_launch_fails_without_inventory_or_cred_access( @pytest.mark.django_db @pytest.mark.job_runtime_vars -def test_job_relaunch_copy_vars(runtime_data, job_template_prompts, project, post, mocker): - job_template = job_template_prompts(True) - - # 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() +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(runtime_data, project, user): - the_cred = Credential.objects.get(pk=runtime_data['credential']) - the_inv = Inventory.objects.get(pk=runtime_data['inventory']) - - original_job = Job.objects.create( - name='existing-job', credential=the_cred, inventory=the_inv - ) +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 - the_cred.use_role.members.add(both_user) - the_inv.use_role.members.add(both_user) - assert both_user.can_access(Job, 'start', original_job) + 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 - the_cred.use_role.members.add(credential_user) - assert not credential_user.can_access(Job, 'start', original_job) + 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 - the_inv.use_role.members.add(inventory_user) - assert not inventory_user.can_access(Job, 'start', original_job) + 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): From 4af1e0a9bf63085569a2de101382244095065aca Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Wed, 27 Apr 2016 15:30:51 -0400 Subject: [PATCH 4/5] remove additional resource creation in test context --- .../functional/api/test_job_runtime_params.py | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) 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 0be655a835..e90c8913d2 100644 --- a/awx/main/tests/functional/api/test_job_runtime_params.py +++ b/awx/main/tests/functional/api/test_job_runtime_params.py @@ -176,32 +176,31 @@ 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_or_cred_access( - job_template_prompts, runtime_data, machine_credential, post, user, mocker): + job_template_prompts, runtime_data, post, user, mocker): 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) + mock_job = mocker.MagicMock(spec=Job, id=968) 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) + 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.' # Assure that giving a credential without access blocks the launch - new_cred = Credential.objects.create(name='machine-cred-you-cant-use', kind='ssh', username='test_user', password='pas4word') + runtime_credential = Credential.objects.get(pk=runtime_data['credential']) response = post(reverse('api:job_template_launch', args=[job_template.pk]), - dict(credential=new_cred.pk), common_user) + 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.' From d5a3951d52841c40b25d353f1b460590415e7a0a Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Wed, 27 Apr 2016 15:58:16 -0400 Subject: [PATCH 5/5] break up test into 2, isolating context --- .../functional/api/test_job_runtime_params.py | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) 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 e90c8913d2..bd79c7aab3 100644 --- a/awx/main/tests/functional/api/test_job_runtime_params.py +++ b/awx/main/tests/functional/api/test_job_runtime_params.py @@ -175,20 +175,11 @@ 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_or_cred_access( - job_template_prompts, runtime_data, 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) - 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 runtime_inventory = Inventory.objects.get(pk=runtime_data['inventory']) response = post(reverse('api:job_template_launch', args=[job_template.pk]), @@ -197,6 +188,13 @@ def test_job_launch_fails_without_inventory_or_cred_access( 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_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) + # 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]),