JT RBAC edits for extra_credentials and vault_credential

awx/main/tests/functional/api/test_job_template.py

@@ -41,6 +41,8 @@ def test_create(post, project, machine_credential, inventory, alice, grant_proje
 def test_create_with_v1_deprecated_credentials(get, post, project, machine_credential, credential, net_credential, inventory, alice):
     project.use_role.members.add(alice)
     machine_credential.use_role.members.add(alice)
+    credential.use_role.members.add(alice)
+    net_credential.use_role.members.add(alice)
     inventory.use_role.members.add(alice)
 
     pk = post(reverse('api:job_template_list', kwargs={'version': 'v1'}), {
@@ -82,7 +84,27 @@ def test_create_with_empty_v1_deprecated_credentials(get, post, project, machine
     assert response.data.get('network_credential') is None
 
 
-# TODO: test this with RBAC and lower-priveleged users
+# TODO: remove in 3.3
+@pytest.mark.django_db
+def test_create_v1_rbac_check(get, post, project, credential, net_credential, rando):
+    project.use_role.members.add(rando)
+
+    base_kwargs = dict(
+        name = 'Made with cloud/net creds I have no access to',
+        project = project.id,
+        ask_inventory_on_launch = True,
+        ask_credential_on_launch = True,
+        playbook = 'helloworld.yml',
+    )
+
+    base_kwargs['cloud_credential'] = credential.pk
+    post(reverse('api:job_template_list', kwargs={'version': 'v1'}), base_kwargs, rando, expect=403)
+
+    base_kwargs.pop('cloud_credential')
+    base_kwargs['network_credential'] = net_credential.pk
+    post(reverse('api:job_template_list', kwargs={'version': 'v1'}), base_kwargs, rando, expect=403)
+
+
 @pytest.mark.django_db
 def test_extra_credential_creation(get, post, organization_factory, job_template_factory, credentialtype_aws):
     objs = organization_factory("org", superusers=['admin'])
@@ -140,7 +162,6 @@ def test_extra_credential_unique_type_xfail(get, post, organization_factory, job
     assert response.data.get('count') == 1
 
 
-# TODO: test this with RBAC and lower-priveleged users
 @pytest.mark.django_db
 def test_attach_extra_credential(get, post, organization_factory, job_template_factory, credential):
     objs = organization_factory("org", superusers=['admin'])
@@ -158,7 +179,6 @@ def test_attach_extra_credential(get, post, organization_factory, job_template_f
     assert response.data.get('count') == 1
 
 
-# TODO: test this with RBAC and lower-priveleged users
 @pytest.mark.django_db
 def test_detach_extra_credential(get, post, organization_factory, job_template_factory, credential):
     objs = organization_factory("org", superusers=['admin'])

awx/main/tests/functional/api/test_rbac_displays.py

@@ -307,24 +307,27 @@ def test_prefetch_group_capabilities(group, rando):
 
 
 @pytest.mark.django_db
-def test_prefetch_jt_copy_capability(job_template, project, inventory, machine_credential, rando):
+def test_prefetch_jt_copy_capability(job_template, project, inventory,
+                                     machine_credential, vault_credential, rando):
     job_template.project = project
     job_template.inventory = inventory
     job_template.credential = machine_credential
+    job_template.vault_credential = vault_credential
     job_template.save()
 
     qs = JobTemplate.objects.all()
     cache_list_capabilities(qs, [{'copy': [
-        'project.use', 'inventory.use', 'credential.use',
+        'project.use', 'inventory.use', 'credential.use', 'vault_credential.use'
     ]}], JobTemplate, rando)
     assert qs[0].capabilities_cache == {'copy': False}
 
     project.use_role.members.add(rando)
     inventory.use_role.members.add(rando)
     machine_credential.use_role.members.add(rando)
+    vault_credential.use_role.members.add(rando)
 
     cache_list_capabilities(qs, [{'copy': [
-        'project.use', 'inventory.use', 'credential.use',
+        'project.use', 'inventory.use', 'credential.use', 'vault_credential.use'
     ]}], JobTemplate, rando)
     assert qs[0].capabilities_cache == {'copy': True}
 

awx/main/tests/functional/conftest.py

@@ -209,6 +209,13 @@ def credentialtype_net():
     return net
 
 
+@pytest.fixture
+def credentialtype_vault():
+    vault_type = CredentialType.defaults['vault']()
+    vault_type.save()
+    return vault_type
+
+
 @pytest.fixture
 def credential(credentialtype_aws):
     return Credential.objects.create(credential_type=credentialtype_aws, name='test-cred',
@@ -221,6 +228,12 @@ def net_credential(credentialtype_net):
                                      inputs={'username': 'something', 'password': 'secret'})
 
 
+@pytest.fixture
+def vault_credential(credentialtype_vault):
+    return Credential.objects.create(credential_type=credentialtype_vault, name='test-cred',
+                                     inputs={'vault_password': 'secret'})
+
+
 @pytest.fixture
 def machine_credential(credentialtype_ssh):
     return Credential.objects.create(credential_type=credentialtype_ssh, name='machine-cred',

awx/main/tests/functional/test_rbac_job_templates.py

@@ -12,11 +12,21 @@ from awx.main.models.schedules import Schedule
 
 
 @pytest.fixture
-def jt_objects(job_template_factory):
+def jt_linked(job_template_factory, credential, net_credential, vault_credential):
+    '''
+    A job template with a reasonably complete set of related objects to
+    test RBAC and other functionality affected by related objects
+    '''
     objects = job_template_factory(
         'testJT', organization='org1', project='proj1', inventory='inventory1',
-        credential='cred1', cloud_credential='aws1', network_credential='juniper1')
-    return objects
+        credential='cred1')
+    jt = objects.job_template
+    jt.vault_credential = vault_credential
+    jt.save()
+    # Add AWS cloud credential and network credential
+    jt.extra_credentials.add(credential)
+    jt.extra_credentials.add(net_credential)
+    return jt
 
 
 @mock.patch.object(BaseAccess, 'check_license', return_value=None)
@@ -32,58 +42,86 @@ def test_job_template_access_superuser(check_license, user, deploy_jobtemplate):
 
 
 @pytest.mark.django_db
-def test_job_template_access_read_level(jt_objects, rando):
+def test_job_template_access_read_level(jt_linked, rando):
 
     access = JobTemplateAccess(rando)
-    jt_objects.project.read_role.members.add(rando)
-    jt_objects.inventory.read_role.members.add(rando)
-    jt_objects.credential.read_role.members.add(rando)
-    jt_objects.cloud_credential.read_role.members.add(rando)
-    jt_objects.network_credential.read_role.members.add(rando)
+    jt_linked.project.read_role.members.add(rando)
+    jt_linked.inventory.read_role.members.add(rando)
+    jt_linked.credential.read_role.members.add(rando)
 
-    proj_pk = jt_objects.project.pk
-    assert not access.can_add(dict(inventory=jt_objects.inventory.pk, project=proj_pk))
-    assert not access.can_add(dict(credential=jt_objects.credential.pk, project=proj_pk))
-    assert not access.can_add(dict(cloud_credential=jt_objects.cloud_credential.pk, project=proj_pk))
-    assert not access.can_add(dict(network_credential=jt_objects.network_credential.pk, project=proj_pk))
+    proj_pk = jt_linked.project.pk
+    assert not access.can_add(dict(inventory=jt_linked.inventory.pk, project=proj_pk))
+    assert not access.can_add(dict(credential=jt_linked.credential.pk, project=proj_pk))
+    assert not access.can_add(dict(vault_credential=jt_linked.vault_credential.pk, project=proj_pk))
+
+    for cred in jt_linked.extra_credentials.all():
+        assert not access.can_unattach(jt_linked, cred, 'extra_credentials', {})
 
 
 @pytest.mark.django_db
-def test_job_template_access_use_level(jt_objects, rando):
+def test_job_template_access_use_level(jt_linked, rando):
 
     access = JobTemplateAccess(rando)
-    jt_objects.project.use_role.members.add(rando)
-    jt_objects.inventory.use_role.members.add(rando)
-    jt_objects.credential.use_role.members.add(rando)
-    jt_objects.cloud_credential.use_role.members.add(rando)
-    jt_objects.network_credential.use_role.members.add(rando)
+    jt_linked.project.use_role.members.add(rando)
+    jt_linked.inventory.use_role.members.add(rando)
+    jt_linked.credential.use_role.members.add(rando)
+    jt_linked.vault_credential.use_role.members.add(rando)
 
-    proj_pk = jt_objects.project.pk
-    assert access.can_add(dict(inventory=jt_objects.inventory.pk, project=proj_pk))
-    assert access.can_add(dict(credential=jt_objects.credential.pk, project=proj_pk))
-    assert access.can_add(dict(cloud_credential=jt_objects.cloud_credential.pk, project=proj_pk))
-    assert access.can_add(dict(network_credential=jt_objects.network_credential.pk, project=proj_pk))
+    proj_pk = jt_linked.project.pk
+    assert access.can_add(dict(inventory=jt_linked.inventory.pk, project=proj_pk))
+    assert access.can_add(dict(credential=jt_linked.credential.pk, project=proj_pk))
+    assert access.can_add(dict(vault_credential=jt_linked.vault_credential.pk, project=proj_pk))
+
+    for cred in jt_linked.extra_credentials.all():
+        assert not access.can_unattach(jt_linked, cred, 'extra_credentials', {})
 
 
 @pytest.mark.django_db
-def test_job_template_access_org_admin(jt_objects, rando):
+def test_job_template_access_org_admin(jt_linked, rando):
     access = JobTemplateAccess(rando)
     # Appoint this user as admin of the organization
-    jt_objects.inventory.organization.admin_role.members.add(rando)
+    jt_linked.inventory.organization.admin_role.members.add(rando)
     # Assign organization permission in the same way the create view does
-    organization = jt_objects.inventory.organization
-    jt_objects.credential.admin_role.parents.add(organization.admin_role)
-    jt_objects.cloud_credential.admin_role.parents.add(organization.admin_role)
-    jt_objects.network_credential.admin_role.parents.add(organization.admin_role)
+    organization = jt_linked.inventory.organization
+    jt_linked.credential.admin_role.parents.add(organization.admin_role)
 
-    proj_pk = jt_objects.project.pk
-    assert access.can_add(dict(inventory=jt_objects.inventory.pk, project=proj_pk))
-    assert access.can_add(dict(credential=jt_objects.credential.pk, project=proj_pk))
-    assert access.can_add(dict(cloud_credential=jt_objects.cloud_credential.pk, project=proj_pk))
-    assert access.can_add(dict(network_credential=jt_objects.network_credential.pk, project=proj_pk))
+    proj_pk = jt_linked.project.pk
+    assert access.can_add(dict(inventory=jt_linked.inventory.pk, project=proj_pk))
+    assert access.can_add(dict(credential=jt_linked.credential.pk, project=proj_pk))
 
-    assert access.can_read(jt_objects.job_template)
-    assert access.can_delete(jt_objects.job_template)
+    for cred in jt_linked.extra_credentials.all():
+        assert access.can_unattach(jt_linked, cred, 'extra_credentials', {})
+
+    assert access.can_read(jt_linked)
+    assert access.can_delete(jt_linked)
+
+
+@pytest.mark.django_db
+class TestJobTemplateCredentials:
+
+    def test_job_template_cannot_add_extra_credentials(self, job_template, credential, rando):
+        job_template.admin_role.members.add(rando)
+        credential.read_role.members.add(rando)
+        # without permission to credential, user can not attach it
+        assert not JobTemplateAccess(rando).can_attach(
+            job_template, credential, 'extra_credentials', {})
+
+    def test_job_template_can_add_extra_credentials(self, job_template, credential, rando):
+        job_template.admin_role.members.add(rando)
+        credential.use_role.members.add(rando)
+        # user has permission to apply credential
+        assert JobTemplateAccess(rando).can_attach(
+            job_template, credential, 'extra_credentials', {})
+
+    def test_job_template_vault_cred_check(self, job_template, vault_credential, rando):
+        job_template.admin_role.members.add(rando)
+        # not allowed to use the vault cred
+        assert not JobTemplateAccess(rando).can_change(
+            job_template, {'vault_credential': vault_credential})
+
+    def test_new_jt_with_vault(self, vault_credential, project, rando):
+        project.admin_role.members.add(rando)
+        assert not JobTemplateAccess(rando).can_add({'vault_credential': vault_credential, 'project': project.pk})
 
 
 @pytest.mark.django_db

awx/main/tests/unit/test_access.py

@@ -123,6 +123,7 @@ def job_template_with_ids(job_template_factory):
     credential = Credential(id=1, pk=1, name='testcred', kind='ssh')
     net_cred = Credential(id=2, pk=2, name='testnetcred', kind='net')
     cloud_cred = Credential(id=3, pk=3, name='testcloudcred', kind='aws')
+    vault_cred = Credential(id=4, pk=4, name='testnetcred', kind='vault')
     inv = Inventory(id=11, pk=11, name='testinv')
     proj = Project(id=14, pk=14, name='testproj')
 
@@ -130,6 +131,7 @@ def job_template_with_ids(job_template_factory):
         'testJT', project=proj, inventory=inv, credential=credential,
         cloud_credential=cloud_cred, network_credential=net_cred,
         persisted=False)
+    jt_objects.job_template.vault_credential = vault_cred
     return jt_objects.job_template
 
 
@@ -159,7 +161,6 @@ def test_jt_existing_values_are_nonsensitive(job_template_with_ids, user_unit):
     assert access.changes_are_non_sensitive(job_template_with_ids, data)
 
 
-@pytest.mark.xfail  # TODO: update this to respect JT.extra_credentials
 def test_change_jt_sensitive_data(job_template_with_ids, mocker, user_unit):
     """Assure that can_add is called with all ForeignKeys."""
 
@@ -178,8 +179,7 @@ def test_change_jt_sensitive_data(job_template_with_ids, mocker, user_unit):
         'inventory': data['inventory'],
         'project': job_template_with_ids.project.id,
         'credential': job_template_with_ids.credential.id,
-        'cloud_credential': job_template_with_ids.cloud_credential.id,
-        'network_credential': job_template_with_ids.network_credential.id
+        'vault_credential': job_template_with_ids.vault_credential.id
     })