diff --git a/.gitignore b/.gitignore index 444f5ee3..c1edadd4 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,8 @@ Thumbs.db # Docker *.log .vercel/ + +# Agent tooling scratch: throwaway git worktrees used to run checks in +# isolation. Full repo copies (including .env.production and coverage +# artifacts) โ€” never commit these. +.claude/worktrees/ diff --git a/CHANGELOG.md b/CHANGELOG.md index dc49e2c1..1c19954a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,20 @@ full release gate; do not backfill a final `v1.2.0` tag. ### ๐Ÿ”ฌ Competitive research, and the positioning correction it forced +- **The interop question now has running code.** + `examples/acta_receipt_interop.py` (plus `tests/test_acta_receipt_interop.py`, + 8 tests) transcodes a signed portable receipt into the ScopeBlind ACTA + receipt envelope as a custom `agentmiddleware:governed_invoke` type โ€” inner + production signature verified over the exact `signing_input` bytes, outer + Ed25519-over-JCS envelope, economic linkage as first-class fields, floats + and unverified or linkage-less bundles refused. Deliberately an experiment + in `examples/`, not a product surface; shipping it remains a product call. +- **The demand question now has an instrument.** + `docs/partner-interview-script.md`: a past-tense-only discovery script with + an incident-first structure, a commitment test priced in engineer-afternoons + rather than opinions, a per-call scoring table, and a decision rule + committed before the interviews so five calls produce a verdict. + - **Second research pass (2026-08-15) answered two of the open questions.** The protect-mcp receipt draft (`draft-farley-acta-signed-receipts`) was read directly: it is an individual diff --git a/docs/market-research-2026-08.md b/docs/market-research-2026-08.md index f618e52a..7c853050 100644 --- a/docs/market-research-2026-08.md +++ b/docs/market-research-2026-08.md @@ -237,6 +237,10 @@ built from ยง3 and ยง4. conversation settles that. 5. Would a buyer with an EU AI Act mandate reject a vendor with no compliance mapping outright, or accept receipts as one input among several? +6. Does the demand exist? The instrument for answering is now written: + [`partner-interview-script.md`](partner-interview-script.md), with a + pre-committed decision rule so five interviews produce a verdict rather + than an argument. --- @@ -308,10 +312,17 @@ thread's author fixed it with idempotency keys and moved on. documents. - **The practitioner quote is the receipt argument verbatim**: "nothing in the transcript says so." Candidate for `/compare/` or the pitch, cited. -- **Interop vs. compete is now a live, cheap experiment**: one serializer - emitting `agentmiddleware:*` receipts in the draft's envelope, offline - verification included. Whether to run it is a product call; the cost side is - no longer unknown. +- **Interop vs. compete is now a live, cheap experiment โ€” and the experiment + has been run.** `examples/acta_receipt_interop.py` transcodes the published + proof receipt into an `agentmiddleware:governed_invoke` receipt in the + draft's envelope (JCS canonicalization, Ed25519, `issuer_id`=`kid`), with + the original signed bundle embedded and independently verifiable โ€” the + signature covers the exact `signing_input` bytes, and the wrapper's + `issuer` label is checked against the issuer bound to the key material in + the trust-keys snapshot rather than trusted on its own โ€” and the + ledger/idempotency linkage carried as first-class fields. Cost measured: + ~200 lines plus tests, no trust-plane changes. Whether to *ship* it stays a + product call; the cost side is now a fact, not an estimate. - **ยง6 risk 1 escalates**: the receipt-format neighbour added spend evidence within two draft revisions. If a registry lands, economic receipt types become table stakes the way signatures did. diff --git a/docs/partner-interview-script.md b/docs/partner-interview-script.md new file mode 100644 index 00000000..8ff58881 --- /dev/null +++ b/docs/partner-interview-script.md @@ -0,0 +1,231 @@ +# Design-Partner Interview Script + +**Purpose:** answer the one question research cannot โ€” +[`market-research-2026-08.md`](market-research-2026-08.md) ยง7 question 4's +remainder: does anyone *pay* for charge-once semantics, or does everyone patch +their own tool layer and move on? + +**Posture:** this is a discovery interview, not a pitch. The rules below exist +because a founder-led interview drifts toward selling within minutes, and a +sold prospect tells you what you want to hear. `ELEVATOR_PITCH.md` already +commits us: *"Existing IAM, gateway, or logging controls may already be +sufficient; the first conversation must establish that they are not before +this product is proposed."* This script is that commitment made operational. + +## Rules (read before every call) + +1. **Past tense only.** Ask what happened, never "would you useโ€ฆ". A yes to a + hypothetical is worth nothing; a story about last March is evidence. +2. **The product does not exist until the reveal (ยง4, ~minute 25).** If + asked earlier, say "we're researching how teams handle agent tool-call + failures โ€” I'll show you what we're building at the end if you want." +3. **Chase money and named people, not opinions.** "Who noticed?" and "what + did it cost?" beat "was it bad?" +4. **Disqualify eagerly.** A clean "this isn't a problem for us" is a + successful interview. Record it with the same care as an enthusiastic one. +5. **Never supply the vocabulary.** Do not say idempotency, receipt, + exactly-once, or audit until they do. Note verbatim which words *they* + reach for โ€” that's the landing-page copy. ยง4 (the reveal) is the **only** + section where product vocabulary is permitted, and only after ยง1โ€“3 are + done. + +## Who to interview + +The qualifying shape from `ELEVATOR_PITCH.md`: platform/AI-infrastructure/ +security teams already running agents against internal tools where **one tool +call costs real money or causes a real side effect** (payments, ticketing, +provisioning, trades, outbound email). Framework users are the richest pool โ€” +the documented gaps are in LangGraph and CrewAI specifically +(`langchain-ai/langgraph#7417`, `crewAIInc/crewAI#5802`). + +Screen out (politely, early): teams whose agent tools are read-only or +idempotent by construction; teams that want an all-tools governance platform +(wrong product); anyone shopping for compliance certifications (send them to +the audit-layer vendors named on `/compare/`). + +--- + +## The script + +### 0. Framing (2 min) + +> "Thanks for the time. I'm researching how teams run AI agents against tools +> that have real side effects โ€” payments, tickets, emails, infrastructure. +> I'm not going to pitch you anything for the next 25 minutes; I want to hear +> what's actually happened in your stack. Some of it may be embarrassing โ€” +> that's the useful part." + +### 1. The incident question (10 min โ€” this is the interview) + +Open broad, so a story that doesn't fit our shape still gets told: + +> **"Tell me about the last time an agent tool call failed, timed out, or +> got retried."** + +Let them finish. Then, if the story didn't already cover it, the pointed +probe: + +> **"Has an agent ever retried something expensive โ€” and either double-did it, +> or left you unable to prove afterward what actually ran?"** + +If yes to either, excavate the story completely before moving on: + +- When? Which framework/orchestrator? Which tool fired twice? +- How did you *find out*? (Alerting? A customer? Reconciliation? Luck?) +- What did it cost โ€” money, hours, trust? Who outside engineering heard + about it? +- What was the ambiguous moment โ€” did you know the call was in flight, or did + the platform re-dispatch behind your back? + *(Listening for the `delivery_uncertain` shape: langgraph#7417 is exactly + this โ€” the platform silently re-ran calls users believed were still + running. If their story rhymes, note it, don't lead them to it.)* +- What did you change afterward? Show me, roughly โ€” decorator? Redis dedup? + Manual runbook? + +If no, probe once, gently, then believe them: + +> "How would you know if it had happened? Walk me through what your logs +> would show for a call that timed out and got retried." + +*(A team that cannot answer this may have the problem invisibly โ€” the +practitioner quote from the CrewAI thread was "two tickets, two emails, two +rows, and nothing in the transcript says so." Note whether their answer is +confident or uncomfortable. Do not quote the thread at them.)* + +### 2. The current-mitigation question (7 min) + +Whatever they built, take it seriously โ€” it is the real competitor. Refer to +it in *their* words from ยง1 (rule 5 still applies here โ€” if they called it +"the dedup thing," so do you): + +- "What, if anything, did you change after the incident?" + *(A real answer may be "nothing" โ€” record `none` for the mitigation field + and move on.)* If they describe a safeguard: "How long did that take? + Who maintains it? Does it survive a worker restart / a second process?" +- "Have you had to prove to anyone *outside* the team what an agent did and + what it cost? Who asked? What did you show them? Did they accept it?" + *(This is the build-vs-buy fork from `/compare/`: an in-process cache is a + reliability fix; the moment a finance owner, auditor, or customer needs to + check the record, evidence becomes the product. If nobody outside eng ever + asks, our wedge may not apply to them โ€” record that honestly.)* +- "What does a tool-call record look like in your system today? Could someone + alter it after the fact? Does that matter to anyone you answer to?" + +### 3. The money questions (5 min โ€” only if ยง1 surfaced a real incident) + +Still past tense. Budgets reveal more than intentions. + +- "After the incident, did anyone propose buying something instead of + building? What happened to that conversation?" +- "What have you already paid for in this general area โ€” observability, + gateways, audit tooling? Roughly what does that cost you a year?" +- "The last tool you bought in this space โ€” who approved it, whose budget + did it come out of, and what did procurement put you through?" + *(This is the buyer question. Adjacent-category spend without a named + approver and budget is not evidence anyone can buy; record all three.)* +- "Has the fix you built ever broken, or needed a rewrite? What did you + actually do โ€” rebuild it, or go looking for a vendor?" + +**Do not ask "would you pay for X?"** The affirmative answer is worthless and +contaminates the rest of the interview. + +### 4. Reveal and test (5 min โ€” optional, only if ยง1โ€“3 qualified them) + +Now, and only now, one paragraph โ€” the debit-first pitch, no superlatives +(the never-claim list in `WEDGE.md` applies verbatim in conversation): + +> "What we're building is a gateway you put in front of *one* tool like that. +> One accepted request key permits at most one dispatch and at most one +> debit from an internal budget, no matter how many times the agent retries โ€” +> and every terminal outcome, including denials and failures, gets a signed +> receipt someone can verify without trusting us or you." + +Then stop talking. The reactions worth recording: + +- Do they map it onto the incident from ยง1 unprompted? +- Do they object with the *right* objections ("what about the upstream + tool's own side effect?" โ€” the honest answer is in `ELEVATOR_PITCH.md`)? +- The commitment test โ€” ask for something that costs them, and make it + observable *on the call*, not aspirational: + > "We take one design partner at a time: one real tool, one engineer, a + > staging environment. If retrying deliberately and verifying the receipt + > in *your* environment doesn't make you want it in prod, we part friends. + > Who on your team would own that โ€” and can we put their afternoon on the + > calendar before we hang up?" + + A pass requires all three: a **named engineer**, a **named staging + environment**, and a **scheduled date or a concrete scheduling step taken + on the call** (an invite sent, a named owner asked to propose a slot). + "Sounds cool, keep me posted" โ€” or a yes with no name and no date โ€” is a + no; write it down as a no. + + This gate is **pre-qualification only**. Pilot readiness is a separate, + stricter contract โ€” the Pilot Acceptance Test in + [`30-day-customer-validation.md`](30-day-customer-validation.md) requires + a partner-owned agent, a partner-owned staging MCP tool, a partner + engineer operating the workflow, and independent partner-side receipt + verification. Passing this interview gate advances someone to that + conversation, not past it. + +### 5. Close (1 min) + +> "Last one: who else do you know who's been bitten by this?" *(Referral = +> the problem is talked about; silence = it may be too niche to spread.)* + +--- + +## Scoring โ€” fill in within one hour of the call + +| Field | Answer | +|---|---| +| Incident in past tense? (yes / no / **unknown** โ€” couldn't tell from their logs story) | | +| Discovery mechanism (how they found out) | | +| `delivery_uncertain` shape present? | | +| Anyone outside eng needed the record? Who? | | +| Current mitigation, and its owner | | +| Prior spend in the category | | +| Buyer evidence (last approver, budget owner, procurement path) | | +| Their words for the problem (verbatim) | | +| Commitment test: engineer named | | +| Commitment test: staging env named | | +| Commitment test: date scheduled, or concrete scheduling step taken | | +| Disqualified at screening? Why? | | + +An interview counts toward the denominator only if it passed screening and +ran through ยง2 โ€” call that number **qualified_n**. Screen-outs are recorded +but never counted. "Unknown" incident answers count as neither yes nor no. + +## Decision rule + +Set **before** the interviews, so the results can't be argued with after. +All counts are over **qualified_n = 5** completed, qualified interviews โ€” +keep interviewing until there are five; do not decide on fewer. + +- **3+ of 5** surface a past-tense incident **and** **2+ of 5** pass the + full commitment test (engineer + staging + date or concrete scheduling + step) โ†’ proceed to a paid-pilot conversation. Selection is deterministic: + take the passer with the **earliest scheduled staging date** first (a + concrete-step-only pass sorts after any dated pass); tie-break on the most + complete buyer-evidence row (approver, budget, procurement โ€” three fields + beat two), then on the earlier interview. The rest queue in the same + order โ€” one design partner at a time. A paid-pilot *conversation* โ€” the + commitment test proves an engineer's afternoon, not a budget; the pilot + conversation is where the ยง3 buyer evidence (approver, budget, procurement) + gets tested for real. +- Incidents at 3+ but **zero or one** commitments โ†’ the problem is real and + the build-it-yourself answer is winning; revisit whether the wedge is + evidence (sell to whoever outside eng asks for the record) rather than + reliability. +- **No past-tense incidents in 5** โ†’ stripe/ai#402 and the framework issues + are ahead of the market; pause the sales motion, keep the gateway as the + proof asset, and re-interview in a quarter. +- **Any other pattern** โ€” 1โ€“2 incidents, a pile of unknowns, commitments + without incidents โ€” is not a decision, it's noise. Default action for + every case not listed above: run five more qualified interviews before + deciding anything. + +What this script must never do: rescue a failing interview with the LangGraph +citation. The evidence in ยง2 of the research doc is for *us* โ€” to know the +problem is real โ€” and for a qualified prospect who asks "has this happened to +anyone?" It is not a crowbar for manufacturing pain in someone who doesn't +have it. diff --git a/examples/acta_receipt_interop.py b/examples/acta_receipt_interop.py new file mode 100644 index 00000000..cbf90c87 --- /dev/null +++ b/examples/acta_receipt_interop.py @@ -0,0 +1,291 @@ +"""Experiment: emit a portable receipt in the ACTA signed-receipt envelope. + +Answers `docs/market-research-2026-08.md` ยง7 question 2 with running code +instead of an estimate. The ACTA format is +draft-farley-acta-signed-receipts (ScopeBlind / protect-mcp): an individual +Internet-Draft with **no IETF standing** โ€” namespaced receipt types over +RFC 8785 (JCS) canonical JSON, signed with Ed25519. + +What this demonstrates +---------------------- +A signed Agent Middleware portable receipt bundle (the artifact served by +``/v1/receipts/{id}/portable`` and published at ``site/proof/receipt.json``) +can be transcoded into an ACTA envelope as a custom +``agentmiddleware:governed_invoke`` receipt type without changing the trust +plane. The economic claims โ€” ledger entry, idempotency record, permit, +credits โ€” ride as first-class fields the draft's namespacing explicitly +permits, which is precisely the room the research pass found. + +What this deliberately does NOT do +---------------------------------- +- It does not re-sign with the production service key. The original bundle + travels inside the envelope untouched, and its Ed25519 signature is + verified over the exact ``signing_input`` bytes against the bundle's own + key material, the same check ``b2a-verify-receipt`` performs. The outer + ACTA signature is applied by whoever transcodes (here: a caller-supplied + key; in any real adoption: the service signing key, server-side). +- It is not a product surface, an API route, or an SDK export. It is an + executable answer to "would interop cost a serializer or a redesign?" + Promoting it is a product decision recorded in the research doc, not here. + +Run the demo against the checked-in public proof artifact: + + python3 examples/acta_receipt_interop.py site/proof/receipt.json +""" + +from __future__ import annotations + +import base64 +import json +import sys +from dataclasses import dataclass +from typing import Any + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + +ACTA_TYPE = "agentmiddleware:governed_invoke" + +# Fields the transcoder requires inside the portable bundle's signing_input. +# Missing any of these is a hard error: an economic receipt without its +# ledger linkage is exactly the artifact this experiment exists to avoid. +REQUIRED_CLAIMS = ( + "receipt_id", + "outcome", + "tool", + "created_at", + "credits_charged", + "ledger_entry_id", + "idempotency_record_id", + "permit_id", +) + + +class TranscodeError(ValueError): + """A portable bundle that cannot be honestly represented in ACTA form.""" + + +def jcs_canonicalize(value: Any) -> bytes: + """Minimal RFC 8785 (JCS) canonicalization for the value shapes we emit. + + The full JCS spec includes IEEE-754 number serialization rules. Every + field this transcoder produces is a string, bool, None, int, list, or + dict, so floats are rejected outright rather than risking a + canonicalization mismatch between implementations. Fail closed: a + signature over ambiguous bytes is worse than no signature. + """ + + def check(node: Any) -> None: + """Recursively reject value shapes outside the supported JCS subset.""" + if isinstance(node, float): + raise TranscodeError( + "float values are not supported by this JCS subset; " + "represent amounts as strings, as the trust plane already does" + ) + if isinstance(node, dict): + for key, item in node.items(): + if not isinstance(key, str): + raise TranscodeError("non-string object keys are not valid JSON") + check(item) + elif isinstance(node, list): + for item in node: + check(item) + + check(value) + return json.dumps( + value, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + + +@dataclass(frozen=True) +class VerifiedBundle: + """A portable bundle whose own Ed25519 signature has been checked.""" + + claims: dict[str, Any] + signing_input: str + signature_b64: str + kid: str + issuer: str + + +def verify_portable_bundle( + bundle: dict[str, Any], public_key_b64: str, expected_issuer: str | None = None +) -> VerifiedBundle: + """Verify the inner receipt exactly as the offline SDK verifier does. + + The signature is checked over the raw ``signing_input`` bytes โ€” never + over a re-serialization โ€” so this function cannot be tricked by a bundle + whose pretty ``claims`` disagree with what was actually signed. + + The bundle's ``issuer`` field is wrapper metadata *outside* the signed + bytes. Pass ``expected_issuer`` โ€” the issuer bound to the key material in + the trust-keys snapshot the public key came from โ€” to refuse a bundle + whose wrapper claims a different issuer. Without it, the issuer is + carried through as unverified routing metadata; trust is anchored in the + key, not the label. + """ + for field in ("signing_input", "signature", "kid", "issuer"): + if not isinstance(bundle.get(field), str) or not bundle[field]: + raise TranscodeError(f"portable bundle is missing {field!r}") + declared_alg = bundle.get("alg") + if declared_alg is not None and declared_alg != "Ed25519": + raise TranscodeError( + f"bundle declares alg {declared_alg!r}; this verifier only " + "performs Ed25519 and will not report a misleading result" + ) + if expected_issuer is not None and bundle["issuer"] != expected_issuer: + raise TranscodeError( + "bundle issuer does not match the issuer bound to the trusted " + f"key record: {bundle['issuer']!r} != {expected_issuer!r}" + ) + + public_key = Ed25519PublicKey.from_public_bytes( + base64.b64decode(public_key_b64, validate=True) + ) + try: + public_key.verify( + base64.b64decode(bundle["signature"], validate=True), + bundle["signing_input"].encode("utf-8"), + ) + except InvalidSignature as exc: + raise TranscodeError( + "portable bundle signature does not verify; refusing to transcode " + "an unverified receipt" + ) from exc + + claims = json.loads(bundle["signing_input"]) + if not isinstance(claims, dict): + raise TranscodeError("signing_input is not a JSON object") + if claims.get("kid") != bundle["kid"]: + raise TranscodeError("bundle kid does not match the signed kid claim") + + missing = [field for field in REQUIRED_CLAIMS if not claims.get(field)] + if missing: + raise TranscodeError(f"signed claims missing required fields: {missing}") + + return VerifiedBundle( + claims=claims, + signing_input=bundle["signing_input"], + signature_b64=bundle["signature"], + kid=bundle["kid"], + issuer=bundle["issuer"], + ) + + +def to_acta_receipt(verified: VerifiedBundle, issuer_id: str) -> dict[str, Any]: + """Map a verified portable receipt into an unsigned ACTA receipt body. + + Field choices follow the draft's conventions where one exists + (``type``, ``issued_at``, ``issuer_id``, ``tool_name``) and carry the + economic linkage โ€” the part no surveyed receipt format defines โ€” as + plainly named custom fields. The untouched original bundle is embedded + so cryptographic trust never depends on the transcoder. Its ``issuer`` + field is wrapper metadata, trustworthy only when the caller verified it + against the key record (``expected_issuer``); the signature covers + ``signing_input`` alone. + """ + claims = verified.claims + return { + "type": ACTA_TYPE, + "issued_at": claims["created_at"], + "issuer_id": issuer_id, + "tool_name": claims["tool"], + "outcome": claims["outcome"], + "credits_charged": claims["credits_charged"], + "receipt_id": claims["receipt_id"], + "permit_id": claims["permit_id"], + "ledger_entry_id": claims["ledger_entry_id"], + "idempotency_record_id": claims["idempotency_record_id"], + "request_hash": claims.get("request_hash"), + "response_hash": claims.get("response_hash"), + "original_bundle": { + "alg": "Ed25519", + "issuer": verified.issuer, + "kid": verified.kid, + "signature": verified.signature_b64, + "signing_input": verified.signing_input, + }, + } + + +def sign_acta_receipt( + receipt: dict[str, Any], private_key: Ed25519PrivateKey, kid: str +) -> dict[str, Any]: + """Attach the draft's signature block: Ed25519 over JCS bytes, hex.""" + if receipt.get("issuer_id") != kid: + raise TranscodeError( + "the draft requires issuer_id to match the signing kid; " + f"got issuer_id={receipt.get('issuer_id')!r} kid={kid!r}" + ) + signature = private_key.sign(jcs_canonicalize(receipt)) + return {**receipt, "signature": {"alg": "Ed25519", "kid": kid, "sig": signature.hex()}} + + +def verify_acta_receipt( + signed: dict[str, Any], public_key: Ed25519PublicKey +) -> dict[str, Any]: + """Verify the outer envelope and return the receipt body.""" + signature_block = signed.get("signature") + if not isinstance(signature_block, dict): + raise TranscodeError("ACTA receipt has no signature block") + if signature_block.get("alg") != "Ed25519": + raise TranscodeError( + "ACTA signature alg must be Ed25519; refusing " + f"{signature_block.get('alg')!r}" + ) + if not isinstance(signature_block.get("sig"), str): + raise TranscodeError("ACTA signature sig must be a hex string") + body = {key: value for key, value in signed.items() if key != "signature"} + issuer_id = body.get("issuer_id") + kid = signature_block.get("kid") + if not isinstance(issuer_id, str) or not issuer_id: + raise TranscodeError("ACTA receipt issuer_id must be a non-empty string") + if not isinstance(kid, str) or not kid: + raise TranscodeError("ACTA signature kid must be a non-empty string") + if issuer_id != kid: + raise TranscodeError("issuer_id does not match the signature kid") + try: + public_key.verify( + bytes.fromhex(signature_block["sig"]), jcs_canonicalize(body) + ) + except (InvalidSignature, ValueError) as exc: + raise TranscodeError("ACTA envelope signature does not verify") from exc + return body + + +def _demo(bundle_path: str) -> int: + """Verify a portable bundle end-to-end and print the ACTA envelope.""" + with open(bundle_path, "r", encoding="utf-8") as handle: + bundle = json.load(handle) + + # The public proof bundle's key snapshot ships beside it. + with open("site/proof/trust-keys.json", "r", encoding="utf-8") as handle: + keys = json.load(handle) + key_b64 = next( + entry["public_key_b64"] + for entry in keys["keys"] + if entry["kid"] == bundle["kid"] + ) + + verified = verify_portable_bundle( + bundle, key_b64, expected_issuer=keys["issuer"] + ) + print(f"inner signature: VERIFIED (kid={verified.kid})") + + transcoder_key = Ed25519PrivateKey.generate() + signed = sign_acta_receipt( + to_acta_receipt(verified, issuer_id="demo-transcoder-key"), + transcoder_key, + kid="demo-transcoder-key", + ) + body = verify_acta_receipt(signed, transcoder_key.public_key()) + print(f"outer envelope: VERIFIED (type={body['type']})") + print(json.dumps(signed, indent=2)[:800]) + return 0 + + +if __name__ == "__main__": + raise SystemExit(_demo(sys.argv[1] if len(sys.argv) > 1 else "site/proof/receipt.json")) diff --git a/tests/test_acta_receipt_interop.py b/tests/test_acta_receipt_interop.py new file mode 100644 index 00000000..1939e96f --- /dev/null +++ b/tests/test_acta_receipt_interop.py @@ -0,0 +1,240 @@ +"""Tests for the ACTA receipt-envelope interop experiment. + +The experiment (`examples/acta_receipt_interop.py`) answers +`docs/market-research-2026-08.md` ยง7 question 2 with running code. These +tests pin the two properties that make the answer trustworthy: the +transcoder refuses anything it cannot verify, and both signatures โ€” the +embedded original and the outer envelope โ€” actually bind their bytes. +""" + +from __future__ import annotations + +import base64 +import importlib.util +import json +import sys +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +ROOT = Path(__file__).resolve().parents[1] + +_spec = importlib.util.spec_from_file_location( + "acta_receipt_interop", ROOT / "examples" / "acta_receipt_interop.py" +) +assert _spec is not None and _spec.loader is not None +interop = importlib.util.module_from_spec(_spec) +# Dataclass creation resolves annotations through sys.modules, so the module +# must be registered before exec, not just held locally. +sys.modules[_spec.name] = interop +_spec.loader.exec_module(interop) + + +def _make_bundle(private_key: Ed25519PrivateKey, kid: str, **overrides): + """Build a synthetic signed portable bundle, with override hooks.""" + claims = { + "alg": "Ed25519", + "created_at": "2026-08-15T00:00:00+00:00", + "credits_charged": "1", + "idempotency_record_id": "idm-test", + "kid": kid, + "ledger_entry_id": "ledger-test", + "outcome": "success", + "permit_id": "permit-test", + "receipt_id": "rcpt-test", + "request_hash": "aa" * 32, + "response_hash": "bb" * 32, + "tool": "partner.echo", + } + claims.update(overrides.pop("claims", {})) + signing_input = json.dumps( + claims, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ) + bundle = { + "issuer": "https://example.invalid", + "kid": kid, + "signature": base64.b64encode( + private_key.sign(signing_input.encode("utf-8")) + ).decode("ascii"), + "signing_input": signing_input, + } + bundle.update(overrides) + return bundle + + +def _public_b64(private_key: Ed25519PrivateKey) -> str: + """Return the raw public key of *private_key*, base64-encoded.""" + from cryptography.hazmat.primitives.serialization import ( + Encoding, + PublicFormat, + ) + + return base64.b64encode( + private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + ).decode("ascii") + + +def test_checked_in_proof_bundle_transcodes_and_roundtrips() -> None: + """The public proof artifact โ€” real signature, real key snapshot.""" + bundle = json.loads((ROOT / "site" / "proof" / "receipt.json").read_text()) + keys = json.loads((ROOT / "site" / "proof" / "trust-keys.json").read_text()) + key_b64 = next( + entry["public_key_b64"] + for entry in keys["keys"] + if entry["kid"] == bundle["kid"] + ) + + verified = interop.verify_portable_bundle( + bundle, key_b64, expected_issuer=keys["issuer"] + ) + transcoder = Ed25519PrivateKey.generate() + signed = interop.sign_acta_receipt( + interop.to_acta_receipt(verified, issuer_id="test-kid"), + transcoder, + kid="test-kid", + ) + body = interop.verify_acta_receipt(signed, transcoder.public_key()) + + assert body["type"] == interop.ACTA_TYPE + # The economic linkage โ€” the row no surveyed format defines โ€” must survive. + assert body["ledger_entry_id"] == verified.claims["ledger_entry_id"] + assert body["idempotency_record_id"] == verified.claims["idempotency_record_id"] + # The untouched original stays independently verifiable. + inner = body["original_bundle"] + assert inner["signing_input"] == bundle["signing_input"] + assert inner["signature"] == bundle["signature"] + + +def test_unverified_inner_signature_is_refused() -> None: + """A bundle signed by a different key must not transcode.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + wrong_key = Ed25519PrivateKey.generate() + with pytest.raises(interop.TranscodeError, match="does not verify"): + interop.verify_portable_bundle(bundle, _public_b64(wrong_key)) + + +def test_tampered_signing_input_is_refused() -> None: + """Altering signing_input after signing must break verification.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + bundle["signing_input"] = bundle["signing_input"].replace('"1"', '"2"', 1) + with pytest.raises(interop.TranscodeError, match="does not verify"): + interop.verify_portable_bundle(bundle, _public_b64(key)) + + +def test_kid_mismatch_between_bundle_and_signed_claim_is_refused() -> None: + """The wrapper kid must match the kid inside the signed claims.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + bundle["kid"] = "kid-b" + with pytest.raises(interop.TranscodeError, match="kid"): + interop.verify_portable_bundle(bundle, _public_b64(key)) + + +def test_missing_economic_linkage_is_refused() -> None: + """An economic receipt without its ledger link must not transcode.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a", claims={"ledger_entry_id": ""}) + with pytest.raises(interop.TranscodeError, match="ledger_entry_id"): + interop.verify_portable_bundle(bundle, _public_b64(key)) + + +def test_issuer_mutated_after_signing_is_refused() -> None: + """The wrapper issuer is unsigned; the key-record binding must catch it.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + bundle["issuer"] = "https://attacker.invalid" + with pytest.raises(interop.TranscodeError, match="issuer"): + interop.verify_portable_bundle( + bundle, _public_b64(key), expected_issuer="https://example.invalid" + ) + + +def test_signature_alg_tamper_is_refused() -> None: + """Changing only signature.alg must fail even though the sig bytes hold.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + verified = interop.verify_portable_bundle(bundle, _public_b64(key)) + transcoder = Ed25519PrivateKey.generate() + signed = interop.sign_acta_receipt( + interop.to_acta_receipt(verified, issuer_id="test-kid"), + transcoder, + kid="test-kid", + ) + signed["signature"] = {**signed["signature"], "alg": "ES256"} + with pytest.raises(interop.TranscodeError, match="alg"): + interop.verify_acta_receipt(signed, transcoder.public_key()) + + +def test_outer_envelope_tamper_is_detected() -> None: + """Changing any envelope field after signing must fail verification.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + verified = interop.verify_portable_bundle(bundle, _public_b64(key)) + transcoder = Ed25519PrivateKey.generate() + signed = interop.sign_acta_receipt( + interop.to_acta_receipt(verified, issuer_id="test-kid"), + transcoder, + kid="test-kid", + ) + signed["credits_charged"] = "999" + with pytest.raises(interop.TranscodeError, match="does not verify"): + interop.verify_acta_receipt(signed, transcoder.public_key()) + + +def test_issuer_id_must_match_signing_kid() -> None: + """The draft binds issuer_id to the signature kid; enforce both ways.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + verified = interop.verify_portable_bundle(bundle, _public_b64(key)) + transcoder = Ed25519PrivateKey.generate() + with pytest.raises(interop.TranscodeError, match="issuer_id"): + interop.sign_acta_receipt( + interop.to_acta_receipt(verified, issuer_id="someone-else"), + transcoder, + kid="test-kid", + ) + + +def test_inner_bundle_declared_alg_mismatch_is_refused() -> None: + """A bundle declaring a non-Ed25519 alg must not verify as Ed25519.""" + key = Ed25519PrivateKey.generate() + bundle = _make_bundle(key, "kid-a") + bundle["alg"] = "ES256" + with pytest.raises(interop.TranscodeError, match="alg"): + interop.verify_portable_bundle(bundle, _public_b64(key)) + + +@pytest.mark.parametrize( + ("issuer_id", "kid", "match"), + [ + (None, None, "issuer_id"), # both absent: None == None must not pass + (None, "test-kid", "issuer_id"), + ("", "test-kid", "issuer_id"), + ("test-kid", None, "kid"), + ("test-kid", "", "kid"), + ], +) +def test_missing_signer_identity_is_refused( + issuer_id: str | None, kid: str | None, match: str +) -> None: + """Each signer-identity field must independently refuse absent/empty.""" + key = Ed25519PrivateKey.generate() + body = {"type": interop.ACTA_TYPE, "receipt_id": "rcpt-test"} + if issuer_id is not None: + body["issuer_id"] = issuer_id + signature = key.sign(interop.jcs_canonicalize(body)) + signature_block = {"alg": "Ed25519", "sig": signature.hex()} + if kid is not None: + signature_block["kid"] = kid + signed = {**body, "signature": signature_block} + with pytest.raises(interop.TranscodeError, match=match): + interop.verify_acta_receipt(signed, key.public_key()) + + +def test_floats_fail_closed_in_canonicalization() -> None: + """Floats are outside the JCS subset and must be refused outright.""" + with pytest.raises(interop.TranscodeError, match="float"): + interop.jcs_canonicalize({"amount": 1.5})