From 51f622bca5f29a904efb0b5ebdde6f8919a60b24 Mon Sep 17 00:00:00 2001 From: Mike Date: Sun, 14 Jun 2026 18:52:48 -0500 Subject: [PATCH] fix(097-C1): commitment hash over ciphertext, version-gated verify compute_commitment_hash now hashes the ciphertext blob, not the plaintext (C-1). create_intent / create_split_key_intent / create_selective_intent hash encrypted_payload after encryption. Verification recomputes SHA-256(ciphertext) and compares ONLY when the API reports commitment_version >= 2; legacy v1 (plaintext-hash) intents skip the check (confirmation-oracle risk) and still verify. Selective-disclosure integrity is now role-independent (the envelope hash needs no plaintext). Tests updated. Co-authored-by: Mike Co-authored-by: Claude Fable 5 --- tests/test_client.py | 60 ++++++++++++++++++++++++----- validpay/client.py | 92 ++++++++++++++++++++++---------------------- validpay/crypto.py | 22 ++++++----- 3 files changed, 108 insertions(+), 66 deletions(-) diff --git a/tests/test_client.py b/tests/test_client.py index 126da1a..f1156b6 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -294,7 +294,7 @@ def test_create_intent_batch_rejects_malformed_items(): client.create_intent_batch([{"document_type": "t"}]) # missing payload -def test_create_intent_sends_commitment_hash_matching_plaintext(): +def test_create_intent_sends_commitment_hash_over_ciphertext(): session = _FakeSession([ _FakeResponse(201, {"retrieval_id": "vp_h", "status": "active"}), ]) @@ -304,8 +304,9 @@ def test_create_intent_sends_commitment_hash_matching_plaintext(): client.create_intent(document_type="check", payload=payload) sent = json.loads(session.calls[0]["data"]) - expected_hash = compute_commitment_hash(json.dumps(payload)) - assert sent["commitment_hash"] == expected_hash + # C-1: the commitment is SHA-256 of the ciphertext blob, NOT the plaintext. + assert sent["commitment_hash"] == compute_commitment_hash(sent["encrypted_payload"]) + assert sent["commitment_hash"] != compute_commitment_hash(json.dumps(payload)) assert len(sent["commitment_hash"]) == 64 @@ -327,21 +328,25 @@ def test_create_intent_batch_includes_per_item_commitment_hash(): {"document_type": "check", "payload": payloads[1]}, ]) sent = json.loads(session.calls[0]["data"])["intents"] - assert sent[0]["commitment_hash"] == compute_commitment_hash(json.dumps(payloads[0])) - assert sent[1]["commitment_hash"] == compute_commitment_hash(json.dumps(payloads[1])) + # C-1: per-item commitment is over each item's ciphertext blob. + assert sent[0]["commitment_hash"] == compute_commitment_hash(sent[0]["encrypted_payload"]) + assert sent[1]["commitment_hash"] == compute_commitment_hash(sent[1]["encrypted_payload"]) def test_verify_intent_with_matching_commitment_hash_sets_integrity_verified(): real_key = generate_key() plaintext = json.dumps({"amount": 1500}) blob = encrypt(plaintext, real_key) - commitment_hash = compute_commitment_hash(plaintext) + # C-1: commitment v2 is over the ciphertext, and the response must carry + # commitment_version >= 2 for the verifier to enforce it. + commitment_hash = compute_commitment_hash(blob) session = _FakeSession([ _FakeResponse(200, { "intent_id": "vp_int_1", "encrypted_payload": blob, "commitment_hash": commitment_hash, + "commitment_version": 2, "issuer": "Acme", "issuer_verified": True, "registered_at": "2026-05-02T00:00:00.000Z", @@ -366,6 +371,7 @@ def test_verify_intent_with_mismatched_commitment_hash_raises_integrity_failure( "intent_id": "vp_int_2", "encrypted_payload": blob, "commitment_hash": bad_hash, + "commitment_version": 2, "issuer": "Acme", "issuer_verified": True, "registered_at": "2026-05-02T00:00:00.000Z", @@ -378,6 +384,32 @@ def test_verify_intent_with_mismatched_commitment_hash_raises_integrity_failure( assert exc.value.code == "integrity_failure" +def test_verify_intent_legacy_v1_commitment_is_skipped(): + # A v1 (or version-less) commitment over plaintext must NOT be enforced — + # it's the confirmation-oracle risk C-1 fixes. Integrity stays False and + # verification still succeeds so legacy QR codes keep working. + real_key = generate_key() + plaintext = json.dumps({"amount": 1500}) + blob = encrypt(plaintext, real_key) + + session = _FakeSession([ + _FakeResponse(200, { + "intent_id": "vp_v1", + "encrypted_payload": blob, + "commitment_hash": compute_commitment_hash(plaintext), # legacy plaintext hash + "commitment_version": 1, + "issuer": "Acme", + "issuer_verified": True, + "registered_at": "2026-05-02T00:00:00.000Z", + "status": "active", + }), + ]) + client = ValidPayClient(api_key="k", base_url="https://api.example.test", session=session) + result = client.verify_intent(retrieval_id="vp_v1", key=real_key) + assert result.integrity_verified is False + assert result.payload == {"amount": 1500} + + def test_verify_intent_legacy_intent_without_commitment_hash_passes_with_integrity_false(): real_key = generate_key() plaintext = json.dumps({"amount": 1500}) @@ -600,6 +632,7 @@ def test_split_key_round_trip(): "status": "active", "split_key": True, "commitment_hash": captured_body["commitment_hash"], + "commitment_version": 2, }), _FakeResponse(200, {"intent_id": "vp_sk_2", "fragment_b": fragment_b}), ]) @@ -639,6 +672,7 @@ def test_verify_intent_delegates_to_split_key_flow(): "status": "active", "split_key": True, "commitment_hash": sent_body["commitment_hash"], + "commitment_version": 2, }), _FakeResponse(200, {"intent_id": "vp_sk_3", "fragment_b": sent_body["key_fragment_b"]}), ]) @@ -797,10 +831,12 @@ def _build_selective_intent_response(payload: dict, policy: dict, *, intent_id: encrypted_fields, field_keys = encrypt_fields(payload) key_map = build_key_map(field_keys, policy) encrypted_key_map = encrypt(json.dumps(key_map), master_key) - commitment_hash = compute_commitment_hash(json.dumps(payload)) + envelope = json.dumps(encrypted_fields) + # C-1: commitment v2 is over the ciphertext envelope, role-independent. + commitment_hash = compute_commitment_hash(envelope) response = { "intent_id": intent_id, - "encrypted_payload": json.dumps(encrypted_fields), + "encrypted_payload": envelope, "issuer": "Acme", "issuer_verified": True, "registered_at": "2026-05-02T00:00:00.000Z", @@ -810,6 +846,7 @@ def _build_selective_intent_response(payload: dict, policy: dict, *, intent_id: "disclosure_policy": json.dumps(policy), "encrypted_key_map": encrypted_key_map, "commitment_hash": commitment_hash, + "commitment_version": 2, } return master_key, response @@ -845,8 +882,10 @@ def test_verify_selective_intent_partial_role_redacts(): assert result.payload["amount"] == 100 assert result.payload["name"] == "[REDACTED]" assert result.payload["ssn"] == "[REDACTED]" - # Integrity check only runs for "full" role. - assert result.integrity_verified is False + # C-1: the commitment is over the ciphertext envelope, so integrity is + # now verified for ANY role — not just "full" (it no longer needs the + # decrypted plaintext). + assert result.integrity_verified is True def test_verify_selective_intent_invalid_role_raises(): @@ -906,6 +945,7 @@ def test_create_selective_intent_with_split_key(): "disclosure_policy": sent["disclosure_policy"], "encrypted_key_map": sent["encrypted_key_map"], "commitment_hash": sent["commitment_hash"], + "commitment_version": 2, }), _FakeResponse(200, {"intent_id": "vp_sel_sk", "fragment_b": sent["key_fragment_b"]}), ]) diff --git a/validpay/client.py b/validpay/client.py index 2cded9d..5fba14d 100644 --- a/validpay/client.py +++ b/validpay/client.py @@ -27,6 +27,29 @@ DEFAULT_TIMEOUT = 30.0 +def _verify_commitment(data: Mapping[str, Any]) -> bool: + """Version-aware commitment check (Prompt 097 C-1). + + v2 commitments are SHA-256(ciphertext): recompute over the received + ``encrypted_payload`` and compare. v1 (legacy SHA-256(plaintext)) is a + confirmation-oracle risk and is intentionally skipped — those documents + expire naturally. Returns whether integrity was confirmed; raises on a + v2 mismatch (the ciphertext was swapped after issuance). + """ + commitment_hash = data.get("commitment_hash") + version = data.get("commitment_version", 1) + if not commitment_hash or not isinstance(version, int) or version < 2: + return False + if compute_commitment_hash(data["encrypted_payload"]) != commitment_hash: + raise ValidPayError( + "integrity_failure", + "INTEGRITY VERIFICATION FAILED — the ciphertext does not match the " + "commitment hash recorded at issuance. This document may have been " + "tampered with.", + ) + return True + + class ValidPayClient: """Client for the ValidPay API. @@ -100,8 +123,9 @@ def create_intent( result_key, share_b = split_key_fn(full_key) plaintext = json.dumps(payload) - commitment_hash = compute_commitment_hash(plaintext) encrypted_payload = encrypt(plaintext, full_key) + # Commitment v2: hash the ciphertext, not the plaintext (C-1). + commitment_hash = compute_commitment_hash(encrypted_payload) body: Dict[str, Any] = { "document_type": document_type, @@ -183,10 +207,12 @@ def create_intent_batch( key = generate_key() keys.append(key) plaintext = json.dumps(item["payload"]) + encrypted_payload = encrypt(plaintext, key) req_item: Dict[str, Any] = { "document_type": doc_type, - "encrypted_payload": encrypt(plaintext, key), - "commitment_hash": compute_commitment_hash(plaintext), + "encrypted_payload": encrypted_payload, + # Commitment v2: hash the ciphertext, not the plaintext (C-1). + "commitment_hash": compute_commitment_hash(encrypted_payload), } if valid_from is not None: req_item["valid_from"] = valid_from @@ -312,23 +338,12 @@ def verify_intent( if data.get("split_key"): key = combine_key_shares(key, self._fetch_fragment_b(retrieval_id)) - decrypted = decrypt(data["encrypted_payload"], key) + # Commitment check over the ciphertext (C-1) — proves the server + # hasn't swapped the blob. Done before decryption since it no longer + # needs the plaintext. Legacy v1 intents skip this check. + integrity_verified = _verify_commitment(data) - # Hybrid Commitment Scheme — proves the server hasn't swapped the - # ciphertext. Server is blind so it can't forge a matching hash. - # Legacy intents (no hash stored) still verify, just without this check. - commitment_hash = data.get("commitment_hash") - integrity_verified = False - if commitment_hash: - actual_hash = compute_commitment_hash(decrypted) - if actual_hash != commitment_hash: - raise ValidPayError( - "integrity_failure", - "INTEGRITY VERIFICATION FAILED — the decrypted payload does not match " - "the commitment hash stored at issuance. This may indicate server-side " - "tampering or payload corruption.", - ) - integrity_verified = True + decrypted = decrypt(data["encrypted_payload"], key) try: payload = json.loads(decrypted) @@ -434,21 +449,12 @@ def verify_split_key_intent( share_b = self._fetch_fragment_b(retrieval_id) + # Commitment check over the ciphertext (C-1); legacy v1 skips. + integrity_verified = _verify_commitment(data) + full_key = combine_key_shares(share_a, share_b) decrypted = decrypt(data["encrypted_payload"], full_key) - commitment_hash = data.get("commitment_hash") - integrity_verified = False - if commitment_hash: - actual_hash = compute_commitment_hash(decrypted) - if actual_hash != commitment_hash: - raise ValidPayError( - "integrity_failure", - "INTEGRITY VERIFICATION FAILED — the decrypted payload does not match " - "the commitment hash stored at issuance.", - ) - integrity_verified = True - try: payload = json.loads(decrypted) except json.JSONDecodeError as exc: @@ -529,9 +535,11 @@ def create_selective_intent( key_map = build_key_map(field_keys, disclosure_policy) encrypted_key_map = encrypt(json.dumps(key_map), master_key) - full_plaintext = json.dumps(payload) - commitment_hash = compute_commitment_hash(full_plaintext) envelope = json.dumps(encrypted_fields) + # Commitment v2: hash the transported ciphertext envelope, not the + # plaintext (C-1). Role-independent — the server hashes this exact + # string and the verifier recomputes it. + commitment_hash = compute_commitment_hash(envelope) qr_key = master_key key_fragment_b: Optional[str] = None @@ -678,20 +686,10 @@ def verify_selective_intent( payload = decrypt_fields(encrypted_fields, field_keys) - integrity_verified = False - commitment_hash = data.get("commitment_hash") - if commitment_hash and role == "full": - all_keys = key_map.get("full", {}) - full_payload = decrypt_fields(encrypted_fields, all_keys) - full_plaintext = json.dumps(full_payload) - actual_hash = compute_commitment_hash(full_plaintext) - if actual_hash != commitment_hash: - raise ValidPayError( - "integrity_failure", - "INTEGRITY VERIFICATION FAILED — the decrypted payload does not match " - "the commitment hash stored at issuance.", - ) - integrity_verified = True + # Commitment over the ciphertext envelope (C-1) — role-independent + # now, since it no longer requires decrypting the full payload. + # Legacy v1 intents skip this check. + integrity_verified = _verify_commitment(data) valid_from_str = data.get("valid_from") valid_until_str = data.get("valid_until") diff --git a/validpay/crypto.py b/validpay/crypto.py index e984a49..c7a053f 100644 --- a/validpay/crypto.py +++ b/validpay/crypto.py @@ -75,16 +75,20 @@ def combine_key_shares(share_a: str, share_b: str) -> str: return base64.b64encode(key_bytes).decode("ascii") -def compute_commitment_hash(plaintext: str) -> str: - """SHA-256 commitment hash of the plaintext payload (Hybrid Commitment Scheme). - - Computed at issuance and stored alongside the ciphertext on the server. - At verification time, the same hash is recomputed against the freshly - decrypted plaintext; a mismatch proves the server tampered with or - swapped the ciphertext, since SHA-256 is one-way and the server cannot - forge a matching hash without the decryption key. +def compute_commitment_hash(ciphertext_b64: str) -> str: + """SHA-256 commitment hash over the *ciphertext* blob (commitment v2). + + Pass the base64 ValidPay wire blob returned by :func:`encrypt` — NOT the + plaintext. Hashing the ciphertext lets the server publish the commitment + on the public verify endpoint without creating a confirmation oracle: + SHA-256(plaintext) over a low-entropy structured document (a check, an + SSN card) can be brute-forced offline to recover contents without the + key, which broke the "we cannot read your documents" promise (Prompt 097 + C-1). The commitment still proves the server hasn't swapped the blob + between issuance and verification — the verifier recomputes + SHA-256(ciphertext) and compares. """ - return hashlib.sha256(plaintext.encode("utf-8")).hexdigest() + return hashlib.sha256(ciphertext_b64.encode("utf-8")).hexdigest() def _decode_key(key: str) -> bytes: