diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index 6337b6eab7..ac75dabaeb 100644
--- a/awx/api/serializers.py
+++ b/awx/api/serializers.py
@@ -10,6 +10,7 @@ import yaml
import urllib.parse
from collections import Counter, OrderedDict
from datetime import timedelta
+from types import MappingProxyType
from uuid import uuid4
# Jinja
@@ -42,7 +43,10 @@ from rest_framework.utils.serializer_helpers import ReturnList
from polymorphic.models import PolymorphicModel
# django-ansible-base
+from ansible_base.lib.serializers.mixins import CleanTextMixin
from ansible_base.lib.utils.models import get_type_for_model
+from ansible_base.lib.utils.settings import get_setting
+from ansible_base.lib.utils.validation import DEFAULT_NAME_FIELDS
from ansible_base.rbac.models import RoleEvaluation, ObjectRole
from ansible_base.rbac import permission_registry
@@ -210,16 +214,52 @@ def reverse_gfk(content_object, request):
return {camelcase_to_underscore(content_object.__class__.__name__): content_object.get_absolute_url(request=request)}
-class CopySerializer(serializers.Serializer):
+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.
+
+ Known limitation: unlike the model-field path, this can't replicate the
+ get_internal_type() exclusion of SlugField/IPAddressField-style fields
+ (DRF fields have no equivalent signal) -- not an issue for any serializer
+ using this mixin today, since none declare such fields.
+ """
+
+ def _classify_fields(self, model):
+ text_fields = [name for name, field in self.fields.items() if isinstance(field, serializers.CharField)]
+ return text_fields, []
+
+
+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
+ CopyAPIView subclasses). Only satisfies the `_meta.app_label`/`object_name`
+ attributes CleanTextMixin's audit-log message reads (AAP-78694) -- never
+ introspected for real fields, since PlainSerializerCleanTextMixin overrides
+ field discovery."""
+
+ _meta = type('Meta', (), {'app_label': 'main', 'object_name': 'CopySerializer'})
+
+
+class CopySerializer(PlainSerializerCleanTextMixin, serializers.Serializer):
+ # The submitted name becomes the new object's name (see
+ # CopyAPIView.post -> copy_model_obj in awx/api/generics.py), so it needs
+ # the same Tier 1 protection as the 'name' field on the resource being
+ # copied (AAP-78694) -- without this, Copy bypassed validation entirely.
name = serializers.CharField()
+ class Meta:
+ model = _CopySerializerFakeModel
+
def validate(self, attrs):
name = attrs.get('name')
view = self.context.get('view', None)
obj = view.get_object()
if name == obj.name:
raise serializers.ValidationError(_('The original object is already named {}, a copy from it cannot have the same name.'.format(name)))
- return attrs
+ return super().validate(attrs)
class BaseSerializerMetaclass(serializers.SerializerMetaclass):
@@ -727,6 +767,52 @@ class BaseSerializer(serializers.ModelSerializer, metaclass=BaseSerializerMetacl
return False
+class PromptFieldCleanTextMixin(CleanTextMixin):
+ """
+ CleanTextMixin for serializers whose Meta.model derives from
+ LaunchTimeConfigBase (WorkflowJobTemplate, WorkflowJobTemplateNode,
+ WorkflowJobNode, Schedule). Those models store scm_branch/limit/job_tags/
+ skip_tags as NullablePromptPseudoField descriptors backed by the
+ char_prompts JSONField, not real django.db.models.Field instances, so
+ CleanTextMixin's model._meta.get_fields() auto-discovery can never see
+ them (AAP-78694). This adds them back in explicitly as Tier 2 free-text
+ fields. They are always declared as plain serializer CharFields (see
+ LaunchConfigurationBaseSerializer below), so self.fields has them
+ whenever applicable, and grandfathering still works because the
+ pseudo-field descriptor's __get__ correctly returns the stored
+ char_prompts value via getattr(self.instance, name).
+ """
+
+ prompt_pseudo_fields = frozenset({'scm_branch', 'limit', 'job_tags', 'skip_tags'})
+
+ def _classify_fields(self, model):
+ text_fields, json_fields = super()._classify_fields(model)
+ extra = [f for f in self.prompt_pseudo_fields if f in self.fields and f not in text_fields]
+ return list(text_fields) + extra, json_fields
+
+ def _run_clean_text_validation(self, attrs):
+ """Run Tier 1/2 validation and raise on failure, without chaining into
+ the next class in the MRO's validate(). Needed by BulkJobNodeSerializer,
+ whose own validate() intentionally skips LaunchConfigurationBaseSerializer.validate()
+ (that method assumes unified_job_template/inventory/execution_environment
+ are model instances, but BulkJobNodeSerializer declares them as raw PKs
+ to avoid a DB lookup per node) via `super(LaunchConfigurationBaseSerializer, self)`.
+ A plain `self.validate(attrs)` call there would still run this mixin's
+ own `super().validate(attrs)` chain straight into the skipped class, since
+ Python's zero-arg super() resolves against the instance's full MRO
+ regardless of how the method was invoked -- so the Tier 1/2 checks are
+ exposed here as a standalone step instead.
+ """
+ model = self.Meta.model
+ text_fields, json_fields = self._classify_fields(model)
+ errors = {}
+ self._validate_text_fields(text_fields, attrs, errors)
+ self._validate_json_fields(json_fields, attrs, errors)
+ if errors and get_setting('ENHANCED_INPUT_VALIDATION_ENABLED', False):
+ raise serializers.ValidationError(errors)
+ return attrs
+
+
class EmptySerializer(serializers.Serializer):
pass
@@ -1031,7 +1117,18 @@ class UnifiedJobStdoutSerializer(UnifiedJobSerializer):
return super(UnifiedJobStdoutSerializer, self).get_types()
-class UserSerializer(BaseSerializer):
+class UserSerializer(CleanTextMixin, BaseSerializer):
+ # password is excluded because it is a raw secret at validation time (pre-hash);
+ # running the Tier 2 blocklist against it would both leak nothing useful and
+ # reject otherwise-valid passwords containing blocklisted characters.
+ excluded_fields = frozenset({'password'})
+
+ # username is deliberately left in Tier 1 (default name_fields) even though
+ # Django's stock UnicodeUsernameValidator (^[\w.@+-]+$) allows '+', which
+ # Tier 1's allowlist does not. This is an intentional hardening decision
+ # (ANSTRAT-1756): new/changed usernames containing '+' will be rejected once
+ # enforcement is enabled. Existing '+' usernames are grandfathered until edited.
+
password = serializers.CharField(required=False, default='', allow_blank=True, help_text=_('Field used to change the password.'))
is_system_auditor = serializers.BooleanField(default=False)
show_capabilities = ['edit', 'delete']
@@ -1192,7 +1289,7 @@ class UserActivityStreamSerializer(UserSerializer):
fields = ('*', '-is_system_auditor')
-class OrganizationSerializer(BaseSerializer, OpaQueryPathMixin):
+class OrganizationSerializer(CleanTextMixin, BaseSerializer, OpaQueryPathMixin):
show_capabilities = ['edit', 'delete']
class Meta:
@@ -1314,7 +1411,7 @@ class ProjectOptionsSerializer(BaseSerializer):
return super(ProjectOptionsSerializer, self).validate(attrs)
-class ExecutionEnvironmentSerializer(BaseSerializer):
+class ExecutionEnvironmentSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['edit', 'delete', 'copy']
managed = serializers.ReadOnlyField()
@@ -1349,7 +1446,7 @@ class ExecutionEnvironmentSerializer(BaseSerializer):
return super(ExecutionEnvironmentSerializer, self).validate(attrs)
-class ProjectSerializer(UnifiedJobTemplateSerializer, ProjectOptionsSerializer):
+class ProjectSerializer(CleanTextMixin, UnifiedJobTemplateSerializer, ProjectOptionsSerializer):
status = serializers.ChoiceField(choices=Project.PROJECT_STATUS_CHOICES, read_only=True)
last_update_failed = serializers.BooleanField(read_only=True)
last_updated = serializers.DateTimeField(read_only=True)
@@ -1551,10 +1648,14 @@ class LabelsListMixin(object):
return res
-class InventorySerializer(LabelsListMixin, BaseSerializerWithVariables, OpaQueryPathMixin):
+class InventorySerializer(CleanTextMixin, LabelsListMixin, BaseSerializerWithVariables, OpaQueryPathMixin):
show_capabilities = ['edit', 'delete', 'adhoc', 'copy']
capabilities_prefetch = [{'edit': 'change'}, {'adhoc': 'adhoc'}, {'copy': 'organization.add_inventory'}]
+ # variables is raw YAML/JSON that legitimately contains Jinja2 syntax,
+ # same conflict class as extra_vars.
+ excluded_fields = frozenset({'variables'})
+
class Meta:
model = Inventory
fields = (
@@ -1795,10 +1896,21 @@ class InventoryScriptSerializer(InventorySerializer):
fields = ()
-class HostSerializer(BaseSerializerWithVariables):
+class HostSerializer(CleanTextMixin, BaseSerializerWithVariables):
show_capabilities = ['edit', 'delete']
capabilities_prefetch = [{'edit': 'inventory.change'}]
+ # variables is raw YAML/JSON that legitimately contains Jinja2 syntax.
+ excluded_fields = frozenset({'variables'})
+ # NOT for the single "host:port" syntax (see _get_host_port_from_name
+ # below) -- that colon is stripped from attrs['name'] in validate()
+ # before it ever reaches Tier 1. This demotion is for names
+ # _get_host_port_from_name passes through untouched: any colon count
+ # other than exactly 1, most notably IPv6 addresses, which its own
+ # comment says it explicitly does not parse ("except IPv6 for now").
+ # Tier 1's allowlist has no colon, so those would otherwise be rejected.
+ name_fields = DEFAULT_NAME_FIELDS - {'name'}
+
has_active_failures = serializers.SerializerMethodField()
has_inventory_sources = serializers.SerializerMethodField()
last_job = serializers.SerializerMethodField()
@@ -1979,10 +2091,13 @@ class AnsibleFactsSerializer(BaseSerializer):
return obj.ansible_facts
-class GroupSerializer(BaseSerializerWithVariables):
+class GroupSerializer(CleanTextMixin, BaseSerializerWithVariables):
show_capabilities = ['copy', 'edit', 'delete']
capabilities_prefetch = [{'edit': 'inventory.change'}, {'adhoc': 'inventory.adhoc'}]
+ # variables is raw YAML/JSON that legitimately contains Jinja2 syntax.
+ excluded_fields = frozenset({'variables'})
+
class Meta:
model = Group
fields = ('*', 'inventory', 'variables')
@@ -2383,13 +2498,16 @@ class InventorySourceOptionsSerializer(BaseSerializer):
return summary_fields
-class InventorySourceSerializer(UnifiedJobTemplateSerializer, InventorySourceOptionsSerializer):
+class InventorySourceSerializer(CleanTextMixin, UnifiedJobTemplateSerializer, InventorySourceOptionsSerializer):
status = serializers.ChoiceField(choices=InventorySource.INVENTORY_SOURCE_STATUS_CHOICES, read_only=True)
last_update_failed = serializers.BooleanField(read_only=True)
last_updated = serializers.DateTimeField(read_only=True)
show_capabilities = ['start', 'schedule', 'edit', 'delete']
capabilities_prefetch = [{'edit': 'inventory.change'}, {'start': 'inventory.update'}]
+ # source_vars is raw YAML/JSON that legitimately contains Jinja2 syntax.
+ excluded_fields = frozenset({'source_vars'})
+
class Meta:
model = InventorySource
fields = ('*', 'name', 'inventory', 'update_on_launch', 'update_cache_timeout', 'source_project') + (
@@ -2651,7 +2769,7 @@ class InventoryUpdateCancelSerializer(InventoryUpdateSerializer):
fields = ('can_cancel',)
-class TeamSerializer(BaseSerializer):
+class TeamSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['edit', 'delete']
class Meta:
@@ -2934,10 +3052,15 @@ class ResourceAccessListElementSerializer(UserSerializer):
return ret
-class CredentialTypeSerializer(BaseSerializer):
+class CredentialTypeSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['edit', 'delete']
managed = serializers.ReadOnlyField()
+ # injectors is Jinja2-templated by design (e.g. "{{ api_token }}" in env/file
+ # injection definitions) -- Tier 2's injection blocklist would break every
+ # custom credential type that defines injectors.
+ excluded_fields = frozenset({'injectors'})
+
class Meta:
model = CredentialType
fields = ('*', 'kind', 'namespace', 'name', 'managed', 'inputs', 'injectors')
@@ -3005,11 +3128,19 @@ class CredentialTypeSerializer(BaseSerializer):
return fields
-class CredentialSerializer(BaseSerializer):
+class CredentialSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['edit', 'delete', 'copy', 'use']
capabilities_prefetch = [{'edit': 'change'}, {'use': 'use'}]
managed = serializers.ReadOnlyField()
+ # inputs is a JSONField whose schema is credential-type-dependent, including
+ # admin-defined custom types. Secret keys cannot be listed at class level
+ # (unlike NotificationTemplateSerializer.notification_configuration).
+ # validate() below sets instance excluded_json_keys from
+ # credential_type.secret_fields (the schema's "secret": true flags) so
+ # non-secret strings (username, host, ...) still get Tier 2. If the type
+ # cannot be resolved, the whole blob is skipped (fail closed).
+
class Meta:
model = Credential
fields = ('*', 'organization', 'credential_type', 'managed', 'inputs', 'kind', 'cloud', 'kubernetes')
@@ -3081,6 +3212,16 @@ class CredentialSerializer(BaseSerializer):
def validate(self, attrs):
if self.instance and self.instance.managed:
raise PermissionDenied(detail=_("Modifications not allowed for managed credentials"))
+
+ # Shadow the class-level MappingProxyType on this instance only -- DRF
+ # constructs a new serializer per request, so this does not leak.
+ # Must run before super().validate() so CleanTextMixin sees it.
+ cred_type = attrs.get('credential_type') or getattr(self.instance, 'credential_type', None)
+ if cred_type is not None:
+ self.excluded_json_keys = MappingProxyType({'inputs': frozenset(cred_type.secret_fields)})
+ else:
+ self.excluded_fields = frozenset(self.excluded_fields) | {'inputs'}
+
return super(CredentialSerializer, self).validate(attrs)
def get_validation_exclusions(self, obj=None):
@@ -3183,7 +3324,7 @@ class CredentialSerializerCreate(CredentialSerializer):
return credential
-class CredentialInputSourceSerializer(BaseSerializer):
+class CredentialInputSourceSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['delete']
class Meta:
@@ -3346,10 +3487,14 @@ class JobTemplateMixin(object):
return super().validate(attrs)
-class JobTemplateSerializer(JobTemplateMixin, UnifiedJobTemplateSerializer, JobOptionsSerializer):
+class JobTemplateSerializer(CleanTextMixin, JobTemplateMixin, UnifiedJobTemplateSerializer, JobOptionsSerializer):
show_capabilities = ['start', 'schedule', 'copy', 'edit', 'delete']
capabilities_prefetch = [{'edit': 'change'}, {'start': 'execute'}, {'copy': ['project.use', 'inventory.use']}]
+ # extra_vars is raw YAML/JSON that legitimately contains Jinja2 syntax
+ # ("{{ var }}"), which Tier 2's injection blocklist would reject.
+ excluded_fields = frozenset({'extra_vars'})
+
status = serializers.ChoiceField(choices=JobTemplate.JOB_TEMPLATE_STATUS_CHOICES, read_only=True, required=False)
class Meta:
@@ -3467,7 +3612,13 @@ class JobTemplateWithSpecSerializer(JobTemplateSerializer):
fields = ('*', 'survey_spec')
-class JobSerializer(UnifiedJobSerializer, JobOptionsSerializer):
+class JobSerializer(CleanTextMixin, UnifiedJobSerializer, JobOptionsSerializer):
+ # extra_vars is raw YAML/JSON that legitimately contains Jinja2 syntax
+ # ("{{ var }}"), same conflict class as JobTemplateSerializer.extra_vars
+ # (AAP-78694). JobDetail allows PUT/PATCH while status is "new" (see
+ # JobDetail.update), which is the real write path this protects.
+ excluded_fields = frozenset({'extra_vars'})
+
passwords_needed_to_start = serializers.ReadOnlyField()
artifacts = serializers.SerializerMethodField()
@@ -3677,7 +3828,12 @@ class JobCreateScheduleSerializer(LabelsListMixin, BaseSerializer):
return {'all': _('Unknown, job may have been run before launch configurations were saved.')}
-class AdHocCommandSerializer(UnifiedJobSerializer):
+class AdHocCommandSerializer(CleanTextMixin, UnifiedJobSerializer):
+ # extra_vars/module_args are raw YAML/JSON or ad hoc arguments that
+ # legitimately contain Jinja2 syntax ("{{ var }}"), which Tier 2's
+ # injection blocklist would reject.
+ excluded_fields = frozenset({'extra_vars', 'module_args'})
+
class Meta:
model = AdHocCommand
fields = (
@@ -3835,9 +3991,13 @@ class SystemJobCancelSerializer(SystemJobSerializer):
fields = ('can_cancel',)
-class WorkflowJobTemplateSerializer(JobTemplateMixin, LabelsListMixin, UnifiedJobTemplateSerializer):
+class WorkflowJobTemplateSerializer(PromptFieldCleanTextMixin, JobTemplateMixin, LabelsListMixin, UnifiedJobTemplateSerializer):
show_capabilities = ['start', 'schedule', 'edit', 'copy', 'delete']
capabilities_prefetch = [{'edit': 'change'}, {'start': 'execute'}, {'copy': 'organization.add_workflowjobtemplate'}]
+
+ # extra_vars is raw YAML/JSON that legitimately contains Jinja2 syntax.
+ excluded_fields = frozenset({'extra_vars'})
+
limit = serializers.CharField(allow_blank=True, allow_null=True, required=False, default=None)
scm_branch = serializers.CharField(allow_blank=True, allow_null=True, required=False, default=None)
@@ -4052,7 +4212,7 @@ class WorkflowApprovalListSerializer(WorkflowApprovalSerializer, UnifiedJobListS
fields = ('*', '-controller_node', '-execution_node', 'can_approve_or_deny', 'approval_expiration', 'timed_out')
-class WorkflowApprovalTemplateSerializer(UnifiedJobTemplateSerializer):
+class WorkflowApprovalTemplateSerializer(CleanTextMixin, UnifiedJobTemplateSerializer):
class Meta:
model = WorkflowApprovalTemplate
fields = ('*', 'timeout', 'name')
@@ -4234,7 +4394,22 @@ class LaunchConfigurationBaseSerializer(BaseSerializer):
return attrs
-class WorkflowJobTemplateNodeSerializer(LaunchConfigurationBaseSerializer):
+class WorkflowJobTemplateNodeSerializer(PromptFieldCleanTextMixin, LaunchConfigurationBaseSerializer):
+ # extra_data plays the same role as extra_vars elsewhere (arbitrary
+ # launch-time variables) and is submitted in the same two formats: a
+ # dict, or a raw YAML/JSON string (see parse_yaml_or_json). Its model
+ # field is JSONBlob, whose get_internal_type() reports "TextField" for
+ # legacy DB-migration reasons (awx/main/fields.py); since CleanTextMixin
+ # classifies fields via get_internal_type(), a dict payload is silently
+ # skipped (not a str) while a string payload hits Tier 2's
+ # template-injection blocklist and incorrectly rejects legitimate Jinja
+ # variable references -- same conflict class as extra_vars (AAP-78694).
+ # NOTE: this must be set directly on this class, not on
+ # LaunchConfigurationBaseSerializer -- CleanTextMixin's own
+ # excluded_fields default sits earlier in the MRO than that shared base,
+ # so a base-class override there is silently shadowed and never applies.
+ excluded_fields = frozenset({'extra_data'})
+
success_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
failure_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
always_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
@@ -4287,7 +4462,12 @@ class WorkflowJobTemplateNodeSerializer(LaunchConfigurationBaseSerializer):
return summary_fields
-class WorkflowJobNodeSerializer(LaunchConfigurationBaseSerializer):
+class WorkflowJobNodeSerializer(PromptFieldCleanTextMixin, LaunchConfigurationBaseSerializer):
+ # See WorkflowJobTemplateNodeSerializer.excluded_fields above -- same
+ # reasoning, and same requirement that this live directly on this class
+ # rather than on the shared LaunchConfigurationBaseSerializer base.
+ excluded_fields = frozenset({'extra_data'})
+
success_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
failure_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
always_nodes = serializers.PrimaryKeyRelatedField(many=True, read_only=True)
@@ -4357,7 +4537,7 @@ class WorkflowJobTemplateNodeDetailSerializer(WorkflowJobTemplateNodeSerializer)
return field_class, field_kwargs
-class WorkflowJobTemplateNodeCreateApprovalSerializer(BaseSerializer):
+class WorkflowJobTemplateNodeCreateApprovalSerializer(CleanTextMixin, BaseSerializer):
class Meta:
model = WorkflowApprovalTemplate
fields = ('timeout', 'name', 'description')
@@ -4579,7 +4759,13 @@ class SystemJobEventSerializer(AdHocCommandEventSerializer):
return res
-class JobLaunchSerializer(BaseSerializer):
+class JobLaunchSerializer(CleanTextMixin, BaseSerializer):
+ # extra_vars is declared as a JSONField below, so it's already a parsed
+ # dict/list (not a str) by the time CleanTextMixin.validate() runs, which
+ # makes this a no-op today -- excluded anyway for clarity/future-proofing,
+ # consistent with JobTemplateSerializer's extra_vars exclusion.
+ excluded_fields = frozenset({'extra_vars'})
+
# Representational fields
passwords_needed_to_start = serializers.ReadOnlyField()
can_start_without_user_input = serializers.BooleanField(read_only=True)
@@ -4795,7 +4981,13 @@ class JobLaunchSerializer(BaseSerializer):
return accepted
-class WorkflowJobLaunchSerializer(BaseSerializer):
+class WorkflowJobLaunchSerializer(PromptFieldCleanTextMixin, BaseSerializer):
+ # extra_vars is declared as a VerbatimField below, so it's already a
+ # parsed dict/list (not a str) by the time CleanTextMixin.validate() runs,
+ # which makes this a no-op today -- excluded anyway for clarity/future-
+ # proofing, consistent with WorkflowJobTemplateSerializer's exclusion.
+ excluded_fields = frozenset({'extra_vars'})
+
can_start_without_user_input = serializers.BooleanField(read_only=True)
defaults = serializers.SerializerMethodField()
variables_needed_to_start = serializers.ReadOnlyField()
@@ -4913,6 +5105,10 @@ class BulkJobNodeSerializer(WorkflowJobNodeSerializer):
fields = ('*', 'credentials', 'labels', 'instance_groups') # m2m fields are not canonical for WJ nodes
def validate(self, attrs):
+ # Deliberately skips LaunchConfigurationBaseSerializer.validate() (see
+ # its comment above) -- run Tier 1/2 validation explicitly first
+ # since that skip would otherwise bypass it too (AAP-78694).
+ attrs = self._run_clean_text_validation(attrs)
return super(LaunchConfigurationBaseSerializer, self).validate(attrs)
def get_validation_exclusions(self, obj=None):
@@ -4921,7 +5117,15 @@ class BulkJobNodeSerializer(WorkflowJobNodeSerializer):
return ret
-class BulkJobLaunchSerializer(serializers.Serializer):
+class BulkJobLaunchSerializer(PromptFieldCleanTextMixin, serializers.Serializer):
+ # extra_vars is raw YAML/JSON that legitimately contains Jinja2 syntax
+ # ("{{ var }}"); by the time this mixin's validate() runs it has already
+ # been converted from a dict to a JSON string by validate() below
+ # (unlike JobLaunchSerializer/WorkflowJobLaunchSerializer's extra_vars,
+ # which stay parsed), so this exclusion is load-bearing here, not
+ # defensive (AAP-78694).
+ excluded_fields = frozenset({'extra_vars'})
+
name = serializers.CharField(default='Bulk Job Launch', max_length=512, write_only=True, required=False, allow_blank=True) # limited by max name of jobs
jobs = BulkJobNodeSerializer(
many=True,
@@ -5159,10 +5363,29 @@ class BulkJobLaunchSerializer(serializers.Serializer):
return objectified_jobs
-class NotificationTemplateSerializer(BaseSerializer):
+class NotificationTemplateSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['edit', 'delete', 'copy']
capabilities_prefetch = [{'copy': 'organization.add_notificationtemplate'}]
+ # messages is user-authored Jinja2 template text ("{{ job.id }}" etc.),
+ # already validated by its own sandboxed Jinja renderer in
+ # validate_messages() below -- a stronger, purpose-built check that
+ # Tier 2 would conflict with.
+ excluded_fields = frozenset({'messages'})
+
+ # notification_configuration's schema is notification_type-dependent, but
+ # unlike Credential.inputs (admin-definable custom types with arbitrary
+ # secret fields), notification_type is a fixed models.CharField(choices=...)
+ # -- every possible backend class ships in awx/main/notifications/ and is
+ # known at code-authoring time. This is the exhaustive union, across all
+ # backends, of every init_parameters key with type "password" (verified:
+ # none of these names are used for a non-secret value in any backend):
+ # email/irc/webhook password, slack/pagerduty token, twilio account_token,
+ # grafana grafana_key, awssns aws_secret_access_key/aws_session_token.
+ excluded_json_keys = MappingProxyType(
+ {'notification_configuration': frozenset({'password', 'token', 'account_token', 'grafana_key', 'aws_secret_access_key', 'aws_session_token'})}
+ )
+
class Meta:
model = NotificationTemplate
fields = ('*', 'organization', 'notification_type', 'notification_configuration', 'messages')
@@ -5334,7 +5557,11 @@ class NotificationTemplateSerializer(BaseSerializer):
password_fields_to_forward = []
error_list = []
if 'notification_configuration' not in attrs:
- return attrs
+ # Still chain into CleanTextMixin.validate() (AAP-78694) so a
+ # PATCH that omits notification_configuration (e.g. name/
+ # description only) doesn't skip Tier 1/2 validation entirely --
+ # only the config-schema checks below are skipped.
+ return super(NotificationTemplateSerializer, self).validate(attrs)
if self.context['view'].kwargs and isinstance(self.context['view'], NotificationTemplateDetail):
object_actual = self.context['view'].get_object()
else:
@@ -5425,7 +5652,7 @@ class NotificationSerializer(BaseSerializer):
return ret
-class LabelSerializer(BaseSerializer):
+class LabelSerializer(CleanTextMixin, BaseSerializer):
class Meta:
model = Label
fields = ('*', '-description', 'organization')
@@ -5505,7 +5732,12 @@ class SchedulePreviewSerializer(BaseSerializer):
return value
-class ScheduleSerializer(LaunchConfigurationBaseSerializer, SchedulePreviewSerializer):
+class ScheduleSerializer(PromptFieldCleanTextMixin, LaunchConfigurationBaseSerializer, SchedulePreviewSerializer):
+ # See WorkflowJobTemplateNodeSerializer.excluded_fields above -- same
+ # reasoning, and same requirement that this live directly on this class
+ # rather than on the shared LaunchConfigurationBaseSerializer base.
+ excluded_fields = frozenset({'extra_data'})
+
show_capabilities = ['edit', 'delete']
timezone = serializers.SerializerMethodField(
@@ -5634,9 +5866,15 @@ class ReceptorAddressSerializer(BaseSerializer):
return obj.get_full_address()
-class InstanceSerializer(BaseSerializer):
+class InstanceSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['edit']
+ # hostname's existing HostnameRegexValidator (see extra_kwargs below)
+ # explicitly accepts IPv6 addresses, which contain colons that Tier 1's
+ # allowlist rejects. Demote to Tier 2 so IPv6 execution-node hostnames
+ # keep working; the existing validator remains the real charset gate.
+ name_fields = DEFAULT_NAME_FIELDS - {'hostname'}
+
consumed_capacity = serializers.SerializerMethodField()
percent_capacity_remaining = serializers.SerializerMethodField()
jobs_running = serializers.IntegerField(help_text=_('Count of jobs in the running or waiting state that are targeted for this instance'), read_only=True)
@@ -5935,7 +6173,7 @@ class HostMetricSummaryMonthlySerializer(BaseSerializer):
fields = read_only_fields
-class InstanceGroupSerializer(BaseSerializer):
+class InstanceGroupSerializer(CleanTextMixin, BaseSerializer):
show_capabilities = ['edit', 'delete']
capacity = serializers.SerializerMethodField()
consumed_capacity = serializers.SerializerMethodField()
diff --git a/awx/main/tests/functional/api/test_credential.py b/awx/main/tests/functional/api/test_credential.py
index f948e7fac3..1aaead8d18 100644
--- a/awx/main/tests/functional/api/test_credential.py
+++ b/awx/main/tests/functional/api/test_credential.py
@@ -1358,3 +1358,96 @@ def test_external_credential_rbac_test_endpoint(post, alice, external_credential
external_credential.use_role.members.add(alice)
assert post(url, data, alice).status_code == 202
+
+
+#
+# CleanTextMixin on Credential.inputs (AAP-78694)
+#
+# Nested JSON strings always go through Tier 2. These tests only assert which
+# keys are skipped: credential_type.secret_fields are excluded per-request via
+# instance excluded_json_keys; everything else is still validated.
+#
+
+UNSAFE_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_credential_inputs_rejects_unsafe_non_secret_key(post, admin, credentialtype_ssh, enforce_clean_text):
+ assert 'username' not in credentialtype_ssh.secret_fields
+ response = post(
+ reverse('api:credential_list'),
+ {
+ 'name': 'ssh-cleantext-username',
+ 'credential_type': credentialtype_ssh.id,
+ 'user': admin.id,
+ 'inputs': {'username': UNSAFE_INPUT, 'password': UNSAFE_INPUT},
+ },
+ admin,
+ )
+ assert response.status_code == 400
+ assert 'username' in response.data['inputs']
+ assert 'password' not in response.data['inputs']
+
+
+@pytest.mark.django_db
+def test_credential_inputs_skips_secret_keys(post, admin, credentialtype_ssh, enforce_clean_text):
+ assert 'password' in credentialtype_ssh.secret_fields
+ response = post(
+ reverse('api:credential_list'),
+ {
+ 'name': 'ssh-cleantext-password',
+ 'credential_type': credentialtype_ssh.id,
+ 'user': admin.id,
+ 'inputs': {'username': 'ok-user', 'password': UNSAFE_INPUT},
+ },
+ admin,
+ )
+ assert response.status_code == 201
+
+
+@pytest.mark.django_db
+def test_credential_inputs_secret_fields_follow_custom_type_schema(post, admin, organization, enforce_clean_text):
+ credential_type = CredentialType(
+ kind='cloud',
+ name='CustomCleanTextType',
+ inputs={
+ 'fields': [
+ {'id': 'endpoint', 'label': 'Endpoint', 'type': 'string'},
+ {'id': 'token', 'label': 'Token', 'type': 'string', 'secret': True},
+ ]
+ },
+ )
+ credential_type.save()
+ assert credential_type.secret_fields == ['token']
+
+ rejected = post(
+ reverse('api:credential_list'),
+ {
+ 'name': 'custom-cleantext-endpoint',
+ 'organization': organization.pk,
+ 'credential_type': credential_type.pk,
+ 'inputs': {'endpoint': UNSAFE_INPUT, 'token': 'ok-token'},
+ },
+ admin,
+ )
+ assert rejected.status_code == 400
+ assert 'endpoint' in rejected.data['inputs']
+ assert 'token' not in rejected.data['inputs']
+
+ accepted = post(
+ reverse('api:credential_list'),
+ {
+ 'name': 'custom-cleantext-token',
+ 'organization': organization.pk,
+ 'credential_type': credential_type.pk,
+ 'inputs': {'endpoint': 'https://example.invalid', 'token': UNSAFE_INPUT},
+ },
+ admin,
+ )
+ assert accepted.status_code == 201
diff --git a/awx/main/tests/functional/api/test_inventory.py b/awx/main/tests/functional/api/test_inventory.py
index f37af494ea..d77f1a9f8c 100644
--- a/awx/main/tests/functional/api/test_inventory.py
+++ b/awx/main/tests/functional/api/test_inventory.py
@@ -64,6 +64,44 @@ def test_inventory_host_name_unique(scm_inventory, post, admin_user):
assert "A Group with that name already exists." in json.dumps(resp.data)
+@pytest.mark.django_db
+def test_inventory_host_inline_port(scm_inventory, post, admin_user):
+ with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True):
+ resp = post(
+ reverse('api:inventory_hosts_list', kwargs={'pk': scm_inventory.id}),
+ {'name': 'web.example.com:2222'},
+ admin_user,
+ expect=201,
+ )
+ assert resp.data['name'] == 'web.example.com'
+ assert '2222' in resp.data['variables']
+
+
+@pytest.mark.django_db
+def test_inventory_host_ipv6_name(scm_inventory, post, admin_user):
+ """HostSerializer demotes name from Tier 1 so IPv6 (multiple colons)
+ is not rejected once enforcement is on (AAP-78694)."""
+ with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True):
+ resp = post(
+ reverse('api:inventory_hosts_list', kwargs={'pk': scm_inventory.id}),
+ {'name': '2001:db8::1'},
+ admin_user,
+ expect=201,
+ )
+ assert resp.data['name'] == '2001:db8::1'
+
+
+@pytest.mark.django_db
+def test_inventory_host_invalid_inline_port(scm_inventory, post, admin_user):
+ resp = post(
+ reverse('api:inventory_hosts_list', kwargs={'pk': scm_inventory.id}),
+ {'name': 'web.example.com:99999'},
+ admin_user,
+ expect=400,
+ )
+ assert 'Invalid port' in json.dumps(resp.data)
+
+
@pytest.mark.django_db
def test_inventory_host_list_ordering(scm_inventory, get, admin_user):
# create 3 hosts, hit the inventory host list view 3 times and get the order visible there each time and compare
diff --git a/awx/main/tests/functional/test_bulk.py b/awx/main/tests/functional/test_bulk.py
index 65eda300e1..a75a06a798 100644
--- a/awx/main/tests/functional/test_bulk.py
+++ b/awx/main/tests/functional/test_bulk.py
@@ -1,6 +1,7 @@
import pytest
from uuid import uuid4
+from unittest import mock
from awx.api.versioning import reverse
@@ -629,3 +630,44 @@ def test_bulk_host_create_performance_large_inventory(organization, inventory, p
assert len(response.data['hosts']) == 10
assert Host.objects.filter(inventory=inventory).count() == 10010
+
+
+@pytest.mark.django_db
+def test_bulk_job_launch_rejects_unsafe_limit(organization, inventory, project, post, user):
+ """PromptFieldCleanTextMixin._run_clean_text_validation must raise when
+ enforcement is on -- BulkJobNodeSerializer skips the normal validate()
+ chain that would otherwise do this (AAP-78694)."""
+ normal_user = user('normal_user', False)
+ organization.member_role.members.add(normal_user)
+ jt = JobTemplate.objects.create(name='my-jt', inventory=inventory, project=project, playbook='helloworld.yml')
+ jt.execute_role.members.add(normal_user)
+ with mock.patch('awx.api.serializers.get_setting', return_value=True):
+ response = post(
+ reverse('api:bulk_job_launch'),
+ {'name': 'Bulk Job Launch', 'jobs': [{'unified_job_template': jt.id, 'limit': ''}]},
+ normal_user,
+ expect=400,
+ )
+ assert 'limit' in str(response.data)
+
+
+@pytest.mark.django_db
+def test_bulk_job_launch_allows_jinja_extra_vars(organization, inventory, project, post, user):
+ """BulkJobLaunchSerializer stringifies extra_vars before CleanTextMixin
+ runs; excluded_fields must keep legitimate Jinja from being rejected
+ (AAP-78694)."""
+ normal_user = user('normal_user', False)
+ organization.member_role.members.add(normal_user)
+ jt = JobTemplate.objects.create(name='my-jt', inventory=inventory, project=project, playbook='helloworld.yml')
+ jt.execute_role.members.add(normal_user)
+ with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True):
+ post(
+ reverse('api:bulk_job_launch'),
+ {
+ 'name': 'Bulk Job Launch',
+ 'jobs': [{'unified_job_template': jt.id}],
+ 'extra_vars': {'foo': '{{ bar }}'},
+ },
+ normal_user,
+ expect=201,
+ )
diff --git a/awx/main/tests/functional/test_copy.py b/awx/main/tests/functional/test_copy.py
index 7c156979a8..adb4aead4c 100644
--- a/awx/main/tests/functional/test_copy.py
+++ b/awx/main/tests/functional/test_copy.py
@@ -271,3 +271,30 @@ def test_notification_template_copy(post, get, notification_template_with_encryp
assert decrypt_field(notification_template_with_encrypt, 'notification_configuration', 'token') == decrypt_field(
notification_template_copy, 'notification_configuration', 'token'
)
+
+
+@pytest.mark.django_db
+def test_job_template_copy_rejects_same_name(post, job_template, admin):
+ response = post(
+ reverse('api:job_template_copy', kwargs={'pk': job_template.pk}),
+ {'name': job_template.name},
+ admin,
+ expect=400,
+ )
+ assert 'already named' in str(response.data)
+
+
+@pytest.mark.django_db
+def test_job_template_copy_rejects_unsafe_name(post, job_template, admin):
+ """CopySerializer.super().validate() must run CleanTextMixin — the same-name
+ check above never reaches it (AAP-78694)."""
+ unsafe_name = ''
+ assert job_template.name != unsafe_name
+ with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True):
+ response = post(
+ reverse('api:job_template_copy', kwargs={'pk': job_template.pk}),
+ {'name': unsafe_name},
+ admin,
+ expect=400,
+ )
+ assert 'name' in response.data
diff --git a/awx/main/tests/functional/test_notifications.py b/awx/main/tests/functional/test_notifications.py
index b66ddc319d..0ee99e03ee 100644
--- a/awx/main/tests/functional/test_notifications.py
+++ b/awx/main/tests/functional/test_notifications.py
@@ -111,6 +111,20 @@ def test_notification_template_simple_patch(patch, notification_template, admin)
patch(reverse('api:notification_template_detail', kwargs={'pk': notification_template.id}), {'name': 'foo'}, admin, expect=200)
+@pytest.mark.django_db
+def test_notification_template_patch_rejects_unsafe_name(patch, notification_template, admin):
+ """Name-only PATCH omits notification_configuration; that branch must still
+ chain into CleanTextMixin.validate() (AAP-78694)."""
+ with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True):
+ response = patch(
+ reverse('api:notification_template_detail', kwargs={'pk': notification_template.id}),
+ {'name': ''},
+ admin,
+ expect=400,
+ )
+ assert 'name' in response.data
+
+
@pytest.mark.django_db
def test_notification_template_invalid_notification_type(patch, notification_template, admin):
patch(reverse('api:notification_template_detail', kwargs={'pk': notification_template.id}), {'notification_type': 'invalid'}, admin, expect=400)
diff --git a/awx/main/tests/unit/api/serializers/test_cleantext_mixin.py b/awx/main/tests/unit/api/serializers/test_cleantext_mixin.py
new file mode 100644
index 0000000000..aa3ede8491
--- /dev/null
+++ b/awx/main/tests/unit/api/serializers/test_cleantext_mixin.py
@@ -0,0 +1,116 @@
+from unittest import mock
+
+import pytest
+from rest_framework.exceptions import PermissionDenied, ValidationError
+
+from ansible_base.lib.serializers.mixins import CleanTextMixin
+
+from awx.api.serializers import (
+ BulkJobNodeSerializer,
+ CopySerializer,
+ CredentialSerializer,
+ ExecutionEnvironmentSerializer,
+ HostSerializer,
+)
+
+
+UNSAFE_INPUT = ''
+
+
+def _passthrough_validate(self, attrs):
+ return attrs
+
+
+class TestCopySerializerValidate:
+ def test_rejects_copy_with_the_same_name(self):
+ original = mock.Mock()
+ original.name = 'keep-me'
+ view = mock.Mock()
+ view.get_object.return_value = original
+ serializer = CopySerializer(context={'view': view})
+
+ with pytest.raises(ValidationError) as exc:
+ serializer.validate({'name': 'keep-me'})
+
+ assert 'already named' in str(exc.value.detail)
+
+
+@pytest.mark.django_db
+class TestHostSerializerPortParsing:
+ def test_extracts_inline_ssh_port(self):
+ serializer = HostSerializer()
+ assert serializer._get_host_port_from_name('web.example.com:2222') == ('web.example.com', 2222)
+
+ @pytest.mark.parametrize('name', ['web.example.com:0', 'web.example.com:65536', 'web.example.com:notaport'])
+ def test_rejects_invalid_inline_port(self, name):
+ serializer = HostSerializer()
+ with pytest.raises(ValidationError):
+ serializer._get_host_port_from_name(name)
+
+ def test_validate_moves_port_into_ansible_ssh_port(self):
+ serializer = HostSerializer()
+ with mock.patch.object(CleanTextMixin, 'validate', _passthrough_validate):
+ attrs = serializer.validate({'name': 'web.example.com:2222'})
+
+ assert attrs['name'] == 'web.example.com'
+ assert '"ansible_ssh_port": 2222' in attrs['variables']
+
+
+@pytest.mark.django_db
+class TestCredentialSerializerValidate:
+ def test_rejects_updates_to_managed_credentials(self):
+ serializer = CredentialSerializer()
+ serializer.instance = mock.Mock(managed=True)
+
+ with pytest.raises(PermissionDenied):
+ serializer.validate({'name': 'nope'})
+
+ def test_fail_closed_excludes_inputs_when_type_is_unknown(self):
+ serializer = CredentialSerializer()
+ with mock.patch.object(CleanTextMixin, 'validate', _passthrough_validate):
+ serializer.validate({'name': 'no-type', 'inputs': {'username': UNSAFE_INPUT}})
+
+ assert 'inputs' in serializer.excluded_fields
+
+
+@pytest.mark.django_db
+class TestBulkJobNodeCleanText:
+ def test_run_clean_text_validation_raises_when_enforced(self):
+ serializer = BulkJobNodeSerializer()
+ with mock.patch('awx.api.serializers.get_setting', return_value=True):
+ with pytest.raises(ValidationError) as exc:
+ serializer._run_clean_text_validation({'limit': UNSAFE_INPUT})
+
+ assert 'limit' in exc.value.detail
+
+
+@pytest.mark.django_db
+class TestExecutionEnvironmentSerializerValidate:
+ def test_rejects_non_registry_credentials(self):
+ serializer = ExecutionEnvironmentSerializer()
+ credential = mock.Mock(kind='ssh')
+ with pytest.raises(ValidationError):
+ serializer.validate_credential(credential)
+
+ def test_accepts_registry_credentials(self):
+ credential = mock.Mock(kind='registry')
+ assert ExecutionEnvironmentSerializer().validate_credential(credential) is credential
+
+ def test_rejects_organization_change(self):
+ serializer = ExecutionEnvironmentSerializer()
+ serializer.instance = mock.Mock(organization_id=1)
+ organization = mock.Mock(pk=2)
+
+ with pytest.raises(ValidationError) as exc:
+ serializer.validate({'organization': organization})
+
+ assert 'organization' in exc.value.detail
+
+ def test_allows_same_organization(self):
+ serializer = ExecutionEnvironmentSerializer()
+ serializer.instance = mock.Mock(organization_id=1)
+ org = mock.Mock(pk=1)
+ with mock.patch.object(CleanTextMixin, 'validate', _passthrough_validate):
+ attrs = serializer.validate({'organization': org})
+
+ assert attrs['organization'] is org
diff --git a/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py b/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py
index 22376c5d72..a6a6c44e15 100644
--- a/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py
+++ b/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py
@@ -30,6 +30,14 @@ class TestNotificationTemplateSerializer:
{'started': {'body': '{{ job_metadata }}'}},
{'started': {'body': '{{ job.summary_fields.inventory.total_hosts }}'}},
{'started': {'body': 'Iñtërnâtiônàlizætiøn'}},
+ {
+ 'workflow_approval': {
+ 'running': {'message': 'running'},
+ 'approved': None,
+ 'timed_out': {'message': 'timed out'},
+ 'denied': {'body': 'denied'},
+ }
+ },
],
)
def test_valid_messages(self, valid_messages):
@@ -55,6 +63,8 @@ class TestNotificationTemplateSerializer:
{'started': {'message': '{{ job.id | bad_filter }}'}},
{'started': {'message': '{{ job.__class__ }}'}},
{'started': {'message': 'Newlines \n not allowed\n'}},
+ {'workflow_approval': {'invalid_subevent': {}}},
+ {'workflow_approval': {'running': 'should_be_dict'}},
],
)
def test_invalid__messages(self, invalid_messages):
diff --git a/awx/main/tests/unit/api/serializers/test_uncovered_paths.py b/awx/main/tests/unit/api/serializers/test_uncovered_paths.py
new file mode 100644
index 0000000000..0542c5fa31
--- /dev/null
+++ b/awx/main/tests/unit/api/serializers/test_uncovered_paths.py
@@ -0,0 +1,351 @@
+from unittest import mock
+
+import pytest
+from rest_framework.exceptions import ValidationError
+
+from awx.api.serializers import (
+ ActivityStreamSerializer,
+ AdHocCommandRelaunchSerializer,
+ AnsibleFactsSerializer,
+ GroupTreeSerializer,
+ HostSerializer,
+ JobCreateScheduleSerializer,
+ JobHostSummarySerializer,
+ JobRelaunchSerializer,
+ NotificationSerializer,
+ OpaQueryPathMixin,
+ ProjectUpdateEventSerializer,
+ UnifiedJobSerializer,
+ UnifiedJobStdoutSerializer,
+ WorkflowApprovalSerializer,
+ WorkflowJobLaunchSerializer,
+ reverse_gfk,
+)
+from awx.main.constants import ACTIVE_STATES
+from awx.main.models import JobLaunchConfig, WorkflowApproval, WorkflowJobTemplate
+
+
+def _mock(**kwargs):
+ """Build a Mock with a real `.name` attribute.
+
+ unittest.mock.Mock(name=...) only sets the mock's repr name, so
+ getattr(obj, 'name') would return another Mock instead of the string.
+ """
+ name = kwargs.pop('name', None)
+ obj = mock.Mock(**kwargs)
+ if name is not None:
+ obj.name = name
+ return obj
+
+
+def test_reverse_gfk_empty_when_object_missing():
+ assert reverse_gfk(None, None) == {}
+ assert reverse_gfk(object(), None) == {}
+
+
+def test_reverse_gfk_uses_absolute_url():
+ class Organization:
+ def get_absolute_url(self, request=None):
+ return '/api/v2/organizations/1/'
+
+ assert reverse_gfk(Organization(), None) == {'organization': '/api/v2/organizations/1/'}
+
+
+def test_opa_query_path_rejects_unencoded_url():
+ serializer = OpaQueryPathMixin()
+ with pytest.raises(ValidationError):
+ serializer.validate_opa_query_path('not a valid path')
+
+
+def test_unified_job_stdout_types():
+ assert UnifiedJobStdoutSerializer().get_types() == [
+ 'project_update',
+ 'inventory_update',
+ 'job',
+ 'ad_hoc_command',
+ 'system_job',
+ ]
+
+
+def test_unified_job_launched_by():
+ serializer = UnifiedJobSerializer()
+ assert serializer.get_launched_by(None) is None
+ obj = mock.Mock(launched_by={'id': 7})
+ assert serializer.get_launched_by(obj) == {'id': 7}
+
+
+def test_unified_job_sub_serializer_workflow_approval():
+ obj = mock.MagicMock(spec=WorkflowApproval)
+ assert UnifiedJobSerializer().get_sub_serializer(obj) is WorkflowApprovalSerializer
+
+
+def test_ansible_facts_to_representation():
+ obj = mock.Mock(ansible_facts={'os': 'rhel'})
+ assert AnsibleFactsSerializer().to_representation(obj) == {'os': 'rhel'}
+
+
+def test_group_tree_children_none():
+ assert GroupTreeSerializer().get_children(None) == {}
+
+
+def test_adhoc_relaunch_to_representation():
+ serializer = AdHocCommandRelaunchSerializer()
+ assert serializer.to_representation(None) == {}
+ obj = mock.Mock(passwords_needed_to_start=['ssh_password'])
+ assert serializer.to_representation(obj) == {'ssh_password': ''}
+
+
+class TestHostToRepresentation:
+ def test_related_includes_last_job_links(self):
+ serializer = HostSerializer()
+ serializer.reverse = mock.Mock(return_value='/x/')
+ obj = mock.Mock(pk=1, instance_id=9)
+ obj.inventory.kind = 'constructed'
+ obj.inventory.pk = 2
+ obj.latest_summary = mock.Mock(pk=3, job_id=4)
+ with mock.patch('awx.api.serializers.BaseSerializer.get_related', return_value={}):
+ related = serializer.get_related(obj)
+ assert 'last_job_host_summary' in related
+ assert 'last_job' in related
+
+ def test_empty_obj_returns_parent_payload(self):
+ serializer = HostSerializer()
+ with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={'inventory': 1}):
+ assert serializer.to_representation(None) == {'inventory': 1}
+
+ def test_nulls_missing_inventory(self):
+ serializer = HostSerializer()
+ obj = mock.Mock(inventory=None)
+ with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={'inventory': 9}):
+ assert serializer.to_representation(obj)['inventory'] is None
+
+
+@pytest.mark.django_db
+class TestJobRelaunchSerializerPaths:
+ def test_missing_credential_passwords(self):
+ serializer = JobRelaunchSerializer()
+ serializer.instance = mock.Mock(passwords_needed_to_start=['ssh_password', 'become_password'])
+ with pytest.raises(ValidationError):
+ serializer.validate_credential_passwords({'ssh_password': ''})
+
+ def test_to_representation_injects_password_keys(self):
+ view = mock.Mock(_raw_data_form_marker=True)
+ serializer = JobRelaunchSerializer(context={'view': view})
+ obj = mock.Mock(passwords_needed_to_start=['ssh_password'])
+ with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={}):
+ assert serializer.to_representation(obj)['ssh_password'] == ''
+
+ def test_passwords_needed_to_start(self):
+ serializer = JobRelaunchSerializer()
+ assert serializer.get_passwords_needed_to_start(None) == ''
+ obj = mock.Mock(passwords_needed_to_start=['ssh_password'])
+ assert serializer.get_passwords_needed_to_start(obj) == ['ssh_password']
+
+ def test_retry_counts_while_running(self):
+ serializer = JobRelaunchSerializer()
+ obj = mock.Mock(status=next(iter(ACTIVE_STATES)))
+ assert 'not available' in str(serializer.get_retry_counts(obj))
+
+ def test_retry_counts_when_finished(self):
+ serializer = JobRelaunchSerializer()
+ obj = mock.Mock(status='successful')
+ obj.retry_qs.return_value.count.return_value = 2
+ counts = serializer.get_retry_counts(obj)
+ assert counts['all'] == 2
+ assert counts['failed'] == 2
+
+ def test_validate_missing_project(self):
+ serializer = JobRelaunchSerializer()
+ serializer.instance = mock.Mock(project=None, inventory=mock.Mock(pending_deletion=False))
+ with pytest.raises(ValidationError):
+ serializer.validate({})
+
+ def test_validate_missing_inventory(self):
+ serializer = JobRelaunchSerializer()
+ serializer.instance = mock.Mock(project=mock.Mock(), inventory=None)
+ with pytest.raises(ValidationError):
+ serializer.validate({})
+
+ def test_validate_inventory_pending_deletion(self):
+ serializer = JobRelaunchSerializer()
+ serializer.instance = mock.Mock(project=mock.Mock(), inventory=mock.Mock(pending_deletion=True))
+ with pytest.raises(ValidationError):
+ serializer.validate({})
+
+
+class TestJobCreateScheduleSerializerPaths:
+ def test_can_schedule(self):
+ obj = mock.Mock(can_schedule=True)
+ assert JobCreateScheduleSerializer().get_can_schedule(obj) is True
+
+ def test_summarize_copies_fk_fields(self):
+ obj = _mock(id=3, name='inv', description='d')
+ summary = JobCreateScheduleSerializer._summarize('host', obj)
+ assert summary['id'] == 3
+ assert summary['name'] == 'inv'
+
+ def test_get_prompts_summarizes_related_objects(self):
+ inventory = _mock(id=1, name='inv', description='', has_active_failures=False)
+ ee = _mock(id=2, name='ee', description='', image='img')
+ cred = _mock(id=3, name='cred', description='', kind='ssh', cloud=False, kubernetes=False, credential_type_id=1)
+ ig = _mock(id=4, name='ig', is_container_group=False)
+ config = mock.Mock()
+ config.prompts_dict.return_value = {
+ 'inventory': inventory,
+ 'execution_environment': ee,
+ 'credentials': [cred],
+ 'instance_groups': [ig],
+ 'labels': True,
+ }
+ job = mock.Mock(launch_config=config)
+ serializer = JobCreateScheduleSerializer()
+ serializer._summary_field_labels = mock.Mock(return_value=[{'id': 9, 'name': 'l'}])
+ prompts = serializer.get_prompts(job)
+ assert prompts['inventory']['id'] == 1
+ assert prompts['execution_environment']['id'] == 2
+ assert prompts['credentials'][0]['id'] == 3
+ assert prompts['instance_groups'][0]['id'] == 4
+ assert prompts['labels'] == [{'id': 9, 'name': 'l'}]
+
+ def test_get_prompts_without_launch_config(self):
+ class JobWithoutConfig:
+ @property
+ def launch_config(self):
+ raise JobLaunchConfig.DoesNotExist()
+
+ prompts = JobCreateScheduleSerializer().get_prompts(JobWithoutConfig())
+ assert 'all' in prompts
+
+
+class TestNotificationSerializerBody:
+ def test_webhook_dict_body(self):
+ obj = mock.Mock(notification_type='webhook', body={'body': 'hello'})
+ assert NotificationSerializer().get_body(obj) == 'hello'
+
+ def test_webhook_json_string_body(self):
+ obj = mock.Mock(notification_type='pagerduty', body='{"ok": true}')
+ assert NotificationSerializer().get_body(obj) == {'ok': True}
+
+ def test_webhook_invalid_json_string_body(self):
+ obj = mock.Mock(notification_type='awssns', body='not-json')
+ assert NotificationSerializer().get_body(obj) == 'not-json'
+
+ def test_plain_body_passthrough(self):
+ obj = mock.Mock(notification_type='email', body='text')
+ assert NotificationSerializer().get_body(obj) == 'text'
+
+
+class TestJobHostSummarySerializerPaths:
+ def test_get_related_includes_host(self):
+ serializer = JobHostSummarySerializer()
+ serializer.reverse = mock.Mock(side_effect=lambda *a, **k: '/x/')
+ obj = mock.Mock(job=mock.Mock(pk=1), host=mock.Mock(pk=2))
+ with mock.patch('awx.api.serializers.BaseSerializer.get_related', return_value={}):
+ related = serializer.get_related(obj)
+ assert 'job' in related
+ assert 'host' in related
+
+ def test_summary_fields_adds_job_template(self):
+ serializer = JobHostSummarySerializer()
+ obj = mock.Mock()
+ obj.job.job_template.id = 11
+ obj.job.job_template.name = 'jt'
+ with mock.patch('awx.api.serializers.BaseSerializer.get_summary_fields', return_value={'job': {}}):
+ summary = serializer.get_summary_fields(obj)
+ assert summary['job']['job_template_id'] == 11
+ assert summary['job']['job_template_name'] == 'jt'
+
+ def test_summary_fields_tolerates_missing_job(self):
+ serializer = JobHostSummarySerializer()
+ obj = mock.Mock()
+ with mock.patch('awx.api.serializers.BaseSerializer.get_summary_fields', return_value={}):
+ assert serializer.get_summary_fields(obj) == {}
+
+
+class TestWorkflowJobLaunchSerializerPaths:
+ def test_survey_enabled(self):
+ serializer = WorkflowJobLaunchSerializer()
+ assert serializer.get_survey_enabled(None) is False
+ obj = mock.Mock(survey_enabled=True, survey_spec={'spec': []})
+ assert serializer.get_survey_enabled(obj) is True
+
+ def test_workflow_job_template_data(self):
+ obj = _mock(name='wf', id=5, description='d')
+ assert WorkflowJobLaunchSerializer().get_workflow_job_template_data(obj) == {'name': 'wf', 'id': 5, 'description': 'd'}
+
+ def test_defaults_inventory_and_labels(self):
+ mapping = {'inventory': None, 'labels': None, 'limit': None}
+ obj = mock.Mock()
+ obj.inventory.name = 'inv'
+ obj.inventory.pk = 8
+ obj.limit = 'webservers'
+ label = _mock(id=1, name='prod')
+ obj.labels.all.return_value = [label]
+ with mock.patch.object(WorkflowJobTemplate, 'get_ask_mapping', return_value=mapping):
+ defaults = WorkflowJobLaunchSerializer().get_defaults(obj)
+ assert defaults['inventory'] == {'name': 'inv', 'id': 8}
+ assert defaults['labels'] == [{'id': 1, 'name': 'prod'}]
+ assert defaults['limit'] == 'webservers'
+
+
+class TestActivityStreamSerializerPaths:
+ def test_get_changes(self):
+ serializer = ActivityStreamSerializer()
+ assert serializer.get_changes(None) == {}
+ assert serializer.get_changes(mock.Mock(changes='{"a": 1}')) == {'a': 1}
+ assert serializer.get_changes(mock.Mock(changes='not-json')) == {}
+
+ def test_object_association(self):
+ serializer = ActivityStreamSerializer()
+ assert serializer.get_object_association(mock.Mock(object_relationship_type='')) == ''
+ assert serializer.get_object_association(mock.Mock(object_relationship_type='awx.main.models.inventory.Inventory.admin_role')) == 'role'
+ rel = 'awx.main.models.organization.Organization_notification_templates_success'
+ assert 'notification' in serializer.get_object_association(mock.Mock(object_relationship_type=rel))
+
+ def test_object_type_role(self):
+ serializer = ActivityStreamSerializer()
+ assert serializer.get_object_type(mock.Mock(object_relationship_type='')) == ''
+ rel = 'awx.main.models.inventory.Inventory.admin_role'
+ assert serializer.get_object_type(mock.Mock(object_relationship_type=rel)) == 'inventory'
+
+
+class TestProjectUpdateEventSanitization:
+ def test_sanitizes_git_event_data(self):
+ serializer = ProjectUpdateEventSerializer()
+ obj = mock.Mock(event_data={'task_action': 'git', 'msg': 'ok'})
+ assert serializer.get_event_data(obj)['task_action'] == 'git'
+
+ def test_returns_empty_on_sanitize_failure(self):
+ serializer = ProjectUpdateEventSerializer()
+ obj = mock.Mock(event_data={'task_action': 'svn'})
+ with mock.patch('awx.api.serializers.json.dumps', side_effect=ValueError('boom')):
+ assert serializer.get_event_data(obj) == {}
+
+ def test_passthrough_for_other_actions(self):
+ serializer = ProjectUpdateEventSerializer()
+ obj = mock.Mock(event_data={'task_action': 'debug', 'msg': 'hi'})
+ assert serializer.get_event_data(obj) == obj.event_data
+
+
+class TestUnifiedJobSummaryFields:
+ def test_spawned_by_workflow_copies_job_summary(self):
+ serializer = UnifiedJobSerializer()
+ workflow_job = _mock(id=10, name='wf', description='', status='successful', failed=False, elapsed=1.5, type='workflow_job', canceled_on=None)
+ obj = mock.Mock(spawned_by_workflow=True, ancestor_job=None)
+ obj.unified_job_node.workflow_job = workflow_job
+ with mock.patch('awx.api.serializers.BaseSerializer.get_summary_fields', return_value={}):
+ summary = serializer.get_summary_fields(obj)
+ assert summary['source_workflow_job']['id'] == 10
+ assert summary['source_workflow_job']['name'] == 'wf'
+
+
+def test_unified_job_elapsed_for_running_job():
+ serializer = UnifiedJobSerializer()
+ started = mock.Mock()
+ td = mock.Mock(microseconds=0, seconds=5, days=0)
+ obj = mock.Mock(pk=1, started=started, finished=None, job_explanation='')
+ with mock.patch.object(UnifiedJobSerializer, 'get_sub_serializer', return_value=None):
+ with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={'elapsed': 0, 'job_explanation': ''}):
+ with mock.patch('awx.api.serializers.now', return_value=mock.Mock(__sub__=lambda self, other: td)):
+ ret = serializer.to_representation(obj)
+ assert ret['elapsed'] == 5.0