server/simulators/workflow_simulator.py

"""Workflow simulator — YAML parse + CI rule checks."""

import re
from typing import Any, Dict, List, Optional

import yaml

from server.models import FileContent


class WorkflowSimulator:
    def validate(self, workflow: Optional[FileContent], files: Dict[str, FileContent]):
        if workflow is None:
            return {"parse_success": True, "execution_success": True}

        content = workflow.content

        # single-brace expressions: ${ } should be ${{ }}
        # Match ${ ... } that is NOT ${{ ... }}
        single_brace = re.findall(r'\$\{(?!\{)\s*[^}]+\}', content)
        if single_brace:
            return {
                "parse_success": False,
                "execution_success": False,
                "error": (
                    "Unrecognized expression syntax. "
                    "Use ${{ expression }} with double braces for GitHub Actions expressions."
                ),
            }

        # parse yaml
        try:
            parsed = yaml.safe_load(content)
        except yaml.YAMLError as exc:
            return {
                "parse_success": False,
                "execution_success": False,
                "error": f"YAML parse error: {exc}",
            }

        if not isinstance(parsed, dict):
            return {
                "parse_success": False,
                "execution_success": False,
                "error": "Workflow root must be a mapping",
            }

        # needs an 'on' trigger
        if "on" not in parsed and True not in parsed:
            # yaml.safe_load converts `on:` to True key in some contexts
            return {
                "parse_success": False,
                "execution_success": False,
                "error": "Workflow must define an 'on' trigger event",
            }

        # validate trigger structure
        on_value = parsed.get("on") or parsed.get(True)
        if isinstance(on_value, dict):
            for event_key, event_config in on_value.items():
                if isinstance(event_config, dict):
                    # Check branches is a list, not a bare string
                    branches_val = event_config.get("branches")
                    if isinstance(branches_val, str):
                        return {
                            "parse_success": False,
                            "execution_success": False,
                            "error": (
                                f"Unexpected value '{branches_val}' for 'on.{event_key}.branches'. "
                                "Expected a sequence (list) value."
                            ),
                        }

        # jobs block
        jobs = parsed.get("jobs")
        if not isinstance(jobs, dict) or not jobs:
            return {
                "parse_success": False,
                "execution_success": False,
                "error": "Workflow must define at least one job",
            }

        # Content-level flags for cross-cutting checks
        has_buildx_setup = "docker/setup-buildx-action" in content
        has_platforms = "platforms:" in content
        has_docker_login = "docker login" in content
        has_docker_push = "docker push" in content
        has_username_secret = "secrets.DOCKER_USERNAME" in content
        has_password_secret = "secrets.DOCKER_PASSWORD" in content
        has_github_token_secret = "secrets.GITHUB_TOKEN" in content

        # Collect job IDs for needs validation
        job_ids = set(jobs.keys())

        for job_name, job in jobs.items():
            if not isinstance(job, dict):
                continue

            # runs-on is required
            if "runs-on" not in job:
                return {
                    "parse_success": False,
                    "execution_success": False,
                    "error": f"Job '{job_name}' is missing required field 'runs-on'",
                }

            # check 'needs' refs point to real jobs
            needs = job.get("needs")
            if needs:
                needed = [needs] if isinstance(needs, str) else (needs if isinstance(needs, list) else [])
                for dep in needed:
                    if dep not in job_ids:
                        return {
                            "parse_success": False,
                            "execution_success": False,
                            "error": f"Job '{job_name}' depends on unknown job '{dep}'",
                        }

            steps = job.get("steps", [])
            if not isinstance(steps, list):
                return {
                    "parse_success": False,
                    "execution_success": False,
                    "error": f"Job '{job_name}' steps must be a list",
                }

            # every step needs 'uses' or 'run'
            for step in steps:
                if not isinstance(step, dict):
                    continue
                has_uses = "uses" in step
                has_run = "run" in step
                if not has_uses and not has_run:
                    step_name = step.get("name", "unnamed")
                    return {
                        "parse_success": False,
                        "execution_success": False,
                        "error": f"Every step must define a 'uses' or 'run' key. Step '{step_name}' has neither.",
                    }

            # checkout must come before docker build
            checkout_index = -1
            build_index = -1
            for idx, step in enumerate(steps):
                if not isinstance(step, dict):
                    continue
                uses = step.get("uses", "")
                run_cmd = step.get("run", "")
                if isinstance(uses, str) and "actions/checkout" in uses:
                    checkout_index = idx
                if (isinstance(run_cmd, str) and "docker build" in run_cmd) or (
                    isinstance(uses, str) and "docker/build-push-action" in uses
                ):
                    build_index = idx

            if build_index != -1 and (checkout_index == -1 or checkout_index > build_index):
                return {
                    "parse_success": True,
                    "execution_success": False,
                    "exec_error": "Checkout must happen before Docker build steps",
                }

        # cross-job artifact dependency: download needs 'needs'
        # If a job uses download-artifact but doesn't declare needs on the upload job
        for job_name, job in jobs.items():
            if not isinstance(job, dict):
                continue
            steps = job.get("steps", [])
            if not isinstance(steps, list):
                continue
            uses_download = any(
                isinstance(s, dict) and "actions/download-artifact" in str(s.get("uses", ""))
                for s in steps
            )
            if uses_download:
                needs = job.get("needs")
                if not needs:
                    return {
                        "parse_success": True,
                        "execution_success": False,
                        "exec_error": (
                            f"Job '{job_name}' uses download-artifact but has no 'needs' dependency — "
                            "add 'needs' to ensure the upload job completes first"
                        ),
                    }

        # docker login needs secrets wired via env
        if has_docker_login:
            # Check if the login step has env block with secrets
            login_has_env_secrets = has_username_secret and has_password_secret
            if not login_has_env_secrets:
                # Check if login uses $DOCKER_USERNAME (env var) without secret mapping
                if "$DOCKER_USERNAME" in content and not has_username_secret:
                    return {
                        "parse_success": True,
                        "execution_success": False,
                        "exec_error": "Docker login secrets not wired — add env block with secrets.DOCKER_USERNAME and secrets.DOCKER_PASSWORD",
                    }

        # push without login
        if has_docker_push and not has_docker_login:
            # Check if using docker/login-action instead
            has_login_action = "docker/login-action" in content
            if not has_login_action:
                return {
                    "parse_success": True,
                    "execution_success": False,
                    "exec_error": "Docker push without login — add a docker login step before pushing",
                }

        # ghcr.io needs GITHUB_TOKEN not DOCKER_PASSWORD
        if "docker login ghcr.io" in content:
            if has_password_secret and not has_github_token_secret:
                return {
                    "parse_success": True,
                    "execution_success": False,
                    "exec_error": "GHCR requires GITHUB_TOKEN for authentication, not DOCKER_PASSWORD",
                }

        # ghcr push needs packages:write permission
        if "ghcr.io" in content and "docker push" in content:
            # Check if permissions block has packages: write
            if "packages: write" not in content and "packages:write" not in content:
                return {
                    "parse_success": True,
                    "execution_success": False,
                    "exec_error": "GITHUB_TOKEN does not have packages:write permission — add permissions block",
                }

        # multi-platform needs buildx
        if has_platforms and not has_buildx_setup:
            return {
                "parse_success": True,
                "execution_success": False,
                "exec_error": "Multi-platform build requires docker/setup-buildx-action",
            }

        # GHA cache export needs mode=max
        if "cache-to:" in content and "cache-from:" in content:
            # Check for mode=max
            if "cache-to: type=gha" in content and "mode=max" not in content:
                return {
                    "parse_success": True,
                    "execution_success": False,
                    "exec_error": "GHA cache export needs mode=max for proper cache support",
                }

        # context vs dockerfile path mismatch
        for job_name, job in jobs.items():
            if not isinstance(job, dict):
                continue
            for step in job.get("steps", []):
                if not isinstance(step, dict):
                    continue
                with_block = step.get("with", {})
                if not isinstance(with_block, dict):
                    continue
                context = with_block.get("context")
                file_path = with_block.get("file")
                if context and file_path and isinstance(context, str) and isinstance(file_path, str):
                    # If context is a subdirectory but file is at root
                    if context not in {".", "./"} and not file_path.startswith(context):
                        return {
                            "parse_success": True,
                            "execution_success": False,
                            "exec_error": f"Dockerfile path '{file_path}' does not match build context '{context}'",
                        }

        # shell env var from secret but not mapped in env block
        for job_name, job in jobs.items():
            if not isinstance(job, dict):
                continue
            for step in job.get("steps", []):
                if not isinstance(step, dict):
                    continue
                run_cmd = step.get("run", "")
                if not isinstance(run_cmd, str):
                    continue
                env_block = step.get("env", {})
                if not isinstance(env_block, dict):
                    env_block = {}
                # Find env vars used in run that look like they should come from secrets
                env_var_refs = re.findall(r'\$([A-Z][A-Z0-9_]+)', run_cmd)
                for var in env_var_refs:
                    # Skip GitHub expression vars (they're in ${{ }})
                    if var in ("GITHUB_SHA", "GITHUB_REF", "GITHUB_ACTOR", "GITHUB_REPOSITORY"):
                        continue
                    # Common secret-backed env vars
                    if var in ("SLACK_WEBHOOK_URL", "DEPLOY_TOKEN", "NPM_TOKEN", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"):
                        if var not in env_block:
                            return {
                                "parse_success": True,
                                "execution_success": False,
                                "exec_error": f"{var} is empty — secret not available in shell environment. Map it via env block.",
                            }

        # node version vs package.json engines
        for job_name, job in jobs.items():
            if not isinstance(job, dict):
                continue
            strategy = job.get("strategy", {})
            if not isinstance(strategy, dict):
                continue
            matrix = strategy.get("matrix", {})
            if not isinstance(matrix, dict):
                continue
            node_versions = matrix.get("node", [])
            if isinstance(node_versions, list):
                # Check package.json engines constraint
                pkg = files.get("package.json")
                if pkg:
                    engines_match = re.search(r'"node"\s*:\s*">=(\d+)"', pkg.content)
                    if engines_match:
                        min_version = int(engines_match.group(1))
                        for v in node_versions:
                            if isinstance(v, int) and v < min_version:
                                return {
                                    "parse_success": True,
                                    "execution_success": False,
                                    "exec_error": f"Matrix job (node: {v}) failed: package.json requires Node >= {min_version}",
                                }

        return {"parse_success": True, "execution_success": True}