disallow launching with other users prompts

awx/main/access.py

@@ -1488,7 +1488,7 @@ class JobAccess(BaseAccess):
         # Obtain prompts used to start original job
         JobLaunchConfig = obj._meta.get_field('launch_config').related_model
         try:
-            config = obj.launch_config
+            config = JobLaunchConfig.objects.prefetch_related('credentials').get(job=obj)
         except JobLaunchConfig.DoesNotExist:
             config = None
 
@@ -1496,6 +1496,12 @@ class JobAccess(BaseAccess):
         if obj.job_template is not None:
             if config is None:
                 prompts_access = False
+            elif config.prompts_dict() == {}:
+                prompts_access = True
+            elif obj.created_by_id != self.user.pk:
+                prompts_access = False
+                if self.save_messages:
+                    self.messages['detail'] = _('Job was launched with prompts provided by another user.')
             else:
                 prompts_access = (
                     JobLaunchConfigAccess(self.user).can_add({'reference_obj': config}) and
@@ -1513,7 +1519,7 @@ class JobAccess(BaseAccess):
 
         # job can be relaunched if user could make an equivalent JT
         ret = org_access and credential_access and project_access
-        if not ret and self.save_messages:
+        if not ret and self.save_messages and not self.messages:
             if not obj.job_template:
                 pretext = _('Job has been orphaned from its job template.')
             elif config is None:

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

@@ -11,6 +11,8 @@ from awx.api.views import RelatedJobsPreventDeleteMixin, UnifiedJobDeletionMixin
 
 from awx.main.models import JobTemplate, User, Job
 
+from crum import impersonate
+
 
 @pytest.mark.django_db
 def test_extra_credentials(get, organization_factory, job_template_factory, credential):
@@ -33,6 +35,7 @@ def test_job_relaunch_permission_denied_response(
     jt.credentials.add(machine_credential)
     jt_user = User.objects.create(username='jobtemplateuser')
     jt.execute_role.members.add(jt_user)
+    with impersonate(jt_user):
         job = jt.create_unified_job()
 
     # User capability is shown for this
@@ -46,6 +49,29 @@ def test_job_relaunch_permission_denied_response(
     assert 'do not have permission' in r.data['detail']
 
 
+@pytest.mark.django_db
+def test_job_relaunch_permission_denied_response_other_user(get, post, inventory, project, alice, bob):
+    '''
+    Asserts custom permission denied message corresponding to
+    awx/main/tests/functional/test_rbac_job.py::TestJobRelaunchAccess::test_other_user_prompts
+    '''
+    jt = JobTemplate.objects.create(
+        name='testjt', inventory=inventory, project=project,
+        ask_credential_on_launch=True,
+        ask_variables_on_launch=True)
+    jt.execute_role.members.add(alice, bob)
+    with impersonate(bob):
+        job = jt.create_unified_job(extra_vars={'job_var': 'foo2'})
+
+    # User capability is shown for this
+    r = get(job.get_absolute_url(), alice, expect=200)
+    assert r.data['summary_fields']['user_capabilities']['start']
+
+    # Job has prompted data, launch denied w/ message
+    r = post(reverse('api:job_relaunch', kwargs={'pk':job.pk}), {}, alice, expect=403)
+    assert 'Job was launched with prompts provided by another user' in r.data['detail']
+
+
 @pytest.mark.django_db
 def test_job_relaunch_without_creds(post, inventory, project, admin_user):
     jt = JobTemplate.objects.create(

awx/main/tests/functional/test_rbac_job.py

@@ -19,6 +19,8 @@ from awx.main.models import (
     Credential
 )
 
+from crum import impersonate
+
 
 @pytest.fixture
 def normal_job(deploy_jobtemplate):
@@ -151,11 +153,14 @@ class TestJobRelaunchAccess:
             ask_inventory_on_launch=True,
             ask_credential_on_launch=True
         )
-        job_with_links = Job.objects.create(name='existing-job', inventory=inventory, job_template=job_template)
+        u = user('user1', False)
+        job_with_links = Job.objects.create(
+            name='existing-job', inventory=inventory, job_template=job_template,
+            created_by=u
+        )
         job_with_links.credentials.add(machine_credential)
         JobLaunchConfig.objects.create(job=job_with_links, inventory=inventory)
         job_with_links.launch_config.credentials.add(machine_credential)  # credential was prompted
-        u = user('user1', False)
         job_template.execute_role.members.add(u)
         if inv_access:
             job_with_links.inventory.use_role.members.add(u)
@@ -225,6 +230,7 @@ class TestJobRelaunchAccess:
 
     @pytest.mark.job_runtime_vars
     def test_callback_relaunchable_by_user(self, job_template, rando):
+        with impersonate(rando):
             job = job_template.create_unified_job(
                 _eager_fields={'launch_type': 'callback'},
                 limit='host2'
@@ -234,6 +240,20 @@ class TestJobRelaunchAccess:
         can_access, messages = rando.can_access_with_errors(Job, 'start', job, validate_license=False)
         assert can_access, messages
 
+    def test_other_user_prompts(self, inventory, project, alice, bob):
+        jt = JobTemplate.objects.create(
+            name='testjt', inventory=inventory, project=project,
+            ask_credential_on_launch=True,
+            ask_variables_on_launch=True)
+        jt.execute_role.members.add(alice, bob)
+
+        with impersonate(bob):
+            job = jt.create_unified_job(extra_vars={'job_var': 'foo2'})
+
+        assert 'job_var' in job.launch_config.extra_data
+        assert bob.can_access(Job, 'start', job, validate_license=False)
+        assert not alice.can_access(Job, 'start', job, validate_license=False)
+
 
 @pytest.mark.django_db
 class TestJobAndUpdateCancels:

awx/main/tests/functional/test_rbac_job_start.py

@@ -87,6 +87,7 @@ class TestJobRelaunchAccess:
         for cred in job_with_prompts.credentials.all():
             cred.use_role.members.add(rando)
         job_with_prompts.inventory.use_role.members.add(rando)
+        job_with_prompts.created_by = rando
         assert rando.can_access(Job, 'start', job_with_prompts)
 
     def test_no_relaunch_after_limit_change(self, inventory, machine_credential, rando):