ansible-lint: Rework validate-all-file into custom rules

Test roles/ subdirectories as well (used by Zuul)

In addition to that container_exec_become will also
fail when "docker exec" or "podman exec" will be used
instead of "{{ kolla_container_engine }} exec"

Generated-By: Claude

Change-Id: I75c1e48c36bb0284e2927b1cb354ce9ec47967d7
Signed-off-by: Michal Nasiadka <mnasiadka@gmail.com>

1 files changed, 271 insertions, 0 deletions

diff --git a/.ansible-lint-rules/container_exec_become.py b/.ansible-lint-rules/container_exec_become.py
new file mode 100644
index 000000000..00d420250
--- /dev/null
+++ b/.ansible-lint-rules/container_exec_become.py
@@ -0,0 +1,271 @@
+# Copyright 2026 Michal Nasiadka
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Custom ansible-lint rule to enforce become on container exec commands."""
+
+from typing import Any
+
+from ansiblelint.errors import MatchError
+from ansiblelint.file_utils import Lintable
+from ansiblelint.rules import AnsibleLintRule
+
+
+class ContainerExecBecomeRule(AnsibleLintRule):
+
+    id = "container-exec-become"
+    description = """shell/command tasks starting with 'docker exec' or
+                     'podman exec' should have become set to True"""
+    severity = "MEDIUM"
+    tags = ["security", "container"]
+    version_added = "v6.0.0"
+    version_changed = "v6.0.0"
+
+    def matchyaml(self, file: Lintable) -> list[MatchError]:
+        """Check if container exec commands have become set to True.
+
+        Args:
+            file: The file being linted
+
+        Returns:
+            List of MatchError objects if violations found
+        """
+        errors = []
+
+        # Get the parsed YAML data
+        data = file.data
+        if not data:
+            return errors
+
+        # Process plays or task files
+        if isinstance(data, list):
+            # Check if this is a playbook (contains plays) or
+            # a task file (just tasks)
+            if data and isinstance(data[0], dict):
+                # If first item has 'hosts' or 'import_playbook',
+                # it's a playbook
+                first_item = data[0]
+                if 'hosts' in first_item or 'import_playbook' in first_item:
+                    # Process as playbook
+                    for play_idx, play in enumerate(data):
+                        if not isinstance(play, dict):
+                            continue
+
+                        # Check tasks at play level
+                        self._check_tasks(play.get('tasks', []), None, errors,
+                                          file, f".[{play_idx}].tasks")
+
+                        # Check pre_tasks
+                        self._check_tasks(play.get('pre_tasks', []), None,
+                                          errors, file,
+                                          f".[{play_idx}].pre_tasks")
+
+                        # Check post_tasks
+                        self._check_tasks(play.get('post_tasks', []), None,
+                                          errors, file,
+                                          f".[{play_idx}].post_tasks")
+
+                        # Check handlers
+                        self._check_tasks(play.get('handlers', []), None,
+                                          errors, file,
+                                          f".[{play_idx}].handlers")
+                else:
+                    # Process as task file (list of tasks directly)
+                    self._check_tasks(data, None, errors, file, ".")
+
+        return errors
+
+    def _check_tasks(
+        self,
+        tasks: list[dict[str, Any]],
+        parent_become: bool | None,
+        errors: list[MatchError],
+        file: Lintable,
+        path: str
+    ) -> None:
+        """Recursively check tasks and blocks.
+
+           Check for container exec commands with become.
+
+        Args:
+            tasks: List of tasks to check
+            parent_become: The become value inherited from parent block
+            errors: List to append errors to
+            file: The file being linted
+            path: Current path in the YAML structure
+        """
+        if not tasks:
+            return
+
+        for task_idx, task in enumerate(tasks):
+            if not isinstance(task, dict):
+                continue
+
+            task_path = f"{path}[{task_idx}]"
+
+            # Determine effective become value for this task
+            task_become = task.get('become')
+            if task_become is not None:
+                effective_become = task_become
+            else:
+                effective_become = parent_become
+
+            # Check if this is a block
+            if 'block' in task:
+                # For blocks, pass down the effective become value
+                block_become = task.get('become')
+                if block_become is not None:
+                    inherited_become = block_become
+                else:
+                    inherited_become = parent_become
+
+                # Recursively check tasks in the block
+                self._check_tasks(
+                    task.get('block', []),
+                    inherited_become,
+                    errors,
+                    file,
+                    f"{task_path}.block"
+                )
+
+                # Also check rescue and always sections
+                if 'rescue' in task:
+                    self._check_tasks(
+                        task.get('rescue', []),
+                        inherited_become,
+                        errors,
+                        file,
+                        f"{task_path}.rescue"
+                    )
+
+                if 'always' in task:
+                    self._check_tasks(
+                        task.get('always', []),
+                        inherited_become,
+                        errors,
+                        file,
+                        f"{task_path}.always"
+                    )
+
+                continue
+
+            # Check if this task uses shell or command module
+            module_name = None
+            command_value = None
+
+            # Check for shell/command modules
+            shell_modules = [
+                'ansible.builtin.shell', 'shell',
+                'ansible.builtin.command', 'command'
+            ]
+
+            for shell_module in shell_modules:
+                if shell_module in task:
+                    module_name = shell_module
+                    # Get the command - can be a string or dict
+                    cmd = task[shell_module]
+                    if isinstance(cmd, str):
+                        command_value = cmd
+                    elif isinstance(cmd, dict):
+                        # Command might be in 'cmd' key or the module args
+                        command_value = (cmd.get('cmd') or
+                                         cmd.get('_raw_params'))
+                    break
+
+            # If this is a shell/command task, check if it starts with
+            # docker/podman exec
+            if module_name and command_value:
+                # Strip leading whitespace and newlines, check if it starts
+                # with docker/podman exec
+                cmd_stripped = command_value.strip()
+
+                # Check for hardcoded docker/podman exec (should use variable)
+                uses_hardcoded_exec = (
+                    cmd_stripped.startswith('docker exec') or
+                    cmd_stripped.startswith('podman exec')
+                )
+
+                if uses_hardcoded_exec:
+                    # Flag hardcoded container engine usage
+                    task_name = task.get('name', 'unnamed task')
+                    if cmd_stripped.startswith('docker exec'):
+                        exec_type = 'docker exec'
+                    else:
+                        exec_type = 'podman exec'
+
+                    message = (f"Task '{task_name}' uses hardcoded "
+                               f"'{exec_type}' - use "
+                               f"'{{{{ kolla_container_engine }}}} exec' "
+                               f"instead")
+
+                    errors.append(
+                        self.create_matcherror(
+                            message=message,
+                            filename=file,
+                            lineno=task.get('__line__', 1)
+                        )
+                    )
+                    # Continue to also check become requirement
+
+                # Check for variable-based container engine followed by exec
+                # e.g., "{{ kolla_container_engine }} exec",
+                # "{{ container_engine }} exec"
+                import re
+                variable_exec_pattern = r'^\{\{[^}]+\}\}\s+exec\s+'
+                uses_variable_exec = re.match(variable_exec_pattern,
+                                              cmd_stripped)
+
+                is_container_exec = uses_hardcoded_exec or uses_variable_exec
+
+                if is_container_exec:
+                    # Check if become is True
+                    if effective_become is not True:
+                        task_name = task.get('name', 'unnamed task')
+
+                        # Determine exec type for error message
+                        if cmd_stripped.startswith('docker exec'):
+                            exec_type = 'docker exec'
+                        elif cmd_stripped.startswith('podman exec'):
+                            exec_type = 'podman exec'
+                        else:
+                            exec_type = 'container exec (via variable)'
+
+                        if task_become is False:
+                            message = (f"Task '{task_name}' uses "
+                                       f"'{exec_type}' but has "
+                                       f"'become: false'")
+                            if parent_become is True:
+                                message += """ which overrides block-level
+                                                'become: true'"""
+                        elif task_become is None and parent_become is False:
+                            message = (f"Task '{task_name}' uses "
+                                       f"'{exec_type}' and inherits "
+                                       f"'become: false' from block")
+                        elif task_become is None and parent_become is None:
+                            message = (f"Task '{task_name}' uses "
+                                       f"'{exec_type}' but is missing "
+                                       f"'become: true' at task or "
+                                       f"block level")
+                        else:
+                            message = (f"Task '{task_name}' uses "
+                                       f"'{exec_type}' but has "
+                                       f"'become: {task_become}' instead of "
+                                       f"'true'")
+
+                        errors.append(
+                            self.create_matcherror(
+                                message=message,
+                                filename=file,
+                                lineno=task.get('__line__', 1)
+                            )
+                        )