From 0b24fabf980882576df1a083a9a7cd4f146ffc7d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 14 Oct 2026 11:17:24 +0900 Subject: [PATCH 1/2] fix(ci): [verified] bind QSR coverage evidence per run --- .../quarantine-sandbox-runtime-ci.yml | 321 ++++++++- ...arantine-sandbox-runtime-self-hosted-ci.md | 92 ++- ..._quarantine_sandbox_runtime_ci_contract.py | 657 +++++++++++++++++- 3 files changed, 1041 insertions(+), 29 deletions(-) diff --git a/.github/workflows/quarantine-sandbox-runtime-ci.yml b/.github/workflows/quarantine-sandbox-runtime-ci.yml index 7e8760f9eb..fd509c9e56 100644 --- a/.github/workflows/quarantine-sandbox-runtime-ci.yml +++ b/.github/workflows/quarantine-sandbox-runtime-ci.yml @@ -109,6 +109,11 @@ jobs: runs-on: group: CWL QSR hostile workload labels: [self-hosted, linux, x64, cwl-hostile-workload] + env: + QSR_RUN_ID: ${{ github.run_id }} + QSR_RUN_ATTEMPT: ${{ github.run_attempt }} + QSR_EVIDENCE_JOB: coverage + QSR_EXPECTED_HEAD: ${{ github.event.pull_request.head.sha || github.sha }} steps: - name: Verify caller and event before checkout env: @@ -149,21 +154,156 @@ jobs: - name: Install pinned coverage tool run: cargo +1.97.1 install cargo-llvm-cov --locked --version 0.8.6 - name: Generate production coverage evidence - run: >- - cargo llvm-cov - --workspace - --lib - --tests - --json - --output-path coverage.json + env: + QSR_EVIDENCE_ROOT: ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-coverage-${{ github.event.pull_request.head.sha || github.sha }} + run: | + python3 -I - <<'PY' + import hashlib + import json + import os + from pathlib import Path + import re + import stat + import subprocess + + identity = {key: os.environ[key] for key in ( + "QSR_RUN_ID", "QSR_RUN_ATTEMPT", "QSR_EVIDENCE_JOB", "QSR_EXPECTED_HEAD", + )} + for key in ("QSR_RUN_ID", "QSR_RUN_ATTEMPT"): + if not re.fullmatch(r"[1-9][0-9]*", identity[key]): + raise SystemExit("SRC03-EVIDENCE-001: invalid run identity") + if not re.fullmatch(r"[0-9a-f]{40}", identity["QSR_EXPECTED_HEAD"]): + raise SystemExit("SRC03-EVIDENCE-001: invalid head identity") + lane = identity["QSR_EVIDENCE_JOB"] + if lane not in ("coverage", "branch-coverage"): + raise SystemExit("SRC03-EVIDENCE-001: invalid evidence lane") + parent = Path(os.environ["RUNNER_TEMP"]).resolve(strict=True) + root = Path(os.environ["QSR_EVIDENCE_ROOT"]) + leaf = "qsr-evidence-{QSR_RUN_ID}-{QSR_RUN_ATTEMPT}-{QSR_EVIDENCE_JOB}-{QSR_EXPECTED_HEAD}".format(**identity) + if not parent.is_dir() or root != parent / leaf: + raise SystemExit("SRC03-EVIDENCE-001: invalid evidence root") + head = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() + if head != identity["QSR_EXPECTED_HEAD"]: + raise SystemExit("SRC03-EVIDENCE-001: checkout drift") + os.mkdir(root, 0o700) + root_stat = root.lstat() + artifact = root / ("branch-coverage.json" if lane == "branch-coverage" else "coverage.json") + command = ["cargo"] + if lane == "branch-coverage": + command += ["+nightly-2026-07-01"] + command += ["llvm-cov"] + if lane == "branch-coverage": + command += ["--branch"] + command += ["--workspace", "--lib", "--tests", "--json", "--output-path", str(artifact)] + completed = subprocess.run(command) + if completed.returncode: + raise SystemExit(completed.returncode) + if subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() != head: + raise SystemExit("SRC03-EVIDENCE-001: checkout drift") + with os.fdopen(os.open(artifact, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK), "rb") as handle: + metadata = os.fstat(handle.fileno()) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size == 0 or metadata.st_nlink != 1: + raise SystemExit("SRC03-EVIDENCE-001: nonregular or empty evidence") + hasher = hashlib.sha256() + counted = 0 + while chunk := handle.read(min(65536, metadata.st_size - counted + 1)): + counted += len(chunk) + if counted > metadata.st_size: + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + hasher.update(chunk) + after = os.fstat(handle.fileno()) + stable = ("st_dev", "st_ino", "st_mode", "st_uid", "st_nlink", "st_size", "st_mtime_ns", "st_ctime_ns") + if counted != metadata.st_size or any(getattr(after, key) != getattr(metadata, key) for key in stable): + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + digest = hasher.hexdigest() + receipt = {**identity, "artifact": artifact.name, "bytes": counted, + "sha256": digest, "producer_exit": completed.returncode, + "root_device": root_stat.st_dev, "root_inode": root_stat.st_ino} + manifest = json.dumps(receipt, sort_keys=True).encode("utf-8") + if len(manifest) > 4096: + raise SystemExit("SRC03-EVIDENCE-001: manifest too large") + with (root / "binding.json").open("x", encoding="utf-8") as handle: + print(manifest.decode("utf-8"), end="", file=handle) + PY + - name: Admit fresh bound coverage evidence before upload + env: + QSR_EVIDENCE_ROOT: ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-coverage-${{ github.event.pull_request.head.sha || github.sha }} + run: | + python3 -I - <<'PY' + import hashlib + import json + import os + from pathlib import Path + import stat + import subprocess + + root = Path(os.environ["QSR_EVIDENCE_ROOT"]) + parent = Path(os.environ["RUNNER_TEMP"]).resolve(strict=True) + identity = {key: os.environ[key] for key in ( + "QSR_RUN_ID", "QSR_RUN_ATTEMPT", "QSR_EVIDENCE_JOB", "QSR_EXPECTED_HEAD", + )} + leaf = "qsr-evidence-{QSR_RUN_ID}-{QSR_RUN_ATTEMPT}-{QSR_EVIDENCE_JOB}-{QSR_EXPECTED_HEAD}".format(**identity) + metadata = root.lstat() + if root != parent / leaf or not stat.S_ISDIR(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o700 or metadata.st_uid != os.getuid(): + raise SystemExit("SRC03-EVIDENCE-001: evidence root changed") + def regular_bytes(path): + with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK), "rb") as handle: + info = os.fstat(handle.fileno()) + if not stat.S_ISREG(info.st_mode) or not info.st_size or info.st_nlink != 1: + raise SystemExit("SRC03-EVIDENCE-001: nonregular or empty evidence") + if path.name == "binding.json": + if info.st_size > 4096: + raise SystemExit("SRC03-EVIDENCE-001: manifest too large") + data = handle.read(4097) + if len(data) > 4096: + raise SystemExit("SRC03-EVIDENCE-001: manifest too large") + after = os.fstat(handle.fileno()) + stable = ("st_dev", "st_ino", "st_mode", "st_uid", "st_nlink", "st_size", "st_mtime_ns", "st_ctime_ns") + if len(data) != info.st_size or any(getattr(after, key) != getattr(info, key) for key in stable): + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + return data + hasher = hashlib.sha256() + counted = 0 + while chunk := handle.read(min(65536, info.st_size - counted + 1)): + counted += len(chunk) + if counted > info.st_size: + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + hasher.update(chunk) + after = os.fstat(handle.fileno()) + stable = ("st_dev", "st_ino", "st_mode", "st_uid", "st_nlink", "st_size", "st_mtime_ns", "st_ctime_ns") + if counted != info.st_size or any(getattr(after, key) != getattr(info, key) for key in stable): + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + return counted, hasher.hexdigest() + def unique_fields(pairs): + result = {} + for key, value in pairs: + if key in result: + raise SystemExit("SRC03-EVIDENCE-001: duplicate binding field") + result[key] = value + return result + receipt = json.loads(regular_bytes(root / "binding.json"), object_pairs_hook=unique_fields) + name = "branch-coverage.json" if identity["QSR_EVIDENCE_JOB"] == "branch-coverage" else "coverage.json" + counted, digest = regular_bytes(root / name) + expected = {**identity, "artifact": name, "bytes": counted, + "sha256": digest, "producer_exit": 0, + "root_device": metadata.st_dev, "root_inode": metadata.st_ino} + if type(receipt) is not dict or any(type(receipt.get(key)) is not type(value) for key, value in expected.items()) or receipt != expected: + raise SystemExit("SRC03-EVIDENCE-001: stale or mismatched evidence binding") + if subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() != identity["QSR_EXPECTED_HEAD"]: + raise SystemExit("SRC03-EVIDENCE-001: checkout drift") + PY - name: Upload production coverage evidence uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 with: - name: coverage-json-${{ github.event.pull_request.head.sha || github.sha }} - path: coverage.json + name: coverage-json-${{ github.event.pull_request.head.sha || github.sha }}-${{ github.run_id }}-${{ github.run_attempt }} + path: | + ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-coverage-${{ github.event.pull_request.head.sha || github.sha }}/coverage.json + ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-coverage-${{ github.event.pull_request.head.sha || github.sha }}/binding.json if-no-files-found: error - name: Explain and enforce complete production coverage - run: python3 scripts/check_coverage.py coverage.json + env: + QSR_EVIDENCE_ROOT: ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-coverage-${{ github.event.pull_request.head.sha || github.sha }} + run: python3 scripts/check_coverage.py "$QSR_EVIDENCE_ROOT/coverage.json" branch-coverage: needs: admission @@ -175,6 +315,11 @@ jobs: runs-on: group: CWL QSR hostile workload labels: [self-hosted, linux, x64, cwl-hostile-workload] + env: + QSR_RUN_ID: ${{ github.run_id }} + QSR_RUN_ATTEMPT: ${{ github.run_attempt }} + QSR_EVIDENCE_JOB: branch-coverage + QSR_EXPECTED_HEAD: ${{ github.event.pull_request.head.sha || github.sha }} steps: - name: Verify caller and event before checkout env: @@ -215,27 +360,161 @@ jobs: - name: Install pinned coverage tool run: cargo +nightly-2026-07-01 install cargo-llvm-cov --locked --version 0.8.6 - name: Generate branch coverage evidence - run: >- - cargo +nightly-2026-07-01 llvm-cov - --branch - --workspace - --lib - --tests - --json - --output-path branch-coverage.json + env: + QSR_EVIDENCE_ROOT: ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-branch-coverage-${{ github.event.pull_request.head.sha || github.sha }} + run: | + python3 -I - <<'PY' + import hashlib + import json + import os + from pathlib import Path + import re + import stat + import subprocess + + identity = {key: os.environ[key] for key in ( + "QSR_RUN_ID", "QSR_RUN_ATTEMPT", "QSR_EVIDENCE_JOB", "QSR_EXPECTED_HEAD", + )} + for key in ("QSR_RUN_ID", "QSR_RUN_ATTEMPT"): + if not re.fullmatch(r"[1-9][0-9]*", identity[key]): + raise SystemExit("SRC03-EVIDENCE-001: invalid run identity") + if not re.fullmatch(r"[0-9a-f]{40}", identity["QSR_EXPECTED_HEAD"]): + raise SystemExit("SRC03-EVIDENCE-001: invalid head identity") + lane = identity["QSR_EVIDENCE_JOB"] + if lane not in ("coverage", "branch-coverage"): + raise SystemExit("SRC03-EVIDENCE-001: invalid evidence lane") + parent = Path(os.environ["RUNNER_TEMP"]).resolve(strict=True) + root = Path(os.environ["QSR_EVIDENCE_ROOT"]) + leaf = "qsr-evidence-{QSR_RUN_ID}-{QSR_RUN_ATTEMPT}-{QSR_EVIDENCE_JOB}-{QSR_EXPECTED_HEAD}".format(**identity) + if not parent.is_dir() or root != parent / leaf: + raise SystemExit("SRC03-EVIDENCE-001: invalid evidence root") + head = subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() + if head != identity["QSR_EXPECTED_HEAD"]: + raise SystemExit("SRC03-EVIDENCE-001: checkout drift") + os.mkdir(root, 0o700) + root_stat = root.lstat() + artifact = root / ("branch-coverage.json" if lane == "branch-coverage" else "coverage.json") + command = ["cargo"] + if lane == "branch-coverage": + command += ["+nightly-2026-07-01"] + command += ["llvm-cov"] + if lane == "branch-coverage": + command += ["--branch"] + command += ["--workspace", "--lib", "--tests", "--json", "--output-path", str(artifact)] + completed = subprocess.run(command) + if completed.returncode: + raise SystemExit(completed.returncode) + if subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() != head: + raise SystemExit("SRC03-EVIDENCE-001: checkout drift") + with os.fdopen(os.open(artifact, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK), "rb") as handle: + metadata = os.fstat(handle.fileno()) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size == 0 or metadata.st_nlink != 1: + raise SystemExit("SRC03-EVIDENCE-001: nonregular or empty evidence") + hasher = hashlib.sha256() + counted = 0 + while chunk := handle.read(min(65536, metadata.st_size - counted + 1)): + counted += len(chunk) + if counted > metadata.st_size: + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + hasher.update(chunk) + after = os.fstat(handle.fileno()) + stable = ("st_dev", "st_ino", "st_mode", "st_uid", "st_nlink", "st_size", "st_mtime_ns", "st_ctime_ns") + if counted != metadata.st_size or any(getattr(after, key) != getattr(metadata, key) for key in stable): + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + digest = hasher.hexdigest() + receipt = {**identity, "artifact": artifact.name, "bytes": counted, + "sha256": digest, "producer_exit": completed.returncode, + "root_device": root_stat.st_dev, "root_inode": root_stat.st_ino} + manifest = json.dumps(receipt, sort_keys=True).encode("utf-8") + if len(manifest) > 4096: + raise SystemExit("SRC03-EVIDENCE-001: manifest too large") + with (root / "binding.json").open("x", encoding="utf-8") as handle: + print(manifest.decode("utf-8"), end="", file=handle) + PY - name: Print missing branch coverage lines run: >- cargo +nightly-2026-07-01 llvm-cov report --branch --show-missing-lines + - name: Admit fresh bound coverage evidence before upload + env: + QSR_EVIDENCE_ROOT: ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-branch-coverage-${{ github.event.pull_request.head.sha || github.sha }} + run: | + python3 -I - <<'PY' + import hashlib + import json + import os + from pathlib import Path + import stat + import subprocess + + root = Path(os.environ["QSR_EVIDENCE_ROOT"]) + parent = Path(os.environ["RUNNER_TEMP"]).resolve(strict=True) + identity = {key: os.environ[key] for key in ( + "QSR_RUN_ID", "QSR_RUN_ATTEMPT", "QSR_EVIDENCE_JOB", "QSR_EXPECTED_HEAD", + )} + leaf = "qsr-evidence-{QSR_RUN_ID}-{QSR_RUN_ATTEMPT}-{QSR_EVIDENCE_JOB}-{QSR_EXPECTED_HEAD}".format(**identity) + metadata = root.lstat() + if root != parent / leaf or not stat.S_ISDIR(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o700 or metadata.st_uid != os.getuid(): + raise SystemExit("SRC03-EVIDENCE-001: evidence root changed") + def regular_bytes(path): + with os.fdopen(os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK), "rb") as handle: + info = os.fstat(handle.fileno()) + if not stat.S_ISREG(info.st_mode) or not info.st_size or info.st_nlink != 1: + raise SystemExit("SRC03-EVIDENCE-001: nonregular or empty evidence") + if path.name == "binding.json": + if info.st_size > 4096: + raise SystemExit("SRC03-EVIDENCE-001: manifest too large") + data = handle.read(4097) + if len(data) > 4096: + raise SystemExit("SRC03-EVIDENCE-001: manifest too large") + after = os.fstat(handle.fileno()) + stable = ("st_dev", "st_ino", "st_mode", "st_uid", "st_nlink", "st_size", "st_mtime_ns", "st_ctime_ns") + if len(data) != info.st_size or any(getattr(after, key) != getattr(info, key) for key in stable): + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + return data + hasher = hashlib.sha256() + counted = 0 + while chunk := handle.read(min(65536, info.st_size - counted + 1)): + counted += len(chunk) + if counted > info.st_size: + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + hasher.update(chunk) + after = os.fstat(handle.fileno()) + stable = ("st_dev", "st_ino", "st_mode", "st_uid", "st_nlink", "st_size", "st_mtime_ns", "st_ctime_ns") + if counted != info.st_size or any(getattr(after, key) != getattr(info, key) for key in stable): + raise SystemExit("SRC03-EVIDENCE-001: evidence changed while reading") + return counted, hasher.hexdigest() + def unique_fields(pairs): + result = {} + for key, value in pairs: + if key in result: + raise SystemExit("SRC03-EVIDENCE-001: duplicate binding field") + result[key] = value + return result + receipt = json.loads(regular_bytes(root / "binding.json"), object_pairs_hook=unique_fields) + name = "branch-coverage.json" if identity["QSR_EVIDENCE_JOB"] == "branch-coverage" else "coverage.json" + counted, digest = regular_bytes(root / name) + expected = {**identity, "artifact": name, "bytes": counted, + "sha256": digest, "producer_exit": 0, + "root_device": metadata.st_dev, "root_inode": metadata.st_ino} + if type(receipt) is not dict or any(type(receipt.get(key)) is not type(value) for key, value in expected.items()) or receipt != expected: + raise SystemExit("SRC03-EVIDENCE-001: stale or mismatched evidence binding") + if subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip() != identity["QSR_EXPECTED_HEAD"]: + raise SystemExit("SRC03-EVIDENCE-001: checkout drift") + PY - name: Upload branch coverage evidence uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 with: - name: branch-coverage-json-${{ github.event.pull_request.head.sha || github.sha }} - path: branch-coverage.json + name: branch-coverage-json-${{ github.event.pull_request.head.sha || github.sha }}-${{ github.run_id }}-${{ github.run_attempt }} + path: | + ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-branch-coverage-${{ github.event.pull_request.head.sha || github.sha }}/branch-coverage.json + ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-branch-coverage-${{ github.event.pull_request.head.sha || github.sha }}/binding.json if-no-files-found: error - name: Enforce complete production branch coverage - run: python3 scripts/check_coverage.py --require-branches branch-coverage.json + env: + QSR_EVIDENCE_ROOT: ${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-branch-coverage-${{ github.event.pull_request.head.sha || github.sha }} + run: python3 scripts/check_coverage.py --require-branches "$QSR_EVIDENCE_ROOT/branch-coverage.json" podman-e2e-positive-lsm: needs: admission diff --git a/docs/doctoring/quarantine-sandbox-runtime-self-hosted-ci.md b/docs/doctoring/quarantine-sandbox-runtime-self-hosted-ci.md index 96c1fdde62..19ca6349c2 100644 --- a/docs/doctoring/quarantine-sandbox-runtime-self-hosted-ci.md +++ b/docs/doctoring/quarantine-sandbox-runtime-self-hosted-ci.md @@ -58,7 +58,8 @@ candidate with the original SHA-pinned checkout action, `persist-credentials: false`, then compares `git rev-parse HEAD` against the event-derived SHA before installing tools or executing product code. The workflow grants only `contents: read`, contains no publication/write commands, credential inheritance -or write permission, and uploads only the original coverage evidence artifacts. +or write permission. Coverage uploads retain their JSON payloads and add a +run/head/content binding manifest under the SRC03 evidence-only slice. Artifact upload is not a repository-write permission grant. Reusable concurrency is workflow-scoped and namespaced `qsr-central-`, then @@ -83,9 +84,11 @@ There is no GitHub-hosted fallback or user-selected runner routing. The fixture remains `docker.io/library/python@sha256:94457973ea8a27a799f0b8ea1fe3e3147fcbebaed63497a8af63b28195a08108`. -All action SHAs, product commands, environment values and coverage artifacts are -retained from the actual product source, with one independently reproduced -cleanup repair: enumerate networks first and fail on enumeration error before +All action SHAs, compiler/tool pins, product acceptance commands and environment +values are retained. SRC03 changes only the two coverage evidence destinations, +adds binding manifests and names the artifacts by head/run/attempt. The earlier +independently reproduced cleanup repair remains: enumerate networks first and +fail on enumeration error before checking names. The old pipeline's `|| true` incorrectly accepted query failure as zero leaks. Container-query errors and visible container/network leaks remain nonpassing. Synthetic shell controls verify these exits, not real isolation. Production statement/function/region/branch @@ -194,8 +197,85 @@ Python isolated mode prevents candidate-root module imports by this wrapper. Synthetic Cargo-output controls test admission, not Podman or kernel isolation. This check detects absent/renamed tests, not a malicious candidate that forges libtest output or weakens the test body. Code review and disposable host custody -remain necessary. Fresh source/tool/evidence directories (SRC03) and actual -positive-LSM execution are separate, unfinished activation work. +remain necessary. Fresh source/tool/process directories and actual positive-LSM +execution are separate, unfinished activation work. The SRC03 slice below +addresses only coverage artifact-path hygiene, not hostile-writer isolation. + +## SRC03 evidence-path slice (2026-10-14) + +The initial source-bound RED replayed the parsed baseline generation commands +with a reached synthetic Cargo producer returning zero and leaving a symlink to +foreign bytes. Both lanes reached the pre-upload boundary without rejecting it; +regular-file positive controls also passed. This is not an Actions upload claim. +A failed generation already suppresses the default-success upload, so stale +files after a nonzero generation are **not** described as an exploitable RED. + +Both coverage lanes now exclusively create a mode-0700 immediate-child directory +under canonical `RUNNER_TEMP`, named with the admitted run ID, attempt, lane and +candidate HEAD. Existing directories or symlinks fail without deletion or +adoption. The trusted isolated inline producer verifies HEAD before/after Cargo, +retains nonzero exits, and requires a nonempty single-link regular output through +no-follow/nonblocking descriptor reads. It emits an exclusively created +`binding.json` containing run/attempt/lane/head, root device/inode, payload name, +byte count, SHA-256 and the zero producer exit. There is no candidate helper +import, new checkout, token permission or failure masking. + +Immediately before the unchanged SHA-pinned upload action, isolated trusted code +checks root type/owner/mode and recorded inode, rejects duplicate/unknown binding +fields and Boolean-as-integer values, rehashes nonempty regular bytes and checks +HEAD again. Upload includes the JSON and manifest, with head/run/attempt in its +name. The strict product parser still follows upload, using the exact admitted +path, and branch coverage still requires nonzero branch instrumentation and +100% owned production coverage. Upload remains default-success only: failed +production or admission cannot publish through `always()`. + +Synthetic source-bound controls cover legitimate output with spaces in the temp +path; successful symlink/empty/missing/directory/FIFO/hardlink output; nonzero +producer exit; prior root and root symlink; and post-production changes +to run/attempt/head/lane/size/hash/producer exit, duplicate/unknown fields, payload +and root nodes. Foreign sentinels are preserved. These are local wrapper controls, +not actual Cargo coverage generation or `actions/upload-artifact` execution. + +SRC03 review repairs B01/B02/R03 retain literal, reviewed producer/guard script +SHA-256 contracts and the exact immutable upload `uses` value. Mechanics compare +actual step projections, complete env/step schemas and ordering; they do not +replace observed runtime steps with baseline objects. Persistent mutations cover +mutable/alternate-SHA uploads, substituted generators/guards and removed/bypassed +admission. The existing behavioral source controls remain required. + +Both producers use portable `hashlib.sha256().update()` chunk loops, not +`file_digest`; the missing-API child test is a synthetic stdlib-capability seam, +not observed minimum-version or actual runner execution. The project Python +minimum is unchanged. Payloads have no arbitrary size cap: both producer and +reader hash chunks of at most 65536 bytes, count bytes against descriptor size, +and compare device/inode/type/owner/link count/size/mtime/ctime before and after. +Reads consume at most the initially observed size plus one detection byte, +refusing growth rather than following an unbounded append. Access time is not a +stability field because the read itself may update it. + +`binding.json` is limited to **4096 UTF-8 bytes** before exclusive publication and +before decoding. Its closed schema contains four identity strings, a fixed +payload name, a 64-character SHA-256 and four integers; ordinary bindings are +well below this generous small bound. Pathological numeric identity lengths are +refused, not silently truncated. The reader bounds the read call itself to 4097 +bytes, so a misleading size observation cannot permit an unsized allocation. +Tests execute actual bodies on a 2162688-byte regular payload, verify repeated +bounded reads and the independent digest/count, reject oversized serialization +before publication and oversized input even through the synthetic metadata seam, +and detect controlled grow/truncate during reads. These are bounded consistency +controls, not hostile-writer isolation or complete product-parser memory bounds. + +**Incomplete:** checkout, Cargo/Rustup homes, installed tools, build target, +positive-LSM `podman-info.json`, and child process settlement remain shared or +unattested. No evidence-root deletion is added: failed/ambiguous roots remain for +operator-owned reconciliation after settlement. A directory name, mode, hash or +inode does not exclude a same-UID hostile process; product code can forge a +manifest and mutate bytes after admission or during upload. The manifest is +byte consistency, not a signed provenance attestation. Dedicated disposable +host custody, trusted tooling and race-safe artifact consumption remain required +before activation. The owner must independently review this slice and execute +Linux/runtime/Actions gates; the private scratch patch is not published source, +full-suite acceptance, App approval, runner provisioning or release evidence. Primary platform references: GitHub Docs, **Secure use reference** (self-hosted PR-code risk), **Reusing workflow configurations** (caller context/concurrency), diff --git a/tests/test_quarantine_sandbox_runtime_ci_contract.py b/tests/test_quarantine_sandbox_runtime_ci_contract.py index 69845d3790..de5815b790 100644 --- a/tests/test_quarantine_sandbox_runtime_ci_contract.py +++ b/tests/test_quarantine_sandbox_runtime_ci_contract.py @@ -1,6 +1,7 @@ """Static contracts for the future dedicated QSR reusable CI lane.""" from copy import deepcopy +import hashlib from pathlib import Path import os import subprocess @@ -436,7 +437,50 @@ def test_preserves_all_four_job_mechanics(): and step.get("name") not in {"Verify exact checkout", "Verify caller and event before checkout"} ] - if name == "podman-e2e-positive-lsm": + if name in SRC03_LANES: + test_src03_upload_is_run_bound_and_preceded_by_admission() + # Literal reviewed script digests, never derived from workflow() at runtime. + # Behavioral producer/admission controls below remain mandatory supplements. + generator_hash = "091ae4e9cb783d06138a6e62be6ad8255b9c5b47637a23da2628b419e6dacbee" + guard_hash = "af57131f5083f77780dc66ed46d2c70679165bef149732c1309e9745132d7d82" + generator_name, filename = SRC03_LANES[name] + root_expression = ( + "${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-" + "${{ github.run_attempt }}-" + name + "-" + CANDIDATE + ) + expected_steps = deepcopy(expected["steps"]) + for step in expected_steps: + if step.get("name") == generator_name: + step["run"] = generator_hash + step["env"] = {"QSR_EVIDENCE_ROOT": root_expression} + elif "uses" in step and step["uses"].startswith("actions/upload-artifact@"): + step["with"] = { + "name": filename.removesuffix(".json") + "-json-" + CANDIDATE + + "-${{ github.run_id }}-${{ github.run_attempt }}", + "path": root_expression + "/" + filename + "\n" + + root_expression + "/binding.json\n", + "if-no-files-found": "error", + } + elif "scripts/check_coverage.py" in step.get("run", ""): + step["run"] = step["run"].replace( + filename, '"$QSR_EVIDENCE_ROOT/' + filename + '"' + ) + step["env"] = {"QSR_EVIDENCE_ROOT": root_expression} + upload_index = next(i for i, s in enumerate(expected_steps) + if s.get("uses", "").startswith("actions/upload-artifact@")) + expected_steps.insert(upload_index, { + "name": "Admit fresh bound coverage evidence before upload", + "env": {"QSR_EVIDENCE_ROOT": root_expression}, "run": guard_hash, + }) + observed_steps = deepcopy(steps) + for step in observed_steps: + if step.get("name") in ( + generator_name, "Admit fresh bound coverage evidence before upload" + ): + step["run"] = hashlib.sha256(step["run"].encode("utf-8")).hexdigest() + # Compare actual projections, not original objects substituted for actual steps. + assert observed_steps == expected_steps, name + elif name == "podman-e2e-positive-lsm": # Only the exact-test witness and the known cleanup bug differ. assert len(steps) == len(expected["steps"]) assert steps[0:3] == expected["steps"][0:3], name @@ -449,7 +493,8 @@ def test_preserves_all_four_job_mechanics(): ) else: assert steps == expected["steps"], name - assert job.get("env", {}) == expected["env"], name + if name not in SRC03_LANES: + assert job.get("env", {}) == expected["env"], name CANDIDATE = "${{ github.event.pull_request.head.sha || github.sha }}" @@ -827,3 +872,611 @@ def test_doctoring_exposes_capacity_loss_and_unresolved_activation_gates(): "production_gated_command_execution_kills_and_reports_a_command_that_exceeds_its_timeout", ): assert required in text, required + + +SRC03_LANES = { + "coverage": ("Generate production coverage evidence", "coverage.json"), + "branch-coverage": ("Generate branch coverage evidence", "branch-coverage.json"), +} + + +def src03_boundary(tmp_path, lane, mode): + """Replay the real YAML producer-to-upload boundary with a reached witness.""" + source = tmp_path / "source" + tools = tmp_path / "tools" + runner = tmp_path / "runner temp" + for directory in (source, tools, runner): + directory.mkdir() + head = "a" * 40 + root = runner / f"qsr-evidence-17-2-{lane}-{head}" + foreign = tmp_path / "foreign.json" + foreign.write_text('{"fixture":"foreign bytes"}\n', encoding="utf-8") + capture = tmp_path / "producer.json" + producer = tools / "cargo" + producer.write_text( + f"#!{sys.executable}\n" + "import json,os,pathlib,sys\n" + "args=sys.argv[1:]\n" + "if 'report' in args: raise SystemExit(0)\n" + f"expected={(['+nightly-2026-07-01'] if lane == 'branch-coverage' else []) + ['llvm-cov'] + (['--branch'] if lane == 'branch-coverage' else []) + ['--workspace', '--lib', '--tests', '--json', '--output-path']!r}\n" + "if args[:-1]!=expected: raise SystemExit(64)\n" + "out=pathlib.Path(args[args.index('--output-path')+1])\n" + "pathlib.Path(os.environ['CAPTURE']).write_text(json.dumps(args))\n" + "mode=os.environ['MODE']\n" + "if mode=='producer_failure': raise SystemExit(42)\n" + "if mode=='no_output': raise SystemExit(0)\n" + "if mode=='symlink':\n" + " out.symlink_to(os.environ['FOREIGN'])\n" + "elif mode=='directory': out.mkdir()\n" + "elif mode=='fifo': os.mkfifo(out)\n" + "elif mode=='hardlink': os.link(os.environ['FOREIGN'],out)\n" + "else: out.write_text('' if mode=='empty' else '{\"fixture\":\"fresh bytes\"}\\n')\n", + encoding="utf-8", + ) + producer.chmod(0o700) + (tools / "python3").symlink_to(sys.executable) + git = tools / "git" + git.write_text('#!/bin/sh\nprintf "%s\\n" "$ACTUAL_HEAD"\n', encoding="utf-8") + git.chmod(0o700) + context = { + "${{ runner.temp }}": str(runner), + "${{ github.run_id }}": "17", + "${{ github.run_attempt }}": "2", + "${{ github.event.pull_request.head.sha || github.sha }}": head, + } + job = workflow()["jobs"][lane] + + def render(value): + """Render only the real fixture contexts present in the parsed YAML.""" + for expression, replacement in context.items(): + value = value.replace(expression, replacement) + assert "${{" not in value, "unrendered real YAML environment" + return value + + env = { + "PATH": str(tools) + os.pathsep + os.defpath, + "RUNNER_TEMP": str(runner), + "ACTUAL_HEAD": head, + "FOREIGN": str(foreign), + "CAPTURE": str(capture), + "MODE": mode, + } + for key, value in job.get("env", {}).items(): + env[key] = render(value) + name, _ = SRC03_LANES[lane] + steps = job["steps"] + if mode == "stale_root": + root.mkdir() + (root / SRC03_LANES[lane][1]).write_text("stale sentinel", encoding="utf-8") + elif mode == "root_symlink": + root.symlink_to(source, target_is_directory=True) + start = next(i for i, step in enumerate(steps) if step.get("name") == name) + boundary = next( + i + for i, step in enumerate(steps) + if step.get("uses", "").startswith("actions/upload-artifact@") + ) + results = [] + for step in steps[start:boundary]: + # Product coverage policy is unchanged and has its own parser suite. + if "scripts/check_coverage.py" in step.get("run", ""): + continue + result = subprocess.run( + ["bash", "-e", "-o", "pipefail", "-c", step["run"]], + cwd=source, + env={ + **env, + **{key: render(value) for key, value in step.get("env", {}).items()}, + }, + capture_output=True, + text=True, + check=False, + timeout=10, + ) + results.append(result) + if result.returncode: + break + return results, capture, foreign, root + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_legitimate_current_producer_reaches_boundary(tmp_path, lane): + """A genuine nonempty regular output must reach the synthetic upload boundary.""" + results, capture, _, _ = src03_boundary(tmp_path, lane, "regular") + assert capture.is_file(), "producer injection did not reach Cargo" + assert all(result.returncode == 0 for result in results), results[-1].stderr + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_successful_producer_symlink_is_rejected_before_upload(tmp_path, lane): + """Zero-exit generation must not admit a symlink to foreign evidence bytes.""" + results, capture, foreign, _ = src03_boundary(tmp_path, lane, "symlink") + assert capture.is_file(), "producer injection did not reach Cargo" + assert foreign.read_text(encoding="utf-8") == '{"fixture":"foreign bytes"}\n' + assert results[-1].returncode != 0, ( + "SRC03-EVIDENCE-001: successful producer admitted a symlink at upload boundary" + ) + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize( + "mode", + ( + "empty", + "no_output", + "directory", + "fifo", + "hardlink", + "producer_failure", + "stale_root", + "root_symlink", + ), +) +def test_src03_rejects_invalid_outputs_and_preserves_foreign_nodes( + tmp_path, lane, mode +): + """Fresh production cannot adopt old nodes, empty output or failed generation.""" + results, capture, foreign, root = src03_boundary(tmp_path, lane, mode) + assert results[-1].returncode != 0, (lane, mode) + if mode in ("stale_root", "root_symlink"): + assert not capture.exists(), "stale root reached producer" + if mode == "stale_root": + assert (root / SRC03_LANES[lane][1]).read_text() == "stale sentinel" + else: + assert root.is_symlink() + else: + assert capture.is_file(), "injection did not reach producer" + if mode == "producer_failure": + assert results[-1].returncode == 42 + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize( + "mutation", + ( + "run", + "attempt", + "head", + "job", + "hash", + "bytes", + "producer", + "extra", + "duplicate", + "empty", + "symlink", + "hardlink", + "root_symlink", + "boolean_exit", + "head_drift", + ), +) +def test_src03_consumer_revalidates_binding_and_bytes(tmp_path, lane, mutation): + """Execute the actual pre-upload guard against post-producer drift.""" + import json + + results, _, foreign, root = src03_boundary(tmp_path, lane, "regular") + assert all(result.returncode == 0 for result in results) + binding = root / "binding.json" + receipt = json.loads(binding.read_text()) + artifact = root / SRC03_LANES[lane][1] + if mutation in ("run", "attempt", "head", "job"): + key = { + "run": "QSR_RUN_ID", + "attempt": "QSR_RUN_ATTEMPT", + "head": "QSR_EXPECTED_HEAD", + "job": "QSR_EVIDENCE_JOB", + }[mutation] + receipt[key] = "different" + elif mutation == "hash": + artifact.write_text("mutated evidence") + elif mutation == "bytes": + receipt["bytes"] += 1 + elif mutation == "producer": + receipt["producer_exit"] = 42 + elif mutation == "boolean_exit": + receipt["producer_exit"] = False + elif mutation == "extra": + receipt["unknown"] = "field" + elif mutation in ("empty", "symlink", "hardlink"): + artifact.unlink() + if mutation == "empty": + artifact.touch() + elif mutation == "symlink": + artifact.symlink_to(foreign) + else: + os.link(foreign, artifact) + elif mutation == "root_symlink": + saved = root.with_name("saved-root") + root.rename(saved) + root.symlink_to(saved, target_is_directory=True) + binding = saved / "binding.json" + if mutation == "duplicate": + binding.write_text(binding.read_text()[:-1] + ',"producer_exit":0}') + else: + binding.write_text(json.dumps(receipt)) + script = next( + step["run"] + for step in workflow()["jobs"][lane]["steps"] + if step.get("name") == "Admit fresh bound coverage evidence before upload" + ) + env = { + "PATH": str(tmp_path / "tools") + os.pathsep + os.defpath, + "RUNNER_TEMP": str(tmp_path / "runner temp"), + "QSR_EVIDENCE_ROOT": str(root), + "QSR_RUN_ID": "17", + "QSR_RUN_ATTEMPT": "2", + "QSR_EVIDENCE_JOB": lane, + "QSR_EXPECTED_HEAD": "a" * 40, + "ACTUAL_HEAD": "b" * 40 if mutation == "head_drift" else "a" * 40, + } + result = subprocess.run( + ["bash", "-e", "-o", "pipefail", "-c", script], + cwd=tmp_path / "source", + env=env, + capture_output=True, + text=True, + timeout=10, + check=False, + ) + assert result.returncode != 0, (lane, mutation) + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +def test_src03_upload_is_run_bound_and_preceded_by_admission(): + """Preserve failure suppression and connect exact evidence paths to upload.""" + for lane, (name, filename) in SRC03_LANES.items(): + job = workflow()["jobs"][lane] + env = job["env"] + assert env["QSR_RUN_ID"] == "${{ github.run_id }}" + assert env["QSR_RUN_ATTEMPT"] == "${{ github.run_attempt }}" + assert env["QSR_EVIDENCE_JOB"] == lane + assert env["QSR_EXPECTED_HEAD"] == CANDIDATE + root_expression = ( + "${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-${{ github.run_attempt }}-" + + lane + + "-" + + CANDIDATE + ) + assert set(env) == { + "QSR_RUN_ID", + "QSR_RUN_ATTEMPT", + "QSR_EVIDENCE_JOB", + "QSR_EXPECTED_HEAD", + } + steps = job["steps"] + upload = next( + step + for step in steps + if step.get("uses", "").startswith("actions/upload-artifact@") + ) + assert set(upload) == {"name", "uses", "with"} + assert upload["uses"] == ( + "actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02" + ) + assert upload["with"] == { + "name": filename.removesuffix(".json") + + "-json-" + + CANDIDATE + + "-${{ github.run_id }}-${{ github.run_attempt }}", + "path": root_expression + + "/" + + filename + + "\n" + + root_expression + + "/binding.json\n", + "if-no-files-found": "error", + } + assert ( + steps[steps.index(upload) - 1]["name"] + == "Admit fresh bound coverage evidence before upload" + ) + assert set(steps[steps.index(upload) - 1]) == {"name", "run", "env"} + for step in steps: + if ( + step.get("name") == name + or step.get("name") + == "Admit fresh bound coverage evidence before upload" + or "scripts/check_coverage.py" in step.get("run", "") + ): + assert step["env"] == {"QSR_EVIDENCE_ROOT": root_expression} + parser = next( + step for step in steps if "scripts/check_coverage.py" in step.get("run", "") + ) + flag = "--require-branches " if lane == "branch-coverage" else "" + assert ( + parser["run"] + == "python3 scripts/check_coverage.py " + + flag + + '"$QSR_EVIDENCE_ROOT/' + + filename + + '"' + ) + assert steps.index(parser) > steps.index(upload) + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize( + "mutation", + ( + "mutable_upload", "alternate_upload", "generator_echo", "guard_echo", + "guard_removed", "guard_if", "generator_if", "guard_env", "step_order", + ), +) +def test_src03_review_b01_rejects_structural_substitution(monkeypatch, lane, mutation): + """Persistent independent desired rejection; actual YAML is the control.""" + data = deepcopy(workflow()) + steps = data["jobs"][lane]["steps"] + generator = next(s for s in steps if s.get("name") == SRC03_LANES[lane][0]) + guard = next(s for s in steps if s.get("name") == + "Admit fresh bound coverage evidence before upload") + upload = next(s for s in steps if s.get("uses", "").startswith( + "actions/upload-artifact@")) + if mutation == "mutable_upload": + upload["uses"] = "actions/upload-artifact@main" + elif mutation == "alternate_upload": + upload["uses"] = "actions/upload-artifact@" + "b" * 40 + elif mutation == "generator_echo": + generator["run"] = "echo substituted-generator\n" + elif mutation == "guard_echo": + guard["run"] = "echo substituted-guard\n" + elif mutation == "guard_removed": + steps.remove(guard) + elif mutation == "guard_if": + guard["if"] = "false" + elif mutation == "generator_if": + generator["if"] = "false" + elif mutation == "guard_env": + guard["env"]["QSR_EVIDENCE_ROOT"] = "alternate" + else: + index = steps.index(generator) + steps[index - 1], steps[index] = steps[index], steps[index - 1] + monkeypatch.setitem(globals(), "workflow", lambda: data) + with pytest.raises(AssertionError): + test_preserves_all_four_job_mechanics() + + +def test_src03_review_b01_accepts_legitimate_literal_control(): + """The complete retained mechanics must remain accepted.""" + test_preserves_all_four_job_mechanics() + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_review_b02_producer_without_file_digest(tmp_path, monkeypatch, lane): + """Synthetic child capability seam, NOT actual runner/minimum-version evidence.""" + import json + + data = deepcopy(workflow()) + producer = next(s for s in data["jobs"][lane]["steps"] + if s.get("name") == SRC03_LANES[lane][0]) + prefix = "python3 -I - <<'PY'\n" + assert producer["run"].startswith(prefix) + producer["run"] = producer["run"].replace( + prefix, prefix + "import hashlib\n" + "if hasattr(hashlib, 'file_digest'):\n del hashlib.file_digest\n", 1 + ) + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, foreign, root = src03_boundary(tmp_path, lane, "regular") + assert capture.is_file(), "capability seam did not reach actual producer" + assert all(r.returncode == 0 for r in results), results[-1].stderr + payload = (root / SRC03_LANES[lane][1]).read_bytes() + receipt = json.loads((root / "binding.json").read_bytes()) + assert receipt["sha256"] == hashlib.sha256(payload).hexdigest() + assert receipt["bytes"] == len(payload) + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +def src03_review_inject(script, code): + """Add an explicit child-only stdlib seam to real embedded Python.""" + prefix = "python3 -I - <<'PY'\n" + assert script.startswith(prefix) + return script.replace(prefix, prefix + code + "\n", 1) + + +def src03_review_guard(tmp_path, lane, root, script=None): + """Execute actual guard with the closed legitimate fixture context.""" + if script is None: + script = next(s["run"] for s in workflow()["jobs"][lane]["steps"] + if s.get("name") == "Admit fresh bound coverage evidence before upload") + return subprocess.run( + ["bash", "-e", "-o", "pipefail", "-c", script], + cwd=tmp_path / "source", + env={ + "PATH": str(tmp_path / "tools") + os.pathsep + os.defpath, + "RUNNER_TEMP": str(tmp_path / "runner temp"), + "QSR_EVIDENCE_ROOT": str(root), "QSR_RUN_ID": "17", + "QSR_RUN_ATTEMPT": "2", "QSR_EVIDENCE_JOB": lane, + "QSR_EXPECTED_HEAD": "a" * 40, "ACTUAL_HEAD": "a" * 40, + }, + capture_output=True, text=True, timeout=10, check=False, + ) + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_review_r03_manifest_reader_refuses_oversize(tmp_path, lane): + """Valid closed binding plus whitespace cannot evade the byte bound.""" + results, _, foreign, root = src03_boundary(tmp_path, lane, "regular") + assert all(r.returncode == 0 for r in results) + binding = root / "binding.json" + assert 0 < binding.stat().st_size < 4096 + binding.write_bytes(binding.read_bytes() + b" " * 4096) + result = src03_review_guard(tmp_path, lane, root) + assert result.returncode != 0, "oversized manifest reached upload boundary" + assert "manifest too large" in result.stderr + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_review_r03_generator_refuses_oversized_serialization( + tmp_path, monkeypatch, lane +): + """Synthetic child serialization seam; refuse before publication.""" + data = deepcopy(workflow()) + generator = next(s for s in data["jobs"][lane]["steps"] + if s.get("name") == SRC03_LANES[lane][0]) + generator["run"] = src03_review_inject(generator["run"], + "import json\n" + "original_dumps = json.dumps\n" + "def inflated_dumps(value, *args, **kwargs):\n" + " return original_dumps(value, *args, **kwargs) + ' ' * 4096\n" + "def inflated_dump(value, handle, *args, **kwargs):\n" + " handle.write(inflated_dumps(value, *args, **kwargs))\n" + "json.dumps = inflated_dumps\njson.dump = inflated_dump\n") + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, foreign, root = src03_boundary(tmp_path, lane, "regular") + assert capture.is_file(), "serialization seam did not reach producer" + assert results[0].returncode != 0, "producer published oversized binding" + assert "manifest too large" in results[0].stderr + assert not (root / "binding.json").exists() + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize("underreported", (False, True)) +def test_src03_review_r03_manifest_read_call_is_bounded(tmp_path, lane, underreported): + """Actual guard read-call seam, including a misleading size observation.""" + results, _, _, root = src03_boundary(tmp_path, lane, "regular") + assert all(r.returncode == 0 for r in results) + if underreported: + binding = root / "binding.json" + binding.write_bytes(binding.read_bytes() + b" " * 4096) + code = '''import os +original_open, original_fdopen, original_fstat = os.open, os.fdopen, os.fstat +manifest_fds = set() +def probe_open(path, *args, **kwargs): + fd = original_open(path, *args, **kwargs) + if str(path).endswith('/binding.json'): + manifest_fds.add(fd) + return fd +class Probe: + def __init__(self, handle): self.handle = handle + def __enter__(self): return self + def __exit__(self, *args): + manifest_fds.discard(self.fileno()) + return self.handle.__exit__(*args) + def fileno(self): return self.handle.fileno() + def read(self, size=-1): + if self.fileno() in manifest_fds: + assert 0 < size <= 4097, 'unbounded manifest read' + return self.handle.read(size) +def probe_fdopen(*args, **kwargs): return Probe(original_fdopen(*args, **kwargs)) +os.open, os.fdopen = probe_open, probe_fdopen +''' + if underreported: + code += '''def probe_fstat(fd): + info = original_fstat(fd) + if fd in manifest_fds: + fields = list(info) + fields[6] = 1 + return os.stat_result(fields) + return info +os.fstat = probe_fstat +''' + script = next(s["run"] for s in workflow()["jobs"][lane]["steps"] + if s.get("name") == "Admit fresh bound coverage evidence before upload") + result = src03_review_guard(tmp_path, lane, root, src03_review_inject(script, code)) + assert "unbounded manifest read" not in result.stderr, result.stderr + if underreported: + assert result.returncode != 0 + assert "manifest too large" in result.stderr + else: + assert result.returncode == 0, result.stderr + + +def src03_review_stream_seam(action="regular"): + """Large real file and observed bounded reads; controlled drift, not hostile race.""" + return '''import os, subprocess, sys +original_run = subprocess.run +original_open, original_fdopen = os.open, os.fdopen +payload_fds = {} +reads = 0 +changed = False +def large_producer(command, *args, **kwargs): + result = original_run(command, *args, **kwargs) + if '--output-path' in command and result.returncode == 0: + path = command[command.index('--output-path') + 1] + with open(path, 'wb') as output: + for _ in range(33): output.write(b'x' * 65536) + return result +subprocess.run = large_producer +def probe_open(path, *args, **kwargs): + fd = original_open(path, *args, **kwargs) + if str(path).endswith('/coverage.json') or str(path).endswith('/branch-coverage.json'): + payload_fds[fd] = path + return fd +class Probe: + def __init__(self, handle): self.handle = handle + def __enter__(self): return self + def __exit__(self, *args): + payload_fds.pop(self.fileno(), None) + return self.handle.__exit__(*args) + def fileno(self): return self.handle.fileno() + def read(self, size=-1): + global reads, changed + fd = self.fileno() + if fd in payload_fds: + assert 0 < size <= 65536, 'unbounded payload read' + reads += 1 + data = self.handle.read(size) + if fd in payload_fds and not changed and ACTION != 'regular': + changed = True + writer = original_open(payload_fds[fd], os.O_WRONLY) + try: + if ACTION == 'grow': + os.lseek(writer, 0, os.SEEK_END) + os.write(writer, b'x') + else: + os.ftruncate(writer, 1) + finally: + os.close(writer) + return data +os.open, os.fdopen = probe_open, lambda *a, **k: Probe(original_fdopen(*a, **k)) +import atexit +atexit.register(lambda: print('SRC03-STREAM-READS=' + str(reads), file=sys.stderr)) +'''.replace("ACTION", repr(action)) + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_review_r03_streams_large_regular_payload(tmp_path, monkeypatch, lane): + """Both actual bodies hash a 2MiB+ regular file with bounded multiple reads.""" + import json + import re + + data = deepcopy(workflow()) + for step in data["jobs"][lane]["steps"]: + if step.get("name") in (SRC03_LANES[lane][0], + "Admit fresh bound coverage evidence before upload"): + step["run"] = src03_review_inject(step["run"], src03_review_stream_seam()) + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, foreign, root = src03_boundary(tmp_path, lane, "regular") + assert capture.is_file() + assert all(r.returncode == 0 for r in results), results[-1].stderr + observed = [int(re.search(r"SRC03-STREAM-READS=(\d+)", r.stderr)[1]) + for r in results if "SRC03-STREAM-READS=" in r.stderr] + assert len(observed) == 2 and all(n > 32 for n in observed), observed + receipt = json.loads((root / "binding.json").read_bytes()) + assert receipt["bytes"] == 33 * 65536 + assert receipt["sha256"] == hashlib.sha256(b"x" * (33 * 65536)).hexdigest() + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize("role", ("producer", "guard")) +@pytest.mark.parametrize("action", ("grow", "truncate")) +def test_src03_review_r03_rejects_counted_payload_drift( + tmp_path, monkeypatch, lane, role, action +): + """Actual descriptor reads detect controlled size changes before/after hashing.""" + data = deepcopy(workflow()) + name = SRC03_LANES[lane][0] if role == "producer" else ( + "Admit fresh bound coverage evidence before upload") + step = next(s for s in data["jobs"][lane]["steps"] if s.get("name") == name) + step["run"] = src03_review_inject(step["run"], src03_review_stream_seam(action)) + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, foreign, _ = src03_boundary(tmp_path, lane, "regular") + assert capture.is_file() + result = results[0] if role == "producer" else results[-1] + assert result.returncode != 0, "size drift accepted" + assert "evidence changed while reading" in result.stderr, result.stderr + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' From fbc6db1a22da5026dc5d2352d18d3d62d0e7292f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 14 Oct 2026 13:41:38 +0900 Subject: [PATCH 2/2] test(ci): cover all QSR evidence guard branches Add 42 parametrized synthetic controls for identity, checkout drift, counted reads, metadata changes, manifest bounds and lane commands. Independent test-delta review: 37d957cce6df45a134c9cb8179c5c2324d254ee723eb5b83dc9ac4c83e276343. Embedded source coverage is 100% per body; central full-gate and runtime acceptance remain separate. --- ..._quarantine_sandbox_runtime_ci_contract.py | 427 ++++++++++++++++-- 1 file changed, 378 insertions(+), 49 deletions(-) diff --git a/tests/test_quarantine_sandbox_runtime_ci_contract.py b/tests/test_quarantine_sandbox_runtime_ci_contract.py index de5815b790..d4c82ae7bb 100644 --- a/tests/test_quarantine_sandbox_runtime_ci_contract.py +++ b/tests/test_quarantine_sandbox_runtime_ci_contract.py @@ -441,8 +441,12 @@ def test_preserves_all_four_job_mechanics(): test_src03_upload_is_run_bound_and_preceded_by_admission() # Literal reviewed script digests, never derived from workflow() at runtime. # Behavioral producer/admission controls below remain mandatory supplements. - generator_hash = "091ae4e9cb783d06138a6e62be6ad8255b9c5b47637a23da2628b419e6dacbee" - guard_hash = "af57131f5083f77780dc66ed46d2c70679165bef149732c1309e9745132d7d82" + generator_hash = ( + "091ae4e9cb783d06138a6e62be6ad8255b9c5b47637a23da2628b419e6dacbee" + ) + guard_hash = ( + "af57131f5083f77780dc66ed46d2c70679165bef149732c1309e9745132d7d82" + ) generator_name, filename = SRC03_LANES[name] root_expression = ( "${{ runner.temp }}/qsr-evidence-${{ github.run_id }}-" @@ -453,12 +457,20 @@ def test_preserves_all_four_job_mechanics(): if step.get("name") == generator_name: step["run"] = generator_hash step["env"] = {"QSR_EVIDENCE_ROOT": root_expression} - elif "uses" in step and step["uses"].startswith("actions/upload-artifact@"): + elif "uses" in step and step["uses"].startswith( + "actions/upload-artifact@" + ): step["with"] = { - "name": filename.removesuffix(".json") + "-json-" + CANDIDATE + "name": filename.removesuffix(".json") + + "-json-" + + CANDIDATE + "-${{ github.run_id }}-${{ github.run_attempt }}", - "path": root_expression + "/" + filename + "\n" - + root_expression + "/binding.json\n", + "path": root_expression + + "/" + + filename + + "\n" + + root_expression + + "/binding.json\n", "if-no-files-found": "error", } elif "scripts/check_coverage.py" in step.get("run", ""): @@ -466,18 +478,28 @@ def test_preserves_all_four_job_mechanics(): filename, '"$QSR_EVIDENCE_ROOT/' + filename + '"' ) step["env"] = {"QSR_EVIDENCE_ROOT": root_expression} - upload_index = next(i for i, s in enumerate(expected_steps) - if s.get("uses", "").startswith("actions/upload-artifact@")) - expected_steps.insert(upload_index, { - "name": "Admit fresh bound coverage evidence before upload", - "env": {"QSR_EVIDENCE_ROOT": root_expression}, "run": guard_hash, - }) + upload_index = next( + i + for i, s in enumerate(expected_steps) + if s.get("uses", "").startswith("actions/upload-artifact@") + ) + expected_steps.insert( + upload_index, + { + "name": "Admit fresh bound coverage evidence before upload", + "env": {"QSR_EVIDENCE_ROOT": root_expression}, + "run": guard_hash, + }, + ) observed_steps = deepcopy(steps) for step in observed_steps: if step.get("name") in ( - generator_name, "Admit fresh bound coverage evidence before upload" + generator_name, + "Admit fresh bound coverage evidence before upload", ): - step["run"] = hashlib.sha256(step["run"].encode("utf-8")).hexdigest() + step["run"] = hashlib.sha256( + step["run"].encode("utf-8") + ).hexdigest() # Compare actual projections, not original objects substituted for actual steps. assert observed_steps == expected_steps, name elif name == "podman-e2e-positive-lsm": @@ -880,7 +902,7 @@ def test_doctoring_exposes_capacity_loss_and_unresolved_activation_gates(): } -def src03_boundary(tmp_path, lane, mode): +def src03_boundary(tmp_path, lane, mode, *, overrides=None): """Replay the real YAML producer-to-upload boundary with a reached witness.""" source = tmp_path / "source" tools = tmp_path / "tools" @@ -967,6 +989,7 @@ def render(value): env={ **env, **{key: render(value) for key, value in step.get("env", {}).items()}, + **(overrides or {}), }, capture_output=True, text=True, @@ -1200,8 +1223,15 @@ def test_src03_upload_is_run_bound_and_preceded_by_admission(): @pytest.mark.parametrize( "mutation", ( - "mutable_upload", "alternate_upload", "generator_echo", "guard_echo", - "guard_removed", "guard_if", "generator_if", "guard_env", "step_order", + "mutable_upload", + "alternate_upload", + "generator_echo", + "guard_echo", + "guard_removed", + "guard_if", + "generator_if", + "guard_env", + "step_order", ), ) def test_src03_review_b01_rejects_structural_substitution(monkeypatch, lane, mutation): @@ -1209,10 +1239,14 @@ def test_src03_review_b01_rejects_structural_substitution(monkeypatch, lane, mut data = deepcopy(workflow()) steps = data["jobs"][lane]["steps"] generator = next(s for s in steps if s.get("name") == SRC03_LANES[lane][0]) - guard = next(s for s in steps if s.get("name") == - "Admit fresh bound coverage evidence before upload") - upload = next(s for s in steps if s.get("uses", "").startswith( - "actions/upload-artifact@")) + guard = next( + s + for s in steps + if s.get("name") == "Admit fresh bound coverage evidence before upload" + ) + upload = next( + s for s in steps if s.get("uses", "").startswith("actions/upload-artifact@") + ) if mutation == "mutable_upload": upload["uses"] = "actions/upload-artifact@main" elif mutation == "alternate_upload": @@ -1248,13 +1282,16 @@ def test_src03_review_b02_producer_without_file_digest(tmp_path, monkeypatch, la import json data = deepcopy(workflow()) - producer = next(s for s in data["jobs"][lane]["steps"] - if s.get("name") == SRC03_LANES[lane][0]) + producer = next( + s for s in data["jobs"][lane]["steps"] if s.get("name") == SRC03_LANES[lane][0] + ) prefix = "python3 -I - <<'PY'\n" assert producer["run"].startswith(prefix) producer["run"] = producer["run"].replace( - prefix, prefix + "import hashlib\n" - "if hasattr(hashlib, 'file_digest'):\n del hashlib.file_digest\n", 1 + prefix, + prefix + "import hashlib\n" + "if hasattr(hashlib, 'file_digest'):\n del hashlib.file_digest\n", + 1, ) monkeypatch.setitem(globals(), "workflow", lambda: data) results, capture, foreign, root = src03_boundary(tmp_path, lane, "regular") @@ -1277,19 +1314,28 @@ def src03_review_inject(script, code): def src03_review_guard(tmp_path, lane, root, script=None): """Execute actual guard with the closed legitimate fixture context.""" if script is None: - script = next(s["run"] for s in workflow()["jobs"][lane]["steps"] - if s.get("name") == "Admit fresh bound coverage evidence before upload") + script = next( + s["run"] + for s in workflow()["jobs"][lane]["steps"] + if s.get("name") == "Admit fresh bound coverage evidence before upload" + ) return subprocess.run( ["bash", "-e", "-o", "pipefail", "-c", script], cwd=tmp_path / "source", env={ "PATH": str(tmp_path / "tools") + os.pathsep + os.defpath, "RUNNER_TEMP": str(tmp_path / "runner temp"), - "QSR_EVIDENCE_ROOT": str(root), "QSR_RUN_ID": "17", - "QSR_RUN_ATTEMPT": "2", "QSR_EVIDENCE_JOB": lane, - "QSR_EXPECTED_HEAD": "a" * 40, "ACTUAL_HEAD": "a" * 40, + "QSR_EVIDENCE_ROOT": str(root), + "QSR_RUN_ID": "17", + "QSR_RUN_ATTEMPT": "2", + "QSR_EVIDENCE_JOB": lane, + "QSR_EXPECTED_HEAD": "a" * 40, + "ACTUAL_HEAD": "a" * 40, }, - capture_output=True, text=True, timeout=10, check=False, + capture_output=True, + text=True, + timeout=10, + check=False, ) @@ -1313,16 +1359,19 @@ def test_src03_review_r03_generator_refuses_oversized_serialization( ): """Synthetic child serialization seam; refuse before publication.""" data = deepcopy(workflow()) - generator = next(s for s in data["jobs"][lane]["steps"] - if s.get("name") == SRC03_LANES[lane][0]) - generator["run"] = src03_review_inject(generator["run"], + generator = next( + s for s in data["jobs"][lane]["steps"] if s.get("name") == SRC03_LANES[lane][0] + ) + generator["run"] = src03_review_inject( + generator["run"], "import json\n" "original_dumps = json.dumps\n" "def inflated_dumps(value, *args, **kwargs):\n" " return original_dumps(value, *args, **kwargs) + ' ' * 4096\n" "def inflated_dump(value, handle, *args, **kwargs):\n" " handle.write(inflated_dumps(value, *args, **kwargs))\n" - "json.dumps = inflated_dumps\njson.dump = inflated_dump\n") + "json.dumps = inflated_dumps\njson.dump = inflated_dump\n", + ) monkeypatch.setitem(globals(), "workflow", lambda: data) results, capture, foreign, root = src03_boundary(tmp_path, lane, "regular") assert capture.is_file(), "serialization seam did not reach producer" @@ -1341,7 +1390,7 @@ def test_src03_review_r03_manifest_read_call_is_bounded(tmp_path, lane, underrep if underreported: binding = root / "binding.json" binding.write_bytes(binding.read_bytes() + b" " * 4096) - code = '''import os + code = """import os original_open, original_fdopen, original_fstat = os.open, os.fdopen, os.fstat manifest_fds = set() def probe_open(path, *args, **kwargs): @@ -1362,9 +1411,9 @@ def read(self, size=-1): return self.handle.read(size) def probe_fdopen(*args, **kwargs): return Probe(original_fdopen(*args, **kwargs)) os.open, os.fdopen = probe_open, probe_fdopen -''' +""" if underreported: - code += '''def probe_fstat(fd): + code += """def probe_fstat(fd): info = original_fstat(fd) if fd in manifest_fds: fields = list(info) @@ -1372,9 +1421,12 @@ def probe_fdopen(*args, **kwargs): return Probe(original_fdopen(*args, **kwargs) return os.stat_result(fields) return info os.fstat = probe_fstat -''' - script = next(s["run"] for s in workflow()["jobs"][lane]["steps"] - if s.get("name") == "Admit fresh bound coverage evidence before upload") +""" + script = next( + s["run"] + for s in workflow()["jobs"][lane]["steps"] + if s.get("name") == "Admit fresh bound coverage evidence before upload" + ) result = src03_review_guard(tmp_path, lane, root, src03_review_inject(script, code)) assert "unbounded manifest read" not in result.stderr, result.stderr if underreported: @@ -1386,7 +1438,7 @@ def probe_fdopen(*args, **kwargs): return Probe(original_fdopen(*args, **kwargs) def src03_review_stream_seam(action="regular"): """Large real file and observed bounded reads; controlled drift, not hostile race.""" - return '''import os, subprocess, sys + return """import os, subprocess, sys original_run = subprocess.run original_open, original_fdopen = os.open, os.fdopen payload_fds = {} @@ -1434,7 +1486,7 @@ def read(self, size=-1): os.open, os.fdopen = probe_open, lambda *a, **k: Probe(original_fdopen(*a, **k)) import atexit atexit.register(lambda: print('SRC03-STREAM-READS=' + str(reads), file=sys.stderr)) -'''.replace("ACTION", repr(action)) +""".replace("ACTION", repr(action)) @pytest.mark.parametrize("lane", SRC03_LANES) @@ -1445,15 +1497,20 @@ def test_src03_review_r03_streams_large_regular_payload(tmp_path, monkeypatch, l data = deepcopy(workflow()) for step in data["jobs"][lane]["steps"]: - if step.get("name") in (SRC03_LANES[lane][0], - "Admit fresh bound coverage evidence before upload"): + if step.get("name") in ( + SRC03_LANES[lane][0], + "Admit fresh bound coverage evidence before upload", + ): step["run"] = src03_review_inject(step["run"], src03_review_stream_seam()) monkeypatch.setitem(globals(), "workflow", lambda: data) results, capture, foreign, root = src03_boundary(tmp_path, lane, "regular") assert capture.is_file() assert all(r.returncode == 0 for r in results), results[-1].stderr - observed = [int(re.search(r"SRC03-STREAM-READS=(\d+)", r.stderr)[1]) - for r in results if "SRC03-STREAM-READS=" in r.stderr] + observed = [ + int(re.search(r"SRC03-STREAM-READS=(\d+)", r.stderr)[1]) + for r in results + if "SRC03-STREAM-READS=" in r.stderr + ] assert len(observed) == 2 and all(n > 32 for n in observed), observed receipt = json.loads((root / "binding.json").read_bytes()) assert receipt["bytes"] == 33 * 65536 @@ -1469,8 +1526,11 @@ def test_src03_review_r03_rejects_counted_payload_drift( ): """Actual descriptor reads detect controlled size changes before/after hashing.""" data = deepcopy(workflow()) - name = SRC03_LANES[lane][0] if role == "producer" else ( - "Admit fresh bound coverage evidence before upload") + name = ( + SRC03_LANES[lane][0] + if role == "producer" + else ("Admit fresh bound coverage evidence before upload") + ) step = next(s for s in data["jobs"][lane]["steps"] if s.get("name") == name) step["run"] = src03_review_inject(step["run"], src03_review_stream_seam(action)) monkeypatch.setitem(globals(), "workflow", lambda: data) @@ -1480,3 +1540,272 @@ def test_src03_review_r03_rejects_counted_payload_drift( assert result.returncode != 0, "size drift accepted" assert "evidence changed while reading" in result.stderr, result.stderr assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize( + ("key", "value", "diagnostic"), + ( + ("QSR_RUN_ID", "0", "invalid run identity"), + ("QSR_RUN_ATTEMPT", "02", "invalid run identity"), + ("QSR_EXPECTED_HEAD", "A" * 40, "invalid head identity"), + ("QSR_EVIDENCE_JOB", "foreign-lane", "invalid evidence lane"), + ("QSR_EVIDENCE_ROOT", "foreign-root", "invalid evidence root"), + ("RUNNER_TEMP", "source-file", "invalid evidence root"), + ), +) +def test_src03_coverage_controls_invalid_producer_identity( + tmp_path, lane, key, value, diagnostic +): + """Reject each malformed identity at its own diagnostic before generation.""" + if key == "QSR_EVIDENCE_ROOT": + value = str(tmp_path / value) + if key == "RUNNER_TEMP": + parent = tmp_path / value + parent.write_text("synthetic non-directory parent", encoding="utf-8") + value = str(parent) + results, capture, foreign, _ = src03_boundary( + tmp_path, lane, "regular", overrides={key: value} + ) + assert results[0].returncode == 1 + assert "SRC03-EVIDENCE-001: " + diagnostic in results[0].stderr + assert not capture.exists(), "invalid identity reached Cargo" + assert foreign.read_text() == '{"fixture":"foreign bytes"}\n' + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize("phase", ("before", "after")) +def test_src03_coverage_controls_producer_checkout_drift( + tmp_path, monkeypatch, lane, phase +): + """Both checkout observations must reject drift, not just the upload guard.""" + data = deepcopy(workflow()) + generator = next( + step + for step in data["jobs"][lane]["steps"] + if step.get("name") == SRC03_LANES[lane][0] + ) + code = """import subprocess +original_check_output = subprocess.check_output +head_reads = 0 +def changed_head(command, *args, **kwargs): + global head_reads + actual = original_check_output(command, *args, **kwargs) + if command == ['git', 'rev-parse', 'HEAD']: + head_reads += 1 + if head_reads == DRIFT_READ: + return 'b' * 40 + '\\n' + return actual +subprocess.check_output = changed_head +""".replace("DRIFT_READ", "1" if phase == "before" else "2") + generator["run"] = src03_review_inject(generator["run"], code) + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, _, root = src03_boundary(tmp_path, lane, "regular") + assert results[0].returncode == 1 + assert "SRC03-EVIDENCE-001: checkout drift" in results[0].stderr + assert capture.exists() == (phase == "after") + assert not (root / "binding.json").exists() + + +def src03_coverage_controls_stat_seam(target, observation): + """Change only an observed stat field, preserving real descriptor reads.""" + return """import os +original_open, original_fstat = os.open, os.fstat +observed_fds = set() +stat_reads = 0 +def observed_open(path, *args, **kwargs): + fd = original_open(path, *args, **kwargs) + if str(path).endswith(TARGET): + observed_fds.add(fd) + return fd +class StatView: + def __init__(self, info): self.info = info + def __getattr__(self, key): + if key == 'st_size': return self.info.st_size + 1 + return getattr(self.info, key) +def observed_stat(fd): + global stat_reads + info = original_fstat(fd) + if fd in observed_fds: + stat_reads += 1 + if stat_reads == OBSERVATION: + return StatView(info) + return info +os.open, os.fstat = observed_open, observed_stat +""".replace("TARGET", repr("/" + target)).replace("OBSERVATION", str(observation)) + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize("role", ("producer", "guard")) +@pytest.mark.parametrize("observation", (1, 2)) +def test_src03_coverage_controls_payload_stat_count_mismatch( + tmp_path, monkeypatch, lane, role, observation +): + """Reject both short counted reads and changed post-read metadata.""" + data = deepcopy(workflow()) + name = ( + SRC03_LANES[lane][0] + if role == "producer" + else ("Admit fresh bound coverage evidence before upload") + ) + step = next(s for s in data["jobs"][lane]["steps"] if s.get("name") == name) + step["run"] = src03_review_inject( + step["run"], + src03_coverage_controls_stat_seam(SRC03_LANES[lane][1], observation), + ) + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, _, root = src03_boundary(tmp_path, lane, "regular") + assert capture.is_file(), "stat seam did not reach producer" + result = results[0] if role == "producer" else results[-1] + assert result.returncode == 1 + assert "SRC03-EVIDENCE-001: evidence changed while reading" in result.stderr + if role == "producer": + assert not (root / "binding.json").exists() + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize("observation", (1, 2)) +def test_src03_coverage_controls_manifest_stat_count_mismatch( + tmp_path, lane, observation +): + """A bounded valid manifest still needs counted-byte and stable-stat checks.""" + results, _, _, root = src03_boundary(tmp_path, lane, "regular") + assert all(result.returncode == 0 for result in results) + script = next( + step["run"] + for step in workflow()["jobs"][lane]["steps"] + if step.get("name") == "Admit fresh bound coverage evidence before upload" + ) + result = src03_review_guard( + tmp_path, + lane, + root, + src03_review_inject( + script, src03_coverage_controls_stat_seam("binding.json", observation) + ), + ) + assert result.returncode == 1 + assert "SRC03-EVIDENCE-001: evidence changed while reading" in result.stderr + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize("actual_lane", SRC03_LANES) +def test_src03_coverage_controls_producer_command_branches( + tmp_path, monkeypatch, lane, actual_lane +): + """Each byte-identical body supports both lanes and exact command argv.""" + import json + + data = deepcopy(workflow()) + source_script = next( + s["run"] + for s in data["jobs"][lane]["steps"] + if s.get("name") == SRC03_LANES[lane][0] + ) + step = next( + s + for s in data["jobs"][actual_lane]["steps"] + if s.get("name") == SRC03_LANES[actual_lane][0] + ) + step["run"] = source_script + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, _, root = src03_boundary(tmp_path, actual_lane, "regular") + assert all(r.returncode == 0 for r in results), results[-1].stderr + argv = json.loads(capture.read_text()) + expected = ["+nightly-2026-07-01"] if actual_lane == "branch-coverage" else [] + expected += ["llvm-cov"] + expected += ["--branch"] if actual_lane == "branch-coverage" else [] + expected += ["--workspace", "--lib", "--tests", "--json", "--output-path"] + expected += [str(root / SRC03_LANES[actual_lane][1])] + assert argv == expected + assert (root / "binding.json").is_file() + + +@pytest.mark.parametrize("lane", SRC03_LANES) +@pytest.mark.parametrize("role", ("producer", "guard")) +def test_src03_coverage_controls_growth_stops_before_second_stat( + tmp_path, monkeypatch, lane, role +): + """Count overflow must reject immediately, not fall through to stat checking.""" + data = deepcopy(workflow()) + name = ( + SRC03_LANES[lane][0] + if role == "producer" + else ("Admit fresh bound coverage evidence before upload") + ) + step = next(s for s in data["jobs"][lane]["steps"] if s.get("name") == name) + code = ( + src03_review_stream_seam("grow") + + """ +original_fstat = os.fstat +payload_stats = 0 +def counted_stat(fd): + global payload_stats + if fd in payload_fds: payload_stats += 1 + return original_fstat(fd) +os.fstat = counted_stat +atexit.register(lambda: print('SRC03-PAYLOAD-STATS=' + str(payload_stats), file=sys.stderr)) +""" + ) + step["run"] = src03_review_inject(step["run"], code) + monkeypatch.setitem(globals(), "workflow", lambda: data) + results, capture, _, _ = src03_boundary(tmp_path, lane, "regular") + assert capture.is_file() + result = results[0] if role == "producer" else results[-1] + assert result.returncode == 1 + assert "SRC03-EVIDENCE-001: evidence changed while reading" in result.stderr + assert "SRC03-PAYLOAD-STATS=1\n" in result.stderr + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_coverage_controls_oversized_binding_not_published( + tmp_path, monkeypatch, lane +): + """Retain the existing serialization fault and its no-publication oracle.""" + test_src03_review_r03_generator_refuses_oversized_serialization( + tmp_path, monkeypatch, lane + ) + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_coverage_controls_manifest_bytes_exceed_observed_stat(tmp_path, lane): + """Actual oversized bytes must fail even when descriptor size is misleading.""" + test_src03_review_r03_manifest_read_call_is_bounded(tmp_path, lane, True) + + +@pytest.mark.parametrize("lane", SRC03_LANES) +def test_src03_coverage_controls_manifest_changes_during_read(tmp_path, lane): + """Real metadata change on the open manifest must reject before binding use.""" + results, _, _, root = src03_boundary(tmp_path, lane, "regular") + assert all(result.returncode == 0 for result in results) + code = """import os +original_open, original_fdopen = os.open, os.fdopen +manifest_fds = set() +def observed_open(path, *args, **kwargs): + fd = original_open(path, *args, **kwargs) + if str(path).endswith('/binding.json'): manifest_fds.add(fd) + return fd +class ChangedManifest: + def __init__(self, handle): self.handle = handle + def __enter__(self): return self + def __exit__(self, *args): + manifest_fds.discard(self.fileno()) + return self.handle.__exit__(*args) + def fileno(self): return self.handle.fileno() + def read(self, size=-1): + data = self.handle.read(size) + if self.fileno() in manifest_fds: + info = os.fstat(self.fileno()) + os.utime(self.fileno(), ns=(info.st_atime_ns, info.st_mtime_ns + 1000000000)) + return data +os.open = observed_open +os.fdopen = lambda *a, **k: ChangedManifest(original_fdopen(*a, **k)) +""" + script = next( + step["run"] + for step in workflow()["jobs"][lane]["steps"] + if step.get("name") == "Admit fresh bound coverage evidence before upload" + ) + result = src03_review_guard(tmp_path, lane, root, src03_review_inject(script, code)) + assert result.returncode == 1 + assert "SRC03-EVIDENCE-001: evidence changed while reading" in result.stderr