diff --git a/.github/scripts/audit_workflow_trust.py b/.github/scripts/audit_workflow_trust.py new file mode 100644 index 00000000..d99886fb --- /dev/null +++ b/.github/scripts/audit_workflow_trust.py @@ -0,0 +1,1109 @@ +#!/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 shlex +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_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*:") +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|&&|\|\||;|(?-)?\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|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|$))) + |\$\( + |` + ) + """, + re.VERBOSE, +) +DYNAMIC_SHELL_RE = re.compile( + r"^\s*(?:(?:[A-Za-z_][A-Za-z0-9_]*=[^\s;|&]+)\s+)*(?:(?:!|do|if|then|until|while)\s+)*(?:eval\b|(?:bash|dash|ksh|sh|zsh)\b[^\n]*\s-c(?:\s|$))" +) + +LOCKFILE_CARGO_SUBCOMMANDS = frozenset( + {"bench", "build", "check", "clippy", "doc", "metadata", "run", "test"} +) +CARGO_INFO_FLAGS = frozenset({"--version", "-V"}) +DYNAMIC_COMMAND_BUILTINS = frozenset({"eval", "alias"}) +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( + { + "require_container_digest", + "require_checkout_credentials_disabled", + "require_external_uses_full_sha", + } +) +SUPPORTED_RULE_KEYS = frozenset({"cargo_locked_subcommands", *BOOLEAN_RULES}) +SUPPORTED_RUNNERS = frozenset({"ubuntu-24.04"}) +MAX_JOB_TIMEOUT_MINUTES = 30 +SUPPORTED_JOB_POLICY_KEYS = frozenset({"permissions", "runner", "timeout_minutes"}) +SUPPORTED_TOP_LEVEL_POLICY_KEYS = frozenset( + {"schema", "rules", "rationales", "workflows", "exceptions"} +) + + +@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( + ( + 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" + + 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]: + """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() + 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) + 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]]: + 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 _block_scalar_line_indexes(lines: list[str]) -> 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_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 ( + QUOTED_USES_RE.search(line) + or FLOW_USES_RE.search(line) + or QUOTED_PERMISSION_KEY_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 + + +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 _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(start + 1, len(lines)): + line = lines[index] + if not line.strip(): + continue + if _indent(line) == step_indent and STEP_LIST_RE.match(line): + end = index + break + 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]: + """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 _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, 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()] + + +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 _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 + + +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): + 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) + except ValueError as error: + findings.append( + Finding( + "unsupported_shell_syntax", + path, + job, + f"cannot safely parse Cargo-containing shell segment: {error}: {segment}", + ) + ) + continue + + 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_indirect", + path, + job, + f"Cargo appears without a statically executable command: {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 + + 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( + "unsupported_cargo_indirect", + path, + job, + 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 + + +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 + 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"}) + + +def _excepted(policy: dict, finding: Finding) -> bool: + 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: + continue + if "detail" in exception and exception["detail"] != finding.detail: + continue + return True + 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 syntax in _unsupported_trust_syntax(lines): + findings.append( + Finding( + "unsupported_trust_syntax", + path, + "", + f"trust-sensitive YAML syntax is not supported by the constrained 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 + ): + 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 with.persist-credentials: false exactly once", + ) + ) + return findings + + +def audit_workflow(path: str, text: str, expected: dict, policy: dict) -> list[Finding]: + findings: list[Finding] = [] + if "\t" in text: + return [Finding("malformed_yaml", path, "", "tab indentation is not supported")] + lines = text.splitlines() + jobs, jobs_error = _job_ranges(lines) + if jobs_error: + return [Finding("malformed_yaml", path, "", jobs_error)] + + expected_jobs = expected.get("jobs") if isinstance(expected, dict) else None + if not isinstance(expected_jobs, dict): + 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") + ) + 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)) + + 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}", + ) + ) + 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 isinstance(timeout_limit, bool) 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 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(rules.get("cargo_locked_subcommands", [])) + 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)] + + +def audit_action_metadata(path: str, text: str, policy: dict) -> 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", + ) + ) + findings.extend(_uses_findings(path, lines, policy)) + return [finding for finding in findings if not _excepted(policy, finding)] + + +def _policy_errors(policy: object) -> list[Finding]: + findings: 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")) + + rules = policy.get("rules") + if not isinstance(rules, dict): + findings.append(Finding("invalid_policy", policy_path, "", "rules 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( + 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["workflows"] + + 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()) \ No newline at end of file diff --git a/.github/scripts/audit_workflow_trust_environment_channels.py b/.github/scripts/audit_workflow_trust_environment_channels.py new file mode 100644 index 00000000..ad9f3e5e --- /dev/null +++ b/.github/scripts/audit_workflow_trust_environment_channels.py @@ -0,0 +1,370 @@ +#!/usr/bin/env python3 +"""Fail closed on environment authority that can alter shell execution across steps.""" + +from __future__ import annotations + +import json +import re +import shlex +import subprocess +import sys +from pathlib import Path +from typing import Iterable + +import audit_workflow_trust as core + +CHANNEL_NAME_RE = re.compile( + r"(?GITHUB_PATH|GITHUB_ENV)(?![A-Za-z0-9_])" +) +SHELL_STARTUP_NAME_RE = re.compile( + r"(?BASH_ENV|ENV|ZDOTDIR)(?![A-Za-z0-9_])" +) +INDIRECT_PARAMETER_RE = re.compile(r"\$\{!") +GITHUB_PREFIX_FRAGMENT_RE = re.compile(r"(? list[str]: + completed = subprocess.run( + ["git", "-C", str(root), "ls-files", "-z"], + check=False, + capture_output=True, + ) + if completed.returncode != 0: + raise RuntimeError("unable to enumerate tracked repository files") + return sorted( + item.decode("utf-8") + for item in completed.stdout.split(b"\0") + if item + ) + + +def _is_yaml_authority(path: str) -> bool: + candidate = Path(path) + return ( + path.startswith(".github/workflows/") and candidate.suffix in {".yml", ".yaml"} + ) or candidate.name in {"action.yml", "action.yaml"} + + +def _read_authority_text(root: Path, path: str) -> tuple[str | None, str | None]: + candidate = root / path + try: + raw = candidate.read_bytes() + except OSError as error: + if _is_yaml_authority(path) or candidate.suffix == ".sh": + return None, f"unable to read tracked authority file: {error}" + return None, None + + is_shell = candidate.suffix == ".sh" + if not is_shell: + prefix = raw[:256].decode("utf-8", errors="ignore") + is_shell = SHELL_SHEBANG_RE.match(prefix) is not None + if not (_is_yaml_authority(path) or is_shell): + return None, None + + try: + return raw.decode("utf-8"), None + except UnicodeDecodeError: + return None, "tracked authority file is not valid UTF-8" + + +def _dynamic_name(value: str) -> bool: + return "$" in value or "`" in value + + +def _basename(token: str) -> str: + return token.rsplit("/", 1)[-1] + + +def _raw_command_index(tokens: list[str]) -> int | None: + index = 0 + while index < len(tokens) and core.SHELL_ASSIGNMENT_RE.fullmatch(tokens[index]): + index += 1 + while index < len(tokens) and tokens[index] in core.SHELL_CONTROL_WORDS: + index += 1 + return index if index < len(tokens) else None + + +def _normalized_writer(tokens: list[str]) -> tuple[str, list[str]] | None: + index = _raw_command_index(tokens) + if index is None: + return None + command = _basename(tokens[index]) + index += 1 + if command in COMMAND_BUILTIN_WRAPPERS and index < len(tokens): + while index < len(tokens) and tokens[index].startswith("-"): + index += 1 + if index >= len(tokens): + return None + command = _basename(tokens[index]) + index += 1 + return command, tokens[index:] + + +def _before_redirection(args: list[str]) -> list[str]: + result: list[str] = [] + for token in args: + if token in {"<", ">", ">>", "<<", "<<<", "<>", ">&", "<&"}: + break + if re.match(r"^(?:\d*)?(?:>>?|<|>&|<&)", token): + break + result.append(token) + return result + + +def _dynamic_writer_detail(tokens: list[str]) -> str | None: + normalized = _normalized_writer(tokens) + if normalized is None: + return None + command, args = normalized + + if command in ASSIGNMENT_BUILTINS: + for arg in args: + if arg.startswith("-"): + continue + name = arg.split("=", 1)[0] + if _dynamic_name(name): + return f"{command} writes a dynamically constructed variable name: {arg}" + return None + + if command == "printf": + for index, arg in enumerate(args): + if arg == "-v" and index + 1 < len(args) and _dynamic_name(args[index + 1]): + return f"printf -v writes a dynamically constructed variable name: {args[index + 1]}" + return None + + if command in VARIABLE_TARGET_BUILTINS: + candidates = _before_redirection(args) + for arg in candidates: + if arg.startswith("-"): + continue + if _dynamic_name(arg): + return f"{command} writes a dynamically constructed variable target: {arg}" + return None + + if command == "getopts": + positional = [arg for arg in args if not arg.startswith("-")] + if len(positional) >= 2 and _dynamic_name(positional[1]): + return f"getopts writes a dynamically constructed variable target: {positional[1]}" + return None + + if command == "env": + for arg in args: + if arg.startswith("-"): + continue + if "=" in arg: + name = arg.split("=", 1)[0] + if _dynamic_name(name): + return f"env constructs a dynamic environment variable name: {arg}" + continue + if _dynamic_name(arg): + return f"env has an unresolved dynamic environment/command operand: {arg}" + break + return None + + +def _shell_scripts(path: str, text: str) -> list[str]: + if _is_yaml_authority(path): + lines = text.splitlines() + return core._run_scripts(lines, -1, len(lines)) + return [text] + + +def _is_shell_boundary(token: str) -> bool: + return bool(token) and all(character in SHELL_BOUNDARY_CHARS for character in token) + + +def _shell_command_tokens(script: str) -> tuple[list[list[str]], str | None]: + """Tokenize shell commands without splitting separators that occur inside quotes.""" + command_text = core._without_heredoc_bodies(script) + joined = re.sub(r"\\[ \t]*\r?\n[ \t]*", " ", command_text) + lexer = shlex.shlex(joined, posix=True, punctuation_chars=";&|\n") + lexer.whitespace = " \t\r" + lexer.whitespace_split = True + lexer.commenters = "#" + + commands: list[list[str]] = [] + current: list[str] = [] + try: + for token in lexer: + if _is_shell_boundary(token): + if current: + commands.append(current) + current = [] + continue + current.append(token) + except ValueError as error: + return [], str(error) + if current: + commands.append(current) + return commands, None + + +def _dynamic_variable_write_findings(path: str, text: str) -> list[dict[str, str]]: + findings: list[dict[str, str]] = [] + for script in _shell_scripts(path, text): + commands, parse_error = _shell_command_tokens(script) + if parse_error is not None: + if "$" in script and any( + name in script + for name in ( + "export", + "declare", + "local", + "readonly", + "typeset", + "printf", + "read", + "mapfile", + "readarray", + "getopts", + "unset", + "env", + ) + ): + findings.append( + { + "channel": "", + "code": "unsupported_dynamic_variable_write", + "detail": f"cannot safely parse dynamic variable-writing shell script: {parse_error}", + "path": path, + } + ) + continue + for tokens in commands: + detail = _dynamic_writer_detail(tokens) + if detail is not None: + findings.append( + { + "channel": "", + "code": "unsupported_dynamic_variable_write", + "detail": detail, + "path": path, + } + ) + return findings + + +def _authority_findings(path: str, text: str) -> list[dict[str, str]]: + findings: list[dict[str, str]] = [] + for matched in CHANNEL_NAME_RE.finditer(text): + channel = matched.group("channel") + findings.append( + { + "channel": channel, + "code": "unsupported_github_environment_channel", + "detail": ( + f"{channel} can mutate later-step environment/command resolution and " + "is outside the AF-01 constrained shell authority" + ), + "path": path, + } + ) + for matched in SHELL_STARTUP_NAME_RE.finditer(text): + startup = matched.group("startup") + findings.append( + { + "channel": startup, + "code": "unsupported_shell_startup_environment", + "detail": ( + f"{startup} can change shell startup-file authority before visible run commands " + "and is outside the AF-01 constrained shell authority" + ), + "path": path, + } + ) + if INDIRECT_PARAMETER_RE.search(text) is not None: + findings.append( + { + "channel": "", + "code": "unsupported_indirect_parameter_expansion", + "detail": ( + "indirect shell parameter expansion can resolve forbidden environment authority " + "and is outside AF-01 authority" + ), + "path": path, + } + ) + if GITHUB_PREFIX_FRAGMENT_RE.search(text) is not None: + findings.append( + { + "channel": "", + "code": "unsupported_github_environment_name_fragment", + "detail": ( + "standalone GITHUB_ name fragments can construct a forbidden GitHub " + "environment channel and are outside AF-01 authority" + ), + "path": path, + } + ) + if SHELL_STARTUP_FRAGMENT_RE.search(text) is not None: + findings.append( + { + "channel": "", + "code": "unsupported_shell_startup_name_fragment", + "detail": ( + "shell-startup variable name fragments can construct hidden startup-file " + "authority and are outside AF-01 authority" + ), + "path": path, + } + ) + findings.extend(_dynamic_variable_write_findings(path, text)) + return findings + + +def audit_repository_environment_channels( + root: Path, tracked_files: Iterable[str] +) -> dict[str, object]: + """Reject cross-step and shell-startup environment authority in tracked shell surfaces.""" + findings: list[dict[str, str]] = [] + for path in sorted(set(tracked_files)): + text, read_error = _read_authority_text(root, path) + if read_error is not None: + findings.append( + { + "channel": "", + "code": "unreadable_environment_channel_authority", + "detail": read_error, + "path": path, + } + ) + continue + if text is None: + continue + findings.extend(_authority_findings(path, text)) + findings.sort(key=lambda item: (item["path"], item["channel"], item["code"], item["detail"])) + return {"findings": findings, "ok": not findings, "schema": 1} + + +def main() -> int: + root = Path(__file__).resolve().parents[2] + try: + tracked = _tracked_files(root) + result = audit_repository_environment_channels(root, tracked) + except RuntimeError as error: + result = { + "findings": [ + { + "channel": "", + "code": "environment_channel_inventory_failed", + "detail": str(error), + "path": "", + } + ], + "ok": False, + "schema": 1, + } + print(json.dumps(result, indent=2, sort_keys=True)) + return 0 if result["ok"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/scripts/audit_workflow_trust_surface.py b/.github/scripts/audit_workflow_trust_surface.py new file mode 100644 index 00000000..be52532a --- /dev/null +++ b/.github/scripts/audit_workflow_trust_surface.py @@ -0,0 +1,720 @@ +#!/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 recursively referenced tracked 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 + +HEREDOC_OPERATOR_RE = re.compile( + r"<<-?\s*[\"']?[A-Za-z_][A-Za-z0-9_]*[\"']?" +) +REDIRECTION_TOKEN_RE = re.compile(r"^(?:\d*)?(?:>>?|<|>&|<&).+$") +DOUBLE_BRACKET_RE = re.compile(r"\[\[(?:(?!\]\]).)*\]\]", re.DOTALL) +ACTION_LOCAL_SCRIPT_RE = re.compile( + r"^\$(?:GITHUB_ACTION_PATH|\{GITHUB_ACTION_PATH\})/" + r"(?P[A-Za-z0-9_.-]+(?:/[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.*)$") +PATH_ASSIGNMENT_RE = re.compile(r"^PATH(?:\+)?=") +ASSIGNMENT_BUILTINS = frozenset({"declare", "export", "local", "readonly", "typeset"}) +VARIABLE_WRITE_BUILTINS = frozenset({"getopts", "mapfile", "read", "readarray"}) +RESOLUTION_MUTATION_BUILTINS = frozenset({"alias", "enable", "hash", "unalias"}) +ACTION_SOURCE_BUILTINS = frozenset({".", "source"}) +EXECUTION_WRAPPERS = frozenset( + {"command", "env", "exec", "nice", "nohup", "stdbuf", "sudo", "timeout"} +) + + +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 _basename(token: str) -> str: + return token.rsplit("/", 1)[-1] + + +def _raw_executable_index(tokens: list[str]) -> int | None: + index = 0 + while index < len(tokens) and core.SHELL_ASSIGNMENT_RE.fullmatch(tokens[index]): + index += 1 + while index < len(tokens) and tokens[index] in core.SHELL_CONTROL_WORDS: + index += 1 + return index if index < len(tokens) else None + + +def _wrapper_hides_cargo(tokens: list[str]) -> bool: + """Detect Cargo behind an execution wrapper the core constrained normalizer did not resolve.""" + raw_index = _raw_executable_index(tokens) + if raw_index is None or _basename(tokens[raw_index]) not in EXECUTION_WRAPPERS: + return False + return any(_basename(token) == "cargo" for token in tokens[raw_index + 1 :]) + + +def _only_redirections(tokens: list[str]) -> bool: + """Allow fixed shell redirections after a Cargo information flag.""" + return all(REDIRECTION_TOKEN_RE.fullmatch(token) is not None for token in tokens) + + +def _heredoc_prefix_executes_shell(prefix: str) -> bool: + """Recognize shell authority before a heredoc and fail closed on execution wrappers.""" + try: + tokens = shlex.split(prefix, comments=True, posix=True) + except ValueError: + return bool( + re.search( + r"(?:^|\s|/)(?:bash|dash|ksh|sh|zsh)(?:\s|$)", + prefix, + ) + ) + if not tokens: + return False + + # Wrapper option grammars have operand-bearing forms (`env -u NAME`, `env -S STRING`, + # `timeout 5 ...`, and others). A heredoc attached through a wrapper is executable input + # authority that this constrained parser intentionally does not normalize. Reject it rather + # than trusting a partially resolved command token. + raw_index = _raw_executable_index(tokens) + if raw_index is not None and _basename(tokens[raw_index]) in EXECUTION_WRAPPERS: + return True + + command_index = core._command_token_index(tokens) + if command_index is not None and command_index < len(tokens): + command = tokens[command_index] + if _basename(command) in core.SHELL_INTERPRETERS: + return True + if "$" in command or "`" in command: + return True + return False + + +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 + matched = HEREDOC_OPERATOR_RE.search(stripped) + if matched is None: + continue + prefix = stripped[: matched.start()] + if _heredoc_prefix_executes_shell(prefix): + findings.append( + _finding( + "unsupported_shell_heredoc", + path, + scope, + f"shell-interpreter heredoc is executable authority and is not supported: {stripped}", + ) + ) + return findings + + +def _mask_double_bracket_tests( + path: str, scope: str, script: str +) -> tuple[str, list[core.Finding]]: + """Mask non-executable [[ expressions while rejecting dynamic command substitution inside them.""" + findings: list[core.Finding] = [] + + def replace(matched: re.Match[str]) -> str: + expression = matched.group(0) + if "$(" in expression or "`" in expression: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"command substitution inside [[ ... ]] is not statically auditable: {expression}", + ) + ) + return "[[ true ]]" + + return DOUBLE_BRACKET_RE.sub(replace, script), findings + + +def _assignment_class(value: str) -> str: + """Classify an executable assignment as Cargo, statically non-Cargo, or unknown.""" + candidate = value.strip().strip("\"'") + if not candidate: + return "unknown" + basename = candidate.rsplit("/", 1)[-1] + if basename == "cargo": + return "cargo" + if "$(" in candidate or "`" in candidate: + return "unknown" + if basename and "$" not in basename: + return "non_cargo" + return "unknown" + + +def _record_assignment_tokens(tokens: list[str], states: dict[str, str]) -> None: + """Record simple shell assignments, including export/readonly/local/declare/typeset forms.""" + if not tokens: + return + candidates: list[str] = [] + if tokens[0] in ASSIGNMENT_BUILTINS: + for token in tokens[1:]: + if token.startswith("-"): + continue + candidates.append(token) + else: + for token in tokens: + if ASSIGNMENT_RE.fullmatch(token): + candidates.append(token) + else: + break + for token in candidates: + matched = ASSIGNMENT_RE.fullmatch(token) + if matched: + states[matched.group("name")] = _assignment_class(matched.group("value")) + + +def _direct_cargo_findings( + path: str, scope: str, script: str, locked_subcommands: set[str] +) -> list[core.Finding]: + """Audit direct Cargo and fail closed on executable indirection that may resolve to Cargo.""" + findings = _shell_heredoc_findings(path, scope, script) + command_script, expression_findings = _mask_double_bracket_tests(path, scope, script) + findings.extend(expression_findings) + variable_states: dict[str, str] = {} + + for segment in core._logical_shell_segments(command_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 + + _record_assignment_tokens(tokens, variable_states) + + command_index = core._command_token_index(tokens) + if command_index is None or command_index >= len(tokens): + if _wrapper_hides_cargo(tokens): + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"execution wrapper hides Cargo from static command normalization: {segment}", + ) + ) + continue + command = tokens[command_index] + + variable = VARIABLE_COMMAND_RE.fullmatch(command) + if variable: + name = variable.group(1) or variable.group(2) + state = variable_states.get(name, "unknown") + if state != "non_cargo": + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"variable-expanded executable is not proven non-Cargo ({name}={state}): {segment}", + ) + ) + continue + + if command.startswith("$(") or command.startswith("`"): + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"command-substituted executable can resolve to Cargo: {segment}", + ) + ) + continue + + if "$" in command or "`" in command: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"dynamic executable path can resolve to Cargo: {segment}", + ) + ) + continue + + command_basename = _basename(command) + if command_basename in core.DYNAMIC_COMMAND_BUILTINS: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"dynamic shell execution can hide Cargo: {segment}", + ) + ) + continue + if command_basename in core.SHELL_INTERPRETERS and "-c" in tokens[command_index + 1 :]: + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"nested shell execution can hide Cargo: {segment}", + ) + ) + continue + + is_direct_cargo = command == "cargo" or command_basename == "cargo" + if not is_direct_cargo: + if _wrapper_hides_cargo(tokens): + findings.append( + _finding( + "unsupported_cargo_indirect", + path, + scope, + f"execution wrapper hides Cargo from static command normalization: {segment}", + ) + ) + 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: + trailing = tokens[subcommand_index + 1 :] + if not trailing or _only_redirections(trailing): + continue + findings.append( + _finding( + "unsupported_cargo_syntax", + path, + scope, + f"Cargo information flag has unsupported trailing syntax: {segment}", + ) + ) + continue + if subcommand.startswith("-"): + findings.append( + _finding( + "unsupported_cargo_syntax", + path, + scope, + f"Cargo global-option syntax requires explicit auditor support: {segment}", + ) + ) + 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 _exact_action_local_target(token: str) -> str | None: + matched = ACTION_LOCAL_SCRIPT_RE.fullmatch(token) + if matched is None: + return None + relative = matched.group("path") + if relative.startswith("/") or ".." in Path(relative).parts: + return None + return relative + + +def _normalized_action_command_is_supported(command: str) -> bool: + return ( + _exact_action_local_target(command) is not None + or command in ACTION_SOURCE_BUILTINS + or _basename(command) in core.SHELL_INTERPRETERS + ) + + +def _unresolved_wrapper_can_delegate_action_script( + tokens: list[str], command_index: int | None +) -> bool: + """Reject wrapper option forms when the core resolver cannot prove the delegated executable.""" + raw_index = _raw_executable_index(tokens) + if raw_index is None or _basename(tokens[raw_index]) not in EXECUTION_WRAPPERS: + return False + if command_index is not None and command_index < len(tokens): + if _normalized_action_command_is_supported(tokens[command_index]): + return False + trailing = tokens[raw_index + 1 :] + return any("GITHUB_ACTION_PATH" in token for token in trailing) or any( + _basename(token) in core.SHELL_INTERPRETERS for token in trailing + ) + + +def _direct_path_executable_is_unsupported(command: str) -> bool: + """Reject direct path execution unless it was already recognized as Action-root authority.""" + return "/" in command + + +def _nameref_builtin_is_unsupported(command: str, args: list[str]) -> bool: + """Reject shell namerefs because they can create an indirect writer for PATH.""" + if command not in ASSIGNMENT_BUILTINS: + return False + for arg in args: + if arg == "--nameref": + return True + if arg.startswith("-") and not arg.startswith("--") and "n" in arg[1:]: + return True + return False + + +def _builtin_writes_path(command: str, args: list[str]) -> bool: + """Detect builtins that can write a caller-selected variable and therefore mutate PATH.""" + if command in VARIABLE_WRITE_BUILTINS: + return any(arg == "PATH" for arg in args) + if command == "printf": + for index, arg in enumerate(args): + if arg == "-v" and index + 1 < len(args) and args[index + 1] == "PATH": + return True + if arg == "-vPATH": + return True + return False + + +def _resolution_mutation_is_unsupported(tokens: list[str]) -> bool: + """Reject command-resolution mutation because bare names must retain runner-owned resolution.""" + raw_index = _raw_executable_index(tokens) + if raw_index is None or raw_index >= len(tokens): + return False + raw_command = _basename(tokens[raw_index]) + raw_args = tokens[raw_index + 1 :] + if raw_command in RESOLUTION_MUTATION_BUILTINS: + return True + if raw_command == "builtin" and any( + _basename(arg) in RESOLUTION_MUTATION_BUILTINS for arg in raw_args if not arg.startswith("-") + ): + return True + command_index = core._command_token_index(tokens) + if command_index is not None and command_index < len(tokens): + if _basename(tokens[command_index]) in RESOLUTION_MUTATION_BUILTINS: + return True + return False + + +def _path_search_mutation_is_unsupported(tokens: list[str]) -> bool: + """Reject PATH/resolution mutation that can redirect a bare executable outside audit authority.""" + if _resolution_mutation_is_unsupported(tokens): + return True + + index = 0 + while index < len(tokens) and tokens[index] in core.SHELL_CONTROL_WORDS: + index += 1 + while index < len(tokens): + token = tokens[index] + if PATH_ASSIGNMENT_RE.match(token): + return True + if core.SHELL_ASSIGNMENT_RE.fullmatch(token) or ASSIGNMENT_RE.fullmatch(token): + index += 1 + continue + break + + raw_index = _raw_executable_index(tokens) + if raw_index is None or raw_index >= len(tokens): + return False + command = _basename(tokens[raw_index]) + args = tokens[raw_index + 1 :] + if _nameref_builtin_is_unsupported(command, args): + return True + if _builtin_writes_path(command, args): + return True + if command in ASSIGNMENT_BUILTINS or command == "env": + return any(PATH_ASSIGNMENT_RE.match(arg) for arg in args) + if command == "unset": + return any(arg == "PATH" for arg in args if not arg.startswith("-")) + return False + + +def _action_local_targets(script: str) -> tuple[list[str], list[str]]: + """Return exact static GITHUB_ACTION_PATH shell targets and unsupported delegation.""" + 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 + if _path_search_mutation_is_unsupported(tokens): + unsupported.append(segment) + continue + command_index = core._command_token_index(tokens) + if _unresolved_wrapper_can_delegate_action_script(tokens, command_index): + unsupported.append(segment) + continue + if command_index is None or command_index >= len(tokens): + continue + + command = tokens[command_index] + command_basename = _basename(command) + args = tokens[command_index + 1 :] + + direct_target = _exact_action_local_target(command) + if direct_target is not None: + if args: + unsupported.append(segment) + else: + targets.append(direct_target) + continue + if "GITHUB_ACTION_PATH" in command: + unsupported.append(segment) + continue + + if command in ACTION_SOURCE_BUILTINS: + if len(args) != 1: + unsupported.append(segment) + continue + source_target = _exact_action_local_target(args[0]) + if source_target is None: + unsupported.append(segment) + else: + targets.append(source_target) + continue + + if command_basename not in core.SHELL_INTERPRETERS: + if _direct_path_executable_is_unsupported(command): + unsupported.append(segment) + continue + if "-c" in args: + unsupported.append(segment) + continue + if HEREDOC_OPERATOR_RE.search(segment): + continue + script_positions = [index for index, arg in enumerate(args) if not arg.startswith("-")] + if not script_positions: + unsupported.append(segment) + continue + script_position = script_positions[0] + script_arg = args[script_position] + target = _exact_action_local_target(script_arg) + if target is None: + unsupported.append(segment) + continue + if any(not arg.startswith("-") for arg in args[script_position + 1 :]): + unsupported.append(segment) + continue + targets.append(target) + return sorted(set(targets)), sorted(set(unsupported)) + + +def _unsupported_action_script_findings( + path: str, scope: str, scripts: Iterable[str] +) -> list[core.Finding]: + findings: list[core.Finding] = [] + for script in scripts: + _, unsupported = _action_local_targets(script) + for segment in unsupported: + findings.append( + _finding( + "unsupported_action_script", + path, + scope, + f"Action shell source is dynamic or not statically repository-owned: {segment}", + ) + ) + return findings + + +def audit_action_text( + path: str, + text: str, + locked_subcommands: set[str], +) -> list[core.Finding]: + lines = text.splitlines() + scripts = _scripts(lines) + findings: list[core.Finding] = [] + for script in scripts: + findings.extend(_direct_cargo_findings(path, "composite-action", script, locked_subcommands)) + findings.extend(_unsupported_action_script_findings(path, "composite-action", scripts)) + return findings + + +def _action_target_path(action_path: str, relative: str) -> str | None: + action_dir = Path(action_path).parent + target_path = action_dir / relative + if target_path.is_absolute() or ".." in target_path.parts: + return None + target = target_path.as_posix() + return target[2:] if target.startswith("./") else target + + +def _audit_local_action_script( + root: Path, + action_path: str, + relative: str, + tracked: set[str], + locked_subcommands: set[str], + seen: set[str], +) -> list[core.Finding]: + target = _action_target_path(action_path, relative) + if target is None: + return [ + _finding( + "unsupported_action_script", + action_path, + "composite-action", + f"Action shell source escapes the Action directory: {relative}", + ) + ] + if target in seen: + return [] + seen.add(target) + 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}", + ) + ] + + findings = _direct_cargo_findings(target, "action-script", source, locked_subcommands) + targets, unsupported = _action_local_targets(source) + for segment in unsupported: + findings.append( + _finding( + "unsupported_action_script", + target, + "action-script", + f"Action shell source delegates dynamically or outside GITHUB_ACTION_PATH: {segment}", + ) + ) + for nested in targets: + findings.extend( + _audit_local_action_script( + root, + action_path, + nested, + tracked, + locked_subcommands, + seen, + ) + ) + return findings + + +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") + action_scripts = _scripts(text.splitlines()) + findings.extend(audit_action_text(action_path, text, locked_subcommands)) + seen: set[str] = set() + for script in action_scripts: + targets, _ = _action_local_targets(script) + for relative in targets: + findings.extend( + _audit_local_action_script( + root, + action_path, + relative, + tracked, + locked_subcommands, + seen, + ) + ) + + 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()) diff --git a/.github/scripts/test-cf08-action-runner.sh b/.github/scripts/test-cf08-action-runner.sh index fcc2bb49..de94e478 100644 --- a/.github/scripts/test-cf08-action-runner.sh +++ b/.github/scripts/test-cf08-action-runner.sh @@ -4,7 +4,9 @@ set -euo pipefail temp="$(mktemp -d)" trap 'rm -rf "$temp"' EXIT -fake="$temp/fake-commandf" +fake_target="$temp/fake-target" +mkdir -p "$fake_target/debug" +fake="$fake_target/debug/commandf" cat > "$fake" <<'FAKE' #!/usr/bin/env bash set -euo pipefail @@ -62,6 +64,7 @@ run_case() { : > "$render_log" set +e + CARGO_TARGET_DIR="$fake_target" \ GITHUB_OUTPUT="$output_file" \ COMMANDF_RESOLVED_REPORT_PATH="$report_path" \ COMMANDF_PACKAGE="$package" \ @@ -75,7 +78,7 @@ run_case() { FAKE_RENDER_CODE="$render_code" \ FAKE_ARGV_LOG="$argv_log" \ FAKE_RENDER_LOG="$render_log" \ - bash scripts/github-action-run.sh "$fake" > "$case_dir/stdout" 2> "$case_dir/stderr" + bash scripts/github-action-run.sh > "$case_dir/stdout" 2> "$case_dir/stderr" actual=$? set -e diff --git a/.github/scripts/test-cf09-action-source-map.sh b/.github/scripts/test-cf09-action-source-map.sh index b8916cd8..e1d3365c 100644 --- a/.github/scripts/test-cf09-action-source-map.sh +++ b/.github/scripts/test-cf09-action-source-map.sh @@ -4,7 +4,9 @@ set -euo pipefail temp="$(mktemp -d)" trap 'rm -rf "$temp"' EXIT -fake="$temp/fake-commandf" +fake_target="$temp/fake-target" +mkdir -p "$fake_target/debug" +fake="$fake_target/debug/commandf" cat > "$fake" <<'FAKE' #!/usr/bin/env bash set -euo pipefail @@ -83,6 +85,7 @@ run_case() { : > "$fsh_index" set +e + CARGO_TARGET_DIR="$fake_target" \ GITHUB_OUTPUT="$output_file" \ GITHUB_WORKSPACE="$case_dir/workspace" \ COMMANDF_RESOLVED_REPORT_PATH="$report_path" \ @@ -101,7 +104,7 @@ run_case() { FAKE_RENDER_CODE="$render_code" \ FAKE_SOURCE_MAP_ARGV_LOG="$source_map_argv" \ FAKE_RENDER_ARGV_LOG="$render_argv" \ - bash scripts/github-action-run.sh "$fake" > "$case_dir/stdout" 2> "$case_dir/stderr" + bash scripts/github-action-run.sh > "$case_dir/stdout" 2> "$case_dir/stderr" actual=$? set -e diff --git a/.github/scripts/test_audit_workflow_trust.py b/.github/scripts/test_audit_workflow_trust.py new file mode 100644 index 00000000..c342b877 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust.py @@ -0,0 +1,421 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import copy +import importlib.util +import sys +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) +sys.modules[SPEC.name] = AUDIT +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: + 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": rules, + "rationales": { + key: f"Test rationale for {key} that is deliberately substantive." + for key in rules + }, + "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_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_trust_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_trust_syntax", 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_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 +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", "") + 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_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) + result = self.run_repo(workflow) + 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_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" + ) + 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 = ( + " run: |\n" + f" cargo test {continuation}\n" + " --locked --workspace" + ) + workflow = valid_workflow().replace( + " run: cargo test --locked --workspace", replacement + ) + 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", + "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_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)) + 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)) + + 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"]) + self.assertIn("invalid_policy", self.codes(result)) + self.assertEqual(result["workflows"], []) + self.assertEqual(result["action_metadata"], []) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/test_audit_workflow_trust_environment_channels.py b/.github/scripts/test_audit_workflow_trust_environment_channels.py new file mode 100644 index 00000000..48a2e230 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust_environment_channels.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import importlib.util +import sys +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("audit_workflow_trust_environment_channels.py") +SPEC = importlib.util.spec_from_file_location("audit_workflow_trust_environment_channels_target", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +CHANNELS = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = CHANNELS +SPEC.loader.exec_module(CHANNELS) + + +def findings_for(files: dict[str, str]) -> list[dict[str, str]]: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for path, text in files.items(): + candidate = root / path + candidate.parent.mkdir(parents=True, exist_ok=True) + candidate.write_text(text, encoding="utf-8") + result = CHANNELS.audit_repository_environment_channels(root, files.keys()) + return result["findings"] + + +def codes(findings: list[dict[str, str]]) -> list[str]: + return [item["code"] for item in findings] + + +class EnvironmentChannelAuditTests(unittest.TestCase): + def test_composite_action_github_path_write_fails_closed(self) -> None: + findings = findings_for( + { + "action.yml": """name: fixture\ndescription: fixture\nruns:\n using: composite\n steps:\n - shell: bash\n run: echo \"$GITHUB_ACTION_PATH/scripts\" >> \"$GITHUB_PATH\"\n""" + } + ) + self.assertIn("GITHUB_PATH", [item["channel"] for item in findings]) + + def test_workflow_github_env_path_write_fails_closed(self) -> None: + findings = findings_for( + { + ".github/workflows/test.yml": """name: test\non: pull_request\njobs:\n test:\n runs-on: ubuntu-24.04\n steps:\n - run: printf 'PATH=%s\\n' \"$GITHUB_ACTION_PATH/scripts:$PATH\" >> \"${GITHUB_ENV}\"\n""" + } + ) + self.assertIn("GITHUB_ENV", [item["channel"] for item in findings]) + + def test_channel_name_then_indirect_parameter_expansion_fails_closed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env bash\nchannel=GITHUB_PATH\nprintf '%s\\n' /tmp/bin >> \"${!channel}\"\n" + } + ) + self.assertIn("GITHUB_PATH", [item["channel"] for item in findings]) + self.assertIn("unsupported_indirect_parameter_expansion", codes(findings)) + + def test_indirect_parameter_expansion_is_always_unsupported(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env bash\nchannel=SAFE_CHANNEL\nprintf '%s\\n' value >> \"${!channel}\"\n" + } + ) + self.assertIn("unsupported_indirect_parameter_expansion", codes(findings)) + + def test_split_github_name_fragment_fails_closed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env bash\ntarget=\"$GITHUB_\"PATH\nprintf '%s\\n' /tmp/bin >> \"$target\"\n" + } + ) + self.assertIn("unsupported_github_environment_name_fragment", codes(findings)) + + def test_literal_github_prefix_fragment_fails_closed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env bash\nprefix=GITHUB_\nchannel=\"${prefix}PATH\"\nprintf '%s\\n' /tmp/bin >> \"${!channel}\"\n" + } + ) + self.assertIn("unsupported_github_environment_name_fragment", codes(findings)) + self.assertIn("unsupported_indirect_parameter_expansion", codes(findings)) + + def _startup_action_findings(self, startup: str, delegated_script: str) -> list[dict[str, str]]: + return findings_for( + { + "action.yml": f"""name: fixture\ndescription: fixture\nruns:\n using: composite\n steps:\n - shell: bash\n env:\n {startup}: $GITHUB_ACTION_PATH/scripts/build.sh\n run: echo ok\n""", + "scripts/build.sh": delegated_script, + } + ) + + def _dynamic_startup_action_findings(self, delegated_script: str) -> list[dict[str, str]]: + return findings_for( + { + "action.yml": """name: fixture\ndescription: fixture\nruns:\n using: composite\n steps:\n - shell: bash\n run: |\n prefix=BASH\n export \"${prefix}_ENV=$GITHUB_ACTION_PATH/scripts/hidden.sh\"\n bash \"$GITHUB_ACTION_PATH/scripts/entry.sh\"\n""", + "scripts/hidden.sh": delegated_script, + "scripts/entry.sh": "#!/usr/bin/env bash\necho entry\n", + } + ) + + def test_bash_env_rejects_locked_hidden_action_script(self) -> None: + findings = self._startup_action_findings( + "BASH_ENV", "#!/usr/bin/env bash\ncargo test --locked --workspace\n" + ) + self.assertIn("unsupported_shell_startup_environment", codes(findings)) + self.assertIn("BASH_ENV", [item["channel"] for item in findings]) + + def test_bash_env_rejects_unlocked_hidden_action_script(self) -> None: + findings = self._startup_action_findings( + "BASH_ENV", "#!/usr/bin/env bash\ncargo test --workspace\n" + ) + self.assertIn("unsupported_shell_startup_environment", codes(findings)) + self.assertIn("BASH_ENV", [item["channel"] for item in findings]) + + def test_dynamic_bash_env_export_rejects_locked_hidden_action_script(self) -> None: + findings = self._dynamic_startup_action_findings( + "#!/usr/bin/env bash\ncargo test --locked --workspace\n" + ) + self.assertIn("unsupported_dynamic_variable_write", codes(findings)) + + def test_dynamic_bash_env_export_rejects_unlocked_hidden_action_script(self) -> None: + findings = self._dynamic_startup_action_findings( + "#!/usr/bin/env bash\ncargo test --workspace\n" + ) + self.assertIn("unsupported_dynamic_variable_write", codes(findings)) + + def test_dynamic_posix_env_export_fails_closed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env ksh\nprefix=E\nexport \"${prefix}NV=/tmp/bootstrap.ksh\"\nprint ok\n" + } + ) + self.assertIn("unsupported_dynamic_variable_write", codes(findings)) + + def test_dynamic_zdotdir_export_fails_closed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env zsh\nprefix=ZDOT\nexport \"${prefix}DIR=/tmp/action-dotfiles\"\nprint ok\n" + } + ) + self.assertIn("unsupported_dynamic_variable_write", codes(findings)) + + def test_dynamic_printf_v_startup_target_fails_closed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env bash\nprefix=BASH\nprintf -v \"${prefix}_ENV\" '%s' /tmp/bootstrap.sh\n" + } + ) + self.assertIn("unsupported_dynamic_variable_write", codes(findings)) + + def test_dynamic_env_assignment_fails_closed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env bash\nprefix=BASH\nenv \"${prefix}_ENV=/tmp/bootstrap.sh\" bash -c 'echo ok'\n" + } + ) + self.assertIn("unsupported_dynamic_variable_write", codes(findings)) + + def test_static_variable_name_with_dynamic_value_remains_allowed(self) -> None: + findings = findings_for( + { + "scripts/build.sh": "#!/usr/bin/env bash\nreport=/tmp/report.json\nexport REPORT_PATH=\"$report\"\nprintf -v REPORT_COPY '%s' \"$report\"\n" + } + ) + self.assertEqual([], findings) + + def test_workflow_bash_env_fails_closed(self) -> None: + findings = findings_for( + { + ".github/workflows/test.yml": """name: test\non: pull_request\njobs:\n test:\n runs-on: ubuntu-24.04\n steps:\n - shell: bash\n env:\n BASH_ENV: scripts/bootstrap.sh\n run: echo ok\n""" + } + ) + self.assertIn("BASH_ENV", [item["channel"] for item in findings]) + + def test_posix_ksh_env_startup_authority_fails_closed(self) -> None: + findings = findings_for( + {"scripts/build.sh": "#!/usr/bin/env ksh\nENV=/tmp/bootstrap.ksh\nprint ok\n"} + ) + self.assertIn("ENV", [item["channel"] for item in findings]) + + def test_zsh_zdotdir_startup_authority_fails_closed(self) -> None: + findings = findings_for( + {"scripts/build.sh": "#!/usr/bin/env zsh\nZDOTDIR=/tmp/action-dotfiles\nprint ok\n"} + ) + self.assertIn("ZDOTDIR", [item["channel"] for item in findings]) + + def test_shell_startup_name_fragments_fail_closed(self) -> None: + bash_findings = findings_for( + {"scripts/build.sh": "#!/usr/bin/env bash\nname=\"BASH_\"ENV\nprintf '%s\\n' \"$name\"\n"} + ) + zsh_findings = findings_for( + {"scripts/build.sh": "#!/usr/bin/env zsh\nname=\"ZDOT\"DIR\nprint -r -- \"$name\"\n"} + ) + self.assertIn("unsupported_shell_startup_name_fragment", codes(bash_findings)) + self.assertIn("unsupported_shell_startup_name_fragment", codes(zsh_findings)) + + def test_extensionless_shell_script_is_covered(self) -> None: + findings = findings_for( + { + "scripts/build": "#!/usr/bin/env bash\nprintf '%s\\n' /tmp/bin >> \"${GITHUB_PATH}\"\n" + } + ) + self.assertIn("GITHUB_PATH", [item["channel"] for item in findings]) + + def test_github_output_remains_allowed(self) -> None: + findings = findings_for( + { + "scripts/github-action.sh": "#!/usr/bin/env bash\nprintf 'passed=true\\n' >> \"$GITHUB_OUTPUT\"\n" + } + ) + self.assertEqual([], findings) + + def test_lowercase_env_mapping_and_similar_names_are_allowed(self) -> None: + findings = findings_for( + { + "action.yml": """name: fixture\ndescription: fixture\nruns:\n using: composite\n steps:\n - shell: bash\n env:\n MY_ENV: safe\n BASH_ENVIRONMENT: safe\n run: echo ok\n""" + } + ) + self.assertEqual([], findings) + + def test_human_diagnostic_without_special_variable_names_is_allowed(self) -> None: + findings = findings_for( + { + "scripts/diagnostic.sh": "#!/usr/bin/env bash\nprintf '%s\\n' 'GitHub path and environment command files are forbidden'\n" + } + ) + self.assertEqual([], findings) + + def test_repeat_output_is_deterministic(self) -> None: + files = { + "action.yml": """name: fixture\ndescription: fixture\nruns:\n using: composite\n steps:\n - shell: bash\n run: echo /tmp/bin >> \"$GITHUB_PATH\"\n""", + "scripts/build.sh": "#!/usr/bin/env bash\nchannel=GITHUB_ENV\nprintf 'PATH=/tmp/bin\\n' >> \"${!channel}\"\n", + } + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + for path, text in files.items(): + candidate = root / path + candidate.parent.mkdir(parents=True, exist_ok=True) + candidate.write_text(text, encoding="utf-8") + first = CHANNELS.audit_repository_environment_channels(root, files.keys()) + second = CHANNELS.audit_repository_environment_channels(root, reversed(list(files.keys()))) + self.assertEqual(first, second) + + def test_live_repository_has_no_cross_step_or_startup_environment_authority(self) -> None: + root = Path(__file__).resolve().parents[2] + tracked = CHANNELS._tracked_files(root) + result = CHANNELS.audit_repository_environment_channels(root, tracked) + self.assertTrue(result["ok"], result["findings"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/test_audit_workflow_trust_environment_channels_precision.py b/.github/scripts/test_audit_workflow_trust_environment_channels_precision.py new file mode 100644 index 00000000..2753dea9 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust_environment_channels_precision.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import importlib.util +import sys +import tempfile +import unittest +from pathlib import Path + +MODULE_PATH = Path(__file__).with_name("audit_workflow_trust_environment_channels.py") +SPEC = importlib.util.spec_from_file_location( + "audit_workflow_trust_environment_channels_precision_target", MODULE_PATH +) +assert SPEC is not None and SPEC.loader is not None +CHANNELS = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = CHANNELS +SPEC.loader.exec_module(CHANNELS) + + +def findings_for(script: str) -> list[dict[str, str]]: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + path = "scripts/build.sh" + candidate = root / path + candidate.parent.mkdir(parents=True, exist_ok=True) + candidate.write_text("#!/usr/bin/env bash\n" + script, encoding="utf-8") + result = CHANNELS.audit_repository_environment_channels(root, [path]) + return result["findings"] + + +class EnvironmentChannelPrecisionTests(unittest.TestCase): + def test_quoted_semicolon_in_static_assignment_is_data(self) -> None: + findings = findings_for( + 'local fsh_index="$case_dir/fsh index;literal.json"\n' + 'export REPORT_PATH="$fsh_index"\n' + ) + self.assertEqual([], findings) + + def test_semicolon_between_commands_still_exposes_dynamic_writer(self) -> None: + findings = findings_for( + 'prefix=BASH; export "${prefix}_ENV=/tmp/bootstrap.sh"; echo ok\n' + ) + self.assertIn( + "unsupported_dynamic_variable_write", + [item["code"] for item in findings], + ) + + +if __name__ == "__main__": + unittest.main() 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..6e359f05 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust_surface.py @@ -0,0 +1,428 @@ +#!/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 = "env bash <<'EOF'\ncargo test --workspace\nEOF" + findings = SURFACE._shell_heredoc_findings("wf.yml", "build", script) + self.assertIn("unsupported_shell_heredoc", codes(findings)) + + def test_env_option_wrapped_bash_heredoc_fails_closed(self) -> None: + script = "env -u UNUSED bash <<'EOF'\ncargo test --workspace\nEOF" + findings = SURFACE._shell_heredoc_findings("wf.yml", "build", script) + self.assertIn("unsupported_shell_heredoc", codes(findings)) + + def test_absolute_bash_heredoc_is_executable_authority(self) -> None: + script = "/bin/bash <<'EOF'\ncargo test --workspace\nEOF" + findings = SURFACE._shell_heredoc_findings("wf.yml", "build", script) + self.assertIn("unsupported_shell_heredoc", codes(findings)) + + def test_command_wrapped_absolute_shell_heredoc_is_executable_authority(self) -> None: + script = "command /bin/sh <<'EOF'\ncargo test --workspace\nEOF" + findings = SURFACE._shell_heredoc_findings("wf.yml", "build", script) + self.assertIn("unsupported_shell_heredoc", codes(findings)) + + def test_python_heredoc_remains_data(self) -> 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_double_bracket_boolean_expression_is_not_executable(self) -> None: + script = 'if [[ "$code" == "0" || "$code" == "2" ]]; then\n echo ok\nfi' + findings = SURFACE._direct_cargo_findings("script.sh", "action-script", script, LOCKED) + self.assertNotIn("unsupported_cargo_indirect", codes(findings)) + + def test_double_bracket_command_substitution_fails_closed(self) -> None: + script = 'if [[ "$(printf test)" == "test" ]]; then\n echo ok\nfi' + findings = SURFACE._direct_cargo_findings("script.sh", "action-script", script, LOCKED) + self.assertIn("unsupported_cargo_indirect", 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_cargo_config_global_option_fails_closed_unlocked(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("cargo --config net.offline=false test --workspace"), LOCKED + ) + self.assertIn("unsupported_cargo_syntax", codes(findings)) + + def test_action_cargo_config_global_option_fails_closed_even_when_locked(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action("cargo --config net.offline=false test --locked --workspace"), + LOCKED, + ) + self.assertIn("unsupported_cargo_syntax", codes(findings)) + + def test_action_cargo_color_global_option_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("cargo --color=always test --workspace"), LOCKED + ) + self.assertIn("unsupported_cargo_syntax", codes(findings)) + + def test_wrapped_cargo_outside_static_command_position_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("env -u UNUSED cargo test --workspace"), LOCKED + ) + self.assertIn("unsupported_cargo_indirect", codes(findings)) + + def test_exported_cargo_variable_executable_fails_closed(self) -> None: + script = 'export tool=cargo\n"$tool" test --workspace' + findings = SURFACE._direct_cargo_findings("action.yml", "composite-action", script, LOCKED) + self.assertIn("unsupported_cargo_indirect", codes(findings)) + + def test_readonly_cargo_variable_executable_fails_closed(self) -> None: + script = 'readonly tool=/usr/bin/cargo\n"$tool" build --workspace' + findings = SURFACE._direct_cargo_findings("action.yml", "composite-action", script, LOCKED) + self.assertIn("unsupported_cargo_indirect", codes(findings)) + + def test_unknown_variable_executable_fails_closed(self) -> None: + findings = SURFACE._direct_cargo_findings( + "action.yml", "composite-action", '"$tool" test --workspace', LOCKED + ) + self.assertIn("unsupported_cargo_indirect", codes(findings)) + + def test_dynamic_path_executable_fails_closed(self) -> None: + findings = SURFACE._direct_cargo_findings( + "action.yml", "composite-action", '"/usr/bin/$tool" test --workspace', LOCKED + ) + self.assertIn("unsupported_cargo_indirect", codes(findings)) + + def test_fixed_non_cargo_variable_executable_is_proven_safe(self) -> None: + script = 'binary="$CARGO_TARGET_DIR/debug/commandf"\n"$binary" check fixture' + findings = SURFACE._direct_cargo_findings("script.sh", "action-script", script, LOCKED) + self.assertNotIn("unsupported_cargo_indirect", 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_action_shell_c_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("bash -c 'cargo test --locked --workspace'"), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_action_relative_shell_source_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("bash scripts/build.sh"), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_action_dynamic_source_builtin_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action('source "$SCRIPT"'), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_action_relative_dot_source_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action(". scripts/build.sh"), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_action_root_script_suffix_expansion_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action('bash "$GITHUB_ACTION_PATH/scripts/build.sh$SUFFIX"'), + LOCKED, + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_action_root_script_prefix_expansion_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action('bash "$PREFIX$GITHUB_ACTION_PATH/scripts/build.sh"'), + 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 / "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"]) + + def test_absolute_shell_local_action_script_is_audited(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('/bin/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_direct_action_root_script_execution_is_audited(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('"$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_sourced_local_action_script_is_audited(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('source "$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_dot_sourced_local_locked_action_script_passes(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('. "$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"]) + + def test_delegated_cargo_global_option_fails_closed(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 --config net.offline=false 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( + "unsupported_cargo_syntax", [item["code"] for item in result["findings"]] + ) + + def test_nested_tracked_local_action_script_is_audited(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('bash "$GITHUB_ACTION_PATH/scripts/entry.sh"'), encoding="utf-8" + ) + (root / "scripts" / "entry.sh").write_text( + '#!/usr/bin/env bash\nexec bash "$GITHUB_ACTION_PATH/scripts/build.sh"\n', + 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/entry.sh", "scripts/build.sh"], + ) + self.assertFalse(result["ok"]) + self.assertIn("cargo_unlocked", [item["code"] for item in result["findings"]]) + + def test_nested_dynamic_local_action_script_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('bash "$GITHUB_ACTION_PATH/scripts/entry.sh"'), encoding="utf-8" + ) + (root / "scripts" / "entry.sh").write_text( + '#!/usr/bin/env bash\nexec bash "$NEXT_SCRIPT"\n', encoding="utf-8" + ) + result = SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/entry.sh"], + ) + self.assertFalse(result["ok"]) + self.assertIn( + "unsupported_action_script", [item["code"] for item in result["findings"]] + ) + + def test_recursive_action_script_cycle_is_bounded(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('bash "$GITHUB_ACTION_PATH/scripts/a.sh"'), encoding="utf-8" + ) + (root / "scripts" / "a.sh").write_text( + '#!/usr/bin/env bash\nexec bash "$GITHUB_ACTION_PATH/scripts/b.sh"\n', + encoding="utf-8", + ) + (root / "scripts" / "b.sh").write_text( + '#!/usr/bin/env bash\nexec bash "$GITHUB_ACTION_PATH/scripts/a.sh"\n', + encoding="utf-8", + ) + result = SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/a.sh", "scripts/b.sh"], + ) + self.assertTrue(result["ok"], result["findings"]) + + def test_recursive_source_cycle_is_bounded(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('source "$GITHUB_ACTION_PATH/scripts/a.sh"'), encoding="utf-8" + ) + (root / "scripts" / "a.sh").write_text( + '#!/usr/bin/env bash\nsource "$GITHUB_ACTION_PATH/scripts/b.sh"\n', + encoding="utf-8", + ) + (root / "scripts" / "b.sh").write_text( + '#!/usr/bin/env bash\n. "$GITHUB_ACTION_PATH/scripts/a.sh"\n', + encoding="utf-8", + ) + result = SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/a.sh", "scripts/b.sh"], + ) + self.assertTrue(result["ok"], result["findings"]) + + def test_live_repository_surface_is_clean(self) -> None: + root = Path(__file__).resolve().parents[2] + policy = json.loads( + (root / ".github" / "workflow-trust-policy.json").read_text(encoding="utf-8") + ) + result = SURFACE.audit_repository_surface(root, policy) + self.assertTrue(result["ok"], result["findings"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/.github/scripts/test_audit_workflow_trust_surface_precision.py b/.github/scripts/test_audit_workflow_trust_surface_precision.py new file mode 100644 index 00000000..7c0c3a36 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust_surface_precision.py @@ -0,0 +1,386 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import importlib.util +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_precision_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 SurfacePrecisionTests(unittest.TestCase): + def test_diagnostic_text_containing_cargo_is_not_executable_authority(self) -> None: + findings = SURFACE._direct_cargo_findings( + "scripts/github-action.sh", + "action-script", + 'emit_operational_failure "rustup and cargo are required to build the source-backed Action"', + LOCKED, + ) + self.assertNotIn("unsupported_cargo_indirect", codes(findings)) + + def test_toolchain_selected_cargo_version_with_fixed_redirections_is_allowed(self) -> None: + findings = SURFACE._direct_cargo_findings( + "scripts/github-action.sh", + "action-script", + "if ! cargo +1.97.1 --version >/dev/null 2>&1", + LOCKED, + ) + self.assertNotIn("unsupported_cargo_syntax", codes(findings)) + self.assertNotIn("cargo_unlocked", codes(findings)) + + def test_cargo_information_flag_with_non_redirection_tail_fails_closed(self) -> None: + findings = SURFACE._direct_cargo_findings( + "action.yml", "composite-action", "cargo --version test", LOCKED + ) + self.assertIn("unsupported_cargo_syntax", codes(findings)) + + def test_unresolved_env_wrapper_cannot_hide_cargo(self) -> None: + findings = SURFACE._direct_cargo_findings( + "action.yml", + "composite-action", + "env -u UNUSED cargo test --workspace", + LOCKED, + ) + self.assertIn("unsupported_cargo_indirect", codes(findings)) + + def _audit_env_delegation(self, delegated_script: str) -> dict: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + action('env -u UNUSED bash "$GITHUB_ACTION_PATH/scripts/build.sh"'), + encoding="utf-8", + ) + (root / "scripts" / "build.sh").write_text(delegated_script, encoding="utf-8") + return SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/build.sh"], + ) + + def test_env_option_operand_wrapper_rejects_locked_action_delegation(self) -> None: + result = self._audit_env_delegation( + "#!/usr/bin/env bash\ncargo test --locked --workspace\n" + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_env_option_operand_wrapper_rejects_unlocked_action_delegation(self) -> None: + result = self._audit_env_delegation("#!/usr/bin/env bash\ncargo test --workspace\n") + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_dot_slash_direct_script_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("./scripts/build.sh"), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_bare_relative_direct_script_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action("scripts/build.sh"), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_workspace_direct_script_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", action('"$GITHUB_WORKSPACE/scripts/build.sh"'), LOCKED + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def _audit_path_resolved_delegation(self, delegated_script: str) -> dict: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + """name: fixture +description: fixture +runs: + using: composite + steps: + - shell: bash + run: | + PATH="$GITHUB_ACTION_PATH/scripts:$PATH" + build.sh +""", + encoding="utf-8", + ) + (root / "scripts" / "build.sh").write_text(delegated_script, encoding="utf-8") + return SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/build.sh"], + ) + + def test_path_resolved_locked_action_script_fails_closed(self) -> None: + result = self._audit_path_resolved_delegation( + "#!/usr/bin/env bash\ncargo test --locked --workspace\n" + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_path_resolved_unlocked_action_script_fails_closed(self) -> None: + result = self._audit_path_resolved_delegation( + "#!/usr/bin/env bash\ncargo test --workspace\n" + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_scoped_path_assignment_before_bare_command_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action('PATH="$GITHUB_ACTION_PATH/scripts:$PATH" build.sh'), + LOCKED, + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_env_path_assignment_before_bare_command_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action('env PATH="$GITHUB_ACTION_PATH/scripts:$PATH" build.sh'), + LOCKED, + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_path_assignment_text_as_argument_is_not_path_mutation(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action("printf 'PATH=/tmp/example\\n'"), + LOCKED, + ) + self.assertNotIn("unsupported_action_script", codes(findings)) + + def _audit_builtin_path_writer(self, writer: str, delegated_script: str) -> dict: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + f"""name: fixture +description: fixture +runs: + using: composite + steps: + - shell: bash + run: | + {writer} + build.sh +""", + encoding="utf-8", + ) + (root / "scripts" / "build.sh").write_text(delegated_script, encoding="utf-8") + return SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/build.sh"], + ) + + def test_read_path_writer_rejects_locked_bare_action_script(self) -> None: + result = self._audit_builtin_path_writer( + 'read -r PATH <<< "$GITHUB_ACTION_PATH/scripts:$PATH"', + "#!/usr/bin/env bash\ncargo test --locked --workspace\n", + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_read_path_writer_rejects_unlocked_bare_action_script(self) -> None: + result = self._audit_builtin_path_writer( + 'read -r PATH <<< "$GITHUB_ACTION_PATH/scripts:$PATH"', + "#!/usr/bin/env bash\ncargo test --workspace\n", + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_printf_v_path_writer_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action("printf -v PATH '%s' /tmp/bin"), + LOCKED, + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_mapfile_path_writer_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action("mapfile PATH None: + findings = SURFACE.audit_action_text( + "action.yml", + action("readarray PATH None: + findings = SURFACE.audit_action_text( + "action.yml", + action("getopts ab PATH"), + LOCKED, + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_nameref_builtin_fails_closed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action("declare -n SEARCH_PATH=PATH"), + LOCKED, + ) + self.assertIn("unsupported_action_script", codes(findings)) + + def test_read_non_path_target_is_allowed(self) -> None: + findings = SURFACE.audit_action_text( + "action.yml", + action("read -r VALUE None: + findings = SURFACE.audit_action_text( + "action.yml", + action("printf -v VALUE '%s' ok"), + LOCKED, + ) + self.assertNotIn("unsupported_action_script", codes(findings)) + + def test_env_unset_operand_shell_heredoc_prefix_fails_closed(self) -> None: + self.assertTrue(SURFACE._heredoc_prefix_executes_shell("env -u UNUSED bash ")) + self.assertTrue(SURFACE._heredoc_prefix_executes_shell("env --unset=UNUSED bash ")) + + def test_wrapper_shell_heredoc_in_action_fails_closed(self) -> None: + text = """name: fixture +description: fixture +runs: + using: composite + steps: + - shell: bash + run: | + env -u UNUSED bash <<'SH' + cargo test --workspace + SH +""" + findings = SURFACE.audit_action_text("action.yml", text, LOCKED) + self.assertIn("unsupported_shell_heredoc", codes(findings)) + + def test_wrapper_shell_heredoc_in_workflow_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + workflow = root / ".github" / "workflows" / "test.yml" + workflow.parent.mkdir(parents=True) + workflow.write_text( + """name: fixture +on: push +jobs: + test: + runs-on: ubuntu-24.04 + steps: + - shell: bash + run: | + env -u UNUSED bash <<'SH' + cargo test --workspace + SH +""", + encoding="utf-8", + ) + result = SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=[".github/workflows/test.yml"], + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_shell_heredoc", [item["code"] for item in result["findings"]]) + + def test_non_shell_data_heredoc_is_not_rejected(self) -> None: + findings = SURFACE._shell_heredoc_findings( + "action.yml", "composite-action", "cat <<'EOF'\ndata\nEOF" + ) + self.assertNotIn("unsupported_shell_heredoc", codes(findings)) + + def _audit_hash_resolution(self, delegated_script: str, prefix: str = "hash") -> dict: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "scripts").mkdir() + (root / "action.yml").write_text( + f"""name: fixture +description: fixture +runs: + using: composite + steps: + - shell: bash + run: | + {prefix} -p "$GITHUB_ACTION_PATH/scripts/build.sh" build.sh + build.sh +""", + encoding="utf-8", + ) + (root / "scripts" / "build.sh").write_text(delegated_script, encoding="utf-8") + return SURFACE.audit_repository_surface( + root, + {"rules": {"cargo_locked_subcommands": sorted(LOCKED)}}, + tracked_files=["action.yml", "scripts/build.sh"], + ) + + def test_hash_resolution_rejects_locked_bare_action_script(self) -> None: + result = self._audit_hash_resolution( + "#!/usr/bin/env bash\ncargo test --locked --workspace\n" + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_hash_resolution_rejects_unlocked_bare_action_script(self) -> None: + result = self._audit_hash_resolution( + "#!/usr/bin/env bash\ncargo test --workspace\n" + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_command_hash_resolution_fails_closed(self) -> None: + result = self._audit_hash_resolution( + "#!/usr/bin/env bash\ncargo test --locked --workspace\n", prefix="command hash" + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_builtin_hash_resolution_fails_closed(self) -> None: + result = self._audit_hash_resolution( + "#!/usr/bin/env bash\ncargo test --locked --workspace\n", prefix="builtin hash" + ) + self.assertFalse(result["ok"]) + self.assertIn("unsupported_action_script", [item["code"] for item in result["findings"]]) + + def test_resolution_mutation_builtins_fail_closed(self) -> None: + for run in ("hash -r", "alias build.sh=true", "unalias build.sh", "enable -n printf"): + with self.subTest(run=run): + findings = SURFACE.audit_action_text("action.yml", action(run), LOCKED) + self.assertIn("unsupported_action_script", codes(findings)) + + +if __name__ == "__main__": + unittest.main() 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..911d0022 --- /dev/null +++ b/.github/scripts/test_audit_workflow_trust_syntax.py @@ -0,0 +1,311 @@ +#!/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_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" + " 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)) + + 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)) + + 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)) + + 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)) + + 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_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_direct_unlocked_cargo_retry_loop_fails(self) -> None: + text = workflow( + " - name: Retry\n" + " 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)) + + 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() diff --git a/.github/workflow-trust-policy.json b/.github/workflow-trust-policy.json new file mode 100644 index 00000000..be17b51f --- /dev/null +++ b/.github/workflow-trust-policy.json @@ -0,0 +1,116 @@ +{ + "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 + }, + "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": { + "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": [] +} \ No newline at end of file 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 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7008cba1..4ebfc227 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,12 +15,23 @@ 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 -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: AF-01 executable authority surface audit + run: python3 .github/scripts/audit_workflow_trust_surface.py + - name: AF-01 cross-step environment-channel audit + run: python3 .github/scripts/audit_workflow_trust_environment_channels.py - name: Format run: cargo fmt --all -- --check - name: Clippy @@ -143,4 +154,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 diff --git a/.github/workflows/registry-download-smoke.yml b/.github/workflows/registry-download-smoke.yml index 6a951d91..84f2f676 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 @@ -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 @@ -112,4 +108,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 diff --git a/scripts/github-action-run.sh b/scripts/github-action-run.sh index 8ea4d2fe..54580518 100644 --- a/scripts/github-action-run.sh +++ b/scripts/github-action-run.sh @@ -1,9 +1,13 @@ #!/usr/bin/env bash set -euo pipefail -binary="${1:-}" -if [[ -z "$binary" || ! -x "$binary" ]]; then - printf '::error title=commandF operational failure::commandF Action runner received no executable binary\n' +if [[ -z "${CARGO_TARGET_DIR:-}" ]]; then + printf '::error title=commandF operational failure::commandF Action runner received no target directory\n' + exit 1 +fi +binary="$CARGO_TARGET_DIR/debug/commandf" +if [[ ! -x "$binary" ]]; then + printf '::error title=commandF operational failure::built commandF binary is missing or not executable\n' exit 1 fi diff --git a/scripts/github-action.sh b/scripts/github-action.sh index 116a0bd9..65b839de 100644 --- a/scripts/github-action.sh +++ b/scripts/github-action.sh @@ -78,9 +78,8 @@ if ! cargo +1.97.1 build --locked \ emit_operational_failure "unable to build exact commandF source with the pinned toolchain" fi -binary="$CARGO_TARGET_DIR/debug/commandf" -if [[ ! -x "$binary" ]]; then +if [[ ! -x "$CARGO_TARGET_DIR/debug/commandf" ]]; then emit_operational_failure "built commandF binary is missing or not executable" fi -exec bash "$GITHUB_ACTION_PATH/scripts/github-action-run.sh" "$binary" +exec bash "$GITHUB_ACTION_PATH/scripts/github-action-run.sh" 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..5241d1c4 --- /dev/null +++ b/specs/015-af-01-trusted-development-baseline/stack-a-inventory.md @@ -0,0 +1,124 @@ +# 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. + +The source-backed root Action delegates through tracked repository-owned shell sources: + +```text +action.yml + -> scripts/github-action.sh + -> scripts/github-action-run.sh +``` + +Because those scripts can execute lockfile-consuming Cargo commands or delegate additional shell authority, Stack A treats the statically exposed `$GITHUB_ACTION_PATH/...` chain as part of the Action trust surface. Delegated Action shell sources must be tracked, recursively auditable, cycle-bounded, and exact-path static. This includes shell-interpreter execution, `source` / `.` inclusion, and direct execution of an exact `$GITHUB_ACTION_PATH/...` script. Absolute shell interpreter paths are normalized by basename for authority checks. Dynamic, relative, command-substituted, shell-`-c`, prefix/suffix-expanded, argument-dependent, or non-Action-root script authority fails closed. The Action runner binds execution to the built `$CARGO_TARGET_DIR/debug/commandf` path rather than accepting a runtime-selected executable argument. + +## 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. + +The source-backed Action also builds commandF from the repository lockfile. Its composite `run:` entries and recursively reachable tracked Action shell sources are therefore part of the same Cargo authority boundary. Variable-expanded executable positions must be proven statically non-Cargo; unknown or Cargo-resolving executable provenance fails closed. Cargo global-option syntax before the subcommand is outside the supported deterministic parser subset and fails closed rather than bypassing `--locked` enforcement. + +AF-01 audit treats at least these cargo subcommands as lockfile-consuming when present in workflow or Action 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`, while unsupported Cargo global-option forms fail closed; +- every tracked `action.yml` and `action.yaml` is scanned for external `uses:` references and composite `run:` Cargo authority; +- every statically delegated `$GITHUB_ACTION_PATH/...` shell source is tracked and recursively audited whether invoked through a shell interpreter, `source` / `.`, or direct execution, while dynamic or non-Action-root shell source selection fails closed; +- shell-interpreter heredocs are rejected even through supported wrappers or absolute interpreter paths so executable heredoc bodies cannot escape Cargo authority checks. + +## Scope boundary + +Stack A changes development-assurance configuration and source-backed Action execution hardening only. It does not change commandF product semantics, CF-06 production oracle identity, the CF-10 frozen corpus, report schemas, or runtime classification authority. 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.