diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c19954a..2e9e278f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -317,6 +317,24 @@ full release gate; do not backfill a final `v1.2.0` tag. ### 🔒 Security +- **The governed AWI HTTP charge is now keyed to its idempotency record.** + `awi_http_governance.py` called `money.charge()` with no `operation_key`, so + the debit had neither the `uq_ledger_wallet_operation_key` constraint nor the + adopt-the-existing-debit recovery the governed MCP path relies on. If the + charge committed but its acknowledgement was lost, the handler released the + permit budget and *completed* the idempotency record as `charge_failed`; the + caller's key then replayed a permanent error, so a retry under a fresh key + was **debited a second time for one logical action**, and because + `response_json` was non-NULL neither reconciler branch ever flagged the + orphan. Six governed endpoints route through this path. Unlike the + read-modify-write races elsewhere in this codebase, this one did not depend + on a row lock being a no-op — it was exposed on PostgreSQL too. +- **The public MCP `verify_receipt` tool no longer describes a caught forgery + as a verifier limitation.** It reports `MISMATCH` before consulting + `is_tampered`, so a bundle whose envelope disagrees with its signed content + is not narrated to the model as a bad signature — or, previously, as + "CANNOT VERIFY (unsupported)". See `b2a_sdk/CHANGELOG.md` for the verifier + change this depends on. - **`.env.production` now sets `ENVIRONMENT=production`.** It omitted the variable entirely, so `Settings.ENVIRONMENT` fell back to its `"local"` default and `is_production_like_environment()` returned `False` — meaning a diff --git a/app/routers/mcp_public.py b/app/routers/mcp_public.py index a13439db..b3c14ab2 100644 --- a/app/routers/mcp_public.py +++ b/app/routers/mcp_public.py @@ -448,17 +448,21 @@ async def _tool_verify_receipt( "signed claims are in structuredContent.claims; report only " "those fields as attested." ) - elif result.is_tampered: - text = ( - f"INVALID — {result.reason}. The bundle is well-formed but its " - "signature does not hold; treat it as altered or forged." - ) elif result.status is _receipt_verifier.VerificationStatus.MISMATCH: + # Checked before is_tampered: a MISMATCH is a rejection, but saying its + # "signature does not hold" would be false. The signature may well have + # verified -- what failed is agreement between the signed bytes and the + # envelope around them, or an explicit caller expectation. text = ( f"CANNOT ACCEPT (mismatch) — {result.reason}. The evidence or " "caller expectation is internally inconsistent; this does not " "mean the signature failed." ) + elif result.is_tampered: + text = ( + f"INVALID — {result.reason}. The bundle is well-formed but its " + "signature does not hold; treat it as altered or forged." + ) else: text = ( f"CANNOT VERIFY ({result.status.value}) — {result.reason}. This " diff --git a/app/services/awi_http_governance.py b/app/services/awi_http_governance.py index b16c6603..c18d6ef4 100644 --- a/app/services/awi_http_governance.py +++ b/app/services/awi_http_governance.py @@ -57,6 +57,10 @@ class AwiHttpGovernedContext: credits: Decimal permit: PermitModel endpoint: str + #: Identity of this request's idempotency record. Used as the ledger + #: ``operation_key`` so a retried charge is deduplicated by the database + #: rather than by hoping the first attempt's outcome was observed. + record_id: str | None = None replay_response: dict[str, Any] | None = None replay_status_code: int | None = None @@ -90,6 +94,7 @@ def _context_from_validation( credits: Decimal, permit: PermitModel, endpoint: str, + record_id: str | None = None, replay: IdempotencyReplay | None = None, ) -> AwiHttpGovernedContext: return AwiHttpGovernedContext( @@ -101,6 +106,7 @@ def _context_from_validation( credits=credits, permit=permit, endpoint=endpoint, + record_id=record_id, replay_response=replay.response_json if replay else None, replay_status_code=replay.status_code if replay else None, ) @@ -178,7 +184,7 @@ async def begin_awi_http_governed( idem = get_idempotency_service() try: - replay = await idem.begin( + begun = await idem.begin_with_record( wallet_id=wallet_id, endpoint=endpoint, idempotency_key=idempotency_key.strip(), @@ -189,6 +195,8 @@ async def begin_awi_http_governed( "permit_id": permit_id.strip(), }, ) + replay = begun.replay + record_id = begun.record_id except (IdempotencyConflictError, IdempotencyInProgressError) as exc: raise HTTPException( status_code=status.HTTP_409_CONFLICT, @@ -205,6 +213,7 @@ async def begin_awi_http_governed( credits=credits, permit=validation.permit, endpoint=endpoint, + record_id=record_id, replay=replay, ) @@ -217,6 +226,7 @@ async def begin_awi_http_governed( credits=credits, permit=validation.permit, endpoint=endpoint, + record_id=record_id, ) @@ -248,6 +258,16 @@ async def complete_awi_http_governed( units=charge_units, request_path=ctx.endpoint, description=f"AWI HTTP {ctx.tool_name}", + # Key the debit to this request's durable identity. Without it this + # path had neither the uq_ledger_wallet_operation_key constraint nor + # the adopt-the-existing-debit recovery that the governed MCP path + # relies on -- so a charge that committed but whose acknowledgement + # was lost took the except branch below, which releases budget and + # *completes* the idempotency record as charge_failed. The caller + # then retried under a fresh key and was debited a second time for + # one logical action. With the key, the second attempt adopts the + # first durable debit instead of creating one. + operation_key=ctx.record_id, ) except Exception as exc: await permits.release_budget(ctx.permit_id, ctx.credits) diff --git a/b2a_sdk/CHANGELOG.md b/b2a_sdk/CHANGELOG.md index d9462aa7..4f2c2eae 100644 --- a/b2a_sdk/CHANGELOG.md +++ b/b2a_sdk/CHANGELOG.md @@ -9,9 +9,22 @@ from inside them. - Distinguish a failed verification (`INVALID`) from an undetermined one (`UNKNOWN_KEY`, `MALFORMED`, `UNSUPPORTED`), so callers do not mistake a - stale key cache for tampering. -- Add the `b2a-verify-receipt` CLI, with exit codes `0` verified, `1` forged, - `2` undetermined. + stale key cache for tampering. A bundle that is internally inconsistent — + or that disagrees with a caller-supplied `expected_issuer` — is `MISMATCH`: + a finding about the bundle, distinct from both families. +- Read `kid`, `alg`, and `canonicalization` from the **signed** payload, never + from the surrounding envelope. The envelope is unauthenticated, so letting it + select the key or gate the capability checks would let one edited byte + downgrade a genuine receipt to `UNSUPPORTED` — an attacker-chosen "your + verifier is too old" that reads as a verifier problem rather than a bundle + problem. Envelope values are cross-checked *after* the signature verifies, + and any disagreement is reported as `MISMATCH`. +- Add `VerificationResult.is_rejected`, true for `INVALID` and `MISMATCH` — + the property callers should branch on for "do not trust this bundle". + `is_tampered` stays narrow (`INVALID` only), because `MISMATCH` also covers + an `expected_issuer` disagreement, which is not a claim about tampering. +- Add the `b2a-verify-receipt` CLI, with exit codes `0` verified, `1` rejected + (`INVALID` or `MISMATCH`), `2` undetermined. - Add the `verify` extra (`pip install "b2a-sdk[verify]"`) for the `cryptography` dependency. The base install is unchanged; importing `b2a_sdk` without the extra still works, and only signature checking diff --git a/b2a_sdk/src/b2a_sdk/receipt_verifier.py b/b2a_sdk/src/b2a_sdk/receipt_verifier.py index fe7b82ce..e3db68ce 100644 --- a/b2a_sdk/src/b2a_sdk/receipt_verifier.py +++ b/b2a_sdk/src/b2a_sdk/receipt_verifier.py @@ -105,9 +105,36 @@ def is_tampered(self) -> bool: """True only for a bundle that is well-formed but does not verify. Distinct from :attr:`ok` being False, which also covers "cannot tell". + + Deliberately narrow: this is a claim that a *signature failed*. It does + not cover ``MISMATCH``, because ``MISMATCH`` also carries caller- + expectation failures such as ``expected_issuer`` -- and reporting "you + asked for a different issuer" as forgery would be the same category + error, in the opposite direction, as reporting a missing key as + tampering. Use :attr:`is_rejected` to decide whether to accept a + bundle; use this only to assert that the cryptography failed. """ return self.status is VerificationStatus.INVALID + @property + def is_rejected(self) -> bool: + """True when the bundle must not be accepted as evidence. + + The verifier reached a verdict and the verdict is no. Covers a failed + signature (``INVALID``) and a bundle that does not hold together + (``MISMATCH`` -- the signature verified but the envelope, the signed + payload, or an explicit caller expectation disagree). + + This is the property most callers want. It is distinct from ``not ok``, + which additionally sweeps in ``UNKNOWN_KEY``, ``MALFORMED`` and + ``UNSUPPORTED`` -- the "I cannot judge this" statuses, where the right + response is to fetch keys or upgrade, not to allege fraud. + """ + return self.status in ( + VerificationStatus.INVALID, + VerificationStatus.MISMATCH, + ) + def canonical_json(payload: Any) -> str: """Serialize already-decoded JSON the way the trust plane signs it. @@ -286,60 +313,86 @@ def verify_bundle( signing_input = bundle.get("signing_input") signature_b64 = bundle.get("signature") - key_id = bundle.get("kid") + envelope_key_id = bundle.get("kid") if not isinstance(signing_input, str) or not signing_input: return _fail(VerificationStatus.MALFORMED, "bundle has no signing_input") if not isinstance(signature_b64, str) or not signature_b64: return _fail(VerificationStatus.MALFORMED, "bundle has no signature") - if not isinstance(key_id, str) or not key_id: + if not isinstance(envelope_key_id, str) or not envelope_key_id: return _fail(VerificationStatus.MALFORMED, "bundle has no kid") - algorithm = bundle.get("alg", "Ed25519") - if algorithm != "Ed25519": - return _fail( - VerificationStatus.UNSUPPORTED, - f"unsupported signature algorithm: {algorithm!r}", - key_id=key_id, - ) - - declared = bundle.get("canonicalization", CANONICALIZATION) - if declared != CANONICALIZATION: - # Verifying under rules the payload was not signed with would produce a - # confident answer to the wrong question. - return _fail( - VerificationStatus.UNSUPPORTED, - f"unsupported canonicalization: {declared!r} (this verifier " - f"implements {CANONICALIZATION!r})", - key_id=key_id, - ) - if expected_issuer is not None and bundle.get("issuer") != expected_issuer: return _fail( VerificationStatus.MISMATCH, f"issuer mismatch: bundle claims {bundle.get('issuer')!r}", - key_id=key_id, + key_id=envelope_key_id, ) - # Parse the signed bytes before checking the signature so a malformed - # payload is reported as malformed rather than as a forgery. + # Parse the signed bytes before anything else that can decide an outcome. + # + # Ordering is the whole security property here. The envelope is + # unauthenticated: anyone in the transport path can edit it. So the + # capability gates below (algorithm, canonicalization) and the key + # selection must read the *signed* payload, never the envelope. An earlier + # version read `alg` and `canonicalization` from the envelope and + # short-circuited to UNSUPPORTED before the signature was ever checked, + # which let one edited byte turn a genuine receipt into "your verifier is + # too old" -- an attacker-chosen status, and the exact integrity/capability + # inversion this taxonomy exists to prevent. Envelope values are still + # checked, but afterwards, and disagreement is MISMATCH (a finding about + # the bundle) rather than UNSUPPORTED (a statement about this verifier). try: payload = json.loads(signing_input) except (json.JSONDecodeError, UnicodeDecodeError, RecursionError, ValueError): return _fail( VerificationStatus.MALFORMED, "signing_input is not valid JSON", - key_id=key_id, + key_id=envelope_key_id, ) if not isinstance(payload, dict): return _fail( VerificationStatus.MALFORMED, "signing_input is not a JSON object", - key_id=key_id, + key_id=envelope_key_id, ) if _contains_invalid_unicode(payload): return _fail( VerificationStatus.MALFORMED, "signing_input contains invalid Unicode text", + key_id=envelope_key_id, + ) + + # Select the verification key by the kid inside the signed bytes. If the + # signature then verifies, that kid is authenticated retroactively, so a + # relabelled envelope cannot steer key selection at all. + key_id = payload.get("kid") + if not isinstance(key_id, str) or not key_id: + return _fail( + VerificationStatus.MALFORMED, + "signed payload has no kid", + key_id=envelope_key_id, + ) + + algorithm = payload.get("alg", "Ed25519") + if algorithm != "Ed25519": + # Signer-declared, and therefore a genuine statement about what this + # verifier can do. + return _fail( + VerificationStatus.UNSUPPORTED, + f"unsupported signature algorithm: {algorithm!r}", + key_id=key_id, + ) + + signed_canonicalization = payload.get("canonicalization") + if isinstance(signed_canonicalization, str) and signed_canonicalization != CANONICALIZATION: + # Verifying under rules the payload was not signed with would produce a + # confident answer to the wrong question. Only trusted because it is + # signed; receipts predating this field carry no counterpart and are + # handled by the envelope cross-check after verification. + return _fail( + VerificationStatus.UNSUPPORTED, + f"unsupported canonicalization: {signed_canonicalization!r} (this " + f"verifier implements {CANONICALIZATION!r})", key_id=key_id, ) @@ -359,6 +412,28 @@ def verify_bundle( raw_public_key = key_set.get(key_id) if raw_public_key is None: + # The payload names a key this verifier does not hold. Normally that is + # "cannot judge" -- but first check whether the envelope points at a + # key we *do* hold that actually signs these bytes. If it does, the + # bundle is provably self-contradicting (signed under one key while + # claiming another), and that is a verdict, not a gap. Skipping this + # would let a forger name a nonexistent kid to downgrade a detectable + # inconsistency into "refresh your key set and try again". + envelope_key = key_set.get(envelope_key_id) if envelope_key_id != key_id else None + if envelope_key is not None and len(envelope_key) == _ED25519_RAW_LEN: + try: + probe_bytes = signing_input.encode("utf-8") + except UnicodeEncodeError: + probe_bytes = None + if probe_bytes is not None and _verify_ed25519_signature( + envelope_key, signature, probe_bytes + ): + return _fail( + VerificationStatus.MISMATCH, + f"signed under kid {envelope_key_id!r} but the signed " + f"payload names kid {key_id!r}", + key_id=key_id, + ) # Not a verdict on the receipt: this verifier simply does not hold the # key. Refresh the key set from the issuer before drawing conclusions. return _fail( @@ -392,12 +467,36 @@ def verify_bundle( key_id=key_id, ) - # The signature holds. Remaining checks are internal consistency of the - # signed payload itself. - if payload.get("kid") != key_id: + # The signature holds, so the signed payload is now authenticated ground + # truth. Every envelope value that claims to describe it is checked against + # it here. Disagreement is a finding about the bundle -- MISMATCH -- never + # UNSUPPORTED, because at this point the verifier demonstrably *could* + # judge the evidence. + if envelope_key_id != key_id: + return _fail( + VerificationStatus.MISMATCH, + "bundle names a different kid than the signed payload", + key_id=key_id, + ) + + envelope_algorithm = bundle.get("alg", "Ed25519") + if envelope_algorithm != algorithm: + return _fail( + VerificationStatus.MISMATCH, + f"bundle claims algorithm {envelope_algorithm!r} but the signed " + f"payload declares {algorithm!r}", + key_id=key_id, + ) + + envelope_canonicalization = bundle.get("canonicalization", CANONICALIZATION) + expected_canonicalization = ( + signed_canonicalization if isinstance(signed_canonicalization, str) else CANONICALIZATION + ) + if envelope_canonicalization != expected_canonicalization: return _fail( VerificationStatus.MISMATCH, - "signed payload names a different kid than the bundle", + f"bundle claims canonicalization {envelope_canonicalization!r} but " + f"the signed payload verifies under {expected_canonicalization!r}", key_id=key_id, ) diff --git a/b2a_sdk/src/b2a_sdk/verify_cli.py b/b2a_sdk/src/b2a_sdk/verify_cli.py index 26cb46a3..02d22296 100644 --- a/b2a_sdk/src/b2a_sdk/verify_cli.py +++ b/b2a_sdk/src/b2a_sdk/verify_cli.py @@ -14,7 +14,9 @@ === ========================================================== 0 signature verified -1 well-formed bundle that does not verify — treat as tampered +1 well-formed bundle that does not hold together — treat as + tampered (failed signature, or a signature that verified over + bytes contradicting the envelope around them) 2 could not determine (unknown key, malformed input, bad usage) === ========================================================== @@ -32,7 +34,6 @@ from .receipt_verifier import ( VerificationError, - VerificationStatus, key_set_from_document, verify_bundle, ) @@ -179,7 +180,12 @@ def main(argv: list[str] | None = None) -> int: if result.ok: return EXIT_VERIFIED - if result.status is VerificationStatus.INVALID: + if result.is_rejected: + # MISMATCH belongs here, not in EXIT_UNDETERMINED. It is a determined + # rejection -- the signature verified and the signed bytes then + # contradicted the envelope -- which is exactly the relabelling attack + # this tool exists to catch. Routing it to "could not determine" told + # a branching script to treat a caught forgery as an outage. return EXIT_INVALID return EXIT_UNDETERMINED diff --git a/b2a_sdk/tests/test_receipt_verifier.py b/b2a_sdk/tests/test_receipt_verifier.py index 7b829907..fa04ee3d 100644 --- a/b2a_sdk/tests/test_receipt_verifier.py +++ b/b2a_sdk/tests/test_receipt_verifier.py @@ -119,8 +119,25 @@ def test_key_set_rejects_a_document_with_no_keys_list(): ({"kid": ""}, VerificationStatus.MALFORMED), ({"signing_input": "not json"}, VerificationStatus.MALFORMED), ({"signing_input": "[1,2]"}, VerificationStatus.MALFORMED), - ({"alg": "RS256"}, VerificationStatus.UNSUPPORTED), - ({"canonicalization": "something-else/9"}, VerificationStatus.UNSUPPORTED), + # UNSUPPORTED must be driven by what the SIGNER declared, inside the + # signed bytes -- never by the envelope. These still resolve before any + # signature check, so they need no key material. + ( + {"signing_input": json.dumps({"receipt_id": "rcpt-1", "kid": "key-1", "alg": "RS256"})}, + VerificationStatus.UNSUPPORTED, + ), + ( + { + "signing_input": json.dumps( + { + "receipt_id": "rcpt-1", + "kid": "key-1", + "canonicalization": "something-else/9", + } + ) + }, + VerificationStatus.UNSUPPORTED, + ), ], ) def test_unusable_bundles_report_why(overrides, expected): diff --git a/tests/test_awi_http_governance.py b/tests/test_awi_http_governance.py index 0c760816..99e32c9f 100644 --- a/tests/test_awi_http_governance.py +++ b/tests/test_awi_http_governance.py @@ -5,8 +5,11 @@ from decimal import Decimal import pytest +from sqlalchemy import func, select from httpx import ASGITransport, AsyncClient +from app.db.database import get_session_factory +from app.db.models import IdempotencyRecordModel, LedgerEntryModel from app.main import app from tests.test_trust_helpers import ( BOOTSTRAP_HEADERS, @@ -141,3 +144,78 @@ async def test_execute_denied_without_wallet_scoped_session(client, clean_databa ) assert resp.status_code == 400 assert resp.json()["detail"]["error"] == "wallet_required" + + +@pytest.mark.anyio +async def test_awi_route_keys_its_debit_to_the_idempotency_record( + client, clean_database +): + """The governed AWI route must key its debit to the request's identity. + + This path used to call ``money.charge()`` with no ``operation_key``, so it + had neither the ``uq_ledger_wallet_operation_key`` constraint nor the + adopt-the-existing-debit recovery the governed MCP path relies on. A charge + that committed but whose acknowledgement was lost took the failure branch, + which releases the permit budget and *completes* the idempotency record as + ``charge_failed`` -- so the caller retried under a fresh key and paid twice + for one logical action. + + Asserted on the real route rather than on ``money.charge`` directly, + because the defect was the wiring: the mechanism already existed and this + caller simply did not use it. + """ + provisioned = await provision_agent_wallet(client) + permit = await create_tool_permit( + client, + wallet_id=provisioned["agent_wallet_id"], + key_id=provisioned["key_id"], + tool_name="awi_rag_query", + max_credits=50, + idem_key="permit-awi-opkey", + ) + resp = await client.post( + "/v1/awi/rag/query", + json={"query": "laptops", "top_k": 3}, + headers={ + **provisioned["agent_headers"], + "X-Wallet-Id": provisioned["agent_wallet_id"], + "X-Permit-Id": permit["permit_id"], + "Idempotency-Key": "awi-opkey-1", + }, + ) + assert resp.status_code == 200, resp.text + + factory = get_session_factory() + async with factory() as session: + record_id = ( + await session.execute( + select(IdempotencyRecordModel.record_id).where( + IdempotencyRecordModel.wallet_id == provisioned["agent_wallet_id"], + IdempotencyRecordModel.idempotency_key == "awi-opkey-1", + ) + ) + ).scalar_one() + keyed = ( + await session.execute( + select(func.count()) + .select_from(LedgerEntryModel) + .where(LedgerEntryModel.operation_key == record_id) + ) + ).scalar_one() + unkeyed = ( + await session.execute( + select(func.count()) + .select_from(LedgerEntryModel) + .where( + LedgerEntryModel.wallet_id == provisioned["agent_wallet_id"], + LedgerEntryModel.action == "charge", + LedgerEntryModel.operation_key.is_(None), + ) + ) + ).scalar_one() + + assert keyed == 1, ( + "the AWI debit is not keyed to its idempotency record, so a retry " + "after a lost acknowledgement would debit again" + ) + assert unkeyed == 0, "a governed AWI charge landed with no operation_key" diff --git a/tests/test_public_mcp_endpoint.py b/tests/test_public_mcp_endpoint.py index 8592b71a..ef29bedf 100644 --- a/tests/test_public_mcp_endpoint.py +++ b/tests/test_public_mcp_endpoint.py @@ -621,9 +621,35 @@ async def test_expected_issuer_mismatch_is_not_a_signature_verdict( ("bundle", "expected_status"), [ ("not-json", "malformed"), - ({**_unverifiable_bundle(), "alg": "RSA"}, "unsupported"), + # The unsupported cases must be declared inside the SIGNED bytes. + # Reading them from the envelope is what let an edited byte downgrade a + # genuine receipt to "this verifier is too old", so the envelope is no + # longer trusted for that decision -- which also means an envelope-only + # bad alg is no longer cheaply rejectable, by design. ( - {**_unverifiable_bundle(), "canonicalization": "unknown/1"}, + { + **_unverifiable_bundle(), + "signing_input": json.dumps( + { + "receipt_id": "rcpt-no-key", + "kid": "no-published-key", + "alg": "RSA", + } + ), + }, + "unsupported", + ), + ( + { + **_unverifiable_bundle(), + "signing_input": json.dumps( + { + "receipt_id": "rcpt-no-key", + "kid": "no-published-key", + "canonicalization": "unknown/1", + } + ), + }, "unsupported", ), ], @@ -716,12 +742,17 @@ async def test_verify_receipt_rejects_a_tampered_bundle(client, public_mcp_enabl @pytest.mark.anyio async def test_unknown_key_is_not_reported_as_tampering(client, public_mcp_enabled): - bundle = await _signed_bundle() await get_signing_key_service().ensure_active_key() - relabeled = {**bundle, "kid": "kid-that-does-not-exist"} + # A key this plane genuinely does not publish, named inside the SIGNED + # bytes, is the real "unknown key" case. + unknown = await _signed_bundle() + payload = json.loads(unknown["signing_input"]) + payload["kid"] = "kid-that-does-not-exist" + unknown["signing_input"] = json.dumps(payload) + unknown["kid"] = "kid-that-does-not-exist" resp = await client.post( PUBLIC_PATH, - json=_call("verify_receipt", {"bundle": relabeled}), + json=_call("verify_receipt", {"bundle": unknown}), headers=MCP_HEADERS, ) structured = resp.json()["result"]["structuredContent"] @@ -731,6 +762,31 @@ async def test_unknown_key_is_not_reported_as_tampering(client, public_mcp_enabl assert "not evidence of tampering" in text.lower() +@pytest.mark.anyio +async def test_relabelled_envelope_kid_is_a_mismatch_not_an_unknown_key( + client, public_mcp_enabled +): + """Editing only the envelope must not look like a key-distribution problem. + + The signed payload still names the real key, so this plane holds it and the + receipt is genuine. Reporting UNKNOWN_KEY here would let anyone in the + transport path turn a valid receipt into "refresh your key set" with one + edited field -- an availability story for what is actually an altered + bundle. + """ + bundle = await _signed_bundle() + await get_signing_key_service().ensure_active_key() + relabeled = {**bundle, "kid": "kid-that-does-not-exist"} + resp = await client.post( + PUBLIC_PATH, + json=_call("verify_receipt", {"bundle": relabeled}), + headers=MCP_HEADERS, + ) + structured = resp.json()["result"]["structuredContent"] + assert structured["ok"] is False + assert structured["status"] == "mismatch" + + @pytest.mark.anyio async def test_verify_receipt_without_bundle_is_invalid_params( client, public_mcp_enabled diff --git a/tests/test_receipt_portability.py b/tests/test_receipt_portability.py index 7e0c570c..d0a65886 100644 --- a/tests/test_receipt_portability.py +++ b/tests/test_receipt_portability.py @@ -697,3 +697,97 @@ def test_cli_emits_machine_readable_output(tmp_path, capsys): assert emitted["ok"] is True assert emitted["status"] == "verified" assert emitted["claims"]["tool"] == "search" + + +@pytest.mark.parametrize( + "overrides,label", + [ + ({"alg": "RS256"}, "algorithm"), + ({"canonicalization": "awi-canonical-json/2"}, "canonicalization"), + ], +) +def test_envelope_cannot_downgrade_a_genuine_receipt_to_unsupported(overrides, label): + """An edited envelope must not let an attacker choose the verdict. + + ``alg`` and ``canonicalization`` were read from the unauthenticated + envelope and short-circuited to UNSUPPORTED *before* the signature was + checked. So one edited byte on a genuine, fully-verifiable receipt + produced "this verifier is too old" -- a statement about the verifier's + own capability, chosen by whoever edited the bytes, carrying + ``is_tampered = False``. + + That is the integrity/capability inversion the six-state taxonomy exists + to prevent, and it is a deniability primitive: anyone on the transport + path could convert a VERIFIED receipt into something an operator resolves + by upgrading rather than by escalating. + + The bundle here is genuinely signed and verifies untouched, so the only + thing claiming otherwise is the envelope. + """ + kid, key_set, sign = _local_signer() + payload = _hashed_payload( + {"receipt_id": "rcpt-downgrade", "kid": kid, "alg": "Ed25519"} + ) + + baseline = verify_bundle(_signed_bundle(payload, kid=kid, sign=sign), key_set) + assert baseline.status is VerificationStatus.VERIFIED, ( + "fixture must verify untouched, or the test proves nothing" + ) + + result = verify_bundle( + _signed_bundle(payload, kid=kid, sign=sign, **overrides), key_set + ) + + assert result.status is not VerificationStatus.UNSUPPORTED, ( + f"an edited envelope {label} downgraded a genuine receipt to " + "UNSUPPORTED, which reports a tampered bundle as a verifier limitation" + ) + assert result.status is VerificationStatus.MISMATCH + # The envelope disagreeing with signed content means the bundle must not + # be accepted. Deliberately is_rejected rather than is_tampered: MISMATCH + # also covers caller-expectation failures like expected_issuer, and + # calling those forgery would be the same category error in reverse. + assert result.is_rejected + assert not result.is_tampered + + +def test_envelope_kid_relabelling_is_a_mismatch_not_a_signature_failure(): + """Selecting the key from the signed payload makes relabelling inert. + + Key selection used to read the envelope's ``kid``, so relabelling a bundle + to a *different published* key made the verifier fetch the wrong key and + report INVALID -- indistinguishable from a forged payload, even though the + signed bytes were untouched. Selecting from the signed payload means the + envelope cannot steer key selection at all. + """ + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + + kid, key_set, sign = _local_signer() + # A genuinely different published key. _local_signer() hardcodes its kid, + # so calling it twice yields one kid with the second key silently + # overwriting the first -- which makes the signature fail for the wrong + # reason and proves nothing about relabelling. + other_kid = "local-test-key-2" + other_public = ( + Ed25519PrivateKey.generate() + .public_key() + .public_bytes(Encoding.Raw, PublicFormat.Raw) + ) + key_set = {**key_set, other_kid: other_public} + + payload = _hashed_payload( + {"receipt_id": "rcpt-relabel", "kid": kid, "alg": "Ed25519"} + ) + # Sign under `kid`, then relabel the envelope to the other published key. + # Built and mutated rather than passed through **overrides, because + # _signed_bundle takes `kid` as a named parameter. + bundle = _signed_bundle(payload, kid=kid, sign=sign) + bundle["kid"] = other_kid + + result = verify_bundle(bundle, key_set) + + assert result.status is VerificationStatus.MISMATCH + assert result.is_rejected + # Reported against the key the signed bytes actually name. + assert result.key_id == kid