diff --git a/awx/api/metadata.py b/awx/api/metadata.py index 53999574a8..b07385fe01 100644 --- a/awx/api/metadata.py +++ b/awx/api/metadata.py @@ -23,6 +23,7 @@ from rest_framework.request import clone_request # AWX from awx.api.fields import ChoiceNullField +from awx.api.validation_patterns import build_survey_spec_options_schema, inject_patterns_into_init_parameters, inject_top_level_clean_text_patterns from awx.main.fields import ImplicitRoleField from awx.main.models import NotificationTemplate from awx.main.utils.execution_environments import get_default_pod_spec @@ -131,7 +132,9 @@ class Metadata(metadata.SimpleMetadata): # are conditional on the type selected. if field.field_name == 'notification_configuration': for notification_type_name, notification_tr_name, notification_type_class in NotificationTemplate.NOTIFICATION_TYPES: - field_info[notification_type_name] = notification_type_class.init_parameters + # Copy before injecting patterns so class-level init_parameters + # dicts are never mutated (AAP-87586). + field_info[notification_type_name] = inject_patterns_into_init_parameters(notification_type_class.init_parameters) # Special handling of notification messages where the required properties # are conditional on the type selected. @@ -178,6 +181,10 @@ class Metadata(metadata.SimpleMetadata): elif isinstance(model_field, BooleanField): field_info['type'] = 'boolean' + # Advertise CleanTextMixin Tier 1/Tier 2 patterns on top-level CharFields (AAP-87586). + # No-op unless ENHANCED_INPUT_VALIDATION_ENABLED is on and the serializer mixes in CleanTextMixin. + field_info = inject_top_level_clean_text_patterns(field, field_info) + return field_info def get_serializer_info(self, serializer, method=None): @@ -240,6 +247,11 @@ class Metadata(metadata.SimpleMetadata): continue actions[method].pop(field) + # Survey spec uses EmptySerializer, so POST has no field metadata unless we + # supply the JSON sub-key catalog (AAP-87586). + if getattr(view, 'advertises_survey_spec_schema', False) and 'POST' in actions: + actions['POST'] = build_survey_spec_options_schema() + return actions def determine_metadata(self, request, view): diff --git a/awx/api/serializers.py b/awx/api/serializers.py index 1c298e0d05..8b9ff33325 100644 --- a/awx/api/serializers.py +++ b/awx/api/serializers.py @@ -51,6 +51,7 @@ from ansible_base.rbac.models import RoleEvaluation, ObjectRole from ansible_base.rbac import permission_registry # AWX +from awx.api.validation_patterns import inject_patterns_into_field_list from awx.main.access import get_user_capabilities from awx.main.constants import ACTIVE_STATES, org_role_to_permission from awx.main.models import ( @@ -3122,6 +3123,11 @@ class CredentialTypeSerializer(CleanTextMixin, BaseSerializer): # Normalize fields and filter out internal fields value['inputs']['fields'] = [f for f in fields if not f.get('internal')] + # Advertise CleanTextMixin Tier 2 patterns on JSON sub-keys (AAP-87586). + # Gated on ENHANCED_INPUT_VALIDATION_ENABLED; secret fields are skipped. + inject_patterns_into_field_list(value['inputs']['fields']) + inject_patterns_into_field_list(value['inputs'].get('metadata')) + return value def filter_field_metadata(self, fields, method): diff --git a/awx/api/validation_patterns.py b/awx/api/validation_patterns.py new file mode 100644 index 0000000000..dfa1e4869a --- /dev/null +++ b/awx/api/validation_patterns.py @@ -0,0 +1,221 @@ +# Copyright (c) 2026 Ansible, Inc. +# All Rights Reserved. + +"""Inject DAB CleanTextMixin frontend patterns into API OPTIONS metadata. + +Controller defines its own ``DEFAULT_METADATA_CLASS`` (``awx.api.metadata.Metadata``) +rather than using DAB's ``CleanTextMetadata``, so top-level CharField OPTIONS +metadata is not handled automatically by DAB -- ``inject_top_level_clean_text_patterns`` +below is called directly from ``awx.api.metadata.Metadata.get_field_info`` to cover it. +JSON sub-keys (credential inputs, notification config, credential input source +metadata, survey_spec question fields) are domain-owned schemas that DAB has no +visibility into at all, so pattern injection for those lives here unconditionally. + +Depends on django-ansible-base PR #1119 (AAP-85987) for +``build_tier2_frontend_pattern`` and ``inject_clean_text_patterns``. Until that +lands on DAB devel, injection is a no-op so Controller can still import and run. +""" + +import copy + +from ansible_base.lib.utils.settings import get_setting + +try: + from ansible_base.lib.metadata import build_tier2_frontend_pattern +except ImportError: # pragma: no cover - DAB without AAP-85987 + build_tier2_frontend_pattern = None + +try: + from ansible_base.lib.metadata import inject_clean_text_patterns as _dab_inject_clean_text_patterns +except ImportError: # pragma: no cover - DAB without AAP-85987 + _dab_inject_clean_text_patterns = None + +try: + from ansible_base.lib.metadata import TIER2_PATTERN_DESCRIPTION +except ImportError: # pragma: no cover - DAB without AAP-85987 + TIER2_PATTERN_DESCRIPTION = "This field can't include HTML tags, script markup, unsafe URI schemes, shell or template syntax, or control characters." + +try: + from ansible_base.lib.utils.validation import validate_free_text as _validate_free_text +except ImportError: # pragma: no cover - DAB without CleanTextMixin validators + _validate_free_text = None + +_STRING_TYPES = frozenset({'string', 'str'}) + +# Survey editor strings advertised on OPTIONS /survey_spec/ and validated on POST. +# default/choices are omitted: they may contain Jinja/playbook text or password values. +SURVEY_SPEC_TOP_LEVEL_TEXT_KEYS = ('name', 'description') +SURVEY_QUESTION_TEXT_KEYS = ('question_name', 'question_description', 'variable') + + +def enhanced_input_validation_enabled(): + return bool(get_setting('ENHANCED_INPUT_VALIDATION_ENABLED', False)) + + +def free_text_pattern_metadata(): + """Return Tier 2 (validate_free_text) pattern keys for API clients, or None.""" + if build_tier2_frontend_pattern is None: + return None + return { + 'pattern': build_tier2_frontend_pattern(), + 'pattern_description': TIER2_PATTERN_DESCRIPTION, + 'flags': 'i', + } + + +def _is_string_schema(field_schema, field_type=None): + if field_type is None: + field_type = field_schema.get('type', 'string') + return field_type in _STRING_TYPES + + +def inject_free_text_pattern(field_schema, *, secret=False, field_type=None): + """Mutate a schema dict in place if it is a non-secret string field. + + No-op when the install-time toggle is off or DAB pattern helpers are missing. + """ + if not isinstance(field_schema, dict): + return field_schema + if not enhanced_input_validation_enabled(): + return field_schema + if secret or field_schema.get('secret') is True: + return field_schema + if not _is_string_schema(field_schema, field_type=field_type): + return field_schema + + metadata = free_text_pattern_metadata() + if metadata is None: + return field_schema + field_schema.update(metadata) + return field_schema + + +def inject_patterns_into_field_list(fields): + """Inject patterns into a credential-type ``fields`` or ``metadata`` list in place. + + Each field is shallow-copied before mutation so shared/module-level schema + dicts (e.g. ``ManagedCredentialType.registry`` entries) are never modified. + """ + if not enhanced_input_validation_enabled(): + return + if not isinstance(fields, list): + return + for i, field in enumerate(fields): + fields[i] = inject_free_text_pattern(copy.copy(field)) + + +def inject_patterns_into_init_parameters(init_parameters): + """Return a copy of notification ``init_parameters`` with patterns injected. + + Copies first so class-level backend dicts are never mutated. + Password-typed keys are treated as secrets and skipped. + """ + params = copy.deepcopy(init_parameters) if isinstance(init_parameters, dict) else {} + if not enhanced_input_validation_enabled(): + return params + for field_schema in params.values(): + if not isinstance(field_schema, dict): + continue + inject_free_text_pattern(field_schema, secret=field_schema.get('type') == 'password') + return params + + +def _survey_string_field_schema(*, required, label): + schema = {'type': 'string', 'required': required, 'label': label} + inject_free_text_pattern(schema) + return schema + + +def build_survey_spec_options_schema(): + """Return OPTIONS ``actions.POST`` metadata for ``/survey_spec/``. + + ``JobTemplateSurveySpec`` uses ``EmptySerializer``, so DRF OPTIONS has no + nested field schema to stamp. This builds the JSON sub-key catalog the UI + survey editor needs (survey title plus question editor strings). + ``default`` / ``choices`` are not included — they may contain Jinja, + playbook text, or password values. + """ + return { + 'name': _survey_string_field_schema(required=True, label='Name'), + 'description': _survey_string_field_schema(required=True, label='Description'), + 'spec': { + 'type': 'json', + 'required': True, + 'label': 'Spec', + 'question_name': _survey_string_field_schema(required=True, label='Question'), + 'question_description': _survey_string_field_schema(required=False, label='Description'), + 'variable': _survey_string_field_schema(required=True, label='Answer variable name'), + }, + } + + +def _survey_questions_list(spec): + questions = spec.get('spec') if isinstance(spec, dict) else None + return questions if isinstance(questions, list) else [] + + +def _survey_questions_by_variable(questions): + return {q['variable']: q for q in questions if isinstance(q, dict) and isinstance(q.get('variable'), str)} + + +def _matching_old_survey_question(item, idx, old_questions, old_by_var): + """Resolve the prior question for grandfathering (by variable, then index).""" + old_item = old_by_var.get(item.get('variable')) + if old_item is None and idx < len(old_questions) and isinstance(old_questions[idx], dict): + return old_questions[idx] + return old_item or {} + + +def _record_survey_free_text_error(errors, error_key, value, old_value): + """Validate ``value`` with Tier 2 rules unless unchanged from ``old_value``.""" + if not isinstance(value, str) or value == old_value: + return + from rest_framework.serializers import ValidationError + + try: + _validate_free_text(value) + except ValidationError as exc: + errors[error_key] = exc.detail + + +def _collect_survey_question_text_errors(errors, new_questions, old_questions): + old_by_var = _survey_questions_by_variable(old_questions) + for idx, item in enumerate(new_questions): + if not isinstance(item, dict): + continue + old_item = _matching_old_survey_question(item, idx, old_questions, old_by_var) + for field in SURVEY_QUESTION_TEXT_KEYS: + _record_survey_free_text_error(errors, f'spec[{idx}].{field}', item.get(field), old_item.get(field)) + + +def collect_survey_spec_text_errors(new_spec, old_spec=None): + """Return field-keyed Tier 2 errors for survey editor strings, or None. + + Grandfathers unchanged values against ``old_spec`` (by variable, then index). + No-op when the install-time toggle is off or DAB validators are missing. + """ + if not enhanced_input_validation_enabled() or _validate_free_text is None: + return None + if not isinstance(new_spec, dict): + return None + + old_spec = old_spec if isinstance(old_spec, dict) else {} + errors = {} + + for field in SURVEY_SPEC_TOP_LEVEL_TEXT_KEYS: + _record_survey_free_text_error(errors, field, new_spec.get(field), old_spec.get(field)) + + _collect_survey_question_text_errors(errors, _survey_questions_list(new_spec), _survey_questions_list(old_spec)) + return errors or None + + +def inject_top_level_clean_text_patterns(field, field_info): + """Advertise DAB CleanTextMixin Tier 1/Tier 2 patterns on a top-level serializer field. + + Delegates entirely to DAB's ``inject_clean_text_patterns``, which no-ops unless + ``ENHANCED_INPUT_VALIDATION_ENABLED`` is on and the field's serializer mixes in + ``CleanTextMixin``. No-op when the DAB helper is missing. + """ + if _dab_inject_clean_text_patterns is None: + return field_info + return _dab_inject_clean_text_patterns(field, field_info) diff --git a/awx/api/views/__init__.py b/awx/api/views/__init__.py index b21d007220..b2c202ddda 100644 --- a/awx/api/views/__init__.py +++ b/awx/api/views/__init__.py @@ -118,6 +118,7 @@ from awx.api.permissions import ( ) from awx.api import renderers from awx.api import serializers +from awx.api.validation_patterns import collect_survey_spec_text_errors from awx.api.metadata import RoleMetadata from awx.main.constants import ACTIVE_STATES, SURVEY_TYPE_MAPPING from awx.main.scheduler.dag_workflow import WorkflowDAG @@ -2815,6 +2816,7 @@ class JobTemplateSurveySpec(GenericAPIView): obj_permission_type = 'admin' serializer_class = serializers.EmptySerializer resource_purpose = 'job template survey specification' + advertises_survey_spec_schema = True @extend_schema_if_available(extensions={"x-ai-description": "Get job template survey specification"}) def get(self, request, *args, **kwargs): @@ -2985,6 +2987,10 @@ class JobTemplateSurveySpec(GenericAPIView): # Submission provides new encrypted default survey_item['default'] = encrypt_value(survey_item['default']) + text_errors = collect_survey_spec_text_errors(new_spec, old_spec) + if text_errors: + return Response(text_errors, status=status.HTTP_400_BAD_REQUEST) + @extend_schema_if_available(extensions={"x-ai-description": "Delete job template survey specification"}) def delete(self, request, *args, **kwargs): obj = self.get_object() diff --git a/awx/main/tests/functional/api/test_credential_type.py b/awx/main/tests/functional/api/test_credential_type.py index bc1637079f..618844089a 100644 --- a/awx/main/tests/functional/api/test_credential_type.py +++ b/awx/main/tests/functional/api/test_credential_type.py @@ -1,6 +1,7 @@ import json import pytest +from django.test import override_settings from ansible_base.lib.testing.util import feature_flag_enabled from awx.main.models.credential import CredentialType, Credential @@ -571,3 +572,67 @@ def test_credential_type_list_filters_internal_fields(get, admin): # Internal field should be filtered out assert 'workload_identity_token' not in field_ids assert 'url' in field_ids + + +FAKE_TIER2_PATTERN = r'^(?!.*<[a-zA-Z/!][^>]*>)[\s\S]*$' + + +@pytest.fixture +def fake_tier2_pattern(monkeypatch): + monkeypatch.setattr('awx.api.validation_patterns.build_tier2_frontend_pattern', lambda: FAKE_TIER2_PATTERN) + + +@pytest.mark.django_db +def test_credential_type_fields_include_patterns_when_toggle_on(get, admin, fake_tier2_pattern): + ct = CredentialType.objects.create( + kind='cloud', + name='Pattern Type', + managed=False, + inputs={ + 'fields': [ + {'id': 'username', 'label': 'Username', 'type': 'string'}, + {'id': 'api_token', 'label': 'API Token', 'type': 'string', 'secret': True}, + {'id': 'verify_ssl', 'label': 'Verify SSL', 'type': 'boolean'}, + ] + }, + ) + url = reverse('api:credential_type_detail', kwargs={'pk': ct.pk}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + response = get(url, admin) + assert response.status_code == 200 + fields = {f['id']: f for f in response.data['inputs']['fields']} + assert fields['username']['pattern'] == FAKE_TIER2_PATTERN + assert 'pattern_description' in fields['username'] + assert 'pattern' not in fields['api_token'] + assert 'pattern' not in fields['verify_ssl'] + + +@pytest.mark.django_db +def test_credential_type_fields_omit_patterns_when_toggle_off(get, admin, fake_tier2_pattern): + ct = CredentialType.objects.create( + kind='cloud', + name='Pattern Type Off', + managed=False, + inputs={'fields': [{'id': 'username', 'label': 'Username', 'type': 'string'}]}, + ) + url = reverse('api:credential_type_detail', kwargs={'pk': ct.pk}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + response = get(url, admin) + assert response.status_code == 200 + username = response.data['inputs']['fields'][0] + assert 'pattern' not in username + assert 'pattern_description' not in username + + +@pytest.mark.django_db +def test_credential_type_metadata_includes_patterns_when_toggle_on(get, admin, credentialtype_external, fake_tier2_pattern): + url = reverse('api:credential_type_detail', kwargs={'pk': credentialtype_external.pk}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + response = get(url, admin) + assert response.status_code == 200 + metadata = {f['id']: f for f in response.data['inputs']['metadata']} + assert metadata['key']['pattern'] == FAKE_TIER2_PATTERN + assert metadata['version']['pattern'] == FAKE_TIER2_PATTERN + fields = {f['id']: f for f in response.data['inputs']['fields']} + assert fields['url']['pattern'] == FAKE_TIER2_PATTERN + assert 'pattern' not in fields['token'] diff --git a/awx/main/tests/functional/api/test_notification_templates.py b/awx/main/tests/functional/api/test_notification_templates.py index f7b74a891b..0961c59742 100644 --- a/awx/main/tests/functional/api/test_notification_templates.py +++ b/awx/main/tests/functional/api/test_notification_templates.py @@ -1,5 +1,6 @@ # -*- coding: utf-8 -*- import pytest +from django.test import override_settings from awx.api.versioning import reverse from awx.main.models import NotificationTemplate, Organization @@ -90,3 +91,36 @@ def test_notification_template_names_unique_per_organization(post, admin_user): expect=400, ) assert 'Notification template with this Organization and Name already exists' in str(resp3.data) + + +FAKE_TIER2_PATTERN = r'^(?!.*<[a-zA-Z/!][^>]*>)[\s\S]*$' + + +@pytest.fixture +def fake_tier2_pattern(monkeypatch): + monkeypatch.setattr('awx.api.validation_patterns.build_tier2_frontend_pattern', lambda: FAKE_TIER2_PATTERN) + + +@pytest.mark.django_db +def test_notification_options_include_patterns_when_toggle_on(options, admin, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + response = options(reverse('api:notification_template_list'), admin) + assert response.status_code == 200 + email = response.data['actions']['POST']['notification_configuration']['email'] + assert email['host']['pattern'] == FAKE_TIER2_PATTERN + assert 'pattern_description' in email['host'] + assert 'pattern' in email['username'] + assert 'pattern' in email['sender'] + assert 'pattern' not in email['password'] + assert 'pattern' not in email['port'] + assert 'pattern' not in email['recipients'] + + +@pytest.mark.django_db +def test_notification_options_omit_patterns_when_toggle_off(options, admin, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + response = options(reverse('api:notification_template_list'), admin) + assert response.status_code == 200 + email = response.data['actions']['POST']['notification_configuration']['email'] + assert 'pattern' not in email['host'] + assert 'pattern_description' not in email['host'] diff --git a/awx/main/tests/functional/api/test_survey_spec.py b/awx/main/tests/functional/api/test_survey_spec.py index ec20806f6b..eaa803f33d 100644 --- a/awx/main/tests/functional/api/test_survey_spec.py +++ b/awx/main/tests/functional/api/test_survey_spec.py @@ -2,6 +2,8 @@ from unittest import mock import pytest import json +from django.test import override_settings + from ansible_base.lib.utils.models import get_type_for_model from awx.api.versioning import reverse @@ -279,3 +281,73 @@ def test_redact_survey_passwords_in_activity_stream(job_template_with_survey_pas changes_dict = json.loads(AS_record.changes) extra_vars = json.loads(changes_dict['extra_vars']) assert extra_vars['secret_key'] == '$encrypted$' + + +FAKE_SURVEY_TIER2_PATTERN = r'^(?!.*<[a-zA-Z/!][^>]*>)[\s\S]*$' + + +@pytest.fixture +def fake_survey_tier2_pattern(monkeypatch): + monkeypatch.setattr('awx.api.validation_patterns.build_tier2_frontend_pattern', lambda: FAKE_SURVEY_TIER2_PATTERN) + + +@pytest.mark.django_db +def test_survey_spec_options_include_patterns_when_toggle_on(options, admin, job_template, fake_survey_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + response = options(reverse('api:job_template_survey_spec', kwargs={'pk': job_template.id}), admin) + assert response.status_code == 200 + post = response.data['actions']['POST'] + assert post['name']['pattern'] == FAKE_SURVEY_TIER2_PATTERN + assert 'pattern_description' in post['name'] + assert post['description']['pattern'] == FAKE_SURVEY_TIER2_PATTERN + assert post['spec']['question_name']['pattern'] == FAKE_SURVEY_TIER2_PATTERN + assert post['spec']['variable']['pattern'] == FAKE_SURVEY_TIER2_PATTERN + assert post['spec']['question_description']['pattern'] == FAKE_SURVEY_TIER2_PATTERN + assert 'default' not in post['spec'] + assert 'choices' not in post['spec'] + + +@pytest.mark.django_db +def test_survey_spec_options_omit_patterns_when_toggle_off(options, admin, job_template, fake_survey_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + response = options(reverse('api:job_template_survey_spec', kwargs={'pk': job_template.id}), admin) + assert response.status_code == 200 + post = response.data['actions']['POST'] + assert 'pattern' not in post['name'] + assert 'pattern' not in post['spec']['question_name'] + assert post['spec']['variable']['type'] == 'string' + + +@pytest.mark.django_db +def test_workflow_survey_spec_options_include_patterns(options, admin, workflow_job_template, fake_survey_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + response = options(reverse('api:workflow_job_template_survey_spec', kwargs={'pk': workflow_job_template.id}), admin) + assert response.status_code == 200 + assert response.data['actions']['POST']['spec']['question_name']['pattern'] == FAKE_SURVEY_TIER2_PATTERN + + +@pytest.mark.django_db +def test_survey_spec_post_rejects_markup_in_question_name(post, admin_user, job_template, survey_spec_factory): + survey_input_data = survey_spec_factory('new_question') + survey_input_data['spec'][0]['question_name'] = '' + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + resp = post( + url=reverse('api:job_template_survey_spec', kwargs={'pk': job_template.id}), + data=survey_input_data, + user=admin_user, + expect=400, + ) + assert 'spec[0].question_name' in resp.data + + +@pytest.mark.django_db +def test_survey_spec_post_allows_markup_when_toggle_off(post, admin_user, job_template, survey_spec_factory): + survey_input_data = survey_spec_factory('new_question') + survey_input_data['spec'][0]['question_name'] = '' + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + post( + url=reverse('api:job_template_survey_spec', kwargs={'pk': job_template.id}), + data=survey_input_data, + user=admin_user, + expect=200, + ) diff --git a/awx/main/tests/unit/api/serializers/test_credential_type_serializer.py b/awx/main/tests/unit/api/serializers/test_credential_type_serializer.py new file mode 100644 index 0000000000..d812105698 --- /dev/null +++ b/awx/main/tests/unit/api/serializers/test_credential_type_serializer.py @@ -0,0 +1,92 @@ +# Copyright (c) 2026 Ansible, Inc. +# All Rights Reserved. + +from unittest.mock import patch, MagicMock + +import pytest +from django.test import override_settings + +from awx.api import validation_patterns + + +FAKE_TIER2_PATTERN = r'^(?!.*<[a-zA-Z/!][^>]*>)[\s\S]*$' + + +@pytest.fixture +def fake_tier2_pattern(monkeypatch): + monkeypatch.setattr(validation_patterns, 'build_tier2_frontend_pattern', lambda: FAKE_TIER2_PATTERN) + + +def _base_representation(fields=None, metadata=None, managed=False): + """Build a dict that mimics what BaseSerializer.to_representation returns.""" + inputs = {'fields': list(fields or [])} + if metadata is not None: + inputs['metadata'] = list(metadata) + return { + 'id': 1, + 'type': 'credential_type', + 'name': 'Test Type', + 'kind': 'cloud', + 'managed': managed, + 'namespace': None, + 'inputs': inputs, + 'injectors': {}, + } + + +class TestCredentialTypeSerializerInjectsPatterns: + @patch('awx.api.serializers.BaseSerializer.to_representation') + def test_inject_patterns_on_fields_and_metadata(self, mock_super, fake_tier2_pattern): + mock_super.return_value = _base_representation( + fields=[ + {'id': 'username', 'label': 'Username', 'type': 'string'}, + {'id': 'token', 'label': 'Token', 'type': 'string', 'secret': True}, + ], + metadata=[ + {'id': 'key', 'label': 'Key', 'type': 'string'}, + ], + ) + + from awx.api.serializers import CredentialTypeSerializer + + serializer = CredentialTypeSerializer() + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = serializer.to_representation(MagicMock()) + + fields = {f['id']: f for f in result['inputs']['fields']} + assert fields['username'].get('pattern') == FAKE_TIER2_PATTERN + assert 'pattern_description' in fields['username'] + assert 'pattern' not in fields['token'] + metadata = {f['id']: f for f in result['inputs']['metadata']} + assert metadata['key'].get('pattern') == FAKE_TIER2_PATTERN + + @patch('awx.api.serializers.BaseSerializer.to_representation') + def test_inject_patterns_skipped_when_toggle_off(self, mock_super, fake_tier2_pattern): + mock_super.return_value = _base_representation( + fields=[{'id': 'username', 'label': 'Username', 'type': 'string'}], + ) + + from awx.api.serializers import CredentialTypeSerializer + + serializer = CredentialTypeSerializer() + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + result = serializer.to_representation(MagicMock()) + + fields = {f['id']: f for f in result['inputs']['fields']} + assert 'pattern' not in fields['username'] + + @patch('awx.api.serializers.BaseSerializer.to_representation') + def test_inject_patterns_handles_none_metadata(self, mock_super, fake_tier2_pattern): + mock_super.return_value = _base_representation( + fields=[{'id': 'host', 'label': 'Host', 'type': 'string'}], + ) + + from awx.api.serializers import CredentialTypeSerializer + + serializer = CredentialTypeSerializer() + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = serializer.to_representation(MagicMock()) + + fields = {f['id']: f for f in result['inputs']['fields']} + assert fields['host'].get('pattern') == FAKE_TIER2_PATTERN + assert 'metadata' not in result['inputs'] diff --git a/awx/main/tests/unit/api/test_metadata.py b/awx/main/tests/unit/api/test_metadata.py new file mode 100644 index 0000000000..d9f17e8109 --- /dev/null +++ b/awx/main/tests/unit/api/test_metadata.py @@ -0,0 +1,125 @@ +# Copyright (c) 2026 Ansible, Inc. +# All Rights Reserved. + +from unittest.mock import MagicMock + +from rest_framework import serializers + +from awx.api import metadata as awx_metadata + + +FAKE_TIER2_PATTERN = r'^(?!.*<[a-zA-Z/!][^>]*>)[\s\S]*$' + + +class PlainSerializer(serializers.Serializer): + name = serializers.CharField() + + +class NotificationConfigSerializer(serializers.Serializer): + notification_configuration = serializers.JSONField() + + +class TestMetadataGetFieldInfoInjectsTopLevelPatterns: + def test_calls_inject_top_level_clean_text_patterns(self, monkeypatch): + calls = [] + + def fake_inject(field, field_info): + calls.append((field, field_info)) + field_info['pattern'] = 'FAKE' + return field_info + + monkeypatch.setattr(awx_metadata, 'inject_top_level_clean_text_patterns', fake_inject) + + serializer = PlainSerializer() + field = serializer.fields['name'] + + field_info = awx_metadata.Metadata().get_field_info(field) + + assert field_info['pattern'] == 'FAKE' + assert len(calls) == 1 + assert calls[0][0] is field + + def test_no_op_when_dab_helper_missing(self, monkeypatch): + # inject_top_level_clean_text_patterns itself no-ops when the DAB helper + # is unavailable; verify Metadata still returns clean field_info in that case. + monkeypatch.setattr(awx_metadata, 'inject_top_level_clean_text_patterns', lambda field, field_info: field_info) + + serializer = PlainSerializer() + field = serializer.fields['name'] + + field_info = awx_metadata.Metadata().get_field_info(field) + + assert 'pattern' not in field_info + + +class TestMetadataNotificationConfigurationInjection: + def test_calls_inject_patterns_into_init_parameters_for_each_type(self, monkeypatch): + calls = [] + + def tracking_inject(init_params): + calls.append(init_params) + return {'host': {'label': 'Host', 'type': 'string', 'pattern': 'FAKE'}} + + monkeypatch.setattr(awx_metadata, 'inject_patterns_into_init_parameters', tracking_inject) + monkeypatch.setattr(awx_metadata, 'inject_top_level_clean_text_patterns', lambda f, fi: fi) + + mock_backend = MagicMock() + mock_backend.init_parameters = {'host': {'label': 'Host', 'type': 'string'}} + + fake_types = [('email', 'Email', mock_backend)] + monkeypatch.setattr(awx_metadata, 'NotificationTemplate', MagicMock(NOTIFICATION_TYPES=fake_types)) + + serializer = NotificationConfigSerializer() + field = serializer.fields['notification_configuration'] + + field_info = awx_metadata.Metadata().get_field_info(field) + + assert len(calls) == 1 + assert calls[0] == mock_backend.init_parameters + assert field_info['email'] == {'host': {'label': 'Host', 'type': 'string', 'pattern': 'FAKE'}} + + def test_injects_patterns_for_multiple_notification_types(self, monkeypatch): + call_count = [] + + def counting_inject(init_params): + call_count.append(1) + return dict(init_params) + + monkeypatch.setattr(awx_metadata, 'inject_patterns_into_init_parameters', counting_inject) + monkeypatch.setattr(awx_metadata, 'inject_top_level_clean_text_patterns', lambda f, fi: fi) + + mock_email = MagicMock() + mock_email.init_parameters = {'host': {'label': 'Host', 'type': 'string'}} + mock_slack = MagicMock() + mock_slack.init_parameters = {'token': {'label': 'Token', 'type': 'password'}} + + fake_types = [('email', 'Email', mock_email), ('slack', 'Slack', mock_slack)] + monkeypatch.setattr(awx_metadata, 'NotificationTemplate', MagicMock(NOTIFICATION_TYPES=fake_types)) + + serializer = NotificationConfigSerializer() + field = serializer.fields['notification_configuration'] + + field_info = awx_metadata.Metadata().get_field_info(field) + + assert len(call_count) == 2 + assert 'email' in field_info + assert 'slack' in field_info + + +class TestMetadataSurveySpecOptions: + def test_post_schema_replaced_when_view_advertises_survey_spec(self, monkeypatch): + sentinel = {'name': {'type': 'string', 'pattern': 'FAKE'}} + monkeypatch.setattr(awx_metadata, 'build_survey_spec_options_schema', lambda: sentinel) + monkeypatch.setattr(awx_metadata, 'clone_request', lambda request, method: request) + + metadata = awx_metadata.Metadata() + view = MagicMock() + view.advertises_survey_spec_schema = True + view.allowed_methods = {'GET', 'POST'} + view.check_permissions.return_value = None + view.get_serializer.return_value = PlainSerializer() + + request = MagicMock() + actions = metadata.determine_actions(request, view) + + assert actions['POST'] is sentinel diff --git a/awx/main/tests/unit/api/test_validation_patterns.py b/awx/main/tests/unit/api/test_validation_patterns.py new file mode 100644 index 0000000000..895600fea9 --- /dev/null +++ b/awx/main/tests/unit/api/test_validation_patterns.py @@ -0,0 +1,334 @@ +# Copyright (c) 2026 Ansible, Inc. +# All Rights Reserved. + +import pytest +from django.test import override_settings + +from awx.api import validation_patterns +from awx.main.notifications.email_backend import CustomEmailBackend + + +FAKE_PATTERN = r'^(?!.*<[a-zA-Z/!][^>]*>)[\s\S]*$' + + +@pytest.fixture +def fake_tier2_pattern(monkeypatch): + monkeypatch.setattr(validation_patterns, 'build_tier2_frontend_pattern', lambda: FAKE_PATTERN) + + +class TestEnhancedInputValidationEnabled: + def test_off_by_default(self): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + assert validation_patterns.enhanced_input_validation_enabled() is False + + def test_on_when_setting_true(self): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.enhanced_input_validation_enabled() is True + + +class TestInjectFreeTextPattern: + def test_no_op_when_toggle_off(self, fake_tier2_pattern): + schema = {'id': 'username', 'label': 'Username', 'type': 'string'} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + validation_patterns.inject_free_text_pattern(schema) + assert 'pattern' not in schema + assert 'pattern_description' not in schema + + def test_injects_tier2_when_toggle_on(self, fake_tier2_pattern): + schema = {'id': 'username', 'label': 'Username', 'type': 'string'} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_free_text_pattern(schema) + assert schema['pattern'] == FAKE_PATTERN + assert schema['pattern_description'] == validation_patterns.TIER2_PATTERN_DESCRIPTION + assert schema['flags'] == 'i' + + def test_skips_secret_fields(self, fake_tier2_pattern): + schema = {'id': 'password', 'label': 'Password', 'type': 'string', 'secret': True} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_free_text_pattern(schema) + assert 'pattern' not in schema + + def test_skips_non_string_fields(self, fake_tier2_pattern): + schema = {'id': 'port', 'label': 'Port', 'type': 'int'} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_free_text_pattern(schema) + assert 'pattern' not in schema + + def test_treats_missing_type_as_string(self, fake_tier2_pattern): + schema = {'id': 'host', 'label': 'Host'} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_free_text_pattern(schema) + assert schema['pattern'] == FAKE_PATTERN + + def test_no_op_when_dab_helper_missing(self, monkeypatch): + monkeypatch.setattr(validation_patterns, 'build_tier2_frontend_pattern', None) + schema = {'id': 'username', 'type': 'string'} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_free_text_pattern(schema) + assert 'pattern' not in schema + + def test_ignores_non_dict(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.inject_free_text_pattern('not-a-dict') == 'not-a-dict' + + +class TestFreeTextPatternMetadata: + def test_returns_pattern_dict_when_helper_available(self, fake_tier2_pattern): + result = validation_patterns.free_text_pattern_metadata() + assert result is not None + assert result['pattern'] == FAKE_PATTERN + assert result['pattern_description'] == validation_patterns.TIER2_PATTERN_DESCRIPTION + assert result['flags'] == 'i' + + def test_returns_none_when_helper_missing(self, monkeypatch): + monkeypatch.setattr(validation_patterns, 'build_tier2_frontend_pattern', None) + assert validation_patterns.free_text_pattern_metadata() is None + + +class TestIsStringSchema: + def test_string_type_in_schema(self): + assert validation_patterns._is_string_schema({'type': 'string'}) is True + + def test_str_type_in_schema(self): + assert validation_patterns._is_string_schema({'type': 'str'}) is True + + def test_non_string_type_in_schema(self): + assert validation_patterns._is_string_schema({'type': 'boolean'}) is False + + def test_missing_type_defaults_to_string(self): + assert validation_patterns._is_string_schema({}) is True + + def test_explicit_field_type_overrides_schema(self): + assert validation_patterns._is_string_schema({'type': 'boolean'}, field_type='string') is True + assert validation_patterns._is_string_schema({'type': 'string'}, field_type='int') is False + + +class TestInjectFreeTextPatternSecretKwarg: + def test_skips_when_secret_kwarg_true(self, fake_tier2_pattern): + schema = {'id': 'token', 'type': 'string'} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_free_text_pattern(schema, secret=True) + assert 'pattern' not in schema + + def test_injects_when_secret_kwarg_false(self, fake_tier2_pattern): + schema = {'id': 'host', 'type': 'string'} + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_free_text_pattern(schema, secret=False) + assert schema['pattern'] == FAKE_PATTERN + + +class TestInjectPatternsIntoFieldList: + def test_injects_only_non_secret_strings(self, fake_tier2_pattern): + fields = [ + {'id': 'username', 'type': 'string'}, + {'id': 'password', 'type': 'string', 'secret': True}, + {'id': 'verify_ssl', 'type': 'boolean'}, + ] + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_patterns_into_field_list(fields) + assert 'pattern' in fields[0] + assert 'pattern' not in fields[1] + assert 'pattern' not in fields[2] + + def test_ignores_non_list(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.inject_patterns_into_field_list('not-a-list') is None + + def test_handles_none_input(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.inject_patterns_into_field_list(None) is None + + def test_handles_empty_list(self, fake_tier2_pattern): + fields = [] + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + validation_patterns.inject_patterns_into_field_list(fields) + assert fields == [] + + +class TestInjectPatternsIntoInitParameters: + def test_does_not_mutate_class_level_dict(self, fake_tier2_pattern): + original = CustomEmailBackend.init_parameters + host_before = dict(original['host']) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + injected = validation_patterns.inject_patterns_into_init_parameters(original) + assert 'pattern' in injected['host'] + assert 'pattern' not in original['host'] + assert original['host'] == host_before + + def test_skips_password_and_non_string_types(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + injected = validation_patterns.inject_patterns_into_init_parameters(CustomEmailBackend.init_parameters) + assert 'pattern' in injected['host'] + assert 'pattern' in injected['username'] + assert 'pattern' in injected['sender'] + assert 'pattern' not in injected['password'] + assert 'pattern' not in injected['port'] + assert 'pattern' not in injected['use_tls'] + assert 'pattern' not in injected['recipients'] + + def test_toggle_off_returns_copy_without_patterns(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + injected = validation_patterns.inject_patterns_into_init_parameters(CustomEmailBackend.init_parameters) + assert 'pattern' not in injected['host'] + assert injected['host'] == CustomEmailBackend.init_parameters['host'] + assert injected is not CustomEmailBackend.init_parameters + + def test_non_dict_input_returns_empty_dict(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_patterns_into_init_parameters('not-a-dict') + assert result == {} + + def test_none_input_returns_empty_dict(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_patterns_into_init_parameters(None) + assert result == {} + + def test_list_input_returns_empty_dict(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_patterns_into_init_parameters([1, 2, 3]) + assert result == {} + + def test_skips_non_dict_values(self, fake_tier2_pattern): + params = { + 'host': {'label': 'Host', 'type': 'string'}, + 'extra_info': 'just-a-string', + 'count': 42, + } + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_patterns_into_init_parameters(params) + assert 'pattern' in result['host'] + assert result['extra_info'] == 'just-a-string' + assert result['count'] == 42 + + def test_non_dict_input_toggle_off_returns_empty_dict(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + result = validation_patterns.inject_patterns_into_init_parameters(None) + assert result == {} + + def test_empty_dict_returns_empty_dict(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_patterns_into_init_parameters({}) + assert result == {} + + +class TestInjectTopLevelCleanTextPatterns: + def test_no_op_when_dab_helper_missing(self, monkeypatch): + monkeypatch.setattr(validation_patterns, '_dab_inject_clean_text_patterns', None) + field_info = {'type': 'string'} + result = validation_patterns.inject_top_level_clean_text_patterns(object(), field_info) + assert result == {'type': 'string'} + + def test_delegates_to_dab_helper(self, monkeypatch): + calls = [] + + def fake_dab_inject(field, field_info): + calls.append((field, field_info)) + field_info['pattern'] = 'FAKE' + return field_info + + monkeypatch.setattr(validation_patterns, '_dab_inject_clean_text_patterns', fake_dab_inject) + sentinel_field = object() + field_info = {'type': 'string'} + + result = validation_patterns.inject_top_level_clean_text_patterns(sentinel_field, field_info) + + assert result['pattern'] == 'FAKE' + assert calls == [(sentinel_field, field_info)] + + +class TestBuildSurveySpecOptionsSchema: + def test_injects_patterns_when_toggle_on(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + schema = validation_patterns.build_survey_spec_options_schema() + assert schema['name']['pattern'] == FAKE_PATTERN + assert schema['name']['pattern_description'] == validation_patterns.TIER2_PATTERN_DESCRIPTION + assert schema['description']['pattern'] == FAKE_PATTERN + assert schema['spec']['type'] == 'json' + assert schema['spec']['question_name']['pattern'] == FAKE_PATTERN + assert schema['spec']['question_description']['pattern'] == FAKE_PATTERN + assert schema['spec']['variable']['pattern'] == FAKE_PATTERN + assert 'default' not in schema['spec'] + assert 'choices' not in schema['spec'] + + def test_omits_patterns_when_toggle_off(self, fake_tier2_pattern): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + schema = validation_patterns.build_survey_spec_options_schema() + assert 'pattern' not in schema['name'] + assert 'pattern' not in schema['description'] + assert 'pattern' not in schema['spec']['question_name'] + assert schema['spec']['variable']['type'] == 'string' + + +def _survey_payload(**overrides): + question = { + 'question_name': 'Question', + 'question_description': 'Help', + 'variable': 'my_var', + 'type': 'text', + 'required': True, + } + question.update(overrides.pop('question', {})) + payload = {'name': 'Survey', 'description': 'Desc', 'spec': [question]} + payload.update(overrides) + return payload + + +class TestCollectSurveySpecTextErrors: + def test_no_op_when_toggle_off(self): + payload = _survey_payload(question={'question_name': ''}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + assert validation_patterns.collect_survey_spec_text_errors(payload) is None + + def test_rejects_markup_in_editor_strings(self): + if validation_patterns._validate_free_text is None: + pytest.skip('DAB validate_free_text is unavailable') + payload = _survey_payload(question={'question_name': ''}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + errors = validation_patterns.collect_survey_spec_text_errors(payload) + assert errors is not None + assert 'spec[0].question_name' in errors + + def test_rejects_top_level_name(self): + if validation_patterns._validate_free_text is None: + pytest.skip('DAB validate_free_text is unavailable') + payload = _survey_payload() + payload['name'] = 'bad' + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + errors = validation_patterns.collect_survey_spec_text_errors(payload) + assert errors is not None + assert 'name' in errors + + def test_skips_default_and_choices(self): + if validation_patterns._validate_free_text is None: + pytest.skip('DAB validate_free_text is unavailable') + payload = _survey_payload(question={'default': '', 'choices': '{{ foo }}'}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.collect_survey_spec_text_errors(payload) is None + + def test_grandfathers_unchanged_invalid_values(self): + if validation_patterns._validate_free_text is None: + pytest.skip('DAB validate_free_text is unavailable') + old = _survey_payload(question={'question_name': ''}) + new = _survey_payload(question={'question_name': '', 'question_description': 'updated'}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.collect_survey_spec_text_errors(new, old) is None + + def test_validates_changed_question_name_against_legacy_sibling(self): + if validation_patterns._validate_free_text is None: + pytest.skip('DAB validate_free_text is unavailable') + old = _survey_payload(question={'question_name': ''}) + new = _survey_payload(question={'question_name': ''}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + errors = validation_patterns.collect_survey_spec_text_errors(new, old) + assert errors is not None + assert 'spec[0].question_name' in errors + + def test_ignores_non_dict_payload(self): + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.collect_survey_spec_text_errors('not-a-dict') is None + + def test_no_op_when_validator_missing(self, monkeypatch): + monkeypatch.setattr(validation_patterns, '_validate_free_text', None) + payload = _survey_payload(question={'question_name': ''}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + assert validation_patterns.collect_survey_spec_text_errors(payload) is None diff --git a/awx/main/tests/unit/api/test_views.py b/awx/main/tests/unit/api/test_views.py index f9dcea2aed..2de6919376 100644 --- a/awx/main/tests/unit/api/test_views.py +++ b/awx/main/tests/unit/api/test_views.py @@ -5,6 +5,8 @@ from unittest import mock from collections import namedtuple +from django.test import override_settings + from awx.api.views.root import ApiVersionRootView from awx.api.views import JobTemplateLabelList, InventoryInventorySourcesUpdate, JobTemplateSurveySpec @@ -317,3 +319,17 @@ class TestSurveySpecValidation: spec = self.spec_from_element({'type': _type, 'default': ''}) r = JobTemplateSurveySpec._validate_spec_data(spec, {}) assert r is None + + def test_rejects_markup_in_question_name_when_toggle_on(self): + spec = self.spec_from_element({'question_name': ''}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + r = JobTemplateSurveySpec._validate_spec_data(spec, {}) + assert r is not None + assert r.status_code == 400 + assert 'spec[0].question_name' in r.data + + def test_allows_markup_in_question_name_when_toggle_off(self): + spec = self.spec_from_element({'question_name': ''}) + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + r = JobTemplateSurveySpec._validate_spec_data(spec, {}) + assert r is None