[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 <noreply@anthropic.com>
Signed-off-by: Chai Bot <chai-bot@redhat.com>

* [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 <noreply@anthropic.com>
Signed-off-by: Chai Bot <chai-bot@redhat.com>

* 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 <noreply@anthropic.com>
Signed-off-by: Chai Bot <chai-bot@redhat.com>

* 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 <redhat-chai-bot@users.noreply.github.com>
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: redhat-chai-bot <redhat-chai-bot@users.noreply.github.com>

* remove jira ref, decouple test from DAB working

---------

Signed-off-by: Chai Bot <chai-bot@redhat.com>
Signed-off-by: redhat-chai-bot <redhat-chai-bot@users.noreply.github.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

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

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

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 = '<script>alert(1)</script>'
+
+# 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'