diff --git a/poetry.lock b/poetry.lock index f4409f179..b7684aaf8 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.4.3 and should not be changed by hand. [[package]] name = "adal" @@ -973,7 +973,7 @@ testing = ["cryptography", "pytest", "pytest-django"] type = "git" url = "https://github.com/ansible/django-ansible-base.git" reference = "devel" -resolved_reference = "405ef5827400f4b10f101cfb0b70f8d607ea828e" +resolved_reference = "7701dd4f239fce2761631a306b7e3a5a48211340" [[package]] name = "django-crum" diff --git a/src/aap_eda/api/serializers/activation.py b/src/aap_eda/api/serializers/activation.py index 44998b3ca..5c1876fe0 100644 --- a/src/aap_eda/api/serializers/activation.py +++ b/src/aap_eda/api/serializers/activation.py @@ -19,6 +19,7 @@ from typing import Optional import yaml +from ansible_base.lib.serializers.mixins import CleanTextMixin from django.conf import settings from django.utils import timezone from rest_framework import serializers @@ -574,12 +575,17 @@ def to_representation(self, activation): class ActivationCreateSerializer( + CleanTextMixin, _K8sPodMetadataWriteFields, OrganizationIdFieldMixin, serializers.ModelSerializer, ): """Serializer for creating the Activation.""" + # extra_var may legitimately contain Jinja2 template syntax + # (e.g. credential injectors), so it is excluded from free-text checks. + excluded_fields = frozenset({"extra_var"}) + class Meta: model = models.Activation fields = [ @@ -665,7 +671,7 @@ def validate(self, data): _validate_sources_with_event_streams(data=data) _validate_persistence_credential(data=data) _normalize_activation_k8s_pod_fields(data) - return data + return super().validate(data) def create(self, validated_data): rulebook_id = validated_data["rulebook_id"] @@ -708,7 +714,26 @@ def create(self, validated_data): return super().create(validated_data) -class ActivationCopySerializer(serializers.ModelSerializer): +class _ActivationCopyTextCheckSerializer( + CleanTextMixin, serializers.ModelSerializer +): + """Internal-only: re-validate copied free-text fields as new content. + + ActivationCopySerializer.is_valid() only ever validates "name", + since it's constructed with instance=, which + would make CleanTextMixin grandfather "description" as unchanged. + This serializer is never exposed to clients; it's only used from + copy() to re-check the copied description with no instance to + grandfather against, the same way EdaCredentialCreateSerializer + re-validates a copied credential's description. + """ + + class Meta: + model = models.Activation + fields = ["description", "k8s_service_name", "source_mappings"] + + +class ActivationCopySerializer(CleanTextMixin, serializers.ModelSerializer): name = serializers.CharField( required=True, validators=[validators.check_if_activation_name_used] ) @@ -719,6 +744,19 @@ class Meta: def copy(self) -> dict: activation: models.Activation = self.instance + + text_check = _ActivationCopyTextCheckSerializer( + data={ + "description": activation.description, + "k8s_service_name": activation.k8s_service_name, + "source_mappings": activation.source_mappings, + } + ) + text_check.is_valid(raise_exception=True) + description = text_check.validated_data["description"] + k8s_service_name = text_check.validated_data["k8s_service_name"] + source_mappings = text_check.validated_data["source_mappings"] + pod_metadata = _activation_k8s_pod_metadata_payload(activation) _normalize_activation_k8s_pod_fields(pod_metadata) validators.check_if_k8s_pod_service_account_name_valid( @@ -740,7 +778,7 @@ def copy(self) -> dict: copied_data = { "name": self.validated_data["name"], - "description": activation.description, + "description": description, "is_enabled": False, "decision_environment": activation.decision_environment, "rulebook": activation.rulebook, @@ -751,8 +789,8 @@ def copy(self) -> dict: "awx_token_id": activation.awx_token, "log_level": activation.log_level, "eda_credentials": activation.eda_credentials.all(), - "k8s_service_name": activation.k8s_service_name, - "source_mappings": activation.source_mappings, + "k8s_service_name": k8s_service_name, + "source_mappings": source_mappings, "event_streams": activation.event_streams.all(), "skip_audit_events": activation.skip_audit_events, "rulebook_name": activation.rulebook.name, @@ -782,12 +820,17 @@ def copy(self) -> dict: class ActivationUpdateSerializer( + CleanTextMixin, _K8sPodMetadataWriteFields, OrganizationIdFieldMixin, serializers.ModelSerializer, ): """Serializer for updating the Activation.""" + # extra_var may legitimately contain Jinja2 template syntax + # (e.g. credential injectors), so it is excluded from free-text checks. + excluded_fields = frozenset({"extra_var"}) + class Meta: model = models.Activation fields = [ @@ -897,7 +940,7 @@ def validate(self, data): _validate_sources_with_event_streams(data=data) _validate_persistence_credential(data=data) _normalize_activation_k8s_pod_fields(data) - return data + return super().validate(data) def prepare_update(self, activation: models.Activation): rulebook_id = self.validated_data.get("rulebook_id") diff --git a/src/aap_eda/api/serializers/credential_input_source.py b/src/aap_eda/api/serializers/credential_input_source.py index 260108529..ed654e9ae 100644 --- a/src/aap_eda/api/serializers/credential_input_source.py +++ b/src/aap_eda/api/serializers/credential_input_source.py @@ -12,6 +12,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from ansible_base.lib.serializers.mixins import CleanTextMixin from drf_spectacular.utils import extend_schema_field from rest_framework import serializers @@ -99,9 +100,15 @@ def to_representation(self, instance): } -class CredentialInputSourceCreateSerializer(serializers.ModelSerializer): +class CredentialInputSourceCreateSerializer( + CleanTextMixin, serializers.ModelSerializer +): """Serializer used during the Create process of the instance.""" + # metadata holds credential input values and is populated + # programmatically, so it is excluded from free-text checks. + excluded_fields = frozenset({"metadata"}) + target_credential = serializers.PrimaryKeyRelatedField( queryset=models.EdaCredential.objects.all(), required=True, @@ -149,7 +156,7 @@ def validate(self, attrs): attrs.get("input_field_name"), ) - return attrs + return super().validate(attrs) class Meta: model = models.CredentialInputSource @@ -163,9 +170,15 @@ class Meta: ] -class CredentialInputSourceUpdateSerializer(serializers.ModelSerializer): +class CredentialInputSourceUpdateSerializer( + CleanTextMixin, serializers.ModelSerializer +): """Serializer used during update of the instance.""" + # metadata holds credential input values and is populated + # programmatically, so it is excluded from free-text checks. + excluded_fields = frozenset({"metadata"}) + organization_id = serializers.IntegerField( required=True, allow_null=False, @@ -185,7 +198,7 @@ def validate(self, attrs): if bool(errors): raise serializers.ValidationError(errors) - return attrs + return super().validate(attrs) class Meta: model = models.CredentialInputSource diff --git a/src/aap_eda/api/serializers/credential_type.py b/src/aap_eda/api/serializers/credential_type.py index 73d3219a8..3b750f71e 100644 --- a/src/aap_eda/api/serializers/credential_type.py +++ b/src/aap_eda/api/serializers/credential_type.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from ansible_base.lib.serializers.mixins import CleanTextMixin from rest_framework import serializers from aap_eda.core import models, validators @@ -38,7 +39,13 @@ class Meta: ] -class CredentialTypeCreateSerializer(serializers.ModelSerializer): +class CredentialTypeCreateSerializer( + CleanTextMixin, serializers.ModelSerializer +): + # injectors commonly contain Jinja2 template syntax, so it is excluded + # from free-text checks. + excluded_fields = frozenset({"injectors"}) + inputs = serializers.JSONField( required=False, default=dict, @@ -63,7 +70,7 @@ def validate(self, data): if bool(errors): raise serializers.ValidationError(errors) - return data + return super().validate(data) class Meta: model = models.CredentialType diff --git a/src/aap_eda/api/serializers/decision_environment.py b/src/aap_eda/api/serializers/decision_environment.py index 2ed335218..7bc3b33fd 100644 --- a/src/aap_eda/api/serializers/decision_environment.py +++ b/src/aap_eda/api/serializers/decision_environment.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from ansible_base.lib.serializers.mixins import CleanTextMixin from rest_framework import serializers from aap_eda.api.serializers.eda_credential import EdaCredentialRefSerializer @@ -58,7 +59,7 @@ def to_representation(self, decision_environment): class DecisionEnvironmentCreateSerializer( - OrganizationIdFieldMixin, serializers.ModelSerializer + CleanTextMixin, OrganizationIdFieldMixin, serializers.ModelSerializer ): """Serializer for creating the DecisionEnvironment.""" @@ -75,7 +76,7 @@ def validate(self, data): image_url = data.get("image_url") or self.instance.image_url validators.check_if_de_valid(image_url, eda_credential_id) - return data + return super().validate(data) class Meta: model = models.DecisionEnvironment diff --git a/src/aap_eda/api/serializers/eda_credential.py b/src/aap_eda/api/serializers/eda_credential.py index c929614f2..0b29d6b66 100644 --- a/src/aap_eda/api/serializers/eda_credential.py +++ b/src/aap_eda/api/serializers/eda_credential.py @@ -13,6 +13,7 @@ # limitations under the License. import logging +from ansible_base.lib.serializers.mixins import CleanTextMixin from django.db.models import Q from django.urls import reverse from drf_spectacular.utils import extend_schema_field @@ -55,6 +56,8 @@ class EdaCredentialReferenceField(serializers.JSONField): class EdaCredentialSerializer(serializers.ModelSerializer): + """Serializer used during a GET.""" + inputs = serializers.SerializerMethodField() credential_type = CredentialTypeRefSerializer( required=False, allow_null=True @@ -120,7 +123,7 @@ def to_representation(self, eda_credential): } -class EdaCredentialCopySerializer(serializers.ModelSerializer): +class EdaCredentialCopySerializer(CleanTextMixin, serializers.ModelSerializer): name = serializers.CharField( required=True, validators=[validators.check_if_credential_name_used], @@ -135,8 +138,12 @@ class Meta: class EdaCredentialCreateSerializer( - OrganizationIdFieldMixin, serializers.ModelSerializer + CleanTextMixin, OrganizationIdFieldMixin, serializers.ModelSerializer ): + # inputs holds arbitrary secret credential values, so it is excluded + # from free-text checks. + excluded_fields = frozenset({"inputs"}) + credential_type_id = serializers.IntegerField( required=True, allow_null=False, @@ -181,7 +188,7 @@ def validate(self, data): if bool(errors): raise serializers.ValidationError(errors) - return data + return super().validate(data) class Meta: model = models.EdaCredential @@ -195,8 +202,12 @@ class Meta: class EdaCredentialUpdateSerializer( - OrganizationIdFieldMixin, serializers.ModelSerializer + CleanTextMixin, OrganizationIdFieldMixin, serializers.ModelSerializer ): + # inputs holds arbitrary secret credential values, so it is excluded + # from free-text checks. + excluded_fields = frozenset({"inputs"}) + inputs = serializers.JSONField() def validate(self, data): @@ -212,7 +223,7 @@ def validate(self, data): inputs = data.get("inputs", {}) # allow empty inputs during updating if self.partial and not bool(inputs): - return data + return super().validate(data) errors = validate_inputs( credential_type, @@ -224,7 +235,7 @@ def validate(self, data): if bool(errors): raise serializers.ValidationError(errors) - return data + return super().validate(data) class Meta: model = models.EdaCredential diff --git a/src/aap_eda/api/serializers/event_stream.py b/src/aap_eda/api/serializers/event_stream.py index 1f8080b14..868a016e1 100644 --- a/src/aap_eda/api/serializers/event_stream.py +++ b/src/aap_eda/api/serializers/event_stream.py @@ -16,6 +16,7 @@ from urllib.parse import urljoin import yaml +from ansible_base.lib.serializers.mixins import CleanTextMixin from django.conf import settings from django.urls import reverse from rest_framework import serializers @@ -27,7 +28,7 @@ from aap_eda.core import enums, models, validators -class EventStreamInSerializer(serializers.ModelSerializer): +class EventStreamInSerializer(CleanTextMixin, serializers.ModelSerializer): organization_id = serializers.IntegerField( required=True, allow_null=False, @@ -80,7 +81,7 @@ def validate(self, data): if not event_stream_type: data["event_stream_type"] = kind - return data + return super().validate(data) class Meta: model = models.EventStream @@ -96,6 +97,8 @@ class Meta: class EventStreamOutSerializer(serializers.ModelSerializer): + """Serializer used during a GET.""" + owner = serializers.SerializerMethodField() organization = serializers.SerializerMethodField() eda_credential = EdaCredentialRefSerializer( diff --git a/src/aap_eda/api/serializers/organization.py b/src/aap_eda/api/serializers/organization.py index b9f4199b3..5c01821b9 100644 --- a/src/aap_eda/api/serializers/organization.py +++ b/src/aap_eda/api/serializers/organization.py @@ -13,6 +13,7 @@ # limitations under the License. from ansible_base.lib.serializers.common import NamedCommonModelSerializer +from ansible_base.lib.serializers.mixins import CleanTextMixin from django.conf import settings from aap_eda.api import exceptions as api_exc @@ -23,6 +24,7 @@ class OrganizationSerializer( + CleanTextMixin, NamedCommonModelSerializer, SharedResourceSerializerMixin, ): @@ -53,7 +55,7 @@ def validate(self, data): raise api_exc.Conflict( "The default organization cannot be modified." ) - return data + return super().validate(data) class OrganizationRefSerializer(NamedCommonModelSerializer): @@ -66,7 +68,7 @@ class Meta: ] -class OrganizationCreateSerializer(NamedCommonModelSerializer): +class OrganizationCreateSerializer(CleanTextMixin, NamedCommonModelSerializer): class Meta: model = Organization fields = [ diff --git a/src/aap_eda/api/serializers/project.py b/src/aap_eda/api/serializers/project.py index a1242db3d..3b5dd3e9d 100644 --- a/src/aap_eda/api/serializers/project.py +++ b/src/aap_eda/api/serializers/project.py @@ -14,6 +14,7 @@ from urllib.parse import urlparse, urlunparse +from ansible_base.lib.serializers.mixins import CleanTextMixin from rest_framework import serializers from rest_framework.validators import UniqueValidator @@ -86,8 +87,10 @@ def to_representation(self, instance): class ProjectCreateRequestSerializer( - OrganizationIdFieldMixin, serializers.ModelSerializer + CleanTextMixin, OrganizationIdFieldMixin, serializers.ModelSerializer ): + excluded_fields = frozenset({"proxy"}) + eda_credential_id = serializers.IntegerField( required=False, allow_null=True, @@ -158,8 +161,10 @@ class Meta: class ProjectUpdateRequestSerializer( - OrganizationIdFieldMixin, serializers.ModelSerializer + CleanTextMixin, OrganizationIdFieldMixin, serializers.ModelSerializer ): + excluded_fields = frozenset({"proxy"}) + name = serializers.CharField( required=False, allow_blank=False, @@ -279,7 +284,7 @@ def validate(self, data): raise serializers.ValidationError( "The password in the proxy field should be unencrypted" ) - return data + return super().validate(data) def to_representation(self, instance): result = super().to_representation(instance) diff --git a/src/aap_eda/api/serializers/team.py b/src/aap_eda/api/serializers/team.py index fa7e90104..13fea6a34 100644 --- a/src/aap_eda/api/serializers/team.py +++ b/src/aap_eda/api/serializers/team.py @@ -11,6 +11,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from ansible_base.lib.serializers.mixins import CleanTextMixin from rest_framework import serializers from rest_framework.validators import UniqueTogetherValidator @@ -41,6 +42,7 @@ class Meta: class TeamCreateSerializer( + CleanTextMixin, OrganizationIdFieldMixin, serializers.ModelSerializer, SharedResourceSerializerMixin, @@ -63,7 +65,7 @@ class Meta: def validate(self, data): self.validate_shared_resource() - return data + return super().validate(data) class TeamDetailSerializer(serializers.ModelSerializer): @@ -86,6 +88,7 @@ class Meta: class TeamUpdateSerializer( + CleanTextMixin, serializers.ModelSerializer, SharedResourceSerializerMixin, ): @@ -98,4 +101,4 @@ class Meta: def validate(self, data): self.validate_shared_resource() - return data + return super().validate(data) diff --git a/src/aap_eda/api/serializers/user.py b/src/aap_eda/api/serializers/user.py index fcf2831f8..b33795597 100644 --- a/src/aap_eda/api/serializers/user.py +++ b/src/aap_eda/api/serializers/user.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from ansible_base.lib.serializers.mixins import CleanTextMixin from django.contrib.auth.hashers import make_password from rest_framework import serializers from rest_framework.exceptions import PermissionDenied @@ -102,9 +103,12 @@ class UserListSerializer(serializers.Serializer): class UserUpdateSerializerBase( + CleanTextMixin, serializers.ModelSerializer, SharedResourceSerializerMixin, ): + excluded_fields = frozenset({"password"}) + username = serializers.CharField( help_text=USERNAME_HELP, ) @@ -128,7 +132,7 @@ def validate_is_superuser(self, value): def validate(self, data): self.validate_shared_resource() - return data + return super().validate(data) class UserCreateUpdateSerializer(UserUpdateSerializerBase): @@ -183,7 +187,11 @@ class Meta: ] -class AwxTokenCreateSerializer(serializers.ModelSerializer): +class AwxTokenCreateSerializer(CleanTextMixin, serializers.ModelSerializer): + # token holds the secret AWX token value, so it is excluded from + # free-text checks. + excluded_fields = frozenset({"token"}) + class Meta: model = models.AwxToken fields = [ @@ -206,4 +214,4 @@ def validate(self, data): if existing_token.exists(): raise Conflict("Token with this name already exists.") - return data + return super().validate(data) diff --git a/tests/integration/api/test_clean_text_mixin.py b/tests/integration/api/test_clean_text_mixin.py new file mode 100644 index 000000000..933eacc06 --- /dev/null +++ b/tests/integration/api/test_clean_text_mixin.py @@ -0,0 +1,951 @@ +# Copyright 2026 Red Hat, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Integration tests verifying CleanTextMixin is correctly wired to EDA +serializers. + +These tests verify that the CleanTextMixin validation (two-tier text +validation with grandfathering) works correctly on all EDA serializers +that were updated in AAP-78702. + +The validation is gated behind ENHANCED_INPUT_VALIDATION_ENABLED, so a +module-level autouse fixture (enable_input_validation) enables it for +every test in this module. +""" + +from types import SimpleNamespace +from unittest.mock import patch + +import pytest +from django.test import override_settings +from rest_framework import status +from rest_framework.test import APIClient + +from aap_eda.api.serializers.credential_input_source import ( + CredentialInputSourceCreateSerializer, + CredentialInputSourceUpdateSerializer, +) +from aap_eda.api.serializers.credential_type import ( + CredentialTypeCreateSerializer, +) +from aap_eda.api.serializers.decision_environment import ( + DecisionEnvironmentCreateSerializer, +) +from aap_eda.api.serializers.eda_credential import ( + EdaCredentialCreateSerializer, + EdaCredentialUpdateSerializer, +) +from aap_eda.api.serializers.event_stream import EventStreamInSerializer +from aap_eda.api.serializers.project import ( + ProjectCreateRequestSerializer, + ProjectUpdateRequestSerializer, +) +from aap_eda.api.serializers.user import ( + AwxTokenCreateSerializer, + UserCreateUpdateSerializer, +) +from aap_eda.core import enums, models +from tests.integration.constants import api_url_v1 + +DANGEROUS_NAME = "" +DANGEROUS_TEXT = "$(rm -rf /)" +VALID_NAME = "Valid Resource Name" +VALID_USERNAME = "valid.user123" + + +@pytest.fixture(autouse=True) +def enable_input_validation(settings): + settings.ENHANCED_INPUT_VALIDATION_ENABLED = True + settings.RULEBOOK_WORKER_QUEUES = [] + + +@pytest.mark.django_db +class TestOrganizationCleanText: + """Test CleanTextMixin integration with OrganizationSerializer.""" + + def test_rejects_invalid_name_on_create( + self, use_local_resource_setting, superuser_client: APIClient + ): + response = superuser_client.post( + f"{api_url_v1}/organizations/", data={"name": DANGEROUS_NAME} + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "name" in response.data + + def test_accepts_valid_name_on_create( + self, use_local_resource_setting, superuser_client: APIClient + ): + response = superuser_client.post( + f"{api_url_v1}/organizations/", data={"name": VALID_NAME} + ) + assert response.status_code == status.HTTP_201_CREATED + assert response.data["name"] == VALID_NAME + + def test_rejects_invalid_description_on_create( + self, use_local_resource_setting, superuser_client: APIClient + ): + response = superuser_client.post( + f"{api_url_v1}/organizations/", + data={"name": VALID_NAME, "description": DANGEROUS_TEXT}, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "description" in response.data + + def test_grandfather_unchanged_name_on_update( + self, + use_local_resource_setting, + new_organization: models.Organization, + superuser_client: APIClient, + ): + models.Organization.objects.filter(pk=new_organization.pk).update( + name="name;semicolon" + ) + response = superuser_client.patch( + f"{api_url_v1}/organizations/{new_organization.id}/", + data={ + "name": "name;semicolon", + "description": "Updated description", + }, + ) + assert response.status_code == status.HTTP_200_OK + + def test_rejects_changed_invalid_name_on_update( + self, + use_local_resource_setting, + new_organization: models.Organization, + superuser_client: APIClient, + ): + response = superuser_client.patch( + f"{api_url_v1}/organizations/{new_organization.id}/", + data={"name": DANGEROUS_NAME}, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "name" in response.data + + +@pytest.mark.django_db +class TestTeamCleanText: + """Test CleanTextMixin integration with TeamSerializer.""" + + def test_rejects_invalid_name_on_create( + self, + use_local_resource_setting, + default_organization: models.Organization, + admin_client: APIClient, + ): + data_in = { + "name": DANGEROUS_NAME, + "organization_id": default_organization.id, + } + response = admin_client.post(f"{api_url_v1}/teams/", data=data_in) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "name" in response.data + + def test_accepts_valid_name_on_create( + self, + use_local_resource_setting, + default_organization: models.Organization, + admin_client: APIClient, + ): + data_in = { + "name": VALID_NAME, + "organization_id": default_organization.id, + } + response = admin_client.post(f"{api_url_v1}/teams/", data=data_in) + assert response.status_code == status.HTTP_201_CREATED + + def test_rejects_invalid_description_on_create( + self, + use_local_resource_setting, + default_organization: models.Organization, + admin_client: APIClient, + ): + data_in = { + "name": VALID_NAME, + "description": DANGEROUS_TEXT, + "organization_id": default_organization.id, + } + response = admin_client.post(f"{api_url_v1}/teams/", data=data_in) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "description" in response.data + + def test_grandfather_unchanged_name_on_update( + self, + use_local_resource_setting, + default_team: models.Team, + admin_client: APIClient, + ): + models.Team.objects.filter(pk=default_team.pk).update( + name="team" + ) + response = admin_client.patch( + f"{api_url_v1}/teams/{default_team.id}/", + data={ + "name": "team", + "description": "Updated", + }, + ) + assert response.status_code == status.HTTP_200_OK + + +@pytest.mark.django_db +class TestUserCleanText: + """Test CleanTextMixin integration with UserSerializer / + UserUpdateSerializerBase. + """ + + def test_rejects_invalid_username_on_create( + self, use_local_resource_setting, admin_client: APIClient + ): + data_in = {"username": DANGEROUS_NAME, "password": "secret"} + response = admin_client.post(f"{api_url_v1}/users/", data=data_in) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "username" in response.data + + def test_accepts_valid_username_on_create( + self, use_local_resource_setting, admin_client: APIClient + ): + data_in = {"username": VALID_USERNAME, "password": "secret"} + response = admin_client.post(f"{api_url_v1}/users/", data=data_in) + assert response.status_code == status.HTTP_201_CREATED + + def test_rejects_invalid_first_name_on_create( + self, use_local_resource_setting, admin_client: APIClient + ): + data_in = { + "username": VALID_USERNAME, + "password": "secret", + "first_name": DANGEROUS_TEXT, + } + response = admin_client.post(f"{api_url_v1}/users/", data=data_in) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "first_name" in response.data + + def test_rejects_invalid_last_name_on_create( + self, use_local_resource_setting, admin_client: APIClient + ): + data_in = { + "username": VALID_USERNAME, + "password": "secret", + "last_name": DANGEROUS_TEXT, + } + response = admin_client.post(f"{api_url_v1}/users/", data=data_in) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "last_name" in response.data + + def test_grandfather_unchanged_username_on_update( + self, + use_local_resource_setting, + admin_client: APIClient, + admin_user: models.User, + ): + models.User.objects.filter(pk=admin_user.pk).update( + username="user;invalid" + ) + response = admin_client.patch( + f"{api_url_v1}/users/{admin_user.id}/", + data={"username": "user;invalid", "first_name": "Updated"}, + ) + assert response.status_code == status.HTTP_200_OK + + def test_rejects_changed_invalid_username_on_update( + self, + use_local_resource_setting, + admin_client: APIClient, + admin_user: models.User, + ): + response = admin_client.patch( + f"{api_url_v1}/users/{admin_user.id}/", + data={"username": DANGEROUS_NAME}, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "username" in response.data + + +@pytest.mark.django_db +class TestActivationCleanText: + """Test CleanTextMixin integration with ActivationCreateSerializer / + ActivationUpdateSerializer. + """ + + @patch( + "aap_eda.api.views.activation.check_dispatcherd_workers_health", + return_value=True, + ) + def test_rejects_invalid_name_on_create( + self, + mock_health_check, + admin_awx_token: models.AwxToken, + activation_payload: dict, + admin_client: APIClient, + ): + activation_payload["name"] = DANGEROUS_NAME + response = admin_client.post( + f"{api_url_v1}/activations/", data=activation_payload + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "name" in response.data + + @patch( + "aap_eda.api.views.activation.check_dispatcherd_workers_health", + return_value=True, + ) + def test_accepts_valid_name_on_create( + self, + mock_health_check, + admin_awx_token: models.AwxToken, + activation_payload: dict, + admin_client: APIClient, + ): + response = admin_client.post( + f"{api_url_v1}/activations/", data=activation_payload + ) + assert response.status_code == status.HTTP_201_CREATED + + @patch( + "aap_eda.api.views.activation.check_dispatcherd_workers_health", + return_value=True, + ) + def test_grandfather_unchanged_description_on_update( + self, + mock_health_check, + admin_awx_token: models.AwxToken, + activation_payload: dict, + admin_client: APIClient, + ): + activation_payload["is_enabled"] = False + response = admin_client.post( + f"{api_url_v1}/activations/", data=activation_payload + ) + assert response.status_code == status.HTTP_201_CREATED + activation_id = response.data["id"] + + models.Activation.objects.filter(pk=activation_id).update( + description="description;semicolon" + ) + response = admin_client.patch( + f"{api_url_v1}/activations/{activation_id}/", + data={"description": "description;semicolon"}, + ) + assert response.status_code == status.HTTP_200_OK + + @patch( + "aap_eda.api.views.activation.check_dispatcherd_workers_health", + return_value=True, + ) + def test_rejects_changed_invalid_description_on_update( + self, + mock_health_check, + admin_awx_token: models.AwxToken, + activation_payload: dict, + admin_client: APIClient, + ): + activation_payload["is_enabled"] = False + response = admin_client.post( + f"{api_url_v1}/activations/", data=activation_payload + ) + assert response.status_code == status.HTTP_201_CREATED + activation_id = response.data["id"] + + response = admin_client.patch( + f"{api_url_v1}/activations/{activation_id}/", + data={"description": DANGEROUS_TEXT}, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert "description" in response.data + + @patch( + "aap_eda.api.views.activation.check_dispatcherd_workers_health", + return_value=True, + ) + def test_copy_propagates_valid_description( + self, + mock_health_check, + admin_awx_token: models.AwxToken, + activation_payload: dict, + admin_client: APIClient, + ): + response = admin_client.post( + f"{api_url_v1}/activations/", data=activation_payload + ) + assert response.status_code == status.HTTP_201_CREATED + activation_id = response.data["id"] + + response = admin_client.post( + f"{api_url_v1}/activations/{activation_id}/copy/", + data={"name": "copied-activation"}, + ) + assert response.status_code == status.HTTP_201_CREATED + copied = models.Activation.objects.get(name="copied-activation") + assert copied.description == activation_payload["description"] + + @patch( + "aap_eda.api.views.activation.check_dispatcherd_workers_health", + return_value=True, + ) + def test_rejects_copy_of_grandfathered_invalid_description( + self, + mock_health_check, + admin_awx_token: models.AwxToken, + activation_payload: dict, + admin_client: APIClient, + ): + """A /copy/ must not propagate blocklisted text into a new row, + even when the source activation's description was grandfathered + in (e.g. it predates ENHANCED_INPUT_VALIDATION_ENABLED). + """ + response = admin_client.post( + f"{api_url_v1}/activations/", data=activation_payload + ) + assert response.status_code == status.HTTP_201_CREATED + activation_id = response.data["id"] + + models.Activation.objects.filter(pk=activation_id).update( + description=DANGEROUS_TEXT + ) + + response = admin_client.post( + f"{api_url_v1}/activations/{activation_id}/copy/", + data={"name": "copied-activation"}, + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + assert not models.Activation.objects.filter( + name="copied-activation" + ).exists() + + +@pytest.mark.django_db +class TestProjectCleanText: + """Test CleanTextMixin integration with ProjectCreateRequestSerializer / + ProjectUpdateRequestSerializer. + """ + + def test_rejects_invalid_name_on_create( + self, default_organization: models.Organization + ): + serializer = ProjectCreateRequestSerializer( + data={ + "name": DANGEROUS_NAME, + "url": "https://git.example.com/acme/project-01", + "organization_id": default_organization.id, + } + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + def test_accepts_valid_name_on_create( + self, default_organization: models.Organization + ): + serializer = ProjectCreateRequestSerializer( + data={ + "name": VALID_NAME, + "url": "https://git.example.com/acme/project-01", + "organization_id": default_organization.id, + } + ) + assert serializer.is_valid(), serializer.errors + assert "name" not in serializer.errors + + def test_rejects_invalid_description_on_create( + self, default_organization: models.Organization + ): + serializer = ProjectCreateRequestSerializer( + data={ + "name": VALID_NAME, + "description": DANGEROUS_TEXT, + "url": "https://git.example.com/acme/project-01", + "organization_id": default_organization.id, + } + ) + assert not serializer.is_valid() + assert "description" in serializer.errors + + def test_grandfather_unchanged_name_on_update( + self, default_project: models.Project + ): + models.Project.objects.filter(pk=default_project.pk).update( + name="project;invalid" + ) + default_project.refresh_from_db() + + serializer = ProjectUpdateRequestSerializer( + instance=default_project, + data={"name": "project;invalid"}, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + def test_rejects_changed_invalid_name_on_update( + self, default_project: models.Project + ): + serializer = ProjectUpdateRequestSerializer( + instance=default_project, + data={"name": DANGEROUS_NAME}, + partial=True, + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + +@pytest.mark.django_db +class TestDecisionEnvironmentCleanText: + """Test CleanTextMixin integration with + DecisionEnvironmentCreateSerializer. + """ + + def test_rejects_invalid_name_on_create( + self, default_organization: models.Organization + ): + serializer = DecisionEnvironmentCreateSerializer( + data={ + "name": DANGEROUS_NAME, + "image_url": "quay.io/ansible/ansible-rulebook:latest", + "organization_id": default_organization.id, + } + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + def test_accepts_valid_name_on_create( + self, default_organization: models.Organization + ): + serializer = DecisionEnvironmentCreateSerializer( + data={ + "name": VALID_NAME, + "image_url": "quay.io/ansible/ansible-rulebook:latest", + "organization_id": default_organization.id, + } + ) + assert serializer.is_valid(), serializer.errors + + def test_grandfather_unchanged_name_on_update( + self, default_decision_environment: models.DecisionEnvironment + ): + models.DecisionEnvironment.objects.filter( + pk=default_decision_environment.pk + ).update(name="de`invalid") + default_decision_environment.refresh_from_db() + + serializer = DecisionEnvironmentCreateSerializer( + instance=default_decision_environment, + data={"name": "de`invalid"}, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + def test_rejects_changed_invalid_name_on_update( + self, default_decision_environment: models.DecisionEnvironment + ): + serializer = DecisionEnvironmentCreateSerializer( + instance=default_decision_environment, + data={"name": DANGEROUS_NAME}, + partial=True, + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + +@pytest.mark.django_db +class TestCredentialTypeCleanText: + """Test CleanTextMixin integration with CredentialTypeCreateSerializer.""" + + def test_rejects_invalid_name_on_create(self): + serializer = CredentialTypeCreateSerializer( + data={"name": DANGEROUS_NAME} + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + def test_accepts_valid_name_on_create(self): + serializer = CredentialTypeCreateSerializer(data={"name": VALID_NAME}) + assert serializer.is_valid(), serializer.errors + + def test_rejects_invalid_description_on_create(self): + serializer = CredentialTypeCreateSerializer( + data={"name": VALID_NAME, "description": DANGEROUS_TEXT} + ) + assert not serializer.is_valid() + assert "description" in serializer.errors + + def test_grandfather_unchanged_name_on_update( + self, credential_type: models.CredentialType + ): + models.CredentialType.objects.filter(pk=credential_type.pk).update( + name="type;invalid" + ) + credential_type.refresh_from_db() + + serializer = CredentialTypeCreateSerializer( + instance=credential_type, + data={"name": "type;invalid"}, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + +@pytest.mark.django_db +class TestEdaCredentialCleanText: + """Test CleanTextMixin integration with EdaCredentialCreateSerializer / + EdaCredentialUpdateSerializer. + """ + + def test_rejects_invalid_name_on_create( + self, + default_organization: models.Organization, + preseed_credential_types, + ): + registry_type = models.CredentialType.objects.get( + name=enums.DefaultCredentialType.REGISTRY + ) + serializer = EdaCredentialCreateSerializer( + data={ + "name": DANGEROUS_NAME, + "credential_type_id": registry_type.id, + "inputs": { + "username": "dummy-user", + "password": "dummy-password", + }, + "organization_id": default_organization.id, + } + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + def test_accepts_valid_name_on_create( + self, + default_organization: models.Organization, + preseed_credential_types, + ): + registry_type = models.CredentialType.objects.get( + name=enums.DefaultCredentialType.REGISTRY + ) + serializer = EdaCredentialCreateSerializer( + data={ + "name": VALID_NAME, + "credential_type_id": registry_type.id, + "inputs": { + "username": "dummy-user", + "password": "dummy-password", + }, + "organization_id": default_organization.id, + } + ) + assert serializer.is_valid(), serializer.errors + + def test_grandfather_unchanged_name_on_update( + self, default_registry_credential: models.EdaCredential + ): + models.EdaCredential.objects.filter( + pk=default_registry_credential.pk + ).update(name="cred;invalid") + default_registry_credential.refresh_from_db() + + serializer = EdaCredentialUpdateSerializer( + instance=default_registry_credential, + data={"name": "cred;invalid"}, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + def test_rejects_changed_invalid_name_on_update( + self, default_registry_credential: models.EdaCredential + ): + serializer = EdaCredentialUpdateSerializer( + instance=default_registry_credential, + data={"name": DANGEROUS_NAME}, + partial=True, + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + +@pytest.mark.django_db +class TestEventStreamCleanText: + """Test CleanTextMixin integration with EventStreamInSerializer.""" + + @override_settings(EVENT_STREAM_BASE_URL="https://www.example.com/") + def test_rejects_invalid_name_on_create( + self, + default_organization: models.Organization, + default_user: models.User, + default_hmac_credential: models.EdaCredential, + ): + serializer = EventStreamInSerializer( + data={ + "name": DANGEROUS_NAME, + "eda_credential_id": default_hmac_credential.id, + "organization_id": default_organization.id, + }, + context={"request": SimpleNamespace(user=default_user, META={})}, + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + @override_settings(EVENT_STREAM_BASE_URL="https://www.example.com/") + def test_accepts_valid_name_on_create( + self, + default_organization: models.Organization, + default_user: models.User, + default_hmac_credential: models.EdaCredential, + ): + serializer = EventStreamInSerializer( + data={ + "name": VALID_NAME, + "eda_credential_id": default_hmac_credential.id, + "organization_id": default_organization.id, + }, + context={"request": SimpleNamespace(user=default_user, META={})}, + ) + assert serializer.is_valid(), serializer.errors + + def test_grandfather_unchanged_name_on_update( + self, default_event_stream: models.EventStream + ): + models.EventStream.objects.filter(pk=default_event_stream.pk).update( + name="es" + ) + default_event_stream.refresh_from_db() + + serializer = EventStreamInSerializer( + instance=default_event_stream, + data={"name": "es"}, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + def test_rejects_changed_invalid_name_on_update( + self, default_event_stream: models.EventStream + ): + serializer = EventStreamInSerializer( + instance=default_event_stream, + data={"name": DANGEROUS_NAME}, + partial=True, + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + +@pytest.mark.django_db +class TestCredentialInputSourceCleanText: + """Test CleanTextMixin integration with + CredentialInputSourceCreateSerializer / + CredentialInputSourceUpdateSerializer. + + Note: CredentialInputSource has no name field, so only the Tier 2 + (free-text) description field is exercised here. + """ + + def test_rejects_invalid_description_on_create( + self, + default_credential_input_source: models.CredentialInputSource, + ): + source = default_credential_input_source + serializer = CredentialInputSourceCreateSerializer( + data={ + "description": DANGEROUS_TEXT, + "source_credential": source.source_credential.id, + "target_credential": source.target_credential.id, + "input_field_name": "username", + "metadata": source.metadata, + "organization_id": source.organization.id, + } + ) + assert not serializer.is_valid() + assert "description" in serializer.errors + + def test_rejects_invalid_description_on_update( + self, + default_credential_input_source: models.CredentialInputSource, + ): + serializer = CredentialInputSourceUpdateSerializer( + instance=default_credential_input_source, + data={"description": DANGEROUS_TEXT}, + partial=True, + ) + assert not serializer.is_valid() + assert "description" in serializer.errors + + def test_accepts_valid_description_on_update( + self, + default_credential_input_source: models.CredentialInputSource, + ): + serializer = CredentialInputSourceUpdateSerializer( + instance=default_credential_input_source, + data={"description": "A perfectly reasonable description"}, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + def test_grandfather_unchanged_description_on_update( + self, + default_credential_input_source: models.CredentialInputSource, + ): + models.CredentialInputSource.objects.filter( + pk=default_credential_input_source.pk + ).update(description="description;semicolon") + default_credential_input_source.refresh_from_db() + + serializer = CredentialInputSourceUpdateSerializer( + instance=default_credential_input_source, + data={"description": "description;semicolon"}, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + +@pytest.mark.django_db +class TestAwxTokenCleanText: + """Test CleanTextMixin integration with AwxTokenCreateSerializer.""" + + def test_rejects_invalid_name_on_create(self, default_user: models.User): + serializer = AwxTokenCreateSerializer( + data={"name": DANGEROUS_NAME, "token": "sometoken"}, + context={"request": SimpleNamespace(user=default_user, META={})}, + ) + assert not serializer.is_valid() + assert "name" in serializer.errors + + def test_accepts_valid_name_on_create(self, default_user: models.User): + serializer = AwxTokenCreateSerializer( + data={"name": VALID_NAME, "token": "sometoken"}, + context={"request": SimpleNamespace(user=default_user, META={})}, + ) + assert serializer.is_valid(), serializer.errors + + def test_rejects_invalid_description_on_create( + self, default_user: models.User + ): + serializer = AwxTokenCreateSerializer( + data={ + "name": VALID_NAME, + "token": "sometoken", + "description": DANGEROUS_TEXT, + }, + context={"request": SimpleNamespace(user=default_user, META={})}, + ) + assert not serializer.is_valid() + assert "description" in serializer.errors + + +@pytest.mark.django_db +class TestExcludedFieldsCleanText: + """Test that fields listed in a serializer's excluded_fields bypass + CleanTextMixin validation. + + Each field below legitimately carries content (Jinja templates, + secrets, opaque tokens) that would otherwise trip the free-text + checks, so it must be accepted even when it contains characters + that are rejected on non-excluded fields like name/description. + """ + + @patch( + "aap_eda.api.views.activation.check_dispatcherd_workers_health", + return_value=True, + ) + def test_activation_extra_var_excluded( + self, + mock_health_check, + admin_awx_token: models.AwxToken, + activation_payload: dict, + admin_client: APIClient, + ): + activation_payload["extra_var"] = f"dangerous: '{DANGEROUS_TEXT}'" + response = admin_client.post( + f"{api_url_v1}/activations/", data=activation_payload + ) + assert response.status_code == status.HTTP_201_CREATED, response.data + + def test_credential_type_injectors_excluded(self): + serializer = CredentialTypeCreateSerializer( + data={ + "name": VALID_NAME, + "inputs": { + "fields": [ + { + "id": "password", + "label": "Password", + "type": "string", + }, + ] + }, + "injectors": {"extra_vars": {"password": DANGEROUS_TEXT}}, + } + ) + assert serializer.is_valid(), serializer.errors + + def test_eda_credential_inputs_excluded( + self, + default_organization: models.Organization, + preseed_credential_types, + ): + registry_type = models.CredentialType.objects.get( + name=enums.DefaultCredentialType.REGISTRY + ) + serializer = EdaCredentialCreateSerializer( + data={ + "name": VALID_NAME, + "credential_type_id": registry_type.id, + "inputs": { + "username": "dummy-user", + "password": DANGEROUS_TEXT, + }, + "organization_id": default_organization.id, + } + ) + assert serializer.is_valid(), serializer.errors + + def test_awx_token_excluded(self, default_user: models.User): + serializer = AwxTokenCreateSerializer( + data={"name": VALID_NAME, "token": DANGEROUS_TEXT}, + context={"request": SimpleNamespace(user=default_user, META={})}, + ) + assert serializer.is_valid(), serializer.errors + + def test_credential_input_source_metadata_excluded( + self, + default_credential_input_source: models.CredentialInputSource, + ): + serializer = CredentialInputSourceUpdateSerializer( + instance=default_credential_input_source, + data={ + "metadata": { + "secret_path": DANGEROUS_TEXT, + "secret_key": "bar", + }, + }, + partial=True, + ) + assert serializer.is_valid(), serializer.errors + + def test_project_proxy_excluded( + self, default_organization: models.Organization + ): + serializer = ProjectCreateRequestSerializer( + data={ + "name": VALID_NAME, + "url": "https://git.example.com/acme/project-01", + "proxy": DANGEROUS_TEXT, + "organization_id": default_organization.id, + } + ) + assert serializer.is_valid(), serializer.errors + + def test_user_password_excluded(self, default_user: models.User): + serializer = UserCreateUpdateSerializer( + data={ + "username": VALID_USERNAME, + "password": DANGEROUS_TEXT, + }, + context={"request": SimpleNamespace(user=default_user, META={})}, + ) + assert serializer.is_valid(), serializer.errors