feat(097-M5): bind document metadata as AES-GCM AAD

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

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)

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

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,14 +137,17 @@ 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)
 
         body: Dict[str, Any] = {
             "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)

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