#!/usr/bin/env python3 """ci_prescan.py — deterministic CI/CD workflow security pre-scan. ARION service artifact. Scans CI/CD workflow definitions for the misconfiguration classes that are checkable offline: * pwn requests: ``pull_request_target`` (or fork-capable) triggers combined with a checkout of the untrusted PR head — forks get a write token and secrets over their own code * script injection: attacker-controlled ``${{ github.event.* }}`` contexts interpolated directly into ``run:`` shell blocks or ``actions/github-script`` ``script:`` bodies * mutable action pins: ``uses: owner/repo@v1`` / ``@main`` — tags and branches move; only a 40-hex SHA pin is immutable * token permissions: ``write-all``, missing ``permissions:`` blocks, and write scopes granted to fork-triggered workflows (incl. ``id-token: write`` OIDC) * self-hosted runners executing pull-request code * pipe-to-shell installs (``curl ... | bash``, ``irm | iex``) * secrets echoed or passed to third-party actions Honest scope: PRE-SCAN, not an audit. The YAML is scanned line/indentation-wise, not evaluated — it does not resolve reusable workflows, matrix expansions, or provenance of an action's contents. A FLAG means "this property deserves human review"; a clean report is not proof the pipeline is safe (compromised actions, OIDC trust-policy gaps, and runner-level issues are out of scope). Usage: ci_prescan.py [--json] ci_prescan.py --selftest Stdlib only. Exit 0 = no HIGH/MEDIUM findings, 1 = findings, 2 = usage error. """ import json import os import re import sys SEV_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2, "INFO": 3} # Attacker-influenceable contexts (PR/issue/comment bodies, branch # names, commit messages, discussion text, workflow_run metadata...). # Matched against the *inside* of ${{ ... }} expressions. UNTRUSTED_CTX = [ r"github\.event\.issue\.(title|body)", r"github\.event\.pull_request\.(title|body)", r"github\.event\.pull_request\.head\.(ref|label)", r"github\.event\.pull_request\.head\.repo\.(name|full_name|description)", r"github\.head_ref\b", r"github\.event\.comment\.(body|user\.login)", r"github\.event\.review(_comment)?\.body", r"github\.event\.discussion\.(title|body)", r"github\.event\.pages\.\w+\.page_name", r"github\.event\.commits\b", r"github\.event\.head_commit\.(message|author)", r"github\.event\.workflow_run\.(head_branch|display_title|name)", r"github\.event\.workflow_run\.head_repository\.(name|description)", r"github\.event\.check_(run|suite)\.", r"github\.event\.release\.(name|body|tag_name)", r"github\.event\.milestone\.(title|description)", r"github\.event\.client_payload\b", r"github\.event\.merge_group\.head_(ref|sha)", ] # Maintainer-controlled but still user-typed (workflow_dispatch / # repository_dispatch inputs) — flagged at lower severity. UNTRUSTED_LOW = [r"github\.event\.inputs?\.", r"(^|[^\w.])inputs\."] EXPR_RE = re.compile(r"\$\{\{(.*?)\}\}", re.S) PIPE_RE = re.compile( r"(curl|wget|Invoke-(?:WebRequest|RestMethod)|irm|iwr)\b[^|\n]*\|" r"\s*(sudo\s+)?(bash|sh|zsh|pwsh|powershell|iex|Invoke-Expression)") ECHO_SECRET_RE = re.compile( r"\b(echo|printf|cat|set-output)\b[^\n]*\$\{\{[^}]*secrets\.") IMAGE_RE = re.compile(r"^\s*(image|container)\s*:\s*(\S+)") SHA40_RE = re.compile(r"@[0-9a-f]{40}$") GHA_TRIGGERS_PR = {"pull_request", "pull_request_target", "pull_request_review", "pull_request_review_comment"} WRITE_PERMS = {"contents", "pull-requests", "issues", "actions", "id-token", "checks", "security-events", "packages", "deployments", "statuses"} def indent(line): return len(line) - len(line.lstrip(" ")) def block_lines(lines, i, base_indent): """Lines following index i that belong to a deeper-indented block.""" out = [] j = i + 1 while j < len(lines): ln = lines[j] if ln.strip() == "" or ln.lstrip().startswith("#"): out.append((j, ln)) j += 1 continue if indent(ln) <= base_indent: break out.append((j, ln)) j += 1 return out def scalar_block(lines, i, base_indent): """Collect a `|`/`>` block scalar after `key: |` on line i.""" return [ln for _, ln in block_lines(lines, i, base_indent)] def find_untrusted(text): hits, low = [], [] for m in EXPR_RE.finditer(text): expr = m.group(1) for pat in UNTRUSTED_CTX: if re.search(pat, expr): hits.append(pat) break else: for pat in UNTRUSTED_LOW: if re.search(pat, expr): low.append(pat) break return hits, low class Finding: def __init__(self, sev, code, path, line, msg): self.sev, self.code, self.path, self.line, self.msg = ( sev, code, path, line, msg) def row(self): return {"severity": self.sev, "code": self.code, "file": self.path, "line": self.line, "msg": self.msg} def parse_on_block(lines): """Return set of trigger event names from the top-level `on:` map.""" events = set() for i, ln in enumerate(lines): m = re.match(r"^on\s*:\s*(.*)$", ln) if not m: continue rest = m.group(1).strip() if rest.startswith("["): # on: [push, pull_request] events.update(re.findall(r"[\w-]+", rest)) return events if rest and not rest.startswith("#"): events.add(rest.split()[0]) return events for _, bl in block_lines(lines, i, 0): s = bl.strip() mm = re.match(r"-\s*([\w-]+)", s) if mm: events.add(mm.group(1)) continue mm = re.match(r"([\w-]+)\s*:", s) if mm: events.add(mm.group(1)) return events return events def parse_permissions(lines, base): """Collect perm->level pairs from every `permissions:` block at indent >= base. Returns dict name->level (last wins).""" perms = {} for i, ln in enumerate(lines): m = re.match(r"(\s*)permissions\s*:\s*(.*)$", ln) if not m: continue ind = len(m.group(1)) if ind < base: continue rest = m.group(2).strip() if rest in ("write-all", "read-all", "{}"): perms["*"] = rest continue for _, bl in block_lines(lines, i, ind): mm = re.match(r"\s*([\w-]+)\s*:\s*(\w+)", bl) if mm: perms[mm.group(1)] = mm.group(2) return perms def split_steps(lines): """Yield (start_idx, step_lines[(idx,line)]) for each `- ` item inside `steps:` blocks of the `jobs:` section.""" for i, ln in enumerate(lines): if not re.match(r"\s*steps\s*:\s*$", ln): continue sind = indent(ln) cur = None for j, bl in block_lines(lines, i, sind): s = bl.strip() if s == "" or s.startswith("#"): if cur is not None: cur.append((j, bl)) continue if re.match(r"-\s", s) and indent(bl) <= sind + 2: if cur: yield cur # rewrite `- key: v` to ` key: v` so field regexes # see a uniform map cur = [(j, " " * (indent(bl) + 2) + s[2:])] elif cur is not None: cur.append((j, bl)) if cur: yield cur def step_field(step_lines, key): """Return (idx, inline_value, block_text) for `key:` inside a step.""" step_ind = indent(step_lines[0][1]) n = len(step_lines) for pos in range(n): idx, ln = step_lines[pos] m = re.match(r"\s*" + re.escape(key) + r"\s*:\s*(.*)$", ln) if not m or indent(ln) > step_ind + 6: continue val = m.group(1).strip() if val.startswith(("|", ">")) or val == "": k = pos + 1 subs = [] while k < n and (step_lines[k][1].strip() == "" or indent(step_lines[k][1]) > indent(ln)): subs.append(step_lines[k][1]) k += 1 return idx, "", "\n".join(subs) return idx, val, "" return None, "", "" def scan_gha(path, lines, findings): rel = path events = parse_on_block(lines) prt = "pull_request_target" in events pr_any = bool(events & GHA_TRIGGERS_PR) or prt if prt: findings.append(Finding( "INFO", "PRT-BASE", rel, 1, "pull_request_target trigger: workflow runs in the base-repo " "context (write-capable token, secrets available) over " "fork-supplied events — every step below must treat PR data " "as hostile")) # permissions top_perms = parse_permissions(lines, 0) if not top_perms: findings.append(Finding( "INFO", "PERMS-DEFAULT", rel, 1, "no explicit permissions: block — token scope falls back to " "the repo/org default (often permissive read-write)")) for name, lvl in top_perms.items(): if name == "*" and lvl == "write-all": sev = "HIGH" if pr_any else "MEDIUM" findings.append(Finding( sev, "PERMS-WRITEALL", rel, 1, "permissions: write-all grants every scope" + (" to a pull-request-triggered workflow" if pr_any else ""))) elif lvl == "write" and name in WRITE_PERMS: if prt: findings.append(Finding( "HIGH", "PRT-WRITE", rel, 1, f"{name}: write granted under pull_request_target — " "fork PRs can reach a write-capable token" + (" (OIDC id-token!)" if name == "id-token" else ""))) # steps for step in split_steps(lines): sidx = step[0][0] + 1 ui, uses_val, _ = step_field(step, "uses") ri, run_val, run_blk = step_field(step, "run") wi, with_val, with_blk = step_field(step, "with") ei, env_val, env_blk = step_field(step, "env") # unpinned action if uses_val: uses_val = uses_val.split("#")[0].strip().strip("'\"") if uses_val.startswith(("./", "docker://")): if uses_val.startswith("docker://") and "@sha256" not in uses_val: findings.append(Finding( "LOW", "DOCKER-UNPINNED", rel, ui + 1, f"docker image not digest-pinned: {uses_val}")) elif "@" not in uses_val: findings.append(Finding( "MEDIUM", "ACTION-UNPINNED", rel, ui + 1, f"action has no ref at all: {uses_val}")) elif not SHA40_RE.search(uses_val): findings.append(Finding( "MEDIUM", "ACTION-TAGPIN", rel, ui + 1, f"action pinned to a mutable ref, not a commit SHA: " f"{uses_val}")) # pwn request: checkout of PR head under pull_request_target if uses_val and uses_val.split("@")[0].rstrip().endswith( "actions/checkout"): wb = with_blk or "" if re.search(r"ref\s*:.*(pull_request\.head|github\.head_ref|" r"\bhead\b.*sha)", wb): if prt: findings.append(Finding( "HIGH", "PWN-REQUEST", rel, ui + 1, "pull_request_target + checkout of the untrusted " "PR head: fork code executes with a write token " "and secrets — the classic pwn request")) elif pr_any: findings.append(Finding( "INFO", "PR-HEAD-CHECKOUT", rel, ui + 1, "explicit checkout of PR head under a read-only " "pull_request context — verify no write scope is " "added later")) # script injection in run: / github-script script: run_text = (run_val + "\n" + run_blk).strip() if run_text: hits, lows = find_untrusted(run_text) if hits: findings.append(Finding( "HIGH", "SCRIPT-INJECTION", rel, (ri or sidx) + 1, "attacker-controlled ${{ }} context interpolated " "directly into a run: shell block — route it through " "an env: var and quote the expansion instead")) for sev, pats in (("LOW", lows),): if pats: findings.append(Finding( sev, "RUN-INPUT-CTX", rel, (ri or sidx) + 1, "workflow_dispatch inputs.* interpolated into " "run: — still user-typed text")) if PIPE_RE.search(run_text): findings.append(Finding( "MEDIUM", "PIPE-EXEC", rel, (ri or sidx) + 1, "remote content piped straight into a shell " "(curl|bash class) — verify the source and pin a " "version/hash")) if ECHO_SECRET_RE.search(run_text): findings.append(Finding( "LOW", "SECRET-ECHO", rel, (ri or sidx) + 1, "a secrets.* value appears to be echoed/written to " "output in run:")) if re.search(r"\bgh\s+pr\s+(checkout|merge|review)", run_text): sev = "MEDIUM" if (prt or any( v == "write" for v in top_perms.values())) else "LOW" findings.append(Finding( sev, "GH-PR-CMD", rel, (ri or sidx) + 1, "gh pr checkout/merge in run: — confirm the token " "scope cannot let a fork merge or execute its code")) # github-script injection: context inside the script: sub-block if uses_val and "github-script" in uses_val: si, sval, sblk = step_field(step, "script") stext = (sval + "\n" + (sblk or "")).strip() if not stext and with_blk: hits, _ = find_untrusted(with_blk) if hits: findings.append(Finding( "HIGH", "GHSCRIPT-INJECTION", rel, sidx, "attacker-controlled ${{ }} context inside " "actions/github-script with: block — pass it via " "env: and read process.env instead")) elif stext: hits, _ = find_untrusted(stext) if hits: findings.append(Finding( "HIGH", "GHSCRIPT-INJECTION", rel, (si or sidx) + 1, "attacker-controlled ${{ }} context inside " "github-script script: — pass it via env: and " "read process.env instead")) # untrusted context bound into env:/with: (the safe pattern, but # still worth one note for data-flow review) for blk, tag in ((env_blk, "env"), (with_blk, "with")): if blk: hits, _ = find_untrusted(blk) if hits: findings.append(Finding( "INFO", "CTX-IN-" + tag.upper(), rel, sidx, f"untrusted context bound via {tag}: — correct " f"indirection; ensure it is consumed quoted " f"(\"$VAR\") and never re-interpolated")) if tag == "with" and uses_val and not uses_val.startswith( ("actions/", "github/", "./", "docker://")): if re.search(r"\$\{\{\s*secrets\.", blk): findings.append(Finding( "LOW", "SECRET-TO-3P", rel, sidx, f"secrets.* passed to third-party action " f"{uses_val.split('@')[0]} — it can read " f"whatever it receives")) # self-hosted runner on PR-triggered jobs for i, ln in enumerate(lines): if re.search(r"runs-on\s*:.*self-hosted", ln): if pr_any: findings.append(Finding( "HIGH", "SELFHOSTED-PR", rel, i + 1, "self-hosted runner executes pull-request-triggered " "jobs — fork code runs on your infrastructure with " "whatever persists there")) else: findings.append(Finding( "INFO", "SELFHOSTED", rel, i + 1, "self-hosted runner in use — confirm job isolation " "and image hygiene")) def scan_generic(path, lines, findings): """Non-GHA CI files (gitlab-ci, circleci, bitbucket, jenkins): generic transport/exec checks only.""" findings.append(Finding( "INFO", "GENERIC-MODE", path, 1, "non-GitHub-Actions CI file — generic checks only " "(pipe-to-shell, unpinned images, secret echo); " "platform-specific trigger/token review is manual")) for i, ln in enumerate(lines): if PIPE_RE.search(ln): findings.append(Finding( "MEDIUM", "PIPE-EXEC", path, i + 1, "remote content piped into a shell (curl|bash class)")) m = IMAGE_RE.match(ln) if m: img = m.group(2).strip("'\"") if ":" not in img or img.endswith(":latest"): findings.append(Finding( "LOW", "IMAGE-UNPINNED", path, i + 1, f"CI image not pinned to a tag/digest: {img}")) if ECHO_SECRET_RE.search(ln) or re.search( r"\b(echo|printf)\b.*\$[A-Z_]*(SECRET|TOKEN|PASSWORD|KEY)", ln): findings.append(Finding( "LOW", "SECRET-ECHO", path, i + 1, "possible secret material echoed to CI logs")) WORKFLOW_RE = re.compile(r"\.(ya?ml)$") def iter_targets(root): if os.path.isfile(root): yield root return gha = os.path.join(root, ".github", "workflows") if os.path.isdir(gha): for f in sorted(os.listdir(gha)): if WORKFLOW_RE.search(f): yield os.path.join(gha, f) for f in sorted(os.listdir(root)): fl = f.lower() if (fl in (".gitlab-ci.yml", "azure-pipelines.yml", "bitbucket-pipelines.yml", "jenkinsfile", "circle.yml") or fl == "config.yml" and os.path.basename(root) == ".circleci"): p = os.path.join(root, f) if os.path.isfile(p): yield p cc = os.path.join(root, ".circleci", "config.yml") if os.path.isfile(cc): yield cc def is_gha(path): return (".github/workflows/" in path.replace(os.sep, "/") or WORKFLOW_RE.search(os.path.basename(path)) and "workflows" in path) def scan_path(root): findings = [] files = list(iter_targets(root)) for p in files: try: with open(p, "r", encoding="utf-8", errors="replace") as fh: lines = fh.read().splitlines() except OSError as e: findings.append(Finding("INFO", "UNREADABLE", p, 0, str(e))) continue rel = os.path.relpath(p, root if os.path.isdir(root) else os.path.dirname(root) or ".") if is_gha(p): scan_gha(rel, lines, findings) else: scan_generic(rel, lines, findings) if not files: findings.append(Finding( "INFO", "NO-CI", ".", 0, "no CI/CD workflow files found at the given path")) findings.sort(key=lambda f: (SEV_ORDER[f.sev], f.path, f.line)) return findings, files def report(findings, files, root): counts = {} for f in findings: counts[f.sev] = counts.get(f.sev, 0) + 1 out = ["# CI/CD security pre-scan report", "", f"Target: `{root}` — {len(files)} workflow file(s) scanned.", ""] out.append("Findings: " + (", ".join( f"**{s}**: {counts[s]}" for s in ("HIGH", "MEDIUM", "LOW", "INFO") if s in counts) or "none")) out.append("") for f in findings: loc = f"{f.path}:{f.line}" if f.line else f.path out.append(f"- **{f.sev}** `{f.code}` — `{loc}` — {f.msg}") out += ["", "---", "*Pre-scan, not an audit: heuristic line-based checks; a clean " "report is not proof of pipeline safety.*"] return "\n".join(out) def selftest(): import tempfile vectors = { "safe.yml": """\ on: [push] permissions: contents: read jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@8f4b7f84864484a7bf31766abe9204da3cbe65b3 - run: echo hello """, "pwn.yml": """\ on: pull_request_target jobs: t: runs-on: ubuntu-latest steps: - uses: actions/checkout@8f4b7f84864484a7bf31766abe9204da3cbe65b3 with: ref: ${{ github.event.pull_request.head.sha }} - run: make test """, "inject.yml": """\ on: issues jobs: t: runs-on: ubuntu-latest steps: - run: echo "${{ github.event.issue.title }}" """, "unpinned.yml": """\ on: push permissions: {} jobs: t: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: docker://alpine:latest """, "writeall.yml": """\ on: push permissions: write-all jobs: t: runs-on: ubuntu-latest steps: - run: echo x """, "prt_write.yml": """\ on: pull_request_target permissions: contents: write id-token: write jobs: t: runs-on: ubuntu-latest steps: - run: echo x """, "selfhost.yml": """\ on: pull_request jobs: t: runs-on: [self-hosted, linux] steps: - run: echo x """, "pipe.yml": """\ on: push permissions: {} jobs: t: runs-on: ubuntu-latest steps: - run: curl -sSL https://example.com/install.sh | bash """, "envsafe.yml": """\ on: issues permissions: {} jobs: t: runs-on: ubuntu-latest steps: - env: TITLE: ${{ github.event.issue.title }} run: echo "$TITLE" """, "secret_echo.yml": """\ on: push permissions: {} jobs: t: runs-on: ubuntu-latest steps: - run: echo "${{ secrets.MY_TOKEN }}" > leak.txt """, "ghscript.yml": """\ on: issue_comment permissions: {} jobs: t: runs-on: ubuntu-latest steps: - uses: actions/github-script@60a0d83039c74a4aee543508d2ffcb1c3799cdea with: script: | const body = "${{ github.event.comment.body }}"; core.info(body); """, } expect = { "safe.yml": {"high": 0, "med": 0}, "pwn.yml": {"high": 1, "med": 0}, "inject.yml": {"high": 1}, "unpinned.yml": {"med": 1}, "writeall.yml": {"med": 1}, "prt_write.yml": {"high": 2}, "selfhost.yml": {"high": 1}, "pipe.yml": {"med": 1}, "envsafe.yml": {"high": 0, "med": 0}, "secret_echo.yml": {"high": 0, "med": 0}, "ghscript.yml": {"high": 1}, } with tempfile.TemporaryDirectory() as td: wf = os.path.join(td, ".github", "workflows") os.makedirs(wf) for name, body in vectors.items(): with open(os.path.join(wf, name), "w") as fh: fh.write(body) findings, files = scan_path(td) assert len(files) == len(vectors), (len(files), len(vectors)) byfile = {} for f in findings: byfile.setdefault(os.path.basename(f.path), []).append(f) ok = True for name, exp in expect.items(): fs = byfile.get(name, []) nh = sum(1 for f in fs if f.sev == "HIGH") nm = sum(1 for f in fs if f.sev == "MEDIUM") if "high" in exp and nh != exp["high"]: ok = False print(f"FAIL {name}: {nh} HIGH (want {exp['high']}) " f"{[f.code for f in fs]}") if "med" in exp and nm != exp["med"]: ok = False print(f"FAIL {name}: {nm} MEDIUM (want {exp['med']}) " f"{[f.code for f in fs]}") # generic-mode gitlab vector gl = os.path.join(td, ".gitlab-ci.yml") with open(gl, "w") as fh: fh.write("image: node:latest\nbuild:\n script:\n" " - curl https://x.sh | sh\n") findings2, files2 = scan_path(td) g = [f for f in findings2 if f.path == ".gitlab-ci.yml" and f.code == "PIPE-EXEC"] if not g: ok = False print("FAIL gitlab PIPE-EXEC") print("SELFTEST " + ("PASS" if ok else "FAIL")) return 0 if ok else 1 def main(argv): args = [a for a in argv[1:] if not a.startswith("--")] if "--selftest" in argv: return selftest() if len(args) != 1: print(__doc__) return 2 root = args[0] if not os.path.exists(root): print(f"not found: {root}") return 2 findings, files = scan_path(root) if "--json" in argv: print(json.dumps({"files": files, "findings": [f.row() for f in findings]}, indent=1)) else: print(report(findings, files, root)) return 1 if any(f.sev in ("HIGH", "MEDIUM") for f in findings) else 0 if __name__ == "__main__": sys.exit(main(sys.argv))