Creating Ansible Playbooks

---
# Ansible Playbook Template
# This template provides a starting point for creating Ansible playbooks.
# It includes common sections and best practice configurations.

- name: "example-value - Playbook Description"
  hosts: all  # Target hosts or groups (e.g., webservers, dbservers)
  become: true  # Enable privilege escalation (sudo)
  become_user: root # Specify the user to become (optional, defaults to root)
  gather_facts: true  # Gather facts about the target hosts

  # Define variables that can be used throughout the playbook
  vars:
    # Example variables
    app_name: "YOUR_APP_NAME"
    app_version: "1.0.0"
    install_dir: "/opt/{{ app_name }}"
    # Add more variables as needed

  # Pre-tasks: Tasks that run before any roles are applied
  pre_tasks:
    - name: "Update apt cache (Debian/Ubuntu)"
      apt:
        update_cache: yes
      when: ansible_os_family == "Debian"

    - name: "Update yum cache (RedHat/CentOS)"
      yum:
        update_cache: yes
      when: ansible_os_family == "RedHat"

  # Roles: Group of tasks to perform a specific function
  roles:
    - role: common  # Example role for common configurations
      # vars:  # Role-specific variables (optional)
      #   some_var: "YOUR_VALUE_HERE"

    # Add more roles as needed (e.g., webserver, database)
    # - role: webserver

  # Tasks: Individual steps to be executed
  tasks:
    - name: "Create installation directory"
      file:
        path: "{{ install_dir }}"
        state: directory
        owner: root
        group: root
        mode: "0755"

    - name: "Copy application files"
      copy:
        src: "files/{{ app_name }}"  # Path to application files on the control node
        dest: "{{ install_dir }}"
        owner: root
        group: root
        mode: "0644"

    # Add more tasks as needed

  # Post-tasks: Tasks that run after all roles and tasks have been applied
  post_tasks:
    - name: "Restart application service"
      service:
        name: "{{ app_name }}"
        state: restarted
      ignore_errors: true # Allows the playbook to continue even if the service restart fails

  # Handlers: Tasks that are triggered by other tasks
  handlers:
    - name: "Restart web server"
      service:
        name: apache2
        state: restarted
      listen: "Restart web server"  # Triggered by tasks that notify "Restart web server"

# validation_rules.yml

# --- General Playbook Structure Rules ---
playbook_structure:
  # Rule: Playbook must have a name
  name_required: true
  # Rule: Playbook must have at least one host
  hosts_required: true
  # Rule: Playbook should have a gather_facts setting (explicitly true or false)
  gather_facts_required: true
  gather_facts_default: true # Consider setting to false if facts are not needed for performance
  # Rule: Playbook should have a become setting (explicitly true or false) if privilege escalation is needed
  become_recommended: true # Recommend setting this, but don't enforce.
  become_default: false # Set to true if most tasks require sudo.

# --- Task Specific Rules ---
task_rules:
  # Rule: Each task must have a name
  name_required: true
  # Rule: Avoid using the 'shell' module unless necessary. Prefer specific modules.
  no_shell_unless_necessary: true
  shell_exceptions: # List of commands where shell is acceptable.  Helps reduce false positives.
    - "ls"
    - "grep"
    - "awk"
    - "sed"
  # Rule: Use 'changed_when' instead of relying on return codes for idempotency.
  changed_when_recommended: true
  # Rule: Use 'failed_when' to handle unexpected errors.
  failed_when_recommended: true

# --- Security Best Practices ---
security_rules:
  # Rule: Avoid storing secrets directly in playbooks. Use Ansible Vault or a secrets management system.
  no_plain_text_secrets: true
  secret_keywords: # List of keywords that indicate a potential secret
    - "password"
    - "secret"
    - "token"
    - "key"
  # Rule: Use 'become' with caution.  Limit its scope to only the tasks that require it.
  become_caution: true
  # Rule: Avoid using '*' in host patterns in production. Be specific.
  no_wildcard_hosts: true
  # Rule: Validate input parameters to prevent injection vulnerabilities.
  validate_input: true
  input_validation_regex: "example-value" # Example regex for validating input.  Should be customized per variable.

# --- Idempotency Rules ---
idempotency_rules:
  # Rule: Ensure tasks are idempotent. They should only make changes when necessary.
  idempotent_tasks: true
  # Rule: Use 'creates' or 'removes' in file/copy/template modules for idempotency.
  file_idempotency: true
  # Rule: Use 'state' parameter where applicable (e.g., present/absent for files/packages).
  state_parameter_required: true
  state_parameter_exceptions:  # Some modules don't use state, so exclude them
    - "debug"
    - "include_tasks"
    - "include_role"

# --- Error Handling Rules ---
error_handling_rules:
  # Rule: Implement proper error handling using 'rescue' and 'always' blocks.
  rescue_blocks_recommended: true
  always_blocks_recommended: true
  # Rule: Use 'ignore_errors' with caution.  Document why it is necessary.
  ignore_errors_caution: true

# --- Variable Usage Rules ---
variable_rules:
  # Rule: Use descriptive variable names.
  descriptive_variable_names: true
  # Rule: Define variables in a structured way (e.g., group_vars, host_vars).
  structured_variables: true
  # Rule: Avoid using hardcoded values directly in tasks. Use variables instead.
  no_hardcoded_values: true

# --- Module Specific Rules (Example for apt module) ---
apt_module_rules:
  # Rule: Ensure 'update_cache' is set to 'yes' when installing packages for the first time.
  update_cache_recommended: true
  # Rule: Specify a state (present/absent) when managing packages.
  state_required: true
  default_package: "YOUR_VALUE_HERE" # Example default package

# --- Platform Specific Rules ---
platform_rules:
  # Rule: Use conditional statements ('when') to handle platform-specific differences.
  conditional_platform_tasks: true
  supported_platforms: # List of supported platforms
    - "Ubuntu"
    - "CentOS"
    - "Windows"

#!/usr/bin/env python3
"""
Ansible playbook security scanner.

Scans playbook for security vulnerabilities including:
- Hardcoded credentials
- Insecure permissions
- Command injection risks
- Unsafe variable handling
- Missing authentication checks
"""

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
import yaml
import re


class PlaybookSecurityScanner:
    """Scans Ansible playbooks for security vulnerabilities."""

    # Security rules
    CREDENTIAL_PATTERNS = [
        r'password\s*[:=]\s*["\']?(?!{{)[^"\'\n]+["\']?',
        r'api_key\s*[:=]\s*["\']?[^"\'\n]+["\']?',
        r'secret\s*[:=]\s*["\']?[^"\'\n]+["\']?',
        r'token\s*[:=]\s*["\']?[^"\'\n]+["\']?',
        r'private_key\s*[:=]',
        r'aws_access_key',
        r'aws_secret_key',
    ]

    DANGEROUS_MODULES = {
        'shell': 'Shell module can lead to command injection',
        'command': 'Command module should use specific modules when possible',
        'raw': 'Raw module bypasses ansible modules',
        'lineinfile': 'Lineinfile can create configuration vulnerabilities',
    }

    UNSAFE_PATTERNS = {
        'become_pass': 'Hardcoded become password detected',
        'no_log: false': 'Logging of sensitive data enabled',
        'validate_certs: false': 'Certificate validation disabled',
        'verify_ssl: false': 'SSL verification disabled',
        'insecure: true': 'Insecure mode enabled',
    }

    def __init__(self):
        """Initialize scanner."""
        self.vulnerabilities = []
        self.warnings = []
        self.recommendations = []

    def scan_file(self, file_path: str) -> bool:
        """
        Scan playbook file for vulnerabilities.

        Args:
            file_path: Path to playbook file

        Returns:
            True if no vulnerabilities found
        """
        try:
            path = Path(file_path)
            if not path.exists():
                self.vulnerabilities.append(f"File not found: {file_path}")
                return False

            if path.suffix.lower() not in ['.yaml', '.yml']:
                self.vulnerabilities.append(f"Expected YAML file, got: {path.suffix}")
                return False

            # Scan file content as text first (for credentials)
            with open(file_path, 'r') as f:
                content = f.read()
                self._scan_raw_content(content)

            # Parse YAML and scan structure
            with open(file_path, 'r') as f:
                playbooks = yaml.safe_load(f)

            if isinstance(playbooks, list):
                for idx, play in enumerate(playbooks):
                    if isinstance(play, dict):
                        self._scan_play(play, idx)

            return len(self.vulnerabilities) == 0

        except yaml.YAMLError as e:
            self.vulnerabilities.append(f"YAML error: {str(e)}")
            return False
        except Exception as e:
            self.vulnerabilities.append(f"Scan error: {str(e)}")
            return False

    def _scan_raw_content(self, content: str) -> None:
        """Scan raw file content for credentials."""
        lines = content.split('\n')

        for idx, line in enumerate(lines, 1):
            # Skip comments
            if line.strip().startswith('#'):
                continue

            # Check for hardcoded credentials
            for pattern in self.CREDENTIAL_PATTERNS:
                if re.search(pattern, line, re.IGNORECASE):
                    # Verify it's not a variable reference or comment
                    if '{{' not in line and '}}' not in line:
                        self.vulnerabilities.append(
                            f"Line {idx}: Possible hardcoded credential: {line.strip()[:50]}"
                        )

            # Check for unsafe patterns
            for unsafe_pattern, description in self.UNSAFE_PATTERNS.items():
                if unsafe_pattern.lower() in line.lower():
                    self.warnings.append(f"Line {idx}: {description}")

    def _scan_play(self, play: Dict[str, Any], play_idx: int) -> None:
        """Scan individual play."""
        if 'tasks' in play:
            self._scan_tasks(play.get('tasks', []), f'Play {play_idx}')

        if 'handlers' in play:
            self._scan_tasks(play.get('handlers', []), f'Play {play_idx} handlers')

        if 'vars' in play:
            self._scan_vars(play.get('vars', {}), f'Play {play_idx}')

        if 'roles' in play:
            self._scan_roles(play.get('roles', []), f'Play {play_idx}')

    def _scan_tasks(self, tasks: List[Any], context: str) -> None:
        """Scan tasks for security issues."""
        if not isinstance(tasks, list):
            return

        for idx, task in enumerate(tasks):
            if not isinstance(task, dict):
                continue

            # Check for dangerous modules
            for module, issue in self.DANGEROUS_MODULES.items():
                if module in task:
                    self.warnings.append(
                        f"{context} task {idx} ({task.get('name', 'unnamed')}): {issue}"
                    )

                    # Check for unquoted shell commands
                    if module in ['shell', 'command']:
                        cmd = task.get(module, '')
                        if self._has_injection_risk(cmd):
                            self.vulnerabilities.append(
                                f"{context} task {idx}: Possible command injection in {module}"
                            )

            # Check for unsafe sudo usage
            if 'become' in task and task.get('become'):
                if 'become_pass' in task:
                    self.vulnerabilities.append(
                        f"{context} task {idx}: Hardcoded become password detected"
                    )
                if 'become_method' not in task:
                    self.warnings.append(
                        f"{context} task {idx}: become without explicit become_method"
                    )

            # Check for unsafe variable handling
            if 'shell' in task or 'command' in task:
                cmd = task.get('shell') or task.get('command', '')
                if '{{' in cmd and '|' in cmd:
                    if 'quote' not in cmd and 'escape' not in cmd:
                        self.warnings.append(
                            f"{context} task {idx}: Variable in {cmd.split()[0] if cmd else ''} may not be properly escaped"
                        )

            # Check for no_log
            if 'register' in task:
                if 'no_log' not in task:
                    if any(sensitive in task.get('name', '').lower()
                           for sensitive in ['password', 'secret', 'token', 'key']):
                        self.warnings.append(
                            f"{context} task {idx}: Sensitive output not protected with no_log"
                        )

    def _scan_vars(self, variables: Dict[str, Any], context: str) -> None:
        """Scan variables for security issues."""
        if not isinstance(variables, dict):
            return

        for key, value in variables.items():
            # Check for credentials in variable names
            if any(cred_term in key.lower() for cred_term in
                   ['password', 'secret', 'key', 'token', 'credential']):
                if isinstance(value, str) and value and not value.startswith('{{'):
                    self.vulnerabilities.append(
                        f"{context}: Hardcoded credential in variable '{key}'"
                    )

    def _scan_roles(self, roles: List[Any], context: str) -> None:
        """Scan role usage."""
        if not isinstance(roles, list):
            return

        for idx, role in enumerate(roles):
            if isinstance(role, dict):
                if 'vars' in role:
                    self._scan_vars(role['vars'], f'{context} role {idx}')

    def _has_injection_risk(self, command: str) -> bool:
        """Check if command has injection risk."""
        injection_patterns = [
            r'\$\{.*\}',  # ${variable}
            r'\$\(.*\)',  # $(command)
            r'`.*`',      # backticks
        ]

        for pattern in injection_patterns:
            if re.search(pattern, command):
                return True

        return False

    def generate_recommendations(self) -> None:
        """Generate security recommendations."""
        if len(self.vulnerabilities) > 0:
            self.recommendations.append(
                "Fix all identified vulnerabilities before deployment"
            )

        if any('password' in v.lower() for v in self.vulnerabilities):
            self.recommendations.append(
                "Use Ansible Vault or Secret Manager for sensitive data"
            )

        if any('shell' in w.lower() or 'command' in w.lower() for w in self.warnings):
            self.recommendations.append(
                "Replace shell/command with specific Ansible modules when possible"
            )

        if any('become' in w.lower() for w in self.warnings):
            self.recommendations.append(
                "Use Ansible Vault for privilege escalation passwords"
            )

        if not self.recommendations:
            self.recommendations.append(
                "Playbook appears to follow security best practices"
            )

    def get_report(self) -> Dict[str, Any]:
        """Get security scan report."""
        self.generate_recommendations()

        return {
            'secure': len(self.vulnerabilities) == 0,
            'vulnerabilities': self.vulnerabilities,
            'warnings': self.warnings,
            'recommendations': self.recommendations,
            'vulnerability_count': len(self.vulnerabilities),
            'warning_count': len(self.warnings),
        }


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description='Scan Ansible playbook for security vulnerabilities'
    )
    parser.add_argument(
        'playbook_file',
        help='Path to Ansible playbook file'
    )
    parser.add_argument(
        '-o', '--output',
        help='Save security report to JSON file'
    )
    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='Print detailed security report'
    )
    parser.add_argument(
        '--fail-on-warning',
        action='store_true',
        help='Exit with error if warnings found'
    )

    args = parser.parse_args()

    try:
        scanner = PlaybookSecurityScanner()
        is_secure = scanner.scan_file(args.playbook_file)
        report = scanner.get_report()

        # Check fail-on-warning
        if args.fail_on_warning and report['warning_count'] > 0:
            is_secure = False

        # Output report
        if args.verbose or not is_secure:
            print(json.dumps(report, indent=2))
        else:
            print(f"Security scan complete: {len(report['vulnerabilities'])} vulnerabilities, "
                  f"{len(report['warnings'])} warnings")

        # Save report if requested
        if args.output:
            with open(args.output, 'w') as f:
                json.dump(report, f, indent=2)
            print(f"Security report saved to: {args.output}")

        sys.exit(0 if is_secure else 1)

    except Exception as e:
        print(f"Error: {str(e)}", file=sys.stderr)
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/bin/bash
#
# Ansible playbook test script
#
# Executes playbook in test environment (container) including:
# - Container setup
# - Playbook execution
# - Result verification
# - Cleanup
#

set -euo pipefail

# Color output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color

# Configuration
PLAYBOOK_FILE=""
CONTAINER_IMAGE="ubuntu:22.04"
CONTAINER_NAME="ansible-test-$$"
INVENTORY="localhost"
EXTRA_VARS=""
TAGS=""
SKIP_TAGS=""
KEEP_CONTAINER=false
VERBOSE=false

usage() {
    cat << EOF
Usage: $0 [OPTIONS]

Execute Ansible playbook in test container environment

OPTIONS:
    -p, --playbook FILE       Path to playbook file [REQUIRED]
    -i, --image IMAGE         Docker image to use (default: ubuntu:22.04)
    -l, --inventory HOSTS     Inventory or hosts (default: localhost)
    -e, --extra-vars VARS     Extra variables (JSON format)
    -t, --tags TAGS           Only run tasks with these tags
    --skip-tags TAGS          Skip tasks with these tags
    -k, --keep                Keep container after test
    -v, --verbose             Verbose output
    -h, --help                Show this help message

EXAMPLES:
    $0 --playbook site.yml
    $0 -p playbook.yml -i hosts.ini -v
    $0 -p playbook.yml --tags deployment

EOF
    exit 1
}

# Parse arguments
while [[ $# -gt 0 ]]; do
    case $1 in
        -p|--playbook)
            PLAYBOOK_FILE="$2"
            shift 2
            ;;
        -i|--image)
            CONTAINER_IMAGE="$2"
            shift 2
            ;;
        -l|--inventory)
            INVENTORY="$2"
            shift 2
            ;;
        -e|--extra-vars)
            EXTRA_VARS="$2"
            shift 2
            ;;
        -t|--tags)
            TAGS="$2"
            shift 2
            ;;
        --skip-tags)
            SKIP_TAGS="$2"
            shift 2
            ;;
        -k|--keep)
            KEEP_CONTAINER=true
            shift
            ;;
        -v|--verbose)
            VERBOSE=true
            shift
            ;;
        -h|--help)
            usage
            ;;
        *)
            echo "Unknown option: $1"
            usage
            ;;
    esac
done

# Validate required arguments
if [[ -z "$PLAYBOOK_FILE" ]]; then
    echo -e "${RED}Error: --playbook argument is required${NC}"
    usage
fi

if [[ ! -f "$PLAYBOOK_FILE" ]]; then
    echo -e "${RED}Error: File not found: $PLAYBOOK_FILE${NC}"
    exit 1
fi

# Function: Log message
log() {
    echo -e "${BLUE}[$(date +'%Y-%m-%d %H:%M:%S')]${NC} $1"
}

# Function: Success message
success() {
    echo -e "${GREEN}✓ $1${NC}"
}

# Function: Error message
error() {
    echo -e "${RED}✗ $1${NC}"
}

# Function: Warning message
warning() {
    echo -e "${YELLOW}⚠ $1${NC}"
}

# Function: Check dependencies
check_dependencies() {
    local missing=false

    if ! command -v docker &> /dev/null; then
        error "docker not found in PATH"
        missing=true
    fi

    if ! command -v ansible-playbook &> /dev/null; then
        error "ansible-playbook not found in PATH"
        missing=true
    fi

    if [[ "$missing" == true ]]; then
        exit 1
    fi

    success "All dependencies found"
}

# Function: Cleanup container
cleanup_container() {
    if [[ "$KEEP_CONTAINER" == true ]]; then
        log "Container kept for inspection: $CONTAINER_NAME"
        return 0
    fi

    log "Cleaning up container: $CONTAINER_NAME"
    docker rm -f "$CONTAINER_NAME" &>/dev/null || true
}

# Function: Setup container
setup_container() {
    log "Setting up test container: $CONTAINER_NAME"

    # Check if image exists
    if ! docker images --format "{{.Repository}}:{{.Tag}}" | grep -q "^${CONTAINER_IMAGE}$"; then
        log "Pulling Docker image: $CONTAINER_IMAGE"
        if ! docker pull "$CONTAINER_IMAGE"; then
            error "Failed to pull Docker image: $CONTAINER_IMAGE"
            return 1
        fi
    fi

    # Create container
    log "Creating container from image: $CONTAINER_IMAGE"
    if ! docker create \
        --name "$CONTAINER_NAME" \
        --hostname test-host \
        --entrypoint /bin/bash \
        "$CONTAINER_IMAGE" \
        -c "sleep infinity" &>/dev/null; then
        error "Failed to create container"
        return 1
    fi

    # Start container
    if ! docker start "$CONTAINER_NAME" &>/dev/null; then
        error "Failed to start container"
        docker rm -f "$CONTAINER_NAME" &>/dev/null || true
        return 1
    fi

    success "Container setup complete"
    return 0
}

# Function: Install Ansible in container
install_ansible() {
    log "Installing Ansible in container..."

    # Update package manager
    docker exec "$CONTAINER_NAME" \
        bash -c "apt-get update && apt-get install -y python3 python3-pip openssh-client" \
        &>/dev/null || {
            warning "apt-get update failed, trying without update"
        }

    # Install Ansible
    docker exec "$CONTAINER_NAME" \
        bash -c "pip install ansible" \
        &>/dev/null || {
            error "Failed to install Ansible"
            cleanup_container
            return 1
        }

    success "Ansible installed"
    return 0
}

# Function: Copy playbook to container
copy_playbook() {
    local playbook_dir=$(dirname "$PLAYBOOK_FILE")
    local playbook_name=$(basename "$PLAYBOOK_FILE")

    log "Copying playbook to container..."

    # Copy entire directory to container
    if ! docker cp "$playbook_dir" "$CONTAINER_NAME:/playbooks" &>/dev/null; then
        error "Failed to copy playbook to container"
        return 1
    fi

    success "Playbook copied to container"
}

# Function: Create inventory
create_inventory() {
    log "Creating inventory..."

    # Create inventory file in container
    docker exec "$CONTAINER_NAME" \
        bash -c "mkdir -p /etc/ansible && echo 'localhost ansible_connection=local' > /etc/ansible/hosts" \
        &>/dev/null

    success "Inventory created"
}

# Function: Run playbook
run_playbook() {
    log "Running playbook: $PLAYBOOK_FILE"

    local playbook_name=$(basename "$PLAYBOOK_FILE")
    local ansible_cmd="ansible-playbook /playbooks/$playbook_name"

    # Add inventory
    if [[ "$INVENTORY" != "localhost" ]]; then
        ansible_cmd="$ansible_cmd -i $INVENTORY"
    fi

    # Add tags
    if [[ -n "$TAGS" ]]; then
        ansible_cmd="$ansible_cmd --tags $TAGS"
    fi

    # Add skip tags
    if [[ -n "$SKIP_TAGS" ]]; then
        ansible_cmd="$ansible_cmd --skip-tags $SKIP_TAGS"
    fi

    # Add extra variables
    if [[ -n "$EXTRA_VARS" ]]; then
        ansible_cmd="$ansible_cmd -e '$EXTRA_VARS'"
    fi

    # Add verbose flag
    if [[ "$VERBOSE" == true ]]; then
        ansible_cmd="$ansible_cmd -vvv"
    fi

    # Run playbook
    local output=$(docker exec "$CONTAINER_NAME" bash -c "$ansible_cmd" 2>&1 || echo "FAILED")

    if echo "$output" | grep -q "FAILED\|ERROR"; then
        error "Playbook execution failed"
        echo ""
        echo "$output"
        return 1
    fi

    if [[ "$VERBOSE" == true ]]; then
        echo ""
        echo "$output"
    fi

    success "Playbook executed successfully"
    return 0
}

# Function: Get execution report
get_report() {
    log "Generating execution report..."

    local report=$(docker exec "$CONTAINER_NAME" \
        bash -c "ansible-playbook /playbooks/$(basename "$PLAYBOOK_FILE") --list-tasks" 2>/dev/null || echo "")

    if [[ -n "$report" ]]; then
        echo ""
        echo "Task List:"
        echo "$report"
    fi
}

# Function: Verify execution
verify_execution() {
    log "Verifying playbook execution..."

    # Check container status
    local status=$(docker inspect -f '{{.State.Status}}' "$CONTAINER_NAME")

    if [[ "$status" == "running" ]]; then
        success "Container is running"
        return 0
    else
        error "Container is not running (status: $status)"
        return 1
    fi
}

# Cleanup on exit
trap cleanup_container EXIT

# Main execution
main() {
    log "=========================================="
    log "Ansible Playbook Test"
    log "=========================================="
    log "Playbook: $PLAYBOOK_FILE"
    log "Container Image: $CONTAINER_IMAGE"
    echo ""

    # Pre-test checks
    check_dependencies

    # Setup environment
    if ! setup_container; then
        error "Container setup failed"
        exit 1
    fi

    # Install Ansible
    if ! install_ansible; then
        error "Ansible installation failed"
        exit 1
    fi

    # Prepare for execution
    copy_playbook
    create_inventory

    # Execute playbook
    if ! run_playbook; then
        error "Playbook test failed"
        exit 1
    fi

    # Verify and report
    get_report
    verify_execution

    echo ""
    success "Test completed successfully"
    log "=========================================="
    exit 0
}

# Run main function
main "$@"

#!/usr/bin/env python3
"""
Ansible playbook validator.

Validates playbook syntax and structure using:
- YAML syntax validation
- Ansible-lint integration
- Task validation
- Handler validation
- Variable validation
"""

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional
import subprocess
import yaml


class PlaybookValidator:
    """Validates Ansible playbooks."""

    def __init__(self, strict: bool = False):
        """
        Initialize validator.

        Args:
            strict: Enable strict validation mode
        """
        self.strict = strict
        self.errors = []
        self.warnings = []
        self.playbooks = []

    def validate_file(self, file_path: str) -> bool:
        """
        Validate playbook file.

        Args:
            file_path: Path to playbook file

        Returns:
            True if valid, False otherwise
        """
        try:
            path = Path(file_path)
            if not path.exists():
                self.errors.append(f"File not found: {file_path}")
                return False

            if path.suffix.lower() not in ['.yaml', '.yml']:
                self.errors.append(f"Expected YAML file, got: {path.suffix}")
                return False

            # Load and validate YAML
            if not self._validate_yaml(file_path):
                return False

            # Validate syntax with ansible-playbook
            if not self._validate_syntax(file_path):
                return False

            # Validate structure
            if not self._validate_structure(file_path):
                return False

            # Run ansible-lint if available
            self._run_ansible_lint(file_path)

            return len(self.errors) == 0

        except Exception as e:
            self.errors.append(f"Validation error: {str(e)}")
            return False

    def _validate_yaml(self, file_path: str) -> bool:
        """Validate YAML syntax."""
        try:
            with open(file_path, 'r') as f:
                data = yaml.safe_load(f)

            if not isinstance(data, list):
                self.errors.append("Playbook must be a YAML list (array)")
                return False

            if not data:
                self.errors.append("Playbook is empty")
                return False

            return True

        except yaml.YAMLError as e:
            self.errors.append(f"YAML syntax error: {str(e)}")
            return False

    def _validate_syntax(self, file_path: str) -> bool:
        """Validate playbook syntax with ansible-playbook."""
        try:
            result = subprocess.run(
                ['ansible-playbook', '--syntax-check', file_path],
                capture_output=True,
                text=True,
                timeout=30
            )

            if result.returncode != 0:
                self.errors.append(f"Playbook syntax error: {result.stderr}")
                return False

            return True

        except FileNotFoundError:
            self.warnings.append("ansible-playbook not found, skipping syntax check")
            return True
        except subprocess.TimeoutExpired:
            self.errors.append("Syntax check timeout")
            return False

    def _validate_structure(self, file_path: str) -> bool:
        """Validate playbook structure."""
        try:
            with open(file_path, 'r') as f:
                playbooks = yaml.safe_load(f)

            if not isinstance(playbooks, list):
                return True

            for idx, play in enumerate(playbooks):
                if not isinstance(play, dict):
                    self.errors.append(f"Play {idx}: must be a dictionary")
                    return False

                # Check required fields
                if 'name' not in play:
                    self.warnings.append(f"Play {idx}: missing 'name' field")

                if 'hosts' not in play:
                    self.errors.append(f"Play {idx}: missing 'hosts' field")
                    return False

                # Validate tasks
                if 'tasks' in play:
                    if not self._validate_tasks(play.get('tasks', []), f'Play {idx}'):
                        return False

                # Validate handlers
                if 'handlers' in play:
                    if not self._validate_handlers(play.get('handlers', []), f'Play {idx}'):
                        return False

                # Validate variables
                if 'vars' in play:
                    if not self._validate_vars(play.get('vars', {}), f'Play {idx}'):
                        return False

                self.playbooks.append({
                    'name': play.get('name', f'unnamed_play_{idx}'),
                    'hosts': play.get('hosts'),
                    'tasks': len(play.get('tasks', [])),
                    'handlers': len(play.get('handlers', [])),
                })

            return True

        except Exception as e:
            self.errors.append(f"Structure validation error: {str(e)}")
            return False

    def _validate_tasks(self, tasks: List[Any], context: str) -> bool:
        """Validate tasks."""
        if not isinstance(tasks, list):
            self.errors.append(f"{context}: tasks must be a list")
            return False

        for idx, task in enumerate(tasks):
            if not isinstance(task, dict):
                self.errors.append(f"{context} task {idx}: must be a dictionary")
                return False

            # Check that task has either 'name' or a module
            has_name = 'name' in task
            has_module = any(k in task for k in self._get_common_modules())

            if not has_name:
                self.warnings.append(f"{context} task {idx}: missing 'name'")

            if not has_module and not has_name:
                self.errors.append(f"{context} task {idx}: missing task module")
                return False

            # Check for common issues
            if 'debug' in task:
                if 'msg' not in task['debug']:
                    self.warnings.append(f"{context} task {idx} (debug): prefer 'msg:' over 'var:'")

            if 'shell' in task or 'command' in task:
                if 'warn' not in task:
                    self.warnings.append(
                        f"{context} task {idx}: shell/command should set warn: false or use module"
                    )

        return True

    def _validate_handlers(self, handlers: List[Any], context: str) -> bool:
        """Validate handlers."""
        if not isinstance(handlers, list):
            self.errors.append(f"{context}: handlers must be a list")
            return False

        for idx, handler in enumerate(handlers):
            if not isinstance(handler, dict):
                self.errors.append(f"{context} handler {idx}: must be a dictionary")
                return False

            if 'name' not in handler:
                self.errors.append(f"{context} handler {idx}: missing 'name'")
                return False

        return True

    def _validate_vars(self, variables: Dict[str, Any], context: str) -> bool:
        """Validate variables."""
        if not isinstance(variables, dict):
            self.errors.append(f"{context}: vars must be a dictionary")
            return False

        # Check for common variable issues
        for key, value in variables.items():
            # Warn about empty string defaults
            if isinstance(value, str) and not value:
                self.warnings.append(f"{context}: variable '{key}' is empty")

        return True

    def _run_ansible_lint(self, file_path: str) -> None:
        """Run ansible-lint if available."""
        try:
            result = subprocess.run(
                ['ansible-lint', file_path],
                capture_output=True,
                text=True,
                timeout=30
            )

            if result.stdout:
                # Parse ansible-lint output
                for line in result.stdout.split('\n'):
                    if 'error' in line.lower():
                        self.errors.append(f"ansible-lint: {line}")
                    elif 'warning' in line.lower():
                        self.warnings.append(f"ansible-lint: {line}")

        except FileNotFoundError:
            self.warnings.append("ansible-lint not found, skipping linting")
        except subprocess.TimeoutExpired:
            self.warnings.append("ansible-lint timeout")

    def _get_common_modules(self) -> List[str]:
        """Get list of common Ansible modules."""
        return [
            'debug', 'shell', 'command', 'copy', 'template', 'file',
            'lineinfile', 'replace', 'service', 'package', 'apt', 'yum',
            'git', 'get_url', 'uri', 'wait_for', 'handlers', 'block',
            'set_fact', 'include', 'import_tasks', 'loop', 'when',
            'register', 'notify', 'changed_when', 'failed_when',
        ]

    def get_report(self) -> Dict[str, Any]:
        """Get validation report."""
        return {
            'valid': len(self.errors) == 0,
            'playbooks': self.playbooks,
            'errors': self.errors,
            'warnings': self.warnings,
            'error_count': len(self.errors),
            'warning_count': len(self.warnings),
            'playbook_count': len(self.playbooks),
        }


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description='Validate Ansible playbook syntax and structure'
    )
    parser.add_argument(
        'playbook_file',
        help='Path to Ansible playbook file (YAML)'
    )
    parser.add_argument(
        '-o', '--output',
        help='Save validation report to JSON file'
    )
    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='Print detailed validation report'
    )
    parser.add_argument(
        '--strict',
        action='store_true',
        help='Treat warnings as errors'
    )

    args = parser.parse_args()

    try:
        validator = PlaybookValidator(strict=args.strict)
        is_valid = validator.validate_file(args.playbook_file)
        report = validator.get_report()

        # Check strict mode
        if args.strict and report['warning_count'] > 0:
            is_valid = False

        # Output report
        if args.verbose or not is_valid:
            print(json.dumps(report, indent=2))

        # Save report if requested
        if args.output:
            with open(args.output, 'w') as f:
                json.dump(report, f, indent=2)
            print(f"Validation report saved to: {args.output}")

        sys.exit(0 if is_valid else 1)

    except Exception as e:
        print(f"Error: {str(e)}", file=sys.stderr)
        sys.exit(1)


if __name__ == '__main__':
    main()