diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 66d0ce5..0b9a177 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,43 +15,11 @@ concurrency: cancel-in-progress: true jobs: - policy-dependencies: - name: Locked policy wheels (${{ matrix.runner }}) - runs-on: ${{ matrix.runner }} - timeout-minutes: 5 - strategy: - fail-fast: false - matrix: - runner: [ubuntu-latest, ubuntu-24.04-arm] - steps: - - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 - with: - persist-credentials: false - - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 - with: - python-version: '3.13' - - name: Create a disposable policy-client venv - run: | - python -I -m venv "$RUNNER_TEMP/policy-venv" - printf '%s\n' "$RUNNER_TEMP/policy-venv/bin" >> "$GITHUB_PATH" - - uses: ./.github/actions/bootstrap-patched-pip # local - - name: Prove isolated installation and wheel integrity - run: python -I tests/test_policy_dependency_install.py - contracts: name: Shared contract tests - permissions: - contents: read - pull-requests: read - needs: [policy-dependencies] - if: always() runs-on: ubuntu-latest timeout-minutes: 10 steps: - - name: Require both native policy-wheel controls - env: - POLICY_RESULT: ${{ needs.policy-dependencies.result }} - run: test "$POLICY_RESULT" = success - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 @@ -81,15 +49,12 @@ jobs: - name: Install Python test dependencies run: >- - python -I -m pip install --only-binary :all: jsonschema==4.23.0 PyYAML==6.0.2 + python -m pip install --only-binary :all: jsonschema==4.23.0 PyYAML==6.0.2 - name: Test retained shared contracts run: >- python -m unittest tests.test_qualification_policy - tests.test_protected_policy_inputs - tests.test_protected_policy_check - tests.test_protected_policy_issue tests.test_regression_corpus tests.test_regression_evidence_schema tests.test_server_codec_counterfactual_schema @@ -103,18 +68,6 @@ jobs: tests.test_composer_container_trust tests.test_composer_container_effective_trust - - name: Exercise the read-only policy client on this public PR - if: github.event_name == 'pull_request' - env: - GITHUB_TOKEN: ${{ github.token }} - POLICY_PULL_REQUEST: ${{ github.event.pull_request.number }} - POLICY_BASE_SHA: ${{ github.event.pull_request.base.sha }} - run: | - python -I scripts/protected_policy_check.py \ - --pull-request "$POLICY_PULL_REQUEST" --base-sha "$POLICY_BASE_SHA" \ - > "$RUNNER_TEMP/protected-policy-read.json" - cat "$RUNNER_TEMP/protected-policy-read.json" - - name: Test actual Composer trust selection run: COMPOSER_BINARY="$(command -v composer)" phpunit --configuration phpunit-composer-trust.xml --do-not-cache-result diff --git a/.github/workflows/protected-policy.yml b/.github/workflows/protected-policy.yml deleted file mode 100644 index a8ea545..0000000 --- a/.github/workflows/protected-policy.yml +++ /dev/null @@ -1,77 +0,0 @@ -name: Protected workflow policy - -on: - pull_request_target: - branches: [main] - types: [opened, reopened, synchronize, ready_for_review] - -permissions: - contents: read - pull-requests: read - -jobs: - protected-policy: - name: Validate bounded workflow inputs - if: >- - github.repository == 'durable-workflow/.github' && - github.ref == 'refs/heads/main' && - github.event_name == 'pull_request_target' - runs-on: ubuntu-24.04 - timeout-minutes: 10 - environment: protected-policy-gate - steps: - - name: Check out the immutable protected base - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6 - with: - ref: ${{ github.sha }} - persist-credentials: false - - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 - with: - python-version: '3.13' - - name: Create the isolated policy runtime - run: | - python -I -m venv "$RUNNER_TEMP/protected-policy-python" - printf '%s\n' "$RUNNER_TEMP/protected-policy-python/bin" >> "$GITHUB_PATH" - - name: Select verified dependency and API trust - uses: ./.github/actions/bootstrap-patched-pip # local - with: - clients: pip,python,curl,git - - name: Install only the protected hash-locked parser - run: >- - python -I -m pip --isolated install --require-hashes - --only-binary=:all: --no-deps -r qualification/requirements.lock - - name: Require protected source and the configured App identity - env: - POLICY_GATE_APP_ID: ${{ vars.POLICY_GATE_APP_ID }} - shell: bash - run: | - set -euo pipefail - test "$GITHUB_WORKFLOW_SHA" = "$GITHUB_SHA" - test "$(git rev-parse HEAD)" = "$GITHUB_SHA" - [[ "$POLICY_GATE_APP_ID" =~ ^[1-9][0-9]{0,18}$ ]] - - name: Create the pilot-only checks token - id: app - uses: actions/create-github-app-token@64ab729a6294000f5aa4a4bc0b73d1f32bc45510 # v3.3.0 - env: - NODE_OPTIONS: --use-openssl-ca - NODE_EXTRA_CA_CERTS: '' - NODE_USE_SYSTEM_CA: '0' - NODE_TLS_REJECT_UNAUTHORIZED: '1' - with: - app-id: ${{ vars.POLICY_GATE_APP_ID }} - private-key: ${{ secrets.POLICY_GATE_APP_KEY }} - owner: durable-workflow - repositories: .github - permission-checks: write - - name: Validate data and issue the bound App check - env: - GITHUB_TOKEN: ${{ github.token }} - POLICY_CHECKS_TOKEN: ${{ steps.app.outputs.token }} - POLICY_GATE_APP_ID: ${{ vars.POLICY_GATE_APP_ID }} - POLICY_PULL_REQUEST: ${{ github.event.pull_request.number }} - POLICY_BASE_SHA: ${{ github.sha }} - POLICY_HEAD_SHA: ${{ github.event.pull_request.head.sha }} - run: >- - python -I scripts/protected_policy_issue.py - --pull-request "$POLICY_PULL_REQUEST" --base-sha "$POLICY_BASE_SHA" - --expected-head "$POLICY_HEAD_SHA" diff --git a/.github/workflows/sonar.yml b/.github/workflows/sonar.yml index 3dc5e87..d622a65 100644 --- a/.github/workflows/sonar.yml +++ b/.github/workflows/sonar.yml @@ -50,8 +50,8 @@ jobs: --coverage-clover build/composer-trust-coverage.xml - name: Produce source coverage run: | - python -I -m pip install --only-binary :all: coverage==7.10.7 jsonschema==4.23.0 PyYAML==6.0.2 - coverage run --source=scripts,conformance/cancellation/temporal,.github/actions/bootstrap-patched-pip -m unittest tests.test_qualification_policy tests.test_protected_policy_inputs tests.test_protected_policy_check tests.test_protected_policy_issue tests.test_regression_corpus tests.test_regression_evidence_schema tests.test_server_codec_counterfactual_schema tests.test_visual_evidence tests.test_temporal_runtime tests.test_sdk_metrics_contract tests.test_ci_trust tests.test_pip_effective_trust tests.test_cli_effective_trust tests.test_composer_trust_contract tests.test_composer_container_trust tests.test_composer_container_effective_trust + python -m pip install --only-binary :all: coverage==7.10.7 jsonschema==4.23.0 PyYAML==6.0.2 + coverage run --source=scripts,conformance/cancellation/temporal,.github/actions/bootstrap-patched-pip -m unittest tests.test_qualification_policy tests.test_regression_corpus tests.test_regression_evidence_schema tests.test_server_codec_counterfactual_schema tests.test_visual_evidence tests.test_temporal_runtime tests.test_sdk_metrics_contract tests.test_ci_trust tests.test_pip_effective_trust tests.test_cli_effective_trust tests.test_composer_trust_contract tests.test_composer_container_trust tests.test_composer_container_effective_trust coverage xml -o coverage.xml - uses: SonarSource/sonarqube-scan-action@d209202bc7d53ff1cc128f7f907dac145c9d6ae9 # v8.3.0 env: diff --git a/qualification/README.md b/qualification/README.md index f92beca..aa68c2e 100644 --- a/qualification/README.md +++ b/qualification/README.md @@ -33,8 +33,7 @@ Cache restores in pull-request workflows must also exclude external forks at the job or step condition. An earlier revision can populate the same PR cache, so event-partitioned keys alone do not protect a later revision's build. An exact shared key remains available for guarded restores and protected-main-only saves. -The scanner rejects `pull_request_target` outside the single content-bound -public pilot described below, mutable container +The scanner rejects `pull_request_target`, mutable container images, unreviewed `workflow_run` consumers, and privileged artifact consumers without exact producer and digest provenance. @@ -49,132 +48,6 @@ plan job. The workflow may have no other trigger. Product repositories own their tests, documentation deployment, publication, and deployment workflows. Editorial wording is outside this policy. -## Isolated, hash-locked policy bootstrap - -Server's policy preflight requires action revision -`fea245682582d0620054e8547a5f3c968f9ef0f0` and exactly: - -```sh -python -I -m pip --isolated install --require-hashes --only-binary=:all: --no-deps -r .central-action-policy/qualification/requirements.lock -``` - -The reviewed central checkout supplies the lock. It permits only the reviewed -PyYAML 6.0.2 CPython 3.13 glibc Linux wheels for amd64 and arm64. Native CI checks -both wheel installations, a tampered hash, and working-directory `pip.py` and -`pip/` shadows. Missing isolation/integrity/binary/no-dependency flags, another -requirements path, extra options, a missing bootstrap or an older bootstrap -revision fail the Server contract. The same command is available for staged -adoption in other targets. Their legacy forms remain explicit migration work, -not evidence of the locked-policy contract. This source check does not establish -an independent protected execution boundary; that gate requires its own issuer -and repository-rule acceptance. - -### Protected gate input layer - -`scripts/protected_policy_inputs.py` captures an open public pull request against -the expected repository ID and exact `main` commit. It reads the candidate's -exact commit/tree identities and regular workflow blobs, plus literal referenced -helper blobs needed by the existing Composer inspection. Paths, file counts, -individual sizes, total bytes and encoding are bounded. Symlinks, submodules, -truncated trees and a changed PR identity refuse. Materialized files are data -with mode `0600`; candidate commands are never executed by this layer. - -The caller must supply a bounded read-only GitHub API client from protected -source, keep candidate data outside the interpreter's import path, and call -`require_current` again immediately before issuing a result. This module does -not create a Check Run, grant merge authority, configure an Environment or prove -an issuer binding. Those live integration requirements remain open. Private -repositories are explicitly refused until their separate gate is qualified. - -### Read-only protected-policy pilot client - -`scripts/protected_policy_check.py` connects that input layer to the reviewed -validator for the public `.github` pilot. It binds the immutable repository ID, -requires the expected protected `main` SHA, and rechecks main and PR identity -after validation. Candidate workflow/helper files are temporary nonexecutable -data outside Python's import path. The policy always comes from the reviewed -central checkout; there is no candidate policy-path override. - -After the authenticated bootstrap exports its CA file, empty additional-root -directory and digest, a read-only caller uses: - -```sh -python -I scripts/protected_policy_check.py \ - --pull-request "$POLICY_PULL_REQUEST" --base-sha "$POLICY_BASE_SHA" -``` - -The client requires curl 8.4 or newer, an explicit matching CA bundle and no -additional roots. Transfers are HTTPS GETs to the fixed GitHub API host, with -no redirects or automatic content decoding. Each response is capped at 8 MiB; -there is a 15-second request budget and a shared 120-second API budget. JSON -depth/node counts, endpoint shapes and request counts are also bounded. A read -token, when supplied as `GITHUB_TOKEN`, travels on stdin rather than process -arguments. The caller grants only Contents read and Pull requests read. Never -give the reader the App signing key or its Checks-write token. - -Success prints a bounded identity/digest report. Failure prints escaped JSON -and exits nonzero; candidate text cannot become a GitHub workflow command. -Ordinary PR CI exercises the real read client and validator. That candidate -controlled job remains a component test. Protected base execution, the App -issuer, immediate identity recheck, environment-denial tests and issuer-bound -repository rules still require separate live acceptance before enforcement. - -### Prepared App issuer component - -`scripts/protected_policy_issue.py` prepares the checks-only issuer for that -same public pilot. It runs the validator itself rather than accepting a -candidate-supplied report. The read-only `GITHUB_TOKEN` and separate -`POLICY_CHECKS_TOKEN` installation token have different roles; the issuer never -accepts or reads an App signing key. `POLICY_GATE_APP_ID` must name the reviewed -installation's App. The App must have Checks write and implicit Metadata read -only, installed solely on the pilot repository. - -The intended protected caller uses: - -```sh -python -I scripts/protected_policy_issue.py \ - --pull-request "$POLICY_PULL_REQUEST" --base-sha "$POLICY_BASE_SHA" \ - --expected-head "$POLICY_HEAD_SHA" -``` - -It checks the protected-main workflow context and exact event head, validates -bounded candidate data, creates one incomplete Check Run, and verifies the -returned check name, original head and App ID. Only then does it re-read the -protected main and complete PR identity immediately before completing that -same check successfully. Read and write credentials use stdin, with explicit -reviewed TLS roots and the existing transfer/time/JSON bounds. The App client -has no arbitrary endpoint or repository input. Refusal exits nonzero and never -reports success; a check already created may remain incomplete. A later head -cannot inherit an earlier head's successful check. - -Caller environment markers are defense in depth, not proof of secret denial. -The prepared workflow checks out `github.sha` from protected main, verifies -both the workflow SHA and actual checkout match it, installs the hash-locked -parser, then uses the pinned official App-token action. Its token is restricted -to Checks write on the one public pilot. The key is an Environment secret and -never enters the reader/issuer program. The token action selects only the -verified OpenSSL CA file and empty additional-root directory; bundled, extra -and additional system roots are excluded for that credentialed Node step. - -The policy permits this one exact UTF-8 workflow SHA-256, target and path. -Another path/repository, changed checkout/permission/credential handling, -different content or additional exception refuses. Updating that root requires -another explicit source review. All other `pull_request_target` workflows -remain forbidden. This does not validate arbitrary changes to central policy -code; root-of-trust changes still require independent maintainer/Security review. - -This bootstrap revision is prepared source, not a live enforcement claim. It -cannot authorize its own adoption. Security must accept the exact source; -the account holder must establish the App and key; and the live pilot must -prove Environment denial, wrong-issuer nonqualification, candidate nonexecution, -stale-head refusal and a safe successful check before the required-check rule -is bound to that App. The known Environment configuration alone is not proof. -No App key, live write or private-repository adoption is exercised by the -constructed-response tests. Existing Server/publication holds remain. - -The transport uses GitHub's [create and update Check Run API](https://docs.github.com/en/rest/checks/runs) -and checks the documented [workflow environment variables](https://docs.github.com/en/actions/reference/workflows-and-actions/variables). - ## Clean trust for dependency requests ### Composer inside a container diff --git a/qualification/policy.json b/qualification/policy.json index 00f34f0..2935d97 100644 --- a/qualification/policy.json +++ b/qualification/policy.json @@ -25,9 +25,6 @@ "actions/configure-pages": { "45bfe0192ca1faeb007ade9deae92b16b8254a0d": "v6" }, - "actions/create-github-app-token": { - "64ab729a6294000f5aa4a4bc0b73d1f32bc45510": "v3.3.0" - }, "actions/deploy-pages": { "cd2ce8fcbc39b97be8ca5fce6e763baed58fa128": "v5" }, @@ -85,7 +82,6 @@ "699d09d602fc0353160bdecaa2b23f07ef4c7c21": "clean-trust", "75587fc33ff40c34f32ef777293dad56bb6f48c7": "clean-trust", "76f333804e68725059039fd54df241d40d855cd3": "clean-trust", - "fea245682582d0620054e8547a5f3c968f9ef0f0": "clean-trust", "ff93e15dd3d3aa0b5ba8991fe5101a9e1be49c10": "patched-pip" }, "helm/chart-testing-action": { @@ -169,13 +165,7 @@ } }, "privileged_workflow_run_consumers": {}, - "pull_request_target_exceptions": [ - { - "target": "github-control-plane", - "path": ".github/workflows/protected-policy.yml", - "sha256": "642d204ea939ca5eba7ef51be9e4ae56a35177413caabe849315086d3d9d2d31" - } - ] + "pull_request_target_exceptions": [] }, "targets": { "ai": { @@ -306,7 +296,6 @@ "workflows": [ { "action_policy_preflight": true, - "locked_policy_dependency_required": true, "matrix_independent": false, "path": "phpunit-feature.yml", "required_check": "PHPUnit feature suite" diff --git a/qualification/requirements.lock b/qualification/requirements.lock deleted file mode 100644 index 2cf2a45..0000000 --- a/qualification/requirements.lock +++ /dev/null @@ -1,5 +0,0 @@ -# PyPI PyYAML 6.0.2 wheels for CPython 3.13 on glibc Linux amd64/arm64. -# Source: https://pypi.org/pypi/PyYAML/6.0.2/json -PyYAML==6.0.2 \ - --hash=sha256:70b189594dbe54f75ab3a1acec5f1e3faa7e8cf2f1e08d9b561cb41b845f69d5 \ - --hash=sha256:0ffe8360bab4910ef1b9e87fb812d8bc0a308b0d0eef8c8f44e0254ab3b07133 diff --git a/qualification/schema.json b/qualification/schema.json index b9c4af0..9fb3517 100644 --- a/qualification/schema.json +++ b/qualification/schema.json @@ -316,17 +316,7 @@ "type": "object" }, "pull_request_target_exceptions": { - "items": { - "additionalProperties": false, - "properties": { - "target": {"const": "github-control-plane"}, - "path": {"const": ".github/workflows/protected-policy.yml"}, - "sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"} - }, - "required": ["target", "path", "sha256"], - "type": "object" - }, - "maxItems": 1, + "maxItems": 0, "type": "array" } }, @@ -377,9 +367,6 @@ "workflow": { "additionalProperties": false, "properties": { - "locked_policy_dependency_required": { - "type": "boolean" - }, "action_policy_preflight": { "type": "boolean" }, diff --git a/scripts/protected_policy_check.py b/scripts/protected_policy_check.py deleted file mode 100644 index 7be09ca..0000000 --- a/scripts/protected_policy_check.py +++ /dev/null @@ -1,239 +0,0 @@ -"""Read-only pilot runner for the protected workflow-policy check. - -An enforcement caller must use the reviewed central checkout. Candidate inputs -remain data in a separate temporary directory. Ordinary CI can exercise this -client, but does not issue a Check Run or establish its App identity. The -protected issuer must independently recheck the PR identity. -""" - -from __future__ import annotations - -import argparse -import hashlib -import json -import os -from pathlib import Path -import re -import subprocess -import sys -import tempfile -import time -from typing import Any - -# Python -I excludes both the working directory and the script directory. -# Add only this reviewed checkout, never the candidate materialization root. -ROOT = Path(__file__).resolve().parents[1] -if __package__ in {None, ""}: - sys.path.insert(0, str(ROOT)) - -from scripts.protected_policy_inputs import ( # noqa: E402 - MAX_FILES, MAX_TREE_ENTRIES, InputRefused, collect_inputs, read_identity, require_current, write_inputs, -) - -PILOT_REPOSITORY = "durable-workflow/.github" -PILOT_REPOSITORY_ID = 575230056 -MAX_RESPONSE_BYTES = 8 * 1024 * 1024 -MAX_REQUESTS = MAX_FILES + 16 -MAX_JSON_DEPTH = 64 -MAX_JSON_NODES = MAX_TREE_ENTRIES * 16 -TOTAL_SECONDS = 120 -REQUEST_SECONDS = 15 -ENDPOINT = re.compile( - r"/repos/[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}/(?:" - r"pulls/[1-9][0-9]*|branches/main|git/commits/[0-9a-f]{40}|" - r"git/trees/[0-9a-f]{40}\?recursive=1|git/blobs/[0-9a-f]{40})" -) - - -def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: - result = {} - for key, value in pairs: - if key in result: - raise InputRefused("GitHub JSON contains a duplicate member") - result[key] = value - return result - - -def reject_constant(_value: str) -> None: - raise InputRefused("GitHub JSON contains a non-finite number") - - -def decode_metadata(raw: str) -> dict[str, Any]: - try: - value = json.loads(raw, object_pairs_hook=unique_object, parse_constant=reject_constant) - except (ValueError, RecursionError) as error: - raise InputRefused("GitHub did not provide unambiguous bounded JSON") from error - if not isinstance(value, dict): - raise InputRefused("The selected GitHub endpoints must return JSON objects") - # Validate the standard decoder's result without recursively walking an - # attacker-controlled structure or relying on Python's parser stack limit. - pending, count = [(value, 0)], 0 - while pending: - item, depth = pending.pop() - count += 1 - if depth > MAX_JSON_DEPTH or count > MAX_JSON_NODES: - raise InputRefused("GitHub JSON exceeds the structure bound") - if isinstance(item, dict): - children = item.values() - elif isinstance(item, list): - children = item - else: - children = () - if len(children) > MAX_JSON_NODES: - raise InputRefused("GitHub JSON exceeds the collection bound") - pending.extend((child, depth + 1) for child in children) - return value - - -class GitHubReadClient: - """Use curl's byte and wall-clock limits, without redirects or decompression.""" - - def __init__(self, bundle: Path, directory: Path, expected_digest: str, token: str = "", *, - run=subprocess.run, clock=time.monotonic): - self.run, self.clock = run, clock - self.deadline = clock() + TOTAL_SECONDS - self.requests = 0 - if token and re.fullmatch(r"[A-Za-z0-9_.-]+", token) is None: - raise InputRefused("The read credential has an invalid format") - self.config = f'header = "Authorization: Bearer {token}"\n' if token else "" - if (bundle.is_symlink() or not bundle.is_file() or bundle.stat().st_size > 2 * 1024 * 1024 - or directory.is_symlink() or not directory.is_dir() or any(directory.iterdir())): - raise InputRefused("Explicit reviewed CA file and empty additional-root directory are required") - if (re.fullmatch(r"[0-9a-f]{64}", expected_digest) is None - or hashlib.sha256(bundle.read_bytes()).hexdigest() != expected_digest): - raise InputRefused("The selected CA file does not match the reviewed bootstrap digest") - self.bundle, self.directory = str(bundle.resolve()), str(directory.resolve()) - try: - version = run(["/usr/bin/curl", "--disable", "--version"], stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, timeout=2, check=False) - except (OSError, subprocess.TimeoutExpired) as error: - raise InputRefused("Cannot inspect the fixed curl executable") from error - match = re.match(rb"curl (\d+)\.(\d+)\.(\d+)\b", version.stdout) - if version.returncode or not match or tuple(map(int, match.groups())) < (8, 4, 0): - raise InputRefused("curl 8.4 or newer is required for unknown-length response bounds") - - def __call__(self, path: str) -> Any: - if ENDPOINT.fullmatch(path) is None or any(part in {".", ".."} for part in path.split("/")): - raise InputRefused("The policy reader permits only exact GitHub metadata/blob endpoints") - return self._request_json(path, method="GET", expected_status=b"200") - - def _request_json(self, path: str, *, method: str, expected_status: bytes, body=None) -> Any: - """Bound the transport used by the reader and the fixed-endpoint issuer. - - This is internal transport, not a candidate-controlled request API. - Each caller restricts its own endpoints and allowed operations. - """ - remaining = self.deadline - self.clock() - if remaining <= 0 or self.requests >= MAX_REQUESTS: - raise InputRefused("The policy API time or request bound is exhausted") - self.requests += 1 - timeout = min(REQUEST_SECONDS, remaining) - with tempfile.TemporaryDirectory(prefix="policy-response-") as temporary: - response = Path(temporary) / "body.json" - data_arguments = [] - if body is not None: - data = json.dumps(body, sort_keys=True, allow_nan=False).encode() - if len(data) > 16 * 1024: - raise InputRefused("The policy check request exceeds its byte bound") - payload = Path(temporary) / "request.json" - with payload.open("xb") as handle: - payload.chmod(0o600) - handle.write(data) - data_arguments = ["--header", "Content-Type: application/json", "--data-binary", "@" + str(payload)] - with (Path(temporary) / "curl-error").open("wb") as errors: - command = [ - "/usr/bin/curl", "--disable", "--silent", "--show-error", "--fail", - "--globoff", "--proto", "=https", "--max-redirs", "0", - "--max-time", str(timeout), "--connect-timeout", str(min(5, timeout)), - "--max-filesize", str(MAX_RESPONSE_BYTES), "--cacert", self.bundle, - "--capath", self.directory, "--header", "Accept: application/vnd.github+json", - "--header", "Accept-Encoding: identity", "--header", "X-GitHub-Api-Version: 2022-11-28", - "--request", method, *data_arguments, - "--config", "-", "--output", str(response), "--write-out", "%{http_code}", - "--url", "https://api.github.com" + path, - ] - try: - result = self.run(command, input=self.config.encode(), stdout=subprocess.PIPE, - stderr=errors, timeout=timeout, check=False) - except (OSError, subprocess.TimeoutExpired) as error: - raise InputRefused("The bounded GitHub request did not complete") from error - if result.returncode or result.stdout != expected_status: - status = result.stdout.decode("ascii", errors="replace")[:3] - raise InputRefused(f"GitHub request refused: curl exit {result.returncode}, HTTP {status}") - if self.clock() >= self.deadline: - raise InputRefused("The total policy API deadline expired") - if not response.is_file() or response.stat().st_size > MAX_RESPONSE_BYTES: - raise InputRefused("The GitHub response is absent or exceeds the byte bound") - try: - value = decode_metadata(response.read_text(encoding="utf-8")) - except InputRefused: - raise - except UnicodeError as error: - raise InputRefused("GitHub JSON must use UTF-8") from error - if self.clock() >= self.deadline: - raise InputRefused("The total policy API deadline expired during decoding") - return value - - -def require_protected_base(read, base_sha: str) -> None: - record = read(f"/repos/{PILOT_REPOSITORY}/branches/main") - try: - if record["name"] != "main" or record["protected"] is not True or record["commit"]["sha"] != base_sha: - raise InputRefused("The pilot's protected main branch moved or cannot be verified") - except (KeyError, TypeError) as error: - raise InputRefused("Incomplete protected-branch metadata") from error - - -def validate_snapshot(read, number: int, base_sha: str, *, validator=None) -> dict[str, Any]: - from scripts.qualification_policy import PolicyError, load_policy, validate_local_action_references - - require_protected_base(read, base_sha) - identity = read_identity(read, PILOT_REPOSITORY, PILOT_REPOSITORY_ID, number, base_sha) - files = collect_inputs(read, identity) - policy_path = ROOT / "qualification/policy.json" - try: - policy = load_policy(policy_path) - except PolicyError as error: - raise InputRefused("The reviewed central policy is invalid") from error - check = validator or validate_local_action_references - with tempfile.TemporaryDirectory(prefix="policy-candidate-") as temporary: - candidate = Path(temporary) / "data" - write_inputs(files, candidate) - try: - actions = check(policy, candidate / ".github/workflows", "github-control-plane") - except (PolicyError, RecursionError) as error: - raise InputRefused(str(error)) from error - require_protected_base(read, base_sha) - require_current(read, identity) - digests = {name: hashlib.sha256(raw).hexdigest() for name, raw in sorted(files.items())} - return { - "schema": "durable-workflow.protected-policy-read/v1", "repository": identity.repository, - "repository_id": identity.repository_id, "pull_request": identity.number, - "base_sha": identity.base_sha, "head_sha": identity.head_sha, - "head_repository": identity.head_repository, "head_repository_id": identity.head_repository_id, - "policy_sha256": hashlib.sha256(policy_path.read_bytes()).hexdigest(), - "inputs_sha256": hashlib.sha256(json.dumps(digests, sort_keys=True).encode()).hexdigest(), - "input_count": len(files), "approved_action_count": len(actions), - } - - -def main() -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--pull-request", type=int, required=True) - parser.add_argument("--base-sha", required=True) - args = parser.parse_args() - try: - read = GitHubReadClient( - Path(os.environ["CURL_CA_BUNDLE"]), Path(os.environ["SSL_CERT_DIR"]), - os.environ["DW_CI_CA_SHA256"], os.environ.get("GITHUB_TOKEN", ""), - ) - report = validate_snapshot(read, args.pull_request, args.base_sha) - print(json.dumps(report, sort_keys=True)) - return 0 - except (InputRefused, OSError, KeyError) as error: - print(json.dumps({"policy_refused": str(error)[:1000]}), file=sys.stderr) - return 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/protected_policy_inputs.py b/scripts/protected_policy_inputs.py deleted file mode 100644 index 00dddf2..0000000 --- a/scripts/protected_policy_inputs.py +++ /dev/null @@ -1,223 +0,0 @@ -"""Read a PR's policy inputs as bounded data, without checking out candidate code.""" - -from __future__ import annotations - -import base64 -import binascii -import re -from collections.abc import Callable -from dataclasses import dataclass -from pathlib import Path, PurePosixPath -from typing import Any - - -MAX_TREE_ENTRIES = 20_000 -MAX_FILES = 256 -MAX_FILE_BYTES = 1024 * 1024 -MAX_TOTAL_BYTES = 8 * 1024 * 1024 -SHA = re.compile(r"[0-9a-f]{40}") -SLUG = re.compile(r"[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}") -HELPERS = re.compile(r"(?:\./)?(?:scripts|\.github)/[A-Za-z0-9_./-]+") -ReadJson = Callable[[str], Any] - - -class InputRefused(ValueError): - """Candidate identity or input data cannot be verified within the bounds.""" - - -def checked_sha(value: object) -> str: - if not isinstance(value, str) or SHA.fullmatch(value) is None: - raise InputRefused("An exact GitHub commit or blob identity is required") - return value - - -def checked_slug(value: object) -> str: - if ( - not isinstance(value, str) - or SLUG.fullmatch(value) is None - or any(part in {".", ".."} for part in value.split("/")) - ): - raise InputRefused("A GitHub owner/repository identity is required") - return value - - -def checked_path(value: object) -> str: - if not isinstance(value, str) or len(value) > 1024 or not value: - raise InputRefused("Invalid policy input path") - path = PurePosixPath(value) - if ( - path.is_absolute() - or not path.parts - or path.as_posix() != value - or any(part in {".", ".."} for part in path.parts) - or "\\" in value - or any(ord(character) < 32 or ord(character) == 127 for character in value) - ): - raise InputRefused("Policy input path must be canonical and repository-relative") - return value - - -def positive_id(value: object) -> int: - if type(value) is not int or value <= 0: - raise InputRefused("A positive GitHub object ID is required") - return value - - -@dataclass(frozen=True) -class PullIdentity: - repository: str - repository_id: int - number: int - base_sha: str - head_sha: str - head_repository: str - head_repository_id: int - - -def read_identity( - read: ReadJson, repository: str, repository_id: int, number: int, expected_base_sha: str, -) -> PullIdentity: - checked_slug(repository) - if not repository.startswith("durable-workflow/"): - raise InputRefused("The protected gate is limited to Durable Workflow repositories") - positive_id(repository_id) - positive_id(number) - checked_sha(expected_base_sha) - record = read(f"/repos/{repository}/pulls/{number}") - try: - base, head = record["base"], record["head"] - if ( - record["state"] != "open" - or positive_id(record["number"]) != number - or base["ref"] != "main" - or base["sha"] != expected_base_sha - or base["repo"]["full_name"] != repository - or positive_id(base["repo"]["id"]) != repository_id - or base["repo"]["private"] is not False - or head["repo"]["private"] is not False - ): - raise InputRefused("The PR must remain open against the expected repository main branch") - return PullIdentity( - repository, repository_id, number, checked_sha(base["sha"]), checked_sha(head["sha"]), - checked_slug(head["repo"]["full_name"]), positive_id(head["repo"]["id"]), - ) - except (KeyError, TypeError) as error: - raise InputRefused("Incomplete GitHub PR identity") from error - - -def require_current(read: ReadJson, identity: PullIdentity) -> None: - if read_identity( - read, identity.repository, identity.repository_id, identity.number, identity.base_sha, - ) != identity: - raise InputRefused("PR head, base or repository changed during policy validation") - - -def _tree_entries(read: ReadJson, identity: PullIdentity) -> dict[str, dict]: - prefix = f"/repos/{identity.head_repository}/git" - commit = read(f"{prefix}/commits/{identity.head_sha}") - try: - if checked_sha(commit["sha"]) != identity.head_sha: - raise InputRefused("GitHub returned another candidate commit") - tree_sha = checked_sha(commit["tree"]["sha"]) - tree = read(f"{prefix}/trees/{tree_sha}?recursive=1") - entries = tree["tree"] - if ( - tree.get("truncated") is not False - or checked_sha(tree["sha"]) != tree_sha - or not isinstance(entries, list) - or len(entries) > MAX_TREE_ENTRIES - ): - raise InputRefused("Candidate tree is incomplete or exceeds the input bound") - result = {} - for entry in entries: - path = checked_path(entry["path"]) - if path in result: - raise InputRefused("Candidate tree contains a duplicate path") - result[path] = entry - return result - except (KeyError, TypeError) as error: - raise InputRefused("Malformed GitHub tree metadata") from error - - -def _text_blob(read: ReadJson, repository: str, entry: dict) -> bytes: - if ( - entry.get("type") != "blob" - or entry.get("mode") not in {"100644", "100755"} - or type(entry.get("size")) is not int - or not 0 <= entry["size"] <= MAX_FILE_BYTES - ): - raise InputRefused("Policy input must be a bounded regular file, never a symlink or submodule") - sha = checked_sha(entry.get("sha")) - blob = read(f"/repos/{repository}/git/blobs/{sha}") - try: - content = blob["content"] - if ( - blob["sha"] != sha - or blob["encoding"] != "base64" - or type(blob["size"]) is not int - or blob["size"] != entry["size"] - or not isinstance(content, str) - or len(content) > MAX_FILE_BYTES * 2 - ): - raise InputRefused("GitHub blob identity, size or encoding does not match the candidate tree") - raw = base64.b64decode(content.replace("\n", ""), validate=True) - if len(raw) != entry["size"]: - raise InputRefused("GitHub blob contents do not match the declared size") - raw.decode("utf-8") - return raw - except (KeyError, TypeError, UnicodeError, binascii.Error) as error: - raise InputRefused("Policy input must be valid UTF-8 in a verified GitHub blob") from error - - -def _referenced_inputs(raw: bytes, entries: dict[str, dict], collected: dict[str, bytes]) -> list[str]: - result = [] - for helper in HELPERS.findall(raw.decode("utf-8")): - helper = checked_path(helper.removeprefix("./")) - if helper in entries and entries[helper].get("type") != "tree" and helper not in collected: - result.append(helper) - return result - - -def collect_inputs(read: ReadJson, identity: PullIdentity) -> dict[str, bytes]: - entries = _tree_entries(read, identity) - pending = sorted( - path for path in entries - if PurePosixPath(path).parent.as_posix() == ".github/workflows" and path.endswith((".yml", ".yaml")) - ) - if not pending: - raise InputRefused("Candidate has no workflow files") - collected: dict[str, bytes] = {} - total = 0 - while pending: - path = pending.pop() - if path in collected: - continue - if len(collected) >= MAX_FILES: - raise InputRefused("Too many workflow or referenced helper inputs") - raw = _text_blob(read, identity.head_repository, entries[path]) - total += len(raw) - if total > MAX_TOTAL_BYTES: - raise InputRefused("Workflow and helper inputs exceed the total byte bound") - collected[path] = raw - pending.extend(_referenced_inputs(raw, entries, collected)) - require_current(read, identity) - return collected - - -def write_inputs(files: dict[str, bytes], destination: Path) -> None: - """Materialize data in a new directory, without preserving executable modes.""" - if len(files) > MAX_FILES: - raise InputRefused("Policy materialization exceeds the file bound") - for name, contents in files.items(): - checked_path(name) - if not isinstance(contents, bytes) or len(contents) > MAX_FILE_BYTES: - raise InputRefused("Policy materialization requires bounded byte strings") - if sum(len(value) for value in files.values()) > MAX_TOTAL_BYTES: - raise InputRefused("Policy materialization exceeds the byte bound") - destination.mkdir(mode=0o700) - for name, contents in files.items(): - path = destination / checked_path(name) - path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) - with path.open("xb") as handle: - handle.write(contents) - path.chmod(0o600) diff --git a/scripts/protected_policy_issue.py b/scripts/protected_policy_issue.py deleted file mode 100644 index 499a270..0000000 --- a/scripts/protected_policy_issue.py +++ /dev/null @@ -1,152 +0,0 @@ -"""Prepared checks-only App issuer for the public central-policy pilot. - -Run only from reviewed central code in the protected execution environment. -This component is not a workflow binding or proof of live enforcement. -""" - -from __future__ import annotations - -import argparse -import json -import os -from pathlib import Path -import re -import sys - -ROOT = Path(__file__).resolve().parents[1] -if __package__ in {None, ""}: - sys.path.insert(0, str(ROOT)) - -from scripts.protected_policy_check import ( # noqa: E402 - PILOT_REPOSITORY, PILOT_REPOSITORY_ID, GitHubReadClient, - require_protected_base, validate_snapshot, -) -from scripts.protected_policy_inputs import ( # noqa: E402 - InputRefused, checked_sha, positive_id, read_identity, require_current, -) - -CHECK_NAME = "Protected workflow policy" -WORKFLOW_REF = f"{PILOT_REPOSITORY}/.github/workflows/protected-policy.yml@refs/heads/main" - - -def require_protected_execution(environment, base_sha: str) -> None: - """Check caller context; actual credential denial still relies on GitHub.""" - expected = { - "GITHUB_ACTIONS": "true", "GITHUB_REPOSITORY": PILOT_REPOSITORY, - "GITHUB_REPOSITORY_ID": str(PILOT_REPOSITORY_ID), "GITHUB_REF": "refs/heads/main", - "GITHUB_SHA": checked_sha(base_sha), "GITHUB_EVENT_NAME": "pull_request_target", - "GITHUB_WORKFLOW_REF": WORKFLOW_REF, - "GITHUB_WORKFLOW_SHA": base_sha, - } - if any(environment.get(name) != value for name, value in expected.items()): - raise InputRefused("The issuer requires the reviewed protected-main workflow context") - - -class CheckRunIssuer(GitHubReadClient): - """Write one Check Run, verifying its App before completing it successfully.""" - - def __init__(self, *args, expected_app_id: int, **options): - self.expected_app_id = positive_id(expected_app_id) - self.check_id = None - self.head_sha = None - super().__init__(*args, **options) - if not self.config: - raise InputRefused("A separate checks-only installation token is required") - - def __call__(self, _path): - raise InputRefused("The App credential is restricted to the check issuer") - - def _verify(self, record, *, status: str, conclusion) -> int: - try: - check_id = positive_id(record["id"]) - if (positive_id(record["app"]["id"]) != self.expected_app_id - or record["name"] != CHECK_NAME or record["head_sha"] != self.head_sha - or record["status"] != status or record["conclusion"] != conclusion - or (self.check_id is not None and check_id != self.check_id)): - raise InputRefused("The Check Run identity or App issuer does not match") - return check_id - except (KeyError, TypeError) as error: - raise InputRefused("GitHub returned incomplete Check Run metadata") from error - - def begin(self, head_sha: str) -> None: - if self.head_sha is not None: - raise InputRefused("The issuer permits only one Check Run") - self.head_sha = checked_sha(head_sha) - record = self._request_json( - f"/repos/{PILOT_REPOSITORY}/check-runs", method="POST", expected_status=b"201", - body={"name": CHECK_NAME, "head_sha": self.head_sha, "status": "in_progress"}, - ) - # A wrong installation may create an incomplete check, never a success. - self.check_id = self._verify(record, status="in_progress", conclusion=None) - - def complete(self, report: dict) -> dict: - if self.check_id is None or report.get("head_sha") != self.head_sha: - raise InputRefused("No matching App-issued check is ready for completion") - summary = "\n".join([ - "Reviewed central policy accepted bounded, data-only workflow inputs.", - f"Policy SHA-256: {report['policy_sha256']}", - f"Inputs SHA-256: {report['inputs_sha256']}", - ]) - record = self._request_json( - f"/repos/{PILOT_REPOSITORY}/check-runs/{self.check_id}", - method="PATCH", expected_status=b"200", - body={"status": "completed", "conclusion": "success", "output": { - "title": "Workflow policy accepted", "summary": summary, - }}, - ) - self._verify(record, status="completed", conclusion="success") - # A completed issuer object cannot be reused to publish another result. - self.check_id = None - return {"check_name": CHECK_NAME, "app_id": self.expected_app_id, "head_sha": self.head_sha} - - -def qualify_and_issue(read, issuer, number: int, base_sha: str, expected_head: str) -> dict: - checked_sha(expected_head) - identity = read_identity(read, PILOT_REPOSITORY, PILOT_REPOSITORY_ID, number, base_sha) - if identity.head_sha != expected_head: - raise InputRefused("The event's expected head is no longer current") - report = validate_snapshot(read, number, base_sha) - if any(report.get(key) != value for key, value in { - "repository": identity.repository, "repository_id": identity.repository_id, - "pull_request": identity.number, "head_sha": identity.head_sha, "base_sha": identity.base_sha, - "head_repository": identity.head_repository, "head_repository_id": identity.head_repository_id, - }.items()): - raise InputRefused("The validated snapshot does not match the original PR identity") - issuer.begin(identity.head_sha) - # Recheck after the App identity is known, immediately before success. - require_protected_base(read, base_sha) - require_current(read, identity) - return {**report, **issuer.complete(report)} - - -def main() -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--pull-request", type=int, required=True) - parser.add_argument("--base-sha", required=True) - parser.add_argument("--expected-head", required=True) - args = parser.parse_args() - try: - require_protected_execution(os.environ, args.base_sha) - bundle = Path(os.environ["CURL_CA_BUNDLE"]) - directory = Path(os.environ["SSL_CERT_DIR"]) - digest = os.environ["DW_CI_CA_SHA256"] - read = GitHubReadClient(bundle, directory, digest, os.environ["GITHUB_TOKEN"]) - app_id = os.environ["POLICY_GATE_APP_ID"] - if re.fullmatch(r"[1-9]\d{0,18}", app_id, flags=re.ASCII) is None: - raise InputRefused("A canonical positive App ID is required") - issuer = CheckRunIssuer( - bundle, directory, digest, os.environ["POLICY_CHECKS_TOKEN"], - expected_app_id=int(app_id), - ) - issuer.deadline = min(issuer.deadline, read.deadline) - print(json.dumps(qualify_and_issue( - read, issuer, args.pull_request, args.base_sha, args.expected_head, - ), sort_keys=True)) - return 0 - except (InputRefused, OSError, KeyError) as error: - print(json.dumps({"policy_refused": str(error)[:1000]}), file=sys.stderr) - return 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/qualification_policy.py b/scripts/qualification_policy.py index d8750c5..e9a17e6 100644 --- a/scripts/qualification_policy.py +++ b/scripts/qualification_policy.py @@ -7,7 +7,6 @@ import contextlib import email.utils import errno -import hashlib import http.client import json import os @@ -18,7 +17,7 @@ import urllib.error import urllib.parse import urllib.request -from collections.abc import Callable, Iterator, Mapping +from collections.abc import Callable, Mapping from datetime import UTC from pathlib import Path from typing import Any @@ -29,23 +28,6 @@ CENTRAL_ACTION_REPOSITORY = "durable-workflow/.github" COMMIT_PATTERN = re.compile(r"^[0-9a-f]{40}$") SUPPORTED_JAVASCRIPT_ACTION_RUNTIMES = ["node24"] -LOCKED_POLICY_BOOTSTRAP_REVISION = "fea245682582d0620054e8547a5f3c968f9ef0f0" -LOCKED_POLICY_INSTALL_COMMAND = ( - "python -I -m pip --isolated install --require-hashes --only-binary=:all: --no-deps " - "-r .central-action-policy/qualification/requirements.lock" -) -# Complete caller-supplied environment for the reviewed preflight. Startup -# settings below are approved only as literal empty clears, never as hooks. -PREFLIGHT_CI_ENVIRONMENT = { - "CI", "CARGO_TERM_COLOR", "RUST_BACKTRACE", "PYTHONUNBUFFERED", - "PYTHONDONTWRITEBYTECODE", "PYTHONUTF8", "FORCE_COLOR", "NO_COLOR", -} -PREFLIGHT_CLEAR_STARTUP_ENVIRONMENT = {"BASH_ENV", "ENV", "LD_PRELOAD", "NODE_OPTIONS"} -PROTECTED_POLICY_EXCEPTION = { - "target": "github-control-plane", - "path": ".github/workflows/protected-policy.yml", - "sha256": "642d204ea939ca5eba7ef51be9e4ae56a35177413caabe849315086d3d9d2d31", -} EXPECTED_TARGETS = { "ai": ("ai", "main"), "cli": ("cli", "main"), @@ -462,7 +444,7 @@ def validate_policy(policy: dict[str, Any]) -> None: "pull_request_target_exceptions", }: raise PolicyError("qualification policy must declare the complete workflow trust contract") - if workflow_trust["pull_request_target_exceptions"] not in ([], [PROTECTED_POLICY_EXCEPTION]): + if workflow_trust["pull_request_target_exceptions"] != []: raise PolicyError("pull_request_target exceptions require a separately reviewed policy revision") artifact_handoffs = workflow_trust["privileged_artifact_handoffs"] if not isinstance(artifact_handoffs, dict) or set(artifact_handoffs) != { @@ -708,69 +690,51 @@ def _is_action_step(step: object, repository: str) -> bool: ) -def _reviewed_preflight_action_options(step: object) -> bool: - return ( - isinstance(step, dict) - and not set(step) - {"name", "id", "uses", "with", "env", "if", "continue-on-error"} - and isinstance(step.get("if", ""), str) - and _condition_without_expression_wrapper(step.get("if", "")) - in {"", "github.server_url == 'https://github.com'"} - and step.get("continue-on-error") in (None, False, "false") - ) - - def _verify_action_policy_preflight( target_name: str, label: str, jobs: dict[str, Any], gate_job_name: str, approved_bootstraps: set[str] | None = None, - locked_dependency_required: bool = False, - workflow_document: Mapping[str, Any] | None = None, ) -> None: preflight = jobs.get("action-policy") if not isinstance(preflight, dict): raise PolicyError(f"{label} has no central action policy preflight job") - workflow_context = workflow_document or {} - _verify_preflight_execution_context(workflow_context, preflight, label) accepted_conditions = {"", "github.server_url == 'https://github.com'"} if ( preflight.get("name") != "Central action policy preflight" or _condition_without_expression_wrapper(preflight.get("if", "")) not in accepted_conditions or preflight.get("runs-on") != "ubuntu-latest" or preflight.get("timeout-minutes") != "5" + or preflight.get("needs") is not None + or preflight.get("permissions") is not None + or preflight.get("env") is not None ): raise PolicyError(f"{label} central action policy preflight job is not the reviewed read-only shape") steps = preflight.get("steps") - if locked_dependency_required and (not isinstance(steps, list) or len(steps) != 6): - raise PolicyError(f"{label} central action policy preflight requires the locked-policy pip bootstrap") if isinstance(steps, list) and len(steps) == 6: bootstrap = steps[3] revisions = approved_bootstraps if approved_bootstraps is not None else {"ff93e15dd3d3aa0b5ba8991fe5101a9e1be49c10"} references = {"durable-workflow/.github/.github/actions/bootstrap-patched-pip@" + sha for sha in revisions} if ( - not _reviewed_preflight_action_options(bootstrap) - or "with" in bootstrap + not isinstance(bootstrap, dict) + or set(bootstrap) - {"name", "uses"} or bootstrap.get("uses") not in references ): raise PolicyError(f"{label} central action policy preflight has an unreviewed pip bootstrap") - if locked_dependency_required and not bootstrap["uses"].endswith("@" + LOCKED_POLICY_BOOTSTRAP_REVISION): - raise PolicyError(f"{label} central action policy preflight requires the locked-policy pip bootstrap") steps = steps[:3] + steps[4:] if not isinstance(steps, list) or len(steps) != 5 or not all(isinstance(step, dict) for step in steps): raise PolicyError(f"{label} central action policy preflight must contain the five reviewed steps") candidate, central, setup_python, install, validate = steps if ( not _is_action_step(candidate, "actions/checkout") - or not _reviewed_preflight_action_options(candidate) or candidate.get("with") != {"persist-credentials": "false"} ): raise PolicyError(f"{label} central action policy preflight must check out credential-free candidate source") if ( not _is_action_step(central, "actions/checkout") - or not _reviewed_preflight_action_options(central) or central.get("with") != { "repository": CENTRAL_ACTION_REPOSITORY, @@ -782,38 +746,22 @@ def _verify_action_policy_preflight( raise PolicyError(f"{label} central action policy preflight must use protected central policy source") if ( not _is_action_step(setup_python, "actions/setup-python") - or not _reviewed_preflight_action_options(setup_python) or setup_python.get("with") != {"python-version": "3.13"} ): raise PolicyError(f"{label} central action policy preflight must use the reviewed Python runtime") reviewed_install_commands = { "python -m pip install PyYAML==6.0.2", "python -m pip install --only-binary :all: PyYAML==6.0.2", - "python -I -m pip install PyYAML==6.0.2", - "python -I -m pip install --only-binary :all: PyYAML==6.0.2", - LOCKED_POLICY_INSTALL_COMMAND, } - if locked_dependency_required: - reviewed_install_commands = {LOCKED_POLICY_INSTALL_COMMAND} - if set(install) - {"name", "run", "working-directory", "shell", "env"} or _normalized_shell(install.get("run")) not in reviewed_install_commands: + if set(install) - {"name", "run"} or _normalized_shell(install.get("run")) not in reviewed_install_commands: raise PolicyError(f"{label} central action policy preflight must install the reviewed policy dependency") expected_validation = ( "python .central-action-policy/scripts/qualification_policy.py validate " f"--policy .central-action-policy/qualification/policy.json --target {target_name} " "--workflow-directory .github/workflows" ) - if set(validate) - {"name", "run", "working-directory", "shell", "env"} or _normalized_shell(validate.get("run")) != expected_validation: + if set(validate) - {"name", "run"} or _normalized_shell(validate.get("run")) != expected_validation: raise PolicyError(f"{label} central action policy preflight does not invoke the reviewed validator") - if not all( - _runs_at_workspace_root(workflow_context, preflight, step) - for step in (install, validate) - ): - raise PolicyError(f"{label} central action policy install and validator must run from the workspace root") - if not all( - _uses_reviewed_run_shell(workflow_context, preflight, step) - for step in (install, validate) - ): - raise PolicyError(f"{label} central action policy install and validator require a reviewed shell") gate = jobs[gate_job_name] if "action-policy" not in _job_needs(gate, f"{label} required check gate"): @@ -828,10 +776,8 @@ def _verify_action_policy_preflight( if not any( isinstance(step, dict) and _condition_without_expression_wrapper(step.get("if", "")) in accepted_conditions - and step.get("continue-on-error") in (None, False, "false") and step.get("env") == reviewed_gate["env"] and _normalized_shell(step.get("run")) == reviewed_gate["run"] - and _uses_reviewed_run_shell(workflow_context, gate, step) for step in gate.get("steps") or [] ): raise PolicyError(f"{label} required check does not fail closed on the central action policy result") @@ -877,11 +823,7 @@ def verify_workflow_source(name: str, branch: str, workflow: dict[str, Any], sou f"matching jobs={stable_jobs}" ) if workflow.get("action_policy_preflight") is True: - _verify_action_policy_preflight( - name, label, jobs, stable_jobs[0], approved_bootstraps, - workflow.get("locked_policy_dependency_required", False), - workflow_document=document, - ) + _verify_action_policy_preflight(name, label, jobs, stable_jobs[0], approved_bootstraps) def _parse_yaml(source: str, label: str) -> Any: @@ -891,33 +833,23 @@ def _parse_yaml(source: str, label: str) -> Any: raise PolicyError(f"cannot parse {label} as YAML: {error}") from error -def _walk_yaml_nodes(document: Any) -> Iterator[Any]: - pending = [document] - visited: set[int] = set() - - # Shared YAML aliases must not multiply traversal work. - while pending: - node = pending.pop() - if not isinstance(node, (dict, list)): - yield node - continue - if id(node) in visited: - continue - visited.add(id(node)) - yield node - children = list(node.values()) if isinstance(node, dict) else node - pending.extend(reversed(children)) - - def _workflow_action_references(source: str, label: str) -> list[str]: + document = _parse_yaml(source, label) references: list[str] = [] - for node in _walk_yaml_nodes(_parse_yaml(source, label)): - if isinstance(node, dict) and "uses" in node: - reference = node["uses"] - if not isinstance(reference, str): - raise PolicyError(f"{label} has a non-string action reference") - references.append(reference) + def visit(node: Any) -> None: + if isinstance(node, dict): + for key, value in node.items(): + if key == "uses": + if not isinstance(value, str): + raise PolicyError(f"{label} has a non-string action reference") + references.append(value) + visit(value) + elif isinstance(node, list): + for value in node: + visit(value) + + visit(document) return references @@ -991,7 +923,16 @@ def _job_needs(job: dict[str, Any], label: str) -> set[str]: def _string_values(value: Any) -> list[str]: - return [node for node in _walk_yaml_nodes(value) if isinstance(node, str)] + values: list[str] = [] + if isinstance(value, dict): + for nested in value.values(): + values.extend(_string_values(nested)) + elif isinstance(value, list): + for nested in value: + values.extend(_string_values(nested)) + elif isinstance(value, str): + values.append(value) + return values def _references_secret_context(value: Any) -> bool: @@ -1462,36 +1403,6 @@ def _run_defaults(scope: Mapping[str, Any]) -> Mapping[str, Any]: return run if isinstance(run, dict) else {} -def _effective_run_setting( - workflow: Mapping[str, Any], job: Mapping[str, Any], step: Mapping[str, Any], - key: str, default: Any, -) -> Any: - for scope in (step, _run_defaults(job), _run_defaults(workflow)): - if key in scope: - return scope[key] - return default - - -def _runs_at_workspace_root( - workflow: Mapping[str, Any], job: Mapping[str, Any], step: Mapping[str, Any] -) -> bool: - directory = _effective_run_setting(workflow, job, step, "working-directory", ".") - return isinstance(directory, str) and ( - directory in {".", "./"} - or re.fullmatch(r"\$\{\{\s*github\.workspace\s*\}\}(?:/\.?)?", directory) is not None - ) - - -def _uses_reviewed_run_shell( - workflow: Mapping[str, Any], job: Mapping[str, Any], step: Mapping[str, Any] -) -> bool: - # Keep the runner default and built-in POSIX script invocation. An arbitrary - # template can ignore the script or change its directory before running it. - runner_default = object() - shell = _effective_run_setting(workflow, job, step, "shell", runner_default) - return shell is runner_default or isinstance(shell, str) and shell in {"bash", "sh"} - - def _uses_only_allowed_environment( scopes: tuple[Mapping[str, Any], ...], allowed_environment: object ) -> bool: @@ -1507,48 +1418,6 @@ def _uses_only_allowed_environment( return True -def _preflight_environment_is_reviewed( - workflow: Mapping[str, Any], job: Mapping[str, Any], step: Mapping[str, Any] -) -> bool: - scopes = (workflow, job, step) - allowed = list(PREFLIGHT_CI_ENVIRONMENT | PREFLIGHT_CLEAR_STARTUP_ENVIRONMENT) - if not _uses_only_allowed_environment(scopes, allowed): - return False - for scope in scopes: - environment = scope.get("env") or {} - if not all( - value == "" if key in PREFLIGHT_CLEAR_STARTUP_ENVIRONMENT - else isinstance(value, (str, bool, int, float)) - for key, value in environment.items() - ): - return False - return True - - -def _verify_preflight_execution_context( - workflow: Mapping[str, Any], job: Mapping[str, Any], label: str -) -> None: - allowed = {"name", "if", "runs-on", "timeout-minutes", "steps", "defaults", "env"} - if set(job) - allowed: - raise PolicyError(f"{label} central action policy preflight job is not the reviewed read-only shape") - for scope in (workflow, job): - defaults = scope.get("defaults") - if defaults is None: - continue - if not isinstance(defaults, dict) or set(defaults) - {"run"}: - raise PolicyError(f"{label} central action policy preflight has unreviewed defaults") - run = defaults.get("run", {}) - if not isinstance(run, dict) or set(run) - {"shell", "working-directory"}: - raise PolicyError(f"{label} central action policy preflight has unreviewed run defaults") - steps = job.get("steps") - if ( - isinstance(steps, list) - and all(isinstance(step, dict) for step in steps) - and not all(_preflight_environment_is_reviewed(workflow, job, step) for step in steps) - ): - raise PolicyError(f"{label} central action policy preflight environment is not reviewed") - - def _trusted_validator_action_step(step: object, expected_step: object) -> bool: if ( not isinstance(step, dict) @@ -1794,17 +1663,6 @@ def output_name_for(expected: str) -> str | None: ) -def _reviewed_policy_workflow( - policy: dict[str, Any], target_name: str, workflow_path: str, source: str, -) -> bool: - return ( - policy["workflow_trust"]["pull_request_target_exceptions"] == [PROTECTED_POLICY_EXCEPTION] - and target_name == PROTECTED_POLICY_EXCEPTION["target"] - and Path(workflow_path).parts[-3:] == Path(PROTECTED_POLICY_EXCEPTION["path"]).parts - and hashlib.sha256(source.encode("utf-8")).hexdigest() == PROTECTED_POLICY_EXCEPTION["sha256"] - ) - - TAG_DISPATCH_BINDING_FIELDS = { "environment", "jobs", @@ -1999,7 +1857,7 @@ def _scan_workflow_trust( triggers = document.get("on") if not isinstance(triggers, dict): raise PolicyError(f"{label} must declare explicit workflow triggers") - if "pull_request_target" in triggers and not _reviewed_policy_workflow(policy, target_name, workflow_path, source): + if "pull_request_target" in triggers: raise PolicyError(f"{label} uses forbidden pull_request_target execution") top_writes = _permissions_with_write(document.get("permissions"), f"{label} top-level") @@ -2384,7 +2242,6 @@ def audit_action_releases( COMPOSER_LOCAL_COMMANDS = {"about", "help", "list"} TRUST_ACTION_PREFIX = "durable-workflow/.github/.github/actions/" COMPOSER_TRUST_FIELDS = {"SSL_CERT_FILE", "SSL_CERT_DIR", "CURL_CA_BUNDLE", "GIT_SSL_CAINFO", "GIT_SSL_CAPATH", "GIT_SSL_NO_VERIFY", "COMPOSER_CAFILE", "COMPOSER_CAPATH", "COMPOSER_DISABLE_TLS", "COMPOSER", "COMPOSER_HOME", "DW_CI_CA_SHA256", "PIP_CERT", "REQUESTS_CA_BUNDLE"} -COMPOSER_HELPER_PATTERN = re.compile(r"(?:\./)?(?:scripts|\.github)/[A-Za-z0-9_./-]+") def _composer_subcommand_needs_network(arguments: list[str]) -> bool: @@ -2425,45 +2282,17 @@ def _is_composer_shell_helper(path: Path, contents: str) -> bool: return path.suffix in {"", ".sh", ".bash", ".ksh", ".zsh"} or re.match(r"^#![^\n]*\b(?:bash|sh|dash|ksh|zsh)\b", contents) is not None -def _shell_changes_working_directory(source: str) -> bool: - try: - lexer = shlex.shlex(source, posix=True, punctuation_chars=";&|()\n") - lexer.whitespace = " \t\r" - lexer.whitespace_split = True - return any( - token in {"cd", "pushd", "popd"} - or re.search(r"(?:^|[;&|()\n])\s*(?:cd|pushd|popd)(?:\s|$)", token) is not None - for token in lexer - ) - except ValueError: - # Shell helpers with an unprovable directory context are not evidence. - return True - - -def _composer_helper_references(source: str) -> list[str]: - # Explicit non-shell interpreters keep their existing program semantics. - # Remove only that invocation, so a following shell helper is still seen. - non_shell = r"(? str: contents = [source] visited = set() if visited is None else visited - helpers = _composer_helper_references(source) - if helpers and _shell_changes_working_directory(source): - raise PolicyError("Composer relative helper inspection requires an unchanged workspace root") - for name in helpers: + for name in re.findall(r"(?:\./)?(?:scripts|\.github)/[A-Za-z0-9_./-]+", source): path = (root / name).resolve() if not path.is_relative_to(root.resolve()): raise PolicyError("Composer source contract must not read outside the candidate repository") if path.is_file() and path not in visited: visited.add(path) helper = path.read_text(encoding="utf-8") - shell_invocation = re.search( - r"\b(?:bash|sh|dash|ksh|zsh)\s+(?:-[\w-]+\s+)*" + re.escape(name) + r"(?=\s|[;&|()]|$)", source, - ) - if _is_composer_shell_helper(path, helper) or shell_invocation: + if _is_composer_shell_helper(path, helper): contents.append(_composer_step_source(helper, root, visited)) return "\n".join(contents) @@ -2514,22 +2343,14 @@ def _require_composer_request_trust(step: dict[str, Any], state: tuple[bool, boo raise PolicyError(f"{label} changes Composer trust inside its request step or helper") -def _composer_job_sinks( - job: dict[str, Any], root: Path, approved: set[str], local: bool, label: str, - workflow: Mapping[str, Any], -) -> list[str]: +def _composer_job_sinks(job: dict[str, Any], root: Path, approved: set[str], local: bool, label: str) -> list[str]: if set(job.get("env") or {}) & COMPOSER_TRUST_FIELDS: raise PolicyError(f"{label} declares unverified job-level Composer trust overrides") evidence = [] state = (False, False, False) for index, step in enumerate(job.get("steps") or []): step_label = f"{label}:step-{index + 1}" - source = step.get("run", "") - if _composer_helper_references(source) and not ( - _runs_at_workspace_root(workflow, job, step) and _uses_reviewed_run_shell(workflow, job, step) - ): - raise PolicyError(f"{step_label} Composer relative helper inspection requires the workspace root and a reviewed shell") - expanded = _composer_step_source(source, root) + expanded = _composer_step_source(step.get("run", ""), root) if _composer_network_commands(expanded): _require_composer_request_trust(step, state, step_label, expanded) evidence.append(step_label) @@ -2551,7 +2372,7 @@ def verify_composer_network_sinks(policy: dict[str, Any], target: str, sources: if set(document.get("env") or {}) & COMPOSER_TRUST_FIELDS: raise PolicyError(f"{filename} declares unverified workflow-level Composer trust overrides") for job_name, job in (document.get("jobs") or {}).items(): - evidence.extend(_composer_job_sinks(job, root, approved, target == "github-control-plane", f"{filename}:{job_name}", document)) + evidence.extend(_composer_job_sinks(job, root, approved, target == "github-control-plane", f"{filename}:{job_name}")) catalog = json.loads((Path(__file__).resolve().parents[1] / "qualification/composer-container-trust.json").read_text()) evidence.extend(verify_composer_container_callers(target, sources, approved, catalog, root)) return evidence @@ -2647,10 +2468,8 @@ def verify_composer_container_callers(target: str, sources: dict[str, str], appr def validate_local_action_references( policy: dict[str, Any], directory: Path, target_name: str = "github-control-plane" ) -> list[str]: - # The reviewed workflow exception hashes Git blob bytes. Text-mode reads - # normalize line endings and would admit unreviewed byte variants. sources = { - path.as_posix(): path.read_bytes().decode("utf-8") for path in sorted(directory.glob("*.y*ml")) if path.is_file() + path.as_posix(): path.read_text(encoding="utf-8") for path in sorted(directory.glob("*.y*ml")) if path.is_file() } evidence = scan_workflow_sources(policy, target_name, sources) verify_composer_network_sinks(policy, target_name, sources, directory.parent.parent) diff --git a/tests/test_composer_trust_contract.py b/tests/test_composer_trust_contract.py index 479b969..a05cb86 100644 --- a/tests/test_composer_trust_contract.py +++ b/tests/test_composer_trust_contract.py @@ -142,12 +142,9 @@ def setUp(self): {"run": "composer install --prefer-dist"}, ] - def check(self, steps=None, jobs=None, defaults=None): + def check(self, steps=None, jobs=None): jobs = jobs or {"test": {"steps": self.steps if steps is None else steps}} - document = {"jobs": jobs} - if defaults is not None: - document["defaults"] = defaults - source = yaml.safe_dump(document) + source = yaml.safe_dump({"jobs": jobs}) return verify_composer_network_sinks(self.policy, "fixture", {"ci.yml": source}, self.root) def test_reviewed_order_is_enumerated(self): @@ -250,119 +247,6 @@ def test_extensionless_and_nested_repository_helpers_are_inspected(self): self.check([{"run": "scripts/build"}]) self.assertEqual(1, len(self.check([*self.steps[:3], {"run": "scripts/build"}]))) - def relative_helper_fixture(self): - (self.root / "scripts").mkdir() - (self.root / "scripts/request.sh").write_text("#!/bin/sh\nprintf done\n") - (self.root / "alternate/scripts").mkdir(parents=True) - (self.root / "alternate/scripts/request.sh").write_text("#!/bin/sh\ncomposer install\n") - - def test_relative_helpers_cannot_be_redirected_by_run_directories(self): - self.relative_helper_fixture() - for scope in ("step", "job", "workflow"): - step = {"run": "scripts/request.sh"} - job = {"steps": [step]} - defaults = {"run": {"working-directory": "alternate"}} - workflow_defaults = None - if scope == "step": - step["working-directory"] = "alternate" - elif scope == "job": - job["defaults"] = defaults - else: - workflow_defaults = defaults - with self.subTest(scope=scope), self.assertRaisesRegex(PolicyError, "workspace root"): - self.check(jobs={"test": job}, defaults=workflow_defaults) - - def test_relative_helpers_refuse_directory_changes_in_shell_source(self): - self.relative_helper_fixture() - for command in ( - "cd alternate\nscripts/request.sh", - "pushd alternate; scripts/request.sh", - "builtin cd alternate && scripts/request.sh", - "bash -c 'cd alternate; scripts/request.sh'", - ): - with self.subTest(command=command), self.assertRaisesRegex(PolicyError, "workspace root"): - self.check([{"run": command}]) - (self.root / "scripts/outer.sh").write_text("cd alternate\nscripts/request.sh\n") - with self.assertRaisesRegex(PolicyError, "workspace root"): - self.check([{"run": "scripts/outer.sh"}]) - - def test_explicit_workspace_roots_keep_guarded_helper_requests(self): - (self.root / "scripts").mkdir() - (self.root / "scripts/request.sh").write_text("composer install\n") - for directory in (".", "./", "${{ github.workspace }}", "${{github.workspace}}/."): - for scope in ("step", "job", "workflow"): - step = {"run": "scripts/request.sh"} - job = {"steps": [*self.steps[:3], step]} - defaults = {"run": {"working-directory": directory}} - workflow_defaults = None - if scope == "step": - step["working-directory"] = directory - elif scope == "job": - job["defaults"] = defaults - else: - workflow_defaults = defaults - with self.subTest(scope=scope, directory=directory): - self.assertEqual(1, len(self.check(jobs={"test": job}, defaults=workflow_defaults))) - - def test_helper_directory_uses_the_most_specific_run_setting(self): - self.relative_helper_fixture() - step = {"run": "scripts/request.sh", "working-directory": "."} - job = {"steps": [step], "defaults": {"run": {"working-directory": "alternate"}}} - self.assertEqual([], self.check(jobs={"test": job})) - del step["working-directory"] - job["defaults"]["run"]["working-directory"] = "${{ github.workspace }}" - self.assertEqual([], self.check(jobs={"test": job}, defaults={"run": {"working-directory": "alternate"}})) - - def test_non_helper_commands_keep_their_existing_directory_behavior(self): - self.assertEqual([], self.check([{"run": "cargo test", "working-directory": "alternate"}])) - self.assertEqual([], self.check([{"run": "cd alternate\ncargo test"}])) - - def test_non_shell_programs_keep_their_existing_directory_behavior(self): - for command in ("python3 scripts/check.py", "python -I scripts/check.py", "node scripts/check.mjs", "php scripts/check.php"): - with self.subTest(command=command): - self.assertEqual([], self.check([{"run": command, "working-directory": "alternate"}])) - with self.assertRaisesRegex(PolicyError, "workspace root"): - self.check([{"run": "python3 scripts/check.py; scripts/request.sh", "working-directory": "alternate"}]) - for command in ("custom-python3 scripts/request.sh", "scripts/python3 scripts/request.sh"): - with self.subTest(command=command), self.assertRaisesRegex(PolicyError, "workspace root"): - self.check([{"run": command, "working-directory": "alternate"}]) - - def test_shell_invocation_keeps_helper_inspection_despite_its_extension(self): - (self.root / "scripts").mkdir() - (self.root / "scripts/request.py").write_text("composer install\n") - with self.assertRaisesRegex(PolicyError, "workspace root"): - self.check([{"run": "bash scripts/request.py", "working-directory": "alternate"}]) - with self.assertRaisesRegex(PolicyError, "lack the reviewed"): - self.check([{"run": "bash scripts/request.py"}]) - self.assertEqual(1, len(self.check([*self.steps[:3], {"run": "bash scripts/request.py"}]))) - - def test_relative_helpers_refuse_custom_shell_context(self): - self.relative_helper_fixture() - for scope in ("step", "job", "workflow"): - step = {"run": "scripts/request.sh"} - job = {"steps": [step]} - shell = "bash -c 'cd alternate; bash {0}' ignored" - defaults = {"run": {"shell": shell}} - workflow_defaults = None - if scope == "step": - step["shell"] = shell - elif scope == "job": - job["defaults"] = defaults - else: - workflow_defaults = defaults - with self.subTest(scope=scope), self.assertRaisesRegex(PolicyError, "reviewed shell"): - self.check(jobs={"test": job}, defaults=workflow_defaults) - - def test_helper_shell_uses_the_most_specific_safe_setting(self): - self.relative_helper_fixture() - for shell in ("bash", "sh"): - step = {"run": "scripts/request.sh", "shell": shell} - job = {"steps": [step], "defaults": {"run": {"shell": "bash -c true {0}"}}} - self.assertEqual([], self.check(jobs={"test": job})) - del step["shell"] - job["defaults"]["run"]["shell"] = shell - self.assertEqual([], self.check(jobs={"test": job}, defaults={"run": {"shell": "bash -c true {0}"}})) - def test_job_and_workflow_overrides_cannot_bypass_the_step_guards(self): with self.assertRaisesRegex(PolicyError, "job-level"): self.check(jobs={"test": {"env": {"GIT_SSL_NO_VERIFY": "1"}, "steps": self.steps}}) diff --git a/tests/test_policy_dependency_install.py b/tests/test_policy_dependency_install.py deleted file mode 100644 index e3a2092..0000000 --- a/tests/test_policy_dependency_install.py +++ /dev/null @@ -1,87 +0,0 @@ -"""Qualify the policy parser wheels in an isolated, task-owned Python venv.""" - -import os -from pathlib import Path -import re -import shutil -import subprocess -import sys -import tempfile -import unittest - - -ROOT = Path(__file__).resolve().parents[1] -REQUIREMENTS = ROOT / "qualification/requirements.lock" -RELATIVE_REQUIREMENTS = ".central-action-policy/qualification/requirements.lock" -INSTALL_FLAGS = ["--require-hashes", "--only-binary=:all:", "--no-deps"] - - -class PolicyDependencyInstallTest(unittest.TestCase): - def setUp(self): - import pip - - self.assertNotEqual(sys.prefix, sys.base_prefix, "Use a disposable venv") - self.assertEqual("26.2.2", pip.__version__) - self.temporary = tempfile.TemporaryDirectory() - self.addCleanup(self.temporary.cleanup) - self.directory = Path(self.temporary.name) - self.environment = dict(os.environ) - for name in ("PYTHONPATH", "PYTHONSAFEPATH", "PIP_EXTRA_INDEX_URL", "PIP_TRUSTED_HOST"): - self.environment.pop(name, None) - self.environment.update({ - "PIP_CONFIG_FILE": os.devnull, - "PIP_INDEX_URL": "https://pypi.org/simple", - "PIP_DISABLE_PIP_VERSION_CHECK": "1", - "PIP_RETRIES": "0", - "PIP_DEFAULT_TIMEOUT": "15", - }) - - def pip(self, arguments, directory=None, isolated=True): - return subprocess.run( - [sys.executable, *(["-I"] if isolated else []), "-m", "pip", - *(["--isolated"] if isolated else []), *arguments], - cwd=directory or self.directory, env=self.environment, - capture_output=True, text=True, timeout=60, - ) - - def test_actual_installer_ignores_both_working_directory_shadows(self): - for kind in ("module", "package"): - with self.subTest(shadow=kind): - removed = self.pip(["uninstall", "--yes", "PyYAML"]) - self.assertEqual(0, removed.returncode, removed.stdout + removed.stderr) - directory = self.directory / kind - directory.mkdir() - lock = directory / RELATIVE_REQUIREMENTS - lock.parent.mkdir(parents=True) - shutil.copyfile(REQUIREMENTS, lock) - module = directory / "pip.py" - if kind == "package": - module = directory / "pip/__init__.py" - module.parent.mkdir() - module.write_text("from pathlib import Path\nPath('shadow-ran').write_text('executed')\nraise SystemExit(29)\n") - raw = self.pip(["--version"], directory, isolated=False) - self.assertEqual(29, raw.returncode, raw.stderr) - marker = directory / "shadow-ran" - self.assertTrue(marker.is_file()) - marker.unlink() - installed = self.pip(["install", *INSTALL_FLAGS, "-r", RELATIVE_REQUIREMENTS], directory) - self.assertEqual(0, installed.returncode, installed.stdout + installed.stderr) - self.assertFalse(marker.exists()) - loaded = subprocess.run( - [sys.executable, "-I", "-c", "import sys,yaml; from pathlib import Path; assert yaml.__version__ == '6.0.2'; assert yaml.__with_libyaml__; assert Path(yaml.__file__).is_relative_to(sys.prefix)"], - cwd=directory, env=self.environment, capture_output=True, text=True, timeout=10, - ) - self.assertEqual(0, loaded.returncode, loaded.stderr) - - def test_a_tampered_wheel_hash_cannot_download(self): - lock = self.directory / "tampered.lock" - lock.write_text(re.sub(r"sha256:[0-9a-f]{64}", "sha256:" + "0" * 64, REQUIREMENTS.read_text())) - destination = self.directory / "rejected" - result = self.pip(["download", *INSTALL_FLAGS, "--no-cache-dir", "--dest", str(destination), "-r", str(lock)]) - self.assertNotEqual(0, result.returncode) - self.assertIn("DO NOT MATCH THE HASHES", result.stdout + result.stderr) - self.assertEqual([], list(destination.glob("*.whl"))) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_protected_policy_check.py b/tests/test_protected_policy_check.py deleted file mode 100644 index f946ec9..0000000 --- a/tests/test_protected_policy_check.py +++ /dev/null @@ -1,286 +0,0 @@ -import contextlib -import hashlib -import io -import json -from pathlib import Path -import subprocess -import sys -import tempfile -import unittest -from unittest.mock import patch - -from scripts import protected_policy_check as gate -from scripts.protected_policy_inputs import InputRefused -from tests import test_protected_policy_inputs as fixtures - - -class FakeCurl: - def __init__(self): - self.version = b"curl 8.5.0 test\n" - self.body = b'{"ok":true}' - self.status = b"200" - self.exit_code = 0 - self.timeout = False - self.calls = [] - self.after_request = lambda: None - - def __call__(self, command, **options): - if "--version" in command: - return subprocess.CompletedProcess(command, 0, self.version) - self.calls.append((command, options)) - if self.timeout: - raise subprocess.TimeoutExpired(command, options["timeout"]) - Path(command[command.index("--output") + 1]).write_bytes(self.body) - self.after_request() - return subprocess.CompletedProcess(command, self.exit_code, self.status) - - -class GitHubReadClientTest(unittest.TestCase): - def setUp(self): - self.temporary = tempfile.TemporaryDirectory() - self.addCleanup(self.temporary.cleanup) - self.directory = Path(self.temporary.name) - self.bundle = self.directory / "reviewed.pem" - # A synthetic digest fixture. No TLS or network connection is made here. - self.bundle.write_bytes(b"synthetic reviewed bundle") - self.digest = hashlib.sha256(self.bundle.read_bytes()).hexdigest() - self.roots = self.directory / "no-additional-roots" - self.roots.mkdir() - self.curl = FakeCurl() - self.now = 0.0 - - def client(self, token=""): - return gate.GitHubReadClient(self.bundle, self.roots, self.digest, token, - run=self.curl, clock=lambda: self.now) - - def read(self, client=None): - return (client or self.client())("/repos/durable-workflow/.github/pulls/7") - - def test_explicit_tls_and_data_only_get_use_no_redirect_or_decompression(self): - self.assertEqual({"ok": True}, self.read()) - command, options = self.curl.calls[0] - self.assertEqual(["/usr/bin/curl", "--disable"], command[:2]) - self.assertEqual("=https", command[command.index("--proto") + 1]) - self.assertEqual(str(self.bundle), command[command.index("--cacert") + 1]) - self.assertEqual(str(self.roots), command[command.index("--capath") + 1]) - self.assertEqual(str(gate.MAX_RESPONSE_BYTES), command[command.index("--max-filesize") + 1]) - self.assertIn("Accept-Encoding: identity", command) - self.assertNotIn("--compressed", command) - self.assertNotIn("--location", command) - self.assertNotIn("--insecure", command) - self.assertLessEqual(options["timeout"], gate.REQUEST_SECONDS) - - def test_read_token_is_passed_only_on_stdin_not_process_arguments(self): - self.read(self.client("synthetic_read_token")) - command, options = self.curl.calls[0] - self.assertNotIn("synthetic_read_token", " ".join(command)) - self.assertEqual(b'header = "Authorization: Bearer synthetic_read_token"\n', options["input"]) - - def test_unsafe_or_write_endpoints_refuse_before_any_http_call(self): - client = self.client() - for path in ("https://elsewhere.test/", "/repos/durable-workflow/.github/check-runs", - "/repos/../.github/pulls/7", "/repos/a/b/pulls/7?redirect=elsewhere", - "/repos/a/b/git/blobs/" + "a" * 39): - with self.subTest(path=path), self.assertRaises(InputRefused): - client(path) - self.assertEqual([], self.curl.calls) - - def test_newline_or_quote_in_read_token_refuses(self): - for token in ('bad"token', "bad\nheader", "bad\\config"): - with self.subTest(token=token), self.assertRaises(InputRefused): - self.client(token) - - def test_old_curl_cannot_claim_unknown_length_response_protection(self): - for version in (b"curl 8.3.0\n", b"curl 7.88.1\n", b"unknown\n"): - self.curl.version = version - with self.subTest(version=version), self.assertRaises(InputRefused): - self.client() - - def test_wrong_bundle_and_additive_roots_refuse_before_requests(self): - self.bundle.write_bytes(b"different") - with self.assertRaises(InputRefused): - self.client() - self.bundle.write_bytes(b"synthetic reviewed bundle") - (self.roots / "extra.pem").write_bytes(b"extra") - with self.assertRaises(InputRefused): - self.client() - - def test_bundle_symlink_refuses(self): - real = self.directory / "real.pem" - self.bundle.rename(real) - self.bundle.symlink_to(real) - with self.assertRaises(InputRefused): - self.client() - - def test_redirect_and_failed_status_never_qualify(self): - for status, exit_code in ((b"302", 0), (b"404", 22), (b"500", 22)): - self.curl.status, self.curl.exit_code = status, exit_code - with self.subTest(status=status), self.assertRaises(InputRefused): - self.read() - - def test_malformed_duplicate_and_nonfinite_json_refuse(self): - for body in (b"not json", b'{"a":1,"a":2}', b'{"x":NaN}', b'{"x":Infinity}', b"\xff", - '{"ok":true}'.encode("utf-16"), b"[]", - b'{"nested":' + b"[" * 2000 + b"]" * 2000 + b"}"): - self.curl.body = body - with self.subTest(size=len(body), prefix=body[:32]), self.assertRaises(InputRefused): - self.read() - - def test_json_collection_bound_is_independent_of_the_transfer_limit(self): - self.curl.body = json.dumps({"items": list(range(100))}).encode() - with patch.object(gate, "MAX_JSON_NODES", 32), self.assertRaises(InputRefused): - self.read() - - def test_oversized_body_refuses_even_if_a_transport_ignores_its_limit(self): - with patch.object(gate, "MAX_RESPONSE_BYTES", 64): - self.curl.body = b"x" * 65 - with self.assertRaisesRegex(InputRefused, "byte bound"): - self.read() - - def test_transport_timeout_is_a_refusal_without_echoing_credentials(self): - self.curl.timeout = True - client = self.client("synthetic_read_token") - with self.assertRaises(InputRefused) as error: - self.read(client) - self.assertNotIn("synthetic_read_token", str(error.exception)) - - def test_total_deadline_and_request_count_do_not_start_another_transfer(self): - client = self.client() - self.now = gate.TOTAL_SECONDS + 1 - with self.assertRaises(InputRefused): - self.read(client) - self.assertEqual([], self.curl.calls) - self.now = 0 - with patch.object(gate, "MAX_REQUESTS", 1): - client = self.client() - self.read(client) - with self.assertRaises(InputRefused): - self.read(client) - self.assertEqual(1, len(self.curl.calls)) - - def test_expiry_during_a_transfer_cannot_return_a_pass(self): - client = self.client() - self.curl.after_request = lambda: setattr(self, "now", gate.TOTAL_SECONDS) - with self.assertRaisesRegex(InputRefused, "deadline"): - self.read(client) - - def test_last_request_uses_only_the_remaining_time(self): - client = self.client() - self.now = gate.TOTAL_SECONDS - 0.5 - self.read(client) - command, options = self.curl.calls[0] - self.assertEqual(0.5, options["timeout"]) - self.assertEqual("0.5", command[command.index("--max-time") + 1]) - - -SAFE_WORKFLOW = b"""name: Candidate -on: - pull_request: -permissions: - contents: read -jobs: - check: - runs-on: ubuntu-latest - steps: - - run: python scripts/candidate.py -""" - - -class ProtectedPolicySnapshotTest(unittest.TestCase): - def setUp(self): - self.api = fixtures.ProtectedPolicyInputTest(methodName="runTest") - self.api.setUp() - self.api.pull()["base"]["repo"]["id"] = gate.PILOT_REPOSITORY_ID - self.branch = {"name": "main", "protected": True, "commit": {"sha": fixtures.BASE}} - self.api.records[f"/repos/{gate.PILOT_REPOSITORY}/branches/main"] = self.branch - self.api.entries.clear() - self.api.add_file(fixtures.WORKFLOW, SAFE_WORKFLOW) - self.api.add_file("scripts/candidate.py", b"raise SystemExit('candidate must not execute')\n", "100755") - - def validate(self, **options): - return gate.validate_snapshot(self.api.read, 7, fixtures.BASE, **options) - - def test_real_validator_reads_candidate_data_without_executing_it(self): - report = self.validate() - self.assertEqual(fixtures.HEAD, report["head_sha"]) - self.assertEqual(2, report["input_count"]) - self.assertEqual(gate.PILOT_REPOSITORY_ID, report["repository_id"]) - self.assertEqual(64, len(report["inputs_sha256"])) - self.assertNotIn("candidate must not execute", json.dumps(report)) - - def test_validator_gets_nonexecutable_files_and_temporary_data_is_retired(self): - from scripts.qualification_policy import validate_local_action_references - - observed = [] - - def inspect(policy, directory, target): - observed.append(directory.parent.parent) - helper = directory.parent.parent / "scripts/candidate.py" - self.assertEqual(0o600, helper.stat().st_mode & 0o777) - return validate_local_action_references(policy, directory, target) - - self.validate(validator=inspect) - self.assertFalse(observed[0].exists()) - - def test_forbidden_workflow_is_a_refusal_not_a_success_report(self): - self.api.entries.clear() - self.api.add_file(fixtures.WORKFLOW, SAFE_WORKFLOW.replace(b"pull_request:", b"pull_request_target:")) - with self.assertRaisesRegex(InputRefused, "forbidden"): - self.validate() - - def test_unprotected_main_refuses_before_candidate_blobs(self): - self.branch["protected"] = False - with self.assertRaises(InputRefused): - self.validate() - self.assertFalse(any("/git/blobs/" in request for request in self.api.requests)) - - def test_base_or_head_movement_after_validation_refuses(self): - def move_base(*_arguments): - self.branch["commit"]["sha"] = "d" * 40 - return [] - - with self.assertRaises(InputRefused): - self.validate(validator=move_base) - self.branch["commit"]["sha"] = fixtures.BASE - - def move_head(*_arguments): - self.api.pull()["head"]["sha"] = "d" * 40 - return [] - - with self.assertRaises(InputRefused): - self.validate(validator=move_head) - - def test_wrong_immutable_repository_id_refuses(self): - self.api.pull()["base"]["repo"]["id"] += 1 - with self.assertRaises(InputRefused): - self.validate() - - def test_error_output_cannot_become_a_github_workflow_command(self): - arguments = ["gate", "--pull-request", "7", "--base-sha", fixtures.BASE] - environment = {"CURL_CA_BUNDLE": "synthetic", "SSL_CERT_DIR": "synthetic", "DW_CI_CA_SHA256": "synthetic"} - output = io.StringIO() - with patch.object(sys, "argv", arguments), patch.dict("os.environ", environment, clear=True): - with patch.object(gate, "GitHubReadClient"), patch.object(gate, "validate_snapshot", side_effect=InputRefused("bad\n::error::candidate")): - with contextlib.redirect_stderr(output): - self.assertEqual(1, gate.main()) - self.assertEqual(1, len(output.getvalue().splitlines())) - self.assertEqual("bad\n::error::candidate", json.loads(output.getvalue())["policy_refused"]) - - def test_isolated_cli_ignores_candidate_python_modules_in_its_working_directory(self): - with tempfile.TemporaryDirectory() as temporary: - candidate = Path(temporary) / "scripts" - candidate.mkdir() - (candidate / "__init__.py").write_text("raise SystemExit('candidate imported')\n") - result = subprocess.run( - [sys.executable, "-I", str(gate.ROOT / "scripts/protected_policy_check.py"), "--help"], - cwd=temporary, capture_output=True, timeout=5, check=False, - ) - self.assertEqual(0, result.returncode, result.stderr.decode()) - self.assertNotIn(b"candidate imported", result.stderr) - - def test_cli_does_not_offer_a_candidate_policy_override(self): - arguments = ["gate", "--pull-request", "7", "--base-sha", fixtures.BASE, "--policy", "candidate.json"] - with patch.object(sys, "argv", arguments), contextlib.redirect_stderr(io.StringIO()): - with self.assertRaises(SystemExit) as error: - gate.main() - self.assertEqual(2, error.exception.code) diff --git a/tests/test_protected_policy_inputs.py b/tests/test_protected_policy_inputs.py deleted file mode 100644 index 5a9966f..0000000 --- a/tests/test_protected_policy_inputs.py +++ /dev/null @@ -1,251 +0,0 @@ -import base64 -import tempfile -import unittest -from copy import deepcopy -from pathlib import Path -from unittest.mock import patch - -from scripts.protected_policy_inputs import ( - MAX_FILE_BYTES, - MAX_FILES, - InputRefused, - checked_path, - checked_slug, - collect_inputs, - read_identity, - require_current, - write_inputs, -) - - -BASE = "a" * 40 -HEAD = "b" * 40 -TREE = "c" * 40 -REPOSITORY = "durable-workflow/.github" -FORK = "contributor/policy-change" -WORKFLOW = ".github/workflows/ci.yml" - - -class ProtectedPolicyInputTest(unittest.TestCase): - def setUp(self): - self.entries = [] - self.records = { - f"/repos/{REPOSITORY}/pulls/7": { - "state": "open", "number": 7, - "base": {"ref": "main", "sha": BASE, "repo": { - "id": 100, "full_name": REPOSITORY, "private": False, - }}, - "head": {"sha": HEAD, "repo": { - "id": 200, "full_name": FORK, "private": False, - }}, - }, - f"/repos/{FORK}/git/commits/{HEAD}": {"sha": HEAD, "tree": {"sha": TREE}}, - f"/repos/{FORK}/git/trees/{TREE}?recursive=1": { - "sha": TREE, "truncated": False, "tree": self.entries, - }, - } - self.requests = [] - self.add_file(WORKFLOW, b"on: push\njobs: {}\n") - self.identity = read_identity(self.read, REPOSITORY, 100, 7, BASE) - - def read(self, path): - self.requests.append(path) - return deepcopy(self.records[path]) - - def add_file(self, path, contents, mode="100644"): - sha = f"{len(self.entries) + 1:040x}" - entry = {"path": path, "sha": sha, "mode": mode, "type": "blob", "size": len(contents)} - self.entries.append(entry) - self.records[f"/repos/{FORK}/git/blobs/{sha}"] = { - "sha": sha, "size": len(contents), "encoding": "base64", - "content": base64.encodebytes(contents).decode("ascii"), - } - return entry - - def blob(self, entry=None): - entry = entry or self.entries[0] - return self.records[f"/repos/{FORK}/git/blobs/{entry['sha']}"] - - def pull(self): - return self.records[f"/repos/{REPOSITORY}/pulls/7"] - - def collect(self): - return collect_inputs(self.read, self.identity) - - def test_only_workflows_and_referenced_helpers_are_fetched_from_the_exact_fork_commit(self): - self.blob()["content"] = base64.encodebytes(b"run: scripts/one.sh\n").decode() - self.entries[0]["size"] = self.blob()["size"] = len(b"run: scripts/one.sh\n") - self.add_file("scripts/one.sh", b"#!/bin/sh\n./scripts/two.sh\n", "100755") - self.add_file("scripts/two.sh", b"#!/bin/sh\nexit 0\n", "100755") - unreferenced = self.add_file("pip.py", b"raise SystemExit('must never load')") - files = self.collect() - self.assertEqual({WORKFLOW, "scripts/one.sh", "scripts/two.sh"}, set(files)) - self.assertNotIn(f"/repos/{FORK}/git/blobs/{unreferenced['sha']}", self.requests) - self.assertIn(f"/repos/{FORK}/git/commits/{HEAD}", self.requests) - - def test_head_movement_refuses_the_result(self): - self.pull()["head"]["sha"] = "d" * 40 - with self.assertRaises(InputRefused): - require_current(self.read, self.identity) - - def test_base_movement_refuses_the_result(self): - self.pull()["base"]["sha"] = "d" * 40 - with self.assertRaises(InputRefused): - self.collect() - - def test_wrong_repository_id_refuses_before_candidate_reads(self): - self.pull()["base"]["repo"]["id"] = 101 - with self.assertRaises(InputRefused): - read_identity(self.read, REPOSITORY, 100, 7, BASE) - - def test_closed_pull_request_cannot_issue_a_result(self): - self.pull()["state"] = "closed" - with self.assertRaises(InputRefused): - require_current(self.read, self.identity) - - def test_private_repository_is_not_silently_onboarded(self): - self.pull()["base"]["repo"]["private"] = True - with self.assertRaises(InputRefused): - require_current(self.read, self.identity) - - def test_private_fork_is_not_silently_onboarded(self): - self.pull()["head"]["repo"]["private"] = True - with self.assertRaises(InputRefused): - require_current(self.read, self.identity) - - def test_external_base_repository_is_not_onboarded(self): - with self.assertRaises(InputRefused): - read_identity(self.read, "elsewhere/repository", 100, 7, BASE) - - def test_deleted_fork_refuses(self): - self.pull()["head"]["repo"] = None - with self.assertRaises(InputRefused): - require_current(self.read, self.identity) - - def test_unsafe_paths_and_repository_names_refuse(self): - for value in (".", "../escape", "/absolute", "a//b", "a/../b", "a\\b", "a\nb", "a\0b"): - with self.subTest(path=value), self.assertRaises(InputRefused): - checked_path(value) - for value in ("../repo", "owner/..", "https://github.com/owner/repo", "owner/repo/more", "x\ny/repo"): - with self.subTest(slug=value), self.assertRaises(InputRefused): - checked_slug(value) - - def test_symlink_workflow_refuses_without_fetching_its_contents(self): - self.entries[0]["mode"] = "120000" - with self.assertRaises(InputRefused): - self.collect() - self.assertFalse(any("/git/blobs/" in path for path in self.requests)) - - def test_referenced_symlink_helper_refuses(self): - contents = b"run: scripts/link.sh\n" - self.entries[0]["size"] = self.blob()["size"] = len(contents) - self.blob()["content"] = base64.encodebytes(contents).decode() - self.add_file("scripts/link.sh", b"../../outside", "120000") - with self.assertRaises(InputRefused): - self.collect() - - def test_submodule_cannot_supply_a_workflow(self): - self.entries[0].update({"mode": "160000", "type": "commit"}) - with self.assertRaises(InputRefused): - self.collect() - - def test_truncated_tree_refuses(self): - self.records[f"/repos/{FORK}/git/trees/{TREE}?recursive=1"]["truncated"] = True - with self.assertRaises(InputRefused): - self.collect() - - def test_another_commit_or_tree_cannot_supply_candidate_inputs(self): - commit = self.records[f"/repos/{FORK}/git/commits/{HEAD}"] - commit["sha"] = "d" * 40 - with self.assertRaises(InputRefused): - self.collect() - commit["sha"] = HEAD - self.records[f"/repos/{FORK}/git/trees/{TREE}?recursive=1"]["sha"] = "d" * 40 - with self.assertRaises(InputRefused): - self.collect() - - def test_duplicate_path_refuses(self): - self.entries.append(deepcopy(self.entries[0])) - with self.assertRaises(InputRefused): - self.collect() - - def test_oversized_file_refuses_before_blob_download(self): - self.entries[0]["size"] = MAX_FILE_BYTES + 1 - with self.assertRaises(InputRefused): - self.collect() - self.assertFalse(any("/git/blobs/" in path for path in self.requests)) - - def test_wrong_declared_blob_size_refuses(self): - self.blob()["size"] += 1 - with self.assertRaises(InputRefused): - self.collect() - - def test_wrong_blob_identity_refuses(self): - self.blob()["sha"] = "d" * 40 - with self.assertRaises(InputRefused): - self.collect() - - def test_invalid_base64_refuses(self): - self.blob()["content"] = "$not-base64" - with self.assertRaises(InputRefused): - self.collect() - - def test_binary_workflow_refuses(self): - self.entries[0]["size"] = self.blob()["size"] = 1 - self.blob()["content"] = "/w==" - with self.assertRaises(InputRefused): - self.collect() - - def test_total_byte_bound_applies_across_files(self): - with patch("scripts.protected_policy_inputs.MAX_TOTAL_BYTES", 4): - with self.assertRaises(InputRefused): - self.collect() - - def test_file_count_bound_applies_to_workflow_fanout(self): - for number in range(MAX_FILES): - self.add_file(f".github/workflows/extra-{number}.yml", b"on: push\n") - with self.assertRaises(InputRefused): - self.collect() - - def test_materialization_is_inert_and_does_not_preserve_executable_bits(self): - with tempfile.TemporaryDirectory() as temporary: - root = Path(temporary) - marker = root / "executed" - files = {"scripts/attack.sh": f"#!/bin/sh\ntouch '{marker}'\n".encode()} - destination = root / "candidate" - write_inputs(files, destination) - self.assertFalse(marker.exists()) - self.assertEqual(0o600, (destination / "scripts/attack.sh").stat().st_mode & 0o777) - self.assertEqual(files["scripts/attack.sh"], (destination / "scripts/attack.sh").read_bytes()) - - def test_materialization_refuses_existing_destination_and_traversal(self): - with tempfile.TemporaryDirectory() as temporary: - root = Path(temporary) - with self.assertRaises(FileExistsError): - write_inputs({WORKFLOW: b"on: push"}, root) - with self.assertRaises(InputRefused): - write_inputs({"../outside": b"must not write"}, root / "candidate") - self.assertFalse((root / "outside").exists()) - - def test_materialization_refuses_a_symlink_destination(self): - with tempfile.TemporaryDirectory() as temporary: - root = Path(temporary) - outside = root / "outside" - outside.mkdir() - redirected = root / "candidate" - redirected.symlink_to(outside, target_is_directory=True) - with self.assertRaises(FileExistsError): - write_inputs({WORKFLOW: b"on: push"}, redirected) - self.assertEqual([], list(outside.iterdir())) - - def test_materialization_refuses_non_bytes_and_oversized_files_before_creation(self): - with tempfile.TemporaryDirectory() as temporary: - destination = Path(temporary) / "candidate" - for contents in ("not bytes", b"x" * (MAX_FILE_BYTES + 1)): - with self.subTest(kind=type(contents).__name__), self.assertRaises(InputRefused): - write_inputs({WORKFLOW: contents}, destination) - self.assertFalse(destination.exists()) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_protected_policy_issue.py b/tests/test_protected_policy_issue.py deleted file mode 100644 index 9fcb6a1..0000000 --- a/tests/test_protected_policy_issue.py +++ /dev/null @@ -1,318 +0,0 @@ -import contextlib -import hashlib -import io -import json -import os -from pathlib import Path -import subprocess -import sys -import tempfile -import unittest -from unittest.mock import Mock, patch - -from scripts import protected_policy_issue as issue -from scripts.protected_policy_inputs import InputRefused -from tests import test_protected_policy_check as snapshot_fixtures -from tests.test_protected_policy_inputs import BASE, HEAD - - -APP_ID = 12345 -CHECK_ID = 67890 - - -class FakeIssuerCurl: - """Only constructed HTTP responses. No App credential or network is used.""" - - def __init__(self): - self.calls = [] - self.app_id = APP_ID - self.head = HEAD - self.name = issue.CHECK_NAME - self.status_override = None - self.after_begin = lambda: None - self.after_complete = lambda: None - - def __call__(self, command, **options): - if "--version" in command: - return subprocess.CompletedProcess(command, 0, b"curl 8.5.0 fixture\n") - payload_file = command[command.index("--data-binary") + 1][1:] - payload = json.loads(Path(payload_file).read_text()) - self.calls.append((command, options, payload)) - beginning = command[command.index("--request") + 1] == "POST" - record = { - "id": CHECK_ID, "name": self.name, "head_sha": self.head, - "app": {"id": self.app_id}, "status": "in_progress" if beginning else "completed", - "conclusion": None if beginning else "success", - } - Path(command[command.index("--output") + 1]).write_text(json.dumps(record)) - (self.after_begin if beginning else self.after_complete)() - status = self.status_override or (b"201" if beginning else b"200") - return subprocess.CompletedProcess(command, 0, status) - - -class ProtectedPolicyIssuerTest(unittest.TestCase): - def setUp(self): - self.temporary = tempfile.TemporaryDirectory() - self.addCleanup(self.temporary.cleanup) - root = Path(self.temporary.name) - self.bundle = root / "ca.pem" - self.bundle.write_bytes(b"synthetic reviewed CA") - self.digest = hashlib.sha256(self.bundle.read_bytes()).hexdigest() - self.roots = root / "empty" - self.roots.mkdir() - self.curl = FakeIssuerCurl() - self.now = 0.0 - self.issuer = issue.CheckRunIssuer( - self.bundle, self.roots, self.digest, "synthetic_app_token", - expected_app_id=APP_ID, run=self.curl, clock=lambda: self.now, - ) - self.snapshot = snapshot_fixtures.ProtectedPolicySnapshotTest(methodName="runTest") - self.snapshot.setUp() - self.api = self.snapshot.api - - def run_gate(self, *, head=HEAD): - return issue.qualify_and_issue(self.api.read, self.issuer, 7, BASE, head) - - def test_real_validator_then_app_check_then_final_identity_recheck(self): - report = self.run_gate() - self.assertEqual(APP_ID, report["app_id"]) - self.assertEqual(HEAD, report["head_sha"]) - self.assertEqual(2, len(self.curl.calls)) - self.assertEqual("in_progress", self.curl.calls[0][2]["status"]) - self.assertNotIn("conclusion", self.curl.calls[0][2]) - self.assertEqual("success", self.curl.calls[1][2]["conclusion"]) - self.assertNotIn("candidate must not execute", json.dumps(report)) - self.assertEqual(f"/repos/{issue.PILOT_REPOSITORY}/pulls/7", self.api.requests[-1]) - - def test_check_write_endpoint_is_fixed_to_the_pilot(self): - self.run_gate() - urls = [command[command.index("--url") + 1] for command, _, _ in self.curl.calls] - self.assertEqual([ - f"https://api.github.com/repos/{issue.PILOT_REPOSITORY}/check-runs", - f"https://api.github.com/repos/{issue.PILOT_REPOSITORY}/check-runs/{CHECK_ID}", - ], urls) - - def test_app_token_is_stdin_only_and_cannot_make_generic_reads(self): - self.run_gate() - for command, options, _ in self.curl.calls: - self.assertNotIn("synthetic_app_token", " ".join(command)) - self.assertEqual(b'header = "Authorization: Bearer synthetic_app_token"\n', options["input"]) - self.assertIn("Accept-Encoding: identity", command) - self.assertNotIn("--location", command) - self.assertNotIn("--compressed", command) - with self.assertRaises(InputRefused): - self.issuer(f"/repos/{issue.PILOT_REPOSITORY}/pulls/7") - - def test_wrong_issuer_can_never_complete_a_success(self): - self.curl.app_id = APP_ID + 1 - with self.assertRaisesRegex(InputRefused, "App issuer"): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - self.assertNotIn("conclusion", self.curl.calls[0][2]) - - def test_ordinary_actions_issuer_is_not_the_expected_app(self): - self.curl.app_id = 15368 - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_empty_installation_token_refuses(self): - with self.assertRaises(InputRefused): - issue.CheckRunIssuer( - self.bundle, self.roots, self.digest, "", expected_app_id=APP_ID, - run=self.curl, clock=lambda: self.now, - ) - self.assertEqual([], self.curl.calls) - - def test_noninteger_app_id_is_not_accepted(self): - self.curl.app_id = str(APP_ID) - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_wrong_returned_name_never_completes(self): - self.curl.name = "unrelated" - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_wrong_returned_head_never_completes(self): - self.curl.head = "d" * 40 - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_head_changed_after_app_identity_refuses_success(self): - self.curl.after_begin = lambda: self.api.pull()["head"].update(sha="d" * 40) - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_base_changed_after_app_identity_refuses_success(self): - self.curl.after_begin = lambda: self.snapshot.branch["commit"].update(sha="d" * 40) - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_unprotected_base_after_app_identity_refuses_success(self): - self.curl.after_begin = lambda: self.snapshot.branch.update(protected=False) - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_stale_event_never_reads_candidate_blobs_or_writes_checks(self): - with self.assertRaises(InputRefused): - self.run_gate(head="d" * 40) - self.assertEqual([], self.curl.calls) - self.assertFalse(any("/git/blobs/" in path for path in self.api.requests)) - - def test_real_forbidden_policy_never_creates_a_check(self): - self.api.entries.clear() - self.api.add_file(".github/workflows/bad.yml", b"on:\n pull_request_target:\njobs: {}\n") - with self.assertRaisesRegex(InputRefused, "forbidden"): - self.run_gate() - self.assertEqual([], self.curl.calls) - - def test_http_redirect_is_not_a_verified_app_response(self): - self.curl.status_override = b"302" - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_expired_transport_cannot_finish_the_pending_check(self): - self.curl.after_begin = lambda: setattr(self, "now", 121.0) - with self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual(1, len(self.curl.calls)) - - def test_completion_response_identity_is_still_verified(self): - self.curl.after_begin = lambda: setattr(self.curl, "app_id", APP_ID + 1) - with self.assertRaises(InputRefused): - self.run_gate() - - def test_failed_completion_is_not_reported_as_a_success(self): - self.curl.after_begin = lambda: setattr(self.curl, "status_override", b"500") - with self.assertRaises(InputRefused): - self.run_gate() - - def test_an_unbound_snapshot_report_cannot_create_a_check(self): - report = issue.validate_snapshot(self.api.read, 7, BASE) - report["head_repository_id"] += 1 - with patch.object(issue, "validate_snapshot", return_value=report), self.assertRaises(InputRefused): - self.run_gate() - self.assertEqual([], self.curl.calls) - - def test_issuer_cannot_be_reused(self): - self.run_gate() - with self.assertRaises(InputRefused): - self.issuer.begin(HEAD) - self.assertEqual(2, len(self.curl.calls)) - - def test_completion_requires_matching_original_head(self): - with self.assertRaises(InputRefused): - self.issuer.complete({"head_sha": HEAD}) - self.issuer.begin(HEAD) - with self.assertRaises(InputRefused): - self.issuer.complete({"head_sha": "d" * 40}) - self.assertEqual(1, len(self.curl.calls)) - - def test_cli_refuses_noncanonical_app_ids_before_creating_an_issuer(self): - environment = { - **ProtectedIssuerCallerTest("runTest").environment(), - "CURL_CA_BUNDLE": str(self.bundle), "SSL_CERT_DIR": str(self.roots), - "DW_CI_CA_SHA256": self.digest, "GITHUB_TOKEN": "synthetic_read_token", - "POLICY_CHECKS_TOKEN": "synthetic_app_token", - } - args = ["issuer", "--pull-request", "7", "--base-sha", BASE, "--expected-head", HEAD] - for app_id in ("12٢", "１２", "١", "0", "01", "1" * 20): - output = io.StringIO() - with self.subTest(app_id=app_id), patch.object(sys, "argv", args): - with patch.dict(os.environ, {**environment, "POLICY_GATE_APP_ID": app_id}, clear=True): - with patch.object(issue, "GitHubReadClient"), patch.object(issue, "CheckRunIssuer") as issuer: - with contextlib.redirect_stderr(output): - self.assertEqual(1, issue.main()) - issuer.assert_not_called() - self.assertEqual( - "A canonical positive App ID is required", - json.loads(output.getvalue())["policy_refused"], - ) - - def test_valid_cli_wires_separate_clients_to_one_total_deadline(self): - environment = { - **ProtectedIssuerCallerTest("runTest").environment(), - "CURL_CA_BUNDLE": str(self.bundle), "SSL_CERT_DIR": str(self.roots), - "DW_CI_CA_SHA256": self.digest, "GITHUB_TOKEN": "synthetic_read_token", - "POLICY_GATE_APP_ID": str(APP_ID), "POLICY_CHECKS_TOKEN": "synthetic_app_token", - } - read = Mock(side_effect=self.api.read) - read.deadline = 80.0 - args = ["issuer", "--pull-request", "7", "--base-sha", BASE, "--expected-head", HEAD] - output = io.StringIO() - with patch.object(sys, "argv", args), patch.dict(os.environ, environment, clear=True): - with patch.object(issue, "GitHubReadClient", return_value=read) as read_factory: - with patch.object(issue, "CheckRunIssuer", return_value=self.issuer) as write_factory: - with contextlib.redirect_stdout(output): - self.assertEqual(0, issue.main()) - self.assertEqual(80.0, self.issuer.deadline) - self.assertEqual("synthetic_read_token", read_factory.call_args.args[3]) - self.assertEqual("synthetic_app_token", write_factory.call_args.args[3]) - self.assertEqual(HEAD, json.loads(output.getvalue())["head_sha"]) - self.assertNotIn("synthetic_", output.getvalue()) - - -class ProtectedIssuerCallerTest(unittest.TestCase): - def environment(self): - return { - "GITHUB_ACTIONS": "true", "GITHUB_REPOSITORY": issue.PILOT_REPOSITORY, - "GITHUB_REPOSITORY_ID": str(issue.PILOT_REPOSITORY_ID), "GITHUB_REF": "refs/heads/main", - "GITHUB_SHA": BASE, "GITHUB_EVENT_NAME": "pull_request_target", - "GITHUB_WORKFLOW_REF": issue.WORKFLOW_REF, - "GITHUB_WORKFLOW_SHA": BASE, - } - - def test_protected_caller_context_matches(self): - issue.require_protected_execution(self.environment(), BASE) - - def test_each_missing_or_wrong_context_field_refuses(self): - environment = self.environment() - for name in environment: - wrong = {**environment, name: "wrong"} - missing = {key: value for key, value in environment.items() if key != name} - with self.subTest(name=name), self.assertRaises(InputRefused): - issue.require_protected_execution(wrong, BASE) - with self.subTest(missing=name), self.assertRaises(InputRefused): - issue.require_protected_execution(missing, BASE) - - def test_nonmain_execution_fails_before_credentials_or_clients(self): - args = ["issuer", "--pull-request", "7", "--base-sha", BASE, "--expected-head", HEAD] - output = io.StringIO() - with patch.object(sys, "argv", args), patch.dict(os.environ, {}, clear=True): - with patch.object(issue, "GitHubReadClient") as client, contextlib.redirect_stderr(output): - self.assertEqual(1, issue.main()) - client.assert_not_called() - self.assertIn("protected-main", json.loads(output.getvalue())["policy_refused"]) - - def test_isolated_cli_does_not_import_candidate_python(self): - with tempfile.TemporaryDirectory() as temporary: - scripts = Path(temporary) / "scripts" - scripts.mkdir() - (scripts / "__init__.py").write_text("raise SystemExit('candidate imported')\n") - result = subprocess.run( - [sys.executable, "-I", str(issue.ROOT / "scripts/protected_policy_issue.py"), "--help"], - cwd=temporary, capture_output=True, timeout=5, check=False, - ) - self.assertEqual(0, result.returncode, result.stderr.decode()) - self.assertNotIn(b"candidate imported", result.stderr) - - def test_malformed_app_id_does_not_echo_its_value(self): - environment = {**self.environment(), "CURL_CA_BUNDLE": "synthetic", "SSL_CERT_DIR": "synthetic", - "DW_CI_CA_SHA256": "synthetic", "GITHUB_TOKEN": "synthetic_read_token", - "POLICY_GATE_APP_ID": "invalid_sensitive_value"} - args = ["issuer", "--pull-request", "7", "--base-sha", BASE, "--expected-head", HEAD] - output = io.StringIO() - with patch.object(sys, "argv", args), patch.dict(os.environ, environment, clear=True): - with patch.object(issue, "GitHubReadClient"), patch.object(issue, "CheckRunIssuer") as issuer: - with contextlib.redirect_stderr(output): - self.assertEqual(1, issue.main()) - issuer.assert_not_called() - self.assertNotIn("invalid_sensitive_value", output.getvalue()) diff --git a/tests/test_qualification_policy.py b/tests/test_qualification_policy.py index 448242a..4da4a83 100644 --- a/tests/test_qualification_policy.py +++ b/tests/test_qualification_policy.py @@ -5,22 +5,13 @@ import tempfile import unittest from pathlib import Path -from unittest.mock import patch - -import yaml from scripts.qualification_policy import ( PolicyError, - LOCKED_POLICY_BOOTSTRAP_REVISION, - LOCKED_POLICY_INSTALL_COMMAND, - _workflow_action_references, - _references_secret_context, - _string_values, _verify_action_policy_preflight, scan_workflow_sources, validate_local_action_references, validate_policy, - verify_workflow_source, ) ROOT = Path(__file__).resolve().parents[1] @@ -128,7 +119,7 @@ def test_current_control_repository_workflows_satisfy_policy(self) -> None: self.assertTrue(actions) self.assertTrue(all("@" in action for action in actions)) - def preflight_jobs(self, target: str = "ai") -> dict: + def test_policy_preflight_accepts_binary_only_without_relaxing_dependency_command(self) -> None: checkout, _ = self.approved_action("actions/checkout") setup_python, _ = self.approved_action("actions/setup-python") jobs = { @@ -150,7 +141,7 @@ def preflight_jobs(self, target: str = "ai") -> dict: {"uses": f"actions/setup-python@{setup_python}", "with": {"python-version": "3.13"}}, {"run": "python -m pip install PyYAML==6.0.2"}, {"run": "python .central-action-policy/scripts/qualification_policy.py validate " - f"--policy .central-action-policy/qualification/policy.json --target {target} " + "--policy .central-action-policy/qualification/policy.json --target ai " "--workflow-directory .github/workflows"}, ], }, @@ -163,15 +154,9 @@ def preflight_jobs(self, target: str = "ai") -> dict: }], }, } - return jobs - - def test_policy_preflight_accepts_binary_only_without_relaxing_dependency_command(self) -> None: - jobs = self.preflight_jobs() for command in ( "python -m pip install PyYAML==6.0.2", "python -m pip install --only-binary :all: PyYAML==6.0.2", - "python -I -m pip install PyYAML==6.0.2", - "python -I -m pip install --only-binary :all: PyYAML==6.0.2", ): with self.subTest(command=command): jobs["action-policy"]["steps"][3]["run"] = command @@ -206,258 +191,6 @@ def test_policy_preflight_accepts_binary_only_without_relaxing_dependency_comman with self.assertRaisesRegex(PolicyError, "unreviewed pip bootstrap"): _verify_action_policy_preflight("ai", "fixture", jobs, "gate") - def test_server_requires_the_isolated_hash_locked_policy_install(self) -> None: - jobs = self.preflight_jobs("server") - steps = jobs["action-policy"]["steps"] - steps.insert(3, {"uses": "durable-workflow/.github/.github/actions/bootstrap-patched-pip@" + LOCKED_POLICY_BOOTSTRAP_REVISION}) - steps[4]["run"] = LOCKED_POLICY_INSTALL_COMMAND - revisions = {LOCKED_POLICY_BOOTSTRAP_REVISION} - _verify_action_policy_preflight("server", "fixture", jobs, "gate", revisions, True) - for flag in ("-I ", "--isolated ", "--require-hashes ", "--only-binary=:all: ", "--no-deps "): - changed = copy.deepcopy(jobs) - changed["action-policy"]["steps"][4]["run"] = LOCKED_POLICY_INSTALL_COMMAND.replace(flag, "") - with self.subTest(removed=flag), self.assertRaisesRegex(PolicyError, "reviewed policy dependency"): - _verify_action_policy_preflight("server", "fixture", changed, "gate", revisions, True) - for command in ( - LOCKED_POLICY_INSTALL_COMMAND.replace("qualification/requirements.lock", "requirements.lock"), - LOCKED_POLICY_INSTALL_COMMAND.replace(".central-action-policy/", ""), - LOCKED_POLICY_INSTALL_COMMAND + " --extra-index-url https://example.com", - LOCKED_POLICY_INSTALL_COMMAND + " --trusted-host example.com", - "python -m pip install PyYAML==6.0.2", - ): - changed = copy.deepcopy(jobs) - changed["action-policy"]["steps"][4]["run"] = command - with self.subTest(command=command), self.assertRaisesRegex(PolicyError, "reviewed policy dependency"): - _verify_action_policy_preflight("server", "fixture", changed, "gate", revisions, True) - changed = copy.deepcopy(jobs) - changed["action-policy"]["steps"].pop(3) - with self.assertRaisesRegex(PolicyError, "locked-policy pip bootstrap"): - _verify_action_policy_preflight("server", "fixture", changed, "gate", revisions, True) - changed = copy.deepcopy(jobs) - changed["action-policy"]["steps"][3]["uses"] = steps[3]["uses"].replace(LOCKED_POLICY_BOOTSTRAP_REVISION, "75587fc33ff40c34f32ef777293dad56bb6f48c7") - with self.assertRaisesRegex(PolicyError, "locked-policy pip bootstrap"): - _verify_action_policy_preflight("server", "fixture", changed, "gate", revisions | {"75587fc33ff40c34f32ef777293dad56bb6f48c7"}, True) - self.assertTrue(self.policy["targets"]["server"]["workflows"][0]["locked_policy_dependency_required"]) - - def locked_preflight_jobs(self) -> dict: - jobs = self.preflight_jobs("server") - jobs["action-policy"]["steps"].insert(3, { - "uses": "durable-workflow/.github/.github/actions/bootstrap-patched-pip@" + LOCKED_POLICY_BOOTSTRAP_REVISION, - }) - jobs["action-policy"]["steps"][4]["run"] = LOCKED_POLICY_INSTALL_COMMAND - jobs["gate"]["name"] = "PHPUnit feature suite" - return jobs - - def verify_server_preflight(self, jobs: dict, defaults: dict | None = None) -> None: - document = {"jobs": jobs} - if defaults is not None: - document["defaults"] = defaults - source = "on:\n push:\n branches: [main]\n pull_request:\n workflow_dispatch:\n" - source += yaml.safe_dump(document, sort_keys=False) - verify_workflow_source( - "server", "main", self.policy["targets"]["server"]["workflows"][0], - source, {LOCKED_POLICY_BOOTSTRAP_REVISION}, - ) - - def test_preflight_refuses_redirected_install_and_validator_directories(self) -> None: - for directory in ("alternate", "../other", "${{ github.workspace }}/alternate", "${{ inputs.directory }}", "$GITHUB_WORKSPACE", None): - for scope in ("workflow", "job", "install", "validate"): - jobs = self.locked_preflight_jobs() - defaults = {"run": {"working-directory": directory}} - workflow_defaults = None - if scope == "workflow": - workflow_defaults = defaults - elif scope == "job": - jobs["action-policy"]["defaults"] = defaults - else: - index = 4 if scope == "install" else 5 - jobs["action-policy"]["steps"][index]["working-directory"] = directory - with self.subTest(scope=scope, directory=directory), self.assertRaisesRegex(PolicyError, "workspace root"): - self.verify_server_preflight(jobs, workflow_defaults) - - def test_preflight_accepts_explicit_workspace_root_directories(self) -> None: - for directory in (".", "./", "${{ github.workspace }}", "${{github.workspace}}/.", "${{ github.workspace }}/"): - for scope in ("workflow", "job", "step"): - jobs = self.locked_preflight_jobs() - defaults = {"run": {"working-directory": directory}} - workflow_defaults = None - if scope == "workflow": - workflow_defaults = defaults - elif scope == "job": - jobs["action-policy"]["defaults"] = defaults - else: - for step in jobs["action-policy"]["steps"][4:]: - step["working-directory"] = directory - with self.subTest(scope=scope, directory=directory): - self.verify_server_preflight(jobs, workflow_defaults) - - def test_preflight_directory_uses_the_most_specific_run_setting(self) -> None: - jobs = self.locked_preflight_jobs() - jobs["action-policy"]["defaults"] = {"run": {"working-directory": "."}} - self.verify_server_preflight(jobs, {"run": {"working-directory": "alternate"}}) - jobs["action-policy"]["defaults"]["run"]["working-directory"] = "alternate" - for step in jobs["action-policy"]["steps"][4:]: - step["working-directory"] = "${{ github.workspace }}" - self.verify_server_preflight(jobs) - del jobs["action-policy"]["steps"][5]["working-directory"] - with self.assertRaisesRegex(PolicyError, "workspace root"): - self.verify_server_preflight(jobs) - - def test_preflight_refuses_inherited_custom_shells(self) -> None: - for scope in ("workflow", "job"): - jobs = self.locked_preflight_jobs() - defaults = {"run": {"shell": "bash -c true {0}"}} - workflow_defaults = defaults if scope == "workflow" else None - if scope == "job": - jobs["action-policy"]["defaults"] = defaults - with self.subTest(scope=scope), self.assertRaisesRegex(PolicyError, "reviewed shell"): - self.verify_server_preflight(jobs, workflow_defaults) - - def test_preflight_refuses_custom_dynamic_and_non_shell_invocation(self) -> None: - for shell in ("bash {0}", "${{ inputs.shell }}", "python", "", None): - for index in (4, 5): - jobs = self.locked_preflight_jobs() - jobs["action-policy"]["steps"][index]["shell"] = shell - with self.subTest(shell=shell, index=index), self.assertRaisesRegex(PolicyError, "reviewed shell"): - self.verify_server_preflight(jobs) - - def test_preflight_accepts_default_and_builtin_safe_shells(self) -> None: - self.verify_server_preflight(self.locked_preflight_jobs()) - for shell in ("bash", "sh"): - for scope in ("workflow", "job", "step"): - jobs = self.locked_preflight_jobs() - defaults = {"run": {"shell": shell}} - workflow_defaults = defaults if scope == "workflow" else None - if scope == "job": - jobs["action-policy"]["defaults"] = defaults - elif scope == "step": - for step in jobs["action-policy"]["steps"][4:]: - step["shell"] = shell - with self.subTest(shell=shell, scope=scope): - self.verify_server_preflight(jobs, workflow_defaults) - - def test_preflight_shell_uses_the_most_specific_safe_setting(self) -> None: - jobs = self.locked_preflight_jobs() - jobs["action-policy"]["defaults"] = {"run": {"shell": "bash"}} - jobs["gate"]["defaults"] = {"run": {"shell": "sh"}} - self.verify_server_preflight(jobs, {"run": {"shell": "bash -c true {0}"}}) - jobs["action-policy"]["defaults"]["run"]["shell"] = "bash -c true {0}" - for step in jobs["action-policy"]["steps"][4:]: - step["shell"] = "sh" - self.verify_server_preflight(jobs) - del jobs["action-policy"]["steps"][5]["shell"] - with self.assertRaisesRegex(PolicyError, "reviewed shell"): - self.verify_server_preflight(jobs) - - def test_preflight_result_gate_cannot_inherit_a_custom_shell(self) -> None: - jobs = self.locked_preflight_jobs() - jobs["gate"]["defaults"] = {"run": {"shell": "bash -c true {0}"}} - with self.assertRaisesRegex(PolicyError, "fail closed"): - self.verify_server_preflight(jobs) - jobs["gate"]["steps"][0]["shell"] = "bash" - self.verify_server_preflight(jobs) - - def test_preflight_result_gate_cannot_ignore_its_failure(self) -> None: - jobs = self.locked_preflight_jobs() - self.verify_server_preflight(jobs) - for ignored in (False, "false"): - jobs["gate"]["steps"][0]["continue-on-error"] = ignored - with self.subTest(ignored=ignored): - self.verify_server_preflight(jobs) - for ignored in (True, "true", "${{ inputs.ignore }}", "${{ false }}"): - jobs["gate"]["steps"][0]["continue-on-error"] = ignored - with self.subTest(ignored=ignored), self.assertRaisesRegex(PolicyError, "fail closed"): - self.verify_server_preflight(jobs) - - def verify_preflight_context(self, jobs: dict, workflow_document: dict | None = None) -> None: - _verify_action_policy_preflight( - "server", "source-only context fixture", jobs, "gate", - {LOCKED_POLICY_BOOTSTRAP_REVISION}, True, workflow_document, - ) - - def test_preflight_job_rejects_unreviewed_execution_fields(self) -> None: - for key, value in ( - ("container", "ubuntu:24.04"), - ("services", {"sidecar": {"image": "busybox:latest"}}), - ("strategy", {"matrix": {"candidate": ["one"]}}), - ("environment", "unreviewed"), - ("continue-on-error", True), - ("concurrency", "unreviewed"), - ("outputs", {"value": "unreviewed"}), - ): - jobs = self.locked_preflight_jobs() - jobs["action-policy"][key] = value - with self.subTest(key=key), self.assertRaisesRegex(PolicyError, "reviewed read-only shape"): - self.verify_preflight_context(jobs) - - def test_preflight_actions_cannot_skip_or_change_the_reviewed_context(self) -> None: - for index in range(4): - for key, value in (("if", "false"), ("if", False), ("continue-on-error", True), ("working-directory", "alternate")): - jobs = self.locked_preflight_jobs() - jobs["action-policy"]["steps"][index][key] = value - with self.subTest(index=index, key=key, value=value), self.assertRaises(PolicyError): - self.verify_preflight_context(jobs) - - def test_preflight_startup_bindings_are_data_only_and_refused(self) -> None: - for name in ("BASH_ENV", "ENV", "LD_PRELOAD", "NODE_OPTIONS", "PYTHONPATH", "PATH", "GIT_CONFIG_COUNT", "GIT_CONFIG_KEY_0", "GIT_CONFIG_VALUE_0", "GIT_CONFIG_SYSTEM", "GIT_CONFIG_PARAMETERS", "UNREVIEWED_SETTING"): - jobs = self.locked_preflight_jobs() - document = {"env": {name: "candidate-controlled-placeholder"}} - with self.subTest(name=name), self.assertRaisesRegex(PolicyError, "environment"): - self.verify_preflight_context(jobs, document) - - def test_preflight_step_startup_bindings_refuse_without_running_them(self) -> None: - for index in range(6): - jobs = self.locked_preflight_jobs() - jobs["action-policy"]["steps"][index]["env"] = {"LD_PRELOAD": "candidate-controlled-placeholder"} - with self.subTest(index=index), self.assertRaisesRegex(PolicyError, "environment"): - self.verify_preflight_context(jobs) - - def test_preflight_preserves_reviewed_diagnostics_and_literal_clears(self) -> None: - jobs = self.locked_preflight_jobs() - document = {"env": {"CI": "true", "CARGO_TERM_COLOR": "always", "RUST_BACKTRACE": "1"}} - jobs["action-policy"]["env"] = {"PYTHONUNBUFFERED": "1"} - for index, step in enumerate(jobs["action-policy"]["steps"]): - step["env"] = {"FORCE_COLOR": "1"} - if index < 4: - step["id"] = f"reviewed_{index}" - step["if"] = "${{ github.server_url == 'https://github.com' }}" - step["continue-on-error"] = "false" - self.verify_preflight_context(jobs, document) - clears = {name: "" for name in ("BASH_ENV", "ENV", "LD_PRELOAD", "NODE_OPTIONS")} - self.verify_preflight_context(self.locked_preflight_jobs(), {"env": clears}) - jobs = self.locked_preflight_jobs() - jobs["action-policy"]["env"] = clears - self.verify_preflight_context(jobs) - - def test_preflight_environment_is_closed_including_shadowed_bindings(self) -> None: - jobs = self.locked_preflight_jobs() - jobs["action-policy"]["env"] = {"BASH_ENV": ""} - for document in ( - {"env": {"BASH_ENV": "candidate-controlled-placeholder"}}, - {"env": {"NODE_OPTIONS": "${{ inputs.options }}"}}, - {"env": {"GIT_CONFIG_COUNT": ""}}, - {"env": {"CI": ["not a scalar"]}}, - {"env": "not a mapping"}, - ): - with self.subTest(document=document), self.assertRaisesRegex(PolicyError, "environment"): - self.verify_preflight_context(jobs, document) - - def test_preflight_defaults_reject_unknown_execution_settings(self) -> None: - for scope in ("workflow", "job"): - jobs = self.locked_preflight_jobs() - defaults = {"run": {"unreviewed": "value"}} - document = {"defaults": defaults} if scope == "workflow" else None - if scope == "job": - jobs["action-policy"]["defaults"] = defaults - with self.subTest(scope=scope), self.assertRaisesRegex(PolicyError, "defaults"): - self.verify_preflight_context(jobs, document) - self.verify_preflight_context(self.locked_preflight_jobs(), {"defaults": {"run": {}}}) - - def test_preflight_shape_does_not_close_unrelated_test_jobs(self) -> None: - jobs = self.locked_preflight_jobs() - jobs["other"] = {"container": "ubuntu:24.04", "env": {"BASH_ENV": "other-job-placeholder"}} - self.verify_preflight_context(jobs) - def test_approved_immutable_action_with_version_comment_is_accepted(self) -> None: commit, version = self.approved_action("actions/checkout") evidence = self.scan(workflow(f"- uses: actions/checkout@{commit} # {version}")) @@ -484,187 +217,6 @@ def test_pull_request_target_is_rejected(self) -> None: with self.assertRaisesRegex(PolicyError, "pull_request_target"): self.scan(workflow("- run: true", trigger="pull_request_target")) - def protected_workflow(self) -> str: - return (ROOT / ".github/workflows/protected-policy.yml").read_text(encoding="utf-8") - - def test_only_exact_reviewed_protected_workflow_is_allowed(self) -> None: - sources = {".github/workflows/protected-policy.yml": self.protected_workflow()} - report = scan_workflow_sources(self.policy, "github-control-plane", sources) - self.assertIn(".github/workflows/protected-policy.yml", report) - - def test_local_reviewed_workflow_keeps_its_original_lf_bytes(self) -> None: - source = (ROOT / ".github/workflows/protected-policy.yml").read_bytes() - with tempfile.TemporaryDirectory() as temporary: - directory = Path(temporary) / ".github/workflows" - directory.mkdir(parents=True) - (directory / "protected-policy.yml").write_bytes(source) - - validate_local_action_references(self.policy, directory) - - def test_local_workflow_line_endings_cannot_reuse_the_reviewed_hash(self) -> None: - source = (ROOT / ".github/workflows/protected-policy.yml").read_bytes() - for ending in (b"\r\n", b"\r"): - changed = source.replace(b"\n", ending) - self.assertNotEqual(source, changed) - with self.subTest(ending=ending), tempfile.TemporaryDirectory() as temporary: - directory = Path(temporary) / ".github/workflows" - directory.mkdir(parents=True) - (directory / "protected-policy.yml").write_bytes(changed) - - with self.assertRaisesRegex(PolicyError, "pull_request_target"): - validate_local_action_references(self.policy, directory) - - def test_ordinary_local_workflows_keep_supported_line_endings(self) -> None: - source = workflow("- run: true").encode("utf-8") - for ending in (b"\n", b"\r\n", b"\r"): - with self.subTest(ending=ending), tempfile.TemporaryDirectory() as temporary: - directory = Path(temporary) - (directory / "ci.yml").write_bytes(source.replace(b"\n", ending)) - - validate_local_action_references(self.policy, directory) - - def test_shared_yaml_nodes_do_not_multiply_action_reference_work(self) -> None: - visits = 0 - - class CountedMapping(dict): - def touch(self): - nonlocal visits - visits += 1 - if visits > 64: - raise AssertionError("Alias traversal exceeded the fixture's work budget") - - def items(self): - self.touch() - return super().items() - - def values(self): - self.touch() - return super().values() - - reference = "actions/checkout@" + self.approved_action("actions/checkout")[0] - document = CountedMapping({"uses": reference}) - for _ in range(14): - document = CountedMapping({"nested": [document, document]}) - - with patch("scripts.qualification_policy._parse_yaml", return_value=document): - self.assertEqual([reference], _workflow_action_references("", "shared graph")) - - def test_real_yaml_aliases_keep_all_distinct_action_references(self) -> None: - first = "actions/checkout@" + self.approved_action("actions/checkout")[0] - second = "actions/setup-python@" + self.approved_action("actions/setup-python")[0] - source = f"shared: &shared [{{uses: {first}}}, {{uses: {second}}}]\n" - previous = "shared" - for index in range(8): - name = f"level{index}" - source += f"{name}: &{name} [*{previous}, *{previous}]\n" - previous = name - - self.assertEqual({first, second}, set(_workflow_action_references(source, "real aliases"))) - - def test_recursive_yaml_alias_keeps_the_existing_parser_refusal(self) -> None: - reference = "actions/checkout@" + self.approved_action("actions/checkout")[0] - source = f"cycle: &cycle [*cycle]\nstep: {{uses: {reference}}}\n" - - with self.assertRaisesRegex(PolicyError, "recursive node"): - _workflow_action_references(source, "recursive alias") - - def test_secret_context_shared_graph_has_bounded_full_scan_work(self) -> None: - visits = 0 - - class CountedMapping(dict): - def values(self): - nonlocal visits - visits += 1 - if visits > 64: - raise AssertionError("Secret-context alias traversal exceeded the fixture work budget") - return super().values() - - shared = CountedMapping({"value": "ordinary text"}) - for _ in range(14): - shared = CountedMapping({"nested": [shared, shared]}) - document = { - "on": {"pull_request": None}, - "permissions": {"contents": "read"}, - "shared": shared, - "jobs": {"test": {"runs-on": "ubuntu-latest", "steps": [{"run": "true"}]}}, - } - with patch("scripts.qualification_policy._parse_yaml", return_value=document): - self.scan("") - self.assertLessEqual(visits, 64) - - def test_shared_alias_scan_keeps_secret_refusals_and_distinct_actions(self) -> None: - first, first_version = self.approved_action("actions/checkout") - second, second_version = self.approved_action("actions/setup-python") - graph = "leaf: &leaf [ordinary text]\n" - previous = "leaf" - for index in range(8): - name = f"level{index}" - graph += f"{name}: &{name} [*{previous}, *{previous}]\n" - previous = name - source = graph + workflow( - f"- uses: actions/checkout@{first} # {first_version}", - f"- uses: actions/setup-python@{second} # {second_version}", - ) - evidence = self.scan(source)[".github/workflows/test.yml"] - self.assertEqual(2, len(evidence["external_actions"])) - with self.assertRaisesRegex(PolicyError, "references a secret"): - self.scan("late: ${{ secrets.EXAMPLE }}\n" + source) - shared_secret_source = source.replace("ordinary text", '"${{ secrets.EXAMPLE }}"') - with self.assertRaisesRegex(PolicyError, "references a secret"): - self.scan(shared_secret_source) - self.assertTrue(_references_secret_context({"nested": ["safe", "${{ secrets.EXAMPLE }}"]})) - self.assertFalse(_references_secret_context({"nested": ["plain secrets.EXAMPLE"]})) - - def test_string_values_preserve_order_for_ordinary_yaml_trees(self) -> None: - self.assertEqual( - ["first", "second", "third"], - _string_values({"a": ["first", ["second"]], "b": {"c": "third"}}), - ) - - def test_preceding_policy_still_refuses_the_protected_workflow(self) -> None: - self.policy["workflow_trust"]["pull_request_target_exceptions"] = [] - sources = {".github/workflows/protected-policy.yml": self.protected_workflow()} - with self.assertRaisesRegex(PolicyError, "pull_request_target"): - scan_workflow_sources(self.policy, "github-control-plane", sources) - - def test_same_workflow_cannot_gain_authority_at_another_path_or_repository(self) -> None: - source = self.protected_workflow() - for target, path in ( - ("github-control-plane", ".github/workflows/copied.yml"), - ("server", ".github/workflows/protected-policy.yml"), - ): - sources = {path: source} - with self.subTest(target=target, path=path), self.assertRaisesRegex(PolicyError, "pull_request_target"): - scan_workflow_sources(self.policy, target, sources) - - def test_protected_source_mutations_cannot_use_the_exception(self) -> None: - source = self.protected_workflow() - for before, after in ( - ("ref: ${{ github.sha }}", "ref: ${{ github.event.pull_request.head.sha }}"), - ("environment: protected-policy-gate", "environment: production"), - ("permission-checks: write", "permission-contents: write"), - ("repositories: .github", "repositories: cloud"), - ("NODE_OPTIONS: --use-openssl-ca", "NODE_OPTIONS: --use-bundled-ca"), - ('--expected-head "$POLICY_HEAD_SHA"', '--expected-head "a"'), - ): - changed = source.replace(before, after) - self.assertNotEqual(source, changed) - sources = {".github/workflows/protected-policy.yml": changed} - with self.subTest(before=before), self.assertRaisesRegex(PolicyError, "pull_request_target"): - scan_workflow_sources(self.policy, "github-control-plane", sources) - - def test_modified_or_additional_exceptions_require_another_review(self) -> None: - original = self.policy["workflow_trust"]["pull_request_target_exceptions"][0] - for changed in ( - [{**original, "target": "server"}], - [{**original, "path": ".github/workflows/copied.yml"}], - [{**original, "sha256": "0" * 64}], - [original, original], - ): - self.policy["workflow_trust"]["pull_request_target_exceptions"] = changed - with self.subTest(changed=changed), self.assertRaisesRegex(PolicyError, "separately reviewed"): - validate_policy(self.policy) - def test_pull_request_jobs_cannot_use_environments_or_secrets(self) -> None: source = workflow("- run: echo ${{ secrets.TOKEN }}") source = source.replace(" runs-on: ubuntu-latest", " runs-on: ubuntu-latest\n environment: production")