Add feature preview play argument spec validation (#85763)

* Add new play keyword validate_argspec

Set to True to use the play name as the argument spec identifier. A play name is required (i.e. a host pattern is not supported as an argument spec name).

Alternatively, set to a specific argument spec name.

A valid argument spec for the play is required. Example:

# playbook.meta.yml
argument_specs:
  name of the play:
    options: {}

* Play argument spec validation runs after fact gathering

Play keywords like tags are inherited and work similarly to fact gathering

changelogs/fragments/play-argument-spec-validation.yml

@@ -0,0 +1,11 @@
+minor_changes:
+- >-
+  Add tech preview play argument spec validation, which can be
+  enabled by setting the play keyword ``validate_argspec`` to ``True``
+  or the name of an argument spec.
+  When ``validate_argspec`` is set to ``True``, a play ``name`` is
+  required and used as the argument spec name.
+  When enabled, the argument spec is loaded from a file matching the
+  pattern <playbook_name>.meta.yml.
+  At minimum, this file should contain ``{"argument_specs": {"name": {"options": {}}}}``,
+  where "name" is the name of the play or configured argument spec.

lib/ansible/executor/play_iterator.py

@@ -174,6 +174,27 @@ class PlayIterator:
             setup_task.when = self._play._included_conditional[:]
         setup_block.block = [setup_task]
 
+        validation_task = Task.load({
+            'name': f'Validating arguments against arg spec {self._play.validate_argspec}',
+            'action': 'ansible.builtin.validate_argument_spec',
+            'args': {
+                # 'provided_arguments': {},  # allow configuration via module_defaults
+                'argument_spec': self._play.argument_spec,
+                'validate_args_context': {
+                    'type': 'play',
+                    'name': self._play.validate_argspec,
+                    'argument_spec_name': self._play.validate_argspec,
+                    'path': self._play._metadata_path,
+                },
+            },
+            'tags': ['always'],
+        }, block=setup_block)
+
+        validation_task.set_loader(self._play._loader)
+        if self._play._included_conditional is not None:
+            validation_task.when = self._play._included_conditional[:]
+        setup_block.block.append(validation_task)
+
         setup_block = setup_block.filter_tagged_tasks(all_vars)
         self._blocks.append(setup_block)
 
@@ -271,35 +292,36 @@ class PlayIterator:
                 return (state, None)
 
             if state.run_state == IteratingStates.SETUP:
-                # First, we check to see if we were pending setup. If not, this is
-                # the first trip through IteratingStates.SETUP, so we set the pending_setup
-                # flag and try to determine if we do in fact want to gather facts for
-                # the specified host.
-                if not state.pending_setup:
-                    state.pending_setup = True
+                # First, we check to see if we completed both setup tasks injected
+                # during play compilation in __init__ above.
+                # If not, below we will determine if we do in fact want to gather
+                # facts or validate arguments for the specified host.
+                state.pending_setup = state.cur_regular_task < len(block.block)
+                if state.pending_setup:
+                    task = block.block[state.cur_regular_task]
 
                     # Gather facts if the default is 'smart' and we have not yet
                     # done it for this host; or if 'explicit' and the play sets
                     # gather_facts to True; or if 'implicit' and the play does
                     # NOT explicitly set gather_facts to False.
-
+                    gather_facts = bool(state.cur_regular_task == 0)
                     gathering = C.DEFAULT_GATHERING
                     implied = self._play.gather_facts is None or boolean(self._play.gather_facts, strict=False)
 
-                    if (gathering == 'implicit' and implied) or \
-                       (gathering == 'explicit' and boolean(self._play.gather_facts, strict=False)) or \
-                       (gathering == 'smart' and implied and not self._variable_manager._facts_gathered_for_host(host.name)):
-                        # The setup block is always self._blocks[0], as we inject it
-                        # during the play compilation in __init__ above.
-                        setup_block = self._blocks[0]
-                        if setup_block.has_tasks() and len(setup_block.block) > 0:
-                            task = setup_block.block[0]
-                else:
-                    # This is the second trip through IteratingStates.SETUP, so we clear
-                    # the flag and move onto the next block in the list while setting
-                    # the run state to IteratingStates.TASKS
-                    state.pending_setup = False
+                    if gather_facts and not (
+                        (gathering == 'implicit' and implied) or
+                        (gathering == 'explicit' and boolean(self._play.gather_facts, strict=False)) or
+                        (gathering == 'smart' and implied and not self._variable_manager._facts_gathered_for_host(host.name))
+                    ):
+                        task = None
+                    elif not gather_facts and not self._play.validate_argspec:
+                        task = None
 
+                    state.cur_regular_task += 1
+                else:
+                    # This is the last trip through IteratingStates.SETUP, so we
+                    # move onto the next block in the list while setting the run
+                    # state to IteratingStates.TASKS
                     state.run_state = IteratingStates.TASKS
                     if not state.did_start_at_task:
                         state.cur_block += 1

lib/ansible/playbook/play.py

@@ -17,11 +17,15 @@
 
 from __future__ import annotations
 
+import functools as _functools
+import pathlib as _pathlib
+
 from ansible import constants as C
 from ansible import context
 from ansible.errors import AnsibleError
-from ansible.errors import AnsibleParserError, AnsibleAssertionError
+from ansible.errors import AnsibleParserError, AnsibleAssertionError, AnsibleValueOmittedError
 from ansible.module_utils.common.collections import is_sequence
+from ansible.module_utils.common.yaml import yaml_dump
 from ansible.playbook.attribute import NonInheritableFieldAttribute
 from ansible.playbook.base import Base
 from ansible.playbook.block import Block
@@ -33,6 +37,8 @@ from ansible.playbook.taggable import Taggable
 from ansible.parsing.vault import EncryptedString
 from ansible.utils.display import Display
 
+from ansible._internal._templating._engine import TemplateEngine as _TE
+
 display = Display()
 
 
@@ -64,6 +70,8 @@ class Play(Base, Taggable, CollectionSearch):
     vars_files = NonInheritableFieldAttribute(isa='list', default=list, priority=99)
     vars_prompt = NonInheritableFieldAttribute(isa='list', default=list, always_post_validate=False)
 
+    validate_argspec = NonInheritableFieldAttribute(isa='string', always_post_validate=True)
+
     # Role Attributes
     roles = NonInheritableFieldAttribute(isa='list', default=list, priority=90)
 
@@ -407,3 +415,92 @@ class Play(Base, Taggable, CollectionSearch):
         new_me._action_groups = self._action_groups
         new_me._group_actions = self._group_actions
         return new_me
+
+    def _post_validate_validate_argspec(self, attr: NonInheritableFieldAttribute, value: object, templar: _TE) -> str | None:
+        """Validate user input is a bool or string, and return the corresponding argument spec name."""
+
+        # Ensure the configuration is valid
+        if isinstance(value, str):
+            try:
+                value = templar.template(value)
+            except AnsibleValueOmittedError:
+                value = False
+
+        if not isinstance(value, (str, bool)):
+            raise AnsibleParserError(f"validate_argspec must be a boolean or string, not {type(value)}", obj=value)
+
+        # Short-circuit if configuration is turned off or inapplicable
+        if not value or self._origin is None:
+            return None
+
+        # Use the requested argument spec or fall back to the play name
+        argspec_name = None
+        if isinstance(value, str):
+            argspec_name = value
+        elif self._ds.get("name"):
+            argspec_name = self.name
+
+        metadata_err = argspec_err = ""
+        if not argspec_name:
+            argspec_err = (
+                "A play name is required when validate_argspec is True. "
+                "Alternatively, set validate_argspec to the name of an argument spec."
+            )
+        if self._metadata_path is None:
+            metadata_err = "A playbook meta file is required. Considered:\n  - "
+            metadata_err += "\n  - ".join([path.as_posix() for path in self._metadata_candidate_paths])
+
+        if metadata_err or argspec_err:
+            error = f"{argspec_err + (' ' if argspec_err else '')}{metadata_err}"
+            raise AnsibleParserError(error, obj=self._origin)
+
+        metadata = self._loader.load_from_file(self._metadata_path)
+
+        try:
+            metadata = metadata['argument_specs']
+            metadata = metadata[argspec_name]
+            options = metadata['options']
+        except (TypeError, KeyError):
+            options = None
+
+        if not isinstance(options, dict):
+            raise AnsibleParserError(
+                f"No argument spec named '{argspec_name}' in {self._metadata_path}. Minimally expected:\n"
+                + yaml_dump({"argument_specs": {f"{argspec_name!s}": {"options": {}}}}),
+                obj=metadata,
+            )
+
+        return argspec_name
+
+    @property
+    def _metadata_candidate_paths(self) -> list[_pathlib.Path]:
+        """A list of possible playbook.meta paths in configured order."""
+        extensions = C.config.get_config_value("YAML_FILENAME_EXTENSIONS")
+        if self._origin.path.endswith(tuple(extensions)):
+            playbook_without_ext = self._origin.path.rsplit('.', 1)[0]
+        else:
+            playbook_without_ext = self._origin.path
+
+        return [_pathlib.Path(playbook_without_ext + ".meta" + ext) for ext in extensions + ['']]
+
+    @_functools.cached_property
+    def _metadata_path(self) -> str | None:
+        """Locate playbook meta path:
+
+        playbook{ext?} -> playbook.meta{ext?}
+        """
+        if self._origin is None:
+            # adhoc, ansible-console don't have an associated playbook
+            return None
+        for candidate in self._metadata_candidate_paths:
+            if candidate.is_file():
+                return candidate.as_posix()
+        return None
+
+    @property
+    def argument_spec(self) -> dict:
+        """Retrieve the argument spec if one is configured."""
+        if not self.validate_argspec:
+            return {}
+
+        return self._loader.load_from_file(self._metadata_path)['argument_specs'][self.validate_argspec]['options']

test/integration/targets/play_arg_spec/aliases

@@ -0,0 +1,2 @@
+shippable/posix/group3
+context/controller

test/integration/targets/play_arg_spec/defaults/main.yml

@@ -0,0 +1,4 @@
+playbook: "{{ role_path }}/playbooks/{{ playbook_name }}.yml"
+playbook_meta: "{{ role_path }}/playbooks/{{ playbook_name }}.meta.yml"
+playbook_name: play_argspec
+play_name: Test play spec

test/integration/targets/play_arg_spec/library/custom_facts.py

@@ -0,0 +1,4 @@
+#!/usr/bin/python
+from __future__ import annotations
+import json
+print(json.dumps({"ansible_facts": {"custom_fact": "required"}}))

test/integration/targets/play_arg_spec/playbooks/missing_metadata.yml

@@ -0,0 +1,3 @@
+- name: Test play argspec
+  hosts: localhost
+  validate_argspec: True

test/integration/targets/play_arg_spec/playbooks/module_defaults.meta.yml

@@ -0,0 +1,5 @@
+argument_specs:
+  Test validate_argspec and module_defaults interaction:
+    options:
+      valid_argument:
+        type: str

test/integration/targets/play_arg_spec/playbooks/module_defaults.yml

@@ -0,0 +1,15 @@
+- name: "Test validate_argspec and module_defaults interaction"
+  hosts: localhost
+  module_defaults:
+    validate_argument_spec:
+      provided_arguments: |
+          {% set defined_vars = {} %}
+          {% set ignore = ['group_names', 'groups', 'inventory_hostname', 'inventory_hostname_short', 'playbook_dir'] %}
+          {% for key, value in hostvars[inventory_hostname].items() %}
+          {% if value is defined and not (key is search("^ansible_.*$") or key in ignore) %}
+          {% set _ = defined_vars.update({key: value}) %}
+          {% endif %}
+          {% endfor %}
+          {{ defined_vars }}
+  gather_facts: no
+  validate_argspec: True

test/integration/targets/play_arg_spec/playbooks/no_play_name.yml

@@ -0,0 +1,2 @@
+- hosts: localhost
+  validate_argspec: True

test/integration/targets/play_arg_spec/playbooks/play_argspec.meta.yml

@@ -0,0 +1,11 @@
+argument_specs:
+  Test play spec:
+    options:
+      required_str:
+        required: True
+        type: str
+      optional:
+        type: str
+        choices:
+          - option1
+          - option2

test/integration/targets/play_arg_spec/playbooks/play_argspec.yml

@@ -0,0 +1,6 @@
+- name: "{{ play_name | default(_play_name) }}"
+  vars:
+    _play_name: "{{ play_name_from_vars }}"
+  hosts: localhost
+  gather_facts: no
+  validate_argspec: True

test/integration/targets/play_arg_spec/playbooks/tagged_play.meta.yml

@@ -0,0 +1,6 @@
+argument_specs:
+  Tagged Play:
+    options:
+      required_str:
+        required: True
+        type: str

test/integration/targets/play_arg_spec/playbooks/tagged_play.yml

@@ -0,0 +1,5 @@
+- name: Tagged Play
+  hosts: localhost
+  tags:
+    - play_level_tag
+  validate_argspec: True

test/integration/targets/play_arg_spec/playbooks/test_keyword.meta.yml

@@ -0,0 +1,3 @@
+argument_specs:
+  Argument Spec Name:
+    options: {}

test/integration/targets/play_arg_spec/playbooks/test_keyword.yml

@@ -0,0 +1,3 @@
+- hosts: localhost
+  gather_facts: no
+  validate_argspec: "{{ test_variable }}"

test/integration/targets/play_arg_spec/playbooks/test_keyword_bool_no.yml

@@ -0,0 +1,3 @@
+- hosts: localhost
+  gather_facts: no
+  validate_argspec: no

test/integration/targets/play_arg_spec/playbooks/test_keyword_str_no.meta.yml

@@ -0,0 +1,3 @@
+argument_specs:
+  "no":
+    options: {}

test/integration/targets/play_arg_spec/playbooks/test_keyword_str_no.yml

@@ -0,0 +1,3 @@
+- hosts: localhost
+  gather_facts: no
+  validate_argspec: "no"

test/integration/targets/play_arg_spec/playbooks/validate_facts.meta.yml

@@ -0,0 +1,6 @@
+argument_specs:
+  Gather Facts:
+    options:
+      custom_fact:
+        required: True
+        type: str

test/integration/targets/play_arg_spec/playbooks/validate_facts.yml

@@ -0,0 +1,7 @@
+- name: Gather Facts
+  hosts: localhost
+  gather_facts: True
+  validate_argspec: True
+  vars:
+    ansible_facts_modules:
+      - custom_facts

test/integration/targets/play_arg_spec/tasks/main.yml

@@ -0,0 +1,197 @@
+- name: Test missing playbook meta file
+  command: ansible-playbook {{ playbook }}
+  vars:
+    playbook_name: missing_metadata
+  register: result
+  ignore_errors: True
+
+- assert:
+    that:
+      - result is failed
+      - result.stderr is search("A playbook meta file is required. Considered:")
+      - result.stderr is search(playbook_meta)
+  vars:
+    playbook_name: missing_metadata
+
+- name: Test configuring validate_argspec as no (False)
+  command: ansible-playbook {{ playbook }}
+  vars:
+    playbook_name: test_keyword_bool_no
+  register: result
+
+- assert:
+    that: "{{ result.stdout is not search('Validating arguments against arg spec') }}"
+
+- name: Test configuring validate_argspec as "no" (string)
+  command: ansible-playbook {{ playbook }}
+  vars:
+    playbook_name: test_keyword_str_no
+  register: result
+
+- assert:
+    that: "{{ result.stdout is search('Validating arguments against arg spec no') }}"
+
+- name: Test configuring argument spec validation
+  vars:
+    playbook_name: test_keyword
+  block:
+    - name: Test configuring argument spec validation
+      command: ansible-playbook {{ playbook }} -e 'test_variable="{{ spec_name }}"'
+      vars:
+        spec_name: !unsafe '{{ "Argument Spec Name" }}'
+      register: result
+
+    - assert:
+        that: "{{ result.stdout is search('Validating arguments against arg spec Argument Spec Name') }}"
+
+    - name: Test turning off argument spec validation
+      command: ansible-playbook {{ playbook }} -e 'test_variable="{{ bool_var }}"'
+      vars:
+        bool_var: !unsafe '{{ False }}'
+      register: result
+
+    - assert:
+        that: "{{ result.stdout is not search('Validating arguments against arg spec') }}"
+
+    - name: Test undefined configuration
+      command: ansible-playbook {{ playbook }} -e 'test_variable="{{ undef_var }}"'
+      vars:
+        undef_var: !unsafe "{{ inventory_hostname != 'localhost' }}"
+      ignore_errors: True
+      register: result
+
+    - assert:
+        that:
+          - result is failed
+          - >-
+            result.stderr is search("Error processing keyword 'validate_argspec': .* is undefined")
+
+    - name: Test omitted configuration
+      command: ansible-playbook {{ playbook }} -e 'test_variable="{{ omitted }}"'
+      vars:
+        omitted: !unsafe "{{ omit }}"
+      register: result
+
+    - assert:
+        that: "{{ result.stdout is not search('Validating arguments against arg spec') }}"
+
+    - name: Test configuring unknown argument spec name
+      command: ansible-playbook {{ playbook }} -e 'test_variable="No argument spec"'
+      ignore_errors: True
+      register: result
+
+    - assert:
+        that:
+          - result is failed
+          - result.stderr is search("No argument spec named 'No argument spec' in " ~ playbook_meta)
+
+- name: Validate missing required argument
+  command: ansible-playbook {{ playbook }} -e 'play_name="{{ play_name }}"'
+  register: result
+  ignore_errors: True
+
+- assert:
+    that:
+      - result is failed
+      - 'result.stdout is search("missing required arguments: required_str")'
+
+- name: Validate incorrect argument choice
+  command: ansible-playbook {{ playbook }} -e 'play_name="{{ play_name }}" required_str="" optional="fail"'
+  register: result
+  ignore_errors: True
+
+- assert:
+    that:
+      - result is failed
+      - 'result.stdout is search("value of optional must be one of: option1, option2, got: fail")'
+
+# Test compatibility with play level fact gathering
+
+- name: Test validating play level fact gathering
+  command: ansible-playbook {{ playbook }}
+  vars:
+    playbook_name: validate_facts
+  environment:
+    ANSIBLE_LIBRARY: "{{ role_path }}/library"
+  register: result
+
+- assert:
+    that: "{{ result.stdout is search('Validating arguments against arg spec Gather Facts') }}"
+
+# Test compatibility with the play name keyword
+
+- name: Test validating a vars variable play name argument spec
+  command: ansible-playbook {{ playbook }} -e 'play_name_from_vars="{{ play_name }}" required_str=""'
+  register: result
+
+- assert:
+    that: "{{ result.stdout is search('Validating arguments against arg spec ' ~ play_name) }}"
+
+- name: Test a play name is required
+  command: ansible-playbook {{ playbook }} -e 'play_name="{{ undef_var }}"'
+  register: result
+  ignore_errors: True
+  loop_control:
+    loop_var: undef_var
+  loop:
+    - !unsafe "{{ inventory_hostname }}"
+    - !unsafe "{{ omit }}"
+    - ""
+
+- assert:
+    that:
+      - result.results[0] is failed
+      - >-
+        result.results[0].stderr is search("Error processing keyword 'name': .* is undefined")
+      - result.results[1] is failed
+      - result.results[1].stderr is search("A play name is required when validate_argspec is True.")
+      - result.results[2] is failed
+      - result.results[2].stderr is search("A play name is required when validate_argspec is True.")
+
+- name: Test host pattern is not used as an argument spec name
+  command: ansible-playbook {{ playbook }}
+  vars:
+    playbook_name: no_play_name
+  register: result
+  ignore_errors: True
+
+- assert:
+    that:
+      - result is failed
+      - result.stderr is search(err)
+  vars:
+    err: >-
+      A play name is required when validate_argspec is True.
+      Alternatively, set validate_argspec to the name of an argument spec.
+      A playbook meta file is required. Considered:
+
+# Test compatibility with the play module_defaults keyword
+
+- name: Test using module_defaults to validate arbitrary variables
+  command: ansible-playbook {{ playbook }} -e 'nodoc="fail"'
+  vars:
+    playbook_name: module_defaults
+  register: result
+  ignore_errors: True
+
+- assert:
+    that:
+      - result is failed
+      - 'result.stdout is search("nodoc. Supported parameters include: valid_argument.")'
+
+# Test compatibility with the play tags keyword
+
+- name: Test skipping the whole play, including argspec validation
+  command: ansible-playbook {{ playbook }} --skip-tags play_level_tag
+  vars:
+    playbook_name: tagged_play
+
+- name: Test validation always runs otherwise
+  command: ansible-playbook {{ playbook }} --tags task_level_tag -e 'required_str="success"'
+  vars:
+    playbook_name: tagged_play
+  register: result
+
+- assert:
+    that:
+      - result.stdout is search("Validating arguments against arg spec Tagged Play")