devx/src/devx/tools/check_ansible_no_log.py

"""Check Ansible tasks for missing no_log on secret-handling tasks.

ansible-lint's built-in ``no-log-password`` rule only fires when a module
parameter is literally named ``*password*`` and there's a loop. It does
NOT catch:

- Shell/command tasks that interpolate ``{{ _secrets.* }}`` or
  ``{{ *password* }}`` variables
- Template/copy tasks that render secret values without ``no_log``

This script fills that gap by scanning all Ansible task files for
variables that look like secrets (``_secrets.*``, ``*password*``,
``*secret*``, ``*token*``, ``*api_key*``) and verifying that the task
has ``no_log`` set to a non-False value.

Usage::

    python -m devx.tools.check_ansible_no_log
    python -m devx.tools.check_ansible_no_log --path ansible/roles/my_role
    python -m devx.tools.check_ansible_no_log --ansible-dir ansible/roles

Exit code 0 if all secret-handling tasks have no_log, 1 otherwise.
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

import click
import yaml

REPO_ROOT = Path.cwd()
DEFAULT_ANSIBLE_DIR = REPO_ROOT / "ansible"

# Patterns that indicate a task is handling secrets.
# We only match Jinja-interpolated variables ({{ ... }}) to avoid false
# positives from field names like "password" in module params or task names.
SECRET_PATTERNS = [
    # {{ _secrets.anything }} or {{ _secrets['anything'] }}
    re.compile(r"\{\{[^}]*_secrets\.", re.IGNORECASE),
    # {{ anything_password }} but NOT the word "password" in a string literal
    re.compile(r"\{\{[^}]*password", re.IGNORECASE),
    # {{ anything_secret }}
    re.compile(r"\{\{[^}]*_secret\b", re.IGNORECASE),
    # {{ anything_api_key }}
    re.compile(r"\{\{[^}]*api_key", re.IGNORECASE),
    # {{ anything_token }} (but not loop tokens like {{ loop_token }})
    re.compile(r"\{\{[^}]*(?:vault_token|auth_token|access_token|bot_token)", re.IGNORECASE),
]

# Task keys whose values might contain secret references
TASK_VALUE_KEYS = {
    "shell",
    "command",
    "ansible.builtin.shell",
    "ansible.builtin.command",
    "ansible.builtin.template",
    "ansible.builtin.copy",
    "ansible.builtin.debug",
    "template",
    "copy",
    "debug",
    "cmd",
    "msg",
    "content",
}

# Keys that are NOT secret-bearing (task metadata, not values)
NON_VALUE_KEYS = {
    "name",
    "when",
    "loop",
    "loop_control",
    "changed_when",
    "failed_when",
    "no_log",
    "register",
    "tags",
    "vars",
    "become",
    "become_user",
    "delegate_to",
    "run_once",
    "environment",
    "with_items",
    "with_dict",
    "with_list",
}


def _contains_secret(value: object) -> bool:
    """Recursively check if a value contains secret-like variable references."""
    if isinstance(value, str):
        return any(p.search(value) for p in SECRET_PATTERNS)
    if isinstance(value, dict):
        return any(_contains_secret(v) for v in value.values())
    if isinstance(value, list):
        return any(_contains_secret(item) for item in value)
    return False


def _has_no_log(task: dict) -> bool:
    """Check if a task has no_log set to a non-False value."""
    no_log = task.get("no_log", False)
    # Jinja expressions (e.g. "{{ not debug_mode }}") count as set
    return no_log is not False and no_log is not None


def _check_task(task: dict, file_path: Path, task_num: int) -> list[str]:
    """Check a single task for missing no_log on secret values.

    Returns a list of violation messages (empty if OK).
    """
    violations: list[str] = []

    # Skip tasks that already have no_log
    if _has_no_log(task):
        return violations

    # Check all string values in the task for secret references
    has_secrets = False
    for key, value in task.items():
        if key in NON_VALUE_KEYS:
            continue
        # Check action module params (shell, command, copy, template, etc.)
        if _contains_secret(value):
            has_secrets = True
            break

    if has_secrets:
        task_name = task.get("name", "<unnamed>")
        violations.append(
            f"{file_path}:{task_num}: Task '{task_name}' references secrets "
            f"but has no no_log. Add `no_log: true` or "
            f'`no_log: "{{{{ not (debug_mode | default(false) | bool) }}}}"` '
            f"to prevent credential leakage in Ansible output."
        )

    return violations


def check_directory(ansible_dir: Path) -> list[str]:
    """Check all Ansible task files in a directory tree."""
    all_violations: list[str] = []

    # Find all task files
    task_files = list(ansible_dir.rglob("tasks/*.yml"))
    task_files += list(ansible_dir.rglob("tasks/*.yaml"))
    # Also check playbook files
    task_files += list(ansible_dir.glob("playbooks/*.yml"))

    for task_file in sorted(task_files):
        # Skip molecule test files
        if "molecule" in task_file.parts:
            continue

        try:
            with task_file.open() as f:
                docs = list(yaml.safe_load_all(f))
        except (yaml.YAMLError, OSError):
            continue

        for doc in docs:
            if not doc:
                continue

            # Task files are bare lists of tasks; playbook files are
            # lists of plays (each play is a dict with 'hosts' key)
            if isinstance(doc, list):
                is_plays = isinstance(doc[0], dict) and "hosts" in doc[0]
                if not is_plays:
                    for i, task in enumerate(doc):
                        if not isinstance(task, dict):
                            continue
                        all_violations.extend(_check_task(task, task_file, i + 1))
                    continue
                plays = doc
            elif isinstance(doc, dict):
                plays = [doc]
            else:
                continue

            for play in plays:
                if not isinstance(play, dict):
                    continue
                for task_section in ("tasks", "pre_tasks", "post_tasks", "handlers"):
                    tasks = play.get(task_section, [])
                    if not isinstance(tasks, list):
                        continue
                    for i, task in enumerate(tasks):
                        if not isinstance(task, dict):
                            continue
                        all_violations.extend(_check_task(task, task_file, i + 1))

    return all_violations


@click.command()
@click.option(
    "--path",
    type=click.Path(exists=True, path_type=Path),
    help="Check a specific file or directory (default: ansible/).",
)
@click.option(
    "--ansible-dir",
    type=click.Path(exists=True, path_type=Path),
    default=None,
    help="Override the default ansible directory (default: ansible/).",
)
def main(path: Path | None, ansible_dir: Path | None) -> None:
    """Check that Ansible tasks handling secrets have no_log set."""
    target = path or ansible_dir or DEFAULT_ANSIBLE_DIR
    if not target.is_dir():
        click.echo(f"Error: {target} is not a directory", err=True)
        sys.exit(2)

    violations = check_directory(target)

    if violations:
        click.echo(f"Found {len(violations)} task(s) handling secrets without no_log:\n")
        for v in violations:
            click.echo(f"  {v}")
        click.echo(f"\nTotal: {len(violations)} violation(s).")
        sys.exit(1)

    click.echo(f"[check-ansible-no-log] All secret-handling tasks have no_log. ({target})")


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