From c47fc64a8e9060be78985b8b5d15cbc3d69853a3 Mon Sep 17 00:00:00 2001 From: Joshua Date: Thu, 22 Oct 2026 09:11:05 -0500 Subject: [PATCH] ci: derive auto-fixes only from trusted base tools --- .github/workflows/auto-fix.yml | 189 +--- .github/workflows/dependabot-auto-merge.yml | 33 +- .github/workflows/release.yml | 4 + .github/workflows/validate.yml | 93 +- CONTRIBUTING.md | 25 +- docs/autofix-threat-model.md | 100 ++ scripts/autofix.py | 304 ++++++ tests/test_autofix_policy_guard.py | 1056 +++++++++++++++++-- 8 files changed, 1448 insertions(+), 356 deletions(-) create mode 100644 docs/autofix-threat-model.md create mode 100644 scripts/autofix.py diff --git a/.github/workflows/auto-fix.yml b/.github/workflows/auto-fix.yml index f5c7709..c64f1e1 100644 --- a/.github/workflows/auto-fix.yml +++ b/.github/workflows/auto-fix.yml @@ -1,22 +1,12 @@ --- name: Auto-fix -# Privileged half of the Dependabot auto-fix flow. React to the Validate workflow -# FAILING on a Dependabot PR and push the mechanical fixes that Validate's -# unprivileged `autofix-patch` job already computed. This job consumes that fix -# set as DATA and rebuilds the commit through the Git Data API: it never checks -# out, fetches, or executes the PR's code (no actions/checkout of the untrusted -# ref, no git fetch/checkout of its SHA, no uv/prek), so no untrusted code runs in -# this privileged, write-credentialled context. workflow_run runs in the base-repo -# context, so the App-token secrets are available; the App-token ref update -# re-triggers Validate (GITHUB_TOKEN pushes would not). Human PRs are never -# auto-mutated. +# A Validate failure is only a wake-up signal, NEVER a source of replacement bytes. +# All code, tools and configuration below come from this workflow's immutable base +# commit. PR blobs are data only. See docs/autofix-threat-model.md. on: - # workflow_run is required by design (react to - # Validate FAILING and re-trigger CI after the App-token ref update); it runs in - # the base-repo context. The PR head is referenced only as a Git Data API - # parameter (base tree + parent) — never checked out, fetched, or executed — so - # no untrusted code runs at elevated privilege. + # Retained intentionally for base-owned code and the existing narrowly scoped App. + # No PR checkout, PR dependency installation, hook execution or artifact download. workflow_run: # zizmor: ignore[dangerous-triggers] workflows: [Validate] types: [completed] @@ -32,169 +22,58 @@ jobs: auto-fix: name: auto-fix runs-on: ubuntu-latest + timeout-minutes: 15 if: >- github.event.workflow_run.event == 'pull_request' && github.event.workflow_run.conclusion == 'failure' && - github.event.workflow_run.actor.login == 'dependabot[bot]' + github.event.workflow_run.actor.id == 49699333 && + github.event.workflow_run.triggering_actor.id == 49699333 permissions: contents: read - actions: read # download-artifact reads the triggering run's artifacts + actions: read # Authenticate the original run and its latest attempt. + pull-requests: read # Verify the current, open, same-repository bot PR. steps: - name: Harden the runner uses: step-security/harden-runner@ccd8616d44fd3846e67624a50d5aad6d37bf2d25 # v2.22.1 with: egress-policy: audit - - name: Guard against auto-fix loops - id: guard - # Abort if the PR head is already an auto-fix commit — prevents an - # infinite fix -> fail -> fix cycle when fixers cannot green the checks. - # Decided from the event payload; no PR checkout needed. - env: - HEAD_MSG: ${{ github.event.workflow_run.head_commit.message }} - run: | - if printf '%s' "$HEAD_MSG" | head -n1 | grep -q '^ci: 🤖 mechanical auto-fix'; then - echo "Head is already an auto-fix commit; skipping to avoid a loop." >&2 - echo "skip=true" >> "$GITHUB_OUTPUT" - else - echo "skip=false" >> "$GITHUB_OUTPUT" - fi + - name: Check out only the trusted workflow commit + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + ref: ${{ github.workflow_sha }} + persist-credentials: false - - name: Download the fix set - id: download - if: steps.guard.outputs.skip == 'false' - # Missing artifact (no fixable changes, or a genuinely red PR) is the - # normal no-op case, not an error — fall through and let later steps skip. - continue-on-error: true - uses: actions/download-artifact@9000827ccba6bdab643e8b6fd33ac0654aef8333 # v8.0.2 + - name: Install the base-locked uv + uses: astral-sh/setup-uv@1c37ad07a6a961277cf70c0d37d6f313000f5884 # v10.3.0 with: - name: autofix-fixes - run-id: ${{ github.event.workflow_run.id }} - github-token: ${{ github.token }} - path: fixes + version-file: uv.lock + enable-cache: false - - name: Read the trusted PR file tree - if: steps.guard.outputs.skip == 'false' && steps.download.outcome == 'success' + - name: Install only trusted base-locked tools env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - HEAD_SHA: ${{ github.event.workflow_run.head_sha }} - REPO: ${{ github.repository }} - run: gh api "repos/$REPO/git/trees/$HEAD_SHA?recursive=1" > trusted-tree.json + UV_MALWARE_CHECK: "1" + run: uv sync --locked --only-group dev --no-install-project --no-cache - - name: Validate mechanical fix scope - id: scope - if: steps.guard.outputs.skip == 'false' && steps.download.outcome == 'success' - # This code comes from the trusted base workflow. Never load a checker - # from the PR artifact or checkout before minting a write credential. - shell: python - run: | - # BEGIN AUTOFIX SCOPE GUARD - import json - import os - from pathlib import Path - - fixes = json.loads(Path("fixes/fixes.json").read_text(encoding="utf-8")) - tree = json.loads(Path("trusted-tree.json").read_text(encoding="utf-8")) - if tree.get("truncated") is not False: - raise ValueError("Cannot verify a partial PR file tree") - tracked = { - item["path"]: item["mode"] - for item in tree["tree"] - if item["type"] == "blob" - } - protected_files = { - ".github/advisory-exceptions.json", - ".github/osv-scanner-empty.toml", - "pyproject.toml", - } - protected_prefixes = ( - ".github/workflows/", - ".github/security-scanner/", - "ci/security-scanner/", - "scripts/", - ) - if not isinstance(fixes, list): - raise ValueError("Mechanical fixes must be a JSON array") - for entry in fixes: - if not isinstance(entry, dict): - raise ValueError("Malformed mechanical fix entry") - path = entry.get("path") - if ( - not isinstance(path, str) - or "\\" in path - or any(part in {"", ".", ".."} for part in path.split("/")) - or path not in tracked - or entry.get("mode") != tracked[path] - or entry.get("status") != "M" - or entry.get("mode") not in {"100644", "100755"} - or path in protected_files - or path.startswith(protected_prefixes) - ): - raise ValueError( - f"Refusing auto-fix of protected, new, or noncanonical path: {path!r}. " - "A maintainer must review policy and automation changes." - ) - with Path(os.environ["GITHUB_OUTPUT"]).open("a", encoding="utf-8") as output: - output.write("safe=true\n") - # END AUTOFIX SCOPE GUARD + - name: Independently derive bounded fixes + id: fixes + env: + GH_READ_TOKEN: ${{ github.token }} + TRUSTED_SHA: ${{ github.workflow_sha }} + run: .venv/bin/python -I scripts/autofix.py prepare - name: Mint a GitHub App token id: app-token - # Only mint a write-capable token once we actually have fixes to push. - if: steps.scope.outputs.safe == 'true' + if: steps.fixes.outputs.has_fixes == 'true' uses: actions/create-github-app-token@64ab729a6294000f5aa4a4bc0b73d1f32bc45510 # v3.3.0 with: client-id: ${{ vars.APP_CLIENT_ID }} private-key: ${{ secrets.APP_PRIVATE_KEY }} - # Least privilege: only Contents write (push mechanical fixes). permission-contents: write - - name: Rebuild the fix commit via the Git Data API and update the branch - if: steps.scope.outputs.safe == 'true' + - name: Revalidate and atomically commit the locally derived fixes + if: steps.fixes.outputs.has_fixes == 'true' env: - # gh api authenticates as the App installation (Contents: write). - GH_TOKEN: ${{ steps.app-token.outputs.token }} - HEAD_BRANCH: ${{ github.event.workflow_run.head_branch }} - HEAD_SHA: ${{ github.event.workflow_run.head_sha }} - REPO: ${{ github.repository }} - run: | - f=fixes/fixes.json - if [ ! -s "$f" ] || [ "$(jq 'length' "$f")" = "0" ]; then - echo "No mechanical fixes to apply; leaving the PR unchanged." >&2 - exit 0 - fi - # Refuse to push a partial commit if the fix set includes deletions — - # rebuilding those safely is out of scope; leave the PR red for a human. - if [ "$(jq '[.[] | select(.status == "D")] | length' "$f")" != "0" ]; then - echo "Fix set includes deletions; refusing a partial auto-fix." >&2 - exit 0 - fi - # Rebuild the fix commit entirely through the Git Data API. The PR head - # is referenced only as an API parameter (base tree + parent) — its code - # is never checked out, fetched, or executed in this privileged job. - head_tree=$(gh api "repos/$REPO/git/commits/$HEAD_SHA" -q '.tree.sha') - tree='[]' - len=$(jq 'length' "$f") - i=0 - while [ "$i" -lt "$len" ]; do - path=$(jq -r ".[$i].path" "$f") - mode=$(jq -r ".[$i].mode" "$f") - blob=$(jq -c ".[$i] | {content, encoding: \"base64\"}" "$f" \ - | gh api "repos/$REPO/git/blobs" --input - -q '.sha') - tree=$(jq -c --arg p "$path" --arg m "$mode" --arg s "$blob" \ - '. + [{path: $p, mode: $m, type: "blob", sha: $s}]' <<<"$tree") - i=$((i + 1)) - done - new_tree=$(jq -nc --arg base "$head_tree" --argjson t "$tree" \ - '{base_tree: $base, tree: $t}' \ - | gh api "repos/$REPO/git/trees" --input - -q '.sha') - new_commit=$(jq -nc --arg t "$new_tree" --arg p "$HEAD_SHA" \ - '{message: "ci: 🤖 mechanical auto-fix", tree: $t, parents: [$p], - author: {name: "optimizer-release-bot[bot]", - email: "optimizer-release-bot[bot]@users.noreply.github.com"}}' \ - | gh api "repos/$REPO/git/commits" --input - -q '.sha') - # force=false => a branch that moved since Validate ran is a non-fast- - # forward the API rejects; we never clobber newer work with a stale fix. - jq -nc --arg s "$new_commit" '{sha: $s, force: false}' \ - | gh api -X PATCH "repos/$REPO/git/refs/heads/$HEAD_BRANCH" --input - - echo "Pushed mechanical auto-fix $new_commit to $HEAD_BRANCH." >&2 + GH_READ_TOKEN: ${{ github.token }} + GH_WRITE_TOKEN: ${{ steps.app-token.outputs.token }} + run: .venv/bin/python -I scripts/autofix.py apply diff --git a/.github/workflows/dependabot-auto-merge.yml b/.github/workflows/dependabot-auto-merge.yml index b5e7565..b5e936d 100644 --- a/.github/workflows/dependabot-auto-merge.yml +++ b/.github/workflows/dependabot-auto-merge.yml @@ -42,20 +42,35 @@ jobs: with: github-token: ${{ secrets.GITHUB_TOKEN }} - - name: Approve the current Dependabot revision + - name: Approve and enable auto-merge for the current revision env: GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} HEAD_SHA: ${{ github.event.pull_request.head.sha }} PR_NUMBER: ${{ github.event.pull_request.number }} + PR_URL: ${{ github.event.pull_request.html_url }} + DEFAULT_BRANCH: ${{ github.event.repository.default_branch }} REPO: ${{ github.repository }} run: | - jq -nc --arg commit_id "$HEAD_SHA" \ + set -euo pipefail + is_current() { + gh api "repos/$REPO/pulls/$PR_NUMBER" | jq -e \ + --arg head "$HEAD_SHA" --arg repo "$REPO" --arg base "$DEFAULT_BRANCH" ' + .state == "open" and .draft == false and .merged == false and + .user.id == 49699333 and .head.repo.full_name == $repo and + .base.repo.full_name == $repo and .base.ref == $base and + .head.sha == $head' >/dev/null + } + if ! is_current; then + echo "The bot event is stale; leaving review and merge unchanged." + exit 0 + fi + review_id=$(jq -nc --arg commit_id "$HEAD_SHA" \ '{event: "APPROVE", commit_id: $commit_id, body: "Approved automatically for a trusted Dependabot revision."}' \ - | gh api --method POST "repos/$REPO/pulls/$PR_NUMBER/reviews" --input - - - - name: Enable auto-merge (squash) - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - PR_URL: ${{ github.event.pull_request.html_url }} - run: gh pr merge --auto --squash "$PR_URL" + | gh api --method POST "repos/$REPO/pulls/$PR_NUMBER/reviews" --input - -q '.id') + if ! is_current; then + jq -nc '{message: "Head changed during automated review; refusing stale approval."}' \ + | gh api --method PUT "repos/$REPO/pulls/$PR_NUMBER/reviews/$review_id/dismissals" --input - + exit 0 + fi + gh pr merge --auto --squash --match-head-commit "$HEAD_SHA" "$PR_URL" diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index f25b848..2273234 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -6,6 +6,10 @@ on: branches: [main] # Operator recovery when GitHub accepts a main push but drops its workflow dispatch. workflow_dispatch: + # Reconcile bot merges whose GITHUB_TOKEN push cannot trigger another workflow. + # This is eventual recovery, not an ordering dependency between workflows. + schedule: + - cron: "47 3 * * *" # Read-only by default; the release job elevates to contents: write, the publish # job to id-token/attestations for OIDC trusted publishing. diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index 67d74b7..c3b26ff 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -7,6 +7,10 @@ on: branches: [main] # Operator recovery when GitHub accepts a main push but drops its workflow dispatch. workflow_dispatch: + # Reconcile bot merges whose GITHUB_TOKEN push cannot trigger another workflow. + # This is eventual recovery, not an ordering dependency between workflows. + schedule: + - cron: "17 3 * * *" # Read-only by default; jobs opt into the least privilege they need. permissions: @@ -144,92 +148,3 @@ jobs: file=sys.stderr, ) raise SystemExit(1) - - # Unprivileged half of the Dependabot auto-fix flow. Runs the SAME mechanical - # fixers as `gate`, but here in the untrusted pull_request context (no secrets, - # read-only token) and captures the result as a patch artifact. The privileged - # Auto-fix workflow later APPLIES this patch without ever executing PR code, so - # untrusted code never runs in a context that holds a write credential. Gated to - # Dependabot PRs; skipped (and never required) on human PRs. continue-on-error - # keeps a compute hiccup from flipping Validate's conclusion — the failure signal - # that drives Auto-fix stays owned by gate/deps-security/pr-title. - autofix-patch: - name: autofix-patch - runs-on: ubuntu-latest - if: >- - github.event_name == 'pull_request' && - github.event.pull_request.user.login == 'dependabot[bot]' - continue-on-error: true - permissions: - contents: read - steps: - - name: Harden the runner - uses: step-security/harden-runner@ccd8616d44fd3846e67624a50d5aad6d37bf2d25 # v2.22.1 - with: - egress-policy: audit - - - name: Check out the PR head commit - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - with: - # Pin to the head SHA (not the merge ref) so the patch applies cleanly - # onto the same commit the privileged job checks out. - ref: ${{ github.event.pull_request.head.sha }} - persist-credentials: false - - - name: Install uv - uses: astral-sh/setup-uv@1c37ad07a6a961277cf70c0d37d6f313000f5884 # v10.3.0 - with: - version-file: uv.lock - - - name: Compute mechanical fixes as self-contained data - id: patch - # Mirror the auto-fixers exactly (lockfile refresh + every autofix hook). - # prek's exit code is ignored: non-fixable failures (real pytest/pyright - # errors) leave the tree unchanged, so the fix set is empty and nothing is - # committed downstream — the PR stays red for a human. - # - # Emit each changed file as {path, mode, status, base64 content} rather - # than a diff, so the privileged Auto-fix job can rebuild the commit via - # the Git Data API WITHOUT ever checking out, fetching, or executing this - # PR's code. (Assumes paths without embedded tabs/newlines — true for this - # repo's tracked files.) - env: - UV_MALWARE_CHECK: "1" - run: | - uv lock - uv sync - uv run prek run --all-files || true - git add -A - git diff --cached --no-renames --raw > raw.txt - entries='[]' - while IFS="$(printf '\t')" read -r meta path; do - [ -n "$path" ] || continue - status=${meta##* } # last token of the metadata field - # shellcheck disable=SC2086 # intentional split: meta is git-controlled - set -- $meta - mode=$2 # new file mode (000000 for deletions) - if [ "$status" = "D" ]; then - entry=$(jq -nc --arg p "$path" '{path:$p, status:"D"}') - else - entry=$(jq -nc --arg p "$path" --arg m "$mode" --arg s "$status" \ - --arg c "$(base64 -w0 "$path")" \ - '{path:$p, mode:$m, status:$s, content:$c}') - fi - entries=$(jq -c --argjson e "$entry" '. + [$e]' <<<"$entries") - done < raw.txt - printf '%s\n' "$entries" > fixes.json - if [ "$entries" != "[]" ]; then - echo "has_patch=true" >> "$GITHUB_OUTPUT" - else - echo "has_patch=false" >> "$GITHUB_OUTPUT" - echo "No mechanical fixes to apply." >&2 - fi - - - name: Upload the fix set - if: steps.patch.outputs.has_patch == 'true' - uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9 # v7.0.2 - with: - name: autofix-fixes - path: fixes.json - retention-days: 1 - if-no-files-found: error diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index d4817c2..635ef1d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -63,11 +63,17 @@ on four automation behaviors: - **Dependabot auto-merge** (`.github/workflows/dependabot-auto-merge.yml`) supplies the required approval only for a trusted Dependabot revision, then squash-merges it once Validate is green. If Validate fails on a bot PR with a mechanically fixable problem, - **auto-fix** (`.github/workflows/auto-fix.yml`) refreshes the lockfile, applies every - autofix hook, and pushes the result back so the checks re-run - no LLM, purely - mechanical. + **auto-fix** (`.github/workflows/auto-fix.yml`) independently derives bounded Python + fixes with trusted base-locked Ruff and pushes only those changes so checks re-run. + It never consumes PR-produced fixes or executes PR dependencies and hooks. Lockfile + refreshes, control files, new files, and new-tool migrations require reviewed + maintenance. See the [trust boundary](docs/autofix-threat-model.md). -The configuration mirrors the local gate exactly, so "green locally" and "green in CI" mean +Validate and Release also reconcile missed main-push runs daily. This handles +GitHub-token bot merges without granting a stronger merger token; the schedules are +eventual recovery, not an ordering dependency or replacement for required PR checks. + +The validation configuration mirrors the local gate exactly, so "green locally" and "green in CI" mean the same thing. For the agent-facing execution contract, see `AGENTS.md`. ## Temporary dependency risk acceptance @@ -77,12 +83,11 @@ Both required dependency scanners use the same empty-by-default policy in escape hatch for an explicitly reviewed, short-lived **no-fix** case. Prefer a canonical dependency or lockfile update whenever a fixed release is available. Neither Dependabot nor the mechanical auto-fixer may create or extend an acceptance. -The privileged auto-fix consumer prevents direct protected-file writes and verifies paths against the actual PR file tree -before obtaining its write token, and refuses policy, scanner, workflow, packaging, -new-file, symlink, or mode changes. Ordinary existing-file formatting and lockfile -repairs remain eligible. This guard does not prove that arbitrary proposed source -changes are mechanically generated. Both PR and weekly dependency audits run in -a separate fresh job from tests and pre-commit hooks. +The auto-fixer independently derives Python-only changes using trusted base code +and configuration. It never accepts proposed PR artifact bytes, and its allowlist +excludes all policy, scanner, workflow, hook, ownership, packaging, and lock files. +New-file, deletion, symlink, and mode changes are not auto-fixed. Both PR and weekly +dependency audits run in a separate fresh job from tests and pre-commit hooks. Each entry must identify one exact GHSA advisory, `PyPI` package, and version already in `uv.lock`, plus an exclusive UTC `expiresAt`, `reason`, `noFixReason`, diff --git a/docs/autofix-threat-model.md b/docs/autofix-threat-model.md new file mode 100644 index 0000000..5be0fe6 --- /dev/null +++ b/docs/autofix-threat-model.md @@ -0,0 +1,100 @@ +# Auto-fix trust boundary + +## Threat model + +A pull-request job is untrusted, including its workflow, dependencies, installation +scripts, hook configuration, test code, artifacts, and claimed results. A compromised +hook could previously upload arbitrary replacement bytes under the expected artifact +name. Binding that artifact to a Validate run and restricting its file paths did not +prove its contents were mechanical fixes. The Contents-write App could then commit +those bytes and the existing bot approval flow could make them eligible for merge. + +The replacement treats a failed Validate run only as a wake-up signal. It neither +produces nor downloads an auto-fix artifact. No PR-supplied patch, filename list, +configuration, executable, or dependency is accepted as an instruction to the writer. + +Trusted inputs are GitHub's authenticated metadata, the immutable base commit that +contains the running workflow, its pinned Actions, and its locked formatter/tools. +Compromise of GitHub, the trusted base, runner image, or those trusted tool releases +is outside this boundary. The ordinary validation jobs still execute PR code with +read-only permissions, never with the auto-fix App credential. + +## Derivation and scope + +1. Check out only `github.workflow_sha`, with no persisted credentials. Disable the + setup-uv cache and install only the base-locked development tools, without + installing the project or reading a PR lockfile. No PR checkout or hook run occurs. +2. Read the current Validate run and workflow from GitHub. Require the exact workflow + ID and path, completed failed `pull_request` event, same numeric repository ID, + latest triggering attempt, and both actor IDs to be Dependabot's stable ID. +3. Resolve exactly one current PR from GitHub's commit-to-PR association, rather + than the often-empty event PR array. Require it to be open, non-draft, unmerged, + authored by Dependabot, in this repository, targeting the default branch, and + still at the exact triggering SHA and `dependabot/` branch. Human pushes and + human reruns cannot inherit the bot's write path. Stop on an auto-fix head commit. +4. Read complete Git trees and immutable blobs through the API. Only existing regular + `100644` Python files in `src/skill_optimizer/`, `tests/`, and the entry-point shim + may be transformed, and their eligible path sets must match the trusted base. + Reject new/deleted eligible files, symlinks, executable modes, malformed or duplicate + paths, partial trees, bad Git object hashes, and excessive input/output sizes. +5. Feed blob bytes only to the absolute base-installed Ruff binary over stdin. Use + the explicit base `pyproject.toml`, safe lint fixes and formatting, no cache, a + scrubbed environment, and a subprocess timeout. Source content is never imported, + installed, or executed. Formatter diagnostics are captured, not emitted as + workflow commands. Python itself runs in isolated mode. +6. Store only those independently derived replacements in a local plan. Recheck the + run and PR before completing preparation. Empty plans or any error never reach + the token-minting step. The plan is not a downloaded artifact. +7. Mint the existing App's Contents-write token only when fixes exist. Recheck the + current run and PR immediately before publishing. Use GitHub's single + `createCommitOnBranch` mutation with `expectedHeadOid`, which atomically rejects + a moved head, including a reset to an ancestor. No force update or retry occurs. + +The allowlist excludes every workflow, script, hook configuration, `CODEOWNERS`, +Dependabot configuration, security policy, packaging file, lockfile, and document. +Those controls cannot be rewritten by this formatter even if a PR supplies a forged +artifact or names them in source text. Existing required checks and review rules +remain authoritative; an auto-fix commit does not bypass them. + +The branch-head comparison is atomic. PR state and run-attempt metadata are checked +immediately before the mutation but GitHub does not atomically compare those fields +with a commit. Closing a PR in that narrow interval can leave a mechanical commit on +its branch; it cannot cause a write to a different branch or head. API errors, +including GraphQL errors in an HTTP 200 response, fail closed without blind retries. +API redirects are refused, so an Authorization header is not forwarded elsewhere. + +## Intentional maintenance tradeoff + +Automatic lock regeneration, arbitrary pre-commit hooks, Markdown rewriting, and +PR-selected tool versions are no longer eligible. Dependabot still updates its +lockfiles and tool pins, and all ordinary checks remain enabled. Base-pinned Ruff +can repair only the rules it knows: a new formatter release's migration or a genuine +code failure may still need an explicitly reviewed maintenance PR. Using a new +PR-selected package in the writer would recreate the trust boundary we removed. + +The Dependabot approval flow trusts its stable sender IDs, rechecks the current PR +before and after reviewing an exact commit, and requests auto-merge only with a +matching head. If the head changes during review it dismisses that stale approval. The auto-fix App sender remains safe only while every +writer using that identity for Dependabot branches observes this constrained model. +This design does not establish trust in arbitrary changes made by other App writers. + +## Regression evidence + +`tests/test_autofix_policy_guard.py` exercises provenance, object validation, +protected paths, actual formatter nonexecution, empty plans, failure paths, and +atomic expected-head writes using offline API fixtures. The normal required gate +runs these tests and enforces the repository's unchanged coverage threshold. + +The implementation follows GitHub's guidance on +[privileged workflow triggers](https://docs.github.com/en/actions/reference/security/secure-use) +and uses its +[atomic commit mutation](https://docs.github.com/en/graphql/reference/commits#createcommitonbranch). + +## Post-merge reconciliation + +A merge performed with `GITHUB_TOKEN` does not dispatch ordinary push-triggered +workflows. Validate and Release retain their push and manual triggers and also run +daily, at 03:17 and 03:47 UTC respectively, to reconcile a missed main run. These +schedules are eventual recovery, not a guarantee of ordering or a new release gate. +Release retains its existing idempotence, permissions, artifact checks, and PyPI +publishing condition; no new credentials or repository settings are required. diff --git a/scripts/autofix.py b/scripts/autofix.py new file mode 100644 index 0000000..1e72a4c --- /dev/null +++ b/scripts/autofix.py @@ -0,0 +1,304 @@ +"""Derive bounded fixes from trusted tools, never from a PR's artifact or hooks. + +Invoked with Python's isolated mode from the workflow's own immutable base commit. +PR blobs are stdin data to Ruff, never files imported, installed, or executed here. +""" + +import base64 +import hashlib +import json +import os +import re +import subprocess +import sys +import urllib.request +from pathlib import Path +from typing import Any, cast + +DEPENDABOT_ID = 49699333 +MAX_FILE_BYTES = 1_000_000 +MAX_FIX_BYTES = 5_000_000 +MAX_SOURCE_FILES = 1_000 +SHA = re.compile(r"[0-9a-f]{40}") +SOURCE_PATH = re.compile( + r"(?:src/skill_optimizer/|tests/)(?:[A-Za-z0-9_]+/)*[A-Za-z0-9_]+\.py" + r"|optimize_description_v2\.py" +) +COMMIT_MESSAGE = "ci: 🤖 mechanical auto-fix" +MUTATION = """mutation($input: CreateCommitOnBranchInput!) { + createCommitOnBranch(input: $input) { commit { oid } } +}""" + + +def require(condition: object, message: str) -> None: + """Stop before writing whenever the trust contract cannot be established.""" + if not condition: + raise ValueError(message) + + +class RejectRedirect(urllib.request.HTTPRedirectHandler): + """Never forward an API credential to a redirected location.""" + + def redirect_request( + self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str + ) -> None: + """Fail closed even if the original endpoint returns a redirect.""" + raise ValueError("GitHub API redirects are forbidden") + + +class GitHub: + """Use only GitHub's fixed API origin, with a separate write credential.""" + + def __init__(self, repository: str, token: str) -> None: + """Bind the client to the workflow repository and one credential.""" + require( + re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository) + and all(part not in {".", ".."} for part in repository.split("/")), + "Bad repo", + ) + self.repository = repository + self.token = token + + def request(self, path: str, payload: dict[str, Any] | None = None) -> Any: + """Read repository metadata or perform the single explicit GraphQL write.""" + route = ( + "/graphql" if payload is not None else f"/repos/{self.repository}/{path}" + ) + request = urllib.request.Request( + "https://api.github.com" + route, + data=None if payload is None else json.dumps(payload).encode(), + headers={ + "Authorization": f"Bearer {self.token}", + "Accept": "application/vnd.github+json", + "Content-Type": "application/json", + "X-GitHub-Api-Version": "2022-11-28", + }, + ) + # The origin is constant, not a URL taken from PR metadata. + opener = urllib.request.build_opener(RejectRedirect()) + with opener.open(request, timeout=30) as response: + raw: object = json.load(response) + require( + isinstance(raw, (dict, list)), + "GitHub API response must be an object or array", + ) + if isinstance(raw, dict): + require(not cast(dict[str, Any], raw).get("errors"), "GitHub API error") + return cast(Any, raw) + + +def current_pr(api: GitHub, event: dict[str, Any]) -> dict[str, Any]: + """Authenticate the current run attempt and its single live Dependabot PR.""" + original = event["workflow_run"] + run_id = original["id"] + require(type(run_id) is int and run_id > 0, "Invalid run ID") + run = api.request(f"actions/runs/{run_id}") + workflow = api.request("actions/workflows/validate.yml") + repository = event["repository"] + require(repository["full_name"] == api.repository, "Wrong repository") + for candidate in (original, run): + require( + candidate["id"] == run_id + and candidate["event"] == "pull_request" + and candidate["status"] == "completed" + and candidate["conclusion"] == "failure" + and candidate["workflow_id"] == workflow["id"] + and candidate["path"] == ".github/workflows/validate.yml" + and candidate["repository"]["id"] == repository["id"] + and candidate["head_repository"]["id"] == repository["id"] + and candidate["actor"]["id"] == DEPENDABOT_ID + and candidate["triggering_actor"]["id"] == DEPENDABOT_ID, + "Untrusted Validate provenance", + ) + require(workflow["state"] == "active", "Validate workflow is inactive") + require( + run["run_attempt"] == original["run_attempt"] + and run["head_sha"] == original["head_sha"] + and run["head_branch"] == original["head_branch"], + "Stale Validate attempt", + ) + require(SHA.fullmatch(run["head_sha"]), "Invalid head SHA") + # Dependabot workflow_run metadata often has an empty pull_requests array. + # Resolve GitHub's authoritative commit association instead, never an artifact. + raw_association: object = api.request( + f"commits/{run['head_sha']}/pulls?per_page=100" + ) + require(isinstance(raw_association, list), "Invalid PR association") + associated = cast(list[dict[str, Any]], raw_association) + require(len(associated) < 100, "Incomplete PR association") + candidates = [ + pr + for pr in associated + if pr["state"] == "open" and pr["head"]["sha"] == run["head_sha"] + ] + require(len(candidates) == 1, "Ambiguous PR association") + number = candidates[0]["number"] + require(type(number) is int and number > 0, "Invalid PR number") + pr = api.request(f"pulls/{number}") + require( + pr["number"] == number + and pr["state"] == "open" + and not pr["draft"] + and not pr["merged"] + and pr["user"]["id"] == DEPENDABOT_ID + and pr["head"]["repo"]["id"] == repository["id"] + and pr["base"]["repo"]["id"] == repository["id"] + and pr["base"]["ref"] == repository["default_branch"] + and pr["head"]["ref"] == run["head_branch"] + and pr["head"]["sha"] == run["head_sha"] + and pr["head"]["ref"].startswith("dependabot/") + and pr["head"]["ref"] != repository["default_branch"], + "PR is not an eligible current Dependabot revision", + ) + commit = api.request(f"git/commits/{run['head_sha']}") + require(not commit["message"].startswith(COMMIT_MESSAGE), "Auto-fix loop") + return pr + + +def source_tree(tree: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Select only canonical regular Python source files from a complete tree.""" + require(tree["truncated"] is False, "Incomplete source tree") + result: dict[str, dict[str, Any]] = {} + seen: set[str] = set() + for item in tree["tree"]: + path = item["path"] + require( + isinstance(path, str) + and path not in seen + and "\\" not in path + and not any(part in {"", ".", ".."} for part in path.split("/")), + "Malformed or duplicate tree path", + ) + seen.add(path) + require(item["type"] in {"blob", "tree", "commit"}, "Unknown tree object") + if SOURCE_PATH.fullmatch(path): + require( + item["type"] == "blob" and item["mode"] == "100644", + "Nonregular source file", + ) + require(SHA.fullmatch(item["sha"]), "Invalid blob SHA") + result[path] = item + require(len(result) <= MAX_SOURCE_FILES, "Too many source files") + return result + + +def fix_source(root: Path, path: str, content: bytes) -> bytes: + """Run only base-locked Ruff with explicit base configuration and stdin data.""" + command = [str(root / ".venv/bin/ruff")] + common = [ + "--no-cache", + "--config", + str(root / "pyproject.toml"), + "--stdin-filename", + path, + ] + # No shell, module import, filename creation, PR config discovery, or PR hooks. + # The executable/config are from the immutable trusted workflow checkout. + for args in (["check", "--fix-only", "--no-unsafe-fixes"], ["format"]): + result = subprocess.run( # noqa: S603 # nosec B603 + [*command, *args, *common, "-"], + input=content, + capture_output=True, + cwd=root, + env={"PATH": os.defpath, "HOME": str(root)}, + timeout=30, + check=True, + ) + content = result.stdout + return content + + +def build_plan( + api: GitHub, event: dict[str, Any], root: Path, trusted_sha: str +) -> dict[str, Any]: + """Independently derive every output byte; no artifact or proposed patch is read.""" + require(SHA.fullmatch(trusted_sha), "Invalid trusted workflow SHA") + pr = current_pr(api, event) + head = pr["head"]["sha"] + trusted = source_tree(api.request(f"git/trees/{trusted_sha}?recursive=1")) + proposed = source_tree(api.request(f"git/trees/{head}?recursive=1")) + # Source additions/deletions/mode changes require review, not partial automation. + require(trusted.keys() == proposed.keys(), "Source file set or modes changed") + additions: list[dict[str, str]] = [] + total = 0 + for path, item in sorted(proposed.items()): + blob = api.request(f"git/blobs/{item['sha']}") + require( + blob["encoding"] == "base64" and blob["size"] <= MAX_FILE_BYTES, + "Oversized or unsupported blob", + ) + content = base64.b64decode("".join(blob["content"].split()), validate=True) + require(len(content) == blob["size"], "Incorrect blob size") + digest = hashlib.sha1( + b"blob " + str(len(content)).encode() + b"\0" + content, + usedforsecurity=False, + ).hexdigest() + require(digest == item["sha"], "Blob does not match its Git object ID") + updated = fix_source(root, path, content) + require(len(updated) <= MAX_FILE_BYTES, "Oversized formatted source") + if updated != content: + additions.append( + {"path": path, "contents": base64.b64encode(updated).decode()} + ) + total += len(updated) + require(total <= MAX_FIX_BYTES, "Oversized fix set") + current_pr(api, event) + return { + "head": head, + "branch": pr["head"]["ref"], + "number": pr["number"], + "additions": additions, + } + + +def apply_plan( + api: GitHub, writer: GitHub, event: dict[str, Any], plan: dict[str, Any] +) -> str: + """Commit a locally derived plan with an atomic expected-head comparison.""" + pr = current_pr(api, event) + require( + (plan["head"], plan["branch"], plan["number"]) + == (pr["head"]["sha"], pr["head"]["ref"], pr["number"]), + "Stale fix plan", + ) + require(plan["additions"], "Empty fix plan") + result = writer.request( + "", + { + "query": MUTATION, + "variables": { + "input": { + "branch": { + "repositoryNameWithOwner": api.repository, + "branchName": plan["branch"], + }, + "expectedHeadOid": plan["head"], + "message": {"headline": COMMIT_MESSAGE}, + "fileChanges": {"additions": plan["additions"]}, + } + }, + }, + ) + return result["data"]["createCommitOnBranch"]["commit"]["oid"] + + +def main() -> None: + """Prepare with read-only credentials, or publish the trusted local plan.""" + root = Path(__file__).resolve().parent.parent + event = json.loads(Path(os.environ["GITHUB_EVENT_PATH"]).read_text()) + api = GitHub(os.environ["GITHUB_REPOSITORY"], os.environ["GH_READ_TOKEN"]) + plan_path = Path(os.environ["RUNNER_TEMP"]) / "trusted-autofix-plan.json" + if sys.argv[1:] == ["prepare"]: + plan = build_plan(api, event, root, os.environ["TRUSTED_SHA"]) + plan_path.write_text(json.dumps(plan), encoding="utf-8") + with Path(os.environ["GITHUB_OUTPUT"]).open("a", encoding="utf-8") as output: + output.write(f"has_fixes={str(bool(plan['additions'])).lower()}\n") + else: + require(sys.argv[1:] == ["apply"], "Expected prepare or apply") + writer = GitHub(api.repository, os.environ["GH_WRITE_TOKEN"]) + oid = apply_plan(api, writer, event, json.loads(plan_path.read_text())) + print(f"Committed trusted mechanical fixes: {oid}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_autofix_policy_guard.py b/tests/test_autofix_policy_guard.py index 5500b11..d719b09 100644 --- a/tests/test_autofix_policy_guard.py +++ b/tests/test_autofix_policy_guard.py @@ -1,146 +1,834 @@ -"""Execute the trusted workflow's exact guard without credentials or GitHub writes.""" +"""Exercise the trusted auto-fixer with hostile PR data and no network credentials.""" +import base64 +import copy +import hashlib +import io import json import os +import runpy +import shutil import subprocess import sys -import textwrap +import urllib.error +import urllib.request +from email.message import Message from pathlib import Path +from typing import Any +from unittest.mock import Mock import pytest -WORKFLOW = Path(__file__).resolve().parent.parent / ".github/workflows/auto-fix.yml" +from scripts import autofix +ROOT = Path(__file__).resolve().parent.parent +WORKFLOW = ROOT / ".github/workflows/auto-fix.yml" +HEAD = "a" * 40 +TRUSTED = "b" * 40 +REPOSITORY = "owner/optimizer" +SOURCE = "src/skill_optimizer/cli.py" -def _run_guard( - tmp_path: Path, path: str, status: str = "M", mode: str = "100644" -) -> subprocess.CompletedProcess[str]: - workflow = WORKFLOW.read_text(encoding="utf-8") - start = workflow.index(" # BEGIN AUTOFIX SCOPE GUARD") - end = workflow.index(" # END AUTOFIX SCOPE GUARD") - script = textwrap.dedent(workflow[start:end]) - tracked = [ + +def _blob(content: bytes) -> tuple[str, dict[str, Any]]: + digest = hashlib.sha1( + b"blob " + str(len(content)).encode() + b"\0" + content, + usedforsecurity=False, + ).hexdigest() + return digest, { + "encoding": "base64", + "size": len(content), + "content": base64.b64encode(content).decode(), + } + + +def _entry(path: str = SOURCE, content: bytes = b"value=1\n") -> dict[str, Any]: + return {"path": path, "type": "blob", "mode": "100644", "sha": _blob(content)[0]} + + +class FakeGitHub(autofix.GitHub): + """Offline API whose explicit route allowlist detects artifact access and writes.""" + + def __init__(self) -> None: + super().__init__(REPOSITORY, "test-read-token") + self.calls: list[tuple[str, dict[str, Any] | None]] = [] + self.event: dict[str, Any] = { + "repository": { + "id": 700, + "full_name": REPOSITORY, + "default_branch": "main", + }, + "workflow_run": { + "id": 123, + "event": "pull_request", + "status": "completed", + "conclusion": "failure", + "workflow_id": 456, + "path": ".github/workflows/validate.yml", + "repository": {"id": 700}, + "head_repository": {"id": 700}, + "actor": {"id": autofix.DEPENDABOT_ID, "login": "dependabot[bot]"}, + "triggering_actor": { + "id": autofix.DEPENDABOT_ID, + "login": "dependabot[bot]", + }, + "run_attempt": 2, + "head_sha": HEAD, + "head_branch": "dependabot/pip/ruff-1", + "pull_requests": [{"number": 42}], + }, + } + self.run = copy.deepcopy(self.event["workflow_run"]) + self.pr: dict[str, Any] = { + "number": 42, + "state": "open", + "draft": False, + "merged": False, + "user": {"id": autofix.DEPENDABOT_ID, "login": "dependabot[bot]"}, + "head": {"repo": {"id": 700}, "ref": self.run["head_branch"], "sha": HEAD}, + "base": {"repo": {"id": 700}, "ref": "main"}, + } + self.associations: Any = [copy.deepcopy(self.pr)] + self.workflow: dict[str, Any] = {"id": 456, "state": "active"} + self.commit: dict[str, Any] = {"message": "build(deps): bump ruff"} + self.trusted_tree: dict[str, Any] = {"truncated": False, "tree": [_entry()]} + self.proposed_tree = copy.deepcopy(self.trusted_tree) + digest, blob = _blob(b"value=1\n") + self.blobs: dict[str, dict[str, Any]] = {digest: blob} + self.result: dict[str, Any] = { + "data": {"createCommitOnBranch": {"commit": {"oid": "c" * 40}}} + } + + def request(self, path: str, payload: dict[str, Any] | None = None) -> Any: + self.calls.append((path, copy.deepcopy(payload))) + if payload is not None: + assert path == "" + return self.result + routes: dict[str, Any] = { + "actions/runs/123": self.run, + "actions/workflows/validate.yml": self.workflow, + "pulls/42": self.pr, + f"commits/{HEAD}/pulls?per_page=100": self.associations, + f"git/commits/{HEAD}": self.commit, + f"git/trees/{TRUSTED}?recursive=1": self.trusted_tree, + f"git/trees/{HEAD}?recursive=1": self.proposed_tree, + **{f"git/blobs/{sha}": blob for sha, blob in self.blobs.items()}, + } + assert path in routes, f"Unexpected API route: {path}" + return copy.deepcopy(routes[path]) + + +def _set_path(document: dict[str, Any], path: str, value: Any) -> None: + parts = path.split(".") + target = document + for part in parts[:-1]: + target = target[part] + target[parts[-1]] = value + + +@pytest.fixture +def api() -> FakeGitHub: + return FakeGitHub() + + +def test_current_bot_revision_uses_stable_ids_not_display_names( + api: FakeGitHub, +) -> None: + for source in (api.run, api.event["workflow_run"]): + source["actor"]["login"] = "renamed-account" + source["triggering_actor"]["login"] = "renamed-account" + api.pr["user"]["login"] = "renamed-account" + assert autofix.current_pr(api, api.event) == api.pr + assert [path for path, _ in api.calls] == [ + "actions/runs/123", + "actions/workflows/validate.yml", + f"commits/{HEAD}/pulls?per_page=100", + "pulls/42", + f"git/commits/{HEAD}", + ] + + +@pytest.mark.parametrize("run_id", [0, -1, True, False, "123", 123.0, None]) +def test_run_identifier_is_a_positive_json_integer( + api: FakeGitHub, run_id: Any +) -> None: + api.event["workflow_run"]["id"] = run_id + with pytest.raises(ValueError, match="Invalid run ID"): + autofix.current_pr(api, api.event) + assert api.calls == [] + + +@pytest.mark.parametrize("source", ["event", "fresh"]) +@pytest.mark.parametrize( + "path,value", + [ + ("event", "push"), + ("status", "in_progress"), + ("conclusion", "success"), + ("workflow_id", 999), + ("path", ".github/workflows/untrusted.yml"), + ("repository.id", 701), + ("head_repository.id", 701), + ("actor.id", 900), + ("triggering_actor.id", 900), + ], +) +def test_original_and_fresh_run_must_have_trusted_provenance( + api: FakeGitHub, source: str, path: str, value: Any +) -> None: + _set_path(api.event["workflow_run"] if source == "event" else api.run, path, value) + with pytest.raises(ValueError, match="Untrusted Validate provenance"): + autofix.current_pr(api, api.event) + + +@pytest.mark.parametrize( + "scenario,message", + [ + ("run_id", "Untrusted Validate provenance"), + ("repository", "Wrong repository"), + ("inactive", "Validate workflow is inactive"), + ("attempt", "Stale Validate attempt"), + ("head", "Stale Validate attempt"), + ("branch", "Stale Validate attempt"), + ("invalid_sha", "Invalid head SHA"), + ("no_pr", "Ambiguous PR association"), + ("multiple_prs", "Ambiguous PR association"), + ("loop", "Auto-fix loop"), + ], +) +def test_stale_ambiguous_or_replayed_run_is_rejected( + api: FakeGitHub, scenario: str, message: str +) -> None: + if scenario == "run_id": + api.run["id"] = 124 + elif scenario == "repository": + api.event["repository"]["full_name"] = "attacker/optimizer" + elif scenario == "inactive": + api.workflow["state"] = "disabled_manually" + elif scenario == "attempt": + api.run["run_attempt"] += 1 + elif scenario == "head": + api.run["head_sha"] = "d" * 40 + elif scenario == "branch": + api.run["head_branch"] = "dependabot/other" + elif scenario == "invalid_sha": + api.run["head_sha"] = api.event["workflow_run"]["head_sha"] = "not-a-sha" + elif scenario == "no_pr": + api.associations = [] + elif scenario == "multiple_prs": + api.associations.append({**api.associations[0], "number": 43}) + else: + api.commit["message"] = autofix.COMMIT_MESSAGE + "\n\nRepeated trigger" + with pytest.raises(ValueError, match=message): + autofix.current_pr(api, api.event) + + +@pytest.mark.parametrize("number", [0, -1, True, False, "42", 42.0, None]) +def test_pr_number_is_a_positive_json_integer(api: FakeGitHub, number: Any) -> None: + api.associations[0]["number"] = number + with pytest.raises(ValueError, match="Invalid PR number"): + autofix.current_pr(api, api.event) + + +@pytest.mark.parametrize( + "path,value", + [ + ("number", 43), + ("state", "closed"), + ("draft", True), + ("merged", True), + ("user.id", 900), + ("head.repo.id", 701), + ("base.repo.id", 701), + ("base.ref", "release"), + ("head.ref", "dependabot/other"), + ("head.sha", "d" * 40), + ], +) +def test_pr_must_remain_open_current_and_same_repository( + api: FakeGitHub, path: str, value: Any +) -> None: + _set_path(api.pr, path, value) + with pytest.raises(ValueError, match="eligible current Dependabot"): + autofix.current_pr(api, api.event) + + +@pytest.mark.parametrize("branch", ["feature/pretend-bot", "dependabot/default"]) +def test_non_bot_or_default_branch_is_never_writable( + api: FakeGitHub, branch: str +) -> None: + api.pr["head"]["ref"] = branch + api.run["head_branch"] = api.event["workflow_run"]["head_branch"] = branch + if branch == "dependabot/default": + api.event["repository"]["default_branch"] = branch + api.pr["base"]["ref"] = branch + with pytest.raises(ValueError, match="eligible current Dependabot"): + autofix.current_pr(api, api.event) + + +@pytest.mark.parametrize( + "path", + [ "uv.lock", - "src/skill_optimizer/cli.py", - "tests/test_cli.py", + "pyproject.toml", "README.md", ".pre-commit-config.yaml", + ".github/workflows/auto-fix.yml", + ".github/workflows/validate.yml", ".github/advisory-exceptions.json", ".github/osv-scanner-empty.toml", - ".github/security-scanner/Dockerfile", "ci/security-scanner/Dockerfile", - ".github/workflows/validate.yml", - ".github/workflows/auto-fix.yml", + "scripts/autofix.py", "scripts/advisory_policy.py", - "scripts/__init__.py", - "scripts/uv_advisory_policy.py", - "scripts/audit-dependencies.sh", - "pyproject.toml", - ] - (tmp_path / "trusted-tree.json").write_text( - json.dumps( - { - "truncated": False, - "tree": [ - {"type": "blob", "mode": "100644", "path": name} for name in tracked - ], - } - ), - encoding="utf-8", + "sitecustomize.py", + "json.py", + ".venv/bin/ruff", + "src/skill_optimizer/evil.py/child.py", + "src/skill_optimizer/evil.py\n", + "src/skill_optimizer/evil\x00.py", + "src/skill_optimizer/évil.py", + "tests/.hidden.py", + "tests/evil-name.py", + ], +) +def test_controls_and_noncanonical_paths_are_never_formatted( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, path: str +) -> None: + api.proposed_tree["tree"].append(_entry(path)) + formatter = Mock(return_value=b"value = 1\n") + monkeypatch.setattr(autofix, "fix_source", formatter) + plan = autofix.build_plan(api, api.event, ROOT, TRUSTED) + assert [item["path"] for item in plan["additions"]] == [SOURCE] + formatter.assert_called_once_with(ROOT, SOURCE, b"value=1\n") + assert not any( + payload is not None or "artifacts" in path for path, payload in api.calls ) - (tmp_path / "fixes").mkdir() - (tmp_path / "fixes/fixes.json").write_text( - json.dumps( - [{"path": path, "status": status, "mode": mode, "content": "dGVzdA=="}] - ), + + +@pytest.mark.parametrize( + "path", + [SOURCE, "tests/test_cli.py", "tests/unit/test_1.py", "optimize_description_v2.py"], +) +def test_canonical_regular_python_paths_are_selected(path: str) -> None: + entry = _entry(path) + assert autofix.source_tree({"truncated": False, "tree": [entry]}) == {path: entry} + + +@pytest.mark.parametrize( + "mode,kind", + [("120000", "blob"), ("100755", "blob"), ("160000", "commit"), ("040000", "tree")], +) +def test_symlink_executable_submodule_and_directory_are_not_source( + mode: str, kind: str +) -> None: + entry = {**_entry(), "mode": mode, "type": kind} + with pytest.raises(ValueError, match="Nonregular source file"): + autofix.source_tree({"truncated": False, "tree": [entry]}) + + +@pytest.mark.parametrize( + "change", ["added", "deleted", "symlink", "executable", "submodule"] +) +def test_source_set_and_mode_changes_require_review( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, change: str +) -> None: + if change == "added": + api.proposed_tree["tree"].append(_entry("tests/new_test.py")) + elif change == "deleted": + api.proposed_tree["tree"] = [] + elif change == "symlink": + api.proposed_tree["tree"][0]["mode"] = "120000" + elif change == "executable": + api.proposed_tree["tree"][0]["mode"] = "100755" + else: + api.proposed_tree["tree"][0].update(type="commit", mode="160000") + formatter = Mock() + monkeypatch.setattr(autofix, "fix_source", formatter) + with pytest.raises( + ValueError, match=r"Source file set or modes changed|Nonregular source file" + ): + autofix.build_plan(api, api.event, ROOT, TRUSTED) + formatter.assert_not_called() + assert not any(path.startswith("git/blobs/") for path, _ in api.calls) + + +@pytest.mark.parametrize("truncated", [True, None, 0, "false"]) +def test_truncated_or_nonboolean_tree_is_rejected(truncated: Any) -> None: + with pytest.raises(ValueError, match="Incomplete source tree"): + autofix.source_tree({"truncated": truncated, "tree": []}) + + +def test_source_difference_is_mechanically_derived_not_copied_from_artifact( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + content = b"value=2\n" + digest, blob = _blob(content) + api.proposed_tree["tree"][0]["sha"] = digest + api.blobs[digest] = blob + (tmp_path / "fixes.json").write_text( + json.dumps([{"path": ".github/workflows/validate.yml", "content": "poison"}]), encoding="utf-8", ) - return subprocess.run( - [sys.executable, "-c", script], - cwd=tmp_path, - check=False, - capture_output=True, - text=True, - env={**os.environ, "GITHUB_OUTPUT": str(tmp_path / "output")}, + formatter = Mock(return_value=b"value = 2\n") + monkeypatch.setattr(autofix, "fix_source", formatter) + plan = autofix.build_plan(api, api.event, tmp_path, TRUSTED) + assert plan == { + "head": HEAD, + "branch": api.pr["head"]["ref"], + "number": 42, + "additions": [ + {"path": SOURCE, "contents": base64.b64encode(b"value = 2\n").decode()} + ], + } + formatter.assert_called_once_with(tmp_path, SOURCE, content) + assert [path for path, _ in api.calls].count("actions/runs/123") == 2 + assert not any( + payload is not None or "artifacts" in path for path, payload in api.calls ) @pytest.mark.parametrize( - "path", + "trusted", ["", "../main", "c" * 39, "G" * 40, "c" * 40 + "\n"] +) +def test_trusted_commit_is_an_immutable_canonical_sha( + api: FakeGitHub, trusted: str +) -> None: + with pytest.raises(ValueError, match="Invalid trusted workflow SHA"): + autofix.build_plan(api, api.event, ROOT, trusted) + assert api.calls == [] + + +def test_blob_identifier_must_be_a_canonical_sha(api: FakeGitHub) -> None: + api.proposed_tree["tree"][0]["sha"] = "../../malicious" + with pytest.raises(ValueError, match="Invalid blob SHA"): + autofix.build_plan(api, api.event, ROOT, TRUSTED) + + +@pytest.mark.parametrize( + "change,value,message", [ - "uv.lock", - "src/skill_optimizer/cli.py", - "tests/test_cli.py", - "README.md", - ".pre-commit-config.yaml", + ("encoding", "utf-8", "Oversized or unsupported blob"), + ("size", autofix.MAX_FILE_BYTES + 1, "Oversized or unsupported blob"), + ("size", -1, "Incorrect blob size"), + ("size", 1, "Incorrect blob size"), + ("content", "not!base64", ""), + ("content", base64.b64encode(b"value=2\n").decode(), "Git object ID"), ], ) -def test_ordinary_existing_mechanical_repairs_remain_allowed( - tmp_path: Path, path: str +def test_corrupt_unsupported_or_oversized_blob_is_never_formatted( + api: FakeGitHub, + monkeypatch: pytest.MonkeyPatch, + change: str, + value: Any, + message: str, +) -> None: + api.blobs[_blob(b"value=1\n")[0]][change] = value + formatter = Mock() + monkeypatch.setattr(autofix, "fix_source", formatter) + with pytest.raises(ValueError, match=message or None): + autofix.build_plan(api, api.event, ROOT, TRUSTED) + formatter.assert_not_called() + + +def test_base64_line_wrapping_is_accepted_and_identical_bytes_produce_no_writes( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch +) -> None: + blob = api.blobs[_blob(b"value=1\n")[0]] + blob["content"] = "\n".join(blob["content"]) + "\n" + monkeypatch.setattr(autofix, "fix_source", Mock(return_value=b"value=1\n")) + plan = autofix.build_plan(api, api.event, ROOT, TRUSTED) + assert plan["additions"] == [] + writer = FakeGitHub() + with pytest.raises(ValueError, match="Empty fix plan"): + autofix.apply_plan(api, writer, api.event, plan) + assert writer.calls == [] + + +@pytest.mark.parametrize("limit", ["file", "total"]) +def test_output_byte_budgets_are_enforced( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, limit: str +) -> None: + monkeypatch.setattr(autofix, "fix_source", Mock(return_value=b"value = 1\n")) + monkeypatch.setattr( + autofix, "MAX_FILE_BYTES" if limit == "file" else "MAX_FIX_BYTES", 8 + ) + with pytest.raises( + ValueError, + match="Oversized formatted source" if limit == "file" else "Oversized fix set", + ): + autofix.build_plan(api, api.event, ROOT, TRUSTED) + + +def test_empty_source_tree_is_a_valid_noop(api: FakeGitHub) -> None: + api.trusted_tree["tree"] = api.proposed_tree["tree"] = [] + assert autofix.build_plan(api, api.event, ROOT, TRUSTED)["additions"] == [] + + +def test_plan_fails_if_head_changes_during_formatting( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch ) -> None: - result = _run_guard(tmp_path, path) - assert result.returncode == 0, result.stderr - assert (tmp_path / "output").read_text(encoding="utf-8") == "safe=true\n" + def advance_head(root: Path, path: str, content: bytes) -> bytes: + api.pr["head"]["sha"] = "d" * 40 + return content + + monkeypatch.setattr(autofix, "fix_source", advance_head) + with pytest.raises(ValueError, match="eligible current Dependabot"): + autofix.build_plan(api, api.event, ROOT, TRUSTED) + assert not any(payload is not None for _, payload in api.calls) + + +def _plan(api: FakeGitHub) -> dict[str, Any]: + return { + "head": HEAD, + "branch": api.pr["head"]["ref"], + "number": 42, + "additions": [{"path": SOURCE, "contents": "dmFsdWUgPSAxCg=="}], + } + + +def test_publish_uses_one_atomic_commit_with_expected_head(api: FakeGitHub) -> None: + writer = FakeGitHub() + plan = _plan(api) + assert autofix.apply_plan(api, writer, api.event, plan) == "c" * 40 + assert writer.calls == [ + ( + "", + { + "query": autofix.MUTATION, + "variables": { + "input": { + "branch": { + "repositoryNameWithOwner": REPOSITORY, + "branchName": plan["branch"], + }, + "expectedHeadOid": HEAD, + "message": {"headline": autofix.COMMIT_MESSAGE}, + "fileChanges": {"additions": plan["additions"]}, + } + }, + }, + ) + ] + assert not any(payload is not None for _, payload in api.calls) @pytest.mark.parametrize( - "path", - [ - ".github/advisory-exceptions.json", - ".github/osv-scanner-empty.toml", - ".github/security-scanner/Dockerfile", - "ci/security-scanner/Dockerfile", - ".github/workflows/validate.yml", - ".github/workflows/auto-fix.yml", - "scripts/advisory_policy.py", - "scripts/__init__.py", - "scripts/uv_advisory_policy.py", - "scripts/audit-dependencies.sh", + "key,value", [("head", "d" * 40), ("branch", "dependabot/other"), ("number", 43)] +) +def test_stale_plan_never_reaches_writer(api: FakeGitHub, key: str, value: Any) -> None: + plan = _plan(api) + plan[key] = value + writer = FakeGitHub() + with pytest.raises(ValueError, match="Stale fix plan"): + autofix.apply_plan(api, writer, api.event, plan) + assert writer.calls == [] + + +def test_compare_and_swap_rejection_never_retries_or_force_pushes( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch +) -> None: + writer = FakeGitHub() + request = Mock(side_effect=ValueError("GitHub API error")) + monkeypatch.setattr(writer, "request", request) + with pytest.raises(ValueError, match="GitHub API error"): + autofix.apply_plan(api, writer, api.event, _plan(api)) + assert request.call_count == 1 + assert request.call_args.args[1]["variables"]["input"]["expectedHeadOid"] == HEAD + + +def test_ruff_command_is_base_owned_stdin_only_and_has_no_token_environment( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setenv("GH_READ_TOKEN", "never-forward-this") + monkeypatch.setenv("GH_WRITE_TOKEN", "never-forward-this-either") + monkeypatch.setenv("PYTHONPATH", "/attacker") + process = Mock( + side_effect=[ + subprocess.CompletedProcess([], 0, b"intermediate", b""), + subprocess.CompletedProcess([], 0, b"formatted", b""), + ] + ) + monkeypatch.setattr(autofix.subprocess, "run", process) + assert autofix.fix_source(tmp_path, SOURCE, b"untrusted source") == b"formatted" + common = [ + "--no-cache", + "--config", + str(tmp_path / "pyproject.toml"), + "--stdin-filename", + SOURCE, + "-", + ] + assert process.call_args_list[0].args[0] == [ + str(tmp_path / ".venv/bin/ruff"), + "check", + "--fix-only", + "--no-unsafe-fixes", + *common, + ] + assert process.call_args_list[1].args[0] == [ + str(tmp_path / ".venv/bin/ruff"), + "format", + *common, + ] + for call, expected_input in zip( + process.call_args_list, [b"untrusted source", b"intermediate"], strict=True + ): + assert call.kwargs == { + "input": expected_input, + "capture_output": True, + "cwd": tmp_path, + "env": {"PATH": os.defpath, "HOME": str(tmp_path)}, + "timeout": 30, + "check": True, + } + + +@pytest.mark.parametrize( + "error", + [subprocess.CalledProcessError(1, "ruff"), subprocess.TimeoutExpired("ruff", 30)], +) +def test_ruff_failure_or_timeout_cannot_create_a_plan( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, error: Exception +) -> None: + process = Mock(side_effect=error) + monkeypatch.setattr(autofix.subprocess, "run", process) + with pytest.raises(type(error)): + autofix.build_plan(api, api.event, ROOT, TRUSTED) + assert process.call_count == 1 + assert not any(payload is not None for _, payload in api.calls) + + +def test_real_ruff_does_not_execute_pr_code_hooks_shadow_imports_or_config( + api: FakeGitHub, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + trusted = tmp_path / "trusted" + (trusted / ".venv/bin").mkdir(parents=True) + (trusted / ".venv/bin/ruff").symlink_to(ROOT / ".venv/bin/ruff") + shutil.copyfile(ROOT / "pyproject.toml", trusted / "pyproject.toml") + hostile = tmp_path / "hostile" + hostile.mkdir() + sentinel = tmp_path / "PR_CODE_EXECUTED" + malicious = f"from pathlib import Path\nPath({str(sentinel)!r}).write_text('executed')\nvalues=[1,2]\n".encode() + digest, blob = _blob(malicious) + api.blobs[digest] = blob + api.proposed_tree["tree"][0]["sha"] = digest + for path in ( "pyproject.toml", - "scripts//advisory_policy.py", - "./scripts/advisory_policy.py", - "src/../scripts/advisory_policy.py", - "/scripts/advisory_policy.py", - "scripts\\advisory_policy.py", - "", + ".pre-commit-config.yaml", "sitecustomize.py", "json.py", - ".venv/lib/python3.14/site-packages/sitecustomize.py", + "scripts/autofix.py", + ".venv/bin/ruff", + ): + api.proposed_tree["tree"].append(_entry(path, malicious)) + (hostile / "sitecustomize.py").write_bytes(malicious) + (hostile / "json.py").write_bytes(malicious) + (hostile / "pyproject.toml").write_text("THIS IS NOT VALID TOML", encoding="utf-8") + monkeypatch.setenv("PYTHONPATH", str(hostile)) + monkeypatch.chdir(hostile) + plan = autofix.build_plan(api, api.event, trusted, TRUSTED) + assert [item["path"] for item in plan["additions"]] == [SOURCE] + output = base64.b64decode(plan["additions"][0]["contents"]) + assert b"values = [1, 2]" in output + assert str(sentinel).encode() in output + assert not sentinel.exists() + assert not (trusted / SOURCE).exists() + assert (hostile / "pyproject.toml").read_text() == "THIS IS NOT VALID TOML" + + +@pytest.mark.parametrize( + "repository", + [ + "", + "owner", + "owner/..", + "./repo", + "owner/repo/extra", + "../repo", + "https://evil.test/repo", + "owner/repo\n", + "owner/rep?x", ], ) -def test_protected_and_noncanonical_paths_never_gain_write_token( - tmp_path: Path, path: str +def test_api_repository_cannot_select_an_external_origin(repository: str) -> None: + with pytest.raises(ValueError, match="Bad repo"): + autofix.GitHub(repository, "secret") + + +@pytest.mark.parametrize("payload", [None, {"query": "query { viewer { login } }"}]) +@pytest.mark.parametrize("response", [{"ok": True}, [], {"errors": []}]) +def test_api_uses_fixed_https_origin_headers_and_json( + monkeypatch: pytest.MonkeyPatch, payload: dict[str, Any] | None, response: Any ) -> None: - result = _run_guard(tmp_path, path) - assert result.returncode != 0 - assert not (tmp_path / "output").exists() + opener = Mock(return_value=io.BytesIO(json.dumps(response).encode())) + monkeypatch.setattr( + urllib.request, "build_opener", Mock(return_value=Mock(open=opener)) + ) + client = autofix.GitHub(REPOSITORY, "private-token") + assert client.request("https://attacker.invalid/path", payload) == response + request = opener.call_args.args[0] + assert request.full_url == ( + "https://api.github.com/graphql" + if payload is not None + else f"https://api.github.com/repos/{REPOSITORY}/https://attacker.invalid/path" + ) + assert request.get_header("Authorization") == "Bearer private-token" + assert request.get_header("Accept") == "application/vnd.github+json" + assert request.get_header("X-github-api-version") == "2022-11-28" + assert request.data == (None if payload is None else json.dumps(payload).encode()) + assert opener.call_args.kwargs == {"timeout": 30} + + +def test_graphql_error_hides_server_supplied_secrets( + monkeypatch: pytest.MonkeyPatch, +) -> None: + opener = Mock( + return_value=io.BytesIO(b'{"errors": [{"message": "private-token"}]}') + ) + monkeypatch.setattr( + urllib.request, "build_opener", Mock(return_value=Mock(open=opener)) + ) + with pytest.raises(ValueError, match=r"^GitHub API error$"): + autofix.GitHub(REPOSITORY, "private-token").request("", {"query": "mutation"}) @pytest.mark.parametrize( - "status,mode", [("A", "100644"), ("D", "000000"), ("T", "120000"), ("M", "120000")] + "error", + [ + urllib.error.HTTPError( + "https://api.github.com/test", 403, "Forbidden", Message(), None + ), + urllib.error.URLError("connection failed"), + ], ) -def test_new_files_deletions_and_symlinks_require_review( - tmp_path: Path, status: str, mode: str +def test_http_failure_is_not_retried_and_does_not_print_token( + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + error: Exception, ) -> None: - assert _run_guard(tmp_path, "sitecustomize.py", status, mode).returncode != 0 - assert not (tmp_path / "output").exists() + opener = Mock(side_effect=error) + monkeypatch.setattr( + urllib.request, "build_opener", Mock(return_value=Mock(open=opener)) + ) + with pytest.raises(type(error)) as caught: + autofix.GitHub(REPOSITORY, "private-token").request("pulls/42") + assert "private-token" not in str(caught.value) + assert opener.call_count == 1 + assert capsys.readouterr() == ("", "") + + +def _wrapper_environment( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> Path: + event = tmp_path / "event.json" + event.write_text(json.dumps(api.event), encoding="utf-8") + for key, value in { + "GITHUB_EVENT_PATH": str(event), + "GITHUB_REPOSITORY": REPOSITORY, + "GH_READ_TOKEN": "read-only-token", + "RUNNER_TEMP": str(tmp_path), + "TRUSTED_SHA": TRUSTED, + "GITHUB_OUTPUT": str(tmp_path / "github-output"), + }.items(): + monkeypatch.setenv(key, value) + monkeypatch.delenv("GH_WRITE_TOKEN", raising=False) + def request( + self: autofix.GitHub, path: str, payload: dict[str, Any] | None = None + ) -> Any: + return api.request(path, payload) -def test_write_steps_require_successful_scope_guard() -> None: + monkeypatch.setattr(autofix.GitHub, "request", request) + return tmp_path / "trusted-autofix-plan.json" + + +@pytest.mark.parametrize("changed", [False, True]) +def test_prepare_wrapper_writes_only_locally_derived_plan_without_write_token( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, tmp_path: Path, changed: bool +) -> None: + plan_path = _wrapper_environment(api, monkeypatch, tmp_path) + plan_path.write_text('{"additions": [{"path": "evil.py"}]}', encoding="utf-8") + monkeypatch.setattr(sys, "argv", ["autofix.py", "prepare"]) + monkeypatch.setattr( + autofix, + "fix_source", + Mock(return_value=b"value = 1\n" if changed else b"value=1\n"), + ) + autofix.main() + plan = json.loads(plan_path.read_text()) + assert plan["head"] == HEAD + assert bool(plan["additions"]) is changed + assert ( + tmp_path / "github-output" + ).read_text() == f"has_fixes={str(changed).lower()}\n" + assert not any(payload is not None for _, payload in api.calls) + assert "GH_WRITE_TOKEN" not in os.environ + + +def test_apply_wrapper_reads_local_plan_and_publishes_once( + api: FakeGitHub, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + plan_path = _wrapper_environment(api, monkeypatch, tmp_path) + plan_path.write_text(json.dumps(_plan(api)), encoding="utf-8") + monkeypatch.setenv("GH_WRITE_TOKEN", "write-only-token") + monkeypatch.setattr(sys, "argv", ["autofix.py", "apply"]) + autofix.main() + assert ( + capsys.readouterr().out == f"Committed trusted mechanical fixes: {'c' * 40}\n" + ) + assert sum(payload is not None for _, payload in api.calls) == 1 + + +@pytest.mark.parametrize("arguments", [[], ["publish"], ["prepare", "apply"]]) +def test_wrapper_rejects_unknown_mode_before_requesting_write_token( + api: FakeGitHub, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + arguments: list[str], +) -> None: + _wrapper_environment(api, monkeypatch, tmp_path) + monkeypatch.setattr(sys, "argv", ["autofix.py", *arguments]) + with pytest.raises(ValueError, match="Expected prepare or apply"): + autofix.main() + assert api.calls == [] + + +def test_script_entrypoint_executes_wrapper( + api: FakeGitHub, monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + _wrapper_environment(api, monkeypatch, tmp_path) + monkeypatch.setattr(sys, "argv", ["autofix.py", "unknown"]) + with pytest.raises(ValueError, match="Expected prepare or apply"): + runpy.run_path(str(ROOT / "scripts/autofix.py"), run_name="__main__") + + +def test_workflow_uses_base_code_and_no_pr_artifacts_or_execution() -> None: workflow = WORKFLOW.read_text(encoding="utf-8") + assert "ref: ${{ github.workflow_sha }}" in workflow + assert "persist-credentials: false" in workflow + assert ( + "uv sync --locked --only-group dev --no-install-project --no-cache" in workflow + ) + assert "python -I scripts/autofix.py prepare" in workflow + assert "python -I scripts/autofix.py apply" in workflow + assert "download-artifact" not in workflow + assert "upload-artifact" not in workflow + assert "git apply" not in workflow + assert "git push" not in workflow + assert "prek run" not in workflow + assert "uv run pytest" not in workflow + assert "ref: ${{ github.event.workflow_run.head_sha }}" not in workflow for step in ( "Mint a GitHub App token", - "Rebuild the fix commit via the Git Data API and update the branch", + "Revalidate and atomically commit the locally derived fixes", ): - assert f"- name: {step}\n" in workflow section = workflow.split(f"- name: {step}\n", 1)[1].split(" - name:", 1)[0] - assert "if: steps.scope.outputs.safe == 'true'" in section - assert workflow.index("- name: Read the trusted PR file tree") < workflow.index( - "- name: Validate mechanical fix scope" - ) - assert workflow.index("- name: Validate mechanical fix scope") < workflow.index( - "- name: Mint a GitHub App token" - ) + assert "if: steps.fixes.outputs.has_fixes == 'true'" in section + assert workflow.index( + "- name: Independently derive bounded fixes" + ) < workflow.index("- name: Mint a GitHub App token") def test_both_dependency_scans_are_isolated_from_test_and_hook_execution() -> None: @@ -158,3 +846,185 @@ def test_both_dependency_scans_are_isolated_from_test_and_hook_execution() -> No full_gate = weekly.split(" full-gate:\n", 1)[1].split(" deps-security:\n", 1)[0] assert "prek run" in full_gate assert "audit-dependencies.sh" not in full_gate + + +@pytest.mark.parametrize( + "path", + [ + None, + 12, + "", + "/tests/file.py", + "./tests/file.py", + "tests//file.py", + "tests/../file.py", + "tests\\file.py", + ], +) +def test_malformed_tree_paths_fail_closed(path: Any) -> None: + entry = {**_entry(), "path": path} + with pytest.raises(ValueError, match="Malformed or duplicate tree path"): + autofix.source_tree({"truncated": False, "tree": [entry]}) + + +def test_duplicate_tree_paths_cannot_hide_a_symlink_or_alternate_blob() -> None: + entries = [_entry(), {**_entry(), "mode": "120000", "sha": "e" * 40}] + with pytest.raises(ValueError, match="Malformed or duplicate tree path"): + autofix.source_tree({"truncated": False, "tree": entries}) + + +def test_unknown_tree_objects_fail_closed_even_outside_write_scope() -> None: + with pytest.raises(ValueError, match="Unknown tree object"): + autofix.source_tree( + {"truncated": False, "tree": [{"path": "README.md", "type": "surprise"}]} + ) + + +@pytest.mark.parametrize( + "tree", + [ + {}, + {"truncated": False}, + {"truncated": False, "tree": [dict[str, Any]()]}, + {"truncated": False, "tree": None}, + ], +) +def test_malformed_tree_objects_never_produce_a_plan(tree: dict[str, Any]) -> None: + with pytest.raises((KeyError, TypeError)): + autofix.source_tree(tree) + + +def test_source_file_count_is_bounded_before_blob_downloads(api: FakeGitHub) -> None: + api.proposed_tree["tree"] = [ + _entry(f"tests/test_{i}.py") for i in range(autofix.MAX_SOURCE_FILES + 1) + ] + with pytest.raises(ValueError, match="Too many source files"): + autofix.build_plan(api, api.event, ROOT, TRUSTED) + assert not any(path.startswith("git/blobs/") for path, _ in api.calls) + assert ( + len( + autofix.source_tree( + {"truncated": False, "tree": api.proposed_tree["tree"][:-1]} + ) + ) + == autofix.MAX_SOURCE_FILES + ) + + +def test_empty_workflow_pr_metadata_uses_authoritative_commit_association( + api: FakeGitHub, +) -> None: + api.run["pull_requests"] = api.event["workflow_run"]["pull_requests"] = [] + api.associations.extend( + [ + {"number": 40, "state": "closed", "head": {"sha": HEAD}}, + {"number": 41, "state": "open", "head": {"sha": "e" * 40}}, + ] + ) + assert autofix.current_pr(api, api.event) == api.pr + + +@pytest.mark.parametrize( + "associations,message", + [ + ({}, "Invalid PR association"), + (None, "Invalid PR association"), + ([dict[str, Any]()] * 100, "Incomplete PR association"), + ], +) +def test_invalid_or_paginated_commit_associations_fail_closed( + api: FakeGitHub, associations: Any, message: str +) -> None: + api.associations = associations + with pytest.raises(ValueError, match=message): + autofix.current_pr(api, api.event) + + +@pytest.mark.parametrize("response", [None, True, 3, "not-an-object"]) +def test_scalar_api_response_is_rejected( + monkeypatch: pytest.MonkeyPatch, response: Any +) -> None: + opener = Mock(open=Mock(return_value=io.BytesIO(json.dumps(response).encode()))) + monkeypatch.setattr(urllib.request, "build_opener", Mock(return_value=opener)) + with pytest.raises( + ValueError, match="GitHub API response must be an object or array" + ): + autofix.GitHub(REPOSITORY, "private-token").request("pulls/42") + + +def test_api_redirect_is_rejected_without_retransmitting_authorization( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class RedirectingTransport: + def __init__(self, handler: autofix.RejectRedirect) -> None: + self.handler = handler + + def open(self, request: urllib.request.Request, timeout: int) -> Any: + assert timeout == 30 + assert request.get_header("Authorization") == "Bearer private-token" + return self.handler.redirect_request( + request, None, 302, "Found", {}, "https://attacker.invalid/steal" + ) + + factory = Mock(side_effect=RedirectingTransport) + monkeypatch.setattr(urllib.request, "build_opener", factory) + with pytest.raises( + ValueError, match="GitHub API redirects are forbidden" + ) as caught: + autofix.GitHub(REPOSITORY, "private-token").request("pulls/42") + assert "private-token" not in str(caught.value) + factory.assert_called_once() + assert isinstance(factory.call_args.args[0], autofix.RejectRedirect) + + +def test_daily_reconciliation_preserves_push_manual_and_pypi_gates() -> None: + validate = (WORKFLOW.parent / "validate.yml").read_text(encoding="utf-8") + release = (WORKFLOW.parent / "release.yml").read_text(encoding="utf-8") + for text, cron in ((validate, '"17 3 * * *"'), (release, '"47 3 * * *"')): + triggers = text.split("on:\n", 1)[1].split("\npermissions:", 1)[0] + assert " push:\n branches: [main]" in triggers + assert " workflow_dispatch:" in triggers + assert f" schedule:\n - cron: {cron}" in triggers + assert ( + " pull_request:" in validate.split("on:\n", 1)[1].split("\npermissions:", 1)[0] + ) + assert "if: needs.release.outputs.released == 'true'" in release + assert "if: vars.PUBLISH_TO_PYPI == 'true'" in release + assert "ref: ${{ needs.release.outputs.tag }}" in release + + +def test_autoapproval_rechecks_current_head_and_binds_review_and_merge() -> None: + workflow = (WORKFLOW.parent / "dependabot-auto-merge.yml").read_text( + encoding="utf-8" + ) + assert "github.event.pull_request.user.id == 49699333" in workflow + assert "github.event.sender.id == 49699333" in workflow + assert "github.event.sender.id == 305056826" in workflow + assert ( + "github.repository == github.event.pull_request.head.repo.full_name" in workflow + ) + step = workflow.split( + "- name: Approve and enable auto-merge for the current revision\n", 1 + )[1] + for check in ( + '.state == "open"', + ".draft == false", + ".merged == false", + ".user.id == 49699333", + ".head.repo.full_name == $repo", + ".base.repo.full_name == $repo", + ".base.ref == $base", + ".head.sha == $head", + ): + assert check in step + assert step.count("if ! is_current; then") == 2 + first_check = step.index("if ! is_current; then") + review = step.index('review_id=$(jq -nc --arg commit_id "$HEAD_SHA"') + second_check = step.index("if ! is_current; then", first_check + 1) + dismissal = step.index('reviews/$review_id/dismissals" --input -') + merge = step.index( + 'gh pr merge --auto --squash --match-head-commit "$HEAD_SHA" "$PR_URL"' + ) + assert first_check < review < second_check < dismissal < merge + assert '{event: "APPROVE", commit_id: $commit_id,' in step + assert "--admin" not in step