mirror of
https://github.com/ansible/awx.git
synced 2026-10-29 22:03:20 -02:30
[AAP-78694] Add CleanTextMixin to Controller Serializers (#16635)
* AAP-78694 - Add CleanTextMixin to AWX resource serializers Applies django-ansible-base's CleanTextMixin to the AWX serializers listed in AAP-78694: Organization, User, Team, Credential(+Create), CredentialType, Project, JobTemplate, WorkflowJobTemplate, Inventory, Host, Group, InventorySource, InstanceGroup, Instance, ExecutionEnvironment, NotificationTemplate, Schedule, AdHocCommand, WorkflowJobTemplateNode, WorkflowJobTemplateNodeCreateApproval, JobLaunch, and WorkflowJobLaunch. Adds a local PromptFieldCleanTextMixin for serializers whose model derives from LaunchTimeConfigBase, where scm_branch/limit/job_tags/ skip_tags are NullablePromptPseudoField descriptors invisible to the mixin's normal model-field auto-discovery. Per-serializer excluded_fields/name_fields overrides are documented inline where a field's existing behavior would otherwise conflict with Tier 1/2 validation (Jinja2-templated fields, secret fields, and existing colon-containing name/hostname syntax). Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * AAP-78694 - Cover CleanTextMixin gaps in Copy, Bulk, and Job Detail paths Three write paths were found to bypass the Tier 1/2 validation applied elsewhere for the same resources: - CopySerializer (used by the Copy button/endpoint for EE, Project, Credential, JobTemplate, WorkflowJobTemplate, NotificationTemplate) accepted a user-supplied name with zero validation. Adds a new PlainSerializerCleanTextMixin for plain serializers.Serializer subclasses with no backing model. - BulkJobNodeSerializer's validate() intentionally skips LaunchConfigurationBaseSerializer.validate() (it assumes unified_job_template/inventory/execution_environment are model instances, but bulk nodes pass raw PKs to avoid a per-node DB lookup), which meant it also skipped CleanTextMixin. Adds PromptFieldCleanTextMixin._run_clean_text_validation() so the Tier 1/2 checks still run despite that skip. BulkJobLaunchSerializer and WorkflowJobNodeSerializer are also covered. - JobSerializer (backing the narrow JobDetail PUT/PATCH window while a job's status is "new") had no coverage at all. Confirmed live against a running controller that a Copy with a malicious name previously succeeded (201) where the equivalent PATCH on the same resource was already rejected (400); both are rejected consistently after this change. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * AAP-78694 - Fix NotificationTemplateSerializer validate() bypass validate() returned early without calling super().validate() whenever notification_configuration was absent from the request (e.g. a PATCH that only changes name/description), which skipped CleanTextMixin entirely on that path. Same bug class as the BulkJobNodeSerializer MRO-skip already fixed, caught by code review. Confirmed live: a PATCH with only a malicious name (no notification_configuration) previously succeeded; now rejected with 400, while a legitimate name-only PATCH still succeeds. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * AAP-78694 - Add CleanTextMixin to the previously out-of-scope API-only serializers LabelSerializer, CredentialInputSourceSerializer, and WorkflowApprovalTemplateSerializer were originally scoped out ("no UI exposure") but are now in scope per updated ticket guidance. WorkflowApprovalTemplateSerializer backs the update path for the same model WorkflowJobTemplateNodeCreateApprovalSerializer already protects on create -- without this, an approval name rejected at creation could be rewritten to anything via PATCH. Confirmed live: Label create and WorkflowApprovalTemplate update both reject a malicious name (400) and accept a valid one. CredentialInputSourceSerializer verified at the code level only (plain CharField input_field_name, no custom validators; metadata is a JSONSchemaField holding vault paths/keys, not secret values) -- setting up a live external-credential-type chain in this environment wasn't done. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * AAP-78694 - Correct HostSerializer name_fields demotion comment The comment claimed the Tier 1 -> Tier 2 demotion was needed for "hostname:port" syntax, but that colon is stripped from attrs['name'] in validate() before it reaches CleanTextMixin -- that case never needed the demotion. What actually needs it: names with any colon count other than exactly 1, which _get_host_port_from_name passes through untouched -- most notably IPv6 addresses, which its own comment says it explicitly does not parse. Caught by code review. Confirmed live: an IPv6-style host name (2 colons) is accepted unmodified (201). Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * AAP-78694 - Validate non-secret notification_configuration keys Replaced the blanket notification_configuration exclusion with a precise excluded_json_keys covering only the actual secret fields. Unlike Credential.inputs (admin-definable custom types with arbitrary secret fields), notification_type is a fixed models.CharField(choices) -- every backend class ships in awx/main/notifications/ and is known at code-authoring time, so the secret key set can be a static list rather than resolved per-instance. Enumerated every init_parameters entry with type "password" across all 8 backends (email/irc/webhook password, slack/pagerduty token, twilio account_token, grafana grafana_key, awssns aws_secret_access_key/aws_session_token) and confirmed none of those names are used for a non-secret value in a different backend. Previously-unvalidated, UI-exposed fields like webhook url and slack channels are now covered. Confirmed live: a javascript: webhook url and a <script>-tagged slack channel are now rejected (400), while a weird-looking token/password value is still accepted unchanged. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * AAP-78694 - Exclude extra_data on launch-config serializers extra_data (WorkflowJobTemplateNode/WorkflowJobNode/Schedule) plays the same role as extra_vars elsewhere and is submitted in the same two formats -- a dict, or a raw YAML/JSON string (parse_yaml_or_json handles both). Its model field is JSONBlob, whose get_internal_type() reports "TextField" for legacy DB-migration reasons, so CleanTextMixin silently skipped dict payloads (not a str) while string payloads hit Tier 2's template-injection blocklist and incorrectly rejected legitimate Jinja variable references -- a real regression once enforcement is on, caught by code review. First attempt put the exclusion on the shared LaunchConfigurationBaseSerializer base and silently did nothing: CleanTextMixin's own excluded_fields = frozenset() default sits earlier in the MRO than that base class for every subclass (mixin is first in the bases tuple), so the base class's value was always shadowed. Verified with a standalone MRO test before settling on setting excluded_fields directly on each leaf serializer (WorkflowJobTemplateNodeSerializer, WorkflowJobNodeSerializer, ScheduleSerializer), matching every other exclusion in this file. Confirmed live on both WorkflowJobTemplateNodeSerializer and ScheduleSerializer: a YAML-string extra_data with a Jinja reference is accepted (previously 400), while other Tier 1/2 fields on the same serializers are still rejected as expected. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * AAP-78694 - Validate non-secret Credential.inputs keys dynamically Unlike NotificationTemplate.notification_configuration (fixed set of backend classes, static excluded_json_keys works), CredentialType is admin-definable via a real "Create a new custom credential type" endpoint, so the set of secret field names in Credential.inputs can't be known at class-definition time -- it varies per credential's own type, via that type's own inputs schema ("secret": true flags, exposed as CredentialType.secret_fields). validate() resolves the credential_type from attrs (create) or the existing instance (update) and sets excluded_json_keys as an instance attribute before chaining into CleanTextMixin.validate(). Setting an instance attribute rather than a class attribute sidesteps the MRO trap found while fixing extra_data on launch-config serializers: CleanTextMixin's own excluded_fields/excluded_json_keys class-level defaults sit earlier in the MRO than any shared base class further down the chain, silently shadowing it -- an instance attribute always wins regardless of MRO position, and correctly threads through CredentialSerializerCreate and its User/Team/Organization subclasses for free, since each of their validate() overrides chains through super() into this one. Falls back to excluding the whole inputs blob (prior, safe behavior) if credential_type can't be resolved. Confirmed live: malicious username (non-secret, including via PATCH with no credential_type in the payload) is rejected; a dangerous- looking password value and a real unencrypted SSH private key (secret fields) are both still accepted unchanged. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * Increase test coverage * Cover remaining serializer branches for Sonar gate Line coverage is already over 90%; Sonar's quality gate uses combined line+branch coverage. These unit tests exercise previously untested methods so the PR can clear the 90% new-code threshold. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> * style(api): format uncovered serializer tests for ruff Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> Assisted-by: Cursor Grok 4.6 (SpaceXAI / Cursor) Co-authored-by: Cursor <cursoragent@cursor.com> * test(api): set Mock.name as an attribute, not a constructor kwarg unittest.mock.Mock(name=...) only sets the mock's repr name, so getattr(obj, 'name') returned another Mock and four coverage tests failed in api-test. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> Assisted-by: Cursor Grok 4.6 (SpaceXAI / Cursor) Co-authored-by: Cursor <cursoragent@cursor.com> * test(api): keep a single call inside pytest.raises blocks Sonar python:S5918 flags exception tests that construct mocks or serializers inside the raises context. Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> Assisted-by: Cursor Grok 4.6 (SpaceXAI / Cursor) Co-authored-by: Cursor <cursoragent@cursor.com> * Add more test coverage --------- Signed-off-by: Pratyush Bhandari <prbhanda@redhat.com> Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Pratyush Bhandari
co-authored by
Cursor
Cursor
parent
fcd7003510
commit
9fdfc95d37
+269
-31
@@ -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()
|
||||
|
||||
@@ -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 = '<script>x</script>'
|
||||
|
||||
|
||||
@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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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': '<script>x</script>'}]},
|
||||
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,
|
||||
)
|
||||
|
||||
@@ -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 = '<script>x</script>'
|
||||
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
|
||||
|
||||
@@ -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': '<script>x</script>'},
|
||||
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)
|
||||
|
||||
@@ -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 = '<script>x</script>'
|
||||
|
||||
|
||||
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
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user