mirror of
https://github.com/ansible/awx.git
synced 2026-10-29 22:03:20 -02:30
[AAP-87586] advertise validation patterns on top level and JSON sub-key fields (#16638)
* feat(api): advertise validation patterns on JSON sub-key fields Credential type inputs.fields/metadata and notification OPTIONS now include pattern and pattern_description for non-secret string sub-keys when ENHANCED_INPUT_VALIDATION_ENABLED is on. * Fix lint issue * Fix sonarqube issue * feat(api): advertise validation patterns on top-level CharFields Wires DAB's inject_clean_text_patterns into awx.api.metadata.Metadata so top-level CharField OPTIONS metadata gets pattern/patternDescription/flags, matching the JSON sub-key pattern exposure already in place. No-op until ENHANCED_INPUT_VALIDATION_ENABLED is on and a serializer mixes in CleanTextMixin. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * Increase test coverage * Fix lint issues * Address review feedback on validation pattern injection Import TIER2_PATTERN_DESCRIPTION from DAB instead of keeping a manually synced duplicate, and short-circuit inject_patterns_into_field_list before iterating fields when enhanced input validation is disabled. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * Avoid mutating shared field dicts in inject_patterns_into_field_list Shallow-copy each field before injecting pattern metadata so callers that pass module-level schema dicts (e.g. ManagedCredentialType.registry entries) don't get them mutated in place, matching the existing copy-before-mutate approach in inject_patterns_into_init_parameters. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * Add validation pattern for survey fields --------- Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> Co-authored-by: Peter Braun <pbraun@redhat.com>
This commit is contained in:
Pratyush Bhandari
co-authored by
Peter Braun
Peter Braun
parent
bbda905e00
commit
32f588e15b
+13
-1
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
|
||||
@@ -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']
|
||||
|
||||
@@ -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']
|
||||
|
||||
@@ -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'] = '<script>alert(1)</script>'
|
||||
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'] = '<script>alert(1)</script>'
|
||||
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,
|
||||
)
|
||||
|
||||
@@ -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']
|
||||
@@ -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
|
||||
@@ -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': '<script>x</script>'})
|
||||
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': '<script>x</script>'})
|
||||
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'] = '<b>bad</b>'
|
||||
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': '<script>x</script>', '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': '<script>old</script>'})
|
||||
new = _survey_payload(question={'question_name': '<script>old</script>', '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': '<script>old</script>'})
|
||||
new = _survey_payload(question={'question_name': '<script>new</script>'})
|
||||
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': '<script>x</script>'})
|
||||
with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True):
|
||||
assert validation_patterns.collect_survey_spec_text_errors(payload) is None
|
||||
@@ -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': '<script>x</script>'})
|
||||
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': '<script>x</script>'})
|
||||
with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False):
|
||||
r = JobTemplateSurveySpec._validate_spec_data(spec, {})
|
||||
assert r is None
|
||||
|
||||
Reference in New Issue
Block a user