Fix validate_argument_spec so the argument spec is validated before

passing it to the ArgumentSpecValidator to prevent unhandled errors
during variable validation.

Make role argument spec errors non-fatal at runtime. Display a warning
and include details with -vvv. Pass ArgumentSpecValidator a pruned
argument spec containing the valid portion.

Give an error and return argument_spec_errors in the task result for
non-implicit validate_argument_spec tasks.

Add tests for role argument spec errors and validate_argument_spec
argument_spec errors.
pull/79657/head
s-hertel 1 year ago
parent 48be6f8b6f
commit 4ba445b748

@ -0,0 +1,6 @@
bugfixes:
- >
roles - give a warning for unsupported fields in the role argument spec during implicit ``validate_argument_spec`` tasks.
The valid portion of the argument spec is used to validate variables to prevent unhandled errors during validation.
(https://github.com/ansible/ansible/issues/79624, https://github.com/ansible/ansible/issues/84103)
- validate_argument_spec - give an error for unsupported fields in the argument spec to prevent unhandled errors during validation.

@ -94,6 +94,14 @@ argument_errors:
- "error message 1" - "error message 1"
- "error message 2" - "error message 2"
argument_spec_errors:
description: A list of errors from the 'argument_spec' arg.
returned: failure
type: list
elements: str
sample:
- "option_str: options. Suboptions are supported for types dict and list."
argument_spec_data: argument_spec_data:
description: A dict of the data from the 'argument_spec' arg. description: A dict of the data from the 'argument_spec' arg.
returned: failure returned: failure

@ -269,11 +269,18 @@ class Role(Base, Conditional, Taggable, CollectionSearch, Delegatable):
if self._should_validate: if self._should_validate:
role_argspecs = self._get_role_argspecs() role_argspecs = self._get_role_argspecs()
task_data = self._prepend_validation_task(task_data, role_argspecs) validation_task_data = self._prepend_validation_task(None, role_argspecs)
validation_blocks = load_list_of_blocks(
validation_task_data, play=self._play, role=self, loader=self._loader, variable_manager=self._variable_manager
)
for block in validation_blocks:
for task in block.block:
task.implicit = True
self._task_blocks = validation_blocks
if task_data: if task_data:
try: try:
self._task_blocks = load_list_of_blocks(task_data, play=self._play, role=self, loader=self._loader, variable_manager=self._variable_manager) self._task_blocks += load_list_of_blocks(task_data, play=self._play, role=self, loader=self._loader, variable_manager=self._variable_manager)
except AssertionError as e: except AssertionError as e:
raise AnsibleParserError("The tasks/main.yml file for role '%s' must contain a list of tasks" % self._role_name, raise AnsibleParserError("The tasks/main.yml file for role '%s' must contain a list of tasks" % self._role_name,
obj=task_data, orig_exc=e) obj=task_data, orig_exc=e)

@ -6,8 +6,111 @@ from __future__ import annotations
from ansible.errors import AnsibleError from ansible.errors import AnsibleError
from ansible.plugins.action import ActionBase from ansible.plugins.action import ActionBase
from ansible.module_utils.common.arg_spec import ArgumentSpecValidator from ansible.module_utils.common.arg_spec import ArgumentSpecValidator
from ansible.utils.display import Display
from ansible.utils.vars import combine_vars from ansible.utils.vars import combine_vars
from contextlib import suppress
from copy import deepcopy
from dataclasses import asdict, field, dataclass, make_dataclass
import typing as t
OPTION_TYPES = [None, 'bits', 'bool', 'bytes', 'dict', 'float', 'int', 'json', 'jsonarg', 'list', 'path', 'raw', 'sid', 'str']
ROLE_OPTION_SPEC = dict(
description=dict(type='list', elements='str'),
type=dict(choices=OPTION_TYPES),
default=dict(type='raw'),
choices=dict(type='list'),
elements=dict(choices=OPTION_TYPES),
options=dict(type='dict'),
required=dict(type='bool'),
version_added=dict(type='raw'),
# some module argument spec features are intentionally omitted since role argument specs are read-only
# no_log, aliases, apply_defaults, and fallback are not supported
)
OPTION_SPEC = dict(ROLE_OPTION_SPEC)
OPTION_SPEC.update(dict(
no_log=dict(type='bool'),
aliases=dict(type='list', elements='str'),
apply_defaults=dict(type='bool'),
# fallback is excluded because tuple[callable, list[str]] can't be set via a task
))
display = Display()
@dataclass
class RoleSpec:
"""Argument spec supported by the ArgumentSpecValidator and roles."""
@classmethod
def from_spec(cls: t.Type[t.Self], fields: dict, role_name: str, spec_name: str, role_path: str) -> t.Self:
"""Load a valid role argument spec from a foreign argument spec."""
fields = deepcopy(fields)
errors: list[str] = []
validator = ArgumentSpecValidator(ROLE_OPTION_SPEC, required_if=[('type', 'list', ('elements',))])
validate_argspec(validator, [], fields, errors, prune=True)
if errors:
display.warning(f"Role '{role_name}' entrypoint '{spec_name}' contains errors in the argument spec. Use -vvv to see details.")
display.vvv(f"Role '{role_name}' ({role_path}) argument spec '{spec_name}' contains errors:")
for error in errors:
display.vvv(f"- {error}")
class_fields = []
for key, value in fields.items():
default = field(default_factory=lambda: value) if is_mutable(value) else field(default=value) # pylint: disable=invalid-field-call
class_fields.append((key, type(value), default))
return make_dataclass(cls.__name__, class_fields)(**fields)
def validate_argspec(validator: ArgumentSpecValidator, context: list[str], argument_spec: dict, errors: list[str], prune: bool = False) -> None:
"""Validate the options in an argument spec and optionally prune those with errors."""
invalid_options = []
for option, fields in argument_spec.items():
if not isinstance(option, str):
invalid_options.append(option)
continue
validate_option_spec(validator, context + [option], fields, errors, prune)
for option in invalid_options:
str_option = f"{option}"
errors.append(f"{'.'.join(context + [str_option])}: type {type(option)}. Option names must be strings.")
if prune:
argument_spec.pop(option)
def validate_option_spec(validator: ArgumentSpecValidator, context: list[str], fields: dict, errors: list[str], prune: bool = False) -> None:
"""Validate the fields for an option and optionally prune those with errors."""
result = validator.validate(fields)
if result.error_messages:
errors.append(f"{'.'.join(context)}: {', '.join(result.error_messages)}")
if prune and result.error_messages:
for error in result.error_messages:
field = error.split('.', 1)[0]
fields.pop(field, None)
if fields.get('type') in ('list', 'dict') and fields.get('options'):
validate_argspec(validator, context, fields['options'], errors, prune)
elif fields.get('options'):
errors.append(f"{'.'.join(context)}: options. Suboptions are supported for types dict and list.")
if prune:
fields.pop('options')
def is_mutable(value: t.Any) -> bool:
"""Return true is a value is not hashable."""
with suppress(TypeError):
hash(value)
return False
return True
class ActionModule(ActionBase): class ActionModule(ActionBase):
""" Validate an arg spec""" """ Validate an arg spec"""
@ -75,8 +178,30 @@ class ActionModule(ActionBase):
if not isinstance(provided_arguments, dict): if not isinstance(provided_arguments, dict):
raise AnsibleError('Incorrect type for provided_arguments, expected dict and got %s' % type(provided_arguments)) raise AnsibleError('Incorrect type for provided_arguments, expected dict and got %s' % type(provided_arguments))
if self._task.implicit:
argument_spec = RoleSpec.from_spec(
argument_spec_data,
result['validate_args_context']['name'],
result['validate_args_context']['argument_spec_name'],
result['validate_args_context']['path']
)
valid_argument_spec = asdict(argument_spec)
else:
errors = []
argspec_option_validator = ArgumentSpecValidator(OPTION_SPEC, required_if=[('type', 'list', ('elements',))])
validate_argspec(argspec_option_validator, [], argument_spec_data, errors)
if errors:
result['failed'] = True
result['msg'] = "Invalid argument_spec cannot be used to validate variables."
result['argument_spec_errors'] = errors
result['argument_spec_data'] = argument_spec_data
return result
valid_argument_spec = argument_spec_data
args_from_vars = self.get_args_from_task_vars(argument_spec_data, task_vars) args_from_vars = self.get_args_from_task_vars(argument_spec_data, task_vars)
validator = ArgumentSpecValidator(argument_spec_data) validator = ArgumentSpecValidator(valid_argument_spec)
validation_result = validator.validate(combine_vars(args_from_vars, provided_arguments), validate_role_argument_spec=True) validation_result = validator.validate(combine_vars(args_from_vars, provided_arguments), validate_role_argument_spec=True)
if validation_result.error_messages: if validation_result.error_messages:

@ -163,6 +163,8 @@ class CallbackModule(CallbackBase):
if not os.path.exists(self._output_dir): if not os.path.exists(self._output_dir):
os.makedirs(self._output_dir) os.makedirs(self._output_dir)
self.wants_implicit_tasks = True
def _start_task(self, task): def _start_task(self, task):
""" record the start of a task for one or more hosts """ """ record the start of a task for one or more hosts """

@ -0,0 +1,67 @@
---
argument_specs:
no_log:
short_description: "Validate no_log in a role argument spec causes a warning."
description:
- "The validation for role argument specs is read-only and does not provide a way to mark a variable as no_log."
options:
password:
description: "Things not to do."
no_log: true
no_log_suboption:
short_description: "Validate no_log in a role argument spec causes a warning."
description:
- "The validation for role argument specs is read-only and does not provide a way to mark a subset of a variable as no_log."
options:
auth:
description: "A dictionary containing a no_log suboption."
type: dict
options:
password:
description: "Things not to do."
no_log: true
apply_defaults:
short_description: "Validate apply_defaults in a role argument spec causes a warning."
description:
- "The validation for role argument specs is read-only and does not provide a way to set a variable."
options:
option_with_suboptions:
description: "Things not to do."
type: dict
apply_defaults: true
options:
sub_option:
description: "A suboption of an option with apply_defaults."
default: "magic"
aliases:
short_description: "Validate aliases in a role argument spec causes a warning."
description:
- "The validation for role argument specs is read-only and does not provide a way to set a variable to an alias."
options:
option_name:
description: "Things not to do."
aliases:
- option_alias
fallback:
short_description: "Validate fallback in a role argument spec causes a warning."
description:
- "The validation for role argument specs is read-only and does not provide a way to set a variable to an environment variable."
options:
option_with_env_fallback:
description: "Things not to do."
fallback:
- ENV_VAR
unknown_field:
short_description: "Validate an unknown field in a role argument spec causes a warning."
description:
- "Help detect typos."
options:
option_name:
description: "Things not to do."
unknown: "yes"
choice: ["typo"]

@ -30,3 +30,38 @@ set -e
# The task is tagged with 'foo' but we use 'bar' in the call below and expect # The task is tagged with 'foo' but we use 'bar' in the call below and expect
# the validation task to run anyway since it is tagged 'always'. # the validation task to run anyway since it is tagged 'always'.
ansible-playbook test_tags.yml -i ../../inventory "$@" --tags bar | grep "a : Validating arguments against arg spec 'main' - Main entry point for role A." ansible-playbook test_tags.yml -i ../../inventory "$@" --tags bar | grep "a : Validating arguments against arg spec 'main' - Main entry point for role A."
# Test unknown and unsupported role spec fields emit a warning
invalid_specs=("no_log" "no_log_suboption" "aliases" "apply_defaults" "fallback" "unknown_field")
for spec in "${invalid_specs[@]}"; do
echo "Testing invalid role argument spec: containing $spec"
# test warning requires no verbosity
ansible localhost -m include_role -a "name=invalid_specs tasks_from=$spec" 2> error
grep "\[WARNING\]: Role 'invalid_specs' entrypoint '${spec}'" error
# test details are included in stdout with at least -vvv
ansible localhost -m include_role -a "name=invalid_specs tasks_from=$spec" -vvv "$@" > out
grep "Role 'invalid_specs' (.*) argument spec '${spec}' contains errors:" out
if [[ "$spec" = "no_log" ]]; then
option_and_field='password: no_log'
elif [[ "$spec" = "no_log_suboption" ]]; then
option_and_field='auth.password: no_log'
elif [[ "$spec" = "aliases" ]]; then
option_and_field="option_name: $spec"
elif [[ "$spec" = "apply_defaults" ]]; then
option_and_field="option_with_suboptions: $spec"
elif [[ "$spec" = "fallback" ]]; then
option_and_field="option_with_env_fallback: $spec"
elif [[ "$spec" = "unknown_field" ]]; then
option_and_field="option_name: choice, unknown"
fi
grep -e "- ${option_and_field}\. Supported parameters include: choices, default, description, elements, options, required, type, version_added\." out
done
# Test a valid spec doesn't cause a warning
ansible-playbook test_tags.yml -i ../../inventory "$@" --tags bar 2> error
if [ -s error ]; then
grep -v "\[WARNING\]: Role 'a' entrypoint 'main'" error
fi

@ -0,0 +1,59 @@
- name: Validate task vars with an argument spec
validate_argument_spec:
argument_spec:
option_with_suboptions:
type: dict
options:
required_suboption:
type: str
required: true
vars:
option_with_suboptions:
required_suboption: specified
- name: Validate an argument spec containing invalid suboptions
validate_argument_spec:
argument_spec:
invalid_suboptions:
description: Valid top level option containing non-string options.
type: dict
options:
true:
description: Invalid suboption name.
invalid_option:
description: Not a dictionary or list.
type: str
options:
suboption:
type: str
ignore_errors: true
register: validation_result
- assert:
that:
- validation_result is failed
- validation_result.msg == "Invalid argument_spec cannot be used to validate variables."
- validation_result.argument_spec_errors|length == 2
- validation_result.argument_spec_errors == [error1, error2]
vars:
error1: "invalid_suboptions.True: type <class 'bool'>. Option names must be strings."
error2: "invalid_option: options. Suboptions are supported for types dict and list."
- name: Validate an argument spec containing an unsupported field for an option
validate_argument_spec:
argument_spec:
option1:
fallback:
- env_fallback # ArgumentSpecValidator expects a callable, so fallback is unsupported for validate_argument_spec
- [ENV_VAR]
ignore_errors: true
register: validation_result
- assert:
that:
- validation_result is failed
- validation_result.msg == "Invalid argument_spec cannot be used to validate variables."
- validation_result.argument_spec_errors|length == 1
- validation_result.argument_spec_errors == ["option1: fallback. " ~ valid_params]
vars:
valid_params: "Supported parameters include: aliases, apply_defaults, choices, default, description, elements, no_log, options, required, type, version_added."
Loading…
Cancel
Save