From 4c290176efb50285c8334cefd6a3c5352395c4ab Mon Sep 17 00:00:00 2001 From: Pratyush Bhandari <112966709+prat98@users.noreply.github.com> Date: Tue, 22 Sep 2026 11:41:47 -0400 Subject: [PATCH] [AAP-78694] Add CleanTextMixin to settings serializer (#16661) * Add cleantextmixin to settings serializer * Make excluded_fields a cached property * Add test coverage for category_slug all * Fix OPTIONS 500 for settings CleanTextMixin pattern injection * Apply validation for json and list fields --------- Co-authored-by: Peter Braun --- awx/api/serializers.py | 30 +++- awx/api/validation_patterns.py | 65 +++++++- awx/conf/serializers.py | 68 +++++++- awx/conf/tests/functional/test_api.py | 145 ++++++++++++++++++ .../unit/api/test_validation_patterns.py | 135 ++++++++++++++++ 5 files changed, 429 insertions(+), 14 deletions(-) diff --git a/awx/api/serializers.py b/awx/api/serializers.py index 8b9ff33325..8474021495 100644 --- a/awx/api/serializers.py +++ b/awx/api/serializers.py @@ -218,9 +218,14 @@ def reverse_gfk(content_object, request): class PlainSerializerCleanTextMixin(CleanTextMixin): """CleanTextMixin for plain `serializers.Serializer` subclasses with no backing Django model (AAP-78694). Overrides ONLY field discovery -- treats - the serializer's own declared CharFields as the text fields to validate. - Everything else (Tier 1/2 dispatch, exclusions, audit logging, the - enforcement gate) is reused unchanged from CleanTextMixin. + the serializer's own declared fields as the text/JSON containers to validate. + Everything else (Tier 1/2 dispatch, nested JSON walk, exclusions, audit + logging, the enforcement gate) is reused unchanged from CleanTextMixin. + + CharFields (including URLField) are Tier 1/2 top-level text. ListField, + DictField, and JSONField values are walked with CleanTextMixin's nested + string validation (same path as model JSONFields) so settings like + StringListField / KeyValueField are covered. Known limitation: unlike the model-field path, this can't replicate the get_internal_type() exclusion of SlugField/IPAddressField-style fields @@ -228,9 +233,24 @@ class PlainSerializerCleanTextMixin(CleanTextMixin): using this mixin today, since none declare such fields. """ + # Structured containers whose string leaves should get validate_free_text. + # Matches model CleanTextMixin's JSONField walk for settings registry + # ListField/DictField/KeyValueField (and DRF JSONField if used). + _NESTED_TEXT_CONTAINER_FIELDS = ( + serializers.ListField, + serializers.DictField, + serializers.JSONField, + ) + def _classify_fields(self, model): - text_fields = [name for name, field in self.fields.items() if isinstance(field, serializers.CharField)] - return text_fields, [] + text_fields = [] + json_fields = [] + for name, field in self.fields.items(): + if isinstance(field, serializers.CharField): + text_fields.append(name) + elif isinstance(field, self._NESTED_TEXT_CONTAINER_FIELDS): + json_fields.append(name) + return text_fields, json_fields class _CopySerializerFakeModel: diff --git a/awx/api/validation_patterns.py b/awx/api/validation_patterns.py index dfa1e4869a..7d9e724c6a 100644 --- a/awx/api/validation_patterns.py +++ b/awx/api/validation_patterns.py @@ -18,6 +18,9 @@ lands on DAB devel, injection is a no-op so Controller can still import and run. import copy +from rest_framework import serializers + +from ansible_base.lib.serializers.mixins import CleanTextMixin from ansible_base.lib.utils.settings import get_setting try: @@ -30,6 +33,13 @@ try: except ImportError: # pragma: no cover - DAB without AAP-85987 _dab_inject_clean_text_patterns = None +try: + from ansible_base.lib.metadata import get_tier1_pattern as _get_tier1_pattern + from ansible_base.lib.metadata import get_tier2_pattern as _get_tier2_pattern +except ImportError: # pragma: no cover - DAB without AAP-85987 + _get_tier1_pattern = None + _get_tier2_pattern = None + try: from ansible_base.lib.metadata import TIER2_PATTERN_DESCRIPTION except ImportError: # pragma: no cover - DAB without AAP-85987 @@ -209,13 +219,62 @@ def collect_survey_spec_text_errors(new_spec, old_spec=None): return errors or None +def _uses_fake_model_meta(serializer): + """True when Meta.model is a stand-in without Django's Options.get_field. + + Plain serializers (SettingSingletonSerializer, CopySerializer) use a fake + Meta.model so CleanTextMixin audit logs have app_label/object_name. DAB's + inject_clean_text_patterns calls model._meta.get_field() and would 500 on + OPTIONS for those endpoints -- handle them here instead. + """ + model = getattr(getattr(serializer, 'Meta', None), 'model', None) + if model is None: + return False + meta = getattr(model, '_meta', None) + return meta is not None and not hasattr(meta, 'get_field') + + +def _inject_declared_char_field_patterns(field, field_info, serializer): + """Tier 1/2 OPTIONS hints for declared CharFields on plain CleanText serializers.""" + if _get_tier1_pattern is None or _get_tier2_pattern is None: + return field_info + if not get_setting('ENHANCED_INPUT_VALIDATION_ENABLED', False): + return field_info + if not isinstance(serializer, CleanTextMixin): + return field_info + if not isinstance(field, serializers.CharField): + return field_info + + field_name = field.field_name + if field_name in getattr(serializer, 'excluded_fields', frozenset()): + return field_info + + if field_name in getattr(serializer, 'name_fields', frozenset()): + tier1 = _get_tier1_pattern() + field_info['pattern'] = tier1['pattern'] + field_info['patternDescription'] = tier1['description'] + field_info['flags'] = tier1['flags'] + field_info['normalize'] = tier1['normalize'] + else: + tier2 = _get_tier2_pattern() + field_info['pattern'] = tier2['pattern'] + field_info['patternDescription'] = tier2['description'] + field_info['flags'] = tier2['flags'] + + return field_info + + 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. + For model-backed CleanTextMixin serializers, delegates to DAB's + ``inject_clean_text_patterns``. For plain serializers with a fake Meta.model + (no ``_meta.get_field``), injects the same Tier 1/2 hints via DAB's public + ``get_tier1_pattern`` / ``get_tier2_pattern`` helpers so OPTIONS still works. """ + serializer = getattr(field, 'parent', None) + if serializer is not None and _uses_fake_model_meta(serializer): + return _inject_declared_char_field_patterns(field, field_info, serializer) if _dab_inject_clean_text_patterns is None: return field_info return _dab_inject_clean_text_patterns(field, field_info) diff --git a/awx/conf/serializers.py b/awx/conf/serializers.py index ea4820aa5b..26685467b4 100644 --- a/awx/conf/serializers.py +++ b/awx/conf/serializers.py @@ -1,9 +1,12 @@ +# Django +from django.utils.functional import cached_property + # Django REST Framework from rest_framework import serializers # AWX from awx.api.fields import VerbatimField -from awx.api.serializers import BaseSerializer +from awx.api.serializers import BaseSerializer, PlainSerializerCleanTextMixin from awx.conf.models import Setting from awx.conf import settings_registry @@ -50,8 +53,58 @@ class SettingFieldMixin(object): return super(SettingFieldMixin, self).to_representation(obj) -class SettingSingletonSerializer(serializers.Serializer): - """Present a group of settings (by category) as a single object.""" +class _SettingSingletonFakeOpts: + """Minimal stand-in for Django model._meta. + + CleanTextMixin audit logs need app_label/object_name. OPTIONS metadata + (awx.api.metadata.get_field_info) also walks Meta.model._meta.fields when + Meta.model is present -- an empty tuple keeps that path from crashing. + """ + + app_label = 'conf' + object_name = 'SettingSingleton' + verbose_name = 'setting singleton' + fields = () + + @property + def concrete_model(self): + return _SettingSingletonFakeModel + + +class _SettingSingletonFakeModel: + """Stand-in for Meta.model on SettingSingletonSerializer (plain Serializer). + + Never introspected for real fields -- PlainSerializerCleanTextMixin discovers + CharField / ListField / DictField / JSONField entries from self.fields instead. + """ + + _meta = _SettingSingletonFakeOpts() + + +class SettingSingletonSerializer(PlainSerializerCleanTextMixin, serializers.Serializer): + """Present a group of settings (by category) as a single object. + + Uses PlainSerializerCleanTextMixin because there is no Django model whose + fields match the dynamic settings registry. Top-level CharFields get Tier + 1/2 text validation; ListField/DictField/JSONField string children are + walked via CleanTextMixin's nested JSON validation path. + """ + + # CUSTOM_LOGIN_INFO deliberately allows HTML fragments (ui/conf.py). + # CUSTOM_LOGO is a data:image/...;base64,... URI; Tier 2 blocks data: schemes. + # AWX_TASK_ENV / GALAXY_TASK_ENV are key/value maps whose values often use + # shell/env expansion (${VAR}, $(...)) or similar — Tier 2's injection + # blocklist would reject legitimate automation env config. + # Encrypted settings (passwords/tokens/PEMs) are excluded dynamically below. + _ALWAYS_EXCLUDED = frozenset({'CUSTOM_LOGIN_INFO', 'CUSTOM_LOGO', 'AWX_TASK_ENV', 'GALAXY_TASK_ENV'}) + + class Meta: + model = _SettingSingletonFakeModel + + @cached_property + def excluded_fields(self): + encrypted = frozenset(s for s in settings_registry.get_registered_settings() if settings_registry.is_setting_encrypted(s)) + return self._ALWAYS_EXCLUDED | encrypted def __init__(self, instance=None, data=serializers.empty, **kwargs): # Instance (if given) should be an object with attributes for all of the @@ -67,9 +120,12 @@ class SettingSingletonSerializer(serializers.Serializer): if self.context['view'].kwargs.get('category_slug', '') == 'all': for validate_func in settings_registry._validate_registry.values(): attrs = validate_func(self, attrs) - return attrs - custom_validate = settings_registry.get_registered_validate_func(category_slug) - return custom_validate(self, attrs) if custom_validate else attrs + else: + custom_validate = settings_registry.get_registered_validate_func(category_slug) + if custom_validate: + attrs = custom_validate(self, attrs) + # Chain into CleanTextMixin (via PlainSerializerCleanTextMixin). + return super().validate(attrs) def get_fields(self): fields = super(SettingSingletonSerializer, self).get_fields() diff --git a/awx/conf/tests/functional/test_api.py b/awx/conf/tests/functional/test_api.py index 9a398d5742..f7bef36767 100644 --- a/awx/conf/tests/functional/test_api.py +++ b/awx/conf/tests/functional/test_api.py @@ -162,3 +162,148 @@ def test_setting_singleton_delete_no_read_only_fields(api_request, dummy_setting api_request('delete', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'})) response = api_request('get', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'})) assert response.data['FOO_BAR'] == 23 + + +# +# CleanTextMixin on SettingSingletonSerializer (via PlainSerializerCleanTextMixin) +# + +UNSAFE_SETTING_INPUT = '' + + +@pytest.fixture +def enforce_clean_text(): + with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + yield + + +@pytest.mark.django_db +def test_setting_singleton_rejects_unsafe_char_field(api_request, dummy_setting, enforce_clean_text): + with ( + dummy_setting('FOO_BAR', field_class=fields.CharField, allow_blank=True, default='', category='FooBar', category_slug='foobar'), + mock.patch('awx.conf.views.clear_setting_cache'), + ): + response = api_request('patch', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), data={'FOO_BAR': UNSAFE_SETTING_INPUT}) + assert response.status_code == 400 + assert 'FOO_BAR' in response.data + + +@pytest.mark.django_db +def test_setting_singleton_rejects_unsafe_char_field_all_category(api_request, dummy_setting, dummy_validate, enforce_clean_text): + # category_slug='all' runs every registered validate_func, then CleanTextMixin. + # A validator on a different category only participates via that loop. + validate_calls = [] + + def record_validate(serializer, attrs): + validate_calls.append(True) + return attrs + + with ( + dummy_setting('FOO_BAR', field_class=fields.CharField, allow_blank=True, default='', category='FooBar', category_slug='foobar'), + dummy_validate('other', record_validate), + mock.patch('awx.conf.views.clear_setting_cache'), + ): + response = api_request('patch', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'all'}), data={'FOO_BAR': UNSAFE_SETTING_INPUT}) + assert validate_calls + assert response.status_code == 400 + assert 'FOO_BAR' in response.data + + +@pytest.mark.django_db +def test_setting_singleton_skips_encrypted_fields(api_request, dummy_setting, enforce_clean_text): + with ( + dummy_setting('FOO_SECRET', field_class=fields.CharField, encrypted=True, allow_blank=True, default='', category='FooBar', category_slug='foobar'), + mock.patch('awx.conf.views.clear_setting_cache'), + ): + response = api_request('patch', reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), data={'FOO_SECRET': UNSAFE_SETTING_INPUT}) + assert response.status_code == 200 + assert decrypt_field(Setting.objects.get(key='FOO_SECRET'), 'value') == UNSAFE_SETTING_INPUT + + +@pytest.mark.django_db +def test_setting_singleton_skips_custom_login_info_html(api_request, enforce_clean_text): + with mock.patch('awx.conf.views.clear_setting_cache'): + response = api_request( + 'patch', + reverse('api:setting_singleton_detail', kwargs={'category_slug': 'ui'}), + data={'CUSTOM_LOGIN_INFO': '

Legal notice

'}, + ) + assert response.status_code == 200 + assert response.data['CUSTOM_LOGIN_INFO'] == '

Legal notice

' + + +@pytest.mark.django_db +def test_setting_singleton_rejects_unsafe_string_list_item(api_request, dummy_setting, enforce_clean_text): + with ( + dummy_setting( + 'FOO_LIST', + field_class=fields.StringListField, + default=[], + category='FooBar', + category_slug='foobar', + ), + mock.patch('awx.conf.views.clear_setting_cache'), + ): + response = api_request( + 'patch', + reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), + data={'FOO_LIST': ['ok-value', UNSAFE_SETTING_INPUT]}, + ) + assert response.status_code == 400 + assert 'FOO_LIST' in response.data + + +@pytest.mark.django_db +def test_setting_singleton_rejects_unsafe_dict_value(api_request, dummy_setting, enforce_clean_text): + with ( + dummy_setting( + 'FOO_MAP', + field_class=fields.KeyValueField, + default={}, + category='FooBar', + category_slug='foobar', + ), + mock.patch('awx.conf.views.clear_setting_cache'), + ): + response = api_request( + 'patch', + reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), + data={'FOO_MAP': {'HTTP_PROXY': UNSAFE_SETTING_INPUT}}, + ) + assert response.status_code == 400 + assert 'FOO_MAP' in response.data + + +@pytest.mark.django_db +def test_setting_singleton_allows_safe_string_list(api_request, dummy_setting, enforce_clean_text): + with ( + dummy_setting( + 'FOO_LIST', + field_class=fields.StringListField, + default=[], + category='FooBar', + category_slug='foobar', + ), + mock.patch('awx.conf.views.clear_setting_cache'), + ): + response = api_request( + 'patch', + reverse('api:setting_singleton_detail', kwargs={'category_slug': 'foobar'}), + data={'FOO_LIST': ['HTTP_X_FORWARDED_FOR', 'https://example.com']}, + ) + assert response.status_code == 200 + assert response.data['FOO_LIST'] == ['HTTP_X_FORWARDED_FOR', 'https://example.com'] + + +@pytest.mark.django_db +def test_setting_singleton_skips_task_env_shell_expansion(api_request, enforce_clean_text): + """AWX_TASK_ENV values may use ${VAR} / $(...) — excluded from CleanText.""" + with mock.patch('awx.conf.views.clear_setting_cache'): + response = api_request( + 'patch', + reverse('api:setting_singleton_detail', kwargs={'category_slug': 'jobs'}), + data={'AWX_TASK_ENV': {'PATH': '${PATH}:/opt/bin', 'CUSTOM': '$(echo hi)'}}, + ) + assert response.status_code == 200 + assert response.data['AWX_TASK_ENV']['PATH'] == '${PATH}:/opt/bin' + assert response.data['AWX_TASK_ENV']['CUSTOM'] == '$(echo hi)' diff --git a/awx/main/tests/unit/api/test_validation_patterns.py b/awx/main/tests/unit/api/test_validation_patterns.py index 895600fea9..cf790e2dfb 100644 --- a/awx/main/tests/unit/api/test_validation_patterns.py +++ b/awx/main/tests/unit/api/test_validation_patterns.py @@ -235,6 +235,141 @@ class TestInjectTopLevelCleanTextPatterns: assert result['pattern'] == 'FAKE' assert calls == [(sentinel_field, field_info)] + def test_plain_serializer_fake_meta_gets_tier2(self, monkeypatch): + from ansible_base.lib.serializers.mixins import CleanTextMixin + from rest_framework import serializers as drf_serializers + + monkeypatch.setattr(validation_patterns, '_get_tier1_pattern', lambda: {'pattern': 'T1', 'description': 'd1', 'flags': 'u', 'normalize': 'NFC'}) + monkeypatch.setattr(validation_patterns, '_get_tier2_pattern', lambda: {'pattern': 'T2', 'description': 'd2', 'flags': 'i'}) + + class FakeOpts: + app_label = 'conf' + object_name = 'SettingSingleton' + + class FakeModel: + _meta = FakeOpts() + + class PlainCleanText(CleanTextMixin, drf_serializers.Serializer): + class Meta: + model = FakeModel + + excluded_fields = frozenset() + name_fields = frozenset({'name'}) + + serializer = PlainCleanText() + field = drf_serializers.CharField() + field.bind('FOO_BAR', serializer) + field_info = {'type': 'string'} + + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_top_level_clean_text_patterns(field, field_info) + + assert result['pattern'] == 'T2' + assert result['patternDescription'] == 'd2' + assert result['flags'] == 'i' + + def test_plain_serializer_name_field_gets_tier1(self, monkeypatch): + from ansible_base.lib.serializers.mixins import CleanTextMixin + from rest_framework import serializers as drf_serializers + + monkeypatch.setattr(validation_patterns, '_get_tier1_pattern', lambda: {'pattern': 'T1', 'description': 'd1', 'flags': 'u', 'normalize': 'NFC'}) + monkeypatch.setattr(validation_patterns, '_get_tier2_pattern', lambda: {'pattern': 'T2', 'description': 'd2', 'flags': 'i'}) + + class FakeOpts: + app_label = 'main' + object_name = 'CopySerializer' + + class FakeModel: + _meta = FakeOpts() + + class PlainCleanText(CleanTextMixin, drf_serializers.Serializer): + class Meta: + model = FakeModel + + excluded_fields = frozenset() + name_fields = frozenset({'name'}) + + serializer = PlainCleanText() + field = drf_serializers.CharField() + field.bind('name', serializer) + field_info = {'type': 'string'} + + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_top_level_clean_text_patterns(field, field_info) + + assert result['pattern'] == 'T1' + assert result['patternDescription'] == 'd1' + assert result['flags'] == 'u' + assert result['normalize'] == 'NFC' + + def test_plain_serializer_skips_excluded_fields(self, monkeypatch): + from ansible_base.lib.serializers.mixins import CleanTextMixin + from rest_framework import serializers as drf_serializers + + monkeypatch.setattr(validation_patterns, '_get_tier2_pattern', lambda: {'pattern': 'T2', 'description': 'd2', 'flags': 'i'}) + + class FakeOpts: + app_label = 'conf' + object_name = 'SettingSingleton' + + class FakeModel: + _meta = FakeOpts() + + class PlainCleanText(CleanTextMixin, drf_serializers.Serializer): + class Meta: + model = FakeModel + + excluded_fields = frozenset({'CUSTOM_LOGIN_INFO'}) + name_fields = frozenset() + + serializer = PlainCleanText() + field = drf_serializers.CharField() + field.bind('CUSTOM_LOGIN_INFO', serializer) + field_info = {'type': 'string'} + + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_top_level_clean_text_patterns(field, field_info) + + assert 'pattern' not in result + + def test_plain_serializer_does_not_call_dab_inject(self, monkeypatch): + from ansible_base.lib.serializers.mixins import CleanTextMixin + from rest_framework import serializers as drf_serializers + + called = [] + + def boom(field, field_info): + called.append(True) + raise AttributeError('should not reach DAB inject') + + monkeypatch.setattr(validation_patterns, '_dab_inject_clean_text_patterns', boom) + monkeypatch.setattr(validation_patterns, '_get_tier2_pattern', lambda: {'pattern': 'T2', 'description': 'd2', 'flags': 'i'}) + monkeypatch.setattr(validation_patterns, '_get_tier1_pattern', lambda: {'pattern': 'T1', 'description': 'd1', 'flags': 'u', 'normalize': 'NFC'}) + + class FakeOpts: + app_label = 'conf' + object_name = 'SettingSingleton' + + class FakeModel: + _meta = FakeOpts() + + class PlainCleanText(CleanTextMixin, drf_serializers.Serializer): + class Meta: + model = FakeModel + + excluded_fields = frozenset() + name_fields = frozenset() + + serializer = PlainCleanText() + field = drf_serializers.CharField() + field.bind('FOO', serializer) + + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + result = validation_patterns.inject_top_level_clean_text_patterns(field, {'type': 'string'}) + + assert called == [] + assert result['pattern'] == 'T2' + class TestBuildSurveySpecOptionsSchema: def test_injects_patterns_when_toggle_on(self, fake_tier2_pattern):