Canonical verify origin → verify.keyhalve.com (source only, no publish)

check21/test_check21.py

@@ -27,7 +27,7 @@
 # --- constants ---------------------------------------------------------------
 
 ENCODED_DATA = (
-    "https://validpay.com/verify/vp_abc123def456"
+    "https://verify.keyhalve.com/verify/vp_abc123def456"
     "#k=dGVzdGtleV9iYXNlNjRfZW5jb2RlZF8zMl9ieXRlcw"
 )
 

tests/test_end_cell.py

@@ -0,0 +1,107 @@
+"""End-Cell tests: crypto round-trip + rail fetch/verify (pinned key)."""
+
+from __future__ import annotations
+
+import base64
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
+
+from validpay.crypto import generate_key, split_key_pieces, combine_key_pieces
+from validpay.errors import ValidPayError
+from validpay.rail import fetch_rail_piece
+
+INTENT = "vp_railtest"
+
+
+def test_split_then_combine_reconstructs_key():
+    key = generate_key()
+    for holders in (1, 2, 3):
+        parts = split_key_pieces(key, holders)
+        assert len(parts) == holders + 1  # share_a + one piece per holder
+        share_a, pieces = parts[0], parts[1:]
+        assert combine_key_pieces(share_a, pieces) == key
+
+
+def test_split_rejects_zero_pieces():
+    with pytest.raises(ValidPayError):
+        split_key_pieces(generate_key(), 0)
+
+
+# --- rail verify --------------------------------------------------------------
+
+def _make_key():
+    priv = Ed25519PrivateKey.generate()
+    spki = priv.public_key().public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo)
+    return priv, base64.b64encode(spki).decode()
+
+
+def _canonical(intent_id: str, piece: str) -> bytes:
+    return f"keyhalve-rail.v1\n{intent_id}\nkeyhalve\n{piece}".encode()
+
+
+class _Resp:
+    def __init__(self, body, status=200):
+        self._body = body
+        self.status_code = status
+        self.ok = 200 <= status < 300
+
+    def json(self):
+        return self._body
+
+
+class _Session:
+    def __init__(self, resp=None, exc=None):
+        self._resp = resp
+        self._exc = exc
+
+    def get(self, *_a, **_k):
+        if self._exc:
+            raise self._exc
+        return self._resp
+
+
+PIECE = base64.b64encode(b"\x07" * 32).decode()
+
+
+def _signed(priv, intent=INTENT, piece=PIECE, holder="keyhalve"):
+    sig = base64.b64encode(priv.sign(_canonical(intent, piece))).decode()
+    return {"holder": holder, "piece": piece, "sig": sig, "alg": "ed25519"}
+
+
+def test_rail_piece_verifies_against_pinned_key():
+    priv, spki = _make_key()
+    sess = _Session(_Resp(_signed(priv)))
+    assert fetch_rail_piece(sess, "https://rail.test", spki, INTENT, 5) == PIECE
+
+
+def test_rail_tampered_signature_fails_closed():
+    priv, spki = _make_key()
+    body = _signed(priv)
+    bad = bytearray(base64.b64decode(body["sig"]))
+    bad[0] ^= 0xFF
+    body["sig"] = base64.b64encode(bytes(bad)).decode()
+    with pytest.raises(ValidPayError, match="signature"):
+        fetch_rail_piece(_Session(_Resp(body)), "https://rail.test", spki, INTENT, 5)
+
+
+def test_rail_non_pinned_key_fails_closed():
+    attacker, _ = _make_key()
+    _, pinned_spki = _make_key()
+    with pytest.raises(ValidPayError, match="signature"):
+        fetch_rail_piece(_Session(_Resp(_signed(attacker))), "https://rail.test", pinned_spki, INTENT, 5)
+
+
+def test_rail_wrong_holder_rejected():
+    priv, spki = _make_key()
+    with pytest.raises(ValidPayError, match="(?i)malformed"):
+        fetch_rail_piece(_Session(_Resp(_signed(priv, holder="platform"))), "https://rail.test", spki, INTENT, 5)
+
+
+def test_rail_404_409_fail_closed():
+    _, spki = _make_key()
+    with pytest.raises(ValidPayError, match="not found"):
+        fetch_rail_piece(_Session(_Resp({}, 404)), "https://rail.test", spki, INTENT, 5)
+    with pytest.raises(ValidPayError, match="revoked"):
+        fetch_rail_piece(_Session(_Resp({}, 409)), "https://rail.test", spki, INTENT, 5)

tests/test_pdf.py

@@ -24,14 +24,14 @@
 
 def test_build_verify_url_basic():
     assert build_verify_url("abc123", "deadbeef") == (
-        "https://validpay.com/verify/abc123#key=deadbeef"
+        "https://verify.keyhalve.com/verify/abc123#key=deadbeef"
     )
 
 
 def test_build_verify_url_base64url_key_and_encoded_id():
     # "K+K/m==" -> base64url "K-K_m"; id is percent-encoded.
     assert build_verify_url("a/b c", "K+K/m==") == (
-        "https://validpay.com/verify/a%2Fb%20c#key=K-K_m"
+        "https://verify.keyhalve.com/verify/a%2Fb%20c#key=K-K_m"
     )
 
 

validpay/__init__.py

@@ -26,6 +26,8 @@
     encrypt_fields,
     generate_key,
     split_key,
+    split_key_pieces,
+    combine_key_pieces,
 )
 from .errors import ValidPayError
 from .offline import OfflineCache, OfflineVerifyResult
@@ -53,6 +55,8 @@
     "build_aad",
     "split_key",
     "combine_key_shares",
+    "split_key_pieces",
+    "combine_key_pieces",
     "encrypt_fields",
     "build_key_map",
     "decrypt_fields",

validpay/client.py

@@ -21,10 +21,19 @@
     encrypt_fields,
     generate_key,
     split_key as split_key_fn,
+    split_key_pieces,
+    combine_key_pieces,
 )
 from .errors import ValidPayError
+from .rail import (
+    fetch_rail_piece,
+    KEYHALVE_RAIL_BASE_URL,
+    KEYHALVE_RAIL_PUBLIC_KEY_SPKI_B64,
+)
 from .types import CreateIntentResult, VerifyIntentResult
 
+_HOLDER_KEYHALVE = "keyhalve"
+
 DEFAULT_BASE_URL = "https://api.validpay.com"
 DEFAULT_TIMEOUT = 30.0
 
@@ -81,13 +90,17 @@ def __init__(
         base_url: str = DEFAULT_BASE_URL,
         timeout: float = DEFAULT_TIMEOUT,
         session: Optional[requests.Session] = None,
+        rail_base_url: str = KEYHALVE_RAIL_BASE_URL,
+        rail_public_key_spki: str = KEYHALVE_RAIL_PUBLIC_KEY_SPKI_B64,
     ) -> None:
         if not api_key:
             raise ValidPayError("invalid_config", "api_key is required")
         self._api_key = api_key
         self._base_url = base_url.rstrip("/")
         self._timeout = timeout
         self._session = session if session is not None else requests.Session()
+        self._rail_base_url = rail_base_url
+        self._rail_public_key_spki = rail_public_key_spki
 
     def create_intent(
         self,
@@ -178,6 +191,62 @@ def create_intent(
 
         return CreateIntentResult(retrieval_id=retrieval_id, key=result_key)
 
+    def create_end_cell_intent(
+        self,
+        document_type: str,
+        payload: Any,
+        *,
+        holders: Optional[List[str]] = None,
+        valid_from: Optional[str] = None,
+        valid_until: Optional[str] = None,
+        on_behalf_of: Optional[Dict[str, str]] = None,
+    ) -> CreateIntentResult:
+        """Seal a document with End-Cell (CVCP Layer 6B): an n-of-n XOR split across
+        ShareA (returned as ``key``, embed in the QR) + one mandatory piece per holder
+        (default: the KeyHalve rail + the platform). No single party can read or
+        assemble the key. Requires API End-Cell issuance to be enabled.
+        """
+        if not document_type:
+            raise ValidPayError("invalid_argument", "document_type is required")
+        _validate_time_lock(valid_from, valid_until)
+
+        holders = holders or ["keyhalve", "platform"]
+        if len(set(holders)) != len(holders):
+            raise ValidPayError("invalid_argument", "holders must be unique")
+        if holders.count(_HOLDER_KEYHALVE) != 1 or not any(h != _HOLDER_KEYHALVE for h in holders):
+            raise ValidPayError(
+                "invalid_argument",
+                'end_cell requires exactly one "keyhalve" rail share and >= 1 platform share',
+            )
+
+        full_key = generate_key()
+        parts = split_key_pieces(full_key, len(holders))  # [share_a, piece_1, ...]
+        share_a = parts[0]
+        pieces = [{"holder": h, "piece": parts[i + 1]} for i, h in enumerate(holders)]
+
+        aad = build_aad(document_type, valid_from, valid_until)
+        encrypted_payload = encrypt(json.dumps(payload), full_key, aad)
+        body: Dict[str, Any] = {
+            "document_type": document_type,
+            "encrypted_payload": encrypted_payload,
+            "commitment_hash": compute_commitment_hash(encrypted_payload),
+            "encryption_version": 2,
+            "end_cell": True,
+            "pieces": pieces,
+        }
+        if valid_from is not None:
+            body["valid_from"] = valid_from
+        if valid_until is not None:
+            body["valid_until"] = valid_until
+        if on_behalf_of is not None:
+            body["on_behalf_of"] = on_behalf_of
+
+        data = self._request("POST", "/v1/intent", body=body, auth=True)
+        retrieval_id = data.get("retrieval_id") if isinstance(data, dict) else None
+        if not retrieval_id:
+            raise ValidPayError("invalid_response", "API response missing retrieval_id", details=data)
+        return CreateIntentResult(retrieval_id=retrieval_id, key=share_a)
+
     def create_file_intent(
         self,
         document_type: str,
@@ -393,6 +462,31 @@ def _fetch_fragment_b(self, retrieval_id: str) -> str:
             )
         return share_b
 
+    def _fetch_pieces(self, retrieval_id: str) -> List[str]:
+        """Fetch the End-Cell PLATFORM share(s) from the API /fragment endpoint.
+
+        The rail share is NOT here — it is fetched separately from the KeyHalve rail.
+        """
+        data = self._request(
+            "GET",
+            f"/v1/intent/{quote(retrieval_id, safe='')}/fragment",
+            auth=False,
+        )
+        if isinstance(data, dict) and data.get("error"):
+            raise ValidPayError(
+                str(data.get("error")),
+                f"Fragment retrieval failed: {data.get('error')}",
+                details=data,
+            )
+        pieces = data.get("pieces") if isinstance(data, dict) else None
+        if not isinstance(pieces, dict) or not pieces:
+            raise ValidPayError("missing_fragment", "Server did not return End-Cell pieces", details=data)
+        order = data.get("holders") or list(pieces.keys())
+        result = [pieces[h] for h in order if pieces.get(h)]
+        if len(result) != len(order):
+            raise ValidPayError("missing_fragment", "An End-Cell piece was missing", details=data)
+        return result
+
     def verify_intent(
         self,
         retrieval_id: str,
@@ -455,7 +549,20 @@ def verify_intent(
         # default issue path, so the key the caller holds is Share A —
         # fetch Share B from the fragment endpoint and XOR-combine, so
         # create_intent -> verify_intent round-trips keep working.
-        if data.get("split_key"):
+        if data.get("end_cell"):
+            # Custody separation: platform share(s) from the API + the rail share from
+            # the independent KeyHalve rail (signature-verified vs the pinned key),
+            # XOR-combined with ShareA. Fails closed if either is missing.
+            platform_pieces = self._fetch_pieces(retrieval_id)
+            rail_piece = fetch_rail_piece(
+                self._session,
+                self._rail_base_url,
+                self._rail_public_key_spki,
+                retrieval_id,
+                self._timeout,
+            )
+            key = combine_key_pieces(key, [*platform_pieces, rail_piece])
+        elif data.get("split_key"):
             key = combine_key_shares(key, self._fetch_fragment_b(retrieval_id))
 
         # Commitment check over the ciphertext (C-1) — proves the server

validpay/crypto.py

@@ -75,6 +75,42 @@ def combine_key_shares(share_a: str, share_b: str) -> str:
     return base64.b64encode(key_bytes).decode("ascii")
 
 
+def split_key_pieces(key: str, piece_count: int) -> list[str]:
+    """Split an AES-256 key for End-Cell (n-of-n XOR): ``[share_a, piece_1, …]``.
+
+    ``share_a`` is the QR key; the pieces go to the server-side holders. The key
+    is recovered as ``share_a XOR piece_1 XOR … XOR piece_n``. Byte-identical to the
+    other ValidPay SDKs and the verifier so a sealed intent reconstructs anywhere.
+    """
+    if piece_count < 1:
+        raise ValidPayError("invalid_argument", "piece_count must be >= 1")
+    key_bytes = _decode_key(key)
+    share_a = os.urandom(_KEY_BYTES)
+    remainder = bytearray(a ^ b for a, b in zip(key_bytes, share_a))  # = K ⊕ ShareA
+    pieces: list[bytes] = []
+    for _ in range(piece_count - 1):
+        piece = os.urandom(_KEY_BYTES)
+        for i in range(_KEY_BYTES):
+            remainder[i] ^= piece[i]
+        pieces.append(piece)
+    pieces.append(bytes(remainder))  # last piece absorbs the remainder
+    return [base64.b64encode(share_a).decode("ascii")] + [
+        base64.b64encode(p).decode("ascii") for p in pieces
+    ]
+
+
+def combine_key_pieces(share_a: str, pieces: list[str]) -> str:
+    """Reconstruct an AES-256 key from ShareA + every server-side piece (XOR)."""
+    if not pieces:
+        raise ValidPayError("invalid_argument", "need at least one piece")
+    combined = bytearray(_decode_key(share_a))
+    for piece_b64 in pieces:
+        piece = _decode_key(piece_b64)
+        for i in range(_KEY_BYTES):
+            combined[i] ^= piece[i]
+    return base64.b64encode(bytes(combined)).decode("ascii")
+
+
 def compute_commitment_hash(ciphertext_b64: str) -> str:
     """SHA-256 commitment hash over the *ciphertext* blob (commitment v2).
 

validpay/pdf.py

@@ -80,7 +80,7 @@ def _to_base64url(b64: str) -> str:
 def build_verify_url(
     retrieval_id: str,
     key: str,
-    base_url: str = "https://validpay.com",
+    base_url: str = "https://verify.keyhalve.com",
 ) -> str:
     """Build the canonical verify URL the QR encodes::
 
@@ -139,7 +139,7 @@ def embed_qr(
     key: str,
     placement: QrPlacement,
     *,
-    base_url: str = "https://validpay.com",
+    base_url: str = "https://verify.keyhalve.com",
     error_correction: str = "M",
     margin: int = 2,
     dark_color: str = "#0A0F1E",