From 3c76e236c270b831f0f71465c3e0d7fae040c83a Mon Sep 17 00:00:00 2001 From: Maeve Hoffer Date: Wed, 30 Sep 2026 10:46:55 -0700 Subject: [PATCH] [AAP-93690] Add CleanText validation and OPTIONS patterns to subscription wizard (#16671) * [AAP-93690] Add CleanText validation and OPTIONS patterns to subscriptions endpoint Add SubscriptionCredentialsSerializer (PlainSerializerCleanTextMixin) for /api/v2/config/subscriptions/ so POST requests validate credential string fields through the CleanText enforcement pipeline. - subscriptions_client_id, subscriptions_username: Tier 2 (free-text) - subscriptions_client_secret, subscriptions_password: excluded (secrets) - name_fields = frozenset() ensures Tier 2, not Tier 1 - Fake Meta.model satisfies CleanTextMixin audit logging and OPTIONS metadata - ApiV2SubscriptionView gains serializer_class and get_serializer() so the existing metadata machinery advertises validation patterns on OPTIONS - post() runs serializer.is_valid() before credential logic, preserving all existing error messages, status codes, and $encrypted$ handling - Functional tests: unsafe-input rejection, secret exclusion, safe passthrough, OPTIONS pattern presence/absence, all existing happy/error paths still pass Co-Authored-By: Claude Opus 4.6 Signed-off-by: Chai Bot * [AAP-93690] Preserve credential whitespace and improve docstring coverage DRF CharField strips leading/trailing whitespace by default, which silently altered subscription credentials before passing them to validate_rh. Set trim_whitespace=False on all four credential fields so serializer.validated_data preserves the exact submitted values. Add four regression tests proving whitespace is passed unchanged to validate_rh for username, password, client_id, and client_secret. Add docstrings to concrete_model(), check_permissions(), post(), and the ApiV2SubscriptionView class to meet the 80% docstring coverage threshold. Co-Authored-By: Claude Opus 4.6 Signed-off-by: Chai Bot * Accept explicit JSON null on SubscriptionCredentialsSerializer fields DRF CharField rejects explicit JSON null by default, causing a 400 "This field may not be null" error when a frontend sends null for the unused credential pair (e.g. client_id=null, client_secret=null with a valid username/password). Add allow_null=True on all four optional fields so the serializer passes null through to the view, where the existing falsy-check credential-selection logic already handles it correctly. Includes a focused regression test proving a valid username/password pair with null client fields is accepted and calls validate_rh with the basic-auth pair. Co-Authored-By: Claude Opus 4.6 Signed-off-by: Chai Bot * fix(subscriptions): supply request/view context to SubscriptionCredentialsSerializer ApiV2SubscriptionView.get_serializer() now injects the standard DRF serializer context (request, format, view) so that CleanTextMixin's _log_validation_failure() can attribute rejected input to the authenticated user and client IP. post() is updated to call self.get_serializer(data=request.data) instead of constructing the serializer class directly, ensuring consistent context propagation. A focused regression test asserts that the CleanText warning emitted for unsafe subscription input includes the authenticated username and client IP address. Signed-off-by: redhat-chai-bot Co-Authored-By: Claude Opus 4.6 Signed-off-by: redhat-chai-bot * remove jira ref, decouple test from DAB working --------- Signed-off-by: Chai Bot Signed-off-by: redhat-chai-bot Co-authored-by: Claude Opus 4.6 --- awx/api/serializers.py | 64 +++++ awx/api/views/root.py | 31 ++- .../tests/functional/api/test_licensing.py | 232 ++++++++++++++++++ 3 files changed, 326 insertions(+), 1 deletion(-) diff --git a/awx/api/serializers.py b/awx/api/serializers.py index 3670082934..1d0fa65349 100644 --- a/awx/api/serializers.py +++ b/awx/api/serializers.py @@ -253,6 +253,70 @@ class PlainSerializerCleanTextMixin(CleanTextMixin): return text_fields, json_fields +class _SubscriptionCredentialsFakeOpts: + """Minimal stand-in for Django model._meta (AAP-93690). + + CleanTextMixin audit logs need app_label/object_name. OPTIONS metadata + (awx.api.metadata.get_field_info) walks Meta.model._meta.fields when + Meta.model is present -- an empty tuple keeps that path safe. + """ + + app_label = 'main' + object_name = 'SubscriptionCredentials' + verbose_name = 'subscription credentials' + fields = () + + @property + def concrete_model(self): + """Return the fake model class so CleanTextMixin can resolve the concrete model.""" + return _SubscriptionCredentialsFakeModel + + +class _SubscriptionCredentialsFakeModel: + """Stand-in for Meta.model on SubscriptionCredentialsSerializer (AAP-93690). + + Never introspected for real fields -- PlainSerializerCleanTextMixin discovers + CharField entries from self.fields instead. + """ + + _meta = _SubscriptionCredentialsFakeOpts() + + +class SubscriptionCredentialsSerializer(PlainSerializerCleanTextMixin, serializers.Serializer): + """Validate subscription credential fields via CleanTextMixin (AAP-93690). + + Uses PlainSerializerCleanTextMixin because there is no Django model behind + the /api/v2/config/subscriptions/ endpoint. All four credential fields are + optional (allow_blank + allow_null + default='') because only one credential + pair is required per request and callers may send explicit JSON ``null`` for + unused fields; mutual-exclusion and required-pair validation lives in the + view's post() handler to preserve its existing error messages/status codes. + + CleanText enforcement: + - subscriptions_client_id, subscriptions_username: Tier 2 (free-text) + - subscriptions_client_secret, subscriptions_password: excluded (secrets) + """ + + subscriptions_client_id = serializers.CharField(required=False, allow_blank=True, allow_null=True, default='', trim_whitespace=False) + subscriptions_client_secret = serializers.CharField(required=False, allow_blank=True, allow_null=True, default='', trim_whitespace=False) + subscriptions_username = serializers.CharField(required=False, allow_blank=True, allow_null=True, default='', trim_whitespace=False) + subscriptions_password = serializers.CharField(required=False, allow_blank=True, allow_null=True, default='', trim_whitespace=False) + + # Secrets must not be validated by CleanText -- they may contain arbitrary + # characters and are never rendered in the UI. + excluded_fields = frozenset({'subscriptions_client_secret', 'subscriptions_password'}) + + # AC3: client_id and username must use Tier 2 (free-text), not Tier 1 + # (resource-name). DEFAULT_NAME_FIELDS includes 'username', which would + # promote *any* field named exactly 'username' to Tier 1 -- our fields are + # 'subscriptions_username' (no match), but an empty set makes the intent + # explicit and future-proof. + name_fields = frozenset() + + class Meta: + model = _SubscriptionCredentialsFakeModel + + class _CopySerializerFakeModel: """Stand-in for Meta.model on CopySerializer, a plain serializers.Serializer with no backing model of its own (it's reused across many resource types via diff --git a/awx/api/views/root.py b/awx/api/views/root.py index a7d36a5817..b2499aef7d 100644 --- a/awx/api/views/root.py +++ b/awx/api/views/root.py @@ -26,6 +26,7 @@ import requests from ansible_base.lib.utils.schema import extend_schema_if_available from awx.api.generics import APIView +from awx.api.serializers import SubscriptionCredentialsSerializer from awx.conf.registry import settings_registry from awx.main.analytics import all_collectors from awx.main.ha import is_ha_environment @@ -177,21 +178,49 @@ class ApiV2PingView(APIView): class ApiV2SubscriptionView(APIView): + """Validate and list AAP subscription credentials via the Red Hat portal.""" + permission_classes = (IsAuthenticated,) + serializer_class = SubscriptionCredentialsSerializer name = _('Subscriptions') swagger_topic = 'System Configuration' resource_purpose = 'aap subscription validation' def check_permissions(self, request): + """Restrict mutating methods to superusers; allow OPTIONS/HEAD for all authenticated users.""" super(ApiV2SubscriptionView, self).check_permissions(request) if not request.user.is_superuser and request.method.lower() not in {'options', 'head'}: self.permission_denied(request) # Raises PermissionDenied exception. + def get_serializer(self, *args, **kwargs): + """Expose the serializer for OPTIONS metadata. + + Plain APIView has no get_serializer(); this minimal implementation lets + Metadata.determine_actions() discover the serializer's fields and inject + CleanText validation patterns into OPTIONS responses. + + Request and view context is included so that CleanTextMixin's audit + logging can attribute rejected input to the authenticated user and + client IP. + """ + kwargs.setdefault( + 'context', + { + 'request': self.request, + 'format': self.format_kwarg, + 'view': self, + }, + ) + return self.serializer_class(*args, **kwargs) + @extend_schema_if_available( extensions={'x-ai-description': 'List valid AAP subscriptions'}, ) def post(self, request): - data = request.data.copy() + """Validate subscription credentials against the Red Hat portal and persist them on success.""" + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + data = serializer.validated_data try: user = None diff --git a/awx/main/tests/functional/api/test_licensing.py b/awx/main/tests/functional/api/test_licensing.py index a752f0d3f2..ec29ec0819 100644 --- a/awx/main/tests/functional/api/test_licensing.py +++ b/awx/main/tests/functional/api/test_licensing.py @@ -1,7 +1,11 @@ +import logging +import re from unittest.mock import patch, MagicMock import pytest +from django.test import override_settings from awx.api.versioning import reverse +from awx.api import validation_patterns from rest_framework import status @@ -242,3 +246,231 @@ class TestApiV2SubscriptionView: # Basic auth settings should be cleared assert settings.SUBSCRIPTIONS_USERNAME == "" assert settings.SUBSCRIPTIONS_PASSWORD == "" + + def test_null_client_fields_with_valid_basic_auth(self, post, admin): + """Explicit JSON null on unused client fields must not block a valid username/password pair.""" + data = { + 'subscriptions_username': 'test_user', + 'subscriptions_password': 'test_password', + 'subscriptions_client_id': None, + 'subscriptions_client_secret': None, + } + + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + + assert response.status_code == status.HTTP_200_OK + mock_licenser.validate_rh.assert_called_once_with('test_user', 'test_password', True) + + +# --------------------------------------------------------------------------- +# CleanTextMixin on SubscriptionCredentialsSerializer (AAP-93690) +# --------------------------------------------------------------------------- + +UNSAFE_INPUT = '' + +# Deterministic stand-in for the real Tier 2 regex so tests don't depend on +# DAB's pattern generation being importable in every test environment. +FAKE_TIER2_PATTERN = r'^[^\<\>]*$' + + +@pytest.fixture +def enforce_clean_text(): + """Enable CleanTextMixin enforcement for the duration of a test.""" + with patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + yield + + +@pytest.mark.django_db +class TestSubscriptionCleanText: + """CleanText validation on /api/v2/config/subscriptions/ (AAP-93690).""" + + def test_rejects_unsafe_client_id(self, post, admin, enforce_clean_text): + """AC1: POST rejects unsafe input in subscriptions_client_id.""" + data = { + 'subscriptions_client_id': UNSAFE_INPUT, + 'subscriptions_client_secret': 'valid_secret', + } + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'subscriptions_client_id' in response.data + + def test_rejects_unsafe_username(self, post, admin, enforce_clean_text): + """AC1: POST rejects unsafe input in subscriptions_username.""" + data = { + 'subscriptions_username': UNSAFE_INPUT, + 'subscriptions_password': 'valid_password', + } + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert 'subscriptions_username' in response.data + + def test_allows_unsafe_client_secret(self, post, admin, enforce_clean_text): + """AC2: client_secret is excluded from CleanText -- unsafe values pass.""" + data = { + 'subscriptions_client_id': 'safe_id', + 'subscriptions_client_secret': UNSAFE_INPUT, + } + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_200_OK + + def test_allows_unsafe_password(self, post, admin, enforce_clean_text): + """AC2: password is excluded from CleanText -- unsafe values pass.""" + data = { + 'subscriptions_username': 'safe_user', + 'subscriptions_password': UNSAFE_INPUT, + } + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_200_OK + + def test_safe_input_passes_with_enforcement(self, post, admin, enforce_clean_text): + """Safe input succeeds even when enforcement is on.""" + data = { + 'subscriptions_username': 'my_rh_user', + 'subscriptions_password': 'my_rh_pass', + } + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_200_OK + + def test_rejected_input_logs_user_and_ip(self, post, admin, enforce_clean_text, caplog): + """CleanText rejection warning includes authenticated username and client IP. + + The assertion checks that a WARNING-level log record is emitted from the + CleanTextMixin logger and that the formatted message contains the + requesting user's identity and a client IP address. It deliberately + avoids hard-coding DAB's exact message template so the test stays stable + across upstream wording changes. + """ + data = { + 'subscriptions_username': UNSAFE_INPUT, + 'subscriptions_password': 'valid_password', + } + with caplog.at_level(logging.WARNING, logger='ansible_base.lib.serializers.mixins'): + post(reverse('api:api_v2_subscription_view'), data, admin) + + mixin_warnings = [r for r in caplog.records if r.levelno >= logging.WARNING and r.name == 'ansible_base.lib.serializers.mixins'] + assert mixin_warnings, 'Expected a CleanTextMixin warning for unsafe input' + msg = mixin_warnings[0].getMessage() + assert admin.username in msg, f'Warning should contain the authenticated username; got: {msg}' + # Client IP: in the test harness this is typically 127.0.0.1 or similar + assert re.search(r'\d{1,3}(?:\.\d{1,3}){3}', msg), f'Warning should contain a client IP address; got: {msg}' + + +@pytest.mark.django_db +class TestSubscriptionWhitespacePreservation: + """Credential values with leading/trailing whitespace must reach validate_rh unchanged.""" + + def test_whitespace_preserved_in_username(self, post, admin): + """Leading/trailing whitespace in subscriptions_username is not stripped.""" + data = { + 'subscriptions_username': ' spaced_user ', + 'subscriptions_password': 'pw', + } + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_200_OK + mock_licenser.validate_rh.assert_called_once_with(' spaced_user ', 'pw', True) + + def test_whitespace_preserved_in_password(self, post, admin): + """Leading/trailing whitespace in subscriptions_password is not stripped.""" + data = { + 'subscriptions_username': 'user', + 'subscriptions_password': ' secret pass ', + } + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_200_OK + mock_licenser.validate_rh.assert_called_once_with('user', ' secret pass ', True) + + def test_whitespace_preserved_in_client_id(self, post, admin): + """Leading/trailing whitespace in subscriptions_client_id is not stripped.""" + data = { + 'subscriptions_client_id': ' cid ', + 'subscriptions_client_secret': 'sec', + } + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_200_OK + mock_licenser.validate_rh.assert_called_once_with(' cid ', 'sec', False) + + def test_whitespace_preserved_in_client_secret(self, post, admin): + """Leading/trailing whitespace in subscriptions_client_secret is not stripped.""" + data = { + 'subscriptions_client_id': 'cid', + 'subscriptions_client_secret': ' sec ', + } + with patch('awx.api.views.root.get_licenser') as mock_get_licenser: + mock_licenser = MagicMock() + mock_licenser.validate_rh.return_value = [] + mock_get_licenser.return_value = mock_licenser + + response = post(reverse('api:api_v2_subscription_view'), data, admin) + assert response.status_code == status.HTTP_200_OK + mock_licenser.validate_rh.assert_called_once_with('cid', ' sec ', False) + + +@pytest.mark.django_db +class TestSubscriptionOptionsPatterns: + """OPTIONS metadata exposes validation patterns (AAP-93690 AC4).""" + + @pytest.fixture(autouse=True) + def _fake_pattern_helpers(self, monkeypatch): + """Provide deterministic Tier 1/2 pattern helpers so tests don't depend + on DAB's build_tier*_frontend_pattern being importable.""" + 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': FAKE_TIER2_PATTERN, 'description': 'd2', 'flags': 'i'}) + + def test_options_shows_patterns_when_enabled(self, options, admin): + """AC4: client_id and username expose Tier 2 patterns when enforcement is on.""" + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=True): + response = options(reverse('api:api_v2_subscription_view'), None, admin) + + assert response.status_code == status.HTTP_200_OK + post_fields = response.data.get('actions', {}).get('POST', {}) + # Non-excluded fields should have patterns + assert post_fields['subscriptions_client_id']['pattern'] == FAKE_TIER2_PATTERN, 'client_id should have Tier 2 pattern' + assert post_fields['subscriptions_username']['pattern'] == FAKE_TIER2_PATTERN, 'username should have Tier 2 pattern' + # Excluded (secret) fields should NOT have patterns + assert 'pattern' not in post_fields.get('subscriptions_client_secret', {}), 'client_secret should not have a pattern' + assert 'pattern' not in post_fields.get('subscriptions_password', {}), 'password should not have a pattern' + + def test_options_no_patterns_when_disabled(self, options, admin): + """Patterns are absent when enforcement is off.""" + with override_settings(ENHANCED_INPUT_VALIDATION_ENABLED=False): + response = options(reverse('api:api_v2_subscription_view'), None, admin) + + assert response.status_code == status.HTTP_200_OK + post_fields = response.data.get('actions', {}).get('POST', {}) + for field_name in ('subscriptions_client_id', 'subscriptions_username'): + assert 'pattern' not in post_fields.get(field_name, {}), f'{field_name} should not have a pattern when disabled'