summaryrefslogtreecommitdiffstats
path: root/lib/ansiblelint/utils.py
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-14 20:04:50 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-14 20:04:50 +0000
commit782f8df6e41f29dce2db4970a3ad84aaeb7d8c5f (patch)
tree3a88a542cd8074743d251881131510157cfc510b /lib/ansiblelint/utils.py
parentInitial commit. (diff)
downloadansible-lint-782f8df6e41f29dce2db4970a3ad84aaeb7d8c5f.tar.xz
ansible-lint-782f8df6e41f29dce2db4970a3ad84aaeb7d8c5f.zip
Adding upstream version 4.3.7.upstream/4.3.7upstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to '')
-rw-r--r--lib/ansiblelint/utils.py836
1 files changed, 836 insertions, 0 deletions
diff --git a/lib/ansiblelint/utils.py b/lib/ansiblelint/utils.py
new file mode 100644
index 0000000..feac4d7
--- /dev/null
+++ b/lib/ansiblelint/utils.py
@@ -0,0 +1,836 @@
+# 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.
+"""Generic utility helpers."""
+
+import contextlib
+import inspect
+import logging
+import os
+import pprint
+import subprocess
+from argparse import Namespace
+from collections import OrderedDict
+from functools import lru_cache
+from pathlib import Path
+from typing import Callable, ItemsView, List, Optional, Tuple
+
+import yaml
+from ansible import constants
+from ansible.errors import AnsibleError, AnsibleParserError
+from ansible.parsing.dataloader import DataLoader
+from ansible.parsing.mod_args import ModuleArgsParser
+from ansible.parsing.splitter import split_args
+from ansible.parsing.yaml.constructor import AnsibleConstructor
+from ansible.parsing.yaml.loader import AnsibleLoader
+from ansible.parsing.yaml.objects import AnsibleSequence
+from ansible.plugins.loader import add_all_plugin_dirs
+from ansible.template import Templar
+from yaml.composer import Composer
+from yaml.representer import RepresenterError
+
+from ansiblelint.constants import (
+ ANSIBLE_FAILURE_RC, CUSTOM_RULESDIR_ENVVAR, DEFAULT_RULESDIR, FileType,
+)
+from ansiblelint.errors import MatchError
+from ansiblelint.file_utils import normpath
+
+# 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'
+
+PLAYBOOK_DIR = os.environ.get('ANSIBLE_PLAYBOOK_DIR', None)
+
+
+_logger = logging.getLogger(__name__)
+
+
+def parse_yaml_from_file(filepath: str) -> dict:
+ dl = DataLoader()
+ if hasattr(dl, 'set_vault_password'):
+ dl.set_vault_password(DEFAULT_VAULT_PASSWORD)
+ return dl.load_from_file(filepath)
+
+
+def path_dwim(basedir: str, given: str) -> str:
+ dl = DataLoader()
+ dl.set_basedir(basedir)
+ return dl.path_dwim(given)
+
+
+def ansible_template(basedir, varname, templatevars, **kwargs):
+ dl = DataLoader()
+ dl.set_basedir(basedir)
+ templar = Templar(dl, variables=templatevars)
+ return templar.template(varname, **kwargs)
+
+
+LINE_NUMBER_KEY = '__line__'
+FILENAME_KEY = '__file__'
+
+VALID_KEYS = [
+ 'name', 'action', 'when', 'async', 'poll', 'notify',
+ 'first_available_file', 'include', 'include_tasks', 'import_tasks', 'import_playbook',
+ 'tags', 'register', 'ignore_errors', 'delegate_to',
+ 'local_action', 'transport', 'remote_user', 'sudo',
+ 'sudo_user', 'sudo_pass', 'when', 'connection', 'environment', 'args', 'always_run',
+ 'any_errors_fatal', 'changed_when', 'failed_when', 'check_mode', 'delay',
+ 'retries', 'until', 'su', 'su_user', 'su_pass', 'no_log', 'run_once',
+ 'become', 'become_user', 'become_method', FILENAME_KEY,
+]
+
+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):
+ 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 = list()
+ kwargs = dict()
+ nonkvfound = False
+ for arg in tokens[1:]:
+ if "=" in arg and not nonkvfound:
+ kv = arg.split("=", 1)
+ kwargs[kv[0]] = kv[1]
+ else:
+ nonkvfound = True
+ args.append(arg)
+ return (command, args, kwargs)
+
+
+def _playbook_items(pb_data: dict) -> ItemsView:
+ if isinstance(pb_data, dict):
+ return pb_data.items()
+ elif not pb_data:
+ return []
+ else:
+ return [item for play in pb_data for item in play.items()]
+
+
+def _rebind_match_filename(filename: str, func) -> Callable:
+ def func_wrapper(*args, **kwargs):
+ try:
+ return func(*args, **kwargs)
+ except MatchError as e:
+ e.filename = filename
+ raise e
+ return func_wrapper
+
+
+def _set_collections_basedir(basedir: str):
+ # Sets the playbook directory as playbook_paths for the collection loader
+ try:
+ # Ansible 2.10+
+ # noqa: # pylint:disable=cyclic-import,import-outside-toplevel
+ from ansible.utils.collection_loader import AnsibleCollectionConfig
+
+ AnsibleCollectionConfig.playbook_paths = basedir
+ except ImportError:
+ # Ansible 2.8 or 2.9
+ # noqa: # pylint:disable=cyclic-import,import-outside-toplevel
+ from ansible.utils.collection_loader import set_collection_playbook_paths
+
+ set_collection_playbook_paths(basedir)
+
+
+def find_children(playbook: Tuple[str, str], playbook_dir: str) -> List:
+ if not os.path.exists(playbook[0]):
+ return []
+ _set_collections_basedir(playbook_dir or '.')
+ add_all_plugin_dirs(playbook_dir or '.')
+ if playbook[1] == 'role':
+ playbook_ds = {'roles': [{'role': playbook[0]}]}
+ else:
+ try:
+ playbook_ds = parse_yaml_from_file(playbook[0])
+ except AnsibleError as e:
+ raise SystemExit(str(e))
+ results = []
+ basedir = os.path.dirname(playbook[0])
+ items = _playbook_items(playbook_ds)
+ for item in items:
+ for child in _rebind_match_filename(playbook[0], play_children)(
+ basedir, item, playbook[1], playbook_dir):
+ if "$" in child['path'] or "{{" in child['path']:
+ continue
+ valid_tokens = list()
+ for token in split_args(child['path']):
+ if '=' in token:
+ break
+ valid_tokens.append(token)
+ path = ' '.join(valid_tokens)
+ results.append({
+ 'path': path_dwim(basedir, path),
+ 'type': child['type']
+ })
+ return results
+
+
+def template(basedir, value, vars, fail_on_undefined=False, **kwargs):
+ try:
+ value = ansible_template(os.path.abspath(basedir), value, vars,
+ **dict(kwargs, fail_on_undefined=fail_on_undefined))
+ # Hack to skip the following exception when using to_json filter on a variable.
+ # I guess the filter doesn't like empty vars...
+ except (AnsibleError, ValueError, RepresenterError):
+ # templating failed, so just keep value as is.
+ pass
+ return value
+
+
+def play_children(basedir, item, parent_type, playbook_dir):
+ delegate_map = {
+ 'tasks': _taskshandlers_children,
+ 'pre_tasks': _taskshandlers_children,
+ 'post_tasks': _taskshandlers_children,
+ 'block': _taskshandlers_children,
+ 'include': _include_children,
+ 'import_playbook': _include_children,
+ 'roles': _roles_children,
+ 'dependencies': _roles_children,
+ 'handlers': _taskshandlers_children,
+ 'include_tasks': _include_children,
+ 'import_tasks': _include_children,
+ }
+ (k, v) = item
+ add_all_plugin_dirs(os.path.abspath(basedir))
+
+ if k in delegate_map:
+ if v:
+ v = template(os.path.abspath(basedir),
+ v,
+ dict(playbook_dir=PLAYBOOK_DIR or os.path.abspath(basedir)),
+ fail_on_undefined=False)
+ return delegate_map[k](basedir, k, v, parent_type)
+ return []
+
+
+def _include_children(basedir, k, v, parent_type):
+ # handle special case include_tasks: name=filename.yml
+ if k == 'include_tasks' and isinstance(v, dict) and 'file' in v:
+ v = v['file']
+
+ # handle include: filename.yml tags=blah
+ (command, args, kwargs) = tokenize("{0}: {1}".format(k, v))
+
+ result = path_dwim(basedir, args[0])
+ if not os.path.exists(result):
+ result = path_dwim(os.path.join(os.path.dirname(basedir)), v)
+ return [{'path': result, 'type': parent_type}]
+
+
+def _taskshandlers_children(basedir, k, v, parent_type: FileType) -> List:
+ results = []
+ for th in v:
+
+ # ignore empty tasks, `-`
+ if not th:
+ continue
+
+ with contextlib.suppress(LookupError):
+ children = _get_task_handler_children_for_tasks_or_playbooks(
+ th, basedir, k, parent_type,
+ )
+ results.append(children)
+ continue
+
+ if 'include_role' in th or 'import_role' in th: # lgtm [py/unreachable-statement]
+ th = normalize_task_v2(th)
+ _validate_task_handler_action_for_role(th['action'])
+ results.extend(_roles_children(basedir, k, [th['action'].get("name")],
+ parent_type,
+ main=th['action'].get('tasks_from', 'main')))
+ continue
+
+ if 'block' not in th:
+ continue
+
+ results.extend(_taskshandlers_children(basedir, k, th['block'], parent_type))
+ if 'rescue' in th:
+ results.extend(_taskshandlers_children(basedir, k, th['rescue'], parent_type))
+ if 'always' in th:
+ results.extend(_taskshandlers_children(basedir, k, th['always'], parent_type))
+
+ return results
+
+
+def _get_task_handler_children_for_tasks_or_playbooks(
+ task_handler, basedir: str, k, parent_type: FileType,
+) -> dict:
+ """Try to get children of taskhandler for include/import tasks/playbooks."""
+ child_type = k if parent_type == 'playbook' else parent_type
+
+ task_include_keys = 'include', 'include_tasks', 'import_playbook', 'import_tasks'
+ for task_handler_key in task_include_keys:
+
+ with contextlib.suppress(KeyError):
+
+ # ignore empty tasks
+ if not task_handler:
+ continue
+
+ return {
+ 'path': path_dwim(basedir, task_handler[task_handler_key]),
+ 'type': child_type,
+ }
+
+ raise LookupError(
+ f'The node contains none of: {", ".join(task_include_keys)}',
+ )
+
+
+def _validate_task_handler_action_for_role(th_action: dict) -> 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(f"Failed to find required 'name' key in {module!s}")
+
+ if not isinstance(th_action['name'], str):
+ raise RuntimeError(
+ f"Value assigned to 'name' key on '{module!s}' is not a string.",
+ )
+
+
+def _roles_children(basedir: str, k, v, parent_type: FileType, main='main') -> list:
+ 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':
+ raise SystemExit('role dict {0} does not contain a "role" '
+ 'or "name" key'.format(role))
+ else:
+ results.extend(_look_for_role_files(basedir, role, main=main))
+ return results
+
+
+def _rolepath(basedir: str, role: str) -> Optional[str]:
+ 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 constants.DEFAULT_ROLES_PATH:
+ search_locations = constants.DEFAULT_ROLES_PATH
+ if isinstance(search_locations, str):
+ search_locations = search_locations.split(os.pathsep)
+ for loc in search_locations:
+ loc = os.path.expanduser(loc)
+ possible_paths.append(path_dwim(loc, role))
+
+ possible_paths.append(path_dwim(basedir, ''))
+
+ for path_option in possible_paths:
+ if os.path.isdir(path_option):
+ role_path = path_option
+ break
+
+ if role_path:
+ add_all_plugin_dirs(role_path)
+
+ return role_path
+
+
+def _look_for_role_files(basedir: str, role: str, main='main') -> list:
+ role_path = _rolepath(basedir, role)
+ if not role_path:
+ return []
+
+ results = []
+
+ for th in ['tasks', 'handlers', 'meta']:
+ current_path = os.path.join(role_path, th)
+ for dir, subdirs, files in os.walk(current_path):
+ for file in files:
+ file_ignorecase = file.lower()
+ if file_ignorecase.endswith(('.yml', '.yaml')):
+ thpath = os.path.join(dir, file)
+ results.append({'path': thpath, 'type': th})
+
+ return results
+
+
+def rolename(filepath):
+ idx = filepath.find('roles/')
+ if idx < 0:
+ return ''
+ role = filepath[idx + 6:]
+ role = role[:role.find('/')]
+ return role
+
+
+def _kv_to_dict(v):
+ (command, args, kwargs) = tokenize(v)
+ return dict(__ansible_module__=command, __ansible_arguments__=args, **kwargs)
+
+
+def _sanitize_task(task: dict) -> dict:
+ """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', FILENAME_KEY, LINE_NUMBER_KEY]:
+ if k in result:
+ del result[k]
+ return result
+
+
+# FIXME: drop noqa once this function is made simpler
+# Ref: https://github.com/ansible/ansible-lint/issues/744
+def normalize_task_v2(task: dict) -> dict: # noqa: C901
+ """Ensure tasks have an action key and strings are converted to python objects."""
+ result = dict()
+ if 'always_run' in task:
+ # FIXME(ssbarnea): Delayed import to avoid circular import
+ # See https://github.com/ansible/ansible-lint/issues/880
+ # noqa: # pylint:disable=cyclic-import,import-outside-toplevel
+ from ansiblelint.rules.AlwaysRunRule import AlwaysRunRule
+
+ raise MatchError(
+ rule=AlwaysRunRule,
+ filename=task[FILENAME_KEY],
+ linenumber=task[LINE_NUMBER_KEY])
+
+ sanitized_task = _sanitize_task(task)
+ mod_arg_parser = ModuleArgsParser(sanitized_task)
+ try:
+ action, arguments, result['delegate_to'] = mod_arg_parser.parse()
+ except AnsibleParserError as e:
+ try:
+ task_info = "%s:%s" % (task[FILENAME_KEY], task[LINE_NUMBER_KEY])
+ except KeyError:
+ task_info = "Unknown"
+ pp = pprint.PrettyPrinter(indent=2)
+ task_pprint = pp.pformat(sanitized_task)
+
+ _logger.critical("Couldn't parse task at %s (%s)\n%s", task_info, e.message, task_pprint)
+ raise SystemExit(ANSIBLE_FAILURE_RC)
+
+ # denormalize shell -> command conversion
+ if '_uses_shell' in arguments:
+ action = 'shell'
+ del arguments['_uses_shell']
+
+ for (k, v) in list(task.items()):
+ if k in ('action', 'local_action', 'args', 'delegate_to') or k == action:
+ # we don't want to re-assign these values, which were
+ # determined by the ModuleArgsParser() above
+ continue
+ else:
+ result[k] = v
+
+ result['action'] = dict(__ansible_module__=action)
+
+ if '_raw_params' in arguments:
+ result['action']['__ansible_arguments__'] = arguments['_raw_params'].split(' ')
+ del arguments['_raw_params']
+ else:
+ result['action']['__ansible_arguments__'] = list()
+
+ if 'argv' in arguments and not result['action']['__ansible_arguments__']:
+ result['action']['__ansible_arguments__'] = arguments['argv']
+ del arguments['argv']
+
+ result['action'].update(arguments)
+ return result
+
+
+# FIXME: drop noqa once this function is made simpler
+# Ref: https://github.com/ansible/ansible-lint/issues/744
+def normalize_task_v1(task): # noqa: C901
+ result = dict()
+ for (k, v) in task.items():
+ if k in VALID_KEYS or k.startswith('with_'):
+ if k == 'local_action' or k == 'action':
+ if not isinstance(v, dict):
+ v = _kv_to_dict(v)
+ v['__ansible_arguments__'] = v.get('__ansible_arguments__', list())
+ result['action'] = v
+ else:
+ result[k] = v
+ else:
+ if isinstance(v, str):
+ v = _kv_to_dict(k + ' ' + v)
+ elif not v:
+ v = dict(__ansible_module__=k)
+ else:
+ if isinstance(v, dict):
+ v.update(dict(__ansible_module__=k))
+ else:
+ if k == '__line__':
+ # Keep the line number stored
+ result[k] = v
+ continue
+
+ else:
+ # Tasks that include playbooks (rather than task files)
+ # can get here
+ # https://github.com/ansible/ansible-lint/issues/138
+ raise RuntimeError("Was not expecting value %s of type %s for key %s\n"
+ "Task: %s. Check the syntax of your playbook using "
+ "ansible-playbook --syntax-check" %
+ (str(v), type(v), k, str(task)))
+ v['__ansible_arguments__'] = v.get('__ansible_arguments__', list())
+ result['action'] = v
+ if 'module' in result['action']:
+ # this happens when a task uses
+ # local_action:
+ # module: ec2
+ # etc...
+ result['action']['__ansible_module__'] = result['action']['module']
+ del result['action']['module']
+ if 'args' in result:
+ result['action'].update(result.get('args'))
+ del result['args']
+ return result
+
+
+def normalize_task(task, filename):
+ 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):
+ name = task.get("name")
+ if name:
+ return name
+ action = task.get("action")
+ args = " ".join([u"{0}={1}".format(k, v) for (k, v) in action.items()
+ if k not in ["__ansible_module__", "__ansible_arguments__"]] +
+ action.get("__ansible_arguments__"))
+ return u"{0} {1}".format(action["__ansible_module__"], args)
+
+
+def extract_from_list(blocks, candidates):
+ results = list()
+ for block in blocks:
+ for candidate in candidates:
+ if isinstance(block, dict) and candidate in block:
+ if isinstance(block[candidate], list):
+ results.extend(add_action_type(block[candidate], candidate))
+ elif block[candidate] is not None:
+ raise RuntimeError(
+ "Key '%s' defined, but bad value: '%s'" %
+ (candidate, str(block[candidate])))
+ return results
+
+
+def add_action_type(actions, action_type):
+ results = list()
+ 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(yaml, file):
+ tasks = list()
+ if file['type'] in ['tasks', 'handlers']:
+ tasks = add_action_type(yaml, file['type'])
+ else:
+ tasks.extend(extract_from_list(yaml, ['tasks', 'handlers', 'pre_tasks', 'post_tasks']))
+
+ # Add sub-elements of block/rescue/always to tasks list
+ tasks.extend(extract_from_list(tasks, ['block', 'rescue', 'always']))
+ # Remove block/rescue/always elements from tasks list
+ block_rescue_always = ('block', 'rescue', 'always')
+ tasks[:] = [task for task in tasks if all(k not in task for k in block_rescue_always)]
+
+ return [task for task in tasks if
+ set(['include', 'include_tasks',
+ 'import_playbook', 'import_tasks']).isdisjoint(task.keys())]
+
+
+def get_normalized_tasks(yaml, file):
+ tasks = get_action_tasks(yaml, file)
+ res = []
+ for task in tasks:
+ # An empty `tags` block causes `None` to be returned if
+ # the `or []` is not present - `task.get('tags', [])`
+ # does not suffice.
+ if 'skip_ansible_lint' in (task.get('tags') or []):
+ # No need to normalize_task is we are skipping it.
+ continue
+ res.append(normalize_task(task, file['path']))
+
+ return res
+
+
+@lru_cache(maxsize=128)
+def parse_yaml_linenumbers(data, filename):
+ """Parse yaml as ansible.utils.parse_yaml but with linenumbers.
+
+ The line numbers are stored in each node's LINE_NUMBER_KEY key.
+ """
+ def compose_node(parent, index):
+ # the line number where the previous token has ended (plus empty lines)
+ line = loader.line
+ node = Composer.compose_node(loader, parent, index)
+ node.__line__ = line + 1
+ return node
+
+ def construct_mapping(node, deep=False):
+ 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
+ mapping[FILENAME_KEY] = filename
+ return mapping
+
+ try:
+ kwargs = {}
+ if 'vault_password' in inspect.getfullargspec(AnsibleLoader.__init__).args:
+ kwargs['vault_password'] = DEFAULT_VAULT_PASSWORD
+ loader = AnsibleLoader(data, **kwargs)
+ loader.compose_node = compose_node
+ loader.construct_mapping = construct_mapping
+ data = loader.get_single_data()
+ except (yaml.parser.ParserError, yaml.scanner.ScannerError) as e:
+ raise SystemExit("Failed to parse YAML in %s: %s" % (filename, str(e)))
+ return data
+
+
+def get_first_cmd_arg(task):
+ try:
+ if 'cmd' in task['action']:
+ first_cmd_arg = task['action']['cmd'].split()[0]
+ else:
+ first_cmd_arg = task['action']['__ansible_arguments__'][0]
+ except IndexError:
+ return None
+ return first_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 e:
+ _logger.warning(
+ "Failed to load %s with %s, assuming is not a playbook.",
+ filename, e)
+ else:
+ if (
+ isinstance(f, AnsibleSequence) and
+ hasattr(f, 'keys') and
+ playbooks_keys.intersection(next(iter(f), {}).keys())
+ ):
+ return True
+ return False
+
+
+def get_yaml_files(options: Namespace) -> dict:
+ """Find all yaml files."""
+ # git is preferred as it also considers .gitignore
+ git_command = ['git', 'ls-files', '*.yaml', '*.yml']
+ _logger.info("Discovering files to lint: %s", ' '.join(git_command))
+
+ out = None
+
+ try:
+ out = subprocess.check_output(
+ git_command,
+ stderr=subprocess.STDOUT,
+ universal_newlines=True
+ ).split()
+ except subprocess.CalledProcessError as exc:
+ _logger.warning(
+ "Failed to discover yaml files to lint using git: %s",
+ exc.output.rstrip('\n')
+ )
+ except FileNotFoundError as exc:
+ if options.verbosity:
+ _logger.warning(
+ "Failed to locate command: %s", exc
+ )
+
+ if out is None:
+ out = [
+ os.path.join(root, name)
+ for root, dirs, files in os.walk('.')
+ for name in files
+ if name.endswith('.yaml') or name.endswith('.yml')
+ ]
+
+ return OrderedDict.fromkeys(sorted(out))
+
+
+# FIXME: drop noqa once this function is made simpler
+# Ref: https://github.com/ansible/ansible-lint/issues/744
+def get_playbooks_and_roles(options=None) -> List[str]: # noqa: C901
+ """Find roles and playbooks."""
+ if options is None:
+ options = {}
+
+ files = get_yaml_files(options)
+
+ playbooks = []
+ role_dirs = []
+ role_internals = {
+ 'defaults',
+ 'files',
+ 'handlers',
+ 'meta',
+ 'tasks',
+ 'templates',
+ 'vars',
+ }
+
+ # detect role in repository root:
+ if 'tasks/main.yml' in files or 'tasks/main.yaml' in files:
+ role_dirs.append('.')
+
+ for p in map(Path, files):
+
+ try:
+ for file_path in options.exclude_paths:
+ if str(p.resolve()).startswith(str(file_path)):
+ raise FileNotFoundError(
+ f'File {file_path} matched exclusion entry: {p}')
+ except FileNotFoundError as e:
+ _logger.debug('Ignored %s due to: %s', p, e)
+ continue
+
+ if (next((i for i in p.parts if i.endswith('playbooks')), None) or
+ 'playbook' in p.parts[-1]):
+ playbooks.append(normpath(p))
+ continue
+
+ # ignore if any folder ends with _vars
+ if next((i for i in p.parts if i.endswith('_vars')), None):
+ continue
+ elif 'roles' in p.parts or '.' in role_dirs:
+ if 'tasks' in p.parts and p.parts[-1] in ['main.yaml', 'main.yml']:
+ role_dirs.append(str(p.parents[1]))
+ continue
+ elif role_internals.intersection(p.parts):
+ continue
+ elif 'tests' in p.parts:
+ playbooks.append(normpath(p))
+ if 'molecule' in p.parts:
+ if p.parts[-1] != 'molecule.yml':
+ playbooks.append(normpath(p))
+ continue
+ # hidden files are clearly not playbooks, likely config files.
+ if p.parts[-1].startswith('.'):
+ continue
+
+ if is_playbook(str(p)):
+ playbooks.append(normpath(p))
+ continue
+
+ _logger.info('Unknown file type: %s', normpath(p))
+
+ _logger.info('Found roles: %s', ' '.join(role_dirs))
+ _logger.info('Found playbooks: %s', ' '.join(playbooks))
+
+ return role_dirs + playbooks
+
+
+def expand_path_vars(path: str) -> str:
+ """Expand the environment or ~ variables in a path string."""
+ # It may be possible for function to be called with a Path object
+ path = str(path).strip()
+ path = os.path.expanduser(path)
+ path = os.path.expandvars(path)
+ return path
+
+
+def expand_paths_vars(paths: List[str]) -> List[str]:
+ """Expand the environment or ~ variables in a list."""
+ paths = [expand_path_vars(p) for p in paths]
+ return paths
+
+
+def get_rules_dirs(rulesdir: List[str], use_default: bool) -> List[str]:
+ """Return a list of rules dirs."""
+ default_ruledirs = [DEFAULT_RULESDIR]
+ default_custom_rulesdir = os.environ.get(
+ CUSTOM_RULESDIR_ENVVAR, os.path.join(DEFAULT_RULESDIR, "custom")
+ )
+ custom_ruledirs = sorted(
+ str(rdir.resolve())
+ for rdir in Path(default_custom_rulesdir).iterdir()
+ if rdir.is_dir() and (rdir / "__init__.py").exists()
+ )
+ if use_default:
+ return rulesdir + custom_ruledirs + default_ruledirs
+
+ return rulesdir or custom_ruledirs + default_ruledirs