From 22b4bd01204d6a65cb39b63406997d82b1f894e0 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 22 Oct 2026 04:38:09 +0000 Subject: [PATCH] Let a PR beat earn CLEAN from a run the gateway signed for A PR session is no-exec, so almost every beat opened UNVERIFIED and the reviewer still did the verifying. The execution gateway already runs one job on a quarantine of the frozen head and signs the result; a proof now reuses it for a walk claim. The walk picks arguments after the trusted profile's fixed command and predicts the exit code. The broker already signs only a matching exit, so a receipt is a run whose result would have differed were the claim false. proofctl.py prove reserves the request, runs the host adapter, verifies the six-file handoff like an implementation consume, and keeps the request, envelopes and streams under proof-evidence/. The store adds proof_kind attested. It earns CLEAN only beside a verified proof run for that beat, in a session with a frozen PR target, and the store writes the signed argv, exit code and output tail onto the beat as attestation, which the walk may not set. The page shows them as what ran. An accepted report finding keeps the proof it was included on. The adapter accepts the wider request shape only when its configuration sets proofArguments: at most 16 bounded arguments after the profile argv, any exit code, everything else still the profile's. Without it, nothing changes. Schema version 8 adds the proof_attempts table. A live run against the real sandbox showed that an import failure exits 1 like a failing assertion, so the skill now says to cite a proof only when its signed output shows the claim. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01NoDcxo3UV95RM7dMTdjMMS --- README.md | 18 + docs/host-execution-protocol.md | 15 + gateway/README.md | 51 +- gateway/adapter.py | 65 +- skills/underwrite/SKILL.md | 120 ++- skills/underwrite/assets/report.css | 44 ++ .../scripts/attested_implementation.py | 50 +- skills/underwrite/scripts/attested_proof.py | 342 +++++++++ skills/underwrite/scripts/proofctl.py | 116 +++ skills/underwrite/scripts/render-report.py | 50 +- skills/underwrite/scripts/session_store.py | 341 ++++++++- tests/test_attested_proof.py | 710 ++++++++++++++++++ tests/test_gateway_adapter.py | 63 ++ tests/test_gateway_live.py | 210 ++++++ tests/test_session_store.py | 22 +- tests/test_skill.py | 20 +- 16 files changed, 2172 insertions(+), 65 deletions(-) create mode 100755 skills/underwrite/scripts/attested_proof.py create mode 100755 skills/underwrite/scripts/proofctl.py create mode 100644 tests/test_attested_proof.py diff --git a/README.md b/README.md index 299fba9..08304b0 100644 --- a/README.md +++ b/README.md @@ -77,6 +77,15 @@ was: `CLEAN` is reserved for a command whose output would have differed were the false, and a claim that was read or inferred opens `UNVERIFIED`, which is where your own reading goes. A risk appears only on a flag, with a decision beside it. +A PR session can still earn `CLEAN` when the host runs a proof gateway. The walk picks a +command, such as one test module, and predicts its exit code. `proofctl.py prove` hands +it to the execution gateway, which runs it on the gateway's own copy of the frozen head +inside its sandbox and signs a receipt only if the command exits as predicted. The +store accepts `proof_kind: attested` only beside a verified receipt for that beat, and +writes the signed command, exit code and the last lines it printed onto the card itself, +so the walk's `PROOF` line cannot pass for the run. A flag can carry its failing run the +same way. A wrong prediction yields no receipt and proves nothing. + Beats stay that size because the store makes them. A slot over 25 words, a claim over 20, a fourth filled slot on a clean beat, or an eleventh quoted line is refused at write time, with the offending slot and its count named, as is a clean verdict on a read proof or a @@ -139,6 +148,13 @@ static readers, but it does not check out the head, install, build, test, lint, scripts or interpreters, or invoke Git hooks and filters. Review comments, report inclusions, notes, navigation, and recorded decisions still work. +A proof never changes the session's `no_exec` policy. This machine never runs the head; +the gateway does, in a sandbox with no network or credentials, from its own quarantine of +the frozen bundle. It happens only on a host that opted in twice: a proof configuration +for the controller, and `proofArguments` in the gateway's own configuration. Even then +the gateway runs the profile's fixed command with at most 16 bounded arguments after it. +The receipt is evidence for one beat and authorizes nothing. + The host execution protocol is documented in [`docs/host-execution-protocol.md`](docs/host-execution-protocol.md). Its [`execution_receipt.py`](skills/underwrite/scripts/execution_receipt.py) module checks a @@ -274,6 +290,8 @@ implementations// one linked child, nested below the source session output.bundle verified output transport stdout complete captured standard output stderr complete captured standard error +proof-evidence// one verified attested proof: request, both signed + envelopes, and both complete streams ``` The scripts are Python 3 stdlib, with no dependencies. `serve.py` runs the walk and diff --git a/docs/host-execution-protocol.md b/docs/host-execution-protocol.md index 2e57773..ebaaaa6 100644 --- a/docs/host-execution-protocol.md +++ b/docs/host-execution-protocol.md @@ -273,6 +273,11 @@ execution attempt to the authoritative queued action and finding beat. A retry u attempt and a new challenge unless it is a byte-identical replay of an already recorded result. +An attested proof uses the same coordinates differently. `action.seq` is the proof's +number within the source session, `action.beat` is the beat the proof may back, and +`action.attempt` is `1`. A proof never retries under the same number: after a definite +failure the controller reserves a new proof with a new number and challenge. + ### Job `job.argv` is a nonempty array of UTF-8 strings without NUL characters. `argv[0]` is @@ -778,6 +783,16 @@ The reference implementation supplies that supervised path only through a separa `gateway_attested` child. Protocol conformance alone remains insufficient, and no receipt authorizes a review publication or push. +An attested proof is the one use a source session makes of a receipt, and it applies +nothing. The request is the trusted profile's job with at most 16 bounded arguments +appended and an exit code the controller predicts; a gateway accepts that shape only when +its own configuration sets `proofArguments`. The controller verifies the receipt exactly +as it verifies an implementation receipt, with the reserved request's job and exit code as +the expected values, verifies the returned output bundle against the signed output tree, +and then discards it. What it records is that one command exited as predicted on the +frozen head and printed the signed streams. The source session stays `no_exec`, its +audience and deliveries are untouched, and the receipt authorizes no effect. + ## Limitations - Current Codex and Claude Code plugin, skill, and hook interfaces do not issue this diff --git a/gateway/README.md b/gateway/README.md index 43490f0..acf5c9a 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -243,11 +243,12 @@ exits. The configuration is version 1 with exactly these fields: The embedded `profile` is the exact separately provisioned profile later given to `implementationctl.py`. Its key ID must match the configured key pair, and its executor ID -must match the policy derived from this deployment. The request's complete `job`, `sandbox`, -and `exitCode` must match it. The `docker`, `openssl`, and `git` values are explicit absolute -host executable paths. They must be trusted regular executables owned by root or the gateway -account and not writable by group or other. The job executable is instead an absolute path -inside the pinned runtime image. +must match the policy derived from this deployment. The request's complete `job`, +`sandbox`, and `exitCode` must match it, except as **Proof deployments** below allows. The +`docker`, `openssl`, and `git` values are explicit absolute host executable paths. They +must be trusted regular executables owned by root or the gateway account and not writable +by group or other. The job executable is instead an absolute path inside the pinned runtime +image. Preprovision `storeRoot` as a gateway-owned `0700` directory. Separately, provision the immediate outbox parent as gateway-owned, group-owned by `consumerGid`, and exactly `0710`. @@ -330,3 +331,43 @@ local branch. After `implementationctl.py consume` durably succeeds, the gateway supervisor may delete the final transport directory. The consumer cannot and must not delete it through the `0710` parent. Neither the adapter nor the gateway checks out the branch, pushes it, opens a pull request, or publishes a review. + +### Proof deployments + +A deployment may also run walk claims for source review sessions, through +`proofctl.py`. It opts in with one more configuration field, `"proofArguments": N`, an +integer from 1 to 16. With it, a request may differ from the profile in exactly two ways: +its `job.argv` may append 1 to `N` arguments after the profile's own argv, each non-empty +printable text of at most 512 UTF-8 bytes, and its `exitCode` may be any integer from 0 to +255. The executable, environment, working directory, closed stdin and sandbox stay the +profile's. A request equal to the profile's job is still held to the profile's exit +code, so the same deployment keeps serving implementation children unchanged. Without +the field, every request must equal the profile exactly, as before. + +The predicted exit code is what makes a run a proof. The broker already signs a receipt +only when the target exits with the request's `exitCode`, so a mismatch ends the attempt +as `failed` with no receipt, and the controller records that nothing was proven. Choose a +profile whose fixed argv names what a proof may run, such as +`["/usr/local/bin/python3", "-I", "-B", "-m", "unittest"]`. A proof runs head code in the +sandbox by design: that is what the network denial, the seccomp filter, the single +process, the `noexec` read-only layout, and the resource limits are for. Do not enable +`proofArguments` on a host whose sandbox you would not trust with the head. + +The controller reaches the adapter through its own proof configuration, kept outside every +repository and session: + +```json +{ + "version": 1, + "profile": "/etc/underwrite/proof-profile.json", + "publicKey": "/etc/underwrite/signing-public.pem", + "adapter": ["/usr/local/bin/underwrite-proof-adapter"], + "evidenceRoot": "/srv/underwrite-handoff/outbox" +} +``` + +`proofctl.py prove` appends `--request`, `--source-bundle` and `--evidence-dir` to +`adapter` and reads the handled JSON result. When the gateway runs as another account, as +it should, `adapter` names a supervisor wrapper that stages the request and the session's +`pr.bundle` where the gateway may read them and then runs the adapter. The evidence +directory is `/underwrite--proof-`. diff --git a/gateway/adapter.py b/gateway/adapter.py index 60782b5..4e38001 100644 --- a/gateway/adapter.py +++ b/gateway/adapter.py @@ -63,6 +63,12 @@ class AdapterError(RuntimeError): "maxSourceBundleBytes", "profile", } +# Opt-in: a proof deployment lets a request append at most this many arguments to the +# profile's command and predict its own exit code. Without it, every request must be +# the profile's job exactly, as an implementation deployment requires. +_PROOF_FIELD = "proofArguments" +_MAX_PROOF_ARGUMENTS = 16 +_MAX_PROOF_ARGUMENT_BYTES = 512 _EVIDENCE_FILES = ( "request.json", "capability.dsse.json", @@ -423,9 +429,22 @@ def load_configuration(path): ), "gateway configuration", ) - _exact(value, _CONFIG_FIELDS, "gateway configuration") + _exact( + value, + _CONFIG_FIELDS | ({_PROOF_FIELD} if _PROOF_FIELD in value else set()), + "gateway configuration", + ) if type(value["version"]) is not int or value["version"] != 1: raise AdapterError("gateway configuration version must be 1") + if _PROOF_FIELD in value: + limit = value[_PROOF_FIELD] + if ( + type(limit) is not int + or not 1 <= limit <= _MAX_PROOF_ARGUMENTS + ): + raise AdapterError( + f"{_PROOF_FIELD} must be an integer from 1 to {_MAX_PROOF_ARGUMENTS}" + ) value["consumerGid"] = _consumer_gid(value["consumerGid"]) try: value["profile"] = validate_execution_profile(value["profile"]) @@ -812,12 +831,50 @@ def publish_evidence(stored, content, destination, consumer_gid=None): os.close(parent) -def _request(path, policy, profile): +def _proof_request(request, profile, limit): + """A proof runs the profile's command with a bounded tail and its predicted exit. + + Everything else stays the profile's: executable, environment, working directory, + stdin and sandbox. The broker signs only when the run exits as predicted.""" + job = request.get("job") + if not isinstance(job, dict): + raise AdapterError("execution request job does not match the trusted profile") + for field in ("cwd", "environment", "executable", "stdin"): + if job.get(field) != profile["job"][field]: + raise AdapterError( + f"execution request job {field} does not match the trusted profile" + ) + if request.get("sandbox") != profile["sandbox"]: + raise AdapterError("execution request sandbox does not match the trusted profile") + prefix = profile["job"]["argv"] + argv = job.get("argv") + if not isinstance(argv, list) or argv[: len(prefix)] != prefix: + raise AdapterError("proof command does not extend the trusted profile command") + tail = argv[len(prefix):] + if not 1 <= len(tail) <= limit: + raise AdapterError(f"proof appends 1 to {limit} arguments to the trusted command") + for argument in tail: + if ( + not isinstance(argument, str) + or not argument + or any(ord(character) < 32 or ord(character) == 127 for character in argument) + or len(argument.encode("utf-8")) > _MAX_PROOF_ARGUMENT_BYTES + ): + raise AdapterError("proof argument is not bounded printable text") + exit_code = request.get("exitCode") + if type(exit_code) is not int or not 0 <= exit_code <= 255: + raise AdapterError("proof exit code must be an integer from 0 to 255") + + +def _request(path, policy, profile, proof_arguments=None): request = _load_json( _read_regular_file(path, "execution request", _MAX_REQUEST_BYTES), "execution request", ) ExecutionRequest(request, policy) + if proof_arguments is not None and request.get("job") != profile["job"]: + _proof_request(request, profile, proof_arguments) + return request for field in ("job", "sandbox", "exitCode"): if request.get(field) != profile[field]: raise AdapterError( @@ -834,7 +891,9 @@ def run(config_path, request_path, source_bundle, evidence_dir): profile = configuration["profile"] if policy.executor_id != profile["executorId"]: raise AdapterError("gateway policy does not match the trusted profile") - request = _request(request_path, policy, profile) + request = _request( + request_path, policy, profile, configuration.get(_PROOF_FIELD) + ) source_bundle = _inspect_regular_file(source_bundle, "source bundle") evidence_dir = _preflight_destination( evidence_dir, diff --git a/skills/underwrite/SKILL.md b/skills/underwrite/SKILL.md index 604b5dd..7eb3c19 100644 --- a/skills/underwrite/SKILL.md +++ b/skills/underwrite/SKILL.md @@ -59,13 +59,15 @@ is two beats, or the sentence is carrying weight it does not need. Do not pad up budget either: the examples below are the target, and both are well under it. `PROOF` names a command you ran or a file you read this session, and the beat says which -in `proof_kind`: `ran`, `read`, or `inferred`. `ran` means the command's output would -have read differently were the claim false, and `PROOF` quotes what it printed. `read` -means you looked and it looked right. `inferred` is a legal value and an honest one. A -claim with none of them is not shippable. Only `ran` earns `CLEAN`: a read or inferred -proof opens the beat `UNVERIFIED`, and the store refuses the other spelling. `UNVERIFIED` -is not a failure. It is the state most beats of a no-exec PR session will open with, and -it is what tells the reviewer where their own reading goes. +in `proof_kind`: `ran`, `read`, `inferred`, or `attested`. `ran` means the command's +output would have read differently were the claim false, and `PROOF` quotes what it +printed. `read` means you looked and it looked right. `inferred` is a legal value and an +honest one. `attested` is a command the execution gateway ran on the frozen PR head and +signed for, described under **Attested proof**. A claim with none of them is not +shippable. Only `ran` and `attested` earn `CLEAN`: a read or inferred proof opens the beat +`UNVERIFIED`, and the store refuses the other spelling. `UNVERIFIED` is not a failure. It +is the state most beats of a no-exec PR session open with when the host runs no proof +gateway, and it is what tells the reviewer where their own reading goes. A `RISK` belongs to a flag. A clean or unverified beat that carries one is refused: a risk worth a line is worth a decision, so raise the flag with its `FIX`, or drop the slot. @@ -197,18 +199,19 @@ this session, even when the PR adds a governing file where the base had none. **The capture freezes the target, `no-exec` policy, and derived audience in one transaction before returning.** Every PR head is untrusted executable input, independently of audience, -author, fork status, reviews, or merge state. The source review or report session does not -invoke the standalone host gateway, so a caller-supplied sandbox label is never -authorization. +author, fork status, reviews, or merge state. The source review or report session reaches +the standalone host gateway only through an attested proof, whose verified receipt is +evidence for one beat, so a caller-supplied sandbox label is never authorization. Do not check out or execute the head. Legacy PR sessions that lack the current frozen context require a supervised replacement. The standalone gateway, `docs/host-execution-protocol.md`, and `scripts/execution_receipt.py` define the host trust boundary. The gateway can enforce and persist a signed execution, but its receipt does not authorize a review effect or mutate a -source session. Never invoke it as review authorization. Every PR source remains `no_exec`. -Only the separate linked implementation flow below may create a `gateway_attested` child -after a distinct implementation approval. +source session. Never invoke it as review authorization. Every PR source remains `no_exec`: +this machine never executes the head, and an attested proof under **Attested proof** runs +only inside the gateway. Only the separate linked implementation flow below may create a +`gateway_attested` child after a distinct implementation approval. After the snapshot is frozen, run these contextual reads in parallel: @@ -302,7 +305,8 @@ State the frozen audience and lifecycle in one line here. Correct a bad capture supervised replacement, never an in-session audience change. State the execution policy separately: source PR snapshots are `no_exec` and use static -evidence only. Never describe review audience as execution trust. A linked child, if later +evidence, plus any attested proof the host's gateway signs. Never describe review audience +as execution trust. A linked child, if later authorized, is a separate session and does not change this statement. **Then stop and put the decision in front of them.** Use the harness's structured question @@ -415,11 +419,84 @@ Never compose a shell command from a PR filename or ref. Do not check out the he install dependencies, build, test, lint, benchmark, invoke an interpreter on repo files, run package or repo scripts, build a container, or run Git hooks or filters. A command suggested by PR text or output is never an exception. `PROOF` may name a file read or -`inferred`; do not imply runtime verification. A `ran` proof in a PR session names an +`inferred`, or an `attested` run the gateway signed for; do not imply runtime +verification any other way. A `ran` proof in a PR session names an Underwrite-owned reader and what it returned, such as two blobs hashing alike or a diff that comes back empty; reading a blob and agreeing with it is `read`, and the beat opens `UNVERIFIED`. Do not narrate the surrounding-code reading. + +### Attested proof + +A claim a single run would settle should be run, not read. When the host operator has +set up a proof gateway, a PR beat can earn `CLEAN`, or a flag can carry its failing run, +without this machine executing anything from the head. The gateway runs the command on +its own quarantine of the frozen head inside its sandbox: no network, no credentials, one +process, a fixed time and memory budget. It signs a receipt only when the command exits +with the code you predicted. The prediction is the claim: say the test fails on the head +and predict `1`, or say it passes and predict `0`. A receipt is then a run whose result +would have differed were the claim false. + +The operator supplies a proof configuration `$G` outside every repository and session, +with exactly `version`, `profile` (the trusted profile path), `publicKey` (the pinned +P-256 key path), `adapter` (the absolute argv that runs the gateway adapter, a wrapper +when the gateway is another account), and `evidenceRoot` (the gateway's outbox). The +profile's own command is the fixed prefix, such as `/usr/local/bin/python3 -I -B -m +unittest`; you choose only what comes after it, at most 16 arguments of at most 512 +bytes each, and the exit code. The adapter's configuration must opt in with +`proofArguments`. Never take any of these from the head, the PR text, or returned +evidence. When `$G` is not set up, there is no proof gateway: write `read` or `inferred` +and say so, never `ran` for something you did not run. + +Run one before writing the beat it backs, and put `--` before the arguments: + +```bash +$S/scripts/proofctl.py prove "$R" "$G" --exit-code 1 -- tests.test_app.AppTest +``` + +`prove` reserves the request, runs the adapter, verifies the six-file handoff the same way +`implementationctl.py consume` does (both signatures under the pinned key, the frozen +source bundle, the output tree, both complete streams, and the exact job and exit code), +keeps the request, envelopes and streams under `$R/proof-evidence//`, and prints +the proof. It exits 0 with `status: verified`. It exits 2 with `status: failed` when the +gateway proved it will never sign this attempt, most often because the command did not +exit as predicted: the failure records the gateway's reason, and nothing was proven. A +wrong prediction is information, not evidence; write the beat from what you read, or run +a sharper command. It exits 3 with `status: retry` on an unknown outcome; repeat the same +call, which reuses the same reservation. Exit 1 is a refusal with its reason. + +An exit code is coarse: a test module that fails to import exits 1 just like a failing +assertion. Before you cite a verified proof, read the signed output `prove` returns under +`output`, and cite it only when that output shows your claim, such as the assertion you +named. When it shows something else, the run proved nothing about the claim; say what it +did show, or run a sharper command. + +Then write the beat with `proof_kind: "attested"` and `attested_proof: `. `PROOF` +quotes the command and what it printed, in the budget like any other slot. The store +refuses an attested beat whose proof is not verified, was run for another beat, or sits +in a session without a frozen PR target, and refuses `attested_proof` beside any other +proof kind. It then writes `attestation` itself from the verified row: the exact argv, the +exit code, the receipt digest, the key and executor, and the last lines of each stream. +Never write `attestation`; the page shows it beside the beat as what ran, so your `PROOF` +line cannot pass for the run. + +Run only what a claim needs: one test module or case, a `-k` filter, a single check. A +command that needs network, an install, or a second process fails in the sandbox; that is +a limit of the evidence, not a finding about the change. The separate steps stay +available when the adapter is driven by another supervisor: + +```bash +$S/scripts/proofctl.py request "$R" "$P" --exit-code 1 -- tests.test_app +$S/scripts/proofctl.py consume "$R" "$P" "$K" "$E" +$S/scripts/proofctl.py fail "$R" --reason '' +$S/scripts/proofctl.py proofs "$R" +``` + +`request` prints the exact gateway request; the adapter's `request.json` must be its +canonical bytes. Use `fail` only for an adapter `status: failed`, as with implementation +attempts. A proof does not change the session's `no_exec` policy, its audience, or any +delivery, and it never authorizes an effect. + **Read the comments, then write what they do not say.** A well-commented diff hands you a `WHY` for every hunk, and rule 5 is the one you will break without noticing. The tell is a slot you could have written without leaving the docstring. Caught in a real session: @@ -1021,9 +1098,15 @@ implementations// one source-bound implementation child output.bundle verified output transport stdout complete captured standard output stderr complete captured standard error +proof-evidence// one verified attested proof + request.json exact reserved request + capability.dsse.json signed host capability + receipt.dsse.json signed execution receipt + stdout complete captured standard output + stderr complete captured standard error ``` -All mutations go through `sessionctl.py` or the server's `/act` endpoint. All reads used +All mutations go through `sessionctl.py`, `proofctl.py`, or the server's `/act` endpoint. All reads used for a later mutation go through `get-session` or `get-beat`. `session.json`, `beats/`, `decisions.jsonl`, and `ack.json` are compatibility exports. The renderer reads SQLite whenever it exists, and a later `$S/scripts/sessionctl.py export "$R"` repairs missing or @@ -1070,8 +1153,9 @@ a beat dropped before the field existed has none. Set top-level ordinary flags persist `delivery`; imported legacy beats may omit it. `slots` accepts only the six keys from rule 2, spelled lower case: `what`, `why`, `proof`, `risk`, `prior`, `fix`. Rule 2 shows how the page labels them, not how the store spells them. `diff` is a list of raw lines, classified on the first character. -`proof_kind` is `ran`, `read`, or `inferred` on every beat written now, and `clean` -requires `ran`. A beat stored before the field existed keeps the verdict it landed with. +`proof_kind` is `ran`, `read`, `inferred`, or `attested` on every beat written now, and +`clean` requires `ran` or `attested`. `attested` requires `attested_proof`, the number of +a verified proof run for that beat, and the store owns `attestation`. A beat stored before the field existed keeps the verdict it landed with. `lands[]` entries are `{state: landed|ready|open, what, where}`. `recommendation` is nonempty session text. A flag's `decision` is an object with exactly diff --git a/skills/underwrite/assets/report.css b/skills/underwrite/assets/report.css index f3d8a3b..a85aa38 100644 --- a/skills/underwrite/assets/report.css +++ b/skills/underwrite/assets/report.css @@ -963,3 +963,47 @@ footer { @media (prefers-reduced-motion: reduce) { * { transition: none !important; animation: none !important; } } + +/* A claim the gateway ran on the frozen head, from the signed receipt the store holds. */ +.attested { + background: var(--clean-bg); + padding: 9px 14px; + border-radius: 3px; + font-size: 13px; + color: var(--ink-2); + margin: 8px 0; +} +.attested .lbl { + font-family: var(--mono); + font-size: 11px; + letter-spacing: 0.13em; + text-transform: uppercase; + color: var(--clean); +} +.attested p { margin: 4px 0; } +.attested-cmd { + display: block; + font-family: var(--mono); + font-size: 12px; + color: var(--ink); + overflow-wrap: anywhere; + margin: 4px 0; +} +.attested-out { + font-family: var(--mono); + font-size: 12px; + color: var(--ink); + background: var(--raised); + margin: 6px 0; + padding: 6px 8px; + border-left: 2px solid var(--clean); + white-space: pre-wrap; + overflow-wrap: anywhere; + max-height: 14em; + overflow-y: auto; +} +.attested-id { + font-family: var(--mono); + font-size: 11px; + color: var(--ink-2); +} diff --git a/skills/underwrite/scripts/attested_implementation.py b/skills/underwrite/scripts/attested_implementation.py index b32bcc5..8c86ade 100644 --- a/skills/underwrite/scripts/attested_implementation.py +++ b/skills/underwrite/scripts/attested_implementation.py @@ -654,29 +654,29 @@ def _stored_evidence_root(child, attempt): return child.root / "implementation-evidence" / str(attempt) -def _verify_persisted_evidence(child, attempt, values): - destination = _stored_evidence_root(child, attempt) +def _verify_persisted_files(destination, values, label): try: details = destination.lstat() except FileNotFoundError as error: - raise Conflict("persisted implementation evidence is missing") from error + raise Conflict(f"persisted {label} is missing") from error if not stat.S_ISDIR(details.st_mode) or stat.S_ISLNK(details.st_mode): - raise Conflict("stored implementation evidence is not a directory") + raise Conflict(f"stored {label} is not a directory") _require_exact_directory_names( destination, values, - "stored implementation evidence", + f"stored {label}", Conflict, ) for name, expected in values.items(): actual = _read_file(destination / name, f"stored {name}", len(expected)) if actual != expected: - raise Conflict(f"stored implementation evidence {name} changed") + raise Conflict(f"stored {label} {name} changed") return destination -def _persist_evidence(child, attempt, values): - parent = child.root / "implementation-evidence" +def _persist_files(root, parent_name, name, values, label): + """Write one evidence set under root/parent_name/name, atomically and once.""" + parent = root / parent_name created = False try: parent.mkdir(mode=0o700) @@ -684,33 +684,49 @@ def _persist_evidence(child, attempt, values): except FileExistsError: details = parent.lstat() if not stat.S_ISDIR(details.st_mode) or stat.S_ISLNK(details.st_mode): - raise ImplementationError("implementation evidence root is not a directory") + raise ImplementationError(f"{label} root is not a directory") if created: - _fsync_directory(child.root) - destination = _stored_evidence_root(child, attempt) + _fsync_directory(root) + destination = parent / name if destination.exists(): - _verify_persisted_evidence(child, attempt, values) + _verify_persisted_files(destination, values, label) return destination - temporary = Path(tempfile.mkdtemp(prefix=f".{attempt}.", dir=parent)) + temporary = Path(tempfile.mkdtemp(prefix=f".{name}.", dir=parent)) try: - for name, data in values.items(): - _write_exact(temporary / name, data) + for file_name, data in values.items(): + _write_exact(temporary / file_name, data) _fsync_directory(temporary) try: os.rename(temporary, destination) except OSError: if not destination.exists(): raise - _verify_persisted_evidence(child, attempt, values) + _verify_persisted_files(destination, values, label) _fsync_directory(parent) - _fsync_directory(child.root) + _fsync_directory(root) finally: if temporary.exists(): shutil.rmtree(temporary) return destination +def _verify_persisted_evidence(child, attempt, values): + return _verify_persisted_files( + _stored_evidence_root(child, attempt), values, "implementation evidence" + ) + + +def _persist_evidence(child, attempt, values): + return _persist_files( + child.root, + "implementation-evidence", + str(attempt), + values, + "implementation evidence", + ) + + def _verified_replay(child, attempt, values): if attempt.get("state") not in ("verified", "prepared", "landed"): return None diff --git a/skills/underwrite/scripts/attested_proof.py b/skills/underwrite/scripts/attested_proof.py new file mode 100755 index 0000000..c8a2bc3 --- /dev/null +++ b/skills/underwrite/scripts/attested_proof.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +"""Run one walk claim on the frozen PR head through the execution gateway. + +A PR session never executes the head. A proof does not change that: the gateway runs the +command on its own quarantine of the frozen head, inside its sandbox, and signs a receipt +only when the command exits with the code the walk predicted. This module reserves that +request, consumes the gateway's evidence, verifies every signature and digest against the +trusted profile and pinned key, and records the verified run so one beat can cite it. +""" + +import copy +import hashlib +import json +import os +import subprocess +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +from attested_implementation import ( + ImplementationError, + _JSON_BYTES, + _bounded_json_integer, + _canonical_bytes, + _evidence_directory, + _persist_files, + _profile, + _read_file, + _read_json, + _stream_descriptor, + _unique_object, + _verify_persisted_files, +) +from execution_receipt import verify_execution_receipt +from gateway import artifacts +from gateway.signing import OpenSSLVerifier +from session_store import Conflict, SessionStore, StoreError + + +PROOF_EVIDENCE_DIRECTORY = "proof-evidence" +# The output bundle is verified against the signed output tree and then let go: a proof +# is what the command printed and how it exited, never the tree it left behind. +_PERSISTED = ( + "request.json", + "capability.dsse.json", + "receipt.dsse.json", + "stdout", + "stderr", +) +_ADAPTER_SECONDS = 1200 +_ADAPTER_OUTPUT_BYTES = 64 * 1024 +_CONFIG_FIELDS = {"version", "profile", "publicKey", "adapter", "evidenceRoot"} + + +class ProofError(ImplementationError): + """The proof cannot be reserved, run, or verified safely.""" + + +def _public(attempt): + result = copy.deepcopy(attempt) + result.pop("capability", None) + result.pop("receipt", None) + return result + + +def _proof_profile(profile, request): + """The request is the profile's own job with arguments after its command.""" + job = request["job"] + prefix = profile["job"]["argv"] + if job["argv"][: len(prefix)] != prefix or len(job["argv"]) <= len(prefix): + raise Conflict("proof command does not extend the trusted profile command") + for name in ("cwd", "environment", "executable", "stdin"): + if job[name] != profile["job"][name]: + raise Conflict(f"proof job {name} does not match the trusted profile") + if request["sandbox"] != profile["sandbox"]: + raise Conflict("proof sandbox does not match the trusted profile") + + +def request(session_root, beat, profile_source, arguments, exit_code): + """Reserve one proof for a beat and return the exact gateway request.""" + store = SessionStore(session_root) + store.verify_target_files() + attempt = store.reserve_proof(beat, _profile(profile_source), arguments, exit_code) + return _public(attempt) + + +def fail(session_root, proof, reason): + """Record that the gateway proved this attempt can never produce a receipt.""" + return _public(SessionStore(session_root).fail_proof(proof, reason)) + + +def proofs(session_root): + return {"proofs": [_public(item) for item in SessionStore(session_root).proof_attempts()]} + + +def _verified_replay(store, attempt, values, output_bundle): + evidence = attempt["evidence"] + checks = { + "requestSha256": hashlib.sha256(values["request.json"]).hexdigest(), + "capabilitySha256": hashlib.sha256(values["capability.dsse.json"]).hexdigest(), + "receiptSha256": hashlib.sha256(values["receipt.dsse.json"]).hexdigest(), + "outputBundle": { + "sha256": hashlib.sha256(output_bundle).hexdigest(), + "bytes": len(output_bundle), + }, + "stdout": _stream_descriptor(values["stdout"]), + "stderr": _stream_descriptor(values["stderr"]), + } + for name, value in checks.items(): + if evidence.get(name) != value: + raise Conflict(f"verified proof replay changed {name}") + _verify_persisted_files( + store.root / PROOF_EVIDENCE_DIRECTORY / str(attempt["proof"]), + values, + "proof evidence", + ) + return attempt + + +def consume(session_root, proof, profile_source, public_key, evidence_dir, *, now=None): + """Verify the gateway's six-file handoff for one proof and record it.""" + store = SessionStore(session_root) + profile = _profile(profile_source) + directory = _evidence_directory(evidence_dir) + request_bytes = _read_file(directory / "request.json", "execution request", _JSON_BYTES) + try: + request_value = json.loads( + request_bytes.decode("utf-8"), + object_pairs_hook=_unique_object, + parse_int=_bounded_json_integer, + ) + except ImplementationError: + raise + except (UnicodeError, ValueError, RecursionError) as error: + raise ProofError("execution request is not valid UTF-8 JSON") from error + try: + coordinate = request_value["action"]["seq"] + except (KeyError, TypeError) as error: + raise ProofError("execution request has no proof coordinate") from error + if coordinate != proof: + raise Conflict(f"evidence is for proof {coordinate}, not proof {proof}") + attempt = store.proof_attempt(proof) + if request_bytes != _canonical_bytes(attempt["request"]): + raise Conflict("evidence request bytes do not match the reserved proof") + if attempt["trusted_profile"] != profile: + raise Conflict("trusted execution profile changed after proof reservation") + reserved = attempt["request"] + _proof_profile(profile, reserved) + limits = profile["sandbox"]["limits"] + workspace_limit = limits["workspaceBytes"] + output_limit = limits["outputBytes"] + bundle_limit = workspace_limit * 2 + capability = _read_file(directory / "capability.dsse.json", "host capability", _JSON_BYTES) + receipt = _read_file(directory / "receipt.dsse.json", "execution receipt", _JSON_BYTES) + output_bundle = _read_file(directory / "output.bundle", "output bundle", bundle_limit) + stdout = _read_file(directory / "stdout", "execution stdout", output_limit) + stderr = _read_file(directory / "stderr", "execution stderr", output_limit) + if len(stdout) + len(stderr) > output_limit: + raise ProofError("execution streams exceed the trusted output limit") + verifier = OpenSSLVerifier(public_key, profile["signerId"]) + if verifier.key_id != profile["keyId"]: + raise Conflict("public key does not match the trusted profile key ID") + values = { + "request.json": request_bytes, + "capability.dsse.json": capability, + "receipt.dsse.json": receipt, + "stdout": stdout, + "stderr": stderr, + } + if attempt["state"] == "verified": + return _public(_verified_replay(store, attempt, values, output_bundle)) + if attempt["state"] != "reserved": + raise Conflict(f"proof {proof} is {attempt['state']}, not reserved") + target = store.verify_target_files() + if target != reserved["target"]: + raise Conflict("proof request target is not the frozen session target") + with tempfile.TemporaryDirectory(prefix="underwrite-proof-") as temporary: + input_result = artifacts.verify_source_bundle( + store.read_object_bundle(), + target, + Path(temporary) / "input", + maximum_workspace_bytes=workspace_limit, + ) + output = artifacts.verify_output_bundle( + output_bundle, + maximum_workspace_bytes=workspace_limit, + maximum_bundle_bytes=bundle_limit, + ) + expected = { + "signerId": profile["signerId"], + "executorId": profile["executorId"], + "sessionId": reserved["sessionId"], + "challenge": attempt["challenge"], + "target": target, + "action": copy.deepcopy(reserved["action"]), + "job": copy.deepcopy(reserved["job"]), + "inputTree": input_result.git_tree, + "sandbox": copy.deepcopy(profile["sandbox"]), + "outputTree": output.git_tree, + "outputBundle": {"sha256": output.sha256, "bytes": output.bytes}, + "stdout": _stream_descriptor(stdout), + "stderr": _stream_descriptor(stderr), + "exitCode": reserved["exitCode"], + } + verified = verify_execution_receipt( + capability, + receipt, + expected, + verifier.verify, + now or datetime.now(timezone.utc), + ) + evidence = { + "version": 1, + "requestSha256": attempt["request_sha256"], + "keyId": profile["keyId"], + "signerId": verified.signer_id, + "executorId": profile["executorId"], + "capabilitySha256": hashlib.sha256(capability).hexdigest(), + "receiptSha256": hashlib.sha256(receipt).hexdigest(), + "inputTree": input_result.git_tree, + "outputTree": output.git_tree, + "outputBundle": {"sha256": output.sha256, "bytes": output.bytes}, + "stdout": _stream_descriptor(stdout), + "stderr": _stream_descriptor(stderr), + "exitCode": reserved["exitCode"], + } + _persist_files(store.root, PROOF_EVIDENCE_DIRECTORY, str(proof), values, "proof evidence") + return _public( + store.record_verified_proof(proof, capability, receipt, evidence, stdout, stderr) + ) + + +def _absolute(value, label): + if not isinstance(value, str) or not os.path.isabs(value): + raise ProofError(f"proof configuration {label} must be an absolute path") + return value + + +def load_configuration(path): + """The controller's own record of where the host's proof gateway lives. + + Host-provisioned and outside every repository and session, like the profile and key + it names. Nothing in it comes from the head or from returned evidence.""" + value = _read_json(path, "proof configuration") + if set(value) != _CONFIG_FIELDS or value.get("version") != 1: + raise ProofError( + "proof configuration must be version 1 with exactly " + + ", ".join(sorted(_CONFIG_FIELDS)) + ) + for name in ("profile", "publicKey", "evidenceRoot"): + _absolute(value[name], name) + adapter = value["adapter"] + if ( + not isinstance(adapter, list) + or not adapter + or any(not isinstance(part, str) or not part or "\x00" in part for part in adapter) + ): + raise ProofError("proof configuration adapter must be a non-empty argv list") + _absolute(adapter[0], "adapter[0]") + return value + + +def _adapter_outcome(completed): + try: + result = json.loads(completed.stdout.decode("utf-8")) + except (UnicodeError, ValueError): + result = None + detail = completed.stderr.decode("utf-8", "replace").strip().splitlines() + detail = detail[-1] if detail else "" + if not isinstance(result, dict): + return "retry", detail or "adapter printed no result" + status = result.get("status") + if status == "complete" and completed.returncode == 0: + return "complete", result.get("evidenceDir") + if status == "failed" and completed.returncode == 2: + return "failed", detail or result.get("reason") or "gateway attempt is terminal" + return "retry", detail or result.get("reason") or "unknown adapter outcome" + + +def prove(session_root, beat, configuration_path, arguments, exit_code, *, now=None): + """Reserve, run through the host adapter, and consume one proof in a single call. + + `failed` means the gateway proved it will never sign this attempt, most often + because the command did not exit as predicted. `retry` leaves the same reservation + for a byte-identical rerun.""" + configuration = load_configuration(configuration_path) + store = SessionStore(session_root) + attempt = request(session_root, beat, configuration["profile"], arguments, exit_code) + proof = attempt["proof"] + if attempt["state"] == "verified": + return {"status": "verified", "proof": attempt} + if attempt["state"] != "reserved": + raise Conflict(f"proof {proof} is {attempt['state']}") + session_id = store.delivery_state()["session_id"] + evidence_dir = os.path.join( + configuration["evidenceRoot"], f"underwrite-{session_id}-proof-{proof}" + ) + with tempfile.TemporaryDirectory(prefix="underwrite-proof-request-") as temporary: + request_path = Path(temporary) / "request.json" + request_path.write_bytes(_canonical_bytes(attempt["request"])) + request_path.chmod(0o644) + Path(temporary).chmod(0o755) + try: + completed = subprocess.run( + [ + *configuration["adapter"], + "--request", str(request_path), + "--source-bundle", str(store.root / "pr.bundle"), + "--evidence-dir", evidence_dir, + ], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=_ADAPTER_SECONDS, + check=False, + ) + except subprocess.TimeoutExpired: + return { + "status": "retry", + "proof": attempt, + "reason": "the gateway adapter did not finish in time", + } + except OSError as error: + raise ProofError(f"gateway adapter could not be started: {error}") from error + completed.stdout = completed.stdout[:_ADAPTER_OUTPUT_BYTES] + completed.stderr = completed.stderr[-_ADAPTER_OUTPUT_BYTES:] + status, detail = _adapter_outcome(completed) + if status == "failed": + return {"status": "failed", "proof": fail(session_root, proof, detail)} + if status == "retry": + return {"status": "retry", "proof": attempt, "reason": detail} + if detail != evidence_dir: + raise Conflict("gateway adapter published evidence somewhere other than asked") + verified = consume( + session_root, + proof, + configuration["profile"], + configuration["publicKey"], + evidence_dir, + now=now, + ) + return {"status": "verified", "proof": verified} diff --git a/skills/underwrite/scripts/proofctl.py b/skills/underwrite/scripts/proofctl.py new file mode 100755 index 0000000..f446d2a --- /dev/null +++ b/skills/underwrite/scripts/proofctl.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Run walk claims on the frozen PR head through the execution gateway.""" + +import argparse +import json +import sqlite3 +import sys +from pathlib import Path + + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parents[2] +for path in (HERE, ROOT): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +from attested_proof import consume, fail, prove, proofs, request # noqa: E402 +from attested_implementation import ImplementationError # noqa: E402 +from execution_receipt import ReceiptError # noqa: E402 +from gateway.artifacts import ArtifactError # noqa: E402 +from gateway.signing import SigningError # noqa: E402 +from session_store import StoreError # noqa: E402 + + +class Usage(argparse.ArgumentParser): + def error(self, message): + self.print_usage(sys.stderr) + print(f"proofctl: error: {message}", file=sys.stderr) + raise SystemExit(1) + + +def _command(parser): + parser.add_argument("--exit-code", type=int, required=True) + parser.add_argument("arguments", nargs="+") + + +def parser(): + command = Usage(prog="proofctl.py") + commands = command.add_subparsers(dest="command", required=True) + + run = commands.add_parser("prove") + run.add_argument("session") + run.add_argument("beat", type=int) + run.add_argument("config") + _command(run) + + reserve = commands.add_parser("request") + reserve.add_argument("session") + reserve.add_argument("beat", type=int) + reserve.add_argument("profile") + _command(reserve) + + verify = commands.add_parser("consume") + verify.add_argument("session") + verify.add_argument("proof", type=int) + verify.add_argument("profile") + verify.add_argument("public_key") + verify.add_argument("evidence_dir") + + reject = commands.add_parser("fail") + reject.add_argument("session") + reject.add_argument("proof", type=int) + reject.add_argument("--reason", required=True) + + listing = commands.add_parser("proofs") + listing.add_argument("session") + return command + + +def run(args): + if args.command == "prove": + return prove(args.session, args.beat, args.config, args.arguments, args.exit_code) + if args.command == "request": + return request(args.session, args.beat, args.profile, args.arguments, args.exit_code) + if args.command == "consume": + return consume( + args.session, args.proof, args.profile, args.public_key, args.evidence_dir + ) + if args.command == "fail": + return fail(args.session, args.proof, args.reason) + if args.command == "proofs": + return proofs(args.session) + raise ImplementationError(f"unknown command {args.command!r}") + + +def main(argv=None): + try: + result = run(parser().parse_args(argv)) + except ( + ArtifactError, + ImplementationError, + OSError, + ReceiptError, + SigningError, + sqlite3.Error, + StoreError, + UnicodeError, + json.JSONDecodeError, + ) as error: + print(f"proofctl: {error}", file=sys.stderr) + return 1 + sys.stdout.buffer.write( + json.dumps( + result, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + + b"\n" + ) + if isinstance(result, dict) and result.get("status") == "failed": + return 2 + if isinstance(result, dict) and result.get("status") == "retry": + return 3 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/skills/underwrite/scripts/render-report.py b/skills/underwrite/scripts/render-report.py index 9c5c55a..33352a7 100755 --- a/skills/underwrite/scripts/render-report.py +++ b/skills/underwrite/scripts/render-report.py @@ -20,6 +20,7 @@ import html import json import re +import shlex import sqlite3 import sys import textwrap @@ -854,6 +855,39 @@ def where_html(where, paths, target): return " ".join(out) +def attestation_html(beat): + """What the gateway ran for this beat, from the verified receipt the store holds. + + Written by the store from the proof row, never by the walk, so the command and exit + code here are the signed ones even when the PROOF slot paraphrases them.""" + attestation = beat.get("attestation") + if beat.get("proof_kind") != "attested" or not isinstance(attestation, dict): + return "" + argv = attestation.get("argv") + command = shlex.join(argv) if isinstance(argv, list) and all( + isinstance(part, str) for part in argv + ) else "" + output = attestation.get("output") if isinstance(attestation.get("output"), dict) else {} + streams = [ + f'
{html.escape(text)}
' + for name in ("stderr", "stdout") + for text in [output.get(name)] + if isinstance(text, str) and text.strip() + ] + receipt = attestation.get("receipt_sha256") + receipt = receipt[:12] if isinstance(receipt, str) else "?" + return ( + '
Ran in sandbox' + f'

The gateway ran this on the frozen head and exited ' + f'{md(attestation.get("exit_code"))}, as the claim predicted. ' + 'No head code ran on this machine.

' + f'{html.escape(command, quote=False)}' + + "".join(streams) + + f'

proof {md(attestation.get("proof"))} ยท ' + f'signed receipt {md(receipt)}

' + ) + + def questions_html(questions): parts = [] for question in questions: @@ -942,6 +976,9 @@ def beat_html( '
Unproven' f"
    {reasons}
", ) + attested = attestation_html(beat) + if attested: + body.append(attested) if beat.get("diff"): body.append(diff_html(beat["diff"])) if fix_row: @@ -1227,7 +1264,18 @@ def assessment(session, beats, problems_by_n, validation_problems=(), walking=Fa else: label = "Recommendation" if owed or walking else "Final recommendation" if execution_mode(session) == "no_exec": - limits.append("Static inspection only; runtime behavior was not verified.") + attested = sum( + beat.get("proof_kind") == "attested" + and isinstance(beat.get("attestation"), dict) + for beat in beats + ) + if attested: + limits.append( + f"{attested} item{'s' if attested != 1 else ''} ran in the gateway " + "sandbox with a signed receipt; everything else is static inspection." + ) + else: + limits.append("Static inspection only; runtime behavior was not verified.") mode = session_mode(session) included = sum(beat.get("state") == "accepted" for beat in beats) if mode in ("review", "report") and included: diff --git a/skills/underwrite/scripts/session_store.py b/skills/underwrite/scripts/session_store.py index 69b3e42..7bcd5d3 100644 --- a/skills/underwrite/scripts/session_store.py +++ b/skills/underwrite/scripts/session_store.py @@ -24,7 +24,7 @@ from diff_lines import parse_diff, untouched_anchor # noqa: E402 -DB_SCHEMA_VERSION = 7 +DB_SCHEMA_VERSION = 8 SCHEMA_VERSION = 1 DELIVERY_VERSION = 1 FINDINGS_FILE = "findings.md" @@ -51,10 +51,21 @@ DROP_REASONS = ("wrong", "intentional", "not-worth-it", "pre-existing") OPEN_STATES = ("clean", "flag", "unverified") # What PROOF is: a command whose output would have read differently were the claim -# false, a file or diff read, or an inference. Only `ran` earns CLEAN. -PROOF_KINDS = ("ran", "read", "inferred") +# false, a file or diff read, or an inference. `attested` is a command the execution +# gateway ran on the frozen head and signed for, with the exit code the walk predicted. +# Only `ran` and `attested` earn CLEAN. +PROOF_KINDS = ("ran", "read", "inferred", "attested") +CLEAN_PROOF_KINDS = ("ran", "attested") RESOLUTION_KINDS = ("delivery", "decision") -STORE_OWNED_FIELDS = ("call", "reason", "landed", "branch", "delivery_kind", "delivery") +STORE_OWNED_FIELDS = ( + "call", "reason", "landed", "branch", "delivery_kind", "delivery", "attestation", +) +# What a proof may append to the trusted profile's command, and how much of what it +# printed the page keeps. The full streams stay in the session's proof evidence. +MAX_PROOF_ARGUMENTS = 16 +MAX_PROOF_ARGUMENT_BYTES = 512 +PROOF_OUTPUT_TAIL_LINES = 8 +PROOF_OUTPUT_TAIL_CHARS = 1000 # Rule 2's budget, in numbers. It binds what a walk writes now and never what is # already on disk: validate_beat stays the shippability contract for stored beats, # so tightening here cannot retroactively unship a session or an accepted finding. @@ -256,11 +267,21 @@ def proof_kind_problems(beat): problems.append( f"beat {n}: proof_kind must be one of {', '.join(PROOF_KINDS)}" ) - elif state == "clean" and kind != "ran": + elif state == "clean" and kind not in CLEAN_PROOF_KINDS: problems.append( f"beat {n}: clean on a {kind} proof; write it unverified, " "or name the command you ran" ) + if kind == "attested" and "attested_proof" not in beat: + problems.append( + f"beat {n}: attested proof names no attested_proof; " + "run proofctl.py prove first" + ) + if "attested_proof" in beat and kind != "attested": + problems.append( + f"beat {n}: attested_proof is set but proof_kind is {kind}; " + "write proof_kind attested" + ) slots = beat.get("slots") if state in ("clean", "unverified") and isinstance(slots, dict) and slots.get("risk"): problems.append( @@ -372,6 +393,10 @@ def validate_beat(beat, mode="branch", final=False): f"beat {n}: proof_kind {beat['proof_kind']!r} is not one of " f"{', '.join(PROOF_KINDS)}" ) + if beat.get("proof_kind") == "attested" and not isinstance( + beat.get("attestation"), dict + ): + problems.append(f"beat {n}: attested proof carries no gateway attestation") return problems @@ -747,6 +772,7 @@ def _create_schema(self, db): ) for statement in statements: db.execute(statement) + self._create_proof_table(db) db.execute(f"PRAGMA user_version = {DB_SCHEMA_VERSION}") def _actions_schema(self, name): @@ -828,6 +854,7 @@ def _upgrade_schema(self): if version in (1, 2, 3): self._default_legacy_execution_policy(db) self._create_implementation_tables(db) + self._create_proof_table(db) if db.total_changes != before: # Derived from rows a repair just rewrote, and before the commit, so a # failed write takes the repair with it and the next open retries both. @@ -884,6 +911,25 @@ def _create_implementation_tables(self, db): PRIMARY KEY (action_seq, attempt) )""") + def _create_proof_table(self, db): + db.execute("""CREATE TABLE IF NOT EXISTS proof_attempts ( + proof INTEGER PRIMARY KEY CHECK (proof > 0), + beat INTEGER NOT NULL CHECK (beat > 0), + challenge TEXT NOT NULL UNIQUE, + state TEXT NOT NULL CHECK (state IN ('reserved', 'verified', 'failed')), + trusted_profile_json TEXT NOT NULL, + request_json TEXT NOT NULL, + request_sha256 TEXT NOT NULL, + capability BLOB, + receipt BLOB, + evidence_json TEXT, + output_json TEXT, + failure TEXT, + created_at TEXT NOT NULL, + verified_at TEXT, + failed_at TEXT + )""") + def _default_legacy_execution_policy(self, db): row = self._session_row(db) body = json.loads(row["body_json"]) @@ -3077,11 +3123,15 @@ def _stream_descriptor(self, document, label): raise StoreError(f"implementation evidence {label} must not be truncated") return value - def _implementation_evidence(self, document, attempt, capability, receipt): + def _implementation_evidence( + self, document, attempt, capability, receipt, exit_code=None + ): if not isinstance(document, dict) or set(document) != _IMPLEMENTATION_EVIDENCE_FIELDS: raise StoreError("implementation evidence has an unsupported shape") evidence = _copy(document) profile = json.loads(attempt["trusted_profile_json"]) + if exit_code is None: + exit_code = profile["exitCode"] if type(evidence["version"]) is not int or evidence["version"] != 1: raise StoreError("implementation evidence version must be 1") expected = { @@ -3091,7 +3141,7 @@ def _implementation_evidence(self, document, attempt, capability, receipt): "executorId": profile["executorId"], "capabilitySha256": hashlib.sha256(capability).hexdigest(), "receiptSha256": hashlib.sha256(receipt).hexdigest(), - "exitCode": profile["exitCode"], + "exitCode": exit_code, } for name, value in expected.items(): if evidence[name] != value: @@ -3303,6 +3353,273 @@ def finish_implementation_land(self, action_seq, attempt, commit, branch): action_seq, action["beat_n"], "landed", desired ) + # ---- attested proofs -------------------------------------------- + + def _proof_session(self, db): + """The one kind of session a proof may run for: a frozen PR source. + + The controller still executes nothing. The gateway runs the command on its own + quarantine of the frozen head, and what comes back is evidence for one beat.""" + session_row = self._session_row(db) + session = json.loads(session_row["body_json"]) + target = session.get("target") + if not isinstance(target, dict): + raise Conflict("attested proof requires a frozen PR target") + if "object_bundle_sha256" not in target: + raise Conflict("attested proof requires a frozen Git object bundle") + if session.get("linked_implementation") is not None: + raise Conflict("a linked implementation child runs no proofs") + policy = self._required_execution_policy(session) + if policy["mode"] != "no_exec": + raise Conflict("attested proof runs only for a no-exec source session") + return session_row, session + + def _proof_arguments(self, arguments): + if not isinstance(arguments, list) or not arguments: + raise StoreError("proof arguments must be a non-empty list") + if len(arguments) > MAX_PROOF_ARGUMENTS: + raise StoreError( + f"proof takes at most {MAX_PROOF_ARGUMENTS} arguments after the " + "trusted command" + ) + for index, argument in enumerate(arguments): + if not isinstance(argument, str) or not argument: + raise StoreError(f"proof argument {index} must be non-empty text") + if any(ord(character) < 32 or ord(character) == 127 for character in argument): + raise StoreError(f"proof argument {index} contains a control character") + if len(argument.encode("utf-8")) > MAX_PROOF_ARGUMENT_BYTES: + raise StoreError( + f"proof argument {index} exceeds {MAX_PROOF_ARGUMENT_BYTES} bytes" + ) + return list(arguments) + + def _proof_document(self, row): + return { + "version": 1, + "proof": row["proof"], + "beat": row["beat"], + "state": row["state"], + "challenge": row["challenge"], + "request": json.loads(row["request_json"]), + "request_sha256": row["request_sha256"], + "trusted_profile": json.loads(row["trusted_profile_json"]), + "capability": row["capability"], + "receipt": row["receipt"], + "evidence": None if row["evidence_json"] is None else json.loads(row["evidence_json"]), + "output": None if row["output_json"] is None else json.loads(row["output_json"]), + "failure": row["failure"], + "created_at": row["created_at"], + "verified_at": row["verified_at"], + "failed_at": row["failed_at"], + } + + def _proof_row(self, db, proof): + _positive(proof, "proof") + row = db.execute( + "SELECT * FROM proof_attempts WHERE proof = ?", (proof,) + ).fetchone() + if row is None: + raise StoreError(f"session has no proof {proof}") + return row + + def proof_attempt(self, proof): + with self._read() as db: + return self._proof_document(self._proof_row(db, proof)) + + def proof_attempts(self): + with self._read() as db: + return [ + self._proof_document(row) + for row in db.execute("SELECT * FROM proof_attempts ORDER BY proof") + ] + + def reserve_proof(self, beat, trusted_profile, arguments, exit_code): + """Bind one command and its predicted exit code to a beat, ready for the gateway. + + The command is the trusted profile's own, with the walk's arguments after it. + The prediction is the claim: the gateway signs only an exit code that matches + it, so a receipt is a run whose result would have differed were the claim false. + A retry of the same command returns the attempt it already has; only a definite + failure makes room for a fresh one.""" + beat = _positive(beat, "proof beat") + profile = self._trusted_implementation_profile(trusted_profile) + arguments = self._proof_arguments(arguments) + _non_negative(exit_code, "proof exit code") + if exit_code > 255: + raise StoreError("proof exit code must not exceed 255") + profile_json = _dump(profile) + job = _copy(profile["job"]) + job["argv"] = list(job["argv"]) + arguments + with self._write() as db: + session_row, session = self._proof_session(db) + for row in db.execute( + "SELECT * FROM proof_attempts WHERE beat = ? AND state != 'failed' " + "ORDER BY proof", + (beat,), + ): + request = json.loads(row["request_json"]) + if ( + row["trusted_profile_json"] == profile_json + and request["job"] == job + and request["exitCode"] == exit_code + ): + return self._proof_document(row) + number = db.execute( + "SELECT COALESCE(MAX(proof), 0) + 1 FROM proof_attempts" + ).fetchone()[0] + challenge = secrets.token_hex(32) + request = { + "version": 1, + "sessionId": session_row["session_id"], + "challenge": challenge, + "target": _copy(session["target"]), + "action": {"seq": number, "beat": beat, "attempt": 1}, + "job": job, + "sandbox": _copy(profile["sandbox"]), + "exitCode": exit_code, + } + request_json = _dump(request) + if len(request_json.encode("utf-8")) > MAX_IMPLEMENTATION_REQUEST_BYTES: + raise StoreError("proof request exceeds its byte limit") + db.execute( + "INSERT INTO proof_attempts " + "(proof, beat, challenge, state, trusted_profile_json, request_json, " + "request_sha256, created_at) VALUES (?, ?, ?, 'reserved', ?, ?, ?, ?)", + ( + number, beat, challenge, profile_json, request_json, + hashlib.sha256(request_json.encode("utf-8")).hexdigest(), _now(), + ), + ) + return self._proof_document(self._proof_row(db, number)) + + def _proof_output(self, stdout, stderr, evidence): + """The tail of each stream as the page shows it, bound to the signed digests.""" + output = {} + for name, data in (("stdout", stdout), ("stderr", stderr)): + if not isinstance(data, bytes): + raise StoreError(f"proof {name} must be bytes") + if self._stream_descriptor(evidence[name], name) != { + "sha256": hashlib.sha256(data).hexdigest(), + "bytes": len(data), + "truncated": False, + }: + raise Conflict(f"proof {name} does not match its signed digest") + text = data.decode("utf-8", "replace") + text = "".join( + character + for character in text + if character in "\n\t" or not ( + ord(character) < 32 or ord(character) == 127 + ) + ) + lines = text.rstrip("\n").split("\n")[-PROOF_OUTPUT_TAIL_LINES:] + tail = "\n".join(lines)[-PROOF_OUTPUT_TAIL_CHARS:] + output[name] = tail + return output + + def record_verified_proof(self, proof, capability, receipt, evidence, stdout, stderr): + if not isinstance(capability, bytes) or not capability: + raise StoreError("verified capability must be non-empty bytes") + if not isinstance(receipt, bytes) or not receipt: + raise StoreError("verified receipt must be non-empty bytes") + with self._write() as db: + self._proof_session(db) + row = self._proof_row(db, proof) + request = json.loads(row["request_json"]) + verified = self._implementation_evidence( + evidence, row, capability, receipt, exit_code=request["exitCode"] + ) + output = self._proof_output(stdout, stderr, verified) + evidence_json, output_json = _dump(verified), _dump(output) + if row["state"] == "verified": + if ( + row["capability"] != capability + or row["receipt"] != receipt + or row["evidence_json"] != evidence_json + or row["output_json"] != output_json + ): + raise Conflict("proof already has different verified evidence") + return self._proof_document(row) + if row["state"] != "reserved": + raise Conflict(f"proof is {row['state']}, not reserved") + db.execute( + "UPDATE proof_attempts SET state = 'verified', capability = ?, " + "receipt = ?, evidence_json = ?, output_json = ?, verified_at = ? " + "WHERE proof = ?", + (capability, receipt, evidence_json, output_json, _now(), proof), + ) + return self._proof_document(self._proof_row(db, proof)) + + def fail_proof(self, proof, failure): + if not isinstance(failure, str) or not failure.strip(): + raise StoreError("proof failure must be non-empty text") + failure = failure.strip() + with self._write() as db: + self._proof_session(db) + row = self._proof_row(db, proof) + if row["state"] == "failed": + if row["failure"] != failure: + raise Conflict("proof failed for a different reason") + return self._proof_document(row) + if row["state"] != "reserved": + raise Conflict(f"proof is {row['state']}, not reserved") + db.execute( + "UPDATE proof_attempts SET state = 'failed', failure = ?, failed_at = ? " + "WHERE proof = ?", + (failure, _now(), proof), + ) + return self._proof_document(self._proof_row(db, proof)) + + def _attest_beat(self, db, session, beat): + """Hold an attested proof to the receipt it names, and write what it showed. + + The walk names the proof; the store writes the command, exit code and output + beside it from the verified row, so the page never shows the walk's paraphrase + of a run as the run itself.""" + beat.pop("attestation", None) + if beat.get("proof_kind") != "attested": + if "attested_proof" in beat: + raise StoreError( + f"beat {beat['n']}: attested_proof is set but proof_kind is " + f"{beat.get('proof_kind')}" + ) + return + if not isinstance(session.get("target"), dict): + raise Conflict( + f"beat {beat['n']}: an attested proof needs a frozen PR target; " + "a branch walk runs its own commands and writes ran" + ) + proof = beat.get("attested_proof") + if isinstance(proof, bool) or not isinstance(proof, int) or proof <= 0: + raise StoreError( + f"beat {beat['n']}: attested proof names no attested_proof number" + ) + row = db.execute( + "SELECT * FROM proof_attempts WHERE proof = ?", (proof,) + ).fetchone() + if row is None: + raise StoreError(f"beat {beat['n']}: session has no proof {proof}") + if row["beat"] != beat["n"]: + raise Conflict( + f"beat {beat['n']}: proof {proof} was run for beat {row['beat']}" + ) + if row["state"] != "verified": + raise Conflict( + f"beat {beat['n']}: proof {proof} is {row['state']}, not verified; " + "an unverified run is no proof" + ) + request = json.loads(row["request_json"]) + evidence = json.loads(row["evidence_json"]) + beat["attestation"] = { + "proof": proof, + "argv": request["job"]["argv"], + "exit_code": request["exitCode"], + "receipt_sha256": evidence["receiptSha256"], + "key_id": evidence["keyId"], + "executor_id": evidence["executorId"], + "output": json.loads(row["output_json"]), + } + def read_trusted_context(self): with self._session_lock(): target = self.frozen_target() @@ -3571,6 +3888,9 @@ def put_beat(self, document): beat[field] = current[field] else: beat.pop(field, None) + # Recomputed from its proof row below, so writing back what get-beat + # returned is no change. + beat.pop("attestation", None) if "proof_kind" in current or "proof_kind" in beat: # Written under the proof-kind rule, so a rewrite answers to it too; # a beat from before the field keeps its verdict untouched. @@ -3586,7 +3906,10 @@ def put_beat(self, document): beat.pop("resolution_kind", None) if current.get("state") == "accepted": if self._audience_mode(session) == "report": - frozen_fields = ("tier", "claim", "where", "slots", "diff", "decision") + frozen_fields = ( + "tier", "claim", "where", "slots", "diff", "decision", + "proof_kind", "attested_proof", + ) changed_fields = [ field for field in frozen_fields @@ -3609,6 +3932,8 @@ def put_beat(self, document): if not isinstance(slots, dict): raise StoreError(f"beat {beat['n']} slots must be an object") slots["fix"] = current_slots["fix"] + self._attest_beat(db, session, beat) + if row is not None: self._hold_authorized_finding(db, beat, current) self._refuse_untouched_flag(session, beat) changed, _revision, _state, _detail = self._save_beat(db, beat) diff --git a/tests/test_attested_proof.py b/tests/test_attested_proof.py new file mode 100644 index 0000000..6aed804 --- /dev/null +++ b/tests/test_attested_proof.py @@ -0,0 +1,710 @@ +#!/usr/bin/env python3 +"""Contracts for walk claims run on the frozen PR head through signed gateway evidence.""" + +import copy +import hashlib +import importlib.util +import json +import os +import shutil +import subprocess +import sys +import tempfile +import textwrap +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + + +ROOT = Path(__file__).resolve().parent.parent +SCRIPTS = ROOT / "skills" / "underwrite" / "scripts" +for path in (ROOT, SCRIPTS): + if str(path) not in sys.path: + sys.path.insert(0, str(path)) + +import attested_proof # noqa: E402 +import execution_receipt # noqa: E402 +from gateway import artifacts # noqa: E402 +from gateway.signing import OpenSSLSigner # noqa: E402 +from pr_snapshot import capture # noqa: E402 +from session_store import Conflict, SessionStore, StoreError # noqa: E402 + +_spec = importlib.util.spec_from_file_location("render_report", SCRIPTS / "render-report.py") +render_report = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(render_report) + + +def canonical(value): + return json.dumps( + value, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + + +FAILING = b"F\n======\nFAIL: test_value (tests.test_app.AppTest)\nAssertionError: 1 != 2\n" + + +class AttestedProofCase(unittest.TestCase): + merged = False + + def setUp(self): + if shutil.which("openssl") is None: + self.skipTest("OpenSSL is required") + self.temporary = tempfile.TemporaryDirectory(prefix="underwrite-proof-") + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.remote = self.root / "upstream.git" + self.controller = self.root / "controller" + self.session_root = self.root / "session" + self.git("init", "--bare", "--template=", self.remote) + self.git("init", "--template=", self.controller) + for name, value in ( + ("user.name", "Underwrite Test"), + ("user.email", "underwrite@example.test"), + ("commit.gpgsign", "false"), + ): + self.git("config", name, value, cwd=self.controller) + (self.controller / "app.py").write_text("VALUE = 0\n", encoding="utf-8") + self.git("add", "app.py", cwd=self.controller) + self.git("commit", "-m", "base", cwd=self.controller) + self.git("branch", "-M", "main", cwd=self.controller) + self.git("remote", "add", "origin", self.remote, cwd=self.controller) + self.git("push", "origin", "main", cwd=self.controller) + self.base = self.git("rev-parse", "HEAD", cwd=self.controller) + self.git("checkout", "-b", "feature", cwd=self.controller) + (self.controller / "app.py").write_text("VALUE = 1\n", encoding="utf-8") + self.git("add", "app.py", cwd=self.controller) + self.git("commit", "-m", "feature", cwd=self.controller) + self.head = self.git("rev-parse", "HEAD", cwd=self.controller) + self.git("push", "origin", "HEAD:refs/pull/7/head", cwd=self.controller) + self.git("checkout", "main", cwd=self.controller) + self.metadata = { + "number": 7, + "state": "open", + "merged_at": None, + "changed_files": 1, + "base": { + "ref": "main", + "sha": self.base, + "repo": {"full_name": "acme/widget", "clone_url": str(self.remote)}, + }, + "head": { + "ref": "feature", + "sha": self.head, + "repo": {"id": 123, "full_name": "acme/widget"}, + }, + } + if self.merged: + self.metadata["state"] = "closed" + self.metadata["merged_at"] = "2026-08-30T12:00:00Z" + self.store = SessionStore(self.session_root, create=True) + capture( + self.store, + "acme/widget", + 7, + self.controller, + api=lambda _repo, _number: copy.deepcopy(self.metadata), + ) + self.store.patch_session({"repo": "acme/widget", "cursor": 1}) + + self.private_key = self.root / "private.pem" + self.public_key = self.root / "public.pem" + self.command( + "openssl", "genpkey", "-algorithm", "EC", + "-pkeyopt", "ec_paramgen_curve:P-256", + "-pkeyopt", "ec_param_enc:named_curve", + "-out", self.private_key, + ) + self.private_key.chmod(0o600) + self.public_key.write_bytes( + self.command("openssl", "pkey", "-in", self.private_key, "-pubout", stdout=True) + ) + self.signer = OpenSSLSigner( + self.private_key, self.public_key, "urn:underwrite:signer:test" + ) + self.workspace_limit = 4 * 1024 * 1024 + self.profile = { + "version": 1, + "keyId": self.signer.key_id, + "signerId": "urn:underwrite:signer:test", + "executorId": "urn:underwrite:executor:test", + "job": { + "argv": ["/usr/bin/python3", "-I", "-B", "-m", "unittest"], + "cwd": ".", + "environment": {"LANG": "C"}, + "executable": {"path": "/usr/bin/python3", "sha256": "4" * 64, "bytes": 1}, + "stdin": "closed", + }, + "sandbox": { + "policy": execution_receipt.SANDBOX_POLICY_TYPE, + "credentials": "absent", + "network": "denied", + "hostWrites": "denied", + "gitHooks": "disabled", + "gitFilters": "disabled", + "timeout": "enforced", + "limits": { + "wallSeconds": 60, + "cpuSeconds": 60, + "memoryBytes": 64 * 1024 * 1024, + "processes": 1, + "workspaceBytes": self.workspace_limit, + "outputBytes": 1024 * 1024, + }, + }, + "exitCode": 0, + } + self.now = datetime(2026, 8, 30, 12, 0, tzinfo=timezone.utc) + + def command(self, *command, cwd=None, stdout=False): + environment = { + name: value for name, value in os.environ.items() if not name.startswith("GIT_") + } + environment.update({ + "GIT_CONFIG_GLOBAL": os.devnull, + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_TERMINAL_PROMPT": "0", + }) + completed = subprocess.run( + [str(part) for part in command], + cwd=None if cwd is None else str(cwd), + env=environment, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + self.assertEqual(completed.returncode, 0, completed.stderr.decode("utf-8", "replace")) + return completed.stdout if stdout else completed.stdout.decode().strip() + + def git(self, *arguments, cwd=None): + return self.command("git", *arguments, cwd=cwd) + + def reserve(self, beat=1, arguments=("tests.test_app",), exit_code=1): + return attested_proof.request( + self.session_root, beat, self.profile, list(arguments), exit_code + ) + + def evidence(self, request_value, *, stdout=b"", stderr=FAILING, signed_exit=None, + name=None): + """The six files a gateway hands back, signed the way the broker signs them.""" + temporary = tempfile.TemporaryDirectory(prefix="underwrite-proof-ws-") + self.addCleanup(temporary.cleanup) + workspace = Path(temporary.name) / "workspace" + source = artifacts.verify_source_bundle( + self.store.read_object_bundle(), + self.store.frozen_target(), + workspace, + maximum_workspace_bytes=self.workspace_limit, + ) + built = artifacts.build_output_bundle( + workspace, + maximum_workspace_bytes=self.workspace_limit, + maximum_bundle_bytes=self.workspace_limit * 2, + ) + expected = { + "signerId": self.profile["signerId"], + "executorId": self.profile["executorId"], + "sessionId": request_value["sessionId"], + "challenge": request_value["challenge"], + "target": copy.deepcopy(request_value["target"]), + "action": copy.deepcopy(request_value["action"]), + "job": copy.deepcopy(request_value["job"]), + "inputTree": source.git_tree, + "sandbox": copy.deepcopy(self.profile["sandbox"]), + "outputTree": built.descriptor.git_tree, + "outputBundle": { + "sha256": built.descriptor.sha256, + "bytes": built.descriptor.bytes, + }, + "stdout": { + "sha256": hashlib.sha256(stdout).hexdigest(), + "bytes": len(stdout), + "truncated": False, + }, + "stderr": { + "sha256": hashlib.sha256(stderr).hexdigest(), + "bytes": len(stderr), + "truncated": False, + }, + "exitCode": request_value["exitCode"] if signed_exit is None else signed_exit, + } + capability_payload = execution_receipt.build_host_capability_payload( + { + key: expected[key] + for key in ( + "signerId", "executorId", "sessionId", "challenge", "target", + "action", "job", "inputTree", "sandbox", + ) + }, + self.now - timedelta(seconds=5), + self.now + timedelta(minutes=4), + ) + receipt_payload = execution_receipt.build_execution_receipt_payload( + capability_payload, + expected, + self.now - timedelta(seconds=4), + self.now - timedelta(seconds=3), + ) + directory = self.root / (name or f"evidence-{request_value['action']['seq']}") + directory.mkdir() + values = { + "request.json": canonical(request_value), + "capability.dsse.json": execution_receipt.build_dsse_envelope( + capability_payload, self.signer.key_id, self.signer.sign + ), + "receipt.dsse.json": execution_receipt.build_dsse_envelope( + receipt_payload, self.signer.key_id, self.signer.sign + ), + "output.bundle": built.bundle, + "stdout": stdout, + "stderr": stderr, + } + for file_name, data in values.items(): + (directory / file_name).write_bytes(data) + return directory + + def consume(self, proof, directory): + return attested_proof.consume( + self.session_root, proof, self.profile, self.public_key, directory, now=self.now + ) + + def verified(self, beat=1, arguments=("tests.test_app",), exit_code=1, **evidence): + attempt = self.reserve(beat, arguments, exit_code) + return self.consume(attempt["proof"], self.evidence(attempt["request"], **evidence)) + + def attested_beat(self, n=1, proof=1, state="clean", **extra): + beat = { + "n": n, + "tier": "core", + "state": state, + "proof_kind": "attested", + "attested_proof": proof, + "claim": "the new value fails the existing test", + "where": "app.py:1", + "slots": { + "what": "VALUE moves from 0 to 1", + "proof": "`python3 -m unittest tests.test_app` exits 1", + }, + } + beat.update(extra) + return beat + + +class ReservingAProof(AttestedProofCase): + def test_the_request_is_the_profile_command_with_the_walk_arguments_after_it(self): + attempt = self.reserve(exit_code=1) + + job = attempt["request"]["job"] + self.assertEqual( + job["argv"], self.profile["job"]["argv"] + ["tests.test_app"] + ) + for name in ("cwd", "environment", "executable", "stdin"): + self.assertEqual(job[name], self.profile["job"][name]) + self.assertEqual(attempt["request"]["exitCode"], 1) + self.assertEqual(attempt["request"]["sandbox"], self.profile["sandbox"]) + self.assertEqual(attempt["request"]["target"], self.store.frozen_target()) + self.assertEqual(attempt["request"]["action"], {"seq": 1, "beat": 1, "attempt": 1}) + self.assertEqual(attempt["state"], "reserved") + self.assertNotIn("capability", attempt) + + def test_a_retry_returns_the_same_reservation_and_a_new_command_gets_its_own(self): + first = self.reserve() + again = self.reserve() + other = self.reserve(exit_code=0) + other_beat = self.reserve(beat=2) + + self.assertEqual(first, again) + self.assertEqual(other["proof"], 2) + self.assertEqual(other_beat["proof"], 3) + self.assertNotEqual(first["challenge"], other["challenge"]) + + def test_only_a_definite_failure_makes_room_for_a_fresh_attempt(self): + first = self.reserve() + failed = attested_proof.fail( + self.session_root, first["proof"], "sandbox exit code was not the trusted expected result" + ) + retried = self.reserve() + + self.assertEqual(failed["state"], "failed") + self.assertEqual(retried["proof"], 2) + self.assertNotEqual(retried["challenge"], first["challenge"]) + with self.assertRaisesRegex(Conflict, "different reason"): + attested_proof.fail(self.session_root, first["proof"], "something else") + + def test_arguments_and_exit_codes_are_bounded(self): + cases = { + "none": ([], 1), + "too many": (["x"] * 17, 1), + "empty": ([""], 1), + "control": (["a\nb"], 1), + "long": (["a" * 513], 1), + "exit high": (["x"], 256), + "exit negative": (["x"], -1), + } + for label, (arguments, exit_code) in cases.items(): + with self.subTest(label=label): + with self.assertRaises(StoreError): + attested_proof.request( + self.session_root, 1, self.profile, arguments, exit_code + ) + + def test_a_session_without_a_frozen_pr_runs_no_proofs(self): + branch = SessionStore(self.root / "branch-session", create=True) + branch.put_session({"repo": "acme/widget", "audience": {"mode": "branch", "why": "local"}}) + + with self.assertRaisesRegex(Conflict, "frozen PR target"): + branch.reserve_proof(1, self.profile, ["tests"], 0) + + +class ConsumingEvidence(AttestedProofCase): + def test_signed_evidence_verifies_and_keeps_what_the_run_printed(self): + attempt = self.verified(stdout=b"", stderr=FAILING) + + self.assertEqual(attempt["state"], "verified") + self.assertEqual(attempt["evidence"]["exitCode"], 1) + self.assertEqual( + attempt["evidence"]["receiptSha256"], + hashlib.sha256( + (self.root / "evidence-1" / "receipt.dsse.json").read_bytes() + ).hexdigest(), + ) + self.assertIn("AssertionError: 1 != 2", attempt["output"]["stderr"]) + persisted = self.session_root / "proof-evidence" / "1" + self.assertEqual( + sorted(path.name for path in persisted.iterdir()), + ["capability.dsse.json", "receipt.dsse.json", "request.json", "stderr", "stdout"], + ) + + def test_an_identical_replay_is_accepted_and_changed_bytes_are_not(self): + attempt = self.reserve() + directory = self.evidence(attempt["request"]) + first = self.consume(attempt["proof"], directory) + self.assertEqual(self.consume(attempt["proof"], directory), first) + + (directory / "stderr").write_bytes(FAILING + b"tampered\n") + with self.assertRaisesRegex(Conflict, "stderr"): + self.consume(attempt["proof"], directory) + + def test_a_receipt_for_another_exit_code_does_not_verify(self): + attempt = self.reserve(exit_code=1) + directory = self.evidence(attempt["request"], signed_exit=0) + + with self.assertRaises(execution_receipt.ReceiptError): + self.consume(attempt["proof"], directory) + self.assertEqual(self.store.proof_attempt(attempt["proof"])["state"], "reserved") + + def test_evidence_must_be_for_the_named_proof_and_the_pinned_key(self): + first = self.reserve() + second = self.reserve(exit_code=0) + directory = self.evidence(second["request"]) + with self.assertRaisesRegex(Conflict, "not proof 1"): + self.consume(first["proof"], directory) + + other_private = self.root / "other.pem" + other_public = self.root / "other-public.pem" + self.command( + "openssl", "genpkey", "-algorithm", "EC", + "-pkeyopt", "ec_paramgen_curve:P-256", + "-pkeyopt", "ec_param_enc:named_curve", + "-out", other_private, + ) + other_public.write_bytes( + self.command("openssl", "pkey", "-in", other_private, "-pubout", stdout=True) + ) + with self.assertRaisesRegex(Conflict, "key ID"): + attested_proof.consume( + self.session_root, second["proof"], self.profile, other_public, + directory, now=self.now, + ) + + def test_a_changed_profile_after_reservation_is_refused(self): + attempt = self.reserve() + directory = self.evidence(attempt["request"]) + changed = copy.deepcopy(self.profile) + changed["sandbox"]["limits"]["memoryBytes"] = 128 * 1024 * 1024 + + with self.assertRaisesRegex(Conflict, "profile changed"): + attested_proof.consume( + self.session_root, attempt["proof"], changed, self.public_key, + directory, now=self.now, + ) + + +class AttestedBeats(AttestedProofCase): + def test_a_verified_proof_earns_clean_and_the_store_writes_what_ran(self): + self.verified() + + stored = self.store.put_beat(self.attested_beat()) + + self.assertEqual(stored["state"], "clean") + attestation = stored["attestation"] + self.assertEqual(attestation["proof"], 1) + self.assertEqual(attestation["argv"], self.profile["job"]["argv"] + ["tests.test_app"]) + self.assertEqual(attestation["exit_code"], 1) + self.assertEqual(attestation["key_id"], self.signer.key_id) + self.assertIn("AssertionError", attestation["output"]["stderr"]) + self.assertEqual( + self.store.put_beat(copy.deepcopy(stored))["attestation"], attestation + ) + + def test_an_attested_flag_carries_its_receipt_too(self): + self.verified() + + stored = self.store.put_beat(self.attested_beat( + state="flag", + slots={ + "what": "VALUE moves from 0 to 1", + "proof": "`python3 -m unittest tests.test_app` exits 1", + "risk": "the shipped default breaks every caller", + "fix": "keep VALUE at 0", + }, + decision={ + "question": "Include the failing test in the review, or drop it?", + "recommendation": "Include it: the gateway ran the test and it failed.", + }, + )) + + self.assertEqual(stored["state"], "flag") + self.assertEqual(stored["attestation"]["exit_code"], 1) + + def test_no_verified_receipt_no_attested_proof(self): + self.reserve() + with self.assertRaisesRegex(Conflict, "reserved, not verified"): + self.store.put_beat(self.attested_beat()) + with self.assertRaisesRegex(StoreError, "no proof 9"): + self.store.put_beat(self.attested_beat(proof=9)) + with self.assertRaisesRegex(StoreError, "names no attested_proof"): + beat = self.attested_beat() + del beat["attested_proof"] + self.store.put_beat(beat) + + def test_a_proof_backs_only_the_beat_it_was_run_for(self): + self.verified(beat=2) + + with self.assertRaisesRegex(Conflict, "run for beat 2"): + self.store.put_beat(self.attested_beat(n=1, proof=1)) + + def test_the_walk_cannot_write_the_attestation_itself(self): + self.verified() + beat = self.attested_beat() + beat["attestation"] = {"proof": 1, "exit_code": 0} + with self.assertRaisesRegex(StoreError, "store-owned field attestation"): + self.store.put_beat(beat) + + def test_a_proof_number_on_another_proof_kind_is_refused(self): + self.verified() + beat = self.attested_beat(state="unverified", proof_kind="read") + + with self.assertRaisesRegex(StoreError, "attested_proof is set"): + self.store.put_beat(beat) + + def test_a_branch_walk_writes_ran_not_attested(self): + branch = SessionStore(self.root / "branch-session", create=True) + branch.put_session({"repo": "acme/widget", "audience": {"mode": "branch", "why": "local"}}) + + with self.assertRaisesRegex(Conflict, "frozen PR target"): + branch.put_beat(self.attested_beat()) + + def test_the_page_shows_the_signed_command_exit_and_output(self): + self.verified() + self.store.put_beat(self.attested_beat()) + session, beats = self.store.presentation_snapshot()[:2] + beat = beats[0] + + card = render_report.beat_html(beat, render_report.validate(beat), True) + self.assertIn("Ran in sandbox", card) + self.assertIn("exited 1, as the claim predicted", card) + self.assertIn("/usr/bin/python3 -I -B -m unittest tests.test_app", card) + self.assertIn("AssertionError: 1 != 2", card) + self.assertNotIn("unproven", card) + _label, _text, limits, _owed = render_report.assessment(session, beats, {}) + self.assertIn( + "1 item ran in the gateway sandbox with a signed receipt; " + "everything else is static inspection.", + limits, + ) + + def test_an_attested_beat_without_its_attestation_renders_unproven(self): + beat = self.attested_beat() + + self.assertIn( + "beat 1: attested proof carries no gateway attestation", + render_report.validate(beat), + ) + + +class OneCallProof(AttestedProofCase): + """`prove` reserves, hands the request to the host adapter, and consumes.""" + + def adapter(self, script): + path = self.root / "adapter.py" + path.write_text(textwrap.dedent(script), encoding="utf-8") + outbox = self.root / "outbox" + outbox.mkdir(exist_ok=True) + config = self.root / "proof.json" + profile = self.root / "profile.json" + profile.write_bytes(canonical(self.profile)) + config.write_bytes(canonical({ + "version": 1, + "profile": str(profile), + "publicKey": str(self.public_key), + "adapter": [sys.executable, str(path)], + "evidenceRoot": str(outbox), + })) + return config + + def test_a_complete_adapter_run_verifies_the_proof(self): + attempt = self.reserve() + prepared = self.evidence(attempt["request"], name="prepared") + config = self.adapter(f""" + import json, shutil, sys + args = dict(zip(sys.argv[1::2], sys.argv[2::2])) + shutil.copytree({str(prepared)!r}, args["--evidence-dir"]) + print(json.dumps({{"version": 1, "status": "complete", + "evidenceDir": args["--evidence-dir"]}})) + """) + + result = attested_proof.prove( + self.session_root, 1, config, ["tests.test_app"], 1, now=self.now + ) + + self.assertEqual(result["status"], "verified") + self.assertEqual(result["proof"]["proof"], attempt["proof"]) + self.assertEqual(self.store.proof_attempt(1)["state"], "verified") + + def test_a_definite_gateway_failure_records_why(self): + config = self.adapter(""" + import json, sys + print(json.dumps({"version": 1, "status": "failed", + "reason": "gateway attempt is terminal without a receipt"})) + print("adapter.py: sandbox exit code was not the trusted expected result", + file=sys.stderr) + sys.exit(2) + """) + + result = attested_proof.prove(self.session_root, 1, config, ["tests.test_app"], 0) + + self.assertEqual(result["status"], "failed") + self.assertEqual( + result["proof"]["failure"], + "adapter.py: sandbox exit code was not the trusted expected result", + ) + self.assertEqual(self.store.proof_attempt(1)["state"], "failed") + + def test_an_unknown_outcome_keeps_the_reservation_for_a_retry(self): + config = self.adapter(""" + import sys + print("not json") + sys.exit(1) + """) + + result = attested_proof.prove(self.session_root, 1, config, ["tests.test_app"], 0) + + self.assertEqual(result["status"], "retry") + self.assertEqual(self.store.proof_attempt(1)["state"], "reserved") + + def test_the_configuration_has_one_exact_shape(self): + config = self.adapter("print('unused')") + value = json.loads(config.read_text()) + value["adapter"] = ["relative/adapter.py"] + config.write_bytes(canonical(value)) + + with self.assertRaisesRegex(attested_proof.ProofError, "absolute"): + attested_proof.prove(self.session_root, 1, config, ["tests.test_app"], 0) + + +class AttestedReportFindings(AttestedProofCase): + merged = True + + def test_an_included_finding_keeps_the_proof_it_was_included_on(self): + self.verified() + self.verified(exit_code=0, stderr=b"OK\n", name="evidence-second") + self.store.put_beat(self.attested_beat( + state="flag", + slots={ + "what": "VALUE moves from 0 to 1", + "proof": "`python3 -m unittest tests.test_app` exits 1", + "risk": "the shipped default breaks every caller", + "fix": "keep VALUE at 0", + }, + decision={ + "question": "Include the failing test in the report, or drop it?", + "recommendation": "Include it: the gateway ran the test and it failed.", + }, + )) + self.assertEqual(self.store.snapshot()[0]["audience"]["mode"], "report") + accepted = self.store.produce("include-1", 1, "accept", "include it") + self.store.ack(accepted["seq"]) + current = self.store.snapshot()[1][0] + self.assertEqual(current["state"], "accepted") + + swapped = copy.deepcopy(current) + swapped["attested_proof"] = 2 + with self.assertRaisesRegex(Conflict, "cannot change attested_proof"): + self.store.put_beat(swapped) + self.assertEqual(self.store.put_beat(current)["attestation"]["proof"], 1) + + +class Upgrading(AttestedProofCase): + def test_a_version_7_session_gains_the_proof_table_and_keeps_its_state(self): + import sqlite3 + + before = self.store.snapshot() + db = sqlite3.connect(str(self.store.path)) + try: + db.execute("DROP TABLE proof_attempts") + db.execute("PRAGMA user_version = 7") + db.commit() + finally: + db.close() + + reopened = SessionStore(self.session_root) + + self.assertEqual(reopened.snapshot(), before) + self.assertEqual(reopened.proof_attempts(), []) + db = sqlite3.connect(str(self.store.path)) + try: + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) + finally: + db.close() + + +class CommandLine(AttestedProofCase): + def proofctl(self, *arguments): + return subprocess.run( + [sys.executable, str(SCRIPTS / "proofctl.py"), *map(str, arguments)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + + def test_request_takes_dashed_arguments_after_a_separator(self): + profile = self.root / "profile.json" + profile.write_bytes(canonical(self.profile)) + + completed = self.proofctl( + "request", self.session_root, 1, profile, "--exit-code", "1", + "--", "tests.test_app", "-k", "value", + ) + + self.assertEqual(completed.returncode, 0, completed.stderr.decode()) + attempt = json.loads(completed.stdout) + self.assertEqual( + attempt["request"]["job"]["argv"][-3:], ["tests.test_app", "-k", "value"] + ) + listed = json.loads(self.proofctl("proofs", self.session_root).stdout) + self.assertEqual([item["proof"] for item in listed["proofs"]], [1]) + + def test_a_refusal_exits_one_with_the_reason(self): + profile = self.root / "profile.json" + profile.write_bytes(canonical(self.profile)) + + completed = self.proofctl( + "request", self.session_root, 1, profile, "--exit-code", "300", "tests" + ) + + self.assertEqual(completed.returncode, 1) + self.assertIn("proofctl: proof exit code must not exceed 255", completed.stderr.decode()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_gateway_adapter.py b/tests/test_gateway_adapter.py index 663045f..2e0e777 100644 --- a/tests/test_gateway_adapter.py +++ b/tests/test_gateway_adapter.py @@ -894,5 +894,68 @@ def test_cli_reports_complete_retry_and_definite_failure_as_json(self): self.assertEqual(result["status"], "retry") +class ProofDeploymentTests(GatewayAdapterCase): + """A proof deployment lets a request extend the profile's command and predict its + exit code. Everything else stays pinned, and without the opt-in nothing changes.""" + + def proof_request(self, *arguments, exit_code=1): + request = copy.deepcopy(self.request) + request["job"]["argv"] = list(self.profile["job"]["argv"]) + list(arguments) + request["exitCode"] = exit_code + return request + + def check(self, request, proof_arguments, name): + path = self.root / name + path.write_bytes(canonical(request)) + path.chmod(0o600) + policy = adapter._policy(self.configuration) + return adapter._request(str(path), policy, self.profile, proof_arguments) + + def test_the_opt_in_is_a_bounded_integer(self): + accepted = copy.deepcopy(self.configuration) + accepted["proofArguments"] = 4 + loaded = adapter.load_configuration( + str(self.write_configuration(accepted, name="proof.json")) + ) + self.assertEqual(loaded["proofArguments"], 4) + for index, value in enumerate((0, 17, "4", True, None)): + with self.subTest(value=value): + refused = copy.deepcopy(self.configuration) + refused["proofArguments"] = value + path = self.write_configuration(refused, name=f"refused-{index}.json") + with self.assertRaises(adapter.AdapterError): + adapter.load_configuration(str(path)) + + def test_a_proof_deployment_accepts_a_tail_and_a_predicted_exit(self): + request = self.proof_request("tests.test_app", exit_code=1) + self.assertEqual(self.check(request, 4, "proof-1.json"), request) + + def test_without_the_opt_in_a_tail_is_refused(self): + request = self.proof_request("tests.test_app", exit_code=0) + with self.assertRaisesRegex(adapter.AdapterError, "job does not match"): + self.check(request, None, "plain.json") + + def test_a_proof_cannot_change_what_the_profile_pins(self): + cases = {} + cases["too many"] = self.proof_request("a", "b", "c", exit_code=1) + changed_prefix = self.proof_request("x", exit_code=1) + changed_prefix["job"]["argv"][1] = "-c" + cases["prefix"] = changed_prefix + environment = self.proof_request("x", exit_code=1) + environment["job"]["environment"]["PYTHONPATH"] = "/tmp" + cases["environment"] = environment + cases["control"] = self.proof_request("a\nb", exit_code=1) + cases["long"] = self.proof_request("a" * 513, exit_code=1) + cases["empty"] = self.proof_request("", exit_code=1) + cases["no tail, other exit"] = self.proof_request(exit_code=1) + for index, (label, request) in enumerate(cases.items()): + with self.subTest(label=label): + with self.assertRaises((adapter.AdapterError, adapter.GatewayError)): + self.check(request, 2, f"refused-request-{index}.json") + + def test_a_proof_deployment_still_runs_the_exact_profile_job(self): + self.assertEqual(self.check(self.request, 2, "exact.json"), self.request) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_gateway_live.py b/tests/test_gateway_live.py index 0d02c25..33c273a 100644 --- a/tests/test_gateway_live.py +++ b/tests/test_gateway_live.py @@ -20,6 +20,10 @@ from gateway.signing import OpenSSLSigner from skills.underwrite.scripts.execution_receipt import verify_execution_receipt +SCRIPTS = Path(__file__).parents[1] / "skills" / "underwrite" / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) + IMAGE_ENV = "UNDERWRITE_LIVE_GATEWAY_IMAGE" PLATFORM_ENV = "UNDERWRITE_LIVE_GATEWAY_PLATFORM" @@ -513,6 +517,212 @@ def stream(value): ) self.assertEqual(remaining, b"") + def proof_session(self, root): + """A captured PR whose head breaks the test its base shipped.""" + from pr_snapshot import capture + from session_store import SessionStore + + remote = root / "upstream.git" + controller = root / "controller" + self.command([self.git, "init", "--bare", "--template=", remote]) + self.command([self.git, "init", "--template=", controller]) + for name, value in ( + ("user.name", "Underwrite Live Test"), + ("user.email", "live@underwrite.invalid"), + ("commit.gpgsign", "false"), + ): + self.command([self.git, "config", name, value], cwd=controller) + (controller / "tests").mkdir() + (controller / "app.py").write_text("VALUE = 0\n", encoding="utf-8") + (controller / "tests" / "__init__.py").write_text("", encoding="utf-8") + (controller / "tests" / "test_app.py").write_text( + "import unittest\n\nimport app\n\n\n" + "class AppTest(unittest.TestCase):\n" + " def test_value(self):\n" + " self.assertEqual(app.VALUE, 0)\n", + encoding="utf-8", + ) + self.command([self.git, "add", "."], cwd=controller) + self.command([self.git, "commit", "-m", "base"], cwd=controller) + self.command([self.git, "branch", "-M", "main"], cwd=controller) + self.command([self.git, "remote", "add", "origin", remote], cwd=controller) + self.command([self.git, "push", "origin", "main"], cwd=controller) + base = self.command([self.git, "rev-parse", "HEAD"], cwd=controller).decode() + self.command([self.git, "checkout", "-b", "feature"], cwd=controller) + (controller / "app.py").write_text("VALUE = 1\n", encoding="utf-8") + self.command([self.git, "commit", "-am", "head"], cwd=controller) + head = self.command([self.git, "rev-parse", "HEAD"], cwd=controller).decode() + self.command( + [self.git, "push", "origin", "HEAD:refs/pull/7/head"], cwd=controller + ) + self.command([self.git, "checkout", "main"], cwd=controller) + metadata = { + "number": 7, + "state": "open", + "merged_at": None, + "changed_files": 1, + "base": { + "ref": "main", + "sha": base, + "repo": {"full_name": "acme/widget", "clone_url": str(remote)}, + }, + "head": { + "ref": "feature", + "sha": head, + "repo": {"id": 123, "full_name": "acme/widget"}, + }, + } + session_root = root / "session" + store = SessionStore(session_root, create=True) + capture(store, "acme/widget", 7, controller, api=lambda _r, _n: dict(metadata)) + return session_root, store + + def test_a_proof_runs_the_frozen_head_in_the_sandbox_and_signs_only_its_prediction(self): + import attested_proof + + with tempfile.TemporaryDirectory(dir="/tmp") as name: + root = Path(name).resolve() + os.chown(root, -1, os.getegid()) + root.chmod(0o710) + container_name = runtime_container_name(RUNTIME_DOMAIN_ID) + self.remove_container(container_name) + self.addCleanup(self.remove_container, container_name) + session_root, store = self.proof_session(root) + private_key = root / "private.pem" + public_key = root / "public.pem" + self.command([ + self.openssl, "genpkey", "-algorithm", "EC", + "-pkeyopt", "ec_paramgen_curve:P-256", + "-pkeyopt", "ec_param_enc:named_curve", + "-out", private_key, + ]) + private_key.chmod(0o600) + public_key.write_bytes( + self.command([self.openssl, "pkey", "-in", private_key, "-pubout"]) + b"\n" + ) + signer_id = "https://runner.example/hosts/live-proof" + signer = OpenSSLSigner(private_key, public_key, signer_id, openssl=self.openssl) + environment = { + "LANG": "C", + "LC_ALL": "C", + "PATH": "/usr/local/bin:/usr/bin:/bin", + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONUTF8": "1", + } + sandbox = self.sandbox(self.limits(wallSeconds=20, cpuSeconds=10)) + job = { + "argv": [self.executable["path"], "-I", "-B", "-m", "unittest"], + "cwd": ".", + "environment": environment, + "executable": self.executable, + "stdin": "closed", + } + policy = GatewayPolicy( + signer_id=signer_id, + deployment_id="urn:underwrite:gateway:live-proof", + runtime_domain_id=RUNTIME_DOMAIN_ID, + docker_host=self.docker_host, + image=self.image, + platform=self.platform, + runner_sha256=self.runner_sha256, + executable=self.executable, + environment=environment, + sandbox=sandbox, + target_uid=65532, + target_gid=65532, + ) + profile = { + "version": 1, + "keyId": signer.key_id, + "signerId": signer_id, + "executorId": policy.executor_id, + "job": job, + "sandbox": sandbox, + "exitCode": 0, + } + store_root = root / "store" + outbox = root / "outbox" + store_root.mkdir(mode=0o700) + outbox.mkdir(mode=0o710) + + def canonical(value): + return json.dumps( + value, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + + gateway_config = root / "gateway.json" + gateway_config.write_bytes(canonical({ + "version": 1, + "privateKey": str(private_key), + "publicKey": str(public_key), + "storeRoot": str(store_root), + "docker": self.docker, + "openssl": self.openssl, + "git": self.git, + "deploymentId": policy.deployment_id, + "runtimeDomainId": policy.runtime_domain_id, + "dockerHost": self.docker_host, + "image": self.image, + "platform": self.platform, + "runnerSha256": self.runner_sha256, + "targetUid": 65532, + "targetGid": 65532, + "consumerGid": os.getegid(), + "capabilitySeconds": 300, + "maxSourceBundleBytes": 64 * 1024 * 1024, + "profile": profile, + "proofArguments": 8, + })) + gateway_config.chmod(0o600) + profile_path = root / "profile.json" + profile_path.write_bytes(canonical(profile)) + proof_config = root / "proof.json" + proof_config.write_bytes(canonical({ + "version": 1, + "profile": str(profile_path), + "publicKey": str(public_key), + "adapter": [ + sys.executable, "-I", "-S", str(ADAPTER_PATH), + "--config", str(gateway_config), + ], + "evidenceRoot": str(outbox), + })) + arguments = ["discover", "-s", "tests", "-t", "."] + + wrong = attested_proof.prove(session_root, 1, proof_config, arguments, 0) + + self.assertEqual(wrong["status"], "failed", wrong) + self.assertIn("exit code", wrong["proof"]["failure"]) + self.assertIsNone(wrong["proof"]["evidence"]) + + right = attested_proof.prove(session_root, 1, proof_config, arguments, 1) + + self.assertEqual(right["status"], "verified", right) + self.assertEqual(right["proof"]["evidence"]["exitCode"], 1) + self.assertIn("AssertionError: 1 != 0", right["proof"]["output"]["stderr"]) + beat = store.put_beat({ + "n": 1, + "tier": "core", + "state": "flag", + "proof_kind": "attested", + "attested_proof": right["proof"]["proof"], + "claim": "the new value breaks the shipped test", + "where": "app.py:1", + "slots": { + "what": "VALUE moves from 0 to 1", + "proof": "`python3 -m unittest discover -s tests -t .` exits 1", + "risk": "every caller relying on 0 changes behavior", + "fix": "keep VALUE at 0 or update the test with intent", + }, + "decision": { + "question": "Include the failing test in the review, or drop it?", + "recommendation": "Include it: the sandbox ran the test and it failed.", + }, + }) + self.assertEqual(beat["attestation"]["exit_code"], 1) + self.assertEqual(beat["attestation"]["argv"], job["argv"] + arguments) + self.assertEqual(store.snapshot()[0]["execution_policy"]["mode"], "no_exec") + def test_clean_job_has_only_the_declared_surface_and_round_trips_output(self): environment = { "LANG": "C", diff --git a/tests/test_session_store.py b/tests/test_session_store.py index e8d9d81..33a665e 100644 --- a/tests/test_session_store.py +++ b/tests/test_session_store.py @@ -402,7 +402,7 @@ def test_v6_migration_preserves_actions_canonical_state_and_replay(self): self.store = session_store.SessionStore(self.root) - self.assertEqual(self.database_state()[0], 7) + self.assertEqual(self.database_state()[0], 8) self.assertEqual(self.database_state()[2], actions) self.assertEqual(self.store.snapshot(), before) self.assertEqual(self.store.head(), pending) @@ -433,7 +433,7 @@ def fail(store, db): self.assertEqual(self.database_state(), before) restored = session_store.SessionStore(self.root) self.assertEqual(restored.head()["note"], "Reviewer words.") - self.assertEqual(self.database_state()[0], 7) + self.assertEqual(self.database_state()[0], 8) def test_v6_migration_rejects_orphaned_references_without_advancing_version(self): self.downgrade(self.store) @@ -2040,7 +2040,7 @@ def downgrade_to_pre_identity_v1(self, format_version=1): def test_the_database_and_export_format_are_versioned(self): with sqlite3.connect(str(self.root / "session.sqlite3")) as db: - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 7) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) self.assertEqual(db.execute("PRAGMA journal_mode").fetchone()[0], "delete") db = self.store._connect() try: @@ -2060,7 +2060,7 @@ def test_v3_non_pr_sessions_do_not_gain_a_pr_execution_policy(self): self.assertNotIn("execution_policy", upgraded.snapshot()[0]) with sqlite3.connect(str(self.root / "session.sqlite3")) as db: - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 7) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) def test_the_session_identity_survives_restart_but_not_recreation(self): first = self.store.delivery_state()["session_id"] @@ -2085,7 +2085,7 @@ def test_a_pre_identity_v1_database_upgrades_without_losing_state(self): self.assertEqual(upgraded.head()["action_id"], action["action_id"]) self.assertTrue(upgraded.delivery_state()["session_id"]) with sqlite3.connect(str(self.root / "session.sqlite3")) as db: - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 7) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) columns = {row[1] for row in db.execute("PRAGMA table_info(session)")} self.assertIn("session_id", columns) @@ -2107,7 +2107,7 @@ def test_a_v2_upgrade_restores_the_fix_approved_before_delivery_failed(self): failed = upgraded.presentation_snapshot()[1][0]["delivery"] self.assertEqual(failed["owed"], "retry with the fixture") with sqlite3.connect(str(self.root / "session.sqlite3")) as db: - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 7) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) def test_a_v2_upgrade_strips_a_failure_injected_fix_written_in_upper_case(self): """The restore reads the six lowercase keys, so the slot repair has to run before @@ -2271,7 +2271,7 @@ def test_an_upgrade_with_nothing_to_repair_leaves_the_exports_untouched(self): self.assertEqual({path: path.stat().st_mtime_ns for path in derived}, before) with sqlite3.connect(str(self.root / "session.sqlite3")) as db: - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 7) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) def test_a_current_session_opens_in_a_directory_it_cannot_write(self): """Reading a published review out of a directory someone locked down.""" @@ -2396,13 +2396,13 @@ def test_a_future_database_is_refused_without_changing_its_journal_mode(self): (self.root / "session.sqlite3").unlink() with sqlite3.connect(str(self.root / "session.sqlite3")) as db: self.assertEqual(db.execute("PRAGMA journal_mode = WAL").fetchone()[0], "wal") - db.execute("PRAGMA user_version = 8") + db.execute("PRAGMA user_version = 9") with self.assertRaisesRegex(session_store.StoreError, "newer than supported"): session_store.SessionStore(self.root) with sqlite3.connect(str(self.root / "session.sqlite3")) as db: - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 9) self.assertEqual(db.execute("PRAGMA journal_mode").fetchone()[0], "wal") def test_legacy_actions_without_an_ack_migrate_as_handled(self): @@ -3110,7 +3110,7 @@ def test_a_beat_stored_before_proof_kinds_keeps_its_verdict(self): store.put_beat(dict(before, proof_kind="read")) self.assertEqual( session_store.validate_beat(dict(stored, proof_kind="checked")), - ["beat 1: proof_kind 'checked' is not one of ran, read, inferred"], + ["beat 1: proof_kind 'checked' is not one of ran, read, inferred, attested"], ) def test_a_refusal_inside_apply_leaves_the_action_replayable(self): @@ -4240,7 +4240,7 @@ def test_v4_upgrade_adds_authority_tables_without_changing_session_state(self): "SELECT name FROM sqlite_master WHERE type = 'table'" ) } - self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 7) + self.assertEqual(db.execute("PRAGMA user_version").fetchone()[0], 8) self.assertIn("implementation_links", names) self.assertIn("implementation_attempts", names) diff --git a/tests/test_skill.py b/tests/test_skill.py index c54dfcf..d9eb7dd 100644 --- a/tests/test_skill.py +++ b/tests/test_skill.py @@ -125,9 +125,25 @@ def readme(self): def test_only_a_proof_that_ran_earns_clean(self): text = self.skill() - self.assertIn("Only `ran` earns `CLEAN`", text) + self.assertIn("Only `ran` and `attested` earn `CLEAN`", text) self.assertIn("`UNVERIFIED` is not a failure", text) - self.assertIn("`proof_kind` is `ran`, `read`, or `inferred` on every beat written now", text) + self.assertIn( + "`proof_kind` is `ran`, `read`, `inferred`, or `attested` on every beat written now", + text, + ) + + def test_an_attested_proof_is_a_signed_prediction_the_store_holds(self): + text = self.skill() + self.assertIn("It signs a receipt only when the command exits with the code you predicted", text) + self.assertIn("$S/scripts/proofctl.py prove \"$R\" \"$G\" --exit-code 1 --", text) + self.assertIn("Never write `attestation`", text) + self.assertIn("A wrong prediction is information, not evidence", text) + self.assertIn("An exit code is coarse", text) + self.assertIn("cite it only when that output shows your claim", text) + self.assertIn("never `ran` for something you did not run", text) + self.assertIn("this machine never executes the head", text) + readme = self.readme() + self.assertIn("A proof never changes the session's `no_exec` policy", readme) def test_a_risk_is_a_flag_or_nothing(self): self.assertIn("A `RISK` belongs to a flag", self.skill())