From 386da782950b59f362ea84f5bfb15aaa6e3489f8 Mon Sep 17 00:00:00 2001 From: RomirJ Date: Fri, 28 Aug 2026 02:35:52 -0700 Subject: [PATCH] ci: add protected quality debt ratchet --- .github/workflows/quality-baseline-update.yml | 399 +++++++++ .github/workflows/quality-ratchet.yml | 277 ++++++ docs/ci_quality_ratchet.md | 68 ++ scripts/ci_quality_ratchet.py | 835 ++++++++++++++++++ tests/fixtures/ci_quality_ratchet/mypy.txt | 3 + .../ci_quality_ratchet/ruff_check.json | 8 + .../ci_quality_ratchet/ruff_format.txt | 2 + tests/test_ci_quality_ratchet.py | 497 +++++++++++ 8 files changed, 2089 insertions(+) create mode 100644 .github/workflows/quality-baseline-update.yml create mode 100644 .github/workflows/quality-ratchet.yml create mode 100644 docs/ci_quality_ratchet.md create mode 100644 scripts/ci_quality_ratchet.py create mode 100644 tests/fixtures/ci_quality_ratchet/mypy.txt create mode 100644 tests/fixtures/ci_quality_ratchet/ruff_check.json create mode 100644 tests/fixtures/ci_quality_ratchet/ruff_format.txt create mode 100644 tests/test_ci_quality_ratchet.py diff --git a/.github/workflows/quality-baseline-update.yml b/.github/workflows/quality-baseline-update.yml new file mode 100644 index 00000000..eddc8430 --- /dev/null +++ b/.github/workflows/quality-baseline-update.yml @@ -0,0 +1,399 @@ +name: Quality ratchet protected policy update + +on: + workflow_dispatch: + inputs: + pull_request_number: + description: Pull request containing the reviewed gate or tool-policy update + required: true + type: number + reason: + description: Why this protected authorization change is required + required: true + type: string + +permissions: + contents: read + +jobs: + validate-candidate: + name: validate candidate without authority + runs-on: ubuntu-latest + timeout-minutes: 20 + permissions: + actions: read + contents: read + pull-requests: read + outputs: + base_sha: ${{ steps.candidate.outputs.base_sha }} + candidate_sha: ${{ steps.candidate.outputs.sha }} + validation_name: ${{ steps.candidate.outputs.validation_name }} + + steps: + - name: Check out protected main + uses: actions/checkout@v4 + with: + fetch-depth: 0 + persist-credentials: false + + - name: Set up uncached Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install pinned validation tools without cache + run: >- + python -m pip install --no-cache-dir + "pytest==8.4.2" "ruff==0.15.10" "mypy==1.20.1" + + - name: Resolve the exact candidate + id: candidate + shell: bash + env: + GH_TOKEN: ${{ github.token }} + PR_NUMBER: ${{ inputs.pull_request_number }} + run: | + set -euo pipefail + [[ "$GITHUB_REF" == "refs/heads/main" ]] + PR_JSON=$(gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}") + jq -e '.state == "open" and .base.ref == "main"' \ + <<< "$PR_JSON" >/dev/null + CANDIDATE_SHA=$(jq -r '.head.sha' <<< "$PR_JSON") + [[ "$CANDIDATE_SHA" =~ ^[0-9a-f]{40}$ ]] + git fetch --no-tags origin \ + "+refs/pull/${PR_NUMBER}/head:refs/remotes/pull/policy-candidate" + git cat-file -e "${CANDIDATE_SHA}^{commit}" + VALIDATION_NAME="quality-policy-validation-${CANDIDATE_SHA}-${GITHUB_RUN_ID}" + echo "base_sha=$GITHUB_SHA" >> "$GITHUB_OUTPUT" + echo "sha=$CANDIDATE_SHA" >> "$GITHUB_OUTPUT" + echo "validation_name=$VALIDATION_NAME" >> "$GITHUB_OUTPUT" + + - name: Compare candidate under protected policy + run: >- + python -I scripts/ci_quality_ratchet.py + --repo . + --base-sha "$GITHUB_SHA" + --head-sha "${{ steps.candidate.outputs.sha }}" + --report "$RUNNER_TEMP/quality-policy-ratchet.json" + --authorization-approved + + - name: Parse proposed policy and statically validate candidate judge + shell: bash + run: | + set -euo pipefail + CANDIDATE_DIR="$RUNNER_TEMP/quality-policy-candidate" + CANDIDATE_POLICY="$RUNNER_TEMP/candidate-pyproject.toml" + git show \ + "${{ steps.candidate.outputs.sha }}:pyproject.toml" > "$CANDIDATE_POLICY" + python -I scripts/ci_quality_ratchet.py \ + --validate-policy-syntax "$CANDIDATE_POLICY" + git worktree add --detach "$CANDIDATE_DIR" "${{ steps.candidate.outputs.sha }}" + trap 'git worktree remove --force "$CANDIDATE_DIR"' EXIT + PROTECTED_POLICY="$CANDIDATE_DIR/.protected-base-pyproject.toml" + rm -f -- "$CANDIDATE_DIR/pyproject.toml" + git show "$GITHUB_SHA:pyproject.toml" > "$PROTECTED_POLICY" + python -I -m ruff check \ + --config="$PROTECTED_POLICY" \ + "$CANDIDATE_DIR/scripts/ci_quality_ratchet.py" \ + "$CANDIDATE_DIR/tests/test_ci_quality_ratchet.py" + python -I -m ruff format \ + --config="$PROTECTED_POLICY" --check \ + "$CANDIDATE_DIR/scripts/ci_quality_ratchet.py" \ + "$CANDIDATE_DIR/tests/test_ci_quality_ratchet.py" + python -I -m mypy \ + --config-file="$PROTECTED_POLICY" \ + "$CANDIDATE_DIR/scripts/ci_quality_ratchet.py" + + - name: Build protected pre-execution validation evidence + shell: bash + env: + CANDIDATE_SHA: ${{ steps.candidate.outputs.sha }} + run: | + set -euo pipefail + REPORT="$RUNNER_TEMP/quality-policy-ratchet.json" + jq -e ' + .passed == true + and (.new_findings | length) == 0 + and (.authorization_input_changes | length) > 0 + ' "$REPORT" >/dev/null + JUDGE_DIGEST=$(sha256sum scripts/ci_quality_ratchet.py | cut -d' ' -f1) + jq -n \ + --arg repository "$GITHUB_REPOSITORY" \ + --arg base_sha "$GITHUB_SHA" \ + --arg candidate_sha "$CANDIDATE_SHA" \ + --argjson run_id "$GITHUB_RUN_ID" \ + --arg judge_sha256 "$JUDGE_DIGEST" \ + --slurpfile ratchet "$REPORT" \ + '{ + schema_version: 1, + repository: $repository, + base_sha: $base_sha, + candidate_sha: $candidate_sha, + run_id: $run_id, + protected_judge_sha256: $judge_sha256, + ratchet_passed: $ratchet[0].passed, + new_findings_count: ($ratchet[0].new_findings | length), + authorization_input_changes: $ratchet[0].authorization_input_changes + }' > "$RUNNER_TEMP/quality-policy-validation.json" + + - name: Upload immutable validation evidence before candidate execution + uses: actions/upload-artifact@v4 + with: + name: ${{ steps.candidate.outputs.validation_name }} + path: ${{ runner.temp }}/quality-policy-validation.json + if-no-files-found: error + retention-days: 30 + + exercise-candidate: + name: exercise candidate only after evidence is sealed + needs: validate-candidate + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + + steps: + - name: Check out protected main without credentials + uses: actions/checkout@v4 + with: + fetch-depth: 0 + persist-credentials: false + + - name: Set up uncached Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install pinned candidate-test tools without cache + run: >- + python -m pip install --no-cache-dir + "pytest==8.4.2" "ruff==0.15.10" "mypy==1.20.1" + + - name: Resolve the already sealed candidate + shell: bash + env: + CANDIDATE_SHA: ${{ needs.validate-candidate.outputs.candidate_sha }} + PR_NUMBER: ${{ inputs.pull_request_number }} + run: | + set -euo pipefail + [[ "$CANDIDATE_SHA" =~ ^[0-9a-f]{40}$ ]] + git fetch --no-tags origin \ + "+refs/pull/${PR_NUMBER}/head:refs/remotes/pull/untrusted-candidate" + [[ "$(git rev-parse refs/remotes/pull/untrusted-candidate)" == "$CANDIDATE_SHA" ]] + + - name: Load candidate policy and run candidate tests without authority + shell: bash + env: + CANDIDATE_SHA: ${{ needs.validate-candidate.outputs.candidate_sha }} + run: | + set -euo pipefail + CANDIDATE_DIR="$RUNNER_TEMP/unprivileged-candidate-tests" + git worktree add --detach "$CANDIDATE_DIR" "$CANDIDATE_SHA" + trap 'cd "$GITHUB_WORKSPACE"; git worktree remove --force "$CANDIDATE_DIR"' EXIT + cd "$CANDIDATE_DIR" + python -I -m ruff check \ + --config=pyproject.toml \ + --show-settings scripts/ci_quality_ratchet.py >/dev/null + python -I -m mypy \ + --config-file=pyproject.toml \ + --no-error-summary -c 'pass' >/dev/null + rm -f -- "$CANDIDATE_DIR/pyproject.toml" + cp "$GITHUB_WORKSPACE/pyproject.toml" "$CANDIDATE_DIR/pyproject.toml" + PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 PYTHONPATH=. \ + python -m pytest -q tests/test_ci_quality_ratchet.py + + authorize-policy-update: + name: authorize exact validated SHA + needs: [validate-candidate, exercise-candidate] + if: >- + needs.validate-candidate.result == 'success' + && needs.exercise-candidate.result == 'success' + && github.ref == 'refs/heads/main' + runs-on: ubuntu-latest + timeout-minutes: 10 + environment: + name: quality-ratchet-policy + permissions: + actions: read + contents: read + pull-requests: read + statuses: write + + steps: + - name: Check out protected main only + uses: actions/checkout@v4 + with: + fetch-depth: 1 + persist-credentials: false + + - name: Verify environment, candidate, and validation artifact + id: validation + shell: bash + env: + GH_TOKEN: ${{ github.token }} + PR_NUMBER: ${{ inputs.pull_request_number }} + EXPECTED_SHA: ${{ needs.validate-candidate.outputs.candidate_sha }} + EXPECTED_BASE: ${{ needs.validate-candidate.outputs.base_sha }} + EXPECTED_NAME: ${{ needs.validate-candidate.outputs.validation_name }} + run: | + set -euo pipefail + ENVIRONMENT="$RUNNER_TEMP/policy-environment.json" + gh api \ + "repos/${GITHUB_REPOSITORY}/environments/quality-ratchet-policy" \ + > "$ENVIRONMENT" + python -I scripts/ci_quality_ratchet.py \ + --verify-policy-environment "$ENVIRONMENT" + + PR_JSON=$(gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}") + CANDIDATE_SHA=$(jq -r '.head.sha' <<< "$PR_JSON") + jq -e '.state == "open" and .base.ref == "main"' \ + <<< "$PR_JSON" >/dev/null + [[ "$CANDIDATE_SHA" == "$EXPECTED_SHA" ]] + [[ "$GITHUB_SHA" == "$EXPECTED_BASE" ]] + EXPECTED_ARTIFACT="quality-policy-validation-${CANDIDATE_SHA}-${GITHUB_RUN_ID}" + [[ "$EXPECTED_NAME" == "$EXPECTED_ARTIFACT" ]] + + ARTIFACTS=$(gh api \ + "repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/artifacts") + MATCH_COUNT=$(jq --arg name "$EXPECTED_ARTIFACT" \ + '[.artifacts[] | select(.name == $name and .expired == false)] | length' \ + <<< "$ARTIFACTS") + [[ "$MATCH_COUNT" == "1" ]] + ARTIFACT_ID=$(jq -r --arg name "$EXPECTED_ARTIFACT" \ + '.artifacts[] | select(.name == $name and .expired == false) | .id' \ + <<< "$ARTIFACTS") + ARTIFACT_DIGEST=$(jq -r --arg name "$EXPECTED_ARTIFACT" \ + '.artifacts[] | select(.name == $name and .expired == false) | .digest' \ + <<< "$ARTIFACTS") + [[ "$ARTIFACT_ID" =~ ^[1-9][0-9]*$ ]] + [[ "$ARTIFACT_DIGEST" =~ ^sha256:[0-9a-f]{64}$ ]] + echo "artifact_id=$ARTIFACT_ID" >> "$GITHUB_OUTPUT" + echo "artifact_digest=$ARTIFACT_DIGEST" >> "$GITHUB_OUTPUT" + echo "candidate_sha=$CANDIDATE_SHA" >> "$GITHUB_OUTPUT" + + - name: Download immutable validation evidence + uses: actions/download-artifact@v4 + with: + artifact-ids: ${{ steps.validation.outputs.artifact_id }} + path: ${{ runner.temp }}/validation-evidence + github-token: ${{ github.token }} + repository: ${{ github.repository }} + run-id: ${{ github.run_id }} + + - name: Verify validation evidence without loading candidate content + shell: bash + env: + CANDIDATE_SHA: ${{ steps.validation.outputs.candidate_sha }} + run: | + set -euo pipefail + EVIDENCE="$RUNNER_TEMP/validation-evidence/quality-policy-validation.json" + jq -e \ + --arg repository "$GITHUB_REPOSITORY" \ + --arg base_sha "$GITHUB_SHA" \ + --arg candidate_sha "$CANDIDATE_SHA" \ + --argjson run_id "$GITHUB_RUN_ID" \ + ' + (keys | sort) == ([ + "authorization_input_changes", "base_sha", "candidate_sha", + "new_findings_count", "protected_judge_sha256", "ratchet_passed", + "repository", "run_id", "schema_version" + ] | sort) + and .schema_version == 1 + and .repository == $repository + and .base_sha == $base_sha + and .candidate_sha == $candidate_sha + and .run_id == $run_id + and (.protected_judge_sha256 | test("^[0-9a-f]{64}$")) + and .ratchet_passed == true + and .new_findings_count == 0 + and (.authorization_input_changes | length) > 0 + ' "$EVIDENCE" >/dev/null + + - name: Reconfirm current protected tip immediately before approval + shell: bash + env: + GH_TOKEN: ${{ github.token }} + CANDIDATE_SHA: ${{ steps.validation.outputs.candidate_sha }} + EXPECTED_BASE: ${{ needs.validate-candidate.outputs.base_sha }} + PR_NUMBER: ${{ inputs.pull_request_number }} + run: | + set -euo pipefail + MAIN_REF="$RUNNER_TEMP/current-main-ref.json" + PULL_REQUEST="$RUNNER_TEMP/current-pull-request.json" + gh api "repos/${GITHUB_REPOSITORY}/git/ref/heads/main" > "$MAIN_REF" + gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}" > "$PULL_REQUEST" + python -I scripts/ci_quality_ratchet.py \ + --verify-main-ref-response "$MAIN_REF" \ + --verify-pull-request-response "$PULL_REQUEST" \ + --expected-current-main-sha "$EXPECTED_BASE" \ + --expected-current-head-sha "$CANDIDATE_SHA" + + - name: Build immutable approval evidence + id: approval + shell: bash + env: + CANDIDATE_SHA: ${{ steps.validation.outputs.candidate_sha }} + VALIDATION_ID: ${{ steps.validation.outputs.artifact_id }} + VALIDATION_DIGEST: ${{ steps.validation.outputs.artifact_digest }} + run: | + set -euo pipefail + APPROVAL_NAME="quality-policy-approval-${CANDIDATE_SHA}-${GITHUB_RUN_ID}" + jq -n \ + --arg repository "$GITHUB_REPOSITORY" \ + --arg base_sha "$GITHUB_SHA" \ + --arg candidate_sha "$CANDIDATE_SHA" \ + --arg workflow_path '.github/workflows/quality-baseline-update.yml' \ + --arg validation_digest "$VALIDATION_DIGEST" \ + --argjson run_id "$GITHUB_RUN_ID" \ + --argjson validation_id "$VALIDATION_ID" \ + '{ + schema_version: 1, + repository: $repository, + base_sha: $base_sha, + candidate_sha: $candidate_sha, + run_id: $run_id, + workflow_path: $workflow_path, + validation_artifact_id: $validation_id, + validation_artifact_digest: $validation_digest + }' > "$RUNNER_TEMP/quality-policy-approval.json" + echo "name=$APPROVAL_NAME" >> "$GITHUB_OUTPUT" + + - name: Upload immutable approval evidence + uses: actions/upload-artifact@v4 + with: + name: ${{ steps.approval.outputs.name }} + path: ${{ runner.temp }}/quality-policy-approval.json + if-no-files-found: error + retention-days: 30 + + - name: Reconfirm current protected tip immediately before status + shell: bash + env: + GH_TOKEN: ${{ github.token }} + CANDIDATE_SHA: ${{ steps.validation.outputs.candidate_sha }} + EXPECTED_BASE: ${{ needs.validate-candidate.outputs.base_sha }} + PR_NUMBER: ${{ inputs.pull_request_number }} + run: | + set -euo pipefail + MAIN_REF="$RUNNER_TEMP/final-main-ref.json" + PULL_REQUEST="$RUNNER_TEMP/final-pull-request.json" + gh api "repos/${GITHUB_REPOSITORY}/git/ref/heads/main" > "$MAIN_REF" + gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}" > "$PULL_REQUEST" + python -I scripts/ci_quality_ratchet.py \ + --verify-main-ref-response "$MAIN_REF" \ + --verify-pull-request-response "$PULL_REQUEST" \ + --expected-current-main-sha "$EXPECTED_BASE" \ + --expected-current-head-sha "$CANDIDATE_SHA" + + - name: Authorize the exact evidence-bound commit + env: + GH_TOKEN: ${{ github.token }} + CANDIDATE_SHA: ${{ steps.validation.outputs.candidate_sha }} + run: | + DESCRIPTION="sha:${CANDIDATE_SHA:0:12} run:${GITHUB_RUN_ID}" + gh api --method POST \ + "repos/${GITHUB_REPOSITORY}/statuses/${CANDIDATE_SHA}" \ + -f state=success \ + -f context='quality-ratchet/policy-approved' \ + -f description="$DESCRIPTION" \ + -f target_url="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" diff --git a/.github/workflows/quality-ratchet.yml b/.github/workflows/quality-ratchet.yml new file mode 100644 index 00000000..9f2e388f --- /dev/null +++ b/.github/workflows/quality-ratchet.yml @@ -0,0 +1,277 @@ +name: Quality ratchet + +# pull_request_target loads this workflow from protected main. The job never +# imports or executes PR code; it runs a protected-base judge over static files. +on: + pull_request_target: + types: [opened, synchronize, reopened, ready_for_review] + push: + branches: [main] + +permissions: + actions: read + contents: read + pull-requests: read + statuses: write + +jobs: + quality-ratchet: + name: ruff + mypy (no new debt) + runs-on: ubuntu-latest + timeout-minutes: 20 + env: + BASE_SHA: ${{ github.event.pull_request.base.sha || github.event.before }} + HEAD_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + PR_NUMBER: ${{ github.event.pull_request.number || 0 }} + + steps: + - name: Check out the protected comparison graph + uses: actions/checkout@v4 + with: + fetch-depth: 0 + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + + - name: Install pinned quality tools + run: python -m pip install "ruff==0.15.10" "mypy==1.20.1" + + - name: Resolve commits and protected judge + id: commits + shell: bash + run: | + set -euo pipefail + if [[ ! "$HEAD_SHA" =~ ^[0-9a-f]{40}$ ]]; then + echo "Invalid head SHA: $HEAD_SHA" >&2 + exit 2 + fi + if [[ "$BASE_SHA" =~ ^0{40}$ ]]; then + BASE_SHA=$(git rev-parse "${HEAD_SHA}^" 2>/dev/null || printf '%s' "$HEAD_SHA") + fi + if [[ ! "$BASE_SHA" =~ ^[0-9a-f]{40}$ ]]; then + echo "Invalid base SHA: $BASE_SHA" >&2 + exit 2 + fi + if ! git cat-file -e "${BASE_SHA}^{commit}" 2>/dev/null; then + git fetch --no-tags --depth=1 origin "$BASE_SHA" + fi + if ! git cat-file -e "${HEAD_SHA}^{commit}" 2>/dev/null; then + if [[ "$GITHUB_EVENT_NAME" == "pull_request_target" ]]; then + git fetch --no-tags origin \ + "+refs/pull/${PR_NUMBER}/head:refs/remotes/pull/quality-head" + else + git fetch --no-tags --depth=1 origin "$HEAD_SHA" + fi + fi + git cat-file -e "${BASE_SHA}^{commit}" + git cat-file -e "${HEAD_SHA}^{commit}" + + JUDGE="$RUNNER_TEMP/protected-ci-quality-ratchet.py" + if git cat-file -e "${BASE_SHA}:scripts/ci_quality_ratchet.py" 2>/dev/null; then + git show "${BASE_SHA}:scripts/ci_quality_ratchet.py" > "$JUDGE" + JUDGE_SOURCE="protected-base" + elif [[ "$GITHUB_EVENT_NAME" == "push" ]]; then + # One-time bootstrap for the merge that first introduces the gate. + git show "${HEAD_SHA}:scripts/ci_quality_ratchet.py" > "$JUDGE" + JUDGE_SOURCE="bootstrap-head" + else + echo "Protected base does not contain the quality judge" >&2 + exit 2 + fi + python -m py_compile "$JUDGE" + echo "base_sha=$BASE_SHA" >> "$GITHUB_OUTPUT" + echo "head_sha=$HEAD_SHA" >> "$GITHUB_OUTPUT" + echo "judge=$JUDGE" >> "$GITHUB_OUTPUT" + echo "judge_source=$JUDGE_SOURCE" >> "$GITHUB_OUTPUT" + + - name: Bind the event to the current protected-main tip + shell: bash + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + MAIN_REF="$RUNNER_TEMP/current-main-ref.json" + gh api "repos/${GITHUB_REPOSITORY}/git/ref/heads/main" > "$MAIN_REF" + ARGS=(--verify-main-ref-response "$MAIN_REF") + if [[ "$GITHUB_EVENT_NAME" == "pull_request_target" ]]; then + PULL_REQUEST="$RUNNER_TEMP/current-pull-request.json" + gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}" > "$PULL_REQUEST" + ARGS+=( + --verify-pull-request-response "$PULL_REQUEST" + --expected-current-main-sha "${{ steps.commits.outputs.base_sha }}" + --expected-current-head-sha "${{ steps.commits.outputs.head_sha }}" + ) + else + ARGS+=(--expected-current-main-sha "${{ steps.commits.outputs.head_sha }}") + fi + python -I "${{ steps.commits.outputs.judge }}" "${ARGS[@]}" + + - name: Verify protected policy authorization + id: authorization + shell: bash + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + if [[ "$GITHUB_EVENT_NAME" == "push" ]]; then + echo "push_approved=true" >> "$GITHUB_OUTPUT" + echo "candidate=false" >> "$GITHUB_OUTPUT" + exit 0 + fi + + STATUS_JSON=$(gh api \ + "repos/${GITHUB_REPOSITORY}/commits/${HEAD_SHA}/status") + STATUS_ENTRY=$(jq -c ' + [.statuses[] + | select(.context == "quality-ratchet/policy-approved") + | select(.state == "success") + | select(.creator.login == "github-actions[bot]")] + | first // {} + ' <<< "$STATUS_JSON") + TARGET_URL=$(jq -r ' + .target_url // "" + ' <<< "$STATUS_ENTRY") + if [[ ! "$TARGET_URL" =~ /actions/runs/([0-9]+)$ ]]; then + echo "push_approved=false" >> "$GITHUB_OUTPUT" + echo "candidate=false" >> "$GITHUB_OUTPUT" + exit 0 + fi + + RUN_ID="${BASH_REMATCH[1]}" + EXPECTED_DESCRIPTION="sha:${HEAD_SHA:0:12} run:${RUN_ID}" + [[ "$(jq -r '.description // ""' <<< "$STATUS_ENTRY")" == "$EXPECTED_DESCRIPTION" ]] || { + echo "push_approved=false" >> "$GITHUB_OUTPUT" + echo "candidate=false" >> "$GITHUB_OUTPUT" + exit 0 + } + RUN_JSON=$(gh api "repos/${GITHUB_REPOSITORY}/actions/runs/${RUN_ID}") + if ! jq -e --arg base_sha "$BASE_SHA" ' + .event == "workflow_dispatch" + and .head_branch == "main" + and .head_sha == $base_sha + and .conclusion == "success" + and .path == ".github/workflows/quality-baseline-update.yml" + ' <<< "$RUN_JSON" >/dev/null; then + echo "push_approved=false" >> "$GITHUB_OUTPUT" + echo "candidate=false" >> "$GITHUB_OUTPUT" + exit 0 + fi + + APPROVAL_NAME="quality-policy-approval-${HEAD_SHA}-${RUN_ID}" + ARTIFACTS=$(gh api \ + "repos/${GITHUB_REPOSITORY}/actions/runs/${RUN_ID}/artifacts") + MATCH_COUNT=$(jq --arg name "$APPROVAL_NAME" \ + '[.artifacts[] | select(.name == $name and .expired == false)] | length' \ + <<< "$ARTIFACTS") + [[ "$MATCH_COUNT" == "1" ]] || { + echo "push_approved=false" >> "$GITHUB_OUTPUT" + echo "candidate=false" >> "$GITHUB_OUTPUT" + exit 0 + } + ARTIFACT_ID=$(jq -r --arg name "$APPROVAL_NAME" \ + '.artifacts[] | select(.name == $name and .expired == false) | .id' \ + <<< "$ARTIFACTS") + ARTIFACT_DIGEST=$(jq -r --arg name "$APPROVAL_NAME" \ + '.artifacts[] | select(.name == $name and .expired == false) | .digest' \ + <<< "$ARTIFACTS") + [[ "$ARTIFACT_ID" =~ ^[1-9][0-9]*$ ]] + [[ "$ARTIFACT_DIGEST" =~ ^sha256:[0-9a-f]{64}$ ]] + echo "push_approved=false" >> "$GITHUB_OUTPUT" + echo "candidate=true" >> "$GITHUB_OUTPUT" + echo "run_id=$RUN_ID" >> "$GITHUB_OUTPUT" + echo "artifact_id=$ARTIFACT_ID" >> "$GITHUB_OUTPUT" + + - name: Download exact approval evidence + if: steps.authorization.outputs.candidate == 'true' + uses: actions/download-artifact@v4 + with: + artifact-ids: ${{ steps.authorization.outputs.artifact_id }} + path: ${{ runner.temp }}/approval-evidence + github-token: ${{ github.token }} + repository: ${{ github.repository }} + run-id: ${{ steps.authorization.outputs.run_id }} + + - name: Verify immutable approval evidence binding + if: steps.authorization.outputs.candidate == 'true' + id: evidence + shell: bash + run: | + set -euo pipefail + python -I "${{ steps.commits.outputs.judge }}" \ + --verify-approval-evidence \ + "$RUNNER_TEMP/approval-evidence/quality-policy-approval.json" \ + --expected-candidate-sha "$HEAD_SHA" \ + --expected-base-sha "$BASE_SHA" \ + --expected-run-id "${{ steps.authorization.outputs.run_id }}" \ + --expected-repository "$GITHUB_REPOSITORY" + echo "approved=true" >> "$GITHUB_OUTPUT" + + - name: Compare protected base and head findings + shell: bash + run: | + set -euo pipefail + ARGS=( + --repo . + --base-sha "${{ steps.commits.outputs.base_sha }}" + --head-sha "${{ steps.commits.outputs.head_sha }}" + --report "$RUNNER_TEMP/quality-ratchet.json" + ) + if [[ "${{ steps.authorization.outputs.push_approved }}" == "true" \ + || "${{ steps.evidence.outputs.approved }}" == "true" ]]; then + ARGS+=(--authorization-approved) + fi + python -I "${{ steps.commits.outputs.judge }}" "${ARGS[@]}" + + - name: Reconfirm current protected-main tip before publishing + shell: bash + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + MAIN_REF="$RUNNER_TEMP/final-main-ref.json" + gh api "repos/${GITHUB_REPOSITORY}/git/ref/heads/main" > "$MAIN_REF" + ARGS=(--verify-main-ref-response "$MAIN_REF") + if [[ "$GITHUB_EVENT_NAME" == "pull_request_target" ]]; then + PULL_REQUEST="$RUNNER_TEMP/final-pull-request.json" + gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}" > "$PULL_REQUEST" + ARGS+=( + --verify-pull-request-response "$PULL_REQUEST" + --expected-current-main-sha "${{ steps.commits.outputs.base_sha }}" + --expected-current-head-sha "${{ steps.commits.outputs.head_sha }}" + ) + else + ARGS+=(--expected-current-main-sha "${{ steps.commits.outputs.head_sha }}") + fi + python -I "${{ steps.commits.outputs.judge }}" "${ARGS[@]}" + + - name: Publish protected result on the exact head SHA + if: always() + env: + GH_TOKEN: ${{ github.token }} + JOB_STATUS: ${{ job.status }} + run: | + STATE=failure + DESCRIPTION='Protected base quality ratchet failed' + if [[ "$JOB_STATUS" == "success" ]]; then + STATE=success + DESCRIPTION='Protected base quality ratchet passed' + fi + gh api --method POST \ + "repos/${GITHUB_REPOSITORY}/statuses/${HEAD_SHA}" \ + -f state="$STATE" \ + -f context='quality-ratchet/protected' \ + -f description="$DESCRIPTION" \ + -f target_url="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" + + - name: Upload ratchet evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: quality-ratchet-${{ steps.commits.outputs.head_sha }} + path: ${{ runner.temp }}/quality-ratchet.json + if-no-files-found: warn + retention-days: 14 diff --git a/docs/ci_quality_ratchet.md b/docs/ci_quality_ratchet.md new file mode 100644 index 00000000..ac9b9aa2 --- /dev/null +++ b/docs/ci_quality_ratchet.md @@ -0,0 +1,68 @@ +# CI quality ratchet + +The quality ratchet compares Ruff check, Ruff format, and mypy findings at the +protected base commit and the exact pull-request head. It recomputes both sides +under the protected base's tool policy; there is no checked-in debt snapshot. + +## Repository settings + +After the bootstrap merge, repository administrators must: + +1. Require the `quality-ratchet/protected` commit status on `main`. +2. Create the `quality-ratchet-policy` environment. +3. Restrict that environment to protected branches and configure + `@rylinjames` as its sole reviewer. + +The manual workflow requires the exact reviewer allowlist before it can +authorize a commit. Missing, duplicate, team, or additional reviewers and a +weaker branch policy all fail closed because every configured reviewer could +otherwise grant deployment approval. + +The protected quality policy forbids `tool.mypy.plugins`. Mypy imports plugin +code during analysis; permitting a repository-local plugin would let a pull +request replace that code and execute it inside the ordinary status-writing +job. Both the current protected policy and proposed policy updates are rejected +before Ruff or mypy starts if a plugin setting is present. + +## Updating the gate or policy + +Ordinary pull requests cannot authorize changes to the ratchet script, +workflows, CODEOWNERS, mutable baseline-like files, or the effective Ruff/mypy +sections of `pyproject.toml`. + +For an intentional update: + +1. Open the pull request and complete code-owner review. +2. From protected `main`, run **Quality ratchet protected policy update** with + the pull-request number and reason. +3. Approve the `quality-ratchet-policy` environment deployment. +4. After the manual run succeeds, rerun the pull request's quality-ratchet job. + +The manual workflow has three deliberately separate jobs. Its first job has no +write authority or Actions cache: it validates the exact current head SHA under +the old policy, parses the proposed Ruff/mypy configuration strictly as TOML +data without invoking either tool or importing configured plugins, statically +validates the proposed judge under protected policy, and uploads immutable +pre-execution evidence. + +Its second job starts on a separate unprivileged runner only after that evidence +is sealed. Candidate tool configuration and tests can execute there, but the +job has no authority, produces no authorization outputs, and cannot mutate the +first runner or its immutable artifact. + +Its third job starts on a fresh environment-protected runner. It checks out +only protected `main`, executes no candidate content, downloads only the exact +validation artifact, and re-queries both `refs/heads/main` and the current pull +request immediately before approval creation. Both must still identify the +sealed base and candidate. It then issues an immutable approval artifact plus a +status whose description binds the candidate SHA to the manual run ID, with a +second live-tip check immediately before that status write. + +The ordinary protected workflow independently queries the current `main` tip +and pull request before comparison, rejecting stale event payloads and reruns +after `main` or the pull-request head advances. It repeats the live-tip check +after analysis and immediately before publishing the protected result. It also +queries the exact manual run and artifact through the GitHub API, downloads it +by artifact ID, and verifies its repository, base SHA, candidate SHA, workflow +path, and run ID with the protected judge. A status from an unrelated run, a +stale approval or event, or missing/expired evidence fails closed. diff --git a/scripts/ci_quality_ratchet.py b/scripts/ci_quality_ratchet.py new file mode 100644 index 00000000..931b0aa6 --- /dev/null +++ b/scripts/ci_quality_ratchet.py @@ -0,0 +1,835 @@ +#!/usr/bin/env python3 +"""Fail CI only when a commit adds Ruff or mypy debt. + +The protected base commit is the baseline. No checked-in finding inventory is +read, so a pull request cannot authorize its own debt by editing a snapshot. +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +import sys +import tempfile +from collections import Counter, defaultdict +from contextlib import contextmanager +from dataclasses import asdict, dataclass +from pathlib import Path, PurePosixPath +from typing import Callable, Iterator, Sequence + +MAX_RELOCATION_LINES = 50 +TRUSTED_POLICY_REVIEWERS = frozenset({"rylinjames"}) +AUTHORIZATION_PATHS = frozenset( + { + ".github/CODEOWNERS", + ".github/workflows/quality-baseline-update.yml", + ".github/workflows/quality-ratchet.yml", + "scripts/ci_quality_ratchet.py", + } +) +POLICY_WORKFLOW_PATH = ".github/workflows/quality-baseline-update.yml" +_APPROVAL_EVIDENCE_FIELDS = frozenset( + { + "base_sha", + "candidate_sha", + "repository", + "run_id", + "schema_version", + "validation_artifact_digest", + "validation_artifact_id", + "workflow_path", + } +) +_SHA256_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") +_COMMIT_SHA = re.compile(r"^[0-9a-f]{40}$") +_MYPY_LINE = re.compile(r"^(.*?):(\d+):(\d+):\s+(error|warning|note):\s+(.*?)(?:\s+\[([^\]]+)\])?$") +_RUFF_FORMAT_LINE = re.compile(r"^Would reformat:\s+(.+?)\s*$") +_DIFF_HUNK = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@") + + +class RatchetError(RuntimeError): + """Raised when a tool or repository operation cannot be trusted.""" + + +@dataclass(frozen=True, slots=True) +class Finding: + """One normalized static-analysis finding.""" + + tool: str + rule: str + path: str + message: str + line: int + column: int + + @property + def fingerprint(self) -> tuple[str, str, str, str]: + """Return the stable identity; source coordinates are metadata only.""" + return (self.tool, self.rule, self.path, self.message) + + +@dataclass(frozen=True, slots=True) +class DiffHunk: + """A zero-context Git diff hunk used to relocate a base line.""" + + old_start: int + old_count: int + new_start: int + new_count: int + + +@dataclass(frozen=True, slots=True) +class Comparison: + """The monotonicity decision and its evidence.""" + + new_findings: tuple[Finding, ...] + removed_findings: tuple[Finding, ...] + relocated_matches: int + base_count: int + head_count: int + base_counts_by_tool: dict[str, int] + head_counts_by_tool: dict[str, int] + + @property + def passed(self) -> bool: + """Return whether both the stable multiset and counts are monotonic.""" + return ( + not self.new_findings + and self.head_count <= self.base_count + and all( + self.head_counts_by_tool.get(tool, 0) <= count + for tool, count in self.base_counts_by_tool.items() + ) + ) + + +def normalize_path(raw_path: str, *, checkout: Path) -> str: + """Normalize a tool path to a repository-relative POSIX path.""" + text = raw_path.strip().replace("\\", "/") + root = checkout.resolve().as_posix().rstrip("/") + if text == root: + return "." + if text.startswith(root + "/"): + text = text[len(root) + 1 :] + while text.startswith("./"): + text = text[2:] + normalized = PurePosixPath(text).as_posix() + return normalized or "." + + +def normalize_message(message: str) -> str: + """Collapse presentation-only whitespace and volatile line references.""" + value = " ".join(message.strip().split()) + return re.sub(r"\bline\s+\d+\b", "line ", value, flags=re.IGNORECASE) + + +def verify_policy_approval_evidence( + value: object, + *, + expected_candidate_sha: str, + expected_base_sha: str, + expected_run_id: int, + expected_repository: str, +) -> None: + """Verify the immutable manual-run envelope for one exact comparison.""" + if not isinstance(value, dict) or set(value) != _APPROVAL_EVIDENCE_FIELDS: + actual = sorted(value) if isinstance(value, dict) else type(value).__name__ + raise RatchetError(f"policy approval evidence fields mismatch: {actual}") + if value["schema_version"] != 1: + raise RatchetError("unsupported policy approval evidence schema") + if value["candidate_sha"] != expected_candidate_sha: + raise RatchetError("policy approval candidate SHA does not match head") + if value["base_sha"] != expected_base_sha: + raise RatchetError("policy approval base SHA does not match protected base") + if value["run_id"] != expected_run_id or isinstance(value["run_id"], bool): + raise RatchetError("policy approval run id does not match status target") + if value["repository"] != expected_repository: + raise RatchetError("policy approval repository does not match") + if value["workflow_path"] != POLICY_WORKFLOW_PATH: + raise RatchetError("policy approval workflow path does not match") + artifact_id = value["validation_artifact_id"] + if isinstance(artifact_id, bool) or not isinstance(artifact_id, int) or artifact_id <= 0: + raise RatchetError("policy approval validation artifact id is invalid") + digest = value["validation_artifact_digest"] + if not isinstance(digest, str) or not _SHA256_DIGEST.fullmatch(digest): + raise RatchetError("policy approval validation artifact digest is invalid") + + +def verify_current_main_binding( + ref_value: object, + *, + expected_main_sha: str, + pull_request_value: object | None = None, + expected_head_sha: str | None = None, +) -> None: + """Reject stale workflow events by binding them to the live main ref.""" + if not _COMMIT_SHA.fullmatch(expected_main_sha): + raise RatchetError("expected protected-main SHA is invalid") + if not isinstance(ref_value, dict) or ref_value.get("ref") != "refs/heads/main": + raise RatchetError("protected-main ref response is invalid") + ref_object = ref_value.get("object") + if not isinstance(ref_object, dict) or ref_object.get("type") != "commit": + raise RatchetError("protected-main ref does not resolve to a commit") + if ref_object.get("sha") != expected_main_sha: + raise RatchetError("workflow event is stale because protected main advanced") + + if pull_request_value is None: + if expected_head_sha is not None: + raise RatchetError("pull-request head was provided without pull-request data") + return + if expected_head_sha is None or not _COMMIT_SHA.fullmatch(expected_head_sha): + raise RatchetError("expected pull-request head SHA is invalid") + if not isinstance(pull_request_value, dict) or pull_request_value.get("state") != "open": + raise RatchetError("pull request is no longer open") + base = pull_request_value.get("base") + head = pull_request_value.get("head") + if not isinstance(base, dict) or base.get("ref") != "main": + raise RatchetError("pull request no longer targets protected main") + if base.get("sha") != expected_main_sha: + raise RatchetError("pull-request base SHA does not match current protected main") + if not isinstance(head, dict) or head.get("sha") != expected_head_sha: + raise RatchetError("pull-request head SHA no longer matches the workflow event") + + +def verify_policy_environment(value: object) -> None: + """Require the exact protected environment reviewer allowlist.""" + if not isinstance(value, dict): + raise RatchetError("policy environment response is not an object") + rules = value.get("protection_rules") + if not isinstance(rules, list): + raise RatchetError("policy environment protection rules are missing") + reviewer_rules = [ + rule + for rule in rules + if isinstance(rule, dict) and rule.get("type") == "required_reviewers" + ] + if len(reviewer_rules) != 1: + raise RatchetError("policy environment must have exactly one reviewer rule") + reviewers = reviewer_rules[0].get("reviewers") + if not isinstance(reviewers, list): + raise RatchetError("policy environment reviewer list is missing") + actual_reviewers: list[str] = [] + for entry in reviewers: + if not isinstance(entry, dict) or entry.get("type") != "User": + raise RatchetError("policy environment reviewers must be individual users") + reviewer = entry.get("reviewer") + login = reviewer.get("login") if isinstance(reviewer, dict) else None + if not isinstance(login, str) or not login: + raise RatchetError("policy environment reviewer login is invalid") + actual_reviewers.append(login.casefold()) + if len(actual_reviewers) != len(set(actual_reviewers)): + raise RatchetError("policy environment reviewer list contains duplicates") + if frozenset(actual_reviewers) != TRUSTED_POLICY_REVIEWERS: + raise RatchetError("policy environment reviewer allowlist does not match") + + branch_policy = value.get("deployment_branch_policy") + if not isinstance(branch_policy, dict): + raise RatchetError("policy environment deployment branch policy is missing") + if ( + branch_policy.get("protected_branches") is not True + or branch_policy.get("custom_branch_policies") is not False + ): + raise RatchetError("policy environment must allow protected branches only") + + +def parse_ruff_check(output: str, *, checkout: Path) -> list[Finding]: + """Parse ``ruff check --output-format=json`` output.""" + try: + values = json.loads(output or "[]") + except json.JSONDecodeError as exc: + raise RatchetError(f"Ruff check emitted invalid JSON: {exc}") from exc + if not isinstance(values, list): + raise RatchetError("Ruff check JSON must be a list") + + findings: list[Finding] = [] + for value in values: + if not isinstance(value, dict): + raise RatchetError("Ruff check JSON contained a non-object finding") + location = value.get("location") or {} + if not isinstance(location, dict): + location = {} + findings.append( + Finding( + tool="ruff-check", + rule=str(value.get("code") or "syntax"), + path=normalize_path(str(value.get("filename") or ""), checkout=checkout), + message=normalize_message(str(value.get("message") or "unknown Ruff finding")), + line=int(location.get("row") or 1), + column=int(location.get("column") or 1), + ) + ) + return findings + + +def parse_ruff_format(output: str, *, checkout: Path) -> list[Finding]: + """Parse ``ruff format --check`` output into one finding per file.""" + findings: list[Finding] = [] + for line in output.splitlines(): + match = _RUFF_FORMAT_LINE.match(line.strip()) + if match: + findings.append( + Finding( + tool="ruff-format", + rule="format", + path=normalize_path(match.group(1), checkout=checkout), + message="File is not formatted", + line=1, + column=1, + ) + ) + return findings + + +def parse_mypy(output: str, *, checkout: Path) -> list[Finding]: + """Parse stable, non-pretty mypy error lines and ignore explanatory notes.""" + findings: list[Finding] = [] + for line in output.splitlines(): + match = _MYPY_LINE.match(line.strip()) + if not match or match.group(4) != "error": + continue + findings.append( + Finding( + tool="mypy", + rule=match.group(6) or "error", + path=normalize_path(match.group(1), checkout=checkout), + message=normalize_message(match.group(5)), + line=int(match.group(2)), + column=int(match.group(3)), + ) + ) + return findings + + +def parse_diff_hunks(diff_text: str) -> list[DiffHunk]: + """Parse zero-context unified diff hunk headers.""" + hunks: list[DiffHunk] = [] + for line in diff_text.splitlines(): + match = _DIFF_HUNK.match(line) + if match: + hunks.append( + DiffHunk( + old_start=int(match.group(1)), + old_count=int(match.group(2) or 1), + new_start=int(match.group(3)), + new_count=int(match.group(4) or 1), + ) + ) + return hunks + + +def relocate_base_line(line: int, hunks: Sequence[DiffHunk]) -> int: + """Map a base coordinate through a diff, including large pure insertions.""" + delta = 0 + for hunk in hunks: + if hunk.old_count == 0: + if line > hunk.old_start: + delta += hunk.new_count + continue + break + old_end = hunk.old_start + hunk.old_count - 1 + if line < hunk.old_start: + break + if line <= old_end: + offset = line - hunk.old_start + if hunk.new_count == 0: + return hunk.new_start + return hunk.new_start + min(offset, hunk.new_count - 1) + delta += hunk.new_count - hunk.old_count + return max(1, line + delta) + + +def _pair_group( + base: Sequence[Finding], + head: Sequence[Finding], + hunks: Sequence[DiffHunk], +) -> tuple[list[Finding], list[Finding], int]: + """Pair one fingerprint multiset, preferring bounded diff relocation.""" + unmatched_base = list(sorted(base, key=lambda item: (item.line, item.column))) + unmatched_head = list(sorted(head, key=lambda item: (item.line, item.column))) + relocated = 0 + + for base_finding in list(unmatched_base): + expected = relocate_base_line(base_finding.line, hunks) + candidates = [ + item for item in unmatched_head if abs(item.line - expected) <= MAX_RELOCATION_LINES + ] + if not candidates: + continue + selected = min(candidates, key=lambda item: (abs(item.line - expected), item.column)) + unmatched_base.remove(base_finding) + unmatched_head.remove(selected) + if selected.line != base_finding.line: + relocated += 1 + + # Coordinates are not part of identity. Pair any remaining identical + # fingerprints deterministically; only surplus occurrences are new debt. + fallback_pairs = min(len(unmatched_base), len(unmatched_head)) + if fallback_pairs: + del unmatched_base[:fallback_pairs] + del unmatched_head[:fallback_pairs] + return unmatched_head, unmatched_base, relocated + + +def compare_findings( + base: Sequence[Finding], + head: Sequence[Finding], + *, + diff_hunks: Callable[[str], Sequence[DiffHunk]], +) -> Comparison: + """Compare stable finding multisets and enforce monotonic non-increase.""" + base_groups: dict[tuple[str, str, str, str], list[Finding]] = defaultdict(list) + head_groups: dict[tuple[str, str, str, str], list[Finding]] = defaultdict(list) + for finding in base: + base_groups[finding.fingerprint].append(finding) + for finding in head: + head_groups[finding.fingerprint].append(finding) + + new: list[Finding] = [] + removed: list[Finding] = [] + relocated = 0 + for fingerprint in sorted(set(base_groups) | set(head_groups)): + base_group = base_groups.get(fingerprint, []) + head_group = head_groups.get(fingerprint, []) + hunks = diff_hunks(fingerprint[2]) if base_group and head_group else () + group_new, group_removed, group_relocated = _pair_group(base_group, head_group, hunks) + new.extend(group_new) + removed.extend(group_removed) + relocated += group_relocated + + return Comparison( + new_findings=tuple( + sorted(new, key=lambda item: (item.tool, item.path, item.line, item.rule)) + ), + removed_findings=tuple( + sorted(removed, key=lambda item: (item.tool, item.path, item.line, item.rule)) + ), + relocated_matches=relocated, + base_count=len(base), + head_count=len(head), + base_counts_by_tool=dict(Counter(item.tool for item in base)), + head_counts_by_tool=dict(Counter(item.tool for item in head)), + ) + + +def is_baseline_artifact(path: str) -> bool: + """Return whether a changed path is reserved for mutable debt snapshots.""" + normalized = PurePosixPath(path.lower()).as_posix() + name = PurePosixPath(normalized).name + return ( + normalized.startswith(".ci/quality-baseline/") + or normalized.startswith(".github/quality-baseline/") + or name.startswith(".quality-baseline") + or bool(re.fullmatch(r"quality[-_.]baselines?\.(json|toml|ya?ml|txt)", name)) + ) + + +def _changed_paths(repo: Path, base_sha: str, head_sha: str) -> list[str]: + """List paths changed between two explicit commits.""" + result = _run( + ["git", "diff", "--name-only", "-z", base_sha, head_sha], + cwd=repo, + accepted={0}, + ) + return sorted(path for path in result.stdout.split("\0") if path) + + +def _git_text(repo: Path, sha: str, path: str) -> str: + """Read one UTF-8 text file exactly from a commit.""" + return _run(["git", "show", f"{sha}:{path}"], cwd=repo, accepted={0}).stdout + + +def validate_tool_policy_text(pyproject_text: str) -> dict[str, dict[str, object]]: + """Parse Ruff/mypy policy as data without importing configured plugins.""" + try: + import tomllib + except ModuleNotFoundError: + try: + import tomli as tomllib # type: ignore[no-redef] + except ModuleNotFoundError as exc: + raise RatchetError("Python 3.11+ or tomli is required to parse tool policy") from exc + try: + value = tomllib.loads(pyproject_text) + except Exception as exc: + raise RatchetError(f"pyproject.toml is not valid TOML: {exc}") from exc + tool = value.get("tool", {}) + if not isinstance(tool, dict): + raise RatchetError("pyproject.toml [tool] must be a table") + policies: dict[str, dict[str, object]] = {} + for name in ("ruff", "mypy"): + policy = tool.get(name, {}) + if not isinstance(policy, dict): + raise RatchetError(f"pyproject.toml [tool.{name}] must be a table") + policies[name] = policy + if "plugins" in policies["mypy"]: + raise RatchetError("mypy plugins are forbidden in the protected quality policy") + return policies + + +def changed_authorization_inputs(repo: Path, base_sha: str, head_sha: str) -> list[str]: + """List gate, ownership, mutable baseline, and effective-policy changes.""" + changed = _changed_paths(repo, base_sha, head_sha) + authorization_changes = [path for path in changed if path in AUTHORIZATION_PATHS] + authorization_changes.extend(path for path in changed if is_baseline_artifact(path)) + + if "pyproject.toml" in changed: + base_policy = validate_tool_policy_text(_git_text(repo, base_sha, "pyproject.toml")) + head_policy = validate_tool_policy_text(_git_text(repo, head_sha, "pyproject.toml")) + for tool in ("ruff", "mypy"): + if base_policy[tool] != head_policy[tool]: + authorization_changes.append(f"pyproject.toml#[tool.{tool}]") + return sorted(set(authorization_changes)) + + +def _run( + command: Sequence[str], + *, + cwd: Path, + accepted: set[int], + env: dict[str, str] | None = None, +) -> subprocess.CompletedProcess[str]: + result = subprocess.run( + list(command), + cwd=cwd, + env=env, + text=True, + capture_output=True, + check=False, + ) + if result.returncode not in accepted: + rendered = " ".join(command) + detail = (result.stdout + "\n" + result.stderr).strip()[-4000:] + raise RatchetError(f"command failed ({result.returncode}): {rendered}\n{detail}") + return result + + +def _tool_version(module: str, *, cwd: Path) -> str: + result = _run( + [sys.executable, "-I", "-m", module, "--version"], + cwd=cwd, + accepted={0}, + ) + return (result.stdout or result.stderr).strip() + + +def collect_findings( + checkout: Path, + *, + cache_dir: Path, + protected_pyproject: str, +) -> tuple[list[Finding], dict[str, str]]: + """Run all three tools in one checkout and parse nonzero finding exits.""" + # Both snapshots must use the protected base policy. These worktrees are + # disposable, so replacing a PR-controlled pyproject cannot mutate a commit. + policy_path = checkout / "pyproject.toml" + policy_path.unlink() + policy_path.write_text(protected_pyproject) + ruff_check = _run( + [ + sys.executable, + "-I", + "-m", + "ruff", + "check", + "--config=pyproject.toml", + "--no-cache", + "--output-format=json", + ".", + ], + cwd=checkout, + accepted={0, 1}, + ) + ruff_check_findings = parse_ruff_check(ruff_check.stdout, checkout=checkout) + if ruff_check.returncode == 1 and not ruff_check_findings: + raise RatchetError("Ruff check returned findings but its JSON parsed empty") + + ruff_format = _run( + [ + sys.executable, + "-I", + "-m", + "ruff", + "format", + "--config=pyproject.toml", + "--check", + ".", + ], + cwd=checkout, + accepted={0, 1}, + ) + ruff_format_output = ruff_format.stdout + "\n" + ruff_format.stderr + ruff_format_findings = parse_ruff_format(ruff_format_output, checkout=checkout) + if ruff_format.returncode == 1 and not ruff_format_findings: + raise RatchetError("Ruff format returned findings but no file paths were parsed") + + mypy = _run( + [ + sys.executable, + "-I", + "-m", + "mypy", + "--config-file=pyproject.toml", + "--show-error-codes", + "--show-column-numbers", + "--no-pretty", + "--no-color-output", + "--no-error-summary", + f"--cache-dir={cache_dir}", + "src/", + ], + cwd=checkout, + accepted={0, 1}, + ) + mypy_output = mypy.stdout + "\n" + mypy.stderr + mypy_findings = parse_mypy(mypy_output, checkout=checkout) + if mypy.returncode == 1 and not mypy_findings: + raise RatchetError("mypy returned findings but no error lines were parsed") + + return ( + ruff_check_findings + ruff_format_findings + mypy_findings, + { + "ruff": _tool_version("ruff", cwd=checkout), + "mypy": _tool_version("mypy", cwd=checkout), + }, + ) + + +@contextmanager +def detached_worktree(repo: Path, sha: str, destination: Path) -> Iterator[Path]: + """Materialize an immutable commit without altering the caller's checkout.""" + _run(["git", "cat-file", "-e", f"{sha}^{{commit}}"], cwd=repo, accepted={0}) + _run( + ["git", "worktree", "add", "--detach", str(destination), sha], + cwd=repo, + accepted={0}, + ) + try: + yield destination + finally: + _run( + ["git", "worktree", "remove", "--force", str(destination)], + cwd=repo, + accepted={0}, + ) + + +def _diff_hunk_provider( + repo: Path, base_sha: str, head_sha: str +) -> Callable[[str], list[DiffHunk]]: + cache: dict[str, list[DiffHunk]] = {} + changed_result = _run( + ["git", "diff", "--no-renames", "--name-only", "-z", base_sha, head_sha], + cwd=repo, + accepted={0}, + ) + changed_paths = {path for path in changed_result.stdout.split("\0") if path} + + def provide(path: str) -> list[DiffHunk]: + if path not in changed_paths: + return [] + if path not in cache: + result = _run( + [ + "git", + "diff", + "--no-ext-diff", + "--no-renames", + "--unified=0", + base_sha, + head_sha, + "--", + path, + ], + cwd=repo, + accepted={0}, + ) + cache[path] = parse_diff_hunks(result.stdout) + return cache[path] + + return provide + + +def _finding_dict(finding: Finding) -> dict[str, object]: + value = asdict(finding) + value["fingerprint"] = list(finding.fingerprint) + return value + + +def run_ratchet( + *, + repo: Path, + base_sha: str, + head_sha: str, + report_path: Path, + authorization_approved: bool = False, +) -> bool: + """Collect both commits independently, compare them, and write evidence.""" + repo = repo.resolve() + authorization_changes = changed_authorization_inputs(repo, base_sha, head_sha) + with tempfile.TemporaryDirectory(prefix="tether-quality-ratchet-") as temporary: + temporary_root = Path(temporary) + with detached_worktree(repo, base_sha, temporary_root / "base") as base_checkout: + protected_pyproject = (base_checkout / "pyproject.toml").read_text() + validate_tool_policy_text(protected_pyproject) + base_findings, base_versions = collect_findings( + base_checkout, + cache_dir=temporary_root / "mypy-base", + protected_pyproject=protected_pyproject, + ) + with detached_worktree(repo, head_sha, temporary_root / "head") as head_checkout: + head_findings, head_versions = collect_findings( + head_checkout, + cache_dir=temporary_root / "mypy-head", + protected_pyproject=protected_pyproject, + ) + + comparison = compare_findings( + base_findings, + head_findings, + diff_hunks=_diff_hunk_provider(repo, base_sha, head_sha), + ) + authorization_allowed = not authorization_changes or authorization_approved + passed = comparison.passed and authorization_allowed + report = { + "schema_version": 1, + "base_sha": base_sha, + "head_sha": head_sha, + "passed": passed, + "authorization_approved": authorization_approved, + "authorization_input_changes": authorization_changes, + "base_count": comparison.base_count, + "head_count": comparison.head_count, + "base_counts_by_tool": comparison.base_counts_by_tool, + "head_counts_by_tool": comparison.head_counts_by_tool, + "new_findings": [_finding_dict(item) for item in comparison.new_findings], + "removed_findings": [_finding_dict(item) for item in comparison.removed_findings], + "relocated_matches": comparison.relocated_matches, + "tool_versions": {"base": base_versions, "head": head_versions}, + } + report_path.parent.mkdir(parents=True, exist_ok=True) + report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n") + + status = "PASS" if passed else "FAIL" + print( + f"quality ratchet {status}: {comparison.base_count} base -> " + f"{comparison.head_count} head; {len(comparison.new_findings)} new, " + f"{len(comparison.removed_findings)} removed" + ) + for finding in comparison.new_findings: + print( + f"NEW {finding.tool} {finding.path}:{finding.line}:{finding.column} " + f"[{finding.rule}] {finding.message}" + ) + if authorization_changes and not authorization_approved: + for path in authorization_changes: + print(f"PROTECTED authorization input requires manual approval: {path}") + print(f"report: {report_path}") + return passed + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--repo", type=Path, default=Path.cwd()) + parser.add_argument("--base-sha") + parser.add_argument("--head-sha") + parser.add_argument("--report", type=Path) + parser.add_argument( + "--authorization-approved", + action="store_true", + help="Accept gate/policy changes approved by the protected manual workflow.", + ) + parser.add_argument("--verify-approval-evidence", type=Path) + parser.add_argument("--validate-policy-syntax", type=Path) + parser.add_argument("--verify-main-ref-response", type=Path) + parser.add_argument("--verify-policy-environment", type=Path) + parser.add_argument("--verify-pull-request-response", type=Path) + parser.add_argument("--expected-current-main-sha") + parser.add_argument("--expected-current-head-sha") + parser.add_argument("--expected-candidate-sha") + parser.add_argument("--expected-base-sha") + parser.add_argument("--expected-run-id", type=int) + parser.add_argument("--expected-repository") + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + """Run the command-line quality ratchet.""" + parser = _build_parser() + args = parser.parse_args(argv) + try: + if args.verify_policy_environment is not None: + try: + environment_value = json.loads(args.verify_policy_environment.read_text()) + except (OSError, json.JSONDecodeError) as exc: + raise RatchetError(f"policy environment evidence is unreadable: {exc}") from exc + verify_policy_environment(environment_value) + print("policy environment reviewer allowlist verified") + return 0 + if args.verify_main_ref_response is not None: + if args.expected_current_main_sha is None: + parser.error("current-main verification requires --expected-current-main-sha") + try: + ref_value = json.loads(args.verify_main_ref_response.read_text()) + pull_request_value = ( + json.loads(args.verify_pull_request_response.read_text()) + if args.verify_pull_request_response is not None + else None + ) + except (OSError, json.JSONDecodeError) as exc: + raise RatchetError(f"current-main API evidence is unreadable: {exc}") from exc + verify_current_main_binding( + ref_value, + expected_main_sha=args.expected_current_main_sha, + pull_request_value=pull_request_value, + expected_head_sha=args.expected_current_head_sha, + ) + print("workflow event is bound to the current protected-main tip") + return 0 + if args.validate_policy_syntax is not None: + try: + policy_text = args.validate_policy_syntax.read_text() + except OSError as exc: + raise RatchetError(f"candidate tool policy is unreadable: {exc}") from exc + validate_tool_policy_text(policy_text) + print("candidate Ruff/mypy policy parsed without executing plugins") + return 0 + if args.verify_approval_evidence is not None: + expected = ( + args.expected_candidate_sha, + args.expected_base_sha, + args.expected_run_id, + args.expected_repository, + ) + if any(value is None for value in expected): + parser.error("approval evidence verification requires every --expected-* value") + try: + evidence = json.loads(args.verify_approval_evidence.read_text()) + except (OSError, json.JSONDecodeError) as exc: + raise RatchetError(f"policy approval evidence is unreadable: {exc}") from exc + verify_policy_approval_evidence( + evidence, + expected_candidate_sha=args.expected_candidate_sha, + expected_base_sha=args.expected_base_sha, + expected_run_id=args.expected_run_id, + expected_repository=args.expected_repository, + ) + print("policy approval evidence verified") + return 0 + if args.base_sha is None or args.head_sha is None or args.report is None: + parser.error("ratchet comparison requires --base-sha, --head-sha, and --report") + passed = run_ratchet( + repo=args.repo, + base_sha=args.base_sha, + head_sha=args.head_sha, + report_path=args.report, + authorization_approved=args.authorization_approved, + ) + except RatchetError as exc: + print(f"quality ratchet operational failure: {exc}", file=sys.stderr) + return 2 + return 0 if passed else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/fixtures/ci_quality_ratchet/mypy.txt b/tests/fixtures/ci_quality_ratchet/mypy.txt new file mode 100644 index 00000000..219845cd --- /dev/null +++ b/tests/fixtures/ci_quality_ratchet/mypy.txt @@ -0,0 +1,3 @@ +src/tether/example.py:18:5: error: Incompatible return value type [return-value] +src/tether/example.py:18:5: note: Expected a string +Found 1 error in 1 file (checked 1 source file) diff --git a/tests/fixtures/ci_quality_ratchet/ruff_check.json b/tests/fixtures/ci_quality_ratchet/ruff_check.json new file mode 100644 index 00000000..e221cf12 --- /dev/null +++ b/tests/fixtures/ci_quality_ratchet/ruff_check.json @@ -0,0 +1,8 @@ +[ + { + "code": "F401", + "filename": "/checkout/src/tether/example.py", + "location": {"column": 8, "row": 12}, + "message": "`os` imported but unused" + } +] diff --git a/tests/fixtures/ci_quality_ratchet/ruff_format.txt b/tests/fixtures/ci_quality_ratchet/ruff_format.txt new file mode 100644 index 00000000..187c5491 --- /dev/null +++ b/tests/fixtures/ci_quality_ratchet/ruff_format.txt @@ -0,0 +1,2 @@ +Would reformat: src/tether/example.py +1 file would be reformatted diff --git a/tests/test_ci_quality_ratchet.py b/tests/test_ci_quality_ratchet.py new file mode 100644 index 00000000..5799867f --- /dev/null +++ b/tests/test_ci_quality_ratchet.py @@ -0,0 +1,497 @@ +"""Unit tests for the protected-SHA quality ratchet.""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from scripts.ci_quality_ratchet import ( + DiffHunk, + Finding, + RatchetError, + changed_authorization_inputs, + compare_findings, + is_baseline_artifact, + main, + parse_mypy, + parse_ruff_check, + parse_ruff_format, + relocate_base_line, + run_ratchet, + validate_tool_policy_text, + verify_current_main_binding, + verify_policy_environment, + verify_policy_approval_evidence, +) + +FIXTURES = Path(__file__).parent / "fixtures" / "ci_quality_ratchet" + + +def _finding(*, rule: str = "F401", line: int = 10, message: str = "unused") -> Finding: + return Finding("ruff-check", rule, "src/tether/example.py", message, line, 1) + + +def _git(repo: Path, *args: str) -> str: + result = subprocess.run( + ["git", *args], + cwd=repo, + text=True, + capture_output=True, + check=True, + ) + return result.stdout.strip() + + +def test_tool_specific_parsers_produce_stable_fingerprints(tmp_path: Path) -> None: + ruff_value = json.loads((FIXTURES / "ruff_check.json").read_text()) + ruff_value[0]["filename"] = str(tmp_path / "src/tether/example.py") + ruff = parse_ruff_check(json.dumps(ruff_value), checkout=tmp_path) + formatted = parse_ruff_format( + (FIXTURES / "ruff_format.txt").read_text(), + checkout=tmp_path, + ) + mypy = parse_mypy((FIXTURES / "mypy.txt").read_text(), checkout=tmp_path) + + assert ruff[0].fingerprint == ( + "ruff-check", + "F401", + "src/tether/example.py", + "`os` imported but unused", + ) + assert formatted[0].fingerprint == ( + "ruff-format", + "format", + "src/tether/example.py", + "File is not formatted", + ) + assert mypy[0].fingerprint == ( + "mypy", + "return-value", + "src/tether/example.py", + "Incompatible return value type", + ) + assert len(mypy) == 1 + + +def test_diff_relocation_handles_large_unrelated_insertions() -> None: + hunk = DiffHunk(old_start=0, old_count=0, new_start=1, new_count=200) + assert relocate_base_line(10, [hunk]) == 210 + comparison = compare_findings( + [_finding(line=10)], + [_finding(line=210)], + diff_hunks=lambda _path: [hunk], + ) + assert comparison.passed + assert comparison.relocated_matches == 1 + assert not comparison.new_findings + + +def test_new_fingerprint_fails_even_when_total_count_decreases() -> None: + comparison = compare_findings( + [_finding(rule="F401"), _finding(rule="F841")], + [_finding(rule="E722", message="bare except")], + diff_hunks=lambda _path: [], + ) + assert not comparison.passed + assert [item.rule for item in comparison.new_findings] == ["E722"] + assert comparison.head_count < comparison.base_count + + +def test_duplicate_fingerprint_increase_is_new_debt() -> None: + comparison = compare_findings( + [_finding(line=10)], + [_finding(line=10), _finding(line=30)], + diff_hunks=lambda _path: [], + ) + assert not comparison.passed + assert len(comparison.new_findings) == 1 + assert comparison.new_findings[0].line == 30 + + +def test_removed_findings_are_allowed() -> None: + comparison = compare_findings( + [_finding(rule="F401"), _finding(rule="F841")], + [_finding(rule="F401")], + diff_hunks=lambda _path: [], + ) + assert comparison.passed + assert [item.rule for item in comparison.removed_findings] == ["F841"] + + +def test_mutable_baseline_artifact_names_are_reserved() -> None: + assert is_baseline_artifact(".ci/quality-baseline/findings.json") + assert is_baseline_artifact("quality-baseline.json") + assert is_baseline_artifact(".github/quality-baseline/ruff.json") + assert not is_baseline_artifact(".github/workflows/quality-ratchet.yml") + assert not is_baseline_artifact("scripts/ci_quality_ratchet.py") + + +def test_workflows_separate_ordinary_and_protected_policy_changes() -> None: + workflows = Path(__file__).parents[1] / ".github" / "workflows" + ordinary = (workflows / "quality-ratchet.yml").read_text() + protected = (workflows / "quality-baseline-update.yml").read_text() + + assert "pull_request_target:" in ordinary + assert "${BASE_SHA}:scripts/ci_quality_ratchet.py" in ordinary + assert "quality-ratchet/policy-approved" in ordinary + assert "quality-policy-approval-${HEAD_SHA}-${RUN_ID}" in ordinary + assert '--expected-candidate-sha "$HEAD_SHA"' in ordinary + assert '--expected-base-sha "$BASE_SHA"' in ordinary + assert "artifact-ids: ${{ steps.authorization.outputs.artifact_id }}" in ordinary + assert "Bind the event to the current protected-main tip" in ordinary + assert "--verify-main-ref-response" in ordinary + assert "--expected-current-main-sha" in ordinary + assert ordinary.index("Bind the event to the current protected-main tip") < ordinary.index( + "Verify protected policy authorization" + ) + assert "Reconfirm current protected-main tip before publishing" in ordinary + assert ordinary.index( + "Reconfirm current protected-main tip before publishing" + ) < ordinary.index("Publish protected result on the exact head SHA") + assert "workflow_dispatch:" in protected + assert "validate-candidate:" in protected + assert "exercise-candidate:" in protected + assert "authorize-policy-update:" in protected + assert "needs: validate-candidate" in protected + assert "needs: [validate-candidate, exercise-candidate]" in protected + assert protected.index("validate-candidate:") < protected.index("exercise-candidate:") + assert protected.index("exercise-candidate:") < protected.index("authorize-policy-update:") + assert "name: quality-ratchet-policy" in protected + assert "--verify-policy-environment" in protected + assert "statuses: write" in protected.split("authorize-policy-update:", 1)[1] + assert "statuses: write" not in protected.split("authorize-policy-update:", 1)[0] + assert "Upload immutable validation evidence before candidate execution" in protected + assert "quality-policy-approval-${CANDIDATE_SHA}-${GITHUB_RUN_ID}" in protected + assert "artifact-ids: ${{ steps.validation.outputs.artifact_id }}" in protected + pre_seal = protected.split("exercise-candidate:", 1)[0] + untrusted = protected.split("exercise-candidate:", 1)[1].split("authorize-policy-update:", 1)[0] + assert "--validate-policy-syntax" in pre_seal + assert "--show-settings" not in pre_seal + assert "--no-error-summary -c 'pass'" not in pre_seal + assert "--show-settings" in untrusted + assert "--no-error-summary -c 'pass'" in untrusted + assert "Reconfirm current protected tip immediately before approval" in protected + assert protected.index( + "Reconfirm current protected tip immediately before approval" + ) < protected.index("Build immutable approval evidence") + assert "Reconfirm current protected tip immediately before status" in protected + assert protected.index( + "Reconfirm current protected tip immediately before status" + ) < protected.index("Authorize the exact evidence-bound commit") + + +def test_policy_approval_evidence_is_bound_to_exact_sha_base_and_run() -> None: + candidate = "a" * 40 + base = "b" * 40 + value = { + "base_sha": base, + "candidate_sha": candidate, + "repository": "FastCrest/tether", + "run_id": 12345, + "schema_version": 1, + "validation_artifact_digest": "sha256:" + "c" * 64, + "validation_artifact_id": 67890, + "workflow_path": ".github/workflows/quality-baseline-update.yml", + } + verify_policy_approval_evidence( + value, + expected_candidate_sha=candidate, + expected_base_sha=base, + expected_run_id=12345, + expected_repository="FastCrest/tether", + ) + + tampered_values = [ + {**value, "candidate_sha": "d" * 40}, + {**value, "base_sha": "d" * 40}, + {**value, "run_id": 54321}, + {**value, "repository": "attacker/fork"}, + {**value, "workflow_path": ".github/workflows/unrelated.yml"}, + {**value, "validation_artifact_digest": "sha256:short"}, + {**value, "extra": "ambiguous"}, + ] + for tampered in tampered_values: + with pytest.raises(RatchetError): + verify_policy_approval_evidence( + tampered, + expected_candidate_sha=candidate, + expected_base_sha=base, + expected_run_id=12345, + expected_repository="FastCrest/tether", + ) + + +def test_current_main_binding_rejects_stale_event_reruns() -> None: + base = "a" * 40 + head = "b" * 40 + current = "c" * 40 + ref_value = { + "ref": "refs/heads/main", + "object": {"type": "commit", "sha": base}, + } + pull_request_value = { + "state": "open", + "base": {"ref": "main", "sha": base}, + "head": {"sha": head}, + } + verify_current_main_binding( + ref_value, + expected_main_sha=base, + pull_request_value=pull_request_value, + expected_head_sha=head, + ) + + stale_cases = [ + ( + {"ref": "refs/heads/main", "object": {"type": "commit", "sha": current}}, + pull_request_value, + ), + (ref_value, {**pull_request_value, "base": {"ref": "main", "sha": current}}), + (ref_value, {**pull_request_value, "head": {"sha": current}}), + ] + for stale_ref, stale_pull_request in stale_cases: + with pytest.raises(RatchetError): + verify_current_main_binding( + stale_ref, + expected_main_sha=base, + pull_request_value=stale_pull_request, + expected_head_sha=head, + ) + + +def test_policy_environment_requires_exact_reviewer_allowlist() -> None: + trusted = {"type": "User", "reviewer": {"login": "rylinjames"}} + valid = { + "protection_rules": [ + {"type": "required_reviewers", "reviewers": [trusted]}, + ], + "deployment_branch_policy": { + "protected_branches": True, + "custom_branch_policies": False, + }, + } + verify_policy_environment(valid) + + untrusted = {"type": "User", "reviewer": {"login": "untrusted-reviewer"}} + invalid_values = [ + {**valid, "protection_rules": []}, + { + **valid, + "protection_rules": [{"type": "required_reviewers", "reviewers": [trusted, untrusted]}], + }, + { + **valid, + "protection_rules": [{"type": "required_reviewers", "reviewers": [trusted, trusted]}], + }, + { + **valid, + "protection_rules": [ + { + "type": "required_reviewers", + "reviewers": [{"type": "Team", "reviewer": {"login": "rylinjames"}}], + } + ], + }, + ] + for invalid in invalid_values: + with pytest.raises(RatchetError): + verify_policy_environment(invalid) + + +def test_candidate_policy_rejects_mypy_plugin_without_executing_or_tampering( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + marker = tmp_path / "plugin-executed" + evidence = tmp_path / "protected-evidence.json" + evidence.write_text('{"sealed": true}\n') + (tmp_path / "malicious_plugin.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed')\n" + f"Path({str(evidence)!r}).write_text('tampered')\n" + "def plugin(version):\n raise RuntimeError(version)\n" + ) + policy = tmp_path / "candidate-pyproject.toml" + policy.write_text( + '[tool.ruff]\nline-length = 88\n[tool.mypy]\nplugins = ["malicious_plugin.py"]\n' + ) + monkeypatch.chdir(tmp_path) + + with pytest.raises(RatchetError): + validate_tool_policy_text(policy.read_text()) + assert main(["--validate-policy-syntax", str(policy)]) == 2 + assert not marker.exists() + assert evidence.read_text() == '{"sealed": true}\n' + + # Control: the same config really does execute the plugin when mypy loads it. + control = subprocess.run( + [ + sys.executable, + "-I", + "-m", + "mypy", + f"--config-file={policy}", + "--no-error-summary", + "-c", + "pass", + ], + cwd=tmp_path, + text=True, + capture_output=True, + check=False, + ) + assert control.returncode != 0 + assert marker.read_text() == "executed" + assert evidence.read_text() == "tampered" + + +def test_pr_modified_mypy_plugin_never_executes_in_ratchet(tmp_path: Path) -> None: + repo = tmp_path / "plugin-head" + repo.mkdir() + _git(repo, "init", "-b", "main") + _git(repo, "config", "user.email", "ratchet@example.test") + _git(repo, "config", "user.name", "Ratchet Test") + (repo / "pyproject.toml").write_text( + '[project]\nname="plugin-head"\nversion="1"\n' + '[tool.ruff]\ntarget-version="py310"\n' + '[tool.mypy]\npython_version="3.10"\nplugins=["local_plugin.py"]\n' + ) + (repo / "local_plugin.py").write_text( + "from mypy.plugin import Plugin\n" + "class LocalPlugin(Plugin):\n pass\n" + "def plugin(version):\n return LocalPlugin\n" + ) + _git(repo, "add", ".") + _git(repo, "commit", "-m", "base with forbidden plugin policy") + base = _git(repo, "rev-parse", "HEAD") + + marker = tmp_path / "head-plugin-executed" + (repo / "local_plugin.py").write_text( + "from pathlib import Path\n" + f"Path({str(marker)!r}).write_text('executed')\n" + "from mypy.plugin import Plugin\n" + "class LocalPlugin(Plugin):\n pass\n" + "def plugin(version):\n return LocalPlugin\n" + ) + _git(repo, "commit", "-am", "modify local plugin in pull request") + head = _git(repo, "rev-parse", "HEAD") + + report = tmp_path / "plugin-head-report.json" + with pytest.raises(RatchetError, match="mypy plugins are forbidden"): + run_ratchet(repo=repo, base_sha=base, head_sha=head, report_path=report) + assert not marker.exists() + assert not report.exists() + + +def test_authorization_inputs_distinguish_dependency_and_tool_policy_changes( + tmp_path: Path, +) -> None: + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init", "-b", "main") + _git(repo, "config", "user.email", "ratchet@example.test") + _git(repo, "config", "user.name", "Ratchet Test") + (repo / "pyproject.toml").write_text( + '[project]\nname="example"\nversion="1"\n' + '[tool.ruff]\nline-length=100\n[tool.mypy]\npython_version="3.10"\n' + ) + _git(repo, "add", ".") + _git(repo, "commit", "-m", "base") + base = _git(repo, "rev-parse", "HEAD") + + (repo / "pyproject.toml").write_text( + '[project]\nname="example"\nversion="2"\n' + '[tool.ruff]\nline-length=100\n[tool.mypy]\npython_version="3.10"\n' + ) + _git(repo, "commit", "-am", "dependency metadata only") + metadata_head = _git(repo, "rev-parse", "HEAD") + assert changed_authorization_inputs(repo, base, metadata_head) == [] + + with (repo / "pyproject.toml").open("a") as handle: + handle.write("ignore_errors=true\n") + _git(repo, "commit", "-am", "weaken mypy") + policy_head = _git(repo, "rev-parse", "HEAD") + assert changed_authorization_inputs(repo, metadata_head, policy_head) == [ + "pyproject.toml#[tool.mypy]" + ] + + +def test_adversarial_head_cannot_hide_new_debt_or_replace_judge(tmp_path: Path) -> None: + repo = tmp_path / "adversarial" + repo.mkdir() + _git(repo, "init", "-b", "main") + _git(repo, "config", "user.email", "ratchet@example.test") + _git(repo, "config", "user.name", "Ratchet Test") + (repo / "src").mkdir() + (repo / "scripts").mkdir() + (repo / ".github/workflows").mkdir(parents=True) + (repo / "pyproject.toml").write_text( + '[project]\nname="adversarial"\nversion="1"\n' + '[tool.ruff]\ntarget-version="py310"\nexclude=["sitecustomize.py"]\n' + '[tool.mypy]\npython_version="3.10"\n' + ) + (repo / "src/example.py").write_text("import os\n\n\ndef existing() -> str:\n return 1\n") + (repo / "scripts/ci_quality_ratchet.py").write_text("PROTECTED = True\n") + (repo / ".github/workflows/quality-ratchet.yml").write_text("name: protected\n") + _git(repo, "add", ".") + _git(repo, "commit", "-m", "base debt") + base = _git(repo, "rev-parse", "HEAD") + + (repo / "pyproject.toml").write_text( + '[project]\nname="adversarial"\nversion="1"\n' + '[tool.ruff]\ntarget-version="py310"\nexclude=["src"]\n' + '[tool.mypy]\npython_version="3.10"\nignore_errors=true\n' + ) + (repo / "scripts/ci_quality_ratchet.py").write_text("raise SystemExit(0)\n") + (repo / ".github/workflows/quality-ratchet.yml").write_text("name: bypass\n") + marker = tmp_path / "sitecustomize-executed" + (repo / "sitecustomize.py").write_text( + f"from pathlib import Path\nPath({str(marker)!r}).write_text('executed')\n" + ) + _git(repo, "add", ".") + _git(repo, "commit", "-m", "attempt policy bypass") + policy_head = _git(repo, "rev-parse", "HEAD") + + policy_report = tmp_path / "policy-report.json" + assert not run_ratchet( + repo=repo, + base_sha=base, + head_sha=policy_head, + report_path=policy_report, + ) + policy_value = json.loads(policy_report.read_text()) + assert policy_value["base_count"] == policy_value["head_count"] == 2 + assert policy_value["new_findings"] == [] + assert not policy_value["authorization_approved"] + + (repo / "src/example.py").write_text( + "import os\nimport sys\n\n\ndef existing() -> str:\n return 1\n" + "\n\ndef added() -> str:\n return 2\n" + ) + _git(repo, "add", "src/example.py") + _git(repo, "commit", "-m", "add hidden debt") + head = _git(repo, "rev-parse", "HEAD") + + report = tmp_path / "adversarial-report.json" + assert not run_ratchet( + repo=repo, + base_sha=base, + head_sha=head, + report_path=report, + authorization_approved=True, + ) + value = json.loads(report.read_text()) + assert value["base_count"] == 2 + assert value["head_count"] == 4 + assert len(value["new_findings"]) == 2 + assert not marker.exists() + assert value["authorization_input_changes"] == [ + ".github/workflows/quality-ratchet.yml", + "pyproject.toml#[tool.mypy]", + "pyproject.toml#[tool.ruff]", + "scripts/ci_quality_ratchet.py", + ]