fix(097-C1): commitment hash over ciphertext, version-gated verify

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"]}),
     ])

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")

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: