path: root/src/ansiblelint/utils.py

# 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))