hide passwords in job_args and activity stream

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

@@ -1,9 +1,11 @@
 import mock
 import pytest
+import json
 
 from django.core.urlresolvers import reverse
 
 from awx.main.models.jobs import JobTemplate, Job
+from awx.main.models.activity_stream import ActivityStream
 from awx.api.license import LicenseForbids
 from awx.main.access import JobTemplateAccess
 
@@ -33,7 +35,7 @@ def test_survey_spec_view_denied(job_template_with_survey, get, admin_user):
 @pytest.mark.django_db
 @pytest.mark.survey
 def test_deny_enabling_survey(deploy_jobtemplate, patch, admin_user):
-    response = patch(url=reverse('api:job_template_detail', args=(deploy_jobtemplate.id,)),
+    response = patch(url=deploy_jobtemplate.get_absolute_url(),
                      data=dict(survey_enabled=True), user=admin_user, expect=402)
     assert response.data['detail'] == 'Feature surveys is not enabled in the active license.'
 
@@ -63,6 +65,7 @@ def test_deny_creating_with_survey(project, post, admin_user):
         user=admin_user, expect=402)
     assert response.data['detail'] == 'Feature surveys is not enabled in the active license.'
 
+# Test normal operations with survey license work
 @mock.patch('awx.api.views.feature_enabled', lambda feature: True)
 @pytest.mark.django_db
 @pytest.mark.survey
@@ -80,6 +83,7 @@ def test_survey_spec_sucessful_creation(survey_spec_factory, job_template, post,
     updated_jt = JobTemplate.objects.get(pk=job_template.pk)
     assert updated_jt.survey_spec == survey_input_data
 
+# Tests related to survey content validation
 @mock.patch('awx.api.views.feature_enabled', lambda feature: True)
 @pytest.mark.django_db
 @pytest.mark.survey
@@ -102,6 +106,7 @@ def test_survey_spec_dual_names_error(survey_spec_factory, deploy_jobtemplate, p
         user=user('admin', True), expect=400)
     assert response.data['error'] == "'variable' 'submitter_email' duplicated in survey question 1."
 
+# Test actions that should be allowed with non-survey license
 @mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
 @pytest.mark.django_db
 @pytest.mark.survey
@@ -165,8 +170,9 @@ def test_launch_survey_enabled_but_no_survey_spec(job_template_factory, post, ad
     obj = objects.job_template
     obj.survey_enabled = True
     obj.save()
-    post(reverse('api:job_template_launch', args=[obj.pk]),
-         dict(extra_vars=dict(survey_var=7)), admin_user, expect=201)
+    response = post(reverse('api:job_template_launch', args=[obj.pk]),
+                    dict(extra_vars=dict(survey_var=7)), admin_user, expect=201)
+    assert 'survey_var' in response.data['ignored_fields']['extra_vars']
 
 @mock.patch('awx.main.access.BaseAccess.check_license', new=mock_no_surveys)
 @mock.patch('awx.main.models.unified_jobs.UnifiedJobTemplate.create_unified_job',
@@ -174,12 +180,21 @@ def test_launch_survey_enabled_but_no_survey_spec(job_template_factory, post, ad
 @mock.patch('awx.api.serializers.JobSerializer.to_representation', lambda self, obj: {})
 @pytest.mark.django_db
 @pytest.mark.survey
-def test_launch_with_non_empty_survey_spec_no_license(survey_spec_factory, job_template_factory, post, admin_user):
+def test_launch_with_non_empty_survey_spec_no_license(job_template_factory, post, admin_user):
     """Assure jobs can still be launched from JTs with a survey_spec
     when the survey is diabled."""
     objects = job_template_factory('jt', organization='org1', project='prj',
-                                   inventory='inv', credential='cred')
+                                   inventory='inv', credential='cred',
+                                   survey='survey_var')
     obj = objects.job_template
-    obj.survey_spec = survey_spec_factory('survey_var')
+    obj.survey_enabled = False
     obj.save()
     post(reverse('api:job_template_launch', args=[obj.pk]), {}, admin_user, expect=201)
+
+@pytest.mark.django_db
+@pytest.mark.survey
+def test_redact_survey_passwords_in_activity_stream(job_with_secret_key):
+    AS_record = ActivityStream.objects.filter(object1='job').all()[0]
+    changes_dict = json.loads(AS_record.changes)
+    extra_vars = json.loads(changes_dict['extra_vars'])
+    assert extra_vars['secret_key'] == '$encrypted$'

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

@@ -1,4 +1,5 @@
 import pytest
+import json
 
 from awx.main.models import UnifiedJob
 from django.core.urlresolvers import reverse
@@ -15,3 +16,8 @@ def test_options_fields_choices(instance, options, user):
     assert 'choice' == response.data['actions']['GET']['status']['type']
     assert UnifiedJob.STATUS_CHOICES == response.data['actions']['GET']['status']['choices']
 
+@pytest.mark.django_db
+@pytest.mark.survey
+def test_job_redacted_survey_passwords(job_with_secret_key, get, admin_user):
+    response = get(reverse('api:job_detail', args=(job_with_secret_key.pk,)), admin_user)
+    assert json.loads(response.data['extra_vars'])['secret_key'] == '$encrypted$'

awx/main/tests/functional/conftest.py

@@ -185,6 +185,16 @@ def notification_template(organization):
                                                notification_type="webhook",
                                                notification_configuration=dict(url="http://localhost",
                                                                                headers={"Test": "Header"}))
+
+@pytest.fixture
+def job_with_secret_key(job_template_factory):
+    "Returns a job from a Job Template with secret_key as password question"
+    objects = job_template_factory(
+        'jt', organization='org1', survey=[{'variable': 'secret_key', 'type': 'password'}])
+    job = objects.job_template.jobs.create(
+        job_type="run", extra_vars=json.dumps({'secret_key': '6kQngg3h8lgiSTvIEb21'}))
+    return job
+
 @pytest.fixture
 def admin(user):
     return user('admin', True)

awx/main/tests/unit/models/test_job_template_unit.py

@@ -1,5 +1,18 @@
-from awx.main.tests.factories import create_job_template
+import pytest
+import json
 
+from awx.main.tests.factories import create_job_template
+from awx.main.models.jobs import Job
+
+
+@pytest.fixture
+def job_template_sensitive_data():
+    return create_job_template(
+        'jt', project='prj', persisted=False,
+        survey=['submitter_email',
+                {'variable': 'secret_key', 'type': 'password'},
+                {'variable': 'SSN', 'type': 'password'}]
+    ).job_template
 
 def test_missing_project_error():
     objects = create_job_template(
@@ -33,3 +46,23 @@ def test_inventory_credential_contradictions():
     validation_errors, resources_needed_to_start = obj.resource_validation_data()
     assert 'inventory' in validation_errors
     assert 'credential' in validation_errors
+
+@pytest.mark.survey
+def test_survey_password_list(job_template_sensitive_data):
+    """Verify the output of the survey_passwords function
+    gives a list of survey variable names which are passwords"""
+    assert job_template_sensitive_data.survey_password_variables() == ['secret_key', 'SSN']
+
+@pytest.mark.survey
+def test_job_redacted_extra_vars(job_template_sensitive_data):
+    """Verify that this method redacts vars marked as passwords in a survey"""
+    job_obj = Job(
+        job_type="run", job_template=job_template_sensitive_data,
+        extra_vars=json.dumps({'submitter_email': 'foobar@redhat.com',
+                               'secret_key': 'b86hpFChM2XSb40Zld9x',
+                               'SSN': '123-45-6789'}))
+    assert json.loads(job_obj.display_extra_vars()) == {
+        'submitter_email': 'foobar@redhat.com',
+        'secret_key': '$encrypted$',
+        'SSN': '$encrypted$'
+    }

awx/main/tests/unit/models/test_job_unit.py

@@ -0,0 +1,40 @@
+import pytest
+import json
+
+from awx.main.models.jobs import Job
+from awx.main.tasks import RunJob
+
+from awx.main.tests.factories import create_job_template
+
+
+@pytest.fixture
+def job_with_secret_vars():
+    job_template = create_job_template(
+        'jt', persisted=False,
+        survey=['submitter_email',
+                {'variable': 'secret_key', 'type': 'password'}]
+    ).job_template
+    job = Job(id=1, job_template=job_template, extra_vars=json.dumps({
+        'submitter_email': 'foobar@redhat.com',
+        'secret_key': '6kQngg3h8lgiSTvIEb21'
+    }))
+    return job
+
+def test_job_args_redacted_passwords(job_with_secret_vars):
+    """Verify that safe_args hides passwords in the job extra_vars"""
+    kwargs = {'ansible_version': '2.1'}
+    run_job = RunJob()
+    safe_args = run_job.build_safe_args(job_with_secret_vars, **kwargs)
+    ev_index = safe_args.index('-e') + 1
+    extra_vars = json.loads(safe_args[ev_index])
+    assert extra_vars['secret_key'] == '$encrypted$'
+    assert extra_vars['submitter_email'] == 'foobar@redhat.com'
+
+def test_job_args_unredacted_passwords(job_with_secret_vars):
+    kwargs = {'ansible_version': '2.1'}
+    run_job = RunJob()
+    safe_args = run_job.build_args(job_with_secret_vars, **kwargs)
+    ev_index = safe_args.index('-e') + 1
+    extra_vars = json.loads(safe_args[ev_index])
+    assert extra_vars['secret_key'] == '6kQngg3h8lgiSTvIEb21'
+    assert extra_vars['submitter_email'] == 'foobar@redhat.com'