← back to Cli Printing Press
.github/scripts/verify-supply-chain/signals.py
426 lines
"""Signal catalog for the supply-chain scan (cli-printing-press mirror).
This file is vendored from mvanhorn/printing-press-library's
.github/scripts/verify-supply-chain/signals.py with scope adaptations for
the generator repo:
- R1 (pull_request_target + PR-head checkout) — applied unchanged.
- R2 (id-token: write outside allowlist) — applied with an empty
allowlist. The generator repo does not currently use OIDC anywhere;
any addition trips the rule. If release.yml migrates to keyless
cosign with OIDC, allowlist that specific workflow at that time.
- R3 (replace directives in library/**/go.mod) — omitted.
- R4 (GOPROXY/GOFLAGS/GONOSUMCHECK in workflows) — applied unchanged.
- R5 (npm lifecycle scripts) — omitted.
- R6 (module-path drift) — omitted.
Detection strategy for R1/R2/R4: parse the YAML structurally with pyyaml
(pre-installed on ubuntu-latest runners) and walk the parsed dict tree.
The earlier regex approach missed valid YAML forms (block-scalar values
`ref: >-` with the value on the next line, etc.) and required a separate
patch for each YAML quirk encountered. Structural parsing eliminates the
whole class.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import PurePosixPath
from typing import Any
import yaml
# ---------------------------------------------------------------------------
# Types
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Finding:
path: str
line: int | None
severity: str # "block" | "advise"
signal_id: str
message: str
remediation: str
def is_block(self) -> bool:
return self.severity == "block"
@dataclass(frozen=True)
class FileChange:
path: str
base_content: str | None
head_content: str | None
added_lines: list[tuple[int, str]]
# ---------------------------------------------------------------------------
# Path-scope helpers
# ---------------------------------------------------------------------------
def is_workflow(path: str) -> bool:
parts = PurePosixPath(path).parts
return (
len(parts) >= 3
and parts[0] == ".github"
and parts[1] == "workflows"
and (path.endswith(".yml") or path.endswith(".yaml"))
)
# Empty allowlist in the generator repo — id-token: write should not appear
# in any current workflow.
ID_TOKEN_ALLOWLIST: set[str] = set()
# ---------------------------------------------------------------------------
# YAML parsing helpers (shared by R1, R2, R4)
# ---------------------------------------------------------------------------
def _parse_workflow(content: str | None) -> Any:
"""Parse a workflow YAML safely. Returns the loaded structure (typically
a dict) or None if the content is absent or unparseable. A malformed
workflow would also fail to load in GitHub Actions itself, so silently
skipping it is safe."""
if content is None:
return None
try:
return yaml.safe_load(content)
except (yaml.YAMLError, TypeError, ValueError):
return None
def _workflow_on(parsed: Any) -> Any:
"""Extract the workflow's `on:` section.
YAML 1.1 (pyyaml's default) interprets unquoted `on` as the boolean
literal True — so `on:` at the top of a workflow becomes the key True
after parsing. Check both forms; GitHub Actions accepts either."""
if not isinstance(parsed, dict):
return None
if "on" in parsed:
return parsed["on"]
if True in parsed:
return parsed[True]
return None
def _has_pr_target_trigger(on_node: Any) -> bool:
"""Detect pull_request_target in any of YAML's trigger declaration forms:
string (`on: pull_request_target`), list (`on: [...]` or
`on:\\n - pull_request_target`), or mapping (`on:\\n pull_request_target:`)."""
if on_node is None:
return False
if isinstance(on_node, str):
return on_node.strip() == "pull_request_target"
if isinstance(on_node, list):
return any(
isinstance(item, str) and item.strip() == "pull_request_target"
for item in on_node
)
if isinstance(on_node, dict):
return "pull_request_target" in on_node
return False
_DANGEROUS_REF_VALUE = re.compile(
# All forms that resolve to PR-author-controlled content under
# pull_request_target. github.head_ref is the shorthand alias for
# event.pull_request.head.ref (Greptile-flagged); merge_commit_sha
# points at GitHub's synthesised merge commit which contains PR code.
r"github\.event\.pull_request\.head\.(sha|ref)"
r"|github\.event\.pull_request\.merge_commit_sha"
r"|github\.head_ref"
# [^\n] (not [^\s]) so the match survives spaces inside `${{ ... }}`
# expressions, e.g., refs/pull/${{ github.event.number }}/merge.
r"|refs/pull/[^\n]*?/(merge|head)"
)
def _is_dangerous_ref_value(value: Any) -> bool:
"""Return True if a checkout step's `ref:` value references the PR head."""
if not isinstance(value, str):
return False
return bool(_DANGEROUS_REF_VALUE.search(value))
def _walk_checkout_refs(parsed: Any) -> list[str]:
"""Walk parsed workflow for actions/checkout steps with dangerous ref
values. Returns the list of dangerous ref values found (one per offending
step). Inspects every job and every step recursively but only flags the
`with.ref` field of `uses: actions/checkout*` steps."""
if not isinstance(parsed, dict):
return []
jobs = parsed.get("jobs")
if not isinstance(jobs, dict):
return []
findings: list[str] = []
for job in jobs.values():
if not isinstance(job, dict):
continue
steps = job.get("steps")
if not isinstance(steps, list):
continue
for step in steps:
if not isinstance(step, dict):
continue
uses = step.get("uses")
if not isinstance(uses, str) or not uses.startswith("actions/checkout"):
continue
with_block = step.get("with")
if not isinstance(with_block, dict):
continue
ref = with_block.get("ref")
if _is_dangerous_ref_value(ref):
findings.append(ref.strip())
return findings
def _walk_id_token_grants(parsed: Any) -> bool:
"""Return True if the parsed workflow grants `id-token: write` at workflow
or job level. (GitHub Actions doesn't honor step-level permissions, so we
skip those.)"""
if not isinstance(parsed, dict):
return False
if _permissions_grant_id_token(parsed.get("permissions")):
return True
jobs = parsed.get("jobs")
if isinstance(jobs, dict):
for job in jobs.values():
if isinstance(job, dict) and _permissions_grant_id_token(job.get("permissions")):
return True
return False
def _permissions_grant_id_token(perm: Any) -> bool:
if isinstance(perm, dict):
value = perm.get("id-token")
return isinstance(value, str) and value.strip() == "write"
# A string value of "write-all" grants every permission, including id-token.
if isinstance(perm, str) and perm.strip() == "write-all":
return True
return False
_GO_ENV_KEYS = ("GOPROXY", "GOFLAGS", "GONOSUMCHECK", "GOSUMDB", "GONOSUMDB")
def _walk_go_env_overrides(parsed: Any) -> list[str]:
"""Return the list of GOPROXY/GOFLAGS/etc env-variable names set anywhere
in the workflow's env blocks (workflow-level, job-level, or step-level)."""
found: list[str] = []
if not isinstance(parsed, dict):
return found
_collect_go_env_from(parsed.get("env"), found)
jobs = parsed.get("jobs")
if isinstance(jobs, dict):
for job in jobs.values():
if not isinstance(job, dict):
continue
_collect_go_env_from(job.get("env"), found)
steps = job.get("steps")
if isinstance(steps, list):
for step in steps:
if isinstance(step, dict):
_collect_go_env_from(step.get("env"), found)
return found
def _collect_go_env_from(env_block: Any, out: list[str]) -> None:
if not isinstance(env_block, dict):
return
for key in env_block:
if isinstance(key, str) and key in _GO_ENV_KEYS:
out.append(key)
def _find_line_in(content: str | None, needle: str) -> int | None:
"""Best-effort line number lookup. Returns the 1-indexed line where the
substring first appears, or None. Used to annotate findings with a line
pointer even when the parsed YAML loses position info."""
if not content or not needle:
return None
for idx, line in enumerate(content.splitlines(), start=1):
if needle in line:
return idx
return None
# ---------------------------------------------------------------------------
# R1: pull_request_target + PR-head checkout (TanStack OIDC theft)
# ---------------------------------------------------------------------------
def signal_workflow_trust(change: FileChange) -> list[Finding]:
"""R1. A workflow that combines pull_request_target with a checkout of
the PR head ref is the TanStack mini-Shai-Hulud attack shape — head
code runs with the elevated permissions of the base context, including
secrets and OIDC.
Structural diff: fires when head has the bad combo AND base didn't have
the same dangerous ref value. (Reformatting the same dangerous YAML in
a different style is not a new attack — only newly-introduced danger
fires.)"""
if not is_workflow(change.path) or change.head_content is None:
return []
head = _parse_workflow(change.head_content)
if not _has_pr_target_trigger(_workflow_on(head)):
return []
head_refs = _walk_checkout_refs(head)
if not head_refs:
return []
# Diff-aware: any dangerous ref present on base is pre-existing, not new.
base = _parse_workflow(change.base_content)
base_refs: set[str] = set()
if base is not None and _has_pr_target_trigger(_workflow_on(base)):
base_refs = set(_walk_checkout_refs(base))
new_dangerous = [r for r in head_refs if r not in base_refs]
if not new_dangerous:
return []
danger_text = new_dangerous[0]
line = _find_line_in(change.head_content, danger_text)
return [
Finding(
path=change.path,
line=line,
severity="block",
signal_id="workflow_trust_pr_head_checkout",
message=(
"pull_request_target workflow checks out PR head code "
"(matched: %r). This is the TanStack mini-Shai-Hulud attack "
"shape — head code runs with base-context secrets and OIDC." % danger_text
),
remediation=(
"Use `pull_request` instead, or omit the `ref:` override on "
"actions/checkout so it stays on the base commit. Never run "
"PR head code under pull_request_target."
),
)
]
# ---------------------------------------------------------------------------
# R2: id-token: write outside the publishing allowlist
# ---------------------------------------------------------------------------
def signal_id_token_outside_allowlist(change: FileChange) -> list[Finding]:
"""R2. id-token: write mints OIDC tokens. It must appear only in the
workflow(s) that actually publish. Generator repo's allowlist is empty.
Structural diff: fires only if the head workflow grants id-token: write
and the base didn't (or didn't exist)."""
if not is_workflow(change.path) or change.head_content is None:
return []
if change.path in ID_TOKEN_ALLOWLIST:
return []
head = _parse_workflow(change.head_content)
if not _walk_id_token_grants(head):
return []
base = _parse_workflow(change.base_content)
if base is not None and _walk_id_token_grants(base):
# Pre-existing grant; not introduced by this PR.
return []
line = _find_line_in(change.head_content, "id-token")
allowlist_label = ", ".join(sorted(ID_TOKEN_ALLOWLIST)) or "(none — generator repo has no OIDC workflows on origin/main)"
return [
Finding(
path=change.path,
line=line,
severity="block",
signal_id="id_token_outside_allowlist",
message=(
"id-token: write is granted in a workflow outside the "
"publishing allowlist (%s)." % allowlist_label
),
remediation=(
"Remove the id-token permission. If a publishing workflow "
"with OIDC is being introduced, add it to ID_TOKEN_ALLOWLIST "
"in signals.py in the same PR with reviewer sign-off."
),
)
]
# ---------------------------------------------------------------------------
# R4: GOPROXY / GOFLAGS / GONOSUMCHECK overrides in workflows
# ---------------------------------------------------------------------------
def signal_go_env_override(change: FileChange) -> list[Finding]:
"""R4. Setting GOPROXY / GOFLAGS / GONOSUMCHECK / GOSUMDB inside a
workflow env block lets an attacker redirect module resolution or
suppress checksum verification (BufferZoneCorp).
Structural diff: fires for each go-env key newly set in head that wasn't
set in base."""
if not is_workflow(change.path) or change.head_content is None:
return []
head_vars = set(_walk_go_env_overrides(_parse_workflow(change.head_content)))
if not head_vars:
return []
base_vars: set[str] = set()
if change.base_content is not None:
base_vars = set(_walk_go_env_overrides(_parse_workflow(change.base_content)))
new_vars = head_vars - base_vars
if not new_vars:
return []
findings: list[Finding] = []
for var in sorted(new_vars):
findings.append(
Finding(
path=change.path,
line=_find_line_in(change.head_content, var),
severity="block",
signal_id="go_env_override_in_workflow",
message=(
"Workflow sets %s in an env block. This can redirect Go "
"module resolution to an attacker proxy or suppress "
"checksum verification (BufferZoneCorp attack shape)." % var
),
remediation=(
"Remove the env override. If a private GOPROXY is required, "
"configure it at the org or runner level under operator review, "
"not in a workflow file that PRs can modify."
),
)
)
return findings
# ---------------------------------------------------------------------------
# Signal dispatch
# ---------------------------------------------------------------------------
ALL_SIGNALS = (
signal_workflow_trust,
signal_id_token_outside_allowlist,
signal_go_env_override,
)
def run_signals(change: FileChange) -> list[Finding]:
findings: list[Finding] = []
for sig in ALL_SIGNALS:
findings.extend(sig(change))
return findings