devx/src/devx/tools/check_ansible_patterns.py

"""Check Ansible tasks for dangerous patterns that mask failures.

This check addresses the gap identified in the testing-strategy audit:
the automated PR review only checks Python files, and ``ansible-lint``
runs at ``profile: basic`` which does not catch dangerous patterns like:

- ``|| true`` on tasks that are NOT cleanup/idempotency operations
- ``failed_when: false`` on critical tasks (e.g. DB operations)
- ``2>/dev/null`` on tasks where stderr contains important diagnostics

Most ``|| true`` and ``2>/dev/null`` instances in the codebase are
legitimate (container removal, journalctl, apt-get, docker prune, SUID
removal). This check flags only instances that are NOT in a known-safe
context. Tasks can also opt out with a ``# lint:allow-failure-masking``
comment.

Usage::

    python -m devx.tools.check_ansible_patterns
    python -m devx.tools.check_ansible_patterns --path ansible/roles/app_container/tasks/main.yml

Exit code 0 if no violations found, 1 otherwise.
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

import click
import yaml

REPO_ROOT = Path.cwd()
DEFAULT_ANSIBLE_DIRS: list[Path] = [
    REPO_ROOT / "ansible" / "playbooks",
    REPO_ROOT / "ansible" / "roles",
]

# Comment marker to explicitly allow a pattern on a specific task
ALLOW_MARKER = "lint:allow-failure-masking"

# Patterns that mask failures when used in shell/command tasks
OR_TRUE_PATTERN = re.compile(r"\|\|\s*true\b", re.IGNORECASE)
REDIRECT_DEVNULL_PATTERN = re.compile(r"2>/dev/null")

# Module keys that accept shell/command strings
SHELL_MODULE_KEYS = frozenset(
    {
        "shell",
        "command",
        "ansible.builtin.shell",
        "ansible.builtin.command",
        "cmd",
        "ansible.builtin.raw",
        "raw",
    }
)

# Task keys whose values might contain shell commands
COMMAND_VALUE_KEYS = frozenset(
    {
        "shell",
        "command",
        "ansible.builtin.shell",
        "ansible.builtin.command",
        "cmd",
        "raw",
        "ansible.builtin.raw",
    }
)

# Legitimate contexts where || true or 2>/dev/null are safe.
# These are command prefixes or task names that indicate cleanup/idempotency.
LEGITIMATE_COMMAND_PREFIXES = (
    # Container/process removal (may not exist)
    "docker rm",
    "docker stop",
    "docker rmi",
    "docker network rm",
    "docker volume rm",
    "pkill",
    "kill",
    # Cleanup commands that are expected to sometimes fail
    "journalctl --vacuum",
    "apt-get clean",
    "apt-get autoremove",
    "docker image prune",
    "docker container prune",
    "docker volume prune",
    "docker builder prune",
    "find / -name",
    # SUID removal (binaries may not exist)
    "chmod",
    "rm -f",
    # Network connection checks (may fail if not connected)
    "docker network connect",
    # Prometheus snapshot API (may fail if no snapshot)
    "curl.*api/v2/admin/tsdb/snapshot",
)

LEGITIMATE_TASK_NAME_KEYWORDS = (
    "remove",
    "cleanup",
    "clean up",
    "prune",
    "purge",
    "disconnect",
    "stop",
    "kill",
    "strip suid",
    "suid",
    "vacuum",
    "ensure.*absent",
    "may not exist",
    "if exists",
    "optional",
    "best effort",
    "no-op",
    "noop",
    "idempotent",
    "sync",
)

# Tasks with failed_when: false that are critical and should not mask failures.
# Only flag operations that SHOULD fail loudly — writing secrets, provisioning
# users, creating OIDC apps. Do NOT flag stop/start/check/wait/migrate/restore
# operations where failed_when: false is legitimate (container may not exist,
# may already be stopped, etc.).
CRITICAL_TASK_KEYWORDS = (
    "password",
    "secret",
    "provision",
    "oidc",
)

# Task name keywords that indicate failed_when: false is legitimate
LEGITIMATE_FAILED_WHEN_KEYWORDS = (
    "stop",
    "start",
    "check",
    "wait",
    "migrate",
    "restart",
    "rebuild",
    "restore",
    "remove",
    "cleanup",
    "sync",
    "download",
    "extract",
    "verify",
)


def _is_legitimate_or_true(command_str: str, task_name: str) -> bool:
    """Check if a || true in a command is in a legitimate context."""
    # Check task name for legitimate keywords
    name_lower = task_name.lower()
    if any(re.search(kw, name_lower) for kw in LEGITIMATE_TASK_NAME_KEYWORDS):
        return True

    # Check command prefix for legitimate patterns
    cmd_lower = command_str.lower()
    return any(re.search(prefix, cmd_lower) for prefix in LEGITIMATE_COMMAND_PREFIXES)


def _is_legitimate_devnull(command_str: str, task_name: str) -> bool:
    """Check if a 2>/dev/null in a command is in a legitimate context."""
    # 2>/dev/null is almost always safe — it suppresses stderr noise.
    # Only flag it if the task is critical (DB, backup, OIDC) AND
    # there's no || true (which is the more dangerous pattern).
    return _is_legitimate_or_true(command_str, task_name)


def _check_task(task: dict, filepath: Path, task_num: int, repo_root: Path) -> list[str]:
    """Check a single task for dangerous failure-masking patterns."""
    violations: list[str] = []

    try:
        display_path = filepath.relative_to(repo_root)
    except ValueError:
        display_path = filepath

    task_name = task.get("name", "<unnamed>")

    # Check for the allow marker in the task name
    # (YAML comments are not preserved by safe_load, so we check the
    # task name for the marker as a workaround)
    if ALLOW_MARKER in task_name:
        return violations

    # Check for || true in command/shell values
    for key in COMMAND_VALUE_KEYS:
        value = task.get(key)
        if value is None:
            continue
        value_str = str(value)
        if OR_TRUE_PATTERN.search(value_str) and not _is_legitimate_or_true(value_str, task_name):
            violations.append(
                f"{display_path}:{task_num} — task '{task_name}' uses "
                f"'|| true' in {key} which may mask real failures. "
                f"If this is a cleanup/idempotency operation, rename the "
                f"task to include 'remove'/'cleanup'/'prune' or add "
                f"#{ALLOW_MARKER} to the task."
            )

    # Check for failed_when: false on critical tasks
    failed_when = task.get("failed_when")
    if failed_when is False:
        name_lower = task_name.lower()
        # Skip if the task name indicates a legitimate failed_when: false context
        is_legitimate = any(kw in name_lower for kw in LEGITIMATE_FAILED_WHEN_KEYWORDS)
        if not is_legitimate:
            for kw in CRITICAL_TASK_KEYWORDS:
                if kw in name_lower:
                    violations.append(
                        f"{display_path}:{task_num} — critical task '{task_name}' "
                        f"has failed_when: false, which masks failures on "
                        f"a {kw}-related operation. Remove failed_when: false "
                        f"or add #{ALLOW_MARKER} if masking is intentional."
                    )
                    break

    return violations


def _check_file(filepath: Path, repo_root: Path) -> list[str]:
    """Check a YAML file for dangerous failure-masking patterns."""
    try:
        content = filepath.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError):
        return []

    # Quick check: if no patterns appear, skip
    if not (
        OR_TRUE_PATTERN.search(content) or "failed_when: false" in content or REDIRECT_DEVNULL_PATTERN.search(content)
    ):
        return []

    # Check for allow markers in comments
    has_allow_marker = ALLOW_MARKER in content

    try:
        docs = list(yaml.safe_load_all(content))
    except yaml.YAMLError:
        return []

    violations: list[str] = []

    for doc in docs:
        if not doc:
            continue
        if isinstance(doc, list):
            for i, item in enumerate(doc):
                if isinstance(item, dict):
                    if any(k in item for k in ("tasks", "pre_tasks", "post_tasks", "handlers")):
                        _check_tasks(item, filepath, violations, repo_root)
                    else:
                        violations.extend(_check_task(item, filepath, i + 1, repo_root))
                        block = item.get("block")
                        if isinstance(block, list):
                            for j, bt in enumerate(block):
                                if isinstance(bt, dict):
                                    violations.extend(_check_task(bt, filepath, i + j + 1, repo_root))
        elif isinstance(doc, dict):
            _check_tasks(doc, filepath, violations, repo_root)

    # Filter out violations if the allow marker is present in the file
    # (coarse-grained opt-out for files with many legitimate uses)
    if has_allow_marker:
        violations = []

    return violations


def _check_tasks(doc: dict, filepath: Path, errors: list[str], repo_root: Path) -> None:
    """Check top-level tasks and nested task sections in a playbook doc."""
    for section_key in ("tasks", "pre_tasks", "post_tasks", "handlers"):
        section = doc.get(section_key)
        if isinstance(section, list):
            for i, task in enumerate(section):
                if isinstance(task, dict):
                    errors.extend(_check_task(task, filepath, i + 1, repo_root))
                    block = task.get("block")
                    if isinstance(block, list):
                        for j, bt in enumerate(block):
                            if isinstance(bt, dict):
                                errors.extend(_check_task(bt, filepath, i + j + 1, repo_root))


def _find_task_files(base: Path) -> list[Path]:
    """Find all YAML task files under a base directory, skipping molecule."""
    if base.is_file() and base.suffix in (".yml", ".yaml"):
        return [base]
    if not base.is_dir():
        return []
    files: list[Path] = []
    for f in sorted(base.rglob("*.yml")) + sorted(base.rglob("*.yaml")):
        if "molecule" in f.parts:
            continue
        files.append(f)
    return files


@click.command()
@click.option(
    "--path",
    type=click.Path(exists=True, path_type=Path),
    help="Check a specific file or directory (default: ansible/playbooks + ansible/roles).",
)
@click.option(
    "--ansible-dir",
    "ansible_dirs",
    type=click.Path(exists=True, path_type=Path),
    multiple=True,
    default=None,
    help="Override the default ansible directories (can be repeated). Defaults to ansible/playbooks and ansible/roles.",
)
def main(path: Path | None, ansible_dirs: tuple[Path, ...]) -> None:
    """Check Ansible tasks for dangerous failure-masking patterns."""
    dirs = list(ansible_dirs) if ansible_dirs else DEFAULT_ANSIBLE_DIRS
    if path:
        files = _find_task_files(path)
    else:
        files: list[Path] = []
        for d in dirs:
            files.extend(_find_task_files(d))

    all_violations: list[str] = []
    for f in files:
        all_violations.extend(_check_file(f, REPO_ROOT))

    if all_violations:
        click.echo("[check-ansible-patterns] FAIL: dangerous failure-masking patterns found:")
        for v in all_violations:
            click.echo(f"  - {v}")
        click.echo(f"\nTotal: {len(all_violations)} violation(s).")
        sys.exit(1)
    else:
        click.echo("[check-ansible-patterns] OK: no dangerous failure-masking patterns.")


if __name__ == "__main__":  # pragma: no cover
    main()