diff --git a/tests/test_client.py b/tests/test_client.py index f1156b6..f143697 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -11,6 +11,7 @@ CreateIntentResult, ValidPayClient, ValidPayError, + build_aad, build_key_map, combine_key_shares, compute_commitment_hash, @@ -96,7 +97,10 @@ def test_create_intent_encrypts_locally_and_never_sends_key(): # Share A (returned) XOR Share B (sent) reconstructs the full key. full_key = combine_key_shares(result.key, sent_body["key_fragment_b"]) assert full_key not in full_call # full key never on the wire either - decrypted = json.loads(decrypt(sent_body["encrypted_payload"], full_key)) + # M-5: the blob is AAD-bound, so pass the same AAD the create call used. + decrypted = json.loads( + decrypt(sent_body["encrypted_payload"], full_key, build_aad("ssn_card")) + ) assert decrypted == payload @@ -120,7 +124,10 @@ def test_create_intent_split_key_false_is_legacy_full_key(): sent_body = json.loads(session.calls[0]["data"]) assert "split_key" not in sent_body assert "key_fragment_b" not in sent_body - decrypted = json.loads(decrypt(sent_body["encrypted_payload"], result.key)) + # M-5: create_intent binds AAD even for the legacy single-key flow. + decrypted = json.loads( + decrypt(sent_body["encrypted_payload"], result.key, build_aad("check")) + ) assert decrypted == payload @@ -273,8 +280,8 @@ def test_create_intent_batch_encrypts_each_payload_with_unique_key(): assert results[0].key not in full_call assert results[1].key not in full_call - assert json.loads(decrypt(sent["intents"][0]["encrypted_payload"], results[0].key)) == inputs[0]["payload"] - assert json.loads(decrypt(sent["intents"][1]["encrypted_payload"], results[1].key)) == inputs[1]["payload"] + assert json.loads(decrypt(sent["intents"][0]["encrypted_payload"], results[0].key, build_aad(inputs[0]["document_type"]))) == inputs[0]["payload"] + assert json.loads(decrypt(sent["intents"][1]["encrypted_payload"], results[1].key, build_aad(inputs[1]["document_type"]))) == inputs[1]["payload"] def test_create_intent_batch_rejects_empty_and_oversized(): @@ -633,6 +640,10 @@ def test_split_key_round_trip(): "split_key": True, "commitment_hash": captured_body["commitment_hash"], "commitment_version": 2, + # M-5: echo the fields the verifier rebuilds the AAD from. The + # create call bound build_aad("ssn_card", None, None). + "encryption_version": 2, + "document_type": "ssn_card", }), _FakeResponse(200, {"intent_id": "vp_sk_2", "fragment_b": fragment_b}), ]) @@ -673,6 +684,9 @@ def test_verify_intent_delegates_to_split_key_flow(): "split_key": True, "commitment_hash": sent_body["commitment_hash"], "commitment_version": 2, + # M-5: verify reconstructs the AAD from these. + "encryption_version": 2, + "document_type": "check", }), _FakeResponse(200, {"intent_id": "vp_sk_3", "fragment_b": sent_body["key_fragment_b"]}), ]) @@ -746,7 +760,8 @@ def test_split_key_xor_relationship_holds_via_combine_helper(): sent = json.loads(session.calls[0]["data"]) full_key = combine_key_shares(result.key, sent["key_fragment_b"]) # Decrypting with the recombined key should work — that's what the verifier does. - assert decrypt(sent["encrypted_payload"], full_key) == json.dumps({"amount": 1}) + # M-5: the blob is AAD-bound, so pass the same AAD the create call used. + assert decrypt(sent["encrypted_payload"], full_key, build_aad("check")) == json.dumps({"amount": 1}) def test_network_error_wrapped_as_validpay_error(): @@ -1134,15 +1149,19 @@ def test_verify_intent_no_time_lock(): def test_verify_split_key_intent_time_lock(): payload = {"a": 1} + # The validity window must be bound at creation (M-5 AAD), so compute it + # first and pass it to create; the verify response echoes the same window. + vf = _iso_in(-3600) + vu = _iso_in(3600) create_session = _FakeSession([ _FakeResponse(201, {"retrieval_id": "vp_tl_sk", "status": "active", "split_key": True}), ]) creator = ValidPayClient(api_key="k", base_url="https://api.example.test", session=create_session) - create_result = creator.create_split_key_intent(document_type="check", payload=payload) + create_result = creator.create_split_key_intent( + document_type="check", payload=payload, valid_from=vf, valid_until=vu + ) sent = json.loads(create_session.calls[0]["data"]) - vf = _iso_in(-3600) - vu = _iso_in(3600) verify_session = _FakeSession([ _FakeResponse(200, { "intent_id": "vp_tl_sk", @@ -1153,6 +1172,9 @@ def test_verify_split_key_intent_time_lock(): "status": "active", "split_key": True, "commitment_hash": sent["commitment_hash"], + "commitment_version": 2, + "encryption_version": 2, + "document_type": "check", "valid_from": vf, "valid_until": vu, }), diff --git a/tests/test_crypto.py b/tests/test_crypto.py index b775bf3..9b9817e 100644 --- a/tests/test_crypto.py +++ b/tests/test_crypto.py @@ -140,6 +140,38 @@ def test_commitment_hash_changes_with_input(): assert compute_commitment_hash("a") != compute_commitment_hash("b") +def test_aad_round_trip(): + from validpay.crypto import build_aad, decrypt, encrypt, generate_key + + key = generate_key() + aad = build_aad("check", None, "2026-08-01T00:00:00Z") + blob = encrypt('{"amount": 100}', key, aad) + assert decrypt(blob, key, aad) == '{"amount": 100}' + + +def test_aad_mismatch_fails(): + from validpay.crypto import ValidPayError, build_aad, decrypt, encrypt, generate_key + import pytest + + key = generate_key() + blob = encrypt('{"amount": 100}', key, build_aad("check")) + # Altered document_type → GCM tag check fails. + with pytest.raises(ValidPayError) as exc: + decrypt(blob, key, build_aad("other")) + assert exc.value.code == "decryption_failed" + + +def test_aad_is_canonical_compact_with_epoch_ms(): + from validpay.crypto import build_aad + + # Compact (no spaces), fixed key order, epoch-ms timestamps — must stay + # byte-identical to the JS SDKs / website verifier. + assert ( + build_aad("check", None, "2026-08-01T00:00:00Z") + == '{"document_type":"check","valid_from":null,"valid_until":1785542400000}' + ) + + def test_split_key_produces_two_32_byte_shares(): key = generate_key() a, b = split_key(key) diff --git a/validpay/__init__.py b/validpay/__init__.py index f5d6c58..2cd1ad7 100644 --- a/validpay/__init__.py +++ b/validpay/__init__.py @@ -14,6 +14,7 @@ ) from .client import ValidPayClient from .crypto import ( + build_aad, build_key_map, combine_key_shares, compute_commitment_hash, @@ -37,6 +38,7 @@ "encrypt", "decrypt", "compute_commitment_hash", + "build_aad", "split_key", "combine_key_shares", "encrypt_fields", diff --git a/validpay/client.py b/validpay/client.py index 5fba14d..2f53bf6 100644 --- a/validpay/client.py +++ b/validpay/client.py @@ -10,6 +10,7 @@ from ._timelock import compute_time_lock_status as _compute_time_lock_status from ._timelock import validate_time_lock as _validate_time_lock from .crypto import ( + build_aad, build_key_map, combine_key_shares, compute_commitment_hash, @@ -50,6 +51,19 @@ def _verify_commitment(data: Mapping[str, Any]) -> bool: return True +def _aad_for(data: Mapping[str, Any]) -> Optional[str]: + """AAD to pass to decrypt (Prompt 097 M-5). For v2 intents, reconstruct it + from the server-returned metadata so a server that altered document_type + or the validity window fails the GCM tag check. None for legacy v1.""" + if not isinstance(data.get("encryption_version"), int) or data["encryption_version"] < 2: + return None + return build_aad( + data.get("document_type", ""), + data.get("valid_from"), + data.get("valid_until"), + ) + + class ValidPayClient: """Client for the ValidPay API. @@ -123,7 +137,9 @@ def create_intent( result_key, share_b = split_key_fn(full_key) plaintext = json.dumps(payload) - encrypted_payload = encrypt(plaintext, full_key) + # M-5: bind document_type + validity window as AAD. + aad = build_aad(document_type, valid_from, valid_until) + encrypted_payload = encrypt(plaintext, full_key, aad) # Commitment v2: hash the ciphertext, not the plaintext (C-1). commitment_hash = compute_commitment_hash(encrypted_payload) @@ -131,6 +147,7 @@ def create_intent( "document_type": document_type, "encrypted_payload": encrypted_payload, "commitment_hash": commitment_hash, + "encryption_version": 2, } if split_key: body["split_key"] = True @@ -207,12 +224,15 @@ def create_intent_batch( key = generate_key() keys.append(key) plaintext = json.dumps(item["payload"]) - encrypted_payload = encrypt(plaintext, key) + # M-5: bind document_type + validity window as AAD per item. + aad = build_aad(doc_type, valid_from, valid_until) + encrypted_payload = encrypt(plaintext, key, aad) req_item: Dict[str, Any] = { "document_type": doc_type, "encrypted_payload": encrypted_payload, # Commitment v2: hash the ciphertext, not the plaintext (C-1). "commitment_hash": compute_commitment_hash(encrypted_payload), + "encryption_version": 2, } if valid_from is not None: req_item["valid_from"] = valid_from @@ -343,7 +363,9 @@ def verify_intent( # needs the plaintext. Legacy v1 intents skip this check. integrity_verified = _verify_commitment(data) - decrypted = decrypt(data["encrypted_payload"], key) + # M-5: pass the reconstructed AAD for v2 intents so altered metadata + # fails the GCM tag check. + decrypted = decrypt(data["encrypted_payload"], key, _aad_for(data)) try: payload = json.loads(decrypted) @@ -453,7 +475,8 @@ def verify_split_key_intent( integrity_verified = _verify_commitment(data) full_key = combine_key_shares(share_a, share_b) - decrypted = decrypt(data["encrypted_payload"], full_key) + # M-5: AAD bound for v2 intents. + decrypted = decrypt(data["encrypted_payload"], full_key, _aad_for(data)) try: payload = json.loads(decrypted) diff --git a/validpay/crypto.py b/validpay/crypto.py index c7a053f..4132fe5 100644 --- a/validpay/crypto.py +++ b/validpay/crypto.py @@ -104,26 +104,36 @@ def _decode_key(key: str) -> bytes: return buf -def encrypt(plaintext: str, key: str) -> str: +def encrypt(plaintext: str, key: str, aad: str | None = None) -> str: """Encrypt ``plaintext`` (UTF-8) with the given base64 AES-256 key. Returns a base64 string in the ValidPay wire format:: base64(iv[12] || authTag[16] || ciphertext) + + ``aad`` (Prompt 097 M-5) is optional Associated Authenticated Data — when + supplied, the same string MUST be passed to :func:`decrypt` or the GCM tag + check fails. Use :func:`build_aad` to bind document metadata. """ key_bytes = _decode_key(key) iv = os.urandom(_IV_BYTES) aesgcm = AESGCM(key_bytes) + aad_bytes = aad.encode("utf-8") if aad else None # cryptography's AESGCM.encrypt returns ``ciphertext || authTag``. # Rearrange to match the ValidPay/Node wire format: iv || authTag || ciphertext. - ct_with_tag = aesgcm.encrypt(iv, plaintext.encode("utf-8"), None) + ct_with_tag = aesgcm.encrypt(iv, plaintext.encode("utf-8"), aad_bytes) auth_tag = ct_with_tag[-_TAG_BYTES:] ciphertext = ct_with_tag[:-_TAG_BYTES] return base64.b64encode(iv + auth_tag + ciphertext).decode("ascii") -def decrypt(blob: str, key: str) -> str: - """Decrypt a ValidPay-format base64 blob and return the plaintext (UTF-8).""" +def decrypt(blob: str, key: str, aad: str | None = None) -> str: + """Decrypt a ValidPay-format base64 blob and return the plaintext (UTF-8). + + ``aad`` must match the value passed to :func:`encrypt` (Prompt 097 M-5); + a mismatch (e.g. a server that altered the bound metadata) raises + ``decryption_failed``. + """ key_bytes = _decode_key(key) try: @@ -142,17 +152,56 @@ def decrypt(blob: str, key: str) -> str: ciphertext = buf[_IV_BYTES + _TAG_BYTES:] aesgcm = AESGCM(key_bytes) + aad_bytes = aad.encode("utf-8") if aad else None try: - plaintext = aesgcm.decrypt(iv, ciphertext + auth_tag, None) + plaintext = aesgcm.decrypt(iv, ciphertext + auth_tag, aad_bytes) except InvalidTag as exc: raise ValidPayError( "decryption_failed", - "Decryption failed — wrong key or tampered blob", + "Decryption failed — wrong key, tampered blob, or altered bound metadata", ) from exc return plaintext.decode("utf-8") +def build_aad( + document_type: str, + valid_from: str | None = None, + valid_until: str | None = None, +) -> str: + """Canonical AAD for AES-GCM metadata binding (Prompt 097 M-5). + + Binds ``document_type`` and the validity window so a blind/compromised + server cannot silently alter them (e.g. change "check" to "other"). The + output MUST be byte-identical across every SDK and the website verifier, + so: + + - keys are emitted in a fixed order (document_type, valid_from, valid_until); + - JSON is compact (no spaces) — matches JS ``JSON.stringify``; + - timestamps are normalized to **epoch milliseconds** rather than raw + ISO strings. The server reformats timestamps (``2026-08-01T00:00:00Z`` + becomes ``...:00.000Z``), so binding the raw string would break + verification of legitimate time-locked documents. Epoch ms parse + identically on both sides regardless of ISO formatting. + """ + payload = { + "document_type": document_type, + "valid_from": _epoch_ms(valid_from), + "valid_until": _epoch_ms(valid_until), + } + return json.dumps(payload, separators=(",", ":")) + + +def _epoch_ms(iso: str | None) -> int | None: + if not iso: + return None + from datetime import datetime + + # Accept the trailing 'Z' (UTC) that fromisoformat rejected before 3.11. + parsed = datetime.fromisoformat(iso.replace("Z", "+00:00")) + return int(parsed.timestamp() * 1000) + + def encrypt_fields(payload: dict, generate_key_fn=None) -> tuple[dict, dict]: """Encrypt each field in payload separately (Selective Field Disclosure, Patent E).