Require the helper at a proven top-level shell position and canonicalize workflow trigger, defaults, job control, and quoted-key semantics with adversarial fixtures.
- commit
- fb928dc01f098e2f7b0212ba92c89dab67b17ba2
- parent
- 85e6a8ecd93723060cb86cf1365cd82dcbdbe13a
- author
- Dami <dami.etoile@gmail.com>
| M | scripts/ | 13 | +- |
| M | scripts/ | 55 | ++++++++- |
| M | scripts/ | 202 | ++++++++++++++++++++++++++++-- |
@@ -5,14 +5,7 @@ script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)"
55 repo_dir="$(cd -- "$script_dir/../.." && pwd -P)"66 77 cd "$repo_dir"8-python3 scripts/verify_ci_cross_build_policy.py .github/workflows/ci.yml8+python3 scripts/verify_ci_cross_build_policy.py \9+ .github/workflows/ci.yml \10+ scripts/verify-demo-build.sh911 python3 -m unittest scripts.tests.test_ci_cross_build_policy -v10-11-gate_call_count="$(12- grep -Fxc './scripts/run-linux-cross-build-gate.sh' \13- scripts/verify-demo-build.sh || true14-)"15-[[ "$gate_call_count" == 1 ]] || {16- echo 'complete repository gate must invoke the Linux cross-build gate exactly once' >&217- exit 118-}@@ -3,13 +3,16 @@ from __future__ import annotations
33 import unittest44 from pathlib import Path55 6-from scripts.verify_ci_cross_build_policy import PolicyError, verify6+from scripts.verify_ci_cross_build_policy import PolicyError, verify, verify_gate_script77 88 99 REPO_DIR = Path(__file__).resolve().parents[2]1010 WORKFLOW = (REPO_DIR / ".github" / "workflows" / "ci.yml").read_text(1111 encoding="utf-8"1212 )13+GATE_SCRIPT = (REPO_DIR / "scripts" / "verify-demo-build.sh").read_text(14+ encoding="utf-8"15+)1316 1417 1518 def replace_once(source: str, old: str, new: str) -> str:@@ -29,6 +32,18 @@ def remove_named_step(source: str, name: str, next_name: str) -> tuple[str, str]
2932 class CiCrossBuildPolicyTests(unittest.TestCase):3033 def test_repository_workflow_passes(self) -> None:3134 verify(WORKFLOW)35+ verify_gate_script(GATE_SCRIPT)36+37+ def test_rejects_cross_build_helper_inside_false_branch(self) -> None:38+ mutated = replace_once(39+ GATE_SCRIPT,40+ "./scripts/run-linux-cross-build-gate.sh",41+ "if false; then\n./scripts/run-linux-cross-build-gate.sh\nfi",42+ )43+ with self.assertRaisesRegex(44+ PolicyError, "must not be nested in conditional or compound"45+ ):46+ verify_gate_script(mutated)3247 3348 def test_rejects_demo_gate_run_bypass_even_with_decoy_comment(self) -> None:3449 mutated = replace_once(@@ -62,6 +77,44 @@ class CiCrossBuildPolicyTests(unittest.TestCase):
6277 with self.assertRaisesRegex(PolicyError, "must not be conditionally disabled"):6378 verify(mutated)6479 80+ def test_rejects_quoted_conditionally_disabled_demo_job(self) -> None:81+ mutated = replace_once(82+ WORKFLOW,83+ " demo-gate:\n name: Complete demo gate",84+ ' demo-gate:\n "if": false\n name: Complete demo gate',85+ )86+ with self.assertRaisesRegex(PolicyError, "must not be conditionally disabled"):87+ verify(mutated)88+89+ def test_rejects_demo_job_continue_on_error(self) -> None:90+ mutated = replace_once(91+ WORKFLOW,92+ " demo-gate:\n name: Complete demo gate",93+ " demo-gate:\n continue-on-error: true\n"94+ " name: Complete demo gate",95+ )96+ with self.assertRaisesRegex(PolicyError, "must not suppress failures"):97+ verify(mutated)98+99+ def test_rejects_missing_pull_request_trigger(self) -> None:100+ mutated = replace_once(WORKFLOW, " pull_request:\n", "")101+ with self.assertRaisesRegex(102+ PolicyError, "unconditional pull_request trigger"103+ ):104+ verify(mutated)105+106+ def test_rejects_failure_suppressing_workflow_shell_default(self) -> None:107+ mutated = replace_once(108+ WORKFLOW,109+ "name: CI\n\non:",110+ "name: CI\n\ndefaults:\n"111+ " run:\n"112+ " shell: 'bash {0} || true'\n\n"113+ "on:",114+ )115+ with self.assertRaisesRegex(PolicyError, "workflow-level defaults are forbidden"):116+ verify(mutated)117+65118 def test_ignores_tokens_in_comments_and_later_jobs(self) -> None:66119 mutated, removed = remove_named_step(67120 WORKFLOW,@@ -13,6 +13,8 @@ from pathlib import Path
1313 CACHE_ACTION = "actions/cache@0400d5f644dc74513175e3cd8d07132dd4860809"1414 CACHE_PATH = "~/.cargo/bin/cargo-zigbuild"1515 CACHE_KEY = "cargo-zigbuild-${{ runner.os }}-${{ runner.arch }}-0.23.0"16+LINUX_CROSS_BUILD_GATE = "./scripts/run-linux-cross-build-gate.sh"17+MAPPING_KEY = r"""(?:[A-Za-z0-9_-]+|"[A-Za-z0-9_-]+"|'[A-Za-z0-9_-]+')"""1618 1719 INSTALL_RUN = """\1820 python3 -m venv "$RUNNER_TEMP/zig-venv"@@ -73,11 +75,71 @@ def _scalar(value: str) -> str:
7375 return value7476 7577 76-def _mapping_key(line: str, indent: int) -> str | None:78+def _mapping_entry(line: str, indent: int) -> tuple[str, str] | None:7779 if _indent(line) != indent or not _is_content(line):7880 return None79- match = re.fullmatch(rf" {{{indent}}}([A-Za-z0-9_-]+):(?:\s.*)?", line)80- return match.group(1) if match else None81+ match = re.fullmatch(82+ rf" {{{indent}}}({MAPPING_KEY}):(?:\s*(.*))?",83+ line,84+ )85+ if not match:86+ return None87+ return _scalar(match.group(1)), (match.group(2) or "").strip()88+89+90+def _mapping_key(line: str, indent: int) -> str | None:91+ entry = _mapping_entry(line, indent)92+ return entry[0] if entry else None93+94+95+def _block_end(lines: list[str], start: int, indent: int) -> int:96+ for index in range(start + 1, len(lines)):97+ if _mapping_entry(lines[index], indent) is not None:98+ return index99+ if _is_content(lines[index]) and _indent(lines[index]) < indent:100+ return index101+ return len(lines)102+103+104+def _validate_workflow_layout(workflow: str) -> None:105+ lines = workflow.splitlines()106+ for line in lines:107+ if _is_content(line) and _indent(line) == 0:108+ if _mapping_entry(line, 0) is None:109+ raise PolicyError(f"cannot parse top-level workflow field: {line}")110+ top_level = [111+ (index, entry)112+ for index, line in enumerate(lines)113+ if (entry := _mapping_entry(line, 0)) is not None114+ ]115+ top_level_names = [key for _, (key, _) in top_level]116+ if len(top_level_names) != len(set(top_level_names)):117+ raise PolicyError("workflow contains a duplicate top-level field")118+119+ if any(key == "defaults" for _, (key, _) in top_level):120+ raise PolicyError(121+ "workflow-level defaults are forbidden because they can suppress gate failures"122+ )123+ allowed_top_level = {"name", "on", "permissions", "concurrency", "jobs"}124+ if set(top_level_names) != allowed_top_level:125+ raise PolicyError("workflow top-level fields changed from the CI policy contract")126+127+ on_entries = [(index, value) for index, (key, value) in top_level if key == "on"]128+ if len(on_entries) != 1:129+ raise PolicyError("workflow must contain exactly one top-level on mapping")130+ on_start, on_value = on_entries[0]131+ if _scalar(on_value):132+ raise PolicyError("workflow on field must be a trigger mapping")133+ on_end = _block_end(lines, on_start, 0)134+ pull_request_entries = [135+ _mapping_entry(line, 2)136+ for line in lines[on_start + 1 : on_end]137+ if _mapping_key(line, 2) == "pull_request"138+ ]139+ if len(pull_request_entries) != 1 or _scalar(pull_request_entries[0][1]):140+ raise PolicyError(141+ "ordinary CI must contain exactly one unconditional pull_request trigger"142+ )81143 82144 83145 def _job_lines(workflow: str, job_name: str) -> list[str]:@@ -85,10 +147,13 @@ def _job_lines(workflow: str, job_name: str) -> list[str]:
85147 jobs_indices = [86148 index87149 for index, line in enumerate(lines)88- if line == "jobs:"150+ if _mapping_key(line, 0) == "jobs"89151 ]90152 if len(jobs_indices) != 1:91153 raise PolicyError("workflow must contain exactly one top-level jobs mapping")154+ jobs_entry = _mapping_entry(lines[jobs_indices[0]], 0)155+ if jobs_entry is None or _scalar(jobs_entry[1]):156+ raise PolicyError("top-level jobs field must be a mapping")92157 93158 jobs_start = jobs_indices[0] + 194159 jobs_end = len(lines)@@ -100,12 +165,15 @@ def _job_lines(workflow: str, job_name: str) -> list[str]:
100165 matches = [101166 index102167 for index in range(jobs_start, jobs_end)103- if re.fullmatch(rf" {re.escape(job_name)}:\s*(?:#.*)?", lines[index])168+ if _mapping_key(lines[index], 2) == job_name104169 ]105170 if len(matches) != 1:106171 raise PolicyError(f"jobs.{job_name} must exist exactly once")107172 108173 job_start = matches[0]174+ job_entry = _mapping_entry(lines[job_start], 2)175+ if job_entry is None or _scalar(job_entry[1]):176+ raise PolicyError(f"jobs.{job_name} must be a mapping")109177 job_end = jobs_end110178 for index in range(job_start + 1, jobs_end):111179 if _mapping_key(lines[index], 2) is not None:@@ -128,10 +196,10 @@ def _parse_step(block: list[str]) -> Step:
128196 continue129197 if _indent(line) != 8:130198 raise PolicyError(f"unexpected YAML structure in step {step.name!r}: {line}")131- match = re.fullmatch(r" ([A-Za-z0-9_-]+):(?:\s*(.*))?", line)199+ match = re.fullmatch(rf" ({MAPPING_KEY}):(?:\s*(.*))?", line)132200 if not match:133201 raise PolicyError(f"cannot parse field in step {step.name!r}: {line}")134- key = match.group(1)202+ key = _scalar(match.group(1))135203 value = (match.group(2) or "").strip()136204 if key in step.field_names:137205 raise PolicyError(f"duplicate field {key!r} in step {step.name!r}")@@ -166,13 +234,13 @@ def _parse_step(block: list[str]) -> Step:
166234 if _indent(child) <= 8:167235 break168236 child_match = re.fullmatch(169- r" ([A-Za-z0-9_-]+):\s*(.+?)\s*", child237+ rf" ({MAPPING_KEY}):\s*(.+?)\s*", child170238 )171239 if not child_match:172240 raise PolicyError(173241 f"cannot parse {key!r} mapping in step {step.name!r}: {child}"174242 )175- child_key = child_match.group(1)243+ child_key = _scalar(child_match.group(1))176244 if child_key in entries:177245 raise PolicyError(178246 f"duplicate {key}.{child_key} in step {step.name!r}"@@ -189,11 +257,28 @@ def _parse_step(block: list[str]) -> Step:
189257 190258 191259 def _parse_steps(job: list[str]) -> list[Step]:192- if any(193- _mapping_key(line, 4) == "if"194- for line in job195- ):260+ job_entries: list[tuple[str, str]] = []261+ for line in job:262+ if not _is_content(line) or _indent(line) != 4:263+ continue264+ entry = _mapping_entry(line, 4)265+ if entry is None:266+ raise PolicyError(f"cannot parse jobs.demo-gate field: {line}")267+ job_entries.append(entry)268+ job_field_list = [key for key, _ in job_entries]269+ if len(job_field_list) != len(set(job_field_list)):270+ raise PolicyError("jobs.demo-gate contains a duplicate field")271+ job_fields = set(job_field_list)272+ if "if" in job_fields:196273 raise PolicyError("jobs.demo-gate must not be conditionally disabled")274+ if "continue-on-error" in job_fields:275+ raise PolicyError("jobs.demo-gate must not suppress failures")276+ if "defaults" in job_fields:277+ raise PolicyError(278+ "jobs.demo-gate must not override run defaults or shell failure behavior"279+ )280+ if job_fields != {"name", "runs-on", "timeout-minutes", "env", "steps"}:281+ raise PolicyError("jobs.demo-gate fields changed from the CI policy contract")197282 198283 steps_indices = [199284 index@@ -230,7 +315,96 @@ def _named_step(steps: list[Step], name: str) -> tuple[int, Step]:
230315 return matches[0]231316 232317 318+def _shell_compound_stack(lines: list[str], stop: int) -> list[str]:319+ stack: list[str] = []320+ closers = {321+ "fi": "if",322+ "done": "loop",323+ "esac": "case",324+ "}": "brace",325+ ")": "subshell",326+ }327+328+ for number, line in enumerate(lines[:stop], start=1):329+ stripped = line.strip()330+ if not stripped or stripped.startswith("#"):331+ continue332+333+ if stripped in closers:334+ expected = closers[stripped]335+ if not stack or stack[-1] != expected:336+ raise PolicyError(337+ f"cannot prove shell control flow before cross-build gate at line {number}"338+ )339+ stack.pop()340+ continue341+342+ if re.match(343+ r"^(?:function\s+)?[A-Za-z_][A-Za-z0-9_]*(?:\(\))?\s*\{\s*$",344+ stripped,345+ ):346+ stack.append("brace")347+ elif re.search(r"(?:^|\|\||&&)\s*\{\s*$", stripped):348+ stack.append("brace")349+ elif re.search(r"(?:^|\|\||&&)\s*\(\s*$", stripped):350+ stack.append("subshell")351+ elif re.match(r"^if\b", stripped):352+ stack.append("if")353+ elif re.match(r"^(?:for|while|until|select)\b", stripped):354+ stack.append("loop")355+ elif re.match(r"^case\b", stripped):356+ stack.append("case")357+358+ return stack359+360+361+def verify_gate_script(gate_script: str) -> None:362+ lines = gate_script.splitlines()363+ calls = [364+ index365+ for index, line in enumerate(lines)366+ if line.strip() == LINUX_CROSS_BUILD_GATE367+ ]368+ if len(calls) != 1:369+ raise PolicyError(370+ "complete repository gate must invoke the Linux cross-build gate exactly once"371+ )372+373+ call = calls[0]374+ if lines[call] != LINUX_CROSS_BUILD_GATE:375+ raise PolicyError(376+ "Linux cross-build gate must be an unindented top-level simple command"377+ )378+ if _shell_compound_stack(lines, call):379+ raise PolicyError(380+ "Linux cross-build gate must not be nested in conditional or compound shell control flow"381+ )382+383+ content_before = [384+ line.strip()385+ for line in lines[:call]386+ if _is_content(line)387+ ]388+ content_after = [389+ line.strip()390+ for line in lines[call + 1 :]391+ if _is_content(line)392+ ]393+ if not content_before or content_before[-1] != "fi":394+ raise PolicyError(395+ "Linux cross-build gate placement changed before the required frontend gates"396+ )397+ if (398+ not content_after399+ or content_after[0] != "while IFS= read -r frontend_source; do"400+ ):401+ raise PolicyError(402+ "Linux cross-build gate placement changed before the required frontend gates"403+ )404+405+233406 def verify(workflow: str) -> None:407+ _validate_workflow_layout(workflow)234408 steps = _parse_steps(_job_lines(workflow, "demo-gate"))235409 236410 dependency_index, dependency = _named_step(@@ -279,9 +453,11 @@ def verify(workflow: str) -> None:
279453 def main() -> int:280454 parser = argparse.ArgumentParser()281455 parser.add_argument("workflow", type=Path)456+ parser.add_argument("gate_script", type=Path)282457 args = parser.parse_args()283458 try:284459 verify(args.workflow.read_text(encoding="utf-8"))460+ verify_gate_script(args.gate_script.read_text(encoding="utf-8"))285461 except (OSError, PolicyError) as error:286462 print(f"CI cross-build policy failed: {error}", file=sys.stderr)287463 return 1