Adding upstream version 6.17.2.upstream/6.17.2

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

1 files changed, 1020 insertions, 0 deletions

diff --git a/src/ansiblelint/utils.py b/src/ansiblelint/utils.py
new file mode 100644
index 0000000..9cb97aa
--- /dev/null
+++ b/src/ansiblelint/utils.py
@@ -0,0 +1,1020 @@
+# Copyright (c) 2013-2014 Will Thames <will@thames.id.au>
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in
+# all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+# THE SOFTWARE.
+# spell-checker:ignore dwim
+# pylint: disable=too-many-lines
+"""Generic utility helpers."""
+from __future__ import annotations
+
+import contextlib
+import inspect
+import logging
+import os
+import re
+from collections.abc import Generator, ItemsView, Iterator, Mapping, Sequence
+from dataclasses import _MISSING_TYPE, dataclass, field
+from functools import cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+from ansible.errors import AnsibleError, AnsibleParserError
+from ansible.module_utils.parsing.convert_bool import boolean
+from ansible.parsing.dataloader import DataLoader
+from ansible.parsing.mod_args import ModuleArgsParser
+from ansible.parsing.yaml.constructor import AnsibleConstructor, AnsibleMapping
+from ansible.parsing.yaml.loader import AnsibleLoader
+from ansible.parsing.yaml.objects import AnsibleBaseYAMLObject, AnsibleSequence
+from ansible.plugins.loader import add_all_plugin_dirs
+from ansible.template import Templar
+from ansible.utils.collection_loader import AnsibleCollectionConfig
+from yaml.composer import Composer
+from yaml.representer import RepresenterError
+
+from ansiblelint._internal.rules import (
+    AnsibleParserErrorRule,
+    RuntimeErrorRule,
+)
+from ansiblelint.app import get_app
+from ansiblelint.config import Options, options
+from ansiblelint.constants import (
+    ANNOTATION_KEYS,
+    FILENAME_KEY,
+    INCLUSION_ACTION_NAMES,
+    LINE_NUMBER_KEY,
+    NESTED_TASK_KEYS,
+    PLAYBOOK_TASK_KEYWORDS,
+    ROLE_IMPORT_ACTION_NAMES,
+    SKIPPED_RULES_KEY,
+    FileType,
+)
+from ansiblelint.errors import MatchError
+from ansiblelint.file_utils import Lintable, discover_lintables
+from ansiblelint.skip_utils import is_nested_task
+from ansiblelint.text import removeprefix
+
+# ansible-lint doesn't need/want to know about encrypted secrets, so we pass a
+# string as the password to enable such yaml files to be opened and parsed
+# successfully.
+DEFAULT_VAULT_PASSWORD = "x"  # noqa: S105
+COLLECTION_PLAY_RE = re.compile(r"^[\w\d_]+\.[\w\d_]+\.[\w\d_]+$")
+
+PLAYBOOK_DIR = os.environ.get("ANSIBLE_PLAYBOOK_DIR", None)
+
+
+_logger = logging.getLogger(__name__)
+
+
+def parse_yaml_from_file(filepath: str) -> AnsibleBaseYAMLObject:
+    """Extract a decrypted YAML object from file."""
+    dataloader = DataLoader()
+    if hasattr(dataloader, "set_vault_password"):
+        dataloader.set_vault_password(DEFAULT_VAULT_PASSWORD)
+    return dataloader.load_from_file(filepath)
+
+
+def path_dwim(basedir: str, given: str) -> str:
+    """Convert a given path do-what-I-mean style."""
+    dataloader = DataLoader()
+    dataloader.set_basedir(basedir)
+    return str(dataloader.path_dwim(given))
+
+
+def ansible_templar(basedir: Path, templatevars: Any) -> Templar:
+    """Create an Ansible Templar using templatevars."""
+    # `basedir` is the directory containing the lintable file.
+    # Therefore, for tasks in a role, `basedir` has the form
+    # `roles/some_role/tasks`. On the other hand, the search path
+    # is `roles/some_role/{files,templates}`. As a result, the
+    # `tasks` part in the basedir should be stripped stripped.
+    if basedir.name == "tasks":
+        basedir = basedir.parent
+
+    dataloader = DataLoader()
+    dataloader.set_basedir(basedir)
+    templar = Templar(dataloader, variables=templatevars)
+    return templar
+
+
+def mock_filter(left: Any, *args: Any, **kwargs: Any) -> Any:  # noqa: ARG001
+    """Mock a filter that can take any combination of args and kwargs.
+
+    This will return x when x | filter(y,z) is called
+    e.g. {{ foo | ansible.utils.ipaddr('address') }}
+
+    :param left: The left hand side of the filter
+    :param args: The args passed to the filter
+    :param kwargs: The kwargs passed to the filter
+    :return: The left hand side of the filter
+    """
+    # pylint: disable=unused-argument
+    return left
+
+
+def ansible_template(
+    basedir: Path,
+    varname: Any,
+    templatevars: Any,
+    **kwargs: Any,
+) -> Any:
+    """Render a templated string by mocking missing filters.
+
+    In the case of a missing lookup, ansible core does an early exit
+    when disable_lookup=True but this happens after the jinja2 syntax already passed
+    return the original string as if it had been templated.
+
+    In the case of a missing filter, extract the missing filter plugin name
+    from the ansible error, 'Could not load "filter"'. Then mock the filter
+    and template the string again. The range allows for up to 10 unknown filters
+    in succession
+
+    :param basedir: The directory containing the lintable file
+    :param varname: The string to be templated
+    :param templatevars: The variables to be used in the template
+    :param kwargs: Additional arguments to be passed to the templating engine
+    :return: The templated string or None
+    :raises: AnsibleError if the filter plugin cannot be extracted or the
+             string could not be templated in 10 attempts
+    """
+    # pylint: disable=too-many-locals
+    filter_error = "template error while templating string:"
+    lookup_error = "was found, however lookups were disabled from templating"
+    re_filter_fqcn = re.compile(r"\w+\.\w+\.\w+")
+    re_filter_in_err = re.compile(r"Could not load \"(\w+)\"")
+    re_valid_filter = re.compile(r"^\w+(\.\w+\.\w+)?$")
+    templar = ansible_templar(basedir=basedir, templatevars=templatevars)
+
+    kwargs["disable_lookups"] = True
+    for _i in range(10):
+        try:
+            templated = templar.template(varname, **kwargs)
+            return templated
+        except AnsibleError as exc:
+            if lookup_error in exc.message:
+                return varname
+            if exc.message.startswith(filter_error):
+                while True:
+                    match = re_filter_in_err.search(exc.message)
+                    if match:
+                        missing_filter = match.group(1)
+                        break
+                    match = re_filter_fqcn.search(exc.message)
+                    if match:
+                        missing_filter = match.group(0)
+                        break
+                    missing_filter = exc.message.split("'")[1]
+                    break
+
+                if not re_valid_filter.match(missing_filter):
+                    err = f"Could not parse missing filter name from error message: {exc.message}"
+                    _logger.warning(err)
+                    raise
+
+                # pylint: disable=protected-access
+                templar.environment.filters._delegatee[  # noqa: SLF001
+                    missing_filter
+                ] = mock_filter
+                # Record the mocked filter so we can warn the user
+                if missing_filter not in options.mock_filters:
+                    _logger.debug("Mocking missing filter %s", missing_filter)
+                    options.mock_filters.append(missing_filter)
+                continue
+            raise
+    return None
+
+
+BLOCK_NAME_TO_ACTION_TYPE_MAP = {
+    "tasks": "task",
+    "handlers": "handler",
+    "pre_tasks": "task",
+    "post_tasks": "task",
+    "block": "meta",
+    "rescue": "meta",
+    "always": "meta",
+}
+
+
+def tokenize(line: str) -> tuple[str, list[str], dict[str, str]]:
+    """Parse a string task invocation."""
+    tokens = line.lstrip().split(" ")
+    if tokens[0] == "-":
+        tokens = tokens[1:]
+    if tokens[0] == "action:" or tokens[0] == "local_action:":
+        tokens = tokens[1:]
+    command = tokens[0].replace(":", "")
+
+    args = []
+    kwargs = {}
+    non_kv_found = False
+    for arg in tokens[1:]:
+        if "=" in arg and not non_kv_found:
+            key_value = arg.split("=", 1)
+            kwargs[key_value[0]] = key_value[1]
+        else:
+            non_kv_found = True
+            args.append(arg)
+    return (command, args, kwargs)
+
+
+def playbook_items(pb_data: AnsibleBaseYAMLObject) -> ItemsView:  # type: ignore[type-arg]
+    """Return a list of items from within the playbook."""
+    if isinstance(pb_data, dict):
+        return pb_data.items()
+    if not pb_data:
+        return []  # type: ignore[return-value]
+
+    # "if play" prevents failure if the play sequence contains None,
+    # which is weird but currently allowed by Ansible
+    # https://github.com/ansible/ansible-lint/issues/849
+    return [item for play in pb_data if play for item in play.items()]  # type: ignore[return-value]
+
+
+def set_collections_basedir(basedir: Path) -> None:
+    """Set the playbook directory as playbook_paths for the collection loader."""
+    # Ansible expects only absolute paths inside `playbook_paths` and will
+    # produce weird errors if we use a relative one.
+    AnsibleCollectionConfig.playbook_paths = str(basedir.resolve())
+
+
+def template(
+    basedir: Path,
+    value: Any,
+    variables: Any,
+    *,
+    fail_on_error: bool = False,
+    fail_on_undefined: bool = False,
+    **kwargs: str,
+) -> Any:
+    """Attempt rendering a value with known vars."""
+    try:
+        value = ansible_template(
+            basedir.resolve(),
+            value,
+            variables,
+            **dict(kwargs, fail_on_undefined=fail_on_undefined),
+        )
+        # Hack to skip the following exception when using to_json filter on a variable. # noqa: FIX004
+        # I guess the filter doesn't like empty vars...
+    except (AnsibleError, ValueError, RepresenterError):
+        # templating failed, so just keep value as is.
+        if fail_on_error:
+            raise
+    return value
+
+
+def _include_children(
+    basedir: str,
+    k: str,
+    v: Any,
+    parent_type: FileType,
+) -> list[Lintable]:
+    # handle special case include_tasks: name=filename.yml
+    if k in INCLUSION_ACTION_NAMES and isinstance(v, dict) and "file" in v:
+        v = v["file"]
+
+    # we cannot really parse any jinja2 in includes, so we ignore them
+    if not v or "{{" in v:
+        return []
+
+    if "import_playbook" in k and COLLECTION_PLAY_RE.match(v):
+        # Any import_playbooks from collections should be ignored as ansible
+        # own syntax check will handle them.
+        return []
+
+    # handle include: filename.yml tags=blah
+    # pylint: disable=unused-variable
+    (command, args, kwargs) = tokenize(f"{k}: {v}")
+
+    result = path_dwim(basedir, args[0])
+    while basedir not in ["", "/"]:
+        if os.path.exists(result):
+            break
+        basedir = os.path.dirname(basedir)
+        result = path_dwim(basedir, args[0])
+
+    return [Lintable(result, kind=parent_type)]
+
+
+def _taskshandlers_children(
+    basedir: str,
+    k: str,
+    v: None | Any,
+    parent_type: FileType,
+) -> list[Lintable]:
+    results: list[Lintable] = []
+    if v is None:
+        raise MatchError(
+            message="A malformed block was encountered while loading a block.",
+            rule=RuntimeErrorRule(),
+        )
+    for task_handler in v:
+        # ignore empty tasks, `-`
+        if not task_handler:
+            continue
+
+        with contextlib.suppress(LookupError):
+            children = _get_task_handler_children_for_tasks_or_playbooks(
+                task_handler,
+                basedir,
+                k,
+                parent_type,
+            )
+            results.append(children)
+            continue
+
+        if any(x in task_handler for x in ROLE_IMPORT_ACTION_NAMES):
+            task_handler = normalize_task_v2(task_handler)
+            _validate_task_handler_action_for_role(task_handler["action"])
+            results.extend(
+                _roles_children(
+                    basedir,
+                    k,
+                    [task_handler["action"].get("name")],
+                    parent_type,
+                    main=task_handler["action"].get("tasks_from", "main"),
+                ),
+            )
+            continue
+
+        if "block" not in task_handler:
+            continue
+
+        results.extend(
+            _taskshandlers_children(basedir, k, task_handler["block"], parent_type),
+        )
+        if "rescue" in task_handler:
+            results.extend(
+                _taskshandlers_children(
+                    basedir,
+                    k,
+                    task_handler["rescue"],
+                    parent_type,
+                ),
+            )
+        if "always" in task_handler:
+            results.extend(
+                _taskshandlers_children(
+                    basedir,
+                    k,
+                    task_handler["always"],
+                    parent_type,
+                ),
+            )
+
+    return results
+
+
+def _get_task_handler_children_for_tasks_or_playbooks(
+    task_handler: dict[str, Any],
+    basedir: str,
+    k: Any,
+    parent_type: FileType,
+) -> Lintable:
+    """Try to get children of taskhandler for include/import tasks/playbooks."""
+    child_type = k if parent_type == "playbook" else parent_type
+
+    # Include the FQCN task names as this happens before normalize
+    for task_handler_key in INCLUSION_ACTION_NAMES:
+        with contextlib.suppress(KeyError):
+            # ignore empty tasks
+            if not task_handler:  # pragma: no branch
+                continue
+
+            file_name = task_handler[task_handler_key]
+            if isinstance(file_name, Mapping) and file_name.get("file", None):
+                file_name = file_name["file"]
+
+            f = path_dwim(basedir, file_name)
+            while basedir not in ["", "/"]:
+                if os.path.exists(f):
+                    break
+                basedir = os.path.dirname(basedir)
+                f = path_dwim(basedir, file_name)
+            return Lintable(f, kind=child_type)
+    msg = f'The node contains none of: {", ".join(sorted(INCLUSION_ACTION_NAMES))}'
+    raise LookupError(msg)
+
+
+def _validate_task_handler_action_for_role(th_action: dict[str, Any]) -> None:
+    """Verify that the task handler action is valid for role include."""
+    module = th_action["__ansible_module__"]
+
+    if "name" not in th_action:
+        raise MatchError(message=f"Failed to find required 'name' key in {module!s}")
+
+    if not isinstance(th_action["name"], str):
+        raise MatchError(
+            message=f"Value assigned to 'name' key on '{module!s}' is not a string.",
+        )
+
+
+def _roles_children(
+    basedir: str,
+    k: str,
+    v: Sequence[Any],
+    parent_type: FileType,  # noqa: ARG001
+    main: str = "main",
+) -> list[Lintable]:
+    # pylint: disable=unused-argument # parent_type)
+    results: list[Lintable] = []
+    if not v:
+        # typing does not prevent junk from being passed in
+        return results
+    for role in v:
+        if isinstance(role, dict):
+            if "role" in role or "name" in role:
+                if "tags" not in role or "skip_ansible_lint" not in role["tags"]:
+                    results.extend(
+                        _look_for_role_files(
+                            basedir,
+                            role.get("role", role.get("name")),
+                            main=main,
+                        ),
+                    )
+            elif k != "dependencies":
+                msg = f'role dict {role} does not contain a "role" or "name" key'
+                raise SystemExit(msg)
+        else:
+            results.extend(_look_for_role_files(basedir, role, main=main))
+    return results
+
+
+def _rolepath(basedir: str, role: str) -> str | None:
+    role_path = None
+
+    possible_paths = [
+        # if included from a playbook
+        path_dwim(basedir, os.path.join("roles", role)),
+        path_dwim(basedir, role),
+        # if included from roles/[role]/meta/main.yml
+        path_dwim(basedir, os.path.join("..", "..", "..", "roles", role)),
+        path_dwim(basedir, os.path.join("..", "..", role)),
+        # if checking a role in the current directory
+        path_dwim(basedir, os.path.join("..", role)),
+    ]
+
+    for loc in get_app(offline=True).runtime.config.default_roles_path:
+        loc = os.path.expanduser(loc)
+        possible_paths.append(path_dwim(loc, role))
+
+    possible_paths.append(path_dwim(basedir, ""))
+
+    for path_option in possible_paths:  # pragma: no branch
+        if os.path.isdir(path_option):
+            role_path = path_option
+            break
+
+    if role_path:  # pragma: no branch
+        add_all_plugin_dirs(role_path)
+
+    return role_path
+
+
+def _look_for_role_files(
+    basedir: str,
+    role: str,
+    main: str | None = "main",  # noqa: ARG001
+) -> list[Lintable]:
+    # pylint: disable=unused-argument # main
+    role_path = _rolepath(basedir, role)
+    if not role_path:  # pragma: no branch
+        return []
+
+    results = []
+
+    for kind in ["tasks", "meta", "handlers", "vars", "defaults"]:
+        current_path = os.path.join(role_path, kind)
+        for folder, _, files in os.walk(current_path):
+            for file in files:
+                file_ignorecase = file.lower()
+                if file_ignorecase.endswith((".yml", ".yaml")):
+                    results.append(Lintable(os.path.join(folder, file)))
+
+    return results
+
+
+def _sanitize_task(task: dict[str, Any]) -> dict[str, Any]:
+    """Return a stripped-off task structure compatible with new Ansible.
+
+    This helper takes a copy of the incoming task and drops
+    any internally used keys from it.
+    """
+    result = task.copy()
+    # task is an AnsibleMapping which inherits from OrderedDict, so we need
+    # to use `del` to remove unwanted keys.
+    for k in [SKIPPED_RULES_KEY, FILENAME_KEY, LINE_NUMBER_KEY]:
+        if k in result:
+            del result[k]
+    return result
+
+
+def _extract_ansible_parsed_keys_from_task(
+    result: dict[str, Any],
+    task: dict[str, Any],
+    keys: tuple[str, ...],
+) -> dict[str, Any]:
+    """Return a dict with existing key in task."""
+    for k, v in list(task.items()):
+        if k in keys:
+            # we don't want to re-assign these values, which were
+            # determined by the ModuleArgsParser() above
+            continue
+        result[k] = v
+    return result
+
+
+def normalize_task_v2(task: dict[str, Any]) -> dict[str, Any]:
+    """Ensure tasks have a normalized action key and strings are converted to python objects."""
+    result: dict[str, Any] = {}
+    ansible_parsed_keys = ("action", "local_action", "args", "delegate_to")
+
+    if is_nested_task(task):
+        _extract_ansible_parsed_keys_from_task(result, task, ansible_parsed_keys)
+        # Add dummy action for block/always/rescue statements
+        result["action"] = {
+            "__ansible_module__": "block/always/rescue",
+            "__ansible_module_original__": "block/always/rescue",
+        }
+
+        return result
+
+    sanitized_task = _sanitize_task(task)
+    mod_arg_parser = ModuleArgsParser(sanitized_task)
+
+    try:
+        action, arguments, result["delegate_to"] = mod_arg_parser.parse(
+            skip_action_validation=options.skip_action_validation,
+        )
+    except AnsibleParserError as exc:
+        # pylint: disable=raise-missing-from
+        raise MatchError(
+            rule=AnsibleParserErrorRule(),
+            message=exc.message,
+            filename=task.get(FILENAME_KEY, "Unknown"),
+            lineno=task.get(LINE_NUMBER_KEY, 0),
+        ) from exc
+
+    # denormalize shell -> command conversion
+    if "_uses_shell" in arguments:
+        action = "shell"
+        del arguments["_uses_shell"]
+
+    _extract_ansible_parsed_keys_from_task(
+        result,
+        task,
+        (*ansible_parsed_keys, action),
+    )
+
+    if not isinstance(action, str):
+        msg = f"Task actions can only be strings, got {action}"
+        raise RuntimeError(msg)
+    action_unnormalized = action
+    # convert builtin fqn calls to short forms because most rules know only
+    # about short calls but in the future we may switch the normalization to do
+    # the opposite. Mainly we currently consider normalized the module listing
+    # used by `ansible-doc -t module -l 2>/dev/null`
+    action = removeprefix(action, "ansible.builtin.")
+    result["action"] = {
+        "__ansible_module__": action,
+        "__ansible_module_original__": action_unnormalized,
+    }
+
+    result["action"].update(arguments)
+    return result
+
+
+def normalize_task(task: dict[str, Any], filename: str) -> dict[str, Any]:
+    """Unify task-like object structures."""
+    ansible_action_type = task.get("__ansible_action_type__", "task")
+    if "__ansible_action_type__" in task:
+        del task["__ansible_action_type__"]
+    task = normalize_task_v2(task)
+    task[FILENAME_KEY] = filename
+    task["__ansible_action_type__"] = ansible_action_type
+    return task
+
+
+def task_to_str(task: dict[str, Any]) -> str:
+    """Make a string identifier for the given task."""
+    name = task.get("name")
+    if name:
+        return str(name)
+    action = task.get("action")
+    if isinstance(action, str) or not isinstance(action, dict):
+        return str(action)
+    args = [
+        f"{k}={v}"
+        for (k, v) in action.items()
+        if k
+        not in [
+            "__ansible_module__",
+            "__ansible_module_original__",
+            "_raw_params",
+            LINE_NUMBER_KEY,
+            FILENAME_KEY,
+        ]
+    ]
+
+    _raw_params = action.get("_raw_params", [])
+    if isinstance(_raw_params, list):
+        for item in _raw_params:
+            args.append(str(item))
+    else:
+        args.append(_raw_params)
+
+    return f"{action['__ansible_module__']} {' '.join(args)}"
+
+
+def extract_from_list(
+    blocks: AnsibleBaseYAMLObject,
+    candidates: list[str],
+    *,
+    recursive: bool = False,
+) -> list[Any]:
+    """Get action tasks from block structures."""
+    results = []
+    for block in blocks:
+        for candidate in candidates:
+            if isinstance(block, dict) and candidate in block:
+                if isinstance(block[candidate], list):
+                    subresults = add_action_type(block[candidate], candidate)
+                    if recursive:
+                        subresults.extend(
+                            extract_from_list(
+                                subresults,
+                                candidates,
+                                recursive=recursive,
+                            ),
+                        )
+                    results.extend(subresults)
+                elif block[candidate] is not None:
+                    msg = f"Key '{candidate}' defined, but bad value: '{block[candidate]!s}'"
+                    raise RuntimeError(msg)
+    return results
+
+
+@dataclass
+class Task(dict[str, Any]):
+    """Class that represents a task from linter point of view.
+
+    raw_task:
+        When looping through the tasks in the file, each "raw_task" is minimally
+        processed to include these special keys: __line__, __file__, skipped_rules.
+    normalized_task:
+        When each raw_task is "normalized", action shorthand (strings) get parsed
+        by ansible into python objects and the action key gets normalized. If the task
+        should be skipped (skipped is True) or normalizing it fails (error is not None)
+        then this is just the raw_task instead of a normalized copy.
+    skip_tags:
+        List of tags found to be skipped, from tags block or noqa comments
+    error:
+        This is normally None. It will be a MatchError when the raw_task cannot be
+        normalized due to an AnsibleParserError.
+    position: Any
+    """
+
+    raw_task: dict[str, Any]
+    filename: str = ""
+    _normalized_task: dict[str, Any] | _MISSING_TYPE = field(init=False, repr=False)
+    error: MatchError | None = None
+    position: Any = None
+
+    @property
+    def name(self) -> str | None:
+        """Return the name of the task."""
+        return self.raw_task.get("name", None)
+
+    @property
+    def action(self) -> str:
+        """Return the resolved action name."""
+        action_name = self.normalized_task["action"]["__ansible_module_original__"]
+        if not isinstance(action_name, str):
+            msg = "Task actions can only be strings."
+            raise RuntimeError(msg)
+        return action_name
+
+    @property
+    def args(self) -> Any:
+        """Return the arguments passed to the task action.
+
+        While we usually expect to return a dictionary, it can also
+        return a templated string when jinja is used.
+        """
+        if "args" in self.raw_task:
+            return self.raw_task["args"]
+        result = {}
+        for k, v in self.normalized_task["action"].items():
+            if k not in ANNOTATION_KEYS:
+                result[k] = v
+        return result
+
+    @property
+    def normalized_task(self) -> dict[str, Any]:
+        """Return the name of the task."""
+        if not hasattr(self, "_normalized_task"):
+            try:
+                self._normalized_task = normalize_task(
+                    self.raw_task,
+                    filename=self.filename,
+                )
+            except MatchError as err:
+                self.error = err
+                # When we cannot normalize it, we just use the raw task instead
+                # to avoid adding extra complexity to the rules.
+                self._normalized_task = self.raw_task
+        if isinstance(self._normalized_task, _MISSING_TYPE):
+            msg = "Task was not normalized"
+            raise RuntimeError(msg)
+        return self._normalized_task
+
+    @property
+    def skip_tags(self) -> list[str]:
+        """Return the list of tags to skip."""
+        skip_tags: list[str] = self.raw_task.get(SKIPPED_RULES_KEY, [])
+        return skip_tags
+
+    def __repr__(self) -> str:
+        """Return a string representation of the task."""
+        return f"Task('{self.name}' [{self.position}])"
+
+    def get(self, key: str, default: Any = None) -> Any:
+        """Get a value from the task."""
+        return self.normalized_task.get(key, default)
+
+    def __getitem__(self, index: str) -> Any:
+        """Allow access as task[...]."""
+        return self.normalized_task[index]
+
+    def __iter__(self) -> Generator[str, None, None]:
+        """Provide support for 'key in task'."""
+        yield from (f for f in self.normalized_task)
+
+
+def task_in_list(
+    data: AnsibleBaseYAMLObject,
+    file: Lintable,
+    kind: str,
+    position: str = ".",
+) -> Iterator[Task]:
+    """Get action tasks from block structures."""
+
+    def each_entry(data: AnsibleBaseYAMLObject, position: str) -> Iterator[Task]:
+        if not data:
+            return
+        for entry_index, entry in enumerate(data):
+            if not entry:
+                continue
+            _pos = f"{position}[{entry_index}]"
+            if isinstance(entry, dict):
+                yield Task(
+                    entry,
+                    position=_pos,
+                )
+            for block in [k for k in entry if k in NESTED_TASK_KEYS]:
+                yield from task_in_list(
+                    data=entry[block],
+                    file=file,
+                    kind="tasks",
+                    position=f"{_pos}.{block}",
+                )
+
+    if not isinstance(data, list):
+        return
+    if kind == "playbook":
+        attributes = ["tasks", "pre_tasks", "post_tasks", "handlers"]
+        for item_index, item in enumerate(data):
+            for attribute in attributes:
+                if not isinstance(item, dict):
+                    continue
+                if attribute in item:
+                    if isinstance(item[attribute], list):
+                        yield from each_entry(
+                            item[attribute],
+                            f"{position }[{item_index}].{attribute}",
+                        )
+                    elif item[attribute] is not None:
+                        msg = f"Key '{attribute}' defined, but bad value: '{item[attribute]!s}'"
+                        raise RuntimeError(msg)
+    else:
+        yield from each_entry(data, position)
+
+
+def add_action_type(actions: AnsibleBaseYAMLObject, action_type: str) -> list[Any]:
+    """Add action markers to task objects."""
+    results = []
+    for action in actions:
+        # ignore empty task
+        if not action:
+            continue
+        action["__ansible_action_type__"] = BLOCK_NAME_TO_ACTION_TYPE_MAP[action_type]
+        results.append(action)
+    return results
+
+
+def get_action_tasks(data: AnsibleBaseYAMLObject, file: Lintable) -> list[Any]:
+    """Get a flattened list of action tasks from the file."""
+    tasks = []
+    if file.kind in ["tasks", "handlers"]:
+        tasks = add_action_type(data, file.kind)
+    else:
+        tasks.extend(extract_from_list(data, PLAYBOOK_TASK_KEYWORDS))
+
+    # Add sub-elements of block/rescue/always to tasks list
+    tasks.extend(extract_from_list(tasks, NESTED_TASK_KEYS, recursive=True))
+
+    return tasks
+
+
+@cache
+def parse_yaml_linenumbers(
+    lintable: Lintable,
+) -> AnsibleBaseYAMLObject:
+    """Parse yaml as ansible.utils.parse_yaml but with linenumbers.
+
+    The line numbers are stored in each node's LINE_NUMBER_KEY key.
+    """
+    result = []
+
+    def compose_node(parent: yaml.nodes.Node, index: int) -> yaml.nodes.Node:
+        # the line number where the previous token has ended (plus empty lines)
+        line = loader.line
+        node = Composer.compose_node(loader, parent, index)
+        if not isinstance(node, yaml.nodes.Node):
+            msg = "Unexpected yaml data."
+            raise RuntimeError(msg)
+        node.__line__ = line + 1  # type: ignore[attr-defined]
+        return node
+
+    def construct_mapping(
+        node: AnsibleBaseYAMLObject,
+        *,
+        deep: bool = False,
+    ) -> AnsibleMapping:
+        mapping = AnsibleConstructor.construct_mapping(loader, node, deep=deep)
+        if hasattr(node, "__line__"):
+            mapping[LINE_NUMBER_KEY] = node.__line__
+        else:
+            mapping[
+                LINE_NUMBER_KEY
+            ] = mapping._line_number  # pylint: disable=protected-access  # noqa: SLF001
+        mapping[FILENAME_KEY] = lintable.path
+        return mapping
+
+    try:
+        kwargs = {}
+        if "vault_password" in inspect.getfullargspec(AnsibleLoader.__init__).args:
+            kwargs["vault_password"] = DEFAULT_VAULT_PASSWORD
+        loader = AnsibleLoader(lintable.content, **kwargs)
+        loader.compose_node = compose_node
+        loader.construct_mapping = construct_mapping
+        # while Ansible only accepts single documents, we also need to load
+        # multi-documents, as we attempt to load any YAML file, not only
+        # Ansible managed ones.
+        while True:
+            data = loader.get_data()
+            if data is None:
+                break
+            result.append(data)
+    except (
+        yaml.parser.ParserError,
+        yaml.scanner.ScannerError,
+        yaml.constructor.ConstructorError,
+    ) as exc:
+        msg = "Failed to load YAML file"
+        raise RuntimeError(msg) from exc
+
+    if len(result) == 0:
+        return None  # empty documents
+    if len(result) == 1:
+        return result[0]
+    return result
+
+
+def get_cmd_args(task: dict[str, Any]) -> str:
+    """Extract the args from a cmd task as a string."""
+    if "cmd" in task["action"]:
+        args = task["action"]["cmd"]
+    else:
+        args = task["action"].get("_raw_params", [])
+    if not isinstance(args, str):
+        return " ".join(args)
+    return args
+
+
+def get_first_cmd_arg(task: dict[str, Any]) -> Any:
+    """Extract the first arg from a cmd task."""
+    try:
+        first_cmd_arg = get_cmd_args(task).split()[0]
+    except IndexError:
+        return None
+    return first_cmd_arg
+
+
+def get_second_cmd_arg(task: dict[str, Any]) -> Any:
+    """Extract the second arg from a cmd task."""
+    try:
+        second_cmd_arg = get_cmd_args(task).split()[1]
+    except IndexError:
+        return None
+    return second_cmd_arg
+
+
+def is_playbook(filename: str) -> bool:
+    """Check if the file is a playbook.
+
+    Given a filename, it should return true if it looks like a playbook. The
+    function is not supposed to raise exceptions.
+    """
+    # we assume is a playbook if we loaded a sequence of dictionaries where
+    # at least one of these keys is present:
+    playbooks_keys = {
+        "gather_facts",
+        "hosts",
+        "import_playbook",
+        "post_tasks",
+        "pre_tasks",
+        "roles",
+        "tasks",
+    }
+
+    # makes it work with Path objects by converting them to strings
+    if not isinstance(filename, str):
+        filename = str(filename)
+
+    try:
+        f = parse_yaml_from_file(filename)
+    except Exception as exc:  # pylint: disable=broad-except # noqa: BLE001
+        _logger.warning(
+            "Failed to load %s with %s, assuming is not a playbook.",
+            filename,
+            exc,
+        )
+    else:
+        if (
+            isinstance(f, AnsibleSequence)
+            and hasattr(next(iter(f), {}), "keys")
+            and playbooks_keys.intersection(next(iter(f), {}).keys())
+        ):
+            return True
+    return False
+
+
+# pylint: disable=too-many-statements
+def get_lintables(
+    opts: Options = options,
+    args: list[str] | None = None,
+) -> list[Lintable]:
+    """Detect files and directories that are lintable."""
+    lintables: list[Lintable] = []
+
+    # passing args bypass auto-detection mode
+    if args:
+        for arg in args:
+            lintable = Lintable(arg)
+            lintables.append(lintable)
+    else:
+        for filename in discover_lintables(opts):
+            path = Path(filename)
+            lintables.append(Lintable(path))
+
+        # stage 2: guess roles from current lintables, as there is no unique
+        # file that must be present in any kind of role.
+        _extend_with_roles(lintables)
+
+    return lintables
+
+
+def _extend_with_roles(lintables: list[Lintable]) -> None:
+    """Detect roles among lintables and adds them to the list."""
+    for lintable in lintables:
+        parts = lintable.path.parent.parts
+        if "roles" in parts:
+            role = lintable.path
+            while role.parent.name != "roles" and role.name:
+                role = role.parent
+            if role.exists() and not role.is_file():
+                lintable = Lintable(role)
+                if lintable.kind == "role" and lintable not in lintables:
+                    _logger.debug("Added role: %s", lintable)
+                    lintables.append(lintable)
+
+
+def convert_to_boolean(value: Any) -> bool:
+    """Use Ansible to convert something to a boolean."""
+    return bool(boolean(value))