Adding upstream version 6.13.1.upstream/6.13.1upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 280 insertions, 0 deletions

diff --git a/src/ansiblelint/rules/args.py b/src/ansiblelint/rules/args.py
new file mode 100644
index 0000000..0ff532e
--- /dev/null
+++ b/src/ansiblelint/rules/args.py
@@ -0,0 +1,280 @@
+"""Rule definition to validate task options."""
+from __future__ import annotations
+
+import contextlib
+import importlib.util
+import io
+import json
+import logging
+import re
+import sys
+from functools import lru_cache
+from typing import Any
+
+# pylint: disable=preferred-module
+from unittest import mock
+from unittest.mock import patch
+
+# pylint: disable=reimported
+import ansible.module_utils.basic as mock_ansible_module
+from ansible.module_utils import basic
+from ansible.plugins import loader
+
+from ansiblelint.constants import LINE_NUMBER_KEY
+from ansiblelint.errors import MatchError
+from ansiblelint.file_utils import Lintable
+from ansiblelint.rules import AnsibleLintRule, RulesCollection
+from ansiblelint.text import has_jinja
+from ansiblelint.yaml_utils import clean_json
+
+_logger = logging.getLogger(__name__)
+
+
+@lru_cache
+def load_module(module_name: str) -> loader.PluginLoadContext:
+    """Load plugin from module name and cache it."""
+    return loader.module_loader.find_plugin_with_context(module_name)
+
+
+class ValidationPassed(Exception):
+    """Exception to be raised when validation passes."""
+
+
+class CustomAnsibleModule(basic.AnsibleModule):  # type: ignore
+    """Mock AnsibleModule class."""
+
+    def __init__(self, *args: str, **kwargs: str) -> None:
+        """Initialize AnsibleModule mock."""
+        super().__init__(*args, **kwargs)
+        raise ValidationPassed
+
+
+class ArgsRule(AnsibleLintRule):
+    """Validating module arguments."""
+
+    id = "args"
+    severity = "HIGH"
+    description = "Check whether tasks are using correct module options."
+    tags = ["syntax", "experimental"]
+    version_added = "v6.10.0"
+    module_aliases: dict[str, str] = {"block/always/rescue": "block/always/rescue"}
+
+    def matchtask(
+        self, task: dict[str, Any], file: Lintable | None = None
+    ) -> list[MatchError]:
+        # pylint: disable=too-many-branches,too-many-locals
+        results: list[MatchError] = []
+        module_name = task["action"]["__ansible_module_original__"]
+        failed_msg = None
+
+        if module_name in self.module_aliases:
+            return []
+
+        loaded_module = load_module(module_name)
+        module_args = {
+            key: value
+            for key, value in task["action"].items()
+            if not key.startswith("__")
+        }
+        # https://github.com/ansible/ansible-lint/issues/2824
+        if loaded_module.resolved_fqcn == "ansible.builtin.async_status":
+            module_args["_async_dir"] = "/tmp/ansible-async"
+        if loaded_module.resolved_fqcn == "ansible.builtin.service":
+            _logger.debug(
+                "Skipped service module validation as not being supported yet."
+            )
+            return []
+
+        with mock.patch.object(
+            mock_ansible_module, "AnsibleModule", CustomAnsibleModule
+        ):
+            spec = importlib.util.spec_from_file_location(
+                name=loaded_module.resolved_fqcn,
+                location=loaded_module.plugin_resolved_path,
+            )
+            if spec:
+                assert spec.loader is not None
+                module = importlib.util.module_from_spec(spec)
+                spec.loader.exec_module(module)
+            else:
+                assert file is not None
+                _logger.warning(
+                    "Unable to load module %s at %s:%s for options validation",
+                    module_name,
+                    file.filename,
+                    task[LINE_NUMBER_KEY],
+                )
+                return []
+
+            try:
+                if not hasattr(module, "main"):
+                    # skip validation for module options that are implemented as action plugin
+                    # as the option values can be changed in action plugin and are not passed
+                    # through `ArgumentSpecValidator` class as in case of modules.
+                    return []
+
+                with patch.object(
+                    sys,
+                    "argv",
+                    ["", json.dumps({"ANSIBLE_MODULE_ARGS": clean_json(module_args)})],
+                ):
+                    fio = io.StringIO()
+                    failed_msg = ""
+                    # Warning: avoid running anything while stdout is redirected
+                    # as what happens may be very hard to debug.
+                    with contextlib.redirect_stdout(fio):
+                        # pylint: disable=protected-access
+                        basic._ANSIBLE_ARGS = None
+                        try:
+                            module.main()
+                        except SystemExit:
+                            failed_msg = fio.getvalue()
+                    if failed_msg:
+                        results.extend(
+                            self._parse_failed_msg(failed_msg, task, module_name, file)
+                        )
+
+                sanitized_results = self._sanitize_results(results, module_name)
+                return sanitized_results
+            except ValidationPassed:
+                return []
+
+    def _sanitize_results(
+        self, results: list[MatchError], module_name: str
+    ) -> list[MatchError]:
+        """Remove results that are false positive."""
+        sanitized_results = []
+        for result in results:
+            result_msg = result.message
+            if result_msg.startswith("Unsupported parameters"):
+                # cmd option is a special case in command module and after option validation is done.
+                if (
+                    "Unsupported parameters for (basic.py) module" in result_msg
+                    and module_name
+                    in ["command", "ansible.builtin.command", "ansible.legacy.command"]
+                ):
+                    continue
+                result.message = result_msg.replace("(basic.py)", f"{module_name}")
+            elif result_msg.startswith("missing required arguments"):
+                if (
+                    "missing required arguments: free_form" in result_msg
+                    and module_name
+                    in [
+                        "raw",
+                        "ansible.builtin.raw",
+                        "ansible.legacy.raw",
+                        "meta",
+                        "ansible.builtin.meta",
+                        "ansible.legacy.meta",
+                    ]
+                ):
+                    # free_form option is a special case in raw module hence ignore this error.
+                    continue
+                if (
+                    "missing required arguments: key_value" in result_msg
+                    and module_name
+                    in [
+                        "set_fact",
+                        "ansible.builtin.set_fact",
+                        "ansible.legacy.set_fact",
+                    ]
+                ):
+                    # handle special case for set_fact module with key and value
+                    continue
+            if "Supported parameters include" in result_msg and module_name in [
+                "set_fact",
+                "ansible.builtin.set_fact",
+                "ansible.legacy.set_fact",
+            ]:
+                continue
+            sanitized_results.append(result)
+
+        return sanitized_results
+
+    def _parse_failed_msg(
+        self,
+        failed_msg: str,
+        task: dict[str, Any],
+        module_name: str,
+        file: Lintable | None = None,
+    ) -> list[MatchError]:
+        """Parse failed message and return list of MatchError."""
+        results: list[MatchError] = []
+        try:
+            failed_obj = json.loads(failed_msg)
+            error_message = failed_obj["msg"]
+        except json.decoder.JSONDecodeError:
+            error_message = failed_msg
+
+        option_type_check_error = re.search(
+            r"argument '(?P<name>.*)' is of type", error_message
+        )
+        if option_type_check_error:
+            # ignore options with templated variable value with type check errors
+            option_key = option_type_check_error.group("name")
+            option_value = task["action"][option_key]
+            if has_jinja(option_value):
+                _logger.debug(
+                    "Type checking ignored for '%s' option in task '%s' at line %s.",
+                    option_key,
+                    module_name,
+                    task[LINE_NUMBER_KEY],
+                )
+                return results
+
+        value_not_in_choices_error = re.search(
+            r"value of (?P<name>.*) must be one of:", error_message
+        )
+        if value_not_in_choices_error:
+            # ignore templated value not in allowed choices
+            choice_key = value_not_in_choices_error.group("name")
+            choice_value = task["action"][choice_key]
+            if has_jinja(choice_value):
+                _logger.debug(
+                    "Value checking ignored for '%s' option in task '%s' at line %s.",
+                    choice_key,
+                    module_name,
+                    task[LINE_NUMBER_KEY],
+                )
+                return results
+
+        results.append(
+            self.create_matcherror(
+                message=error_message,
+                linenumber=task[LINE_NUMBER_KEY],
+                tag="args[module]",
+                filename=file,
+            )
+        )
+        return results
+
+
+# testing code to be loaded only with pytest or when executed the rule file
+if "pytest" in sys.modules:
+    from ansiblelint.runner import Runner  # pylint: disable=ungrouped-imports
+
+    def test_args_module_fail() -> None:
+        """Test rule invalid module options."""
+        collection = RulesCollection()
+        collection.register(ArgsRule())
+        success = "examples/playbooks/rule-args-module-fail-1.yml"
+        results = Runner(success, rules=collection).run()
+        assert len(results) == 5
+        assert results[0].tag == "args[module]"
+        assert "missing required arguments" in results[0].message
+        assert results[1].tag == "args[module]"
+        assert "missing parameter(s) required by " in results[1].message
+        assert results[2].tag == "args[module]"
+        assert "Unsupported parameters for" in results[2].message
+        assert results[3].tag == "args[module]"
+        assert "Unsupported parameters for" in results[2].message
+        assert results[4].tag == "args[module]"
+        assert "value of state must be one of" in results[4].message
+
+    def test_args_module_pass() -> None:
+        """Test rule valid module options."""
+        collection = RulesCollection()
+        collection.register(ArgsRule())
+        success = "examples/playbooks/rule-args-module-pass-1.yml"
+        results = Runner(success, rules=collection).run()
+        assert len(results) == 0, results