jester1177/cloud-native-debug-env

1"""Workflow simulator — YAML parse + CI rule checks."""2 3import re4from typing import Any, Dict, List, Optional5 6import yaml7 8from server.models import FileContent9 10 11class WorkflowSimulator:12    def validate(self, workflow: Optional[FileContent], files: Dict[str, FileContent]):13        if workflow is None:14            return {"parse_success": True, "execution_success": True}15 16        content = workflow.content17 18        # single-brace expressions: ${ } should be ${{ }}19        # Match ${ ... } that is NOT ${{ ... }}20        single_brace = re.findall(r'\$\{(?!\{)\s*[^}]+\}', content)21        if single_brace:22            return {23                "parse_success": False,24                "execution_success": False,25                "error": (26                    "Unrecognized expression syntax. "27                    "Use ${{ expression }} with double braces for GitHub Actions expressions."28                ),29            }30 31        # parse yaml32        try:33            parsed = yaml.safe_load(content)34        except yaml.YAMLError as exc:35            return {36                "parse_success": False,37                "execution_success": False,38                "error": f"YAML parse error: {exc}",39            }40 41        if not isinstance(parsed, dict):42            return {43                "parse_success": False,44                "execution_success": False,45                "error": "Workflow root must be a mapping",46            }47 48        # needs an 'on' trigger49        if "on" not in parsed and True not in parsed:50            # yaml.safe_load converts `on:` to True key in some contexts51            return {52                "parse_success": False,53                "execution_success": False,54                "error": "Workflow must define an 'on' trigger event",55            }56 57        # validate trigger structure58        on_value = parsed.get("on") or parsed.get(True)59        if isinstance(on_value, dict):60            for event_key, event_config in on_value.items():61                if isinstance(event_config, dict):62                    # Check branches is a list, not a bare string63                    branches_val = event_config.get("branches")64                    if isinstance(branches_val, str):65                        return {66                            "parse_success": False,67                            "execution_success": False,68                            "error": (69                                f"Unexpected value '{branches_val}' for 'on.{event_key}.branches'. "70                                "Expected a sequence (list) value."71                            ),72                        }73 74        # jobs block75        jobs = parsed.get("jobs")76        if not isinstance(jobs, dict) or not jobs:77            return {78                "parse_success": False,79                "execution_success": False,80                "error": "Workflow must define at least one job",81            }82 83        # Content-level flags for cross-cutting checks84        has_buildx_setup = "docker/setup-buildx-action" in content85        has_platforms = "platforms:" in content86        has_docker_login = "docker login" in content87        has_docker_push = "docker push" in content88        has_username_secret = "secrets.DOCKER_USERNAME" in content89        has_password_secret = "secrets.DOCKER_PASSWORD" in content90        has_github_token_secret = "secrets.GITHUB_TOKEN" in content91 92        # Collect job IDs for needs validation93        job_ids = set(jobs.keys())94 95        for job_name, job in jobs.items():96            if not isinstance(job, dict):97                continue98 99            # runs-on is required100            if "runs-on" not in job:101                return {102                    "parse_success": False,103                    "execution_success": False,104                    "error": f"Job '{job_name}' is missing required field 'runs-on'",105                }106 107            # check 'needs' refs point to real jobs108            needs = job.get("needs")109            if needs:110                needed = [needs] if isinstance(needs, str) else (needs if isinstance(needs, list) else [])111                for dep in needed:112                    if dep not in job_ids:113                        return {114                            "parse_success": False,115                            "execution_success": False,116                            "error": f"Job '{job_name}' depends on unknown job '{dep}'",117                        }118 119            steps = job.get("steps", [])120            if not isinstance(steps, list):121                return {122                    "parse_success": False,123                    "execution_success": False,124                    "error": f"Job '{job_name}' steps must be a list",125                }126 127            # every step needs 'uses' or 'run'128            for step in steps:129                if not isinstance(step, dict):130                    continue131                has_uses = "uses" in step132                has_run = "run" in step133                if not has_uses and not has_run:134                    step_name = step.get("name", "unnamed")135                    return {136                        "parse_success": False,137                        "execution_success": False,138                        "error": f"Every step must define a 'uses' or 'run' key. Step '{step_name}' has neither.",139                    }140 141            # checkout must come before docker build142            checkout_index = -1143            build_index = -1144            for idx, step in enumerate(steps):145                if not isinstance(step, dict):146                    continue147                uses = step.get("uses", "")148                run_cmd = step.get("run", "")149                if isinstance(uses, str) and "actions/checkout" in uses:150                    checkout_index = idx151                if (isinstance(run_cmd, str) and "docker build" in run_cmd) or (152                    isinstance(uses, str) and "docker/build-push-action" in uses153                ):154                    build_index = idx155 156            if build_index != -1 and (checkout_index == -1 or checkout_index > build_index):157                return {158                    "parse_success": True,159                    "execution_success": False,160                    "exec_error": "Checkout must happen before Docker build steps",161                }162 163        # cross-job artifact dependency: download needs 'needs'164        # If a job uses download-artifact but doesn't declare needs on the upload job165        for job_name, job in jobs.items():166            if not isinstance(job, dict):167                continue168            steps = job.get("steps", [])169            if not isinstance(steps, list):170                continue171            uses_download = any(172                isinstance(s, dict) and "actions/download-artifact" in str(s.get("uses", ""))173                for s in steps174            )175            if uses_download:176                needs = job.get("needs")177                if not needs:178                    return {179                        "parse_success": True,180                        "execution_success": False,181                        "exec_error": (182                            f"Job '{job_name}' uses download-artifact but has no 'needs' dependency — "183                            "add 'needs' to ensure the upload job completes first"184                        ),185                    }186 187        # docker login needs secrets wired via env188        if has_docker_login:189            # Check if the login step has env block with secrets190            login_has_env_secrets = has_username_secret and has_password_secret191            if not login_has_env_secrets:192                # Check if login uses $DOCKER_USERNAME (env var) without secret mapping193                if "$DOCKER_USERNAME" in content and not has_username_secret:194                    return {195                        "parse_success": True,196                        "execution_success": False,197                        "exec_error": "Docker login secrets not wired — add env block with secrets.DOCKER_USERNAME and secrets.DOCKER_PASSWORD",198                    }199 200        # push without login201        if has_docker_push and not has_docker_login:202            # Check if using docker/login-action instead203            has_login_action = "docker/login-action" in content204            if not has_login_action:205                return {206                    "parse_success": True,207                    "execution_success": False,208                    "exec_error": "Docker push without login — add a docker login step before pushing",209                }210 211        # ghcr.io needs GITHUB_TOKEN not DOCKER_PASSWORD212        if "docker login ghcr.io" in content:213            if has_password_secret and not has_github_token_secret:214                return {215                    "parse_success": True,216                    "execution_success": False,217                    "exec_error": "GHCR requires GITHUB_TOKEN for authentication, not DOCKER_PASSWORD",218                }219 220        # ghcr push needs packages:write permission221        if "ghcr.io" in content and "docker push" in content:222            # Check if permissions block has packages: write223            if "packages: write" not in content and "packages:write" not in content:224                return {225                    "parse_success": True,226                    "execution_success": False,227                    "exec_error": "GITHUB_TOKEN does not have packages:write permission — add permissions block",228                }229 230        # multi-platform needs buildx231        if has_platforms and not has_buildx_setup:232            return {233                "parse_success": True,234                "execution_success": False,235                "exec_error": "Multi-platform build requires docker/setup-buildx-action",236            }237 238        # GHA cache export needs mode=max239        if "cache-to:" in content and "cache-from:" in content:240            # Check for mode=max241            if "cache-to: type=gha" in content and "mode=max" not in content:242                return {243                    "parse_success": True,244                    "execution_success": False,245                    "exec_error": "GHA cache export needs mode=max for proper cache support",246                }247 248        # context vs dockerfile path mismatch249        for job_name, job in jobs.items():250            if not isinstance(job, dict):251                continue252            for step in job.get("steps", []):253                if not isinstance(step, dict):254                    continue255                with_block = step.get("with", {})256                if not isinstance(with_block, dict):257                    continue258                context = with_block.get("context")259                file_path = with_block.get("file")260                if context and file_path and isinstance(context, str) and isinstance(file_path, str):261                    # If context is a subdirectory but file is at root262                    if context not in {".", "./"} and not file_path.startswith(context):263                        return {264                            "parse_success": True,265                            "execution_success": False,266                            "exec_error": f"Dockerfile path '{file_path}' does not match build context '{context}'",267                        }268 269        # shell env var from secret but not mapped in env block270        for job_name, job in jobs.items():271            if not isinstance(job, dict):272                continue273            for step in job.get("steps", []):274                if not isinstance(step, dict):275                    continue276                run_cmd = step.get("run", "")277                if not isinstance(run_cmd, str):278                    continue279                env_block = step.get("env", {})280                if not isinstance(env_block, dict):281                    env_block = {}282                # Find env vars used in run that look like they should come from secrets283                env_var_refs = re.findall(r'\$([A-Z][A-Z0-9_]+)', run_cmd)284                for var in env_var_refs:285                    # Skip GitHub expression vars (they're in ${{ }})286                    if var in ("GITHUB_SHA", "GITHUB_REF", "GITHUB_ACTOR", "GITHUB_REPOSITORY"):287                        continue288                    # Common secret-backed env vars289                    if var in ("SLACK_WEBHOOK_URL", "DEPLOY_TOKEN", "NPM_TOKEN", "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"):290                        if var not in env_block:291                            return {292                                "parse_success": True,293                                "execution_success": False,294                                "exec_error": f"{var} is empty — secret not available in shell environment. Map it via env block.",295                            }296 297        # node version vs package.json engines298        for job_name, job in jobs.items():299            if not isinstance(job, dict):300                continue301            strategy = job.get("strategy", {})302            if not isinstance(strategy, dict):303                continue304            matrix = strategy.get("matrix", {})305            if not isinstance(matrix, dict):306                continue307            node_versions = matrix.get("node", [])308            if isinstance(node_versions, list):309                # Check package.json engines constraint310                pkg = files.get("package.json")311                if pkg:312                    engines_match = re.search(r'"node"\s*:\s*">=(\d+)"', pkg.content)313                    if engines_match:314                        min_version = int(engines_match.group(1))315                        for v in node_versions:316                            if isinstance(v, int) and v < min_version:317                                return {318                                    "parse_success": True,319                                    "execution_success": False,320                                    "exec_error": f"Matrix job (node: {v}) failed: package.json requires Node >= {min_version}",321                                }322 323        return {"parse_success": True, "execution_success": True}324