diff options
Diffstat (limited to 'src/ansiblelint/rules/args.py')
| -rw-r--r-- | src/ansiblelint/rules/args.py | 280 |
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 |