From daae1efa4df3665f4b08fb3ffaa9a4e21616fe80 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:35:03 +0300 Subject: [PATCH 01/37] docs(af01): inventory workflow trust authority --- .../stack-a-inventory.md | 110 ++++++++++++++++++ 1 file changed, 110 insertions(+) create mode 100644 specs/015-af-01-trusted-development-baseline/stack-a-inventory.md diff --git a/specs/015-af-01-trusted-development-baseline/stack-a-inventory.md b/specs/015-af-01-trusted-development-baseline/stack-a-inventory.md new file mode 100644 index 00000000..5eb0c916 --- /dev/null +++ b/specs/015-af-01-trusted-development-baseline/stack-a-inventory.md @@ -0,0 +1,110 @@ +# AF-01 Stack A Workflow Trust Inventory + +Status: IMPLEMENTATION_EVIDENCE / T010 + +Canonical inventory base: + +```text +main: eeecb0bc03c7040bb18b70bce8b69d618384f783 +tree: d5abe932f1436a9612f45bf130ba29aadbc5a133 +AF-01 planning: CANONICAL +``` + +This inventory records the tracked GitHub workflow/action authority that AF-01 Stack A must make machine-checkable. It is not a substitute for the repository-owned discovery audit: the audit must discover future workflow and Action metadata files automatically. + +## Tracked workflow files + +1. `.github/workflows/ci.yml` +2. `.github/workflows/cf06-oracle.yml` +3. `.github/workflows/cf11-multi-version-proof.yml` +4. `.github/workflows/cf11g-context-proof.yml` +5. `.github/workflows/cf12-impact-proof.yml` +6. `.github/workflows/cf13-quality-gate-proof.yml` +7. `.github/workflows/registry-download-smoke.yml` + +## Tracked Action metadata + +- `action.yml` + +No `action.yaml` metadata file is present at the inventory base. AF-01 discovery still treats both `action.yml` and `action.yaml` at any tracked path as authoritative scan inputs. + +## Job authority inventory + +| Workflow | Job | Current effective permission | Current runner | Current container | Current timeout | Stack A disposition | +|---|---|---|---|---|---|---| +| `ci.yml` | `rust` | `contents: read` | `ubuntu-latest` | none | none | pin runner/actions, disable checkout credentials, add timeout | +| `cf06-oracle.yml` | `oracle-self-smoke` | `contents: read` | `ubuntu-latest` | none | none | fixed runner, explicit job permission, timeout | +| `cf06-oracle.yml` | `oracle-changed-profile` | `contents: read` | `ubuntu-latest` | none | none | fixed runner, explicit job permission, timeout | +| `cf06-oracle.yml` | `oracle-proof` | inherited `contents: read` | `ubuntu-latest` | none | none | reduce to no repository permission, fixed runner, timeout | +| `cf11-multi-version-proof.yml` | `real-package-graph` | `contents: read` | `ubuntu-24.04` | Rust digest pinned | 20m | retain | +| `cf11g-context-proof.yml` | `deterministic-context-cli` | `contents: read` | `ubuntu-24.04` | Rust digest pinned | 15m | retain | +| `cf12-impact-proof.yml` | `deterministic-impact-cli` | `contents: read` | `ubuntu-24.04` | Rust digest pinned | 15m | retain | +| `cf13-quality-gate-proof.yml` | `deterministic-quality-gate` | `contents: read` | `ubuntu-24.04` | Rust digest pinned | 15m | retain | +| `registry-download-smoke.yml` | `registry-download` | `contents: read` | `ubuntu-latest` | none | 15m | fixed runner | + +## External Action / reusable-workflow references + +Immutable references already used by current proof workflows: + +```text +actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 +actions/setup-java@b6effb05e454b25005698d916606bdc6ffcbf961 +dtolnay/rust-toolchain@032958afbdc797a9164d3bc0b56325c1308924a5 +actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 +``` + +Current `ci.yml` is the outlier: + +```text +actions/checkout@v4 +dtolnay/rust-toolchain@1.97.1 +``` + +`action.yml` is a local composite Action and contains no external `uses:` reference at this base. + +## Checkout credential inventory + +Every current checkout in the proof/oracle/registry workflows sets: + +```yaml +persist-credentials: false +``` + +`ci.yml` does not and must be reconciled. + +## Container identity inventory + +Current containerized proof jobs use digest-bound Rust images. No service containers are present at this base. AF-01 policy applies the digest rule to every future job or service container that appears; a mutable tag is never accepted merely because it is new or non-proof-labeled. + +## Cargo lockfile-consuming command inventory + +Current workflow commands that build/check/test/run against the Rust dependency graph already use `--locked` in the proof/oracle/registry workflows and in the relevant `ci.yml` clippy/test/run invocations. `cargo fmt` and `cargo --version` are not lockfile-consuming commands and are outside this rule. + +AF-01 audit treats at least these cargo subcommands as lockfile-consuming when present in workflow shell commands: + +```text +bench +build +check +clippy +doc +metadata +run +test +``` + +## Machine-checkable target after T014/T015 + +- every discovered workflow has an exact policy entry for every discovered job; +- effective workflow/job permissions equal the policy declaration, with no unresolved GitHub default authority; +- all current runners become `ubuntu-24.04`; +- every job has an explicit bounded `timeout-minutes`; +- all external `uses:` references are full 40-hex commit SHAs; +- all checkout steps persist no credentials; +- every job/service container reference, if present, is digest-bound with `sha256`; +- all lockfile-consuming cargo invocations use `--locked`; +- every tracked `action.yml` and `action.yaml` is scanned for external `uses:` references. + +## Scope boundary + +Stack A changes development-assurance configuration only. It does not change commandF product semantics, CF-06 production oracle identity, the CF-10 frozen corpus, report schemas, or runtime authority. From 0f373b618a7c38b116e325fe785ab4cd3342b64f Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:35:47 +0300 Subject: [PATCH 02/37] feat(af01): define workflow trust policy --- .github/workflow-trust-policy.json | 110 +++++++++++++++++++++++++++++ 1 file changed, 110 insertions(+) create mode 100644 .github/workflow-trust-policy.json diff --git a/.github/workflow-trust-policy.json b/.github/workflow-trust-policy.json new file mode 100644 index 00000000..7fdcb295 --- /dev/null +++ b/.github/workflow-trust-policy.json @@ -0,0 +1,110 @@ +{ + "schema": 1, + "rules": { + "cargo_locked_subcommands": [ + "bench", + "build", + "check", + "clippy", + "doc", + "metadata", + "run", + "test" + ], + "require_container_digest": true, + "require_checkout_credentials_disabled": true, + "require_external_uses_full_sha": true + }, + "workflows": { + ".github/workflows/cf06-oracle.yml": { + "jobs": { + "oracle-changed-profile": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 25 + }, + "oracle-proof": { + "permissions": {}, + "runner": "ubuntu-24.04", + "timeout_minutes": 5 + }, + "oracle-self-smoke": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 20 + } + } + }, + ".github/workflows/cf11-multi-version-proof.yml": { + "jobs": { + "real-package-graph": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 20 + } + } + }, + ".github/workflows/cf11g-context-proof.yml": { + "jobs": { + "deterministic-context-cli": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 15 + } + } + }, + ".github/workflows/cf12-impact-proof.yml": { + "jobs": { + "deterministic-impact-cli": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 15 + } + } + }, + ".github/workflows/cf13-quality-gate-proof.yml": { + "jobs": { + "deterministic-quality-gate": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 15 + } + } + }, + ".github/workflows/ci.yml": { + "jobs": { + "rust": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 30 + } + } + }, + ".github/workflows/registry-download-smoke.yml": { + "jobs": { + "registry-download": { + "permissions": { + "contents": "read" + }, + "runner": "ubuntu-24.04", + "timeout_minutes": 15 + } + } + } + }, + "exceptions": [] +} From d602343272512f338472c2e525a991ce3effc02d Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:36:54 +0300 Subject: [PATCH 03/37] feat(af01): add deterministic workflow trust audit --- .github/scripts/audit_workflow_trust.py | 411 ++++++++++++++++++++++++ 1 file changed, 411 insertions(+) create mode 100644 .github/scripts/audit_workflow_trust.py diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py new file mode 100644 index 00000000..a30c0c53 --- /dev/null +++ b/.github/scripts/audit_workflow_trust.py @@ -0,0 +1,411 @@ +#!/usr/bin/env python3 +"""Deterministic repository-owned GitHub workflow trust audit for AF-01.""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path +from typing import Iterable + +FULL_SHA_RE = re.compile(r"^[0-9a-fA-F]{40}$") +DIGEST_IMAGE_RE = re.compile(r"@sha256:[0-9a-fA-F]{64}$") +JOB_RE = re.compile(r"^ ([A-Za-z0-9_.-]+):\s*(?:#.*)?$") +USES_RE = re.compile(r"^(\s*)(?:-\s*)?uses:\s*(.+?)\s*$") +STEP_USES_RE = re.compile(r"^(\s*)-\s+uses:\s*(.+?)\s*$") +PERMISSION_RE = re.compile(r"^([A-Za-z0-9_-]+):\s*(read|write|none)\s*$") +CARGO_RE = re.compile(r"\bcargo\s+(bench|build|check|clippy|doc|metadata|run|test)\b") + + +@dataclass(frozen=True, order=True) +class Finding: + code: str + path: str + job: str + detail: str + + def as_dict(self) -> dict[str, str]: + result = {"code": self.code, "path": self.path, "detail": self.detail} + if self.job: + result["job"] = self.job + return result + + +def _indent(line: str) -> int: + return len(line) - len(line.lstrip(" ")) + + +def _scalar(value: str) -> str: + value = value.strip() + if " #" in value: + value = value.split(" #", 1)[0].rstrip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}: + value = value[1:-1] + return value + + +def _tracked_files(root: Path) -> list[str]: + completed = subprocess.run( + ["git", "ls-files", "-z"], + cwd=root, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + return sorted(path for path in completed.stdout.decode("utf-8").split("\0") if path) + + +def discover_security_files(paths: Iterable[str]) -> tuple[list[str], list[str]]: + workflows: list[str] = [] + actions: list[str] = [] + for path in sorted(paths): + name = Path(path).name + if path.startswith(".github/workflows/") and name.endswith((".yml", ".yaml")): + workflows.append(path) + if name in {"action.yml", "action.yaml"}: + actions.append(path) + return workflows, actions + + +def _parse_permissions( + lines: list[str], start: int, end: int, indent: int +) -> tuple[dict[str, str] | None, str | None]: + prefix = " " * indent + "permissions:" + for index in range(start, end): + raw = lines[index] + if not raw.startswith(prefix) or _indent(raw) != indent: + continue + value = _scalar(raw[len(prefix) :]) + if value == "{}": + return {}, None + if value: + return None, f"unsupported permissions scalar {value!r}" + permissions: dict[str, str] = {} + cursor = index + 1 + while cursor < end: + child = lines[cursor] + if not child.strip() or child.lstrip().startswith("#"): + cursor += 1 + continue + child_indent = _indent(child) + if child_indent <= indent: + break + if child_indent != indent + 2: + return None, "permissions block contains unsupported nested syntax" + parsed = PERMISSION_RE.match(child.strip()) + if not parsed: + return None, f"unsupported permission entry {child.strip()!r}" + key, permission = parsed.groups() + if key in permissions: + return None, f"duplicate permission key {key!r}" + permissions[key] = permission + cursor += 1 + return permissions, None + return None, None + + +def _job_ranges(lines: list[str]) -> tuple[dict[str, tuple[int, int]], str | None]: + jobs_index = next((i for i, line in enumerate(lines) if line.strip() == "jobs:" and _indent(line) == 0), None) + if jobs_index is None: + return {}, "workflow has no top-level jobs mapping" + + starts: list[tuple[str, int]] = [] + for index in range(jobs_index + 1, len(lines)): + line = lines[index] + if line.strip() and _indent(line) == 0: + break + matched = JOB_RE.match(line) + if matched: + starts.append((matched.group(1), index)) + if not starts: + return {}, "workflow jobs mapping has no statically named jobs" + if len({name for name, _ in starts}) != len(starts): + return {}, "workflow contains duplicate statically named jobs" + + ranges: dict[str, tuple[int, int]] = {} + for position, (name, start) in enumerate(starts): + end = starts[position + 1][1] if position + 1 < len(starts) else len(lines) + for index in range(start + 1, end): + line = lines[index] + if line.strip() and _indent(line) == 0: + end = index + break + ranges[name] = (start, end) + return ranges, None + + +def _job_scalar(lines: list[str], start: int, end: int, key: str) -> str | None: + prefix = " " + key + ":" + for index in range(start + 1, end): + line = lines[index] + if _indent(line) == 4 and line.startswith(prefix): + return _scalar(line[len(prefix) :]) + return None + + +def _container_images(lines: list[str], start: int, end: int) -> list[str]: + images: list[str] = [] + for index in range(start + 1, end): + line = lines[index] + stripped = line.strip() + if _indent(line) == 4 and stripped.startswith("container:"): + value = _scalar(stripped.split(":", 1)[1]) + if value: + images.append(value) + if stripped.startswith("image:") and _indent(line) >= 6: + value = _scalar(stripped.split(":", 1)[1]) + if value: + images.append(value) + return images + + +def _all_uses(lines: list[str], start: int = 0, end: int | None = None) -> list[tuple[int, str]]: + if end is None: + end = len(lines) + result: list[tuple[int, str]] = [] + for index in range(start, end): + matched = USES_RE.match(lines[index]) + if matched: + result.append((index, _scalar(matched.group(2)))) + return result + + +def _external_ref_is_immutable(reference: str) -> bool: + if reference.startswith("./"): + return True + if reference.startswith("docker://"): + return bool(DIGEST_IMAGE_RE.search(reference)) + if "@" not in reference: + return False + _, revision = reference.rsplit("@", 1) + return bool(FULL_SHA_RE.fullmatch(revision)) + + +def _checkout_has_credentials_disabled(lines: list[str], uses_index: int) -> bool: + matched = STEP_USES_RE.match(lines[uses_index]) + if not matched: + return False + step_indent = len(matched.group(1)) + end = len(lines) + for index in range(uses_index + 1, len(lines)): + line = lines[index] + if re.match(rf"^ {{{step_indent}}}-\s+", line): + end = index + break + for line in lines[uses_index + 1 : end]: + if re.match(r"^\s*persist-credentials:\s*false\s*(?:#.*)?$", line): + return True + return False + + +def _valid_exception(exception: object) -> bool: + if not isinstance(exception, dict): + return False + required = {"rule", "path", "reason", "revisit"} + if not required.issubset(exception): + return False + if not all(isinstance(exception[key], str) and exception[key].strip() for key in required): + return False + if len(exception["reason"].strip()) < 10 or len(exception["revisit"].strip()) < 5: + return False + return set(exception).issubset(required | {"job", "detail"}) + + +def _excepted(policy: dict, finding: Finding) -> bool: + for exception in policy.get("exceptions", []): + if exception.get("rule") != finding.code or exception.get("path") != finding.path: + continue + if exception.get("job", finding.job) != finding.job: + continue + if "detail" in exception and exception["detail"] != finding.detail: + continue + return True + return False + + +def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[Finding]: + findings: list[Finding] = [] + if "\t" in text: + findings.append(Finding("malformed_yaml", path, "", "tab indentation is not supported")) + return findings + lines = text.splitlines() + jobs, jobs_error = _job_ranges(lines) + if jobs_error: + findings.append(Finding("malformed_yaml", path, "", jobs_error)) + return findings + + expected_jobs = expected.get("jobs") + if not isinstance(expected_jobs, dict): + findings.append(Finding("invalid_policy", path, "", "workflow policy must contain a jobs object")) + return findings + actual_names = set(jobs) + expected_names = set(expected_jobs) + for name in sorted(actual_names - expected_names): + findings.append(Finding("unplanned_job", path, name, "job is not declared in workflow trust policy")) + for name in sorted(expected_names - actual_names): + findings.append(Finding("missing_job", path, name, "policy job is missing from workflow")) + + top_permissions, top_permission_error = _parse_permissions(lines, 0, len(lines), 0) + if top_permission_error: + findings.append(Finding("permissions_syntax", path, "", top_permission_error)) + + for job in sorted(actual_names & expected_names): + start, end = jobs[job] + expected_job = expected_jobs[job] + runner = _job_scalar(lines, start, end, "runs-on") + if runner != expected_job.get("runner"): + findings.append( + Finding("runner_mismatch", path, job, f"expected runner {expected_job.get('runner')!r}, found {runner!r}") + ) + if runner and runner.endswith("-latest"): + findings.append(Finding("mutable_runner", path, job, f"runner {runner!r} is a mutable latest label")) + + timeout = _job_scalar(lines, start, end, "timeout-minutes") + try: + timeout_value = int(timeout) if timeout is not None else None + except ValueError: + timeout_value = None + timeout_limit = expected_job.get("timeout_minutes") + if not isinstance(timeout_limit, int) or timeout_limit <= 0: + findings.append(Finding("invalid_policy", path, job, "timeout_minutes must be a positive integer")) + elif timeout_value is None or timeout_value <= 0 or timeout_value > timeout_limit: + findings.append( + Finding("timeout_policy", path, job, f"timeout must be 1..{timeout_limit} minutes, found {timeout!r}") + ) + + job_permissions, job_permission_error = _parse_permissions(lines, start + 1, end, 4) + if job_permission_error: + findings.append(Finding("permissions_syntax", path, job, job_permission_error)) + effective_permissions = job_permissions if job_permissions is not None else top_permissions + if effective_permissions is None: + findings.append( + Finding("unresolved_permissions", path, job, "job inherits undocumented GitHub default token permissions") + ) + elif effective_permissions != expected_job.get("permissions"): + findings.append( + Finding( + "permission_mismatch", + path, + job, + f"expected effective permissions {expected_job.get('permissions')!r}, found {effective_permissions!r}", + ) + ) + + if policy["rules"].get("require_container_digest", False): + for image in _container_images(lines, start, end): + if not DIGEST_IMAGE_RE.search(image): + findings.append( + Finding("mutable_container", path, job, f"container image is not sha256 digest-bound: {image}") + ) + + locked_subcommands = set(policy["rules"].get("cargo_locked_subcommands", [])) + for line in lines[start + 1 : end]: + command = line.strip() + for match in CARGO_RE.finditer(command): + subcommand = match.group(1) + if subcommand in locked_subcommands and "--locked" not in command[match.start() :]: + findings.append( + Finding("cargo_unlocked", path, job, f"cargo {subcommand} invocation omits --locked: {command}") + ) + + for index, reference in _all_uses(lines): + if policy["rules"].get("require_external_uses_full_sha", False) and not _external_ref_is_immutable(reference): + findings.append(Finding("mutable_uses", path, "", f"uses reference is not immutable: {reference}")) + if reference.startswith("actions/checkout@") and policy["rules"].get( + "require_checkout_credentials_disabled", False + ): + if not _checkout_has_credentials_disabled(lines, index): + findings.append( + Finding("checkout_credentials", path, "", "checkout step does not set persist-credentials: false") + ) + + return [finding for finding in findings if not _excepted(policy, finding)] + + +def audit_action_metadata(path: str, text: str, policy: dict) -> list[Finding]: + findings: list[Finding] = [] + if "\t" in text: + return [Finding("malformed_yaml", path, "", "tab indentation is not supported")] + lines = text.splitlines() + if not any(line.strip() == "runs:" for line in lines): + findings.append(Finding("malformed_action_metadata", path, "", "Action metadata has no runs mapping")) + for _, reference in _all_uses(lines): + if policy["rules"].get("require_external_uses_full_sha", False) and not _external_ref_is_immutable(reference): + findings.append(Finding("mutable_uses", path, "", f"uses reference is not immutable: {reference}")) + return [finding for finding in findings if not _excepted(policy, finding)] + + +def audit_repository(root: Path, policy: dict, tracked_files: Iterable[str] | None = None) -> dict: + findings: list[Finding] = [] + if policy.get("schema") != 1 or not isinstance(policy.get("rules"), dict): + findings.append(Finding("invalid_policy", ".github/workflow-trust-policy.json", "", "unsupported policy schema")) + exceptions = policy.get("exceptions", []) + if not isinstance(exceptions, list) or any(not _valid_exception(item) for item in exceptions): + findings.append( + Finding("invalid_policy", ".github/workflow-trust-policy.json", "", "every exception requires bounded rule/path/reason/revisit fields") + ) + + paths = list(tracked_files) if tracked_files is not None else _tracked_files(root) + workflows, actions = discover_security_files(paths) + expected_workflows = policy.get("workflows", {}) + if not isinstance(expected_workflows, dict): + expected_workflows = {} + findings.append(Finding("invalid_policy", ".github/workflow-trust-policy.json", "", "workflows must be an object")) + + for path in sorted(set(workflows) - set(expected_workflows)): + findings.append(Finding("unplanned_workflow", path, "", "tracked workflow is absent from policy")) + for path in sorted(set(expected_workflows) - set(workflows)): + findings.append(Finding("missing_workflow", path, "", "policy workflow is not tracked")) + + for path in sorted(set(workflows) & set(expected_workflows)): + findings.extend(audit_workflow(path, (root / path).read_text(encoding="utf-8"), expected_workflows[path], policy)) + for path in actions: + findings.extend(audit_action_metadata(path, (root / path).read_text(encoding="utf-8"), policy)) + + filtered = sorted(finding for finding in findings if not _excepted(policy, finding)) + return { + "schema": 1, + "ok": not filtered, + "workflows": workflows, + "action_metadata": actions, + "findings": [finding.as_dict() for finding in filtered], + } + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--root", type=Path, default=Path(".")) + parser.add_argument("--policy", type=Path, default=Path(".github/workflow-trust-policy.json")) + args = parser.parse_args() + + root = args.root.resolve() + policy_path = args.policy if args.policy.is_absolute() else root / args.policy + try: + policy = json.loads(policy_path.read_text(encoding="utf-8")) + result = audit_repository(root, policy) + except (OSError, UnicodeError, json.JSONDecodeError, subprocess.CalledProcessError) as error: + result = { + "schema": 1, + "ok": False, + "workflows": [], + "action_metadata": [], + "findings": [ + { + "code": "audit_operational_failure", + "path": str(args.policy), + "detail": str(error), + } + ], + } + + rendered = json.dumps(result, indent=2, sort_keys=True, separators=(",", ": ")) + "\n" + sys.stdout.write(rendered) + return 0 if result["ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) From cbac27bf7e93d20000387b2b4e89b908bb3040e5 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:52:33 +0300 Subject: [PATCH 04/37] test(af01): cover workflow trust counterexamples --- .github/scripts/test_audit_workflow_trust.py | 261 +++++++++++++++++++ 1 file changed, 261 insertions(+) create mode 100644 .github/scripts/test_audit_workflow_trust.py diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py new file mode 100644 index 00000000..48e6e455 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust.py @@ -0,0 +1,261 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import copy +import importlib.util +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("audit_workflow_trust.py") +SPEC = importlib.util.spec_from_file_location("audit_workflow_trust", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +AUDIT = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(AUDIT) + +WORKFLOW = ".github/workflows/example.yml" +ACTION_YAML = "tools/example/action.yaml" +CHECKOUT_SHA = "fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09" +RUST_SHA = "032958afbdc797a9164d3bc0b56325c1308924a5" +CONTAINER_DIGEST = "9" * 64 + + +def policy() -> dict: + return { + "schema": 1, + "rules": { + "cargo_locked_subcommands": [ + "bench", + "build", + "check", + "clippy", + "doc", + "metadata", + "run", + "test", + ], + "require_container_digest": True, + "require_checkout_credentials_disabled": True, + "require_external_uses_full_sha": True, + }, + "workflows": { + WORKFLOW: { + "jobs": { + "build": { + "permissions": {"contents": "read"}, + "runner": "ubuntu-24.04", + "timeout_minutes": 10, + } + } + } + }, + "exceptions": [], + } + + +def valid_workflow(extra_steps: str = "") -> str: + return f"""name: example +on: + pull_request: +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-24.04 + timeout-minutes: 10 + container: + image: docker.io/library/rust@sha256:{CONTAINER_DIGEST} + steps: + - uses: actions/checkout@{CHECKOUT_SHA} + with: + persist-credentials: false + - uses: dtolnay/rust-toolchain@{RUST_SHA} + - name: Test + run: cargo test --locked --workspace +{extra_steps}""" + + +def valid_action() -> str: + return """name: example action +description: fixture +runs: + using: composite + steps: + - shell: bash + run: echo ok +""" + + +class WorkflowTrustAuditTests(unittest.TestCase): + def run_repo( + self, + workflow: str | None = None, + action: str | None = None, + audit_policy: dict | None = None, + tracked: list[str] | None = None, + ) -> dict: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + paths: list[str] = [] + if workflow is not None: + path = root / WORKFLOW + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(workflow, encoding="utf-8") + paths.append(WORKFLOW) + if action is not None: + path = root / ACTION_YAML + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(action, encoding="utf-8") + paths.append(ACTION_YAML) + return AUDIT.audit_repository( + root, + copy.deepcopy(audit_policy if audit_policy is not None else policy()), + tracked_files=tracked if tracked is not None else paths, + ) + + @staticmethod + def codes(result: dict) -> list[str]: + return [finding["code"] for finding in result["findings"]] + + def test_valid_fixture_passes_and_is_deterministic(self) -> None: + first = self.run_repo(valid_workflow(), valid_action()) + second = self.run_repo(valid_workflow(), valid_action()) + self.assertTrue(first["ok"], first) + self.assertEqual(first, second) + self.assertEqual(first["workflows"], [WORKFLOW]) + self.assertEqual(first["action_metadata"], [ACTION_YAML]) + + def test_discovers_both_action_metadata_filenames_anywhere(self) -> None: + workflows, actions = AUDIT.discover_security_files( + [ + ".github/workflows/a.yml", + ".github/workflows/b.yaml", + "action.yml", + "nested/one/action.yaml", + "nested/two/not-action.yml", + ] + ) + self.assertEqual(workflows, [".github/workflows/a.yml", ".github/workflows/b.yaml"]) + self.assertEqual(actions, ["action.yml", "nested/one/action.yaml"]) + + def test_new_workflow_not_in_policy_fails_closed(self) -> None: + result = self.run_repo( + valid_workflow(), + tracked=[WORKFLOW, ".github/workflows/unplanned.yaml"], + ) + self.assertIn("unplanned_workflow", self.codes(result)) + self.assertFalse(result["ok"]) + + def test_mutable_external_action_tag_is_rejected(self) -> None: + workflow = valid_workflow().replace( + f"actions/checkout@{CHECKOUT_SHA}", "actions/checkout@v5" + ) + result = self.run_repo(workflow) + self.assertIn("mutable_uses", self.codes(result)) + + def test_short_sha_and_branch_refs_are_rejected(self) -> None: + for reference in ("owner/action@abc1234", "owner/action@main"): + with self.subTest(reference=reference): + workflow = valid_workflow( + f" - uses: {reference}\n" + ) + result = self.run_repo(workflow) + self.assertIn("mutable_uses", self.codes(result)) + + def test_mutable_reusable_workflow_reference_is_rejected(self) -> None: + workflow = valid_workflow( + " - uses: owner/repository/.github/workflows/reuse.yml@main\n" + ) + result = self.run_repo(workflow) + self.assertIn("mutable_uses", self.codes(result)) + + def test_mutable_external_uses_in_nested_action_yaml_is_rejected(self) -> None: + action = """name: nested +description: fixture +runs: + using: composite + steps: + - uses: owner/action@v1 +""" + result = self.run_repo(valid_workflow(), action) + self.assertIn("mutable_uses", self.codes(result)) + + def test_checkout_credentials_must_be_disabled(self) -> None: + workflow = valid_workflow().replace( + " with:\n persist-credentials: false\n", "" + ) + result = self.run_repo(workflow) + self.assertIn("checkout_credentials", self.codes(result)) + + def test_unresolved_default_permissions_fail_closed(self) -> None: + workflow = valid_workflow().replace( + "permissions:\n contents: read\n", "" + ) + result = self.run_repo(workflow) + self.assertIn("unresolved_permissions", self.codes(result)) + + def test_overbroad_permission_is_rejected(self) -> None: + workflow = valid_workflow().replace( + "permissions:\n contents: read\n", + "permissions:\n contents: write\n", + ) + result = self.run_repo(workflow) + self.assertIn("permission_mismatch", self.codes(result)) + + def test_mutable_runner_is_rejected(self) -> None: + workflow = valid_workflow().replace("ubuntu-24.04", "ubuntu-latest", 1) + expected = policy() + expected["workflows"][WORKFLOW]["jobs"]["build"]["runner"] = "ubuntu-latest" + result = self.run_repo(workflow, audit_policy=expected) + self.assertIn("mutable_runner", self.codes(result)) + + def test_missing_or_excessive_timeout_is_rejected(self) -> None: + for workflow in ( + valid_workflow().replace(" timeout-minutes: 10\n", ""), + valid_workflow().replace("timeout-minutes: 10", "timeout-minutes: 11"), + ): + with self.subTest(): + result = self.run_repo(workflow) + self.assertIn("timeout_policy", self.codes(result)) + + def test_mutable_job_and_service_container_images_are_rejected(self) -> None: + mutable_job = valid_workflow().replace( + f"docker.io/library/rust@sha256:{CONTAINER_DIGEST}", "rust:1.97.1" + ) + result = self.run_repo(mutable_job) + self.assertIn("mutable_container", self.codes(result)) + + mutable_service = valid_workflow().replace( + " steps:\n", + " services:\n database:\n image: postgres:18\n steps:\n", + ) + result = self.run_repo(mutable_service) + self.assertIn("mutable_container", self.codes(result)) + + def test_unlocked_cargo_command_is_rejected(self) -> None: + workflow = valid_workflow().replace( + "cargo test --locked --workspace", "cargo test --workspace" + ) + result = self.run_repo(workflow) + self.assertIn("cargo_unlocked", self.codes(result)) + + def test_malformed_workflow_fails_closed(self) -> None: + result = self.run_repo("name: broken\n\tjobs:\n") + self.assertIn("malformed_yaml", self.codes(result)) + self.assertFalse(result["ok"]) + + def test_missing_jobs_mapping_fails_closed(self) -> None: + result = self.run_repo("name: broken\non:\n pull_request:\n") + self.assertIn("malformed_yaml", self.codes(result)) + + def test_exception_requires_reason_and_revisit(self) -> None: + broken = policy() + broken["exceptions"] = [ + {"rule": "mutable_runner", "path": WORKFLOW, "reason": "short"} + ] + result = self.run_repo(valid_workflow(), audit_policy=broken) + self.assertIn("invalid_policy", self.codes(result)) + + +if __name__ == "__main__": + unittest.main() From ad8d0fc45a3e3d1dcde8b475b2c5f2fc3e61e58d Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:53:08 +0300 Subject: [PATCH 05/37] ci(af01): harden mandatory workflow trust --- .github/workflows/ci.yml | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7008cba1..1409a08f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,12 +15,19 @@ permissions: jobs: rust: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 + timeout-minutes: 30 steps: - - uses: actions/checkout@v4 - - uses: dtolnay/rust-toolchain@1.97.1 + - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5 / Node 24 + with: + persist-credentials: false + - uses: dtolnay/rust-toolchain@032958afbdc797a9164d3bc0b56325c1308924a5 # 1.97.1 with: components: rustfmt, clippy + - name: AF-01 workflow trust audit tests + run: python3 .github/scripts/test_audit_workflow_trust.py + - name: AF-01 repository workflow trust audit + run: python3 .github/scripts/audit_workflow_trust.py - name: Format run: cargo fmt --all -- --check - name: Clippy @@ -143,4 +150,4 @@ jobs: assert report["decision"]["passed"] is True assert report["decision"]["blocking_findings"] == 0 assert report["compatibility"]["findings"] == [] - PY + PY \ No newline at end of file From 916361113565fd0fc353d170aa770bdaf5786aec Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:53:38 +0300 Subject: [PATCH 06/37] ci(af01): pin registry smoke runner --- .github/workflows/registry-download-smoke.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/registry-download-smoke.yml b/.github/workflows/registry-download-smoke.yml index 6a951d91..afc6216d 100644 --- a/.github/workflows/registry-download-smoke.yml +++ b/.github/workflows/registry-download-smoke.yml @@ -20,7 +20,7 @@ permissions: jobs: registry-download: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 timeout-minutes: 15 steps: - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5 / node24 @@ -112,4 +112,4 @@ jobs: fi echo "VSAC fallback resolve/verify failed on attempt ${attempt}; retrying" >&2 sleep $((attempt * 5)) - done + done \ No newline at end of file From 6ba53646d459b555981ba33c9147396b77e15f92 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:54:33 +0300 Subject: [PATCH 07/37] ci(af01): harden oracle workflow authority --- .github/workflows/cf06-oracle.yml | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/.github/workflows/cf06-oracle.yml b/.github/workflows/cf06-oracle.yml index cd5c0b89..664e3039 100644 --- a/.github/workflows/cf06-oracle.yml +++ b/.github/workflows/cf06-oracle.yml @@ -12,7 +12,8 @@ permissions: jobs: oracle-self-smoke: name: oracle-self-smoke - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 + timeout-minutes: 20 steps: - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5 / Node 24 with: @@ -94,7 +95,8 @@ jobs: oracle-changed-profile: name: oracle-changed-profile - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 + timeout-minutes: 25 steps: - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5 / Node 24 with: @@ -277,7 +279,9 @@ jobs: needs: - oracle-self-smoke - oracle-changed-profile - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 + timeout-minutes: 5 + permissions: {} steps: - name: Enforce all CF-06 oracle validation suites env: @@ -286,4 +290,4 @@ jobs: run: | set -euo pipefail test "$SELF_SMOKE_RESULT" = success - test "$CHANGED_PROFILE_RESULT" = success + test "$CHANGED_PROFILE_RESULT" = success \ No newline at end of file From e844f5b306cbf201cc31fa893539d2ed0875486c Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:55:42 +0300 Subject: [PATCH 08/37] test(af01): register audit module before execution --- .github/scripts/test_audit_workflow_trust.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py index 48e6e455..74f6570a 100644 --- a/.github/scripts/test_audit_workflow_trust.py +++ b/.github/scripts/test_audit_workflow_trust.py @@ -3,6 +3,7 @@ import copy import importlib.util +import sys import tempfile import unittest from pathlib import Path @@ -11,6 +12,7 @@ SPEC = importlib.util.spec_from_file_location("audit_workflow_trust", MODULE_PATH) assert SPEC is not None and SPEC.loader is not None AUDIT = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = AUDIT SPEC.loader.exec_module(AUDIT) WORKFLOW = ".github/workflows/example.yml" From 0be4766d3f2631f1dba24f03706da8ffe81f3279 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 18:59:01 +0300 Subject: [PATCH 09/37] fix(af01): close action metadata checkout authority gap --- .github/scripts/audit_workflow_trust.py | 252 +++++++++++++++++++----- 1 file changed, 200 insertions(+), 52 deletions(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index a30c0c53..ff428250 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -109,7 +109,14 @@ def _parse_permissions( def _job_ranges(lines: list[str]) -> tuple[dict[str, tuple[int, int]], str | None]: - jobs_index = next((i for i, line in enumerate(lines) if line.strip() == "jobs:" and _indent(line) == 0), None) + jobs_index = next( + ( + index + for index, line in enumerate(lines) + if line.strip() == "jobs:" and _indent(line) == 0 + ), + None, + ) if jobs_index is None: return {}, "workflow has no top-level jobs mapping" @@ -148,22 +155,48 @@ def _job_scalar(lines: list[str], start: int, end: int, key: str) -> str | None: def _container_images(lines: list[str], start: int, end: int) -> list[str]: + """Return job-container and service-container image scalars only.""" images: list[str] = [] + service_indent: int | None = None + in_job_container = False + for index in range(start + 1, end): line = lines[index] stripped = line.strip() - if _indent(line) == 4 and stripped.startswith("container:"): + indent = _indent(line) + if not stripped or stripped.startswith("#"): + continue + + if indent == 4: + in_job_container = False + service_indent = None + if stripped.startswith("container:"): + value = _scalar(stripped.split(":", 1)[1]) + if value: + images.append(value) + else: + in_job_container = True + elif stripped == "services:": + service_indent = 4 + continue + + if in_job_container and indent == 6 and stripped.startswith("image:"): value = _scalar(stripped.split(":", 1)[1]) if value: images.append(value) - if stripped.startswith("image:") and _indent(line) >= 6: + continue + + if service_indent is not None and indent == 8 and stripped.startswith("image:"): value = _scalar(stripped.split(":", 1)[1]) if value: images.append(value) + return images -def _all_uses(lines: list[str], start: int = 0, end: int | None = None) -> list[tuple[int, str]]: +def _all_uses( + lines: list[str], start: int = 0, end: int | None = None +) -> list[tuple[int, str]]: if end is None: end = len(lines) result: list[tuple[int, str]] = [] @@ -208,7 +241,9 @@ def _valid_exception(exception: object) -> bool: required = {"rule", "path", "reason", "revisit"} if not required.issubset(exception): return False - if not all(isinstance(exception[key], str) and exception[key].strip() for key in required): + if not all( + isinstance(exception[key], str) and exception[key].strip() for key in required + ): return False if len(exception["reason"].strip()) < 10 or len(exception["revisit"].strip()) < 5: return False @@ -216,7 +251,12 @@ def _valid_exception(exception: object) -> bool: def _excepted(policy: dict, finding: Finding) -> bool: - for exception in policy.get("exceptions", []): + exceptions = policy.get("exceptions", []) + if not isinstance(exceptions, list): + return False + for exception in exceptions: + if not isinstance(exception, dict): + continue if exception.get("rule") != finding.code or exception.get("path") != finding.path: continue if exception.get("job", finding.job) != finding.job: @@ -227,25 +267,59 @@ def _excepted(policy: dict, finding: Finding) -> bool: return False +def _uses_findings(path: str, lines: list[str], policy: dict) -> list[Finding]: + findings: list[Finding] = [] + rules = policy.get("rules", {}) + if not isinstance(rules, dict): + return findings + for index, reference in _all_uses(lines): + if rules.get("require_external_uses_full_sha", False) and not _external_ref_is_immutable( + reference + ): + findings.append( + Finding("mutable_uses", path, "", f"uses reference is not immutable: {reference}") + ) + if reference.startswith("actions/checkout@") and rules.get( + "require_checkout_credentials_disabled", False + ): + if not _checkout_has_credentials_disabled(lines, index): + findings.append( + Finding( + "checkout_credentials", + path, + "", + "checkout step does not set persist-credentials: false", + ) + ) + return findings + + def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[Finding]: findings: list[Finding] = [] if "\t" in text: - findings.append(Finding("malformed_yaml", path, "", "tab indentation is not supported")) - return findings + return [Finding("malformed_yaml", path, "", "tab indentation is not supported")] lines = text.splitlines() jobs, jobs_error = _job_ranges(lines) if jobs_error: - findings.append(Finding("malformed_yaml", path, "", jobs_error)) - return findings + return [Finding("malformed_yaml", path, "", jobs_error)] - expected_jobs = expected.get("jobs") + expected_jobs = expected.get("jobs") if isinstance(expected, dict) else None if not isinstance(expected_jobs, dict): - findings.append(Finding("invalid_policy", path, "", "workflow policy must contain a jobs object")) - return findings + return [ + Finding( + "invalid_policy", + path, + "", + "workflow policy must contain a jobs object", + ) + ] + actual_names = set(jobs) expected_names = set(expected_jobs) for name in sorted(actual_names - expected_names): - findings.append(Finding("unplanned_job", path, name, "job is not declared in workflow trust policy")) + findings.append( + Finding("unplanned_job", path, name, "job is not declared in workflow trust policy") + ) for name in sorted(expected_names - actual_names): findings.append(Finding("missing_job", path, name, "policy job is missing from workflow")) @@ -253,16 +327,33 @@ def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[F if top_permission_error: findings.append(Finding("permissions_syntax", path, "", top_permission_error)) + rules = policy.get("rules", {}) + if not isinstance(rules, dict): + rules = {} + for job in sorted(actual_names & expected_names): start, end = jobs[job] expected_job = expected_jobs[job] + if not isinstance(expected_job, dict): + findings.append( + Finding("invalid_policy", path, job, "job policy must be an object") + ) + continue + runner = _job_scalar(lines, start, end, "runs-on") if runner != expected_job.get("runner"): findings.append( - Finding("runner_mismatch", path, job, f"expected runner {expected_job.get('runner')!r}, found {runner!r}") + Finding( + "runner_mismatch", + path, + job, + f"expected runner {expected_job.get('runner')!r}, found {runner!r}", + ) ) if runner and runner.endswith("-latest"): - findings.append(Finding("mutable_runner", path, job, f"runner {runner!r} is a mutable latest label")) + findings.append( + Finding("mutable_runner", path, job, f"runner {runner!r} is a mutable latest label") + ) timeout = _job_scalar(lines, start, end, "timeout-minutes") try: @@ -271,10 +362,17 @@ def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[F timeout_value = None timeout_limit = expected_job.get("timeout_minutes") if not isinstance(timeout_limit, int) or timeout_limit <= 0: - findings.append(Finding("invalid_policy", path, job, "timeout_minutes must be a positive integer")) + findings.append( + Finding("invalid_policy", path, job, "timeout_minutes must be a positive integer") + ) elif timeout_value is None or timeout_value <= 0 or timeout_value > timeout_limit: findings.append( - Finding("timeout_policy", path, job, f"timeout must be 1..{timeout_limit} minutes, found {timeout!r}") + Finding( + "timeout_policy", + path, + job, + f"timeout must be 1..{timeout_limit} minutes, found {timeout!r}", + ) ) job_permissions, job_permission_error = _parse_permissions(lines, start + 1, end, 4) @@ -283,7 +381,12 @@ def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[F effective_permissions = job_permissions if job_permissions is not None else top_permissions if effective_permissions is None: findings.append( - Finding("unresolved_permissions", path, job, "job inherits undocumented GitHub default token permissions") + Finding( + "unresolved_permissions", + path, + job, + "job inherits undocumented GitHub default token permissions", + ) ) elif effective_permissions != expected_job.get("permissions"): findings.append( @@ -295,76 +398,119 @@ def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[F ) ) - if policy["rules"].get("require_container_digest", False): + if rules.get("require_container_digest", False): for image in _container_images(lines, start, end): if not DIGEST_IMAGE_RE.search(image): findings.append( - Finding("mutable_container", path, job, f"container image is not sha256 digest-bound: {image}") + Finding( + "mutable_container", + path, + job, + f"container image is not sha256 digest-bound: {image}", + ) ) - locked_subcommands = set(policy["rules"].get("cargo_locked_subcommands", [])) + locked_subcommands = set(rules.get("cargo_locked_subcommands", [])) for line in lines[start + 1 : end]: command = line.strip() for match in CARGO_RE.finditer(command): subcommand = match.group(1) if subcommand in locked_subcommands and "--locked" not in command[match.start() :]: findings.append( - Finding("cargo_unlocked", path, job, f"cargo {subcommand} invocation omits --locked: {command}") + Finding( + "cargo_unlocked", + path, + job, + f"cargo {subcommand} invocation omits --locked: {command}", + ) ) - for index, reference in _all_uses(lines): - if policy["rules"].get("require_external_uses_full_sha", False) and not _external_ref_is_immutable(reference): - findings.append(Finding("mutable_uses", path, "", f"uses reference is not immutable: {reference}")) - if reference.startswith("actions/checkout@") and policy["rules"].get( - "require_checkout_credentials_disabled", False - ): - if not _checkout_has_credentials_disabled(lines, index): - findings.append( - Finding("checkout_credentials", path, "", "checkout step does not set persist-credentials: false") - ) - + findings.extend(_uses_findings(path, lines, policy)) return [finding for finding in findings if not _excepted(policy, finding)] def audit_action_metadata(path: str, text: str, policy: dict) -> list[Finding]: - findings: list[Finding] = [] if "\t" in text: return [Finding("malformed_yaml", path, "", "tab indentation is not supported")] lines = text.splitlines() + findings: list[Finding] = [] if not any(line.strip() == "runs:" for line in lines): - findings.append(Finding("malformed_action_metadata", path, "", "Action metadata has no runs mapping")) - for _, reference in _all_uses(lines): - if policy["rules"].get("require_external_uses_full_sha", False) and not _external_ref_is_immutable(reference): - findings.append(Finding("mutable_uses", path, "", f"uses reference is not immutable: {reference}")) + findings.append( + Finding( + "malformed_action_metadata", + path, + "", + "Action metadata has no runs mapping", + ) + ) + findings.extend(_uses_findings(path, lines, policy)) return [finding for finding in findings if not _excepted(policy, finding)] -def audit_repository(root: Path, policy: dict, tracked_files: Iterable[str] | None = None) -> dict: +def _policy_errors(policy: object) -> list[Finding]: findings: list[Finding] = [] - if policy.get("schema") != 1 or not isinstance(policy.get("rules"), dict): - findings.append(Finding("invalid_policy", ".github/workflow-trust-policy.json", "", "unsupported policy schema")) + policy_path = ".github/workflow-trust-policy.json" + if not isinstance(policy, dict): + return [Finding("invalid_policy", policy_path, "", "policy root must be an object")] + if policy.get("schema") != 1: + findings.append(Finding("invalid_policy", policy_path, "", "unsupported policy schema")) + if not isinstance(policy.get("rules"), dict): + findings.append(Finding("invalid_policy", policy_path, "", "rules must be an object")) + if not isinstance(policy.get("workflows"), dict): + findings.append(Finding("invalid_policy", policy_path, "", "workflows must be an object")) exceptions = policy.get("exceptions", []) if not isinstance(exceptions, list) or any(not _valid_exception(item) for item in exceptions): findings.append( - Finding("invalid_policy", ".github/workflow-trust-policy.json", "", "every exception requires bounded rule/path/reason/revisit fields") + Finding( + "invalid_policy", + policy_path, + "", + "every exception requires bounded rule/path/reason/revisit fields", + ) ) + return findings + + +def audit_repository( + root: Path, policy: dict, tracked_files: Iterable[str] | None = None +) -> dict: + findings = _policy_errors(policy) + if findings: + filtered = sorted(findings) + return { + "schema": 1, + "ok": False, + "workflows": [], + "action_metadata": [], + "findings": [finding.as_dict() for finding in filtered], + } paths = list(tracked_files) if tracked_files is not None else _tracked_files(root) workflows, actions = discover_security_files(paths) - expected_workflows = policy.get("workflows", {}) - if not isinstance(expected_workflows, dict): - expected_workflows = {} - findings.append(Finding("invalid_policy", ".github/workflow-trust-policy.json", "", "workflows must be an object")) + expected_workflows = policy["workflows"] for path in sorted(set(workflows) - set(expected_workflows)): - findings.append(Finding("unplanned_workflow", path, "", "tracked workflow is absent from policy")) + findings.append( + Finding("unplanned_workflow", path, "", "tracked workflow is absent from policy") + ) for path in sorted(set(expected_workflows) - set(workflows)): - findings.append(Finding("missing_workflow", path, "", "policy workflow is not tracked")) + findings.append( + Finding("missing_workflow", path, "", "policy workflow is not tracked") + ) for path in sorted(set(workflows) & set(expected_workflows)): - findings.extend(audit_workflow(path, (root / path).read_text(encoding="utf-8"), expected_workflows[path], policy)) + findings.extend( + audit_workflow( + path, + (root / path).read_text(encoding="utf-8"), + expected_workflows[path], + policy, + ) + ) for path in actions: - findings.extend(audit_action_metadata(path, (root / path).read_text(encoding="utf-8"), policy)) + findings.extend( + audit_action_metadata(path, (root / path).read_text(encoding="utf-8"), policy) + ) filtered = sorted(finding for finding in findings if not _excepted(policy, finding)) return { @@ -379,7 +525,9 @@ def audit_repository(root: Path, policy: dict, tracked_files: Iterable[str] | No def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--root", type=Path, default=Path(".")) - parser.add_argument("--policy", type=Path, default=Path(".github/workflow-trust-policy.json")) + parser.add_argument( + "--policy", type=Path, default=Path(".github/workflow-trust-policy.json") + ) args = parser.parse_args() root = args.root.resolve() From 95a2b4829c7b1a832b1460a4a97aada107e23877 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:00:24 +0300 Subject: [PATCH 10/37] test(af01): prove metadata checkout and policy fail closed --- .github/scripts/test_audit_workflow_trust.py | 47 +++++++++++++++++--- 1 file changed, 41 insertions(+), 6 deletions(-) diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py index 74f6570a..b6ddbf0b 100644 --- a/.github/scripts/test_audit_workflow_trust.py +++ b/.github/scripts/test_audit_workflow_trust.py @@ -158,9 +158,7 @@ def test_mutable_external_action_tag_is_rejected(self) -> None: def test_short_sha_and_branch_refs_are_rejected(self) -> None: for reference in ("owner/action@abc1234", "owner/action@main"): with self.subTest(reference=reference): - workflow = valid_workflow( - f" - uses: {reference}\n" - ) + workflow = valid_workflow(f" - uses: {reference}\n") result = self.run_repo(workflow) self.assertIn("mutable_uses", self.codes(result)) @@ -189,10 +187,32 @@ def test_checkout_credentials_must_be_disabled(self) -> None: result = self.run_repo(workflow) self.assertIn("checkout_credentials", self.codes(result)) + def test_action_metadata_checkout_credentials_must_be_disabled(self) -> None: + action = f"""name: nested +description: fixture +runs: + using: composite + steps: + - uses: actions/checkout@{CHECKOUT_SHA} +""" + result = self.run_repo(valid_workflow(), action) + self.assertIn("checkout_credentials", self.codes(result)) + + def test_action_metadata_checkout_with_credentials_disabled_passes(self) -> None: + action = f"""name: nested +description: fixture +runs: + using: composite + steps: + - uses: actions/checkout@{CHECKOUT_SHA} + with: + persist-credentials: false +""" + result = self.run_repo(valid_workflow(), action) + self.assertTrue(result["ok"], result) + def test_unresolved_default_permissions_fail_closed(self) -> None: - workflow = valid_workflow().replace( - "permissions:\n contents: read\n", "" - ) + workflow = valid_workflow().replace("permissions:\n contents: read\n", "") result = self.run_repo(workflow) self.assertIn("unresolved_permissions", self.codes(result)) @@ -234,6 +254,14 @@ def test_mutable_job_and_service_container_images_are_rejected(self) -> None: result = self.run_repo(mutable_service) self.assertIn("mutable_container", self.codes(result)) + def test_non_container_image_key_does_not_false_positive(self) -> None: + workflow = valid_workflow().replace( + " steps:\n", + " env:\n image: mutable-but-not-a-container-authority\n steps:\n", + ) + result = self.run_repo(workflow) + self.assertTrue(result["ok"], result) + def test_unlocked_cargo_command_is_rejected(self) -> None: workflow = valid_workflow().replace( "cargo test --locked --workspace", "cargo test --workspace" @@ -258,6 +286,13 @@ def test_exception_requires_reason_and_revisit(self) -> None: result = self.run_repo(valid_workflow(), audit_policy=broken) self.assertIn("invalid_policy", self.codes(result)) + def test_invalid_policy_root_fails_closed_without_exception(self) -> None: + result = self.run_repo(valid_workflow(), audit_policy={"schema": 1}) + self.assertFalse(result["ok"]) + self.assertIn("invalid_policy", self.codes(result)) + self.assertEqual(result["workflows"], []) + self.assertEqual(result["action_metadata"], []) + if __name__ == "__main__": unittest.main() From d76afa1c16dbed39ff9bfacbf7188a39170ea935 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:04:45 +0300 Subject: [PATCH 11/37] fix(af01): audit logical cargo invocations --- .github/scripts/audit_workflow_trust.py | 94 +++++++++++++++++++++---- 1 file changed, 81 insertions(+), 13 deletions(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index ff428250..91ca5d11 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -19,6 +19,7 @@ STEP_USES_RE = re.compile(r"^(\s*)-\s+uses:\s*(.+?)\s*$") PERMISSION_RE = re.compile(r"^([A-Za-z0-9_-]+):\s*(read|write|none)\s*$") CARGO_RE = re.compile(r"\bcargo\s+(bench|build|check|clippy|doc|metadata|run|test)\b") +SHELL_SEPARATOR_RE = re.compile(r"(?:\r?\n|&&|\|\||;|(? boo return False +def _run_scripts(lines: list[str], start: int, end: int) -> list[str]: + """Extract inline and block step `run:` scripts from a statically structured job.""" + scripts: list[str] = [] + index = start + 1 + while index < end: + line = lines[index] + stripped = line.strip() + indent = _indent(line) + if indent < 6 or not stripped.startswith("run:"): + index += 1 + continue + + value = _scalar(stripped.split(":", 1)[1]) + if value in {"|", "|-", "|+", ">", ">-", ">+"}: + block: list[str] = [] + cursor = index + 1 + while cursor < end: + child = lines[cursor] + if child.strip() and _indent(child) <= indent: + break + if not child.strip(): + block.append("") + else: + child_indent = _indent(child) + if child_indent < indent + 2: + break + block.append(child[indent + 2 :]) + cursor += 1 + scripts.append("\n".join(block)) + index = cursor + continue + if value: + scripts.append(value) + index += 1 + return scripts + + +def _logical_shell_segments(script: str) -> list[str]: + """Join backslash continuations, then split at shell command boundaries.""" + joined = re.sub(r"\\[ \t]*\r?\n[ \t]*", " ", script) + return [segment.strip() for segment in SHELL_SEPARATOR_RE.split(joined) if segment.strip()] + + +def _cargo_findings( + path: str, + job: str, + lines: list[str], + start: int, + end: int, + locked_subcommands: set[str], +) -> list[Finding]: + findings: list[Finding] = [] + for script in _run_scripts(lines, start, end): + for segment in _logical_shell_segments(script): + matches = list(CARGO_RE.finditer(segment)) + for position, match in enumerate(matches): + subcommand = match.group(1) + if subcommand not in locked_subcommands: + continue + invocation_end = ( + matches[position + 1].start() + if position + 1 < len(matches) + else len(segment) + ) + invocation = segment[match.start() : invocation_end].strip() + if "--locked" not in invocation: + findings.append( + Finding( + "cargo_unlocked", + path, + job, + f"cargo {subcommand} invocation omits --locked: {invocation}", + ) + ) + return findings + + def _valid_exception(exception: object) -> bool: if not isinstance(exception, dict): return False @@ -411,19 +489,9 @@ def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[F ) locked_subcommands = set(rules.get("cargo_locked_subcommands", [])) - for line in lines[start + 1 : end]: - command = line.strip() - for match in CARGO_RE.finditer(command): - subcommand = match.group(1) - if subcommand in locked_subcommands and "--locked" not in command[match.start() :]: - findings.append( - Finding( - "cargo_unlocked", - path, - job, - f"cargo {subcommand} invocation omits --locked: {command}", - ) - ) + findings.extend( + _cargo_findings(path, job, lines, start, end, locked_subcommands) + ) findings.extend(_uses_findings(path, lines, policy)) return [finding for finding in findings if not _excepted(policy, finding)] From 0d639dc3fbc013ce1b9a9ea2e203e5a69a5e7e6b Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:05:29 +0300 Subject: [PATCH 12/37] test(af01): cover logical cargo command parsing --- .github/scripts/test_audit_workflow_trust.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py index b6ddbf0b..c954784f 100644 --- a/.github/scripts/test_audit_workflow_trust.py +++ b/.github/scripts/test_audit_workflow_trust.py @@ -269,6 +269,26 @@ def test_unlocked_cargo_command_is_rejected(self) -> None: result = self.run_repo(workflow) self.assertIn("cargo_unlocked", self.codes(result)) + def test_multiline_locked_cargo_command_is_accepted(self) -> None: + workflow = valid_workflow().replace( + " run: cargo test --locked --workspace", + " run: |\n cargo test \\\n --locked --workspace", + ) + result = self.run_repo(workflow) + self.assertTrue(result["ok"], result) + + def test_later_locked_cargo_command_cannot_mask_unlocked_command(self) -> None: + workflow = valid_workflow().replace( + "cargo test --locked --workspace", + "cargo test --workspace && cargo test --locked -p commandf-pkg", + ) + result = self.run_repo(workflow) + cargo_findings = [ + finding for finding in result["findings"] if finding["code"] == "cargo_unlocked" + ] + self.assertEqual(len(cargo_findings), 1, result) + self.assertIn("cargo test --workspace", cargo_findings[0]["detail"]) + def test_malformed_workflow_fails_closed(self) -> None: result = self.run_repo("name: broken\n\tjobs:\n") self.assertIn("malformed_yaml", self.codes(result)) From 8550a648aa4aeffa9506efc50e5edee423a58c92 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:06:35 +0300 Subject: [PATCH 13/37] test(af01): construct literal cargo continuation --- .github/scripts/test_audit_workflow_trust.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py index c954784f..365f54d1 100644 --- a/.github/scripts/test_audit_workflow_trust.py +++ b/.github/scripts/test_audit_workflow_trust.py @@ -270,9 +270,14 @@ def test_unlocked_cargo_command_is_rejected(self) -> None: self.assertIn("cargo_unlocked", self.codes(result)) def test_multiline_locked_cargo_command_is_accepted(self) -> None: + continuation = chr(92) + replacement = ( + " run: |\n" + f" cargo test {continuation}\n" + " --locked --workspace" + ) workflow = valid_workflow().replace( - " run: cargo test --locked --workspace", - " run: |\n cargo test \\\n --locked --workspace", + " run: cargo test --locked --workspace", replacement ) result = self.run_repo(workflow) self.assertTrue(result["ok"], result) From bb7183ec2aa7de7ea571a65f1e5ac1215a730ffb Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:10:35 +0300 Subject: [PATCH 14/37] fix(af01): close workflow policy parser bypasses --- .github/scripts/audit_workflow_trust.py | 376 ++++++++++++++++++++++-- 1 file changed, 349 insertions(+), 27 deletions(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index 91ca5d11..571fc770 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -6,6 +6,7 @@ import argparse import json import re +import shlex import subprocess import sys from dataclasses import dataclass @@ -16,11 +17,30 @@ DIGEST_IMAGE_RE = re.compile(r"@sha256:[0-9a-fA-F]{64}$") JOB_RE = re.compile(r"^ ([A-Za-z0-9_.-]+):\s*(?:#.*)?$") USES_RE = re.compile(r"^(\s*)(?:-\s*)?uses:\s*(.+?)\s*$") -STEP_USES_RE = re.compile(r"^(\s*)-\s+uses:\s*(.+?)\s*$") +STEP_LIST_RE = re.compile(r"^(\s*)-\s+\S") PERMISSION_RE = re.compile(r"^([A-Za-z0-9_-]+):\s*(read|write|none)\s*$") -CARGO_RE = re.compile(r"\bcargo\s+(bench|build|check|clippy|doc|metadata|run|test)\b") +FLOW_USES_RE = re.compile(r"[\[{,]\s*[\"']?uses[\"']?\s*:") +BLOCK_SCALAR_RE = re.compile(r":\s*[|>][+-]?\s*(?:#.*)?$") SHELL_SEPARATOR_RE = re.compile(r"(?:\r?\n|&&|\|\||;|(? set[int]: + indexes: set[int] = set() + for index, line in enumerate(lines): + if not BLOCK_SCALAR_RE.search(line): + continue + indent = _indent(line) + cursor = index + 1 + while cursor < len(lines): + child = lines[cursor] + if child.strip() and _indent(child) <= indent: + break + indexes.add(cursor) + cursor += 1 + return indexes + + +def _unsupported_flow_uses(lines: list[str]) -> list[str]: + """Reject flow-style uses keys instead of silently under-parsing them.""" + block_lines = _block_scalar_line_indexes(lines) + unsupported: list[str] = [] + for index, line in enumerate(lines): + if index in block_lines: + continue + stripped = line.lstrip() + if not stripped or stripped.startswith("#"): + continue + if stripped.startswith("run:") or re.match(r"^-\s+run:\s*", stripped): + continue + if FLOW_USES_RE.search(line): + unsupported.append(line.strip()) + return unsupported + + def _external_ref_is_immutable(reference: str) -> bool: if reference.startswith("./"): return True @@ -219,21 +272,70 @@ def _external_ref_is_immutable(reference: str) -> bool: return bool(FULL_SHA_RE.fullmatch(revision)) -def _checkout_has_credentials_disabled(lines: list[str], uses_index: int) -> bool: - matched = STEP_USES_RE.match(lines[uses_index]) - if not matched: - return False - step_indent = len(matched.group(1)) +def _step_bounds(lines: list[str], uses_index: int) -> tuple[int, int, int] | None: + uses_line = lines[uses_index] + uses_indent = _indent(uses_line) + if STEP_LIST_RE.match(uses_line) and uses_line.lstrip().startswith("- uses:"): + start = uses_index + step_indent = uses_indent + else: + start = -1 + step_indent = -1 + for index in range(uses_index - 1, -1, -1): + line = lines[index] + if not line.strip(): + continue + indent = _indent(line) + if indent >= uses_indent: + continue + if STEP_LIST_RE.match(line): + start = index + step_indent = indent + break + if indent < uses_indent and line.strip().endswith(":"): + break + if start < 0: + return None + end = len(lines) - for index in range(uses_index + 1, len(lines)): + for index in range(start + 1, len(lines)): line = lines[index] - if re.match(rf"^ {{{step_indent}}}-\s+", line): + if not line.strip(): + continue + if _indent(line) == step_indent and STEP_LIST_RE.match(line): end = index break - for line in lines[uses_index + 1 : end]: - if re.match(r"^\s*persist-credentials:\s*false\s*(?:#.*)?$", line): - return True - return False + if _indent(line) < step_indent: + end = index + break + return start, end, step_indent + + +def _checkout_has_credentials_disabled(lines: list[str], uses_index: int) -> bool: + bounds = _step_bounds(lines, uses_index) + if bounds is None: + return False + start, end, step_indent = bounds + with_indexes = [ + index + for index in range(start + 1, end) + if _indent(lines[index]) == step_indent + 2 and lines[index].strip() == "with:" + ] + if len(with_indexes) != 1: + return False + + with_index = with_indexes[0] + entries: list[str] = [] + for index in range(with_index + 1, end): + line = lines[index] + if not line.strip() or line.lstrip().startswith("#"): + continue + indent = _indent(line) + if indent <= step_indent + 2: + break + if indent == step_indent + 4 and line.strip().startswith("persist-credentials:"): + entries.append(_scalar(line.strip().split(":", 1)[1])) + return entries == ["false"] def _run_scripts(lines: list[str], start: int, end: int) -> list[str]: @@ -290,24 +392,66 @@ def _cargo_findings( findings: list[Finding] = [] for script in _run_scripts(lines, start, end): for segment in _logical_shell_segments(script): - matches = list(CARGO_RE.finditer(segment)) - for position, match in enumerate(matches): - subcommand = match.group(1) + if "cargo" not in segment: + continue + try: + tokens = shlex.split(segment, comments=True, posix=True) + except ValueError as error: + findings.append( + Finding( + "unsupported_shell_syntax", + path, + job, + f"cannot safely parse Cargo-containing shell segment: {error}: {segment}", + ) + ) + continue + + for index, token in enumerate(tokens): + if token != "cargo": + continue + command_index = index + 1 + if command_index < len(tokens) and tokens[command_index].startswith("+"): + command_index += 1 + if command_index >= len(tokens): + findings.append( + Finding( + "unsupported_cargo_syntax", + path, + job, + f"cannot identify Cargo subcommand: {segment}", + ) + ) + continue + subcommand = tokens[command_index] + if subcommand.startswith("-"): + findings.append( + Finding( + "unsupported_cargo_syntax", + path, + job, + f"Cargo global-option syntax requires explicit auditor support: {segment}", + ) + ) + continue if subcommand not in locked_subcommands: continue - invocation_end = ( - matches[position + 1].start() - if position + 1 < len(matches) - else len(segment) + next_cargo = next( + ( + position + for position in range(command_index + 1, len(tokens)) + if tokens[position] == "cargo" + ), + len(tokens), ) - invocation = segment[match.start() : invocation_end].strip() + invocation = tokens[index:next_cargo] if "--locked" not in invocation: findings.append( Finding( "cargo_unlocked", path, job, - f"cargo {subcommand} invocation omits --locked: {invocation}", + f"cargo {subcommand} invocation omits --locked: {' '.join(invocation)}", ) ) return findings @@ -323,6 +467,11 @@ def _valid_exception(exception: object) -> bool: isinstance(exception[key], str) and exception[key].strip() for key in required ): return False + for optional in ("job", "detail"): + if optional in exception and ( + not isinstance(exception[optional], str) or not exception[optional].strip() + ): + return False if len(exception["reason"].strip()) < 10 or len(exception["revisit"].strip()) < 5: return False return set(exception).issubset(required | {"job", "detail"}) @@ -350,6 +499,17 @@ def _uses_findings(path: str, lines: list[str], policy: dict) -> list[Finding]: rules = policy.get("rules", {}) if not isinstance(rules, dict): return findings + + for syntax in _unsupported_flow_uses(lines): + findings.append( + Finding( + "unsupported_uses_syntax", + path, + "", + f"flow-style uses syntax is not supported by the trust parser: {syntax}", + ) + ) + for index, reference in _all_uses(lines): if rules.get("require_external_uses_full_sha", False) and not _external_ref_is_immutable( reference @@ -366,7 +526,7 @@ def _uses_findings(path: str, lines: list[str], policy: dict) -> list[Finding]: "checkout_credentials", path, "", - "checkout step does not set persist-credentials: false", + "checkout step does not set with.persist-credentials: false exactly once", ) ) return findings @@ -439,7 +599,7 @@ def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[F except ValueError: timeout_value = None timeout_limit = expected_job.get("timeout_minutes") - if not isinstance(timeout_limit, int) or timeout_limit <= 0: + if not isinstance(timeout_limit, int) or isinstance(timeout_limit, bool) or timeout_limit <= 0: findings.append( Finding("invalid_policy", path, job, "timeout_minutes must be a positive integer") ) @@ -520,12 +680,174 @@ def _policy_errors(policy: object) -> list[Finding]: policy_path = ".github/workflow-trust-policy.json" if not isinstance(policy, dict): return [Finding("invalid_policy", policy_path, "", "policy root must be an object")] + + unknown_top = set(policy) - SUPPORTED_TOP_LEVEL_POLICY_KEYS + if unknown_top: + findings.append( + Finding( + "invalid_policy", + policy_path, + "", + f"unsupported top-level policy keys: {sorted(unknown_top)!r}", + ) + ) if policy.get("schema") != 1: findings.append(Finding("invalid_policy", policy_path, "", "unsupported policy schema")) - if not isinstance(policy.get("rules"), dict): + + rules = policy.get("rules") + if not isinstance(rules, dict): findings.append(Finding("invalid_policy", policy_path, "", "rules must be an object")) - if not isinstance(policy.get("workflows"), dict): - findings.append(Finding("invalid_policy", policy_path, "", "workflows must be an object")) + else: + if set(rules) != SUPPORTED_RULE_KEYS: + findings.append( + Finding( + "invalid_policy", + policy_path, + "", + f"rules must contain exactly {sorted(SUPPORTED_RULE_KEYS)!r}", + ) + ) + cargo_rules = rules.get("cargo_locked_subcommands") + if ( + not isinstance(cargo_rules, list) + or any(not isinstance(item, str) for item in cargo_rules) + or len(cargo_rules) != len(set(cargo_rules)) + or set(cargo_rules) != LOCKFILE_CARGO_SUBCOMMANDS + ): + findings.append( + Finding( + "invalid_policy", + policy_path, + "", + "cargo_locked_subcommands must list the complete supported lockfile-consuming command set exactly once", + ) + ) + for key in BOOLEAN_RULES: + if rules.get(key) is not True: + findings.append( + Finding( + "invalid_policy", + policy_path, + "", + f"security rule {key!r} must be boolean true; use a reviewed exception for a narrow waiver", + ) + ) + + rationales = policy.get("rationales") + if not isinstance(rationales, dict): + findings.append( + Finding("invalid_policy", policy_path, "", "rationales must be an object") + ) + elif set(rationales) != SUPPORTED_RULE_KEYS or any( + not isinstance(value, str) or len(value.strip()) < 20 for value in rationales.values() + ): + findings.append( + Finding( + "invalid_policy", + policy_path, + "", + "rationales must provide a substantive string for every supported rule", + ) + ) + + workflows = policy.get("workflows") + if not isinstance(workflows, dict) or not workflows: + findings.append( + Finding("invalid_policy", policy_path, "", "workflows must be a non-empty object") + ) + else: + for workflow_path, workflow_policy in workflows.items(): + if not isinstance(workflow_path, str) or not workflow_path: + findings.append( + Finding("invalid_policy", policy_path, "", "workflow paths must be non-empty strings") + ) + continue + if not isinstance(workflow_policy, dict) or set(workflow_policy) != {"jobs"}: + findings.append( + Finding( + "invalid_policy", + policy_path, + "", + f"workflow {workflow_path!r} policy must contain only a jobs object", + ) + ) + continue + jobs = workflow_policy.get("jobs") + if not isinstance(jobs, dict) or not jobs: + findings.append( + Finding( + "invalid_policy", + policy_path, + "", + f"workflow {workflow_path!r} jobs must be a non-empty object", + ) + ) + continue + for job_name, job_policy in jobs.items(): + if not isinstance(job_name, str) or not job_name: + findings.append( + Finding("invalid_policy", policy_path, "", "job names must be non-empty strings") + ) + continue + if not isinstance(job_policy, dict) or set(job_policy) != SUPPORTED_JOB_POLICY_KEYS: + findings.append( + Finding( + "invalid_policy", + policy_path, + job_name, + f"job policy must contain exactly {sorted(SUPPORTED_JOB_POLICY_KEYS)!r}", + ) + ) + continue + permissions = job_policy.get("permissions") + if not isinstance(permissions, dict) or any( + not isinstance(key, str) + or not key + or value not in {"read", "write", "none"} + for key, value in permissions.items() + ): + findings.append( + Finding( + "invalid_policy", + policy_path, + job_name, + "permissions must be a string-to-read/write/none object", + ) + ) + elif permissions not in ({}, {"contents": "read"}): + findings.append( + Finding( + "invalid_policy", + policy_path, + job_name, + "AF-01 Stack A permits only no token permissions or contents: read", + ) + ) + runner = job_policy.get("runner") + if runner not in SUPPORTED_RUNNERS: + findings.append( + Finding( + "invalid_policy", + policy_path, + job_name, + f"runner must be one of {sorted(SUPPORTED_RUNNERS)!r}", + ) + ) + timeout = job_policy.get("timeout_minutes") + if ( + type(timeout) is not int + or timeout <= 0 + or timeout > MAX_JOB_TIMEOUT_MINUTES + ): + findings.append( + Finding( + "invalid_policy", + policy_path, + job_name, + f"timeout_minutes must be 1..{MAX_JOB_TIMEOUT_MINUTES}", + ) + ) + exceptions = policy.get("exceptions", []) if not isinstance(exceptions, list) or any(not _valid_exception(item) for item in exceptions): findings.append( From 47fe525af13658ee6756be0cb13a4536e9779ece Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:10:58 +0300 Subject: [PATCH 15/37] docs(af01): bind trust rules to rationale --- .github/workflow-trust-policy.json | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/.github/workflow-trust-policy.json b/.github/workflow-trust-policy.json index 7fdcb295..be17b51f 100644 --- a/.github/workflow-trust-policy.json +++ b/.github/workflow-trust-policy.json @@ -15,6 +15,12 @@ "require_checkout_credentials_disabled": true, "require_external_uses_full_sha": true }, + "rationales": { + "cargo_locked_subcommands": "Lockfile-consuming Cargo commands must use the checked-in Cargo.lock rather than silently resolving a different dependency graph during assurance execution.", + "require_container_digest": "Job and service containers on proof-relevant workflows must resolve to an immutable image generation so a mutable tag cannot silently change the execution environment.", + "require_checkout_credentials_disabled": "Checkout must not persist the GitHub token into repository Git configuration because later steps do not require ambient repository write credentials.", + "require_external_uses_full_sha": "External Actions and reusable workflows must be bound to an immutable 40-hex commit so mutable tags or branches cannot change executed code without a repository diff." + }, "workflows": { ".github/workflows/cf06-oracle.yml": { "jobs": { @@ -107,4 +113,4 @@ } }, "exceptions": [] -} +} \ No newline at end of file From c6d84344717e507c3361d3c4d4205e0e0f17d7db Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:11:52 +0300 Subject: [PATCH 16/37] test(af01): regress all trust parser bypasses --- .github/scripts/test_audit_workflow_trust.py | 132 ++++++++++++++++--- 1 file changed, 115 insertions(+), 17 deletions(-) diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py index 365f54d1..08b9adb9 100644 --- a/.github/scripts/test_audit_workflow_trust.py +++ b/.github/scripts/test_audit_workflow_trust.py @@ -23,22 +23,27 @@ def policy() -> dict: + rules = { + "cargo_locked_subcommands": [ + "bench", + "build", + "check", + "clippy", + "doc", + "metadata", + "run", + "test", + ], + "require_container_digest": True, + "require_checkout_credentials_disabled": True, + "require_external_uses_full_sha": True, + } return { "schema": 1, - "rules": { - "cargo_locked_subcommands": [ - "bench", - "build", - "check", - "clippy", - "doc", - "metadata", - "run", - "test", - ], - "require_container_digest": True, - "require_checkout_credentials_disabled": True, - "require_external_uses_full_sha": True, + "rules": rules, + "rationales": { + key: f"Test rationale for {key} that is deliberately substantive." + for key in rules }, "workflows": { WORKFLOW: { @@ -180,6 +185,23 @@ def test_mutable_external_uses_in_nested_action_yaml_is_rejected(self) -> None: result = self.run_repo(valid_workflow(), action) self.assertIn("mutable_uses", self.codes(result)) + def test_flow_style_workflow_uses_fails_closed(self) -> None: + workflow = valid_workflow().replace( + " steps:\n - uses:", " steps: [{uses: owner/action@v1}]\n ignored:\n - uses:" + ) + result = self.run_repo(workflow) + self.assertIn("unsupported_uses_syntax", self.codes(result)) + + def test_flow_style_action_metadata_uses_fails_closed(self) -> None: + action = """name: nested +description: fixture +runs: + using: composite + steps: [{uses: owner/action@v1}] +""" + result = self.run_repo(valid_workflow(), action) + self.assertIn("unsupported_uses_syntax", self.codes(result)) + def test_checkout_credentials_must_be_disabled(self) -> None: workflow = valid_workflow().replace( " with:\n persist-credentials: false\n", "" @@ -187,6 +209,22 @@ def test_checkout_credentials_must_be_disabled(self) -> None: result = self.run_repo(workflow) self.assertIn("checkout_credentials", self.codes(result)) + def test_unscoped_persist_credentials_key_does_not_satisfy_checkout(self) -> None: + workflow = valid_workflow().replace( + " with:\n persist-credentials: false", + " env:\n persist-credentials: false", + ) + result = self.run_repo(workflow) + self.assertIn("checkout_credentials", self.codes(result)) + + def test_named_checkout_step_with_scoped_input_passes(self) -> None: + workflow = valid_workflow().replace( + f" - uses: actions/checkout@{CHECKOUT_SHA}", + f" - name: Checkout\n uses: actions/checkout@{CHECKOUT_SHA}", + ) + result = self.run_repo(workflow) + self.assertTrue(result["ok"], result) + def test_action_metadata_checkout_credentials_must_be_disabled(self) -> None: action = f"""name: nested description: fixture @@ -224,11 +262,17 @@ def test_overbroad_permission_is_rejected(self) -> None: result = self.run_repo(workflow) self.assertIn("permission_mismatch", self.codes(result)) + def test_policy_cannot_authorize_write_permission_in_stack_a(self) -> None: + broken = policy() + broken["workflows"][WORKFLOW]["jobs"]["build"]["permissions"] = { + "contents": "write" + } + result = self.run_repo(valid_workflow(), audit_policy=broken) + self.assertIn("invalid_policy", self.codes(result)) + def test_mutable_runner_is_rejected(self) -> None: workflow = valid_workflow().replace("ubuntu-24.04", "ubuntu-latest", 1) - expected = policy() - expected["workflows"][WORKFLOW]["jobs"]["build"]["runner"] = "ubuntu-latest" - result = self.run_repo(workflow, audit_policy=expected) + result = self.run_repo(workflow) self.assertIn("mutable_runner", self.codes(result)) def test_missing_or_excessive_timeout_is_rejected(self) -> None: @@ -269,6 +313,13 @@ def test_unlocked_cargo_command_is_rejected(self) -> None: result = self.run_repo(workflow) self.assertIn("cargo_unlocked", self.codes(result)) + def test_comment_cannot_fake_cargo_locked_flag(self) -> None: + workflow = valid_workflow().replace( + "cargo test --locked --workspace", "cargo test --workspace # --locked" + ) + result = self.run_repo(workflow) + self.assertIn("cargo_unlocked", self.codes(result)) + def test_multiline_locked_cargo_command_is_accepted(self) -> None: continuation = chr(92) replacement = ( @@ -282,6 +333,19 @@ def test_multiline_locked_cargo_command_is_accepted(self) -> None: result = self.run_repo(workflow) self.assertTrue(result["ok"], result) + def test_cargo_and_subcommand_line_continuation_is_accepted(self) -> None: + continuation = chr(92) + replacement = ( + " run: |\n" + f" cargo {continuation}\n" + " test --locked --workspace" + ) + workflow = valid_workflow().replace( + " run: cargo test --locked --workspace", replacement + ) + result = self.run_repo(workflow) + self.assertTrue(result["ok"], result) + def test_later_locked_cargo_command_cannot_mask_unlocked_command(self) -> None: workflow = valid_workflow().replace( "cargo test --locked --workspace", @@ -294,6 +358,14 @@ def test_later_locked_cargo_command_cannot_mask_unlocked_command(self) -> None: self.assertEqual(len(cargo_findings), 1, result) self.assertIn("cargo test --workspace", cargo_findings[0]["detail"]) + def test_cargo_global_option_syntax_fails_closed(self) -> None: + workflow = valid_workflow().replace( + "cargo test --locked --workspace", + "cargo --color always test --locked --workspace", + ) + result = self.run_repo(workflow) + self.assertIn("unsupported_cargo_syntax", self.codes(result)) + def test_malformed_workflow_fails_closed(self) -> None: result = self.run_repo("name: broken\n\tjobs:\n") self.assertIn("malformed_yaml", self.codes(result)) @@ -311,6 +383,32 @@ def test_exception_requires_reason_and_revisit(self) -> None: result = self.run_repo(valid_workflow(), audit_policy=broken) self.assertIn("invalid_policy", self.codes(result)) + def test_non_object_exception_fails_closed(self) -> None: + broken = policy() + broken["exceptions"] = [None] + result = self.run_repo(valid_workflow(), audit_policy=broken) + self.assertFalse(result["ok"]) + self.assertIn("invalid_policy", self.codes(result)) + + def test_malformed_rule_types_fail_closed(self) -> None: + for key, value in ( + ("cargo_locked_subcommands", "test"), + ("require_external_uses_full_sha", "true"), + ("require_container_digest", False), + ): + with self.subTest(key=key, value=value): + broken = policy() + broken["rules"][key] = value + result = self.run_repo(valid_workflow(), audit_policy=broken) + self.assertFalse(result["ok"]) + self.assertIn("invalid_policy", self.codes(result)) + + def test_missing_rule_rationale_fails_closed(self) -> None: + broken = policy() + del broken["rationales"]["require_container_digest"] + result = self.run_repo(valid_workflow(), audit_policy=broken) + self.assertIn("invalid_policy", self.codes(result)) + def test_invalid_policy_root_fails_closed_without_exception(self) -> None: result = self.run_repo(valid_workflow(), audit_policy={"schema": 1}) self.assertFalse(result["ok"]) From 3c84f9a93674a5957fe731f70330ea94404fcf55 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Wed, 26 Aug 2026 19:12:57 +0300 Subject: [PATCH 17/37] docs(af01): record completed Stack A implementation tasks --- .../tasks.md | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/specs/015-af-01-trusted-development-baseline/tasks.md b/specs/015-af-01-trusted-development-baseline/tasks.md index 5f1de8af..d5999196 100644 --- a/specs/015-af-01-trusted-development-baseline/tasks.md +++ b/specs/015-af-01-trusted-development-baseline/tasks.md @@ -16,19 +16,19 @@ Status: PLANNING_CANDIDATE - [x] **T002** Audit current repository assurance gaps: unprotected `main`; mixed mutable/immutable workflow references; missing cargo-deny/cargo-audit/zizmor/Scorecard; no fuzz/mutation/coverage/portability/release assurance program; stale README capability surface. - [x] **T003** Research current primary guidance for GitHub Actions full-SHA pinning, SLSA v1.2, Sigstore bundles, Rust fuzzing/mutation/coverage/security tooling, and HL7 FHIR release status. - [x] **T004** Add `docs/COMMAND_F_ASSURANCE_PROGRAM_2026-08-26.md` and AF-01 Spec Kit planning package; preserve CF-14/15/16 identities. -- [ ] **T005** Planning gate: exact final planning head passes all path-applicable CI, independent CodeRabbit/Qodo review truth is recorded without invented PASS, zero unresolved substantive planning findings remain, and planning PR is merged to canonical `main`. +- [x] **T005** Planning gate: exact final planning head passes all path-applicable CI, independent CodeRabbit/Qodo review truth is recorded without invented PASS, zero unresolved substantive planning findings remain, and planning PR is merged to canonical `main`. ## Phase 1 / Stack A — workflow trust audit and baseline hardening Depends on T005. -- [ ] **T010** Inventory every tracked `.github/workflows/*.yml|*.yaml`, every tracked Action metadata file named `action.yml` or `action.yaml` at any repository depth, every external `uses:` reference, runner label, workflow/job permission, checkout credential setting, job/service container image identity, and cargo lockfile-consuming command on canonical planning main. -- [ ] **T011** Define a minimal checked-in AF-01 workflow-trust policy format that makes allowed workflow/job permissions and proof-container identity modes machine-checkable, including any narrowly scoped exception schema with reason/revisit condition. -- [ ] **T012** Implement repository-owned deterministic workflow-trust audit with complete workflow plus `action.yml`/`action.yaml` discovery, local-action allowance, full-40-hex external action/reusable-workflow requirement, checkout credential check, effective workflow/job permission normalization plus allowlist enforcement, proof-critical job/service container digest enforcement, and proof-runner policy. -- [ ] **T013** Add positive and counterexample tests for T012, including mutable external `uses:` in `action.yaml`, tag/branch/short-SHA rejection, missing `persist-credentials: false`, overbroad permission rejection, unresolved inherited/default permission rejection, proof-critical mutable job/service container rejection, new-workflow/action-metadata coverage, malformed input fail-closed behavior, and deterministic repeat output. -- [ ] **T014** Harden `.github/workflows/ci.yml` to full-SHA external Actions, credentialless checkout, explicit machine-checkable least permissions, fixed supported runner label, bounded timeout, and preserved existing semantic/test steps. -- [ ] **T015** Reconcile every other existing workflow and repository Action metadata file to the AF-01 baseline, including permission declarations and proof-critical container digest identity, without changing its product/oracle/proof semantics or path-filter authority except where later universal required-check aggregation is explicitly introduced. -- [ ] **T016** Add a regression that discovers both `action.yml` and `action.yaml` anywhere in the tracked tree and fails if a future workflow, Action metadata file, permission grant, external Action ref, checkout credential setting, or proof-critical container identity escapes AF-01 trust auditing. +- [x] **T010** Inventory every tracked `.github/workflows/*.yml|*.yaml`, every tracked Action metadata file named `action.yml` or `action.yaml` at any repository depth, every external `uses:` reference, runner label, workflow/job permission, checkout credential setting, job/service container image identity, and cargo lockfile-consuming command on canonical planning main. +- [x] **T011** Define a minimal checked-in AF-01 workflow-trust policy format that makes allowed workflow/job permissions and proof-container identity modes machine-checkable, including any narrowly scoped exception schema with reason/revisit condition. +- [x] **T012** Implement repository-owned deterministic workflow-trust audit with complete workflow plus `action.yml`/`action.yaml` discovery, local-action allowance, full-40-hex external action/reusable-workflow requirement, checkout credential check, effective workflow/job permission normalization plus allowlist enforcement, proof-critical job/service container digest enforcement, and proof-runner policy. +- [x] **T013** Add positive and counterexample tests for T012, including mutable external `uses:` in `action.yaml`, tag/branch/short-SHA rejection, missing `persist-credentials: false`, overbroad permission rejection, unresolved inherited/default permission rejection, proof-critical mutable job/service container rejection, new-workflow/action-metadata coverage, malformed input fail-closed behavior, and deterministic repeat output. +- [x] **T014** Harden `.github/workflows/ci.yml` to full-SHA external Actions, credentialless checkout, explicit machine-checkable least permissions, fixed supported runner label, bounded timeout, and preserved existing semantic/test steps. +- [x] **T015** Reconcile every other existing workflow and repository Action metadata file to the AF-01 baseline, including permission declarations and proof-critical container digest identity, without changing its product/oracle/proof semantics or path-filter authority except where later universal required-check aggregation is explicitly introduced. +- [x] **T016** Add a regression that discovers both `action.yml` and `action.yaml` anywhere in the tracked tree and fails if a future workflow, Action metadata file, permission grant, external Action ref, checkout credential setting, or proof-critical container identity escapes AF-01 trust auditing. - [ ] **T017** Run mandatory workspace gates and every path-applicable existing proof/oracle workflow on the exact Stack A head. - [ ] **T018** Request CodeRabbit and Qodo on exact Stack A head; disposition every substantive returned finding and require zero unresolved material review threads. - [ ] **T019** Merge Stack A only from its exact qualified head and record canonical merge/main/tree. From 81509057e28ea48f83be3186316f3e2948772d54 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:16:48 +0300 Subject: [PATCH 18/37] fix(af01): reject quoted trust syntax --- .github/scripts/audit_workflow_trust.py | 43 +++++++++++++++++++++---- 1 file changed, 36 insertions(+), 7 deletions(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index 571fc770..cc8d9c69 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -20,6 +20,10 @@ STEP_LIST_RE = re.compile(r"^(\s*)-\s+\S") PERMISSION_RE = re.compile(r"^([A-Za-z0-9_-]+):\s*(read|write|none)\s*$") FLOW_USES_RE = re.compile(r"[\[{,]\s*[\"']?uses[\"']?\s*:") +QUOTED_USES_RE = re.compile(r"^\s*(?:-\s*)?[\"']uses[\"']\s*:") +QUOTED_JOB_CONTAINER_RE = re.compile(r"^\s{4}[\"'](?:container|services)[\"']\s*:") +QUOTED_IMAGE_RE = re.compile(r"^\s*(?:[\"']image[\"'])\s*:") +FLOW_SERVICES_RE = re.compile(r"^\s{4}services\s*:\s*[\[{]") BLOCK_SCALAR_RE = re.compile(r":\s*[|>][+-]?\s*(?:#.*)?$") SHELL_SEPARATOR_RE = re.compile(r"(?:\r?\n|&&|\|\||;|(? set[int]: return indexes -def _unsupported_flow_uses(lines: list[str]) -> list[str]: - """Reject flow-style uses keys instead of silently under-parsing them.""" +def _unsupported_trust_syntax(lines: list[str]) -> list[str]: + """Reject valid YAML forms that the constrained trust parser cannot safely normalize.""" block_lines = _block_scalar_line_indexes(lines) unsupported: list[str] = [] + in_job_container = False + in_services = False + for index, line in enumerate(lines): if index in block_lines: continue stripped = line.lstrip() + indent = _indent(line) if not stripped or stripped.startswith("#"): continue if stripped.startswith("run:") or re.match(r"^-\s+run:\s*", stripped): continue - if FLOW_USES_RE.search(line): + + if QUOTED_USES_RE.search(line) or FLOW_USES_RE.search(line): + unsupported.append(line.strip()) + continue + + if indent == 4: + in_job_container = False + in_services = False + if QUOTED_JOB_CONTAINER_RE.search(line) or FLOW_SERVICES_RE.search(line): + unsupported.append(line.strip()) + continue + if stripped == "container:": + in_job_container = True + elif stripped == "services:": + in_services = True + continue + + if in_job_container and indent == 6 and QUOTED_IMAGE_RE.search(line): + unsupported.append(line.strip()) + continue + if in_services and indent == 8 and QUOTED_IMAGE_RE.search(line): unsupported.append(line.strip()) + return unsupported @@ -500,13 +529,13 @@ def _uses_findings(path: str, lines: list[str], policy: dict) -> list[Finding]: if not isinstance(rules, dict): return findings - for syntax in _unsupported_flow_uses(lines): + for syntax in _unsupported_trust_syntax(lines): findings.append( Finding( - "unsupported_uses_syntax", + "unsupported_trust_syntax", path, "", - f"flow-style uses syntax is not supported by the trust parser: {syntax}", + f"trust-sensitive YAML syntax is not supported by the constrained parser: {syntax}", ) ) @@ -946,4 +975,4 @@ def main() -> int: if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file From bc74c4ce12157940fa6e74440da5e7f4c83bb3f8 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:17:24 +0300 Subject: [PATCH 19/37] test(af01): cover quoted trust syntax --- .../test_audit_workflow_trust_syntax.py | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 .github/scripts/test_audit_workflow_trust_syntax.py diff --git a/.github/scripts/test_audit_workflow_trust_syntax.py b/.github/scripts/test_audit_workflow_trust_syntax.py new file mode 100644 index 00000000..40e67f33 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import importlib.util +import sys +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("audit_workflow_trust.py") +SPEC = importlib.util.spec_from_file_location("audit_workflow_trust_syntax_target", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +AUDIT = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = AUDIT +SPEC.loader.exec_module(AUDIT) + +CHECKOUT_SHA = "fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09" +CONTAINER_DIGEST = "9" * 64 +PATH = ".github/workflows/example.yml" +EXPECTED = { + "jobs": { + "build": { + "permissions": {"contents": "read"}, + "runner": "ubuntu-24.04", + "timeout_minutes": 10, + } + } +} +POLICY = { + "rules": { + "cargo_locked_subcommands": [ + "bench", + "build", + "check", + "clippy", + "doc", + "metadata", + "run", + "test", + ], + "require_container_digest": True, + "require_checkout_credentials_disabled": True, + "require_external_uses_full_sha": True, + }, + "exceptions": [], +} + + +def workflow(step: str, container: str | None = None) -> str: + container_block = container or ( + " container:\n" + f" image: docker.io/library/rust@sha256:{CONTAINER_DIGEST}\n" + ) + return f"""name: example +on: + pull_request: +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-24.04 + timeout-minutes: 10 +{container_block} steps: +{step} +""" + + +class QuotedTrustSyntaxTests(unittest.TestCase): + @staticmethod + def codes(findings: list[object]) -> list[str]: + return [finding.code for finding in findings] + + def test_quoted_uses_in_workflow_fails_closed(self) -> None: + text = workflow( + f" - \"uses\": actions/checkout@{CHECKOUT_SHA}\n" + " with:\n" + " persist-credentials: false" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_trust_syntax", self.codes(findings)) + + def test_quoted_uses_in_action_metadata_fails_closed(self) -> None: + text = f"""name: example +description: fixture +runs: + using: composite + steps: + - 'uses': actions/checkout@{CHECKOUT_SHA} + with: + persist-credentials: false +""" + findings = AUDIT.audit_action_metadata("nested/action.yaml", text, POLICY) + self.assertIn("unsupported_trust_syntax", self.codes(findings)) + + def test_quoted_job_container_key_fails_closed(self) -> None: + text = workflow( + " - name: Test\n" + " run: cargo test --locked --workspace", + container=( + " \"container\":\n" + f" image: docker.io/library/rust@sha256:{CONTAINER_DIGEST}\n" + ), + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_trust_syntax", self.codes(findings)) + + def test_quoted_container_image_key_fails_closed(self) -> None: + text = workflow( + " - name: Test\n" + " run: cargo test --locked --workspace", + container=( + " container:\n" + f" \"image\": docker.io/library/rust@sha256:{CONTAINER_DIGEST}\n" + ), + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_trust_syntax", self.codes(findings)) + + def test_flow_style_services_fails_closed(self) -> None: + text = workflow( + " - name: Test\n" + " run: cargo test --locked --workspace" + ).replace( + " steps:\n", + " services: {db: {image: postgres:18}}\n steps:\n", + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_trust_syntax", self.codes(findings)) + + def test_quoted_image_under_env_is_not_container_authority(self) -> None: + text = workflow( + " - name: Test\n" + " env:\n" + " \"image\": harmless-string\n" + " run: cargo test --locked --workspace" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertNotIn("unsupported_trust_syntax", self.codes(findings)) + + def test_quoted_uses_text_inside_run_block_is_ignored(self) -> None: + text = workflow( + " - name: Test\n" + " run: |\n" + " printf '%s\\n' '- \"uses\": owner/action@v1'\n" + " cargo test --locked --workspace" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertNotIn("unsupported_trust_syntax", self.codes(findings)) + + +if __name__ == "__main__": + unittest.main() From 315d8cbc8b9cf408d593e7d82cca3d15ea27ab49 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:18:00 +0300 Subject: [PATCH 20/37] test(af01): discover all trust audit tests --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1409a08f..748db22b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,7 +25,7 @@ jobs: with: components: rustfmt, clippy - name: AF-01 workflow trust audit tests - run: python3 .github/scripts/test_audit_workflow_trust.py + run: python3 -m unittest discover -s .github/scripts -p 'test_audit_workflow_trust*.py' - name: AF-01 repository workflow trust audit run: python3 .github/scripts/audit_workflow_trust.py - name: Format From 35e1db516e9983cd6f6a495450a54c613b88a316 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:20:28 +0300 Subject: [PATCH 21/37] test(af01): align trust syntax findings --- .github/scripts/test_audit_workflow_trust.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py index 08b9adb9..c342b877 100644 --- a/.github/scripts/test_audit_workflow_trust.py +++ b/.github/scripts/test_audit_workflow_trust.py @@ -190,7 +190,7 @@ def test_flow_style_workflow_uses_fails_closed(self) -> None: " steps:\n - uses:", " steps: [{uses: owner/action@v1}]\n ignored:\n - uses:" ) result = self.run_repo(workflow) - self.assertIn("unsupported_uses_syntax", self.codes(result)) + self.assertIn("unsupported_trust_syntax", self.codes(result)) def test_flow_style_action_metadata_uses_fails_closed(self) -> None: action = """name: nested @@ -200,7 +200,7 @@ def test_flow_style_action_metadata_uses_fails_closed(self) -> None: steps: [{uses: owner/action@v1}] """ result = self.run_repo(valid_workflow(), action) - self.assertIn("unsupported_uses_syntax", self.codes(result)) + self.assertIn("unsupported_trust_syntax", self.codes(result)) def test_checkout_credentials_must_be_disabled(self) -> None: workflow = valid_workflow().replace( From 3b64052210c7b03b7e45f9c8444f16dc988da4a6 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:22:41 +0300 Subject: [PATCH 22/37] fix(af01): allow cargo version introspection --- .github/scripts/audit_workflow_trust.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index cc8d9c69..4d7d035b 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -30,6 +30,7 @@ LOCKFILE_CARGO_SUBCOMMANDS = frozenset( {"bench", "build", "check", "clippy", "doc", "metadata", "run", "test"} ) +CARGO_INFO_FLAGS = frozenset({"--version", "-V"}) BOOLEAN_RULES = frozenset( { "require_container_digest", @@ -453,6 +454,8 @@ def _cargo_findings( ) continue subcommand = tokens[command_index] + if subcommand in CARGO_INFO_FLAGS and command_index == len(tokens) - 1: + continue if subcommand.startswith("-"): findings.append( Finding( From 732c5bbb150fab66e502f1ce7b7a7a5144e696c4 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:23:05 +0300 Subject: [PATCH 23/37] test(af01): cover cargo version introspection --- .../test_audit_workflow_trust_syntax.py | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/.github/scripts/test_audit_workflow_trust_syntax.py b/.github/scripts/test_audit_workflow_trust_syntax.py index 40e67f33..bb2e938b 100644 --- a/.github/scripts/test_audit_workflow_trust_syntax.py +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -146,6 +146,34 @@ def test_quoted_uses_text_inside_run_block_is_ignored(self) -> None: findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) self.assertNotIn("unsupported_trust_syntax", self.codes(findings)) + def test_cargo_version_is_non_lockfile_introspection(self) -> None: + text = workflow( + " - name: Version\n" + " run: cargo --version" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + codes = self.codes(findings) + self.assertNotIn("unsupported_cargo_syntax", codes) + self.assertNotIn("cargo_unlocked", codes) + + def test_toolchain_selected_cargo_version_is_non_lockfile_introspection(self) -> None: + text = workflow( + " - name: Version\n" + " run: cargo +1.97.1 --version" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + codes = self.codes(findings) + self.assertNotIn("unsupported_cargo_syntax", codes) + self.assertNotIn("cargo_unlocked", codes) + + def test_cargo_version_with_extra_tokens_fails_closed(self) -> None: + text = workflow( + " - name: Invalid\n" + " run: cargo --version test" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_cargo_syntax", self.codes(findings)) + if __name__ == "__main__": unittest.main() From 283d425aad03289e19530212b8358bcaadb3090f Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:24:54 +0300 Subject: [PATCH 24/37] fix(af01): reject quoted permission overrides --- .github/scripts/audit_workflow_trust.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index 4d7d035b..d32a8736 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -23,6 +23,7 @@ QUOTED_USES_RE = re.compile(r"^\s*(?:-\s*)?[\"']uses[\"']\s*:") QUOTED_JOB_CONTAINER_RE = re.compile(r"^\s{4}[\"'](?:container|services)[\"']\s*:") QUOTED_IMAGE_RE = re.compile(r"^\s*(?:[\"']image[\"'])\s*:") +QUOTED_PERMISSION_KEY_RE = re.compile(r"^(?: {4})?[\"']permissions[\"']\s*:") FLOW_SERVICES_RE = re.compile(r"^\s{4}services\s*:\s*[\[{]") BLOCK_SCALAR_RE = re.compile(r":\s*[|>][+-]?\s*(?:#.*)?$") SHELL_SEPARATOR_RE = re.compile(r"(?:\r?\n|&&|\|\||;|(? list[str]: if stripped.startswith("run:") or re.match(r"^-\s+run:\s*", stripped): continue - if QUOTED_USES_RE.search(line) or FLOW_USES_RE.search(line): + if ( + QUOTED_USES_RE.search(line) + or FLOW_USES_RE.search(line) + or QUOTED_PERMISSION_KEY_RE.search(line) + ): unsupported.append(line.strip()) continue From f6239ab64504eb326d2765c83925e097f737f7cd Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:25:17 +0300 Subject: [PATCH 25/37] test(af01): cover quoted permission override --- .github/scripts/test_audit_workflow_trust_syntax.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/.github/scripts/test_audit_workflow_trust_syntax.py b/.github/scripts/test_audit_workflow_trust_syntax.py index bb2e938b..86e2febc 100644 --- a/.github/scripts/test_audit_workflow_trust_syntax.py +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -126,6 +126,17 @@ def test_flow_style_services_fails_closed(self) -> None: findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) self.assertIn("unsupported_trust_syntax", self.codes(findings)) + def test_quoted_job_permission_override_fails_closed(self) -> None: + text = workflow( + " - name: Test\n" + " run: cargo test --locked --workspace" + ).replace( + " steps:\n", + " \"permissions\":\n contents: write\n steps:\n", + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_trust_syntax", self.codes(findings)) + def test_quoted_image_under_env_is_not_container_authority(self) -> None: text = workflow( " - name: Test\n" From d0753ed7d7839c9958d830285b05f59e7a9af83f Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:35:57 +0300 Subject: [PATCH 26/37] fix(af01): fail closed on indirect cargo execution --- .github/scripts/audit_workflow_trust.py | 95 ++++++++++++++++++++++++- 1 file changed, 93 insertions(+), 2 deletions(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index d32a8736..0259b954 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -27,11 +27,18 @@ FLOW_SERVICES_RE = re.compile(r"^\s{4}services\s*:\s*[\[{]") BLOCK_SCALAR_RE = re.compile(r":\s*[|>][+-]?\s*(?:#.*)?$") SHELL_SEPARATOR_RE = re.compile(r"(?:\r?\n|&&|\|\||;|(? list[str]: return [segment.strip() for segment in SHELL_SEPARATOR_RE.split(joined) if segment.strip()] +def _command_token_index(tokens: list[str]) -> int | None: + """Locate a statically visible executable token in the supported shell subset.""" + index = 0 + while index < len(tokens) and SHELL_ASSIGNMENT_RE.fullmatch(tokens[index]): + index += 1 + while index < len(tokens) and tokens[index] in SHELL_CONTROL_WORDS: + index += 1 + if index >= len(tokens): + return None + + if tokens[index] == "env": + index += 1 + while index < len(tokens) and ( + tokens[index].startswith("-") or SHELL_ASSIGNMENT_RE.fullmatch(tokens[index]) + ): + index += 1 + if index >= len(tokens): + return None + + if tokens[index] in SHELL_COMMAND_WRAPPERS: + index += 1 + while index < len(tokens) and tokens[index].startswith("-"): + index += 1 + if index >= len(tokens): + return None + return index + + +def _indirect_cargo_finding(path: str, job: str, segment: str, tokens: list[str]) -> Finding | None: + """Fail closed when shell indirection could hide Cargo from static command analysis.""" + command_index = _command_token_index(tokens) + if command_index is not None: + command = tokens[command_index] + if ( + BARE_DYNAMIC_COMMAND_RE.fullmatch(command) + or command.startswith("$(") + or command.startswith("`") + ): + return Finding( + "unsupported_cargo_indirect", + path, + job, + f"dynamic executable token could resolve to Cargo: {segment}", + ) + if command in DYNAMIC_COMMAND_BUILTINS: + return Finding( + "unsupported_cargo_indirect", + path, + job, + f"dynamic shell execution is not statically auditable for Cargo: {segment}", + ) + if command in SHELL_INTERPRETERS and "-c" in tokens[command_index + 1 :]: + return Finding( + "unsupported_cargo_indirect", + path, + job, + f"nested shell execution is not statically auditable for Cargo: {segment}", + ) + + for token in tokens: + if token == "cargo": + continue + if CARGO_WORD_RE.search(token) and ( + "=" in token + or " " in token + or "$(" in token + or "`" in token + or token.endswith(")") + ): + return Finding( + "unsupported_cargo_indirect", + path, + job, + f"Cargo appears inside an indirect shell token: {segment}", + ) + return None + + def _cargo_findings( path: str, job: str, @@ -427,7 +512,8 @@ def _cargo_findings( findings: list[Finding] = [] for script in _run_scripts(lines, start, end): for segment in _logical_shell_segments(script): - if "cargo" not in segment: + interesting = "cargo" in segment or "$" in segment or "`" in segment + if not interesting: continue try: tokens = shlex.split(segment, comments=True, posix=True) @@ -437,11 +523,16 @@ def _cargo_findings( "unsupported_shell_syntax", path, job, - f"cannot safely parse Cargo-containing shell segment: {error}: {segment}", + f"cannot safely parse potentially Cargo-related shell segment: {error}: {segment}", ) ) continue + indirect = _indirect_cargo_finding(path, job, segment, tokens) + if indirect is not None: + findings.append(indirect) + continue + for index, token in enumerate(tokens): if token != "cargo": continue From 2d0ee7aab06ac88a456678cb7a234aab763a467c Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 04:36:27 +0300 Subject: [PATCH 27/37] test(af01): cover indirect cargo execution --- .../test_audit_workflow_trust_syntax.py | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/.github/scripts/test_audit_workflow_trust_syntax.py b/.github/scripts/test_audit_workflow_trust_syntax.py index 86e2febc..2add604d 100644 --- a/.github/scripts/test_audit_workflow_trust_syntax.py +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -185,6 +185,48 @@ def test_cargo_version_with_extra_tokens_fails_closed(self) -> None: findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) self.assertIn("unsupported_cargo_syntax", self.codes(findings)) + def test_variable_expanded_cargo_command_fails_closed(self) -> None: + text = workflow( + " - name: Indirect\n" + " run: |\n" + " tool=cargo\n" + " \"$tool\" test --workspace" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_cargo_indirect", self.codes(findings)) + + def test_command_substitution_cargo_command_fails_closed(self) -> None: + text = workflow( + " - name: Indirect\n" + " run: $(command -v cargo) test --workspace" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_cargo_indirect", self.codes(findings)) + + def test_eval_cargo_command_fails_closed(self) -> None: + text = workflow( + " - name: Indirect\n" + " run: eval 'cargo test --workspace'" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_cargo_indirect", self.codes(findings)) + + def test_nested_shell_cargo_command_fails_closed(self) -> None: + text = workflow( + " - name: Indirect\n" + " run: bash -c 'cargo test --workspace'" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_cargo_indirect", self.codes(findings)) + + def test_dynamic_path_executable_does_not_false_positive(self) -> None: + text = workflow( + " - name: Java\n" + " run: \"$JAVA_HOME/bin/java\" -version" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertNotIn("unsupported_cargo_indirect", self.codes(findings)) + if __name__ == "__main__": unittest.main() From 6a0882d87f7f6efffd11c5cd32173ece24ecc9e6 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:17:15 +0300 Subject: [PATCH 28/37] fix(af01): scope indirect Cargo detection to executable authority --- .github/scripts/audit_workflow_trust.py | 192 +++++++++++++----------- 1 file changed, 102 insertions(+), 90 deletions(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index 0259b954..0d48b2df 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -30,6 +30,23 @@ SHELL_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=.*$") BARE_DYNAMIC_COMMAND_RE = re.compile(r"^\$(?:[A-Za-z_][A-Za-z0-9_]*|\{[A-Za-z_][A-Za-z0-9_]*\})$") CARGO_WORD_RE = re.compile(r"(? int | None: return index -def _indirect_cargo_finding(path: str, job: str, segment: str, tokens: list[str]) -> Finding | None: - """Fail closed when shell indirection could hide Cargo from static command analysis.""" - command_index = _command_token_index(tokens) - if command_index is not None: - command = tokens[command_index] - if ( - BARE_DYNAMIC_COMMAND_RE.fullmatch(command) - or command.startswith("$(") - or command.startswith("`") - ): - return Finding( - "unsupported_cargo_indirect", - path, - job, - f"dynamic executable token could resolve to Cargo: {segment}", - ) - if command in DYNAMIC_COMMAND_BUILTINS: - return Finding( - "unsupported_cargo_indirect", - path, - job, - f"dynamic shell execution is not statically auditable for Cargo: {segment}", - ) - if command in SHELL_INTERPRETERS and "-c" in tokens[command_index + 1 :]: - return Finding( - "unsupported_cargo_indirect", - path, - job, - f"nested shell execution is not statically auditable for Cargo: {segment}", - ) - - for token in tokens: - if token == "cargo": - continue - if CARGO_WORD_RE.search(token) and ( - "=" in token - or " " in token - or "$(" in token - or "`" in token - or token.endswith(")") - ): - return Finding( - "unsupported_cargo_indirect", - path, - job, - f"Cargo appears inside an indirect shell token: {segment}", - ) +def _raw_indirect_cargo_finding(path: str, job: str, segment: str) -> Finding | None: + """Reject executable-position indirection without parsing unrelated shell arguments.""" + if DYNAMIC_EXECUTABLE_RE.match(segment): + return Finding( + "unsupported_cargo_indirect", + path, + job, + f"dynamic executable position could resolve to Cargo: {segment}", + ) + if DYNAMIC_SHELL_RE.match(segment): + return Finding( + "unsupported_cargo_indirect", + path, + job, + f"dynamic shell execution is not statically auditable for Cargo: {segment}", + ) return None @@ -512,8 +498,11 @@ def _cargo_findings( findings: list[Finding] = [] for script in _run_scripts(lines, start, end): for segment in _logical_shell_segments(script): - interesting = "cargo" in segment or "$" in segment or "`" in segment - if not interesting: + indirect = _raw_indirect_cargo_finding(path, job, segment) + if indirect is not None: + findings.append(indirect) + continue + if "cargo" not in segment: continue try: tokens = shlex.split(segment, comments=True, posix=True) @@ -523,65 +512,88 @@ def _cargo_findings( "unsupported_shell_syntax", path, job, - f"cannot safely parse potentially Cargo-related shell segment: {error}: {segment}", + f"cannot safely parse Cargo-containing shell segment: {error}: {segment}", ) ) continue - indirect = _indirect_cargo_finding(path, job, segment, tokens) - if indirect is not None: - findings.append(indirect) - continue - - for index, token in enumerate(tokens): - if token != "cargo": - continue - command_index = index + 1 - if command_index < len(tokens) and tokens[command_index].startswith("+"): - command_index += 1 - if command_index >= len(tokens): + command_index = _command_token_index(tokens) + if command_index is None: + if any(CARGO_WORD_RE.search(token) for token in tokens): findings.append( Finding( - "unsupported_cargo_syntax", + "unsupported_cargo_indirect", path, job, - f"cannot identify Cargo subcommand: {segment}", + f"Cargo appears without a statically executable command: {segment}", ) ) - continue - subcommand = tokens[command_index] - if subcommand in CARGO_INFO_FLAGS and command_index == len(tokens) - 1: - continue - if subcommand.startswith("-"): - findings.append( - Finding( - "unsupported_cargo_syntax", - path, - job, - f"Cargo global-option syntax requires explicit auditor support: {segment}", - ) + continue + + command = tokens[command_index] + if "$" in command or "`" in command: + findings.append( + Finding( + "unsupported_cargo_indirect", + path, + job, + f"dynamic Cargo executable path is not supported: {segment}", ) - continue - if subcommand not in locked_subcommands: - continue - next_cargo = next( - ( - position - for position in range(command_index + 1, len(tokens)) - if tokens[position] == "cargo" - ), - len(tokens), ) - invocation = tokens[index:next_cargo] - if "--locked" not in invocation: + continue + + is_direct_cargo = command == "cargo" or command.rsplit("/", 1)[-1] == "cargo" + if not is_direct_cargo: + if any(CARGO_WORD_RE.search(token) for token in tokens): findings.append( Finding( - "cargo_unlocked", + "unsupported_cargo_indirect", path, job, - f"cargo {subcommand} invocation omits --locked: {' '.join(invocation)}", + f"Cargo appears outside the statically executable command position: {segment}", ) ) + continue + + index = command_index + subcommand_index = index + 1 + if subcommand_index < len(tokens) and tokens[subcommand_index].startswith("+"): + subcommand_index += 1 + if subcommand_index >= len(tokens): + findings.append( + Finding( + "unsupported_cargo_syntax", + path, + job, + f"cannot identify Cargo subcommand: {segment}", + ) + ) + continue + subcommand = tokens[subcommand_index] + if subcommand in CARGO_INFO_FLAGS and subcommand_index == len(tokens) - 1: + continue + if subcommand.startswith("-"): + findings.append( + Finding( + "unsupported_cargo_syntax", + path, + job, + f"Cargo global-option syntax requires explicit auditor support: {segment}", + ) + ) + continue + if subcommand not in locked_subcommands: + continue + invocation = tokens[index:] + if "--locked" not in invocation: + findings.append( + Finding( + "cargo_unlocked", + path, + job, + f"cargo {subcommand} invocation omits --locked: {' '.join(invocation)}", + ) + ) return findings From 3460b695d49c29f73894b9256f44f30335b402da Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:17:46 +0300 Subject: [PATCH 29/37] test(af01): cover unrelated shell substitutions --- .../test_audit_workflow_trust_syntax.py | 20 +++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/.github/scripts/test_audit_workflow_trust_syntax.py b/.github/scripts/test_audit_workflow_trust_syntax.py index 2add604d..5f83be8a 100644 --- a/.github/scripts/test_audit_workflow_trust_syntax.py +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -227,6 +227,26 @@ def test_dynamic_path_executable_does_not_false_positive(self) -> None: findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) self.assertNotIn("unsupported_cargo_indirect", self.codes(findings)) + def test_unrelated_command_substitution_does_not_false_positive(self) -> None: + text = workflow( + " - name: Metadata\n" + " run: |\n" + " VALUE=\"$(python -c 'print(1)')\"\n" + " test \"$VALUE\" = \"1\"" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + codes = self.codes(findings) + self.assertNotIn("unsupported_cargo_indirect", codes) + self.assertNotIn("unsupported_shell_syntax", codes) + + def test_argument_variable_does_not_false_positive(self) -> None: + text = workflow( + " - name: Retry\n" + " run: echo \"attempt ${attempt}\"" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertNotIn("unsupported_cargo_indirect", self.codes(findings)) + if __name__ == "__main__": unittest.main() From e69e4b9a57302ad9683321a8472ffb56409a2b33 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:21:23 +0300 Subject: [PATCH 30/37] fix(af01): normalize heredocs and retry wrapper --- .github/scripts/audit_workflow_trust.py | 28 +++++++++++++++++++++---- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index 0d48b2df..2e266998 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -30,11 +30,12 @@ SHELL_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=.*$") BARE_DYNAMIC_COMMAND_RE = re.compile(r"^\$(?:[A-Za-z_][A-Za-z0-9_]*|\{[A-Za-z_][A-Za-z0-9_]*\})$") CARGO_WORD_RE = re.compile(r"(?-)?\s*(?P['\"]?)(?P[A-Za-z_][A-Za-z0-9_]*)\2") DYNAMIC_EXECUTABLE_RE = re.compile( r"""^\s* (?:(?:[A-Za-z_][A-Za-z0-9_]*=[^\s;|&]+)\s+)* (?:(?:!|do|if|then|until|while)\s+)* - (?:(?:env|command|exec|nohup)(?:\s+-[^\s]+|\s+[A-Za-z_][A-Za-z0-9_]*=[^\s;|&]+)*\s+)* + (?:(?:env|command|exec|nohup|retry)(?:\s+-[^\s]+|\s+[A-Za-z_][A-Za-z0-9_]*=[^\s;|&]+)*\s+)* [\"']? (?: \$(?:[A-Za-z_][A-Za-z0-9_]*(?=[\"']?(?:\s|$))|\{[A-Za-z_][A-Za-z0-9_]*\}(?=[\"']?(?:\s|$))) @@ -53,7 +54,7 @@ ) CARGO_INFO_FLAGS = frozenset({"--version", "-V"}) DYNAMIC_COMMAND_BUILTINS = frozenset({"eval", "alias"}) -SHELL_COMMAND_WRAPPERS = frozenset({"command", "exec", "nohup"}) +SHELL_COMMAND_WRAPPERS = frozenset({"command", "exec", "nohup", "retry"}) SHELL_INTERPRETERS = frozenset({"bash", "dash", "ksh", "sh", "zsh"}) SHELL_CONTROL_WORDS = frozenset({"!", "do", "if", "then", "until", "while"}) BOOLEAN_RULES = frozenset( @@ -434,9 +435,28 @@ def _run_scripts(lines: list[str], start: int, end: int) -> list[str]: return scripts +def _without_heredoc_bodies(script: str) -> str: + """Remove heredoc bodies because their contents are data, not shell commands.""" + kept: list[str] = [] + pending: list[tuple[str, bool]] = [] + for line in script.splitlines(): + if pending: + delimiter, strip_tabs = pending[0] + candidate = line.lstrip("\t") if strip_tabs else line + if candidate == delimiter: + pending.pop(0) + continue + + kept.append(line) + for matched in HEREDOC_RE.finditer(line): + pending.append((matched.group("delimiter"), matched.group("tabs") is not None)) + return "\n".join(kept) + + def _logical_shell_segments(script: str) -> list[str]: - """Join backslash continuations, then split at shell command boundaries.""" - joined = re.sub(r"\\[ \t]*\r?\n[ \t]*", " ", script) + """Join backslash continuations, strip heredoc data, then split shell boundaries.""" + command_text = _without_heredoc_bodies(script) + joined = re.sub(r"\\[ \t]*\r?\n[ \t]*", " ", command_text) return [segment.strip() for segment in SHELL_SEPARATOR_RE.split(joined) if segment.strip()] From 02150096eee0980dd4da6f5a4152e0152366ccf7 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:21:56 +0300 Subject: [PATCH 31/37] test(af01): cover heredoc data and retry Cargo authority --- .../test_audit_workflow_trust_syntax.py | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) diff --git a/.github/scripts/test_audit_workflow_trust_syntax.py b/.github/scripts/test_audit_workflow_trust_syntax.py index 5f83be8a..df0bc18f 100644 --- a/.github/scripts/test_audit_workflow_trust_syntax.py +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -247,6 +247,47 @@ def test_argument_variable_does_not_false_positive(self) -> None: findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) self.assertNotIn("unsupported_cargo_indirect", self.codes(findings)) + def test_heredoc_cargo_text_is_data_not_shell_authority(self) -> None: + text = workflow( + " - name: Evidence\n" + " run: |\n" + " python3 - <<'PY'\n" + " import subprocess\n" + " evidence = {'cargo': subprocess.check_output(['cargo', '--version'], text=True)}\n" + " print(evidence)\n" + " PY" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + codes = self.codes(findings) + self.assertNotIn("unsupported_cargo_indirect", codes) + self.assertNotIn("cargo_unlocked", codes) + + def test_retry_wrapped_locked_cargo_is_enforced_and_allowed(self) -> None: + text = workflow( + " - name: Retry\n" + " run: retry cargo test --locked --workspace" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + codes = self.codes(findings) + self.assertNotIn("unsupported_cargo_indirect", codes) + self.assertNotIn("cargo_unlocked", codes) + + def test_retry_wrapped_unlocked_cargo_fails(self) -> None: + text = workflow( + " - name: Retry\n" + " run: retry cargo test --workspace" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("cargo_unlocked", self.codes(findings)) + + def test_retry_wrapped_dynamic_executable_fails_closed(self) -> None: + text = workflow( + " - name: Retry\n" + " run: retry \"$tool\" test --workspace" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_cargo_indirect", self.codes(findings)) + if __name__ == "__main__": unittest.main() From b62a464581c278efb05145c1822775de91250811 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:27:39 +0300 Subject: [PATCH 32/37] ci(af01): remove implicit retry wrapper trust --- .github/workflows/registry-download-smoke.yml | 50 +++++++++---------- 1 file changed, 23 insertions(+), 27 deletions(-) diff --git a/.github/workflows/registry-download-smoke.yml b/.github/workflows/registry-download-smoke.yml index afc6216d..7c6e06e4 100644 --- a/.github/workflows/registry-download-smoke.yml +++ b/.github/workflows/registry-download-smoke.yml @@ -26,7 +26,7 @@ jobs: - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5 / node24 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@032958afbdc797a9164d3bc0b56325c1308924a5 # 1.97.1 + - uses: dtolnay/rust-toolchain@032958afbdc797a916606bdc6ffcbf961 # 1.97.1 with: components: rustfmt, clippy @@ -43,37 +43,33 @@ jobs: shell: bash run: | set -euo pipefail - retry() { - local attempt=1 - until "$@"; do - if (( attempt >= 3 )); then - echo "real primary registry probe failed after ${attempt} attempts" >&2 - return 1 - fi - echo "real primary registry probe failed on attempt ${attempt}; retrying" >&2 - sleep $((attempt * 5)) - attempt=$((attempt + 1)) - done - } - retry cargo test --locked -p commandf-pkg registry::tests::real_primary_us_core_is_direct_gzip -- --ignored --exact + for attempt in 1 2 3; do + if cargo test --locked -p commandf-pkg registry::tests::real_primary_us_core_is_direct_gzip -- --ignored --exact; then + break + fi + if (( attempt >= 3 )); then + echo "real primary registry probe failed after ${attempt} attempts" >&2 + exit 1 + fi + echo "real primary registry probe failed on attempt ${attempt}; retrying" >&2 + sleep $((attempt * 5)) + done - name: Real secondary redirect-to-tarball response shell: bash run: | set -euo pipefail - retry() { - local attempt=1 - until "$@"; do - if (( attempt >= 3 )); then - echo "real secondary registry probe failed after ${attempt} attempts" >&2 - return 1 - fi - echo "real secondary registry probe failed on attempt ${attempt}; retrying" >&2 - sleep $((attempt * 5)) - attempt=$((attempt + 1)) - done - } - retry cargo test --locked -p commandf-pkg registry::tests::real_secondary_us_core_follows_only_expected_tarball -- --ignored --exact + for attempt in 1 2 3; do + if cargo test --locked -p commandf-pkg registry::tests::real_secondary_us_core_follows_only_expected_tarball -- --ignored --exact; then + break + fi + if (( attempt >= 3 )); then + echo "real secondary registry probe failed after ${attempt} attempts" >&2 + exit 1 + fi + echo "real secondary registry probe failed on attempt ${attempt}; retrying" >&2 + sleep $((attempt * 5)) + done - name: End-to-end exact VSAC fallback resolve and verify shell: bash From 1b550f631a8430c7b28a3a44c7bfb0fba6f68f25 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:27:55 +0300 Subject: [PATCH 33/37] fix(af01): restore exact rust toolchain pin --- .github/workflows/registry-download-smoke.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/registry-download-smoke.yml b/.github/workflows/registry-download-smoke.yml index 7c6e06e4..84f2f676 100644 --- a/.github/workflows/registry-download-smoke.yml +++ b/.github/workflows/registry-download-smoke.yml @@ -26,7 +26,7 @@ jobs: - uses: actions/checkout@fbc6f3992d24b796d5a048ff273f7fcc4a7b6c09 # v5 / node24 with: persist-credentials: false - - uses: dtolnay/rust-toolchain@032958afbdc797a916606bdc6ffcbf961 # 1.97.1 + - uses: dtolnay/rust-toolchain@032958afbdc797a9164d3bc0b56325c1308924a5 # 1.97.1 with: components: rustfmt, clippy From ab52bc6a69acf2f93124fa6f2f9f145805a13b78 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:29:14 +0300 Subject: [PATCH 34/37] fix(af01): fail closed on unverified retry wrappers --- .github/scripts/audit_workflow_trust.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py index 2e266998..d99886fb 100644 --- a/.github/scripts/audit_workflow_trust.py +++ b/.github/scripts/audit_workflow_trust.py @@ -54,7 +54,7 @@ ) CARGO_INFO_FLAGS = frozenset({"--version", "-V"}) DYNAMIC_COMMAND_BUILTINS = frozenset({"eval", "alias"}) -SHELL_COMMAND_WRAPPERS = frozenset({"command", "exec", "nohup", "retry"}) +SHELL_COMMAND_WRAPPERS = frozenset({"command", "exec", "nohup"}) SHELL_INTERPRETERS = frozenset({"bash", "dash", "ksh", "sh", "zsh"}) SHELL_CONTROL_WORDS = frozenset({"!", "do", "if", "then", "until", "while"}) BOOLEAN_RULES = frozenset( From 3de35108f5457d1d86de7b437ece8d359a395632 Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:29:44 +0300 Subject: [PATCH 35/37] test(af01): require direct Cargo retry authority --- .../test_audit_workflow_trust_syntax.py | 24 ++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/.github/scripts/test_audit_workflow_trust_syntax.py b/.github/scripts/test_audit_workflow_trust_syntax.py index df0bc18f..911d0022 100644 --- a/.github/scripts/test_audit_workflow_trust_syntax.py +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -262,20 +262,38 @@ def test_heredoc_cargo_text_is_data_not_shell_authority(self) -> None: self.assertNotIn("unsupported_cargo_indirect", codes) self.assertNotIn("cargo_unlocked", codes) - def test_retry_wrapped_locked_cargo_is_enforced_and_allowed(self) -> None: + def test_unverified_retry_wrapper_fails_closed(self) -> None: text = workflow( " - name: Retry\n" " run: retry cargo test --locked --workspace" ) findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) + self.assertIn("unsupported_cargo_indirect", self.codes(findings)) + + def test_direct_locked_cargo_retry_loop_is_allowed(self) -> None: + text = workflow( + " - name: Retry\n" + " run: |\n" + " for attempt in 1 2 3; do\n" + " if cargo test --locked --workspace; then\n" + " break\n" + " fi\n" + " done" + ) + findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) codes = self.codes(findings) self.assertNotIn("unsupported_cargo_indirect", codes) self.assertNotIn("cargo_unlocked", codes) - def test_retry_wrapped_unlocked_cargo_fails(self) -> None: + def test_direct_unlocked_cargo_retry_loop_fails(self) -> None: text = workflow( " - name: Retry\n" - " run: retry cargo test --workspace" + " run: |\n" + " for attempt in 1 2 3; do\n" + " if cargo test --workspace; then\n" + " break\n" + " fi\n" + " done" ) findings = AUDIT.audit_workflow(PATH, text, EXPECTED, POLICY) self.assertIn("cargo_unlocked", self.codes(findings)) From 9d1f14b16ff04e21315170486b76947488f47bbc Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:38:05 +0300 Subject: [PATCH 36/37] fix(af01): audit executable shell surfaces --- .../scripts/audit_workflow_trust_surface.py | 336 ++++++++++++++++++ 1 file changed, 336 insertions(+) create mode 100644 .github/scripts/audit_workflow_trust_surface.py diff --git a/.github/scripts/audit_workflow_trust_surface.py b/.github/scripts/audit_workflow_trust_surface.py new file mode 100644 index 00000000..e1bf5f7d --- /dev/null +++ b/.github/scripts/audit_workflow_trust_surface.py @@ -0,0 +1,336 @@ +#!/usr/bin/env python3 +"""Complement AF-01 trust auditing for executable shell surfaces. + +The primary workflow audit intentionally uses a constrained parser. This companion gate closes +shell-authority boundaries that require source-aware handling: shell-interpreter heredocs, +composite Action run steps, and statically referenced local Action shell scripts. +""" + +from __future__ import annotations + +import json +import re +import shlex +import sys +from pathlib import Path +from typing import Iterable + +import audit_workflow_trust as core + +SHELL_HEREDOC_RE = re.compile( + r"(?:^|\s)(?:bash|dash|ksh|sh|zsh)\b[^\n;]*(?:<<-?\s*[\"']?[A-Za-z_][A-Za-z0-9_]*[\"']?)" +) +ACTION_LOCAL_SCRIPT_RE = re.compile( + r"\$\{?GITHUB_ACTION_PATH\}?/(?P[A-Za-z0-9_./-]+)" +) +VARIABLE_COMMAND_RE = re.compile( + r"^[\"']?\$(?:([A-Za-z_][A-Za-z0-9_]*)|\{([A-Za-z_][A-Za-z0-9_]*)\})[\"']?$" +) +ASSIGNMENT_RE = re.compile(r"^(?P[A-Za-z_][A-Za-z0-9_]*)=(?P.*)$") + + +def _finding(code: str, path: str, scope: str, detail: str) -> core.Finding: + return core.Finding(code, path, scope, detail) + + +def _scripts(lines: list[str]) -> list[str]: + return core._run_scripts(lines, -1, len(lines)) + + +def _shell_heredoc_findings(path: str, scope: str, script: str) -> list[core.Finding]: + findings: list[core.Finding] = [] + for line in script.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith("#"): + continue + if SHELL_HEREDOC_RE.search(stripped): + findings.append( + _finding( + "unsupported_shell_heredoc", + path, + scope, + f"shell-interpreter heredoc is executable authority and is not supported: {stripped}", + ) + ) + return findings + + +def _direct_cargo_findings( + path: str, scope: str, script: str, locked_subcommands: set[str] +) -> list[core.Finding]: + """Audit direct/obviously indirect Cargo in a shell source without treating text as authority.""" + findings = _shell_heredoc_findings(path, scope, script) + cargo_variables: set[str] = set() + + for segment in core._logical_shell_segments(script): + try: + tokens = shlex.split(segment, comments=True, posix=True) + except ValueError as error: + if "cargo" in segment: + findings.append( + _finding( + "unsupported_shell_syntax", + path, + scope, + f"cannot safely parse Cargo-containing shell segment: {error}: {segment}", + ) + ) + continue + if not tokens: + continue + + if len(tokens) == 1: + assignment = ASSIGNMENT_RE.fullmatch(tokens[0]) + if assignment: + value = assignment.group("value").strip("\"'") + if value == "cargo" or value.rsplit("/", 1)[-1] == "cargo": + cargo_variables.add(assignment.group("name")) + continue + + command_index = core._command_token_index(tokens) + if command_index is None or command_index >= len(tokens): + continue + command = tokens[command_index] + + variable = VARIABLE_COMMAND_RE.fullmatch(command) + if variable: + name = variable.group(1) or variable.group(2) + if name in cargo_variables: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"variable-expanded executable resolves to Cargo: {segment}", + ) + ) + continue + + if command.startswith("$(") or command.startswith("`"): + if "cargo" in segment: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"command-substituted executable can resolve to Cargo: {segment}", + ) + ) + continue + + if command in core.DYNAMIC_COMMAND_BUILTINS: + if "cargo" in segment: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"dynamic shell execution can hide Cargo: {segment}", + ) + ) + continue + if command in core.SHELL_INTERPRETERS and "-c" in tokens[command_index + 1 :]: + if "cargo" in segment: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"nested shell execution can hide Cargo: {segment}", + ) + ) + continue + + if not (command == "cargo" or command.rsplit("/", 1)[-1] == "cargo"): + continue + + subcommand_index = command_index + 1 + if subcommand_index < len(tokens) and tokens[subcommand_index].startswith("+"): + subcommand_index += 1 + if subcommand_index >= len(tokens): + findings.append( + _finding( + "unsupported_cargo_syntax", + path, + scope, + f"cannot identify Cargo subcommand: {segment}", + ) + ) + continue + subcommand = tokens[subcommand_index] + if subcommand in core.CARGO_INFO_FLAGS and subcommand_index == len(tokens) - 1: + continue + if subcommand.startswith("-"): + # `command -v cargo` is normalized by the primary scanner and is not a Cargo invocation. + continue + if subcommand in locked_subcommands and "--locked" not in tokens[command_index:]: + findings.append( + _finding( + "cargo_unlocked", + path, + scope, + f"cargo {subcommand} invocation omits --locked: {' '.join(tokens[command_index:])}", + ) + ) + return findings + + +def _action_local_targets(script: str) -> tuple[list[str], list[str]]: + """Return statically exposed GITHUB_ACTION_PATH shell targets and unsupported dynamic targets.""" + targets: list[str] = [] + unsupported: list[str] = [] + for segment in core._logical_shell_segments(script): + try: + tokens = shlex.split(segment, comments=True, posix=True) + except ValueError: + continue + command_index = core._command_token_index(tokens) + if command_index is None or command_index >= len(tokens): + continue + if tokens[command_index] not in core.SHELL_INTERPRETERS: + continue + args = tokens[command_index + 1 :] + if "-c" in args or "<<" in segment: + continue + script_arg = next((arg for arg in args if not arg.startswith("-")), None) + if script_arg is None: + unsupported.append(segment) + continue + matched = ACTION_LOCAL_SCRIPT_RE.search(script_arg) + if matched: + relative = matched.group("path") + if relative.startswith("/") or ".." in Path(relative).parts: + unsupported.append(segment) + else: + targets.append(relative) + continue + if "$" in script_arg or "`" in script_arg or script_arg.startswith("$("): + unsupported.append(segment) + return sorted(set(targets)), sorted(set(unsupported)) + + +def audit_action_text( + path: str, + text: str, + locked_subcommands: set[str], +) -> list[core.Finding]: + lines = text.splitlines() + findings: list[core.Finding] = [] + for script in _scripts(lines): + findings.extend(_direct_cargo_findings(path, "composite-action", script, locked_subcommands)) + _, unsupported = _action_local_targets(script) + for segment in unsupported: + findings.append( + _finding( + "unsupported_action_script", + path, + "composite-action", + f"Action shell source is dynamic or not statically exposed: {segment}", + ) + ) + return findings + + +def _audit_local_action_script( + root: Path, + action_path: str, + relative: str, + tracked: set[str], + locked_subcommands: set[str], +) -> list[core.Finding]: + action_dir = Path(action_path).parent + target = (action_dir / relative).as_posix() + if target.startswith("./"): + target = target[2:] + if target not in tracked: + return [ + _finding( + "untracked_action_script", + action_path, + "composite-action", + f"statically referenced Action shell source is not tracked: {target}", + ) + ] + try: + source = (root / target).read_text(encoding="utf-8") + except (OSError, UnicodeError) as error: + return [ + _finding( + "unreadable_action_script", + action_path, + "composite-action", + f"cannot read tracked Action shell source {target}: {error}", + ) + ] + return _direct_cargo_findings(target, "action-script", source, locked_subcommands) + + +def audit_repository_surface( + root: Path, + policy: dict, + tracked_files: Iterable[str] | None = None, +) -> dict: + paths = list(tracked_files) if tracked_files is not None else core._tracked_files(root) + tracked = set(paths) + workflows, actions = core.discover_security_files(paths) + rules = policy.get("rules", {}) if isinstance(policy, dict) else {} + raw_subcommands = rules.get("cargo_locked_subcommands", []) if isinstance(rules, dict) else [] + locked_subcommands = set(raw_subcommands) if isinstance(raw_subcommands, list) else set() + findings: list[core.Finding] = [] + + for workflow_path in workflows: + text = (root / workflow_path).read_text(encoding="utf-8") + lines = text.splitlines() + jobs, _ = core._job_ranges(lines) + for job, (start, end) in sorted(jobs.items()): + for script in core._run_scripts(lines, start, end): + findings.extend(_shell_heredoc_findings(workflow_path, job, script)) + + for action_path in actions: + text = (root / action_path).read_text(encoding="utf-8") + findings.extend(audit_action_text(action_path, text, locked_subcommands)) + for script in _scripts(text.splitlines()): + targets, _ = _action_local_targets(script) + for relative in targets: + findings.extend( + _audit_local_action_script( + root, + action_path, + relative, + tracked, + locked_subcommands, + ) + ) + + ordered = sorted(set(findings)) + return { + "schema": 1, + "ok": not ordered, + "findings": [finding.as_dict() for finding in ordered], + } + + +def main() -> int: + root = Path(".").resolve() + policy_path = root / ".github/workflow-trust-policy.json" + try: + policy = json.loads(policy_path.read_text(encoding="utf-8")) + result = audit_repository_surface(root, policy) + except (OSError, UnicodeError, json.JSONDecodeError) as error: + result = { + "schema": 1, + "ok": False, + "findings": [ + { + "code": "surface_audit_operational_failure", + "path": str(policy_path.relative_to(root)), + "detail": str(error), + } + ], + } + sys.stdout.write(json.dumps(result, indent=2, sort_keys=True) + "\n") + return 0 if result["ok"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) From 00ea8f865e3e31fb24fc1e1d9eead0ff4faa784c Mon Sep 17 00:00:00 2001 From: "Abdulaziz M. Shehri" Date: Thu, 27 Aug 2026 05:38:32 +0300 Subject: [PATCH 37/37] test(af01): cover composite and heredoc shell authority --- .../test_audit_workflow_trust_surface.py | 124 ++++++++++++++++++ 1 file changed, 124 insertions(+) create mode 100644 .github/scripts/test_audit_workflow_trust_surface.py diff --git a/.github/scripts/test_audit_workflow_trust_surface.py b/.github/scripts/test_audit_workflow_trust_surface.py new file mode 100644 index 00000000..e142a10f --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust_surface.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import importlib.util +import json +import sys +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("audit_workflow_trust_surface.py") +SPEC = importlib.util.spec_from_file_location("audit_workflow_trust_surface_target", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +SURFACE = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = SURFACE +SPEC.loader.exec_module(SURFACE) + +LOCKED = {"bench", "build", "check", "clippy", "doc", "metadata", "run", "test"} + + +def codes(findings: list[object]) -> list[str]: + return [finding.code for finding in findings] + + +def action(run: str) -> str: + return f"""name: fixture +description: fixture +runs: + using: composite + steps: + - shell: bash + run: {run} +""" + + +class ShellAuthoritySurfaceTests(unittest.TestCase): + def test_bash_heredoc_is_executable_authority(self) -> None: + script = "bash <<'SCRIPT'\ncargo test --workspace\nSCRIPT" + findings = SURFACE._shell_heredoc_findings("wf.yml", "build", script) + self.assertIn("unsupported_shell_heredoc", codes(findings)) + + def test_sh_heredoc_is_executable_authority(self) -> None: + script = "sh < None: + script = "python3 - <<'PY'\nprint('cargo test --workspace')\nPY" + findings = SURFACE._shell_heredoc_findings("wf.yml", "build", script) + self.assertNotIn("unsupported_shell_heredoc", codes(findings)) + + def test_action_yml_unlocked_cargo_fails(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("cargo test --workspace"), LOCKED + ) + self.assertIn("cargo_unlocked", codes(findings)) + + def test_action_yaml_unlocked_cargo_fails(self) -> None: + findings = SURFACE.audit_action_text( + "nested/action.yaml", action("cargo build --workspace"), LOCKED + ) + self.assertIn("cargo_unlocked", codes(findings)) + + def test_action_yml_locked_cargo_passes(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("cargo test --locked --workspace"), LOCKED + ) + self.assertNotIn("cargo_unlocked", codes(findings)) + + def test_action_yaml_locked_cargo_passes(self) -> None: + findings = SURFACE.audit_action_text( + "nested/action.yaml", action("cargo build --locked --workspace"), LOCKED + ) + self.assertNotIn("cargo_unlocked", codes(findings)) + + def test_action_dynamic_shell_source_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action('bash "$SCRIPT"'), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_tracked_local_action_script_is_audited(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / ".github").mkdir() + (root / ".github" / "workflow-trust-policy.json").write_text( + json.dumps({"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}), + encoding="utf-8", + ) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('bash "$GITHUB_ACTION_PATH/scripts/build.sh"'), encoding="utf-8" + ) + (root / "scripts" / "build.sh").write_text( + "#!/usr/bin/env bash\ncargo test --workspace\n", encoding="utf-8" + ) + result = SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/build.sh"], + ) + self.assertFalse(result["ok"]) + self.assertIn("cargo_unlocked", [item["code"] for item in result["findings"]]) + + def test_tracked_local_locked_action_script_passes(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('bash "$GITHUB_ACTION_PATH/scripts/build.sh"'), encoding="utf-8" + ) + (root / "scripts" / "build.sh").write_text( + "#!/usr/bin/env bash\ncargo test --locked --workspace\n", encoding="utf-8" + ) + result = SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/build.sh"], + ) + self.assertTrue(result["ok"], result["findings"]) + + +if __name__ == "__main__": + unittest.main()