fix: make receipt export bounded and evidence-honest

killinchu_receipt_export.py

@@ -0,0 +1,256 @@
+"""Bounded, evidence-honest Killinchu receipt export contract.
+
+This module is deliberately pure stdlib so the export state can be tested
+without importing the full FastAPI application. A transport-success response
+does not imply that a receipt or signature exists: callers must inspect
+``export_state``, ``receipt_available`` and ``verification.state``.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+from datetime import datetime, timezone
+from typing import Any, Mapping, Sequence
+
+
+SCHEMA = "szl.killinchu.receipt-export/v1"
+DEFAULT_PUBLIC_KEY_URL = "https://szlholdings-killinchu.hf.space/cosign.pub"
+
+
+def _now_utc() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _base(doctrine: str, public_key_url: str) -> dict[str, Any]:
+    return {
+        "schema": SCHEMA,
+        "ok": True,
+        "doctrine": doctrine,
+        "generated_at": _now_utc(),
+        "public_key_url": public_key_url,
+    }
+
+
+def build_receipt_export(
+    dag: Sequence[Mapping[str, Any]],
+    *,
+    index: int = -1,
+    doctrine: str = "v11",
+    dsse_module: Any = None,
+    khipu_root: str | None = None,
+    public_key_url: str = DEFAULT_PUBLIC_KEY_URL,
+) -> tuple[dict[str, Any], int]:
+    """Return ``(body, status_code)`` for the public receipt export route.
+
+    The empty ledger is a valid, explicit runtime state and therefore returns
+    HTTP 200 with no receipt and no signature claim. A non-empty export is
+    marked signed only when the configured DSSE verifier validates it.
+    Out-of-range indexes are rejected rather than silently selecting a
+    different receipt.
+    """
+
+    out = _base(doctrine, public_key_url)
+    ledger_size = len(dag)
+
+    if ledger_size == 0:
+        out.update(
+            {
+                "export_state": "EMPTY",
+                "receipt_available": False,
+                "ledger_size": 0,
+                "node_index": None,
+                "node_digest": None,
+                "khipu_root": None,
+                "dsse": None,
+                "payload_b64": None,
+                "signed": False,
+                "keyid": None,
+                "verification": {
+                    "state": "NOT_APPLICABLE",
+                    "verified": False,
+                    "reason": "no receipt exists in the in-memory ledger",
+                },
+                "verify_offline": [],
+                "limits": [
+                    {
+                        "code": "NO_RECEIPTS",
+                        "detail": "No receipt has been emitted since this runtime started.",
+                    },
+                    {
+                        "code": "EPHEMERAL_LEDGER",
+                        "detail": "The in-memory Khipu ledger resets on Space restart.",
+                    },
+                ],
+                "honesty": (
+                    "The export route is reachable, but no receipt exists. "
+                    "This response is not a receipt and carries no signature."
+                ),
+            }
+        )
+        return out, 200
+
+    if index < -ledger_size or index >= ledger_size:
+        out.update(
+            {
+                "ok": False,
+                "export_state": "INDEX_OUT_OF_RANGE",
+                "receipt_available": False,
+                "ledger_size": ledger_size,
+                "requested_index": index,
+                "valid_index": {"min": -ledger_size, "max": ledger_size - 1},
+                "signed": False,
+                "verification": {
+                    "state": "NOT_APPLICABLE",
+                    "verified": False,
+                    "reason": "requested index is outside the bounded ledger",
+                },
+                "limits": [
+                    {
+                        "code": "INDEX_OUT_OF_RANGE",
+                        "detail": "Select an index inside valid_index; no fallback receipt was substituted.",
+                    }
+                ],
+            }
+        )
+        return out, 422
+
+    node = dag[index]
+    receipt_obj = node.get("receipt")
+    raw_dsse = node.get("dsse")
+    dsse: dict[str, Any] = dict(raw_dsse) if isinstance(raw_dsse, Mapping) else {}
+    raw_signatures = dsse.get("signatures")
+    signatures = [dict(s) for s in raw_signatures or [] if isinstance(s, Mapping)]
+
+    # A literal placeholder is an honesty marker, never a signature. Do not
+    # expose it in the DSSE signatures array where generic clients may mistake
+    # presence for cryptographic proof.
+    has_placeholder = any(
+        s.get("keyid") in (None, "PENDING")
+        or "PLACEHOLDER" in str(s.get("sig", "")).upper()
+        for s in signatures
+    )
+    if has_placeholder:
+        signatures = []
+    dsse["signatures"] = signatures
+
+    reconstruction_error: str | None = None
+    payload_b64: str | None = None
+    pae_sha256: str | None = None
+    if receipt_obj is None:
+        reconstruction_error = "receipt object missing from ledger node"
+    elif dsse_module is None:
+        reconstruction_error = "DSSE module unavailable in this runtime"
+    else:
+        try:
+            body = dsse_module.canonical_json(receipt_obj)
+            payload_type = dsse.get("payloadType", "application/vnd.szl.receipt+json")
+            payload_b64 = base64.b64encode(body).decode("ascii")
+            pae_sha256 = hashlib.sha256(dsse_module.pae(payload_type, body)).hexdigest()
+            dsse.update(
+                {
+                    "payloadType": payload_type,
+                    "payload": payload_b64,
+                    "_dsse": "DSSEv1",
+                    "_pae_sha256": pae_sha256,
+                }
+            )
+        except Exception:
+            reconstruction_error = "DSSE payload reconstruction failed"
+
+    verification: dict[str, Any]
+    if not signatures:
+        verification = {
+            "state": "UNSIGNED",
+            "verified": False,
+            "reason": "no genuine signature is present",
+        }
+    elif reconstruction_error:
+        verification = {
+            "state": "UNAVAILABLE",
+            "verified": None,
+            "reason": reconstruction_error,
+        }
+    elif not hasattr(dsse_module, "verify_envelope"):
+        verification = {
+            "state": "UNAVAILABLE",
+            "verified": None,
+            "reason": "DSSE verifier unavailable in this runtime",
+        }
+    else:
+        try:
+            verdict = dsse_module.verify_envelope(dsse)
+            verified = bool(isinstance(verdict, Mapping) and verdict.get("verified") is True)
+            verification = {
+                "state": "VERIFIED" if verified else "FAILED",
+                "verified": verified,
+                "reason": (
+                    "signature verified against the published public key"
+                    if verified
+                    else str(verdict.get("reason", "signature verification failed"))
+                ),
+                "pae_sha256": verdict.get("pae_sha256", pae_sha256),
+            }
+        except Exception:
+            verification = {
+                "state": "UNAVAILABLE",
+                "verified": None,
+                "reason": "DSSE verifier raised; no signed claim emitted",
+            }
+
+    signed = verification.get("verified") is True
+    dsse["signed"] = signed
+    export_state = (
+        "SIGNED_VERIFIED"
+        if signed
+        else "UNSIGNED"
+        if not signatures
+        else "SIGNATURE_UNVERIFIED"
+    )
+    keyid = signatures[0].get("keyid") if signatures else None
+    limits = [
+        {
+            "code": "EPHEMERAL_LEDGER",
+            "detail": "The in-memory Khipu ledger resets on Space restart.",
+        }
+    ]
+    if not signed:
+        limits.append(
+            {
+                "code": "NO_VERIFIED_SIGNATURE",
+                "detail": verification["reason"],
+            }
+        )
+
+    verify_offline = []
+    if signed:
+        verify_offline = [
+            f"curl -s {public_key_url} -o cosign.pub",
+            "base64 -d payload.b64 > payload.bin",
+            "cosign verify-blob --key cosign.pub --signature sig.b64 payload.bin",
+        ]
+
+    out.update(
+        {
+            "export_state": export_state,
+            "receipt_available": True,
+            "ledger_size": ledger_size,
+            "node_index": node.get("index", index % ledger_size),
+            "node_digest": node.get("digest"),
+            "khipu_root": khipu_root,
+            "dsse": dsse,
+            "payload_b64": payload_b64,
+            "signed": signed,
+            "keyid": keyid,
+            "verification": verification,
+            "verify_offline": verify_offline,
+            "limits": limits,
+            "honesty": (
+                "The DSSE signature was verified against the published public key."
+                if signed
+                else "A receipt exists, but this response makes no verified-signature claim."
+            ),
+        }
+    )
+    return out, 200
+

serve.py

@@ -41,6 +41,7 @@ from fastapi.staticfiles import StaticFiles
 from starlette.middleware.cors import CORSMiddleware
 
 import killinchu_protocols as kp
+from killinchu_receipt_export import build_receipt_export
 
 _APP_ROOT = Path(os.environ.get("KILLINCHU_ROOT", "/app"))
 STATIC_DIR = _APP_ROOT / "static"
@@ -2217,64 +2218,20 @@ async def cosign_pub() -> Response:
 
 @app.get("/api/killinchu/v1/receipt/export")
 async def receipt_export(index: int = -1) -> JSONResponse:
-    """Export ONE signed receipt + step-by-step offline verify instructions.
+    """Export one bounded receipt or an explicit, typed empty state.
 
-    The on-stage 'verify it yourself' artifact: gives the judge the DSSE
-    envelope, the public-key URL, and the exact two commands to verify the
-    signature with cosign — no trust in killinchu's infrastructure required.
+    ``signed`` is true only after the reconstructed DSSE envelope verifies
+    against the public key. Missing receipts or signing capability are never
+    represented by a fabricated envelope.
     """
-    if not _KHIPU_DAG:
-        return JSONResponse({"ok": False, "error": "no receipts yet — run a /beyond demo first"},
-                            status_code=404)
-    try:
-        node = _KHIPU_DAG[index]
-    except IndexError:
-        node = _KHIPU_DAG[-1]
-    dsse = dict(node.get("dsse") or {})
-    sigs = dsse.get("signatures") or []
-    signed = bool(sigs and sigs[0].get("keyid") not in (None, "PENDING"))
-    # RECONSTRUCT the full verifiable DSSE envelope. The signature was computed by
-    # szl_dsse.sign_payload over canonical_json(node["receipt"]); the stored node
-    # kept only the signatures, so re-derive the exact base64 payload here. Without
-    # this, an offline `cosign verify-blob` would have nothing to verify against.
-    payload_b64 = None
-    pae_sha256 = None
-    receipt_obj = node.get("receipt")
-    if receipt_obj is not None and _szl_dsse is not None:
-        try:
-            import base64 as _b64_exp
-            body = _szl_dsse.canonical_json(receipt_obj)
-            payload_b64 = _b64_exp.b64encode(body).decode("ascii")
-            ptype = dsse.get("payloadType", "application/vnd.szl.receipt+json")
-            pae_sha256 = hashlib.sha256(_szl_dsse.pae(ptype, body)).hexdigest()
-            dsse["payload"] = payload_b64
-            dsse["_dsse"] = "DSSEv1"
-            dsse["_pae_sha256"] = pae_sha256
-        except Exception:
-            pass
-    return JSONResponse({
-        "ok": True,
-        "node_index": node.get("index"),
-        "node_digest": node.get("digest"),
-        "khipu_root": _khipu_root(),
-        "dsse": dsse,
-        "payload_b64": payload_b64,
-        "signed": signed,
-        "keyid": (sigs[0].get("keyid") if sigs else None),
-        "public_key_url": "https://szlholdings-killinchu.hf.space/cosign.pub",
-        "verify_offline": [
-            "# 1. Save the public key",
-            "curl -s https://szlholdings-killinchu.hf.space/cosign.pub -o cosign.pub",
-            "# 2. Save this DSSE envelope's payload + signature, then:",
-            "cosign verify-blob --key cosign.pub --signature sig.b64 payload.bin",
-            "# OR verify the whole DAG via the live endpoint:",
-            "curl -s -X POST https://szlholdings-killinchu.hf.space/khipu/verify -H 'Content-Type: application/json' -d @receipt.json",
-        ],
-        "honesty": ("REAL ECDSA-P256-SHA256 DSSE when signed=true (keyid szlholdings-cosign); "
-                    "an honest UNSIGNED placeholder otherwise. The public key is the same one "
-                    "published at szl-holdings/.github/cosign.pub — verify without trusting us."),
-        "doctrine": DOCTRINE,
-    })
+    body, status_code = build_receipt_export(
+        _KHIPU_DAG,
+        index=index,
+        doctrine=DOCTRINE,
+        dsse_module=_szl_dsse,
+        khipu_root=_khipu_root(),
+    )
+    return JSONResponse(body, status_code=status_code)
 
 
 @app.get("/api/killinchu/v1/lambda")

tests/test_receipt_export_contract.py

@@ -0,0 +1,116 @@
+import base64
+import hashlib
+import hmac
+import json
+import unittest
+
+from killinchu_receipt_export import SCHEMA, build_receipt_export
+
+
+class _TestDSSE:
+    """Deterministic verifier used to test the route contract without secrets."""
+
+    secret = b"test-only-key"
+
+    @staticmethod
+    def canonical_json(obj):
+        return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")
+
+    @staticmethod
+    def pae(payload_type, body):
+        t = payload_type.encode("utf-8")
+        return b"DSSEv1 " + str(len(t)).encode() + b" " + t + b" " + str(len(body)).encode() + b" " + body
+
+    @classmethod
+    def signature(cls, payload_type, receipt):
+        body = cls.canonical_json(receipt)
+        return base64.b64encode(hmac.new(cls.secret, cls.pae(payload_type, body), hashlib.sha256).digest()).decode()
+
+    @classmethod
+    def verify_envelope(cls, env):
+        body = base64.b64decode(env["payload"])
+        expected = hmac.new(cls.secret, cls.pae(env["payloadType"], body), hashlib.sha256).digest()
+        actual = base64.b64decode(env["signatures"][0]["sig"])
+        verified = hmac.compare_digest(actual, expected)
+        return {
+            "verified": verified,
+            "reason": "test signature mismatch" if not verified else "verified",
+            "pae_sha256": hashlib.sha256(cls.pae(env["payloadType"], body)).hexdigest(),
+        }
+
+
+def _node(signature):
+    receipt = {
+        "schema": "szl.killinchu.receipt/v1",
+        "kind": "contract-test",
+        "payload": {"bounded": True},
+        "doctrine": "v11",
+        "ts_utc": "2026-07-11T00:00:00Z",
+    }
+    return receipt, {
+        "index": 0,
+        "digest": "a" * 64,
+        "receipt": receipt,
+        "dsse": {
+            "payloadType": "application/vnd.szl.receipt+json",
+            "signatures": [{"keyid": "test-key", "sig": signature}],
+        },
+    }
+
+
+class ReceiptExportContractTests(unittest.TestCase):
+    def test_empty_ledger_is_typed_200_with_explicit_limit(self):
+        body, status = build_receipt_export([], dsse_module=_TestDSSE)
+        self.assertEqual(status, 200)
+        self.assertEqual(body["schema"], SCHEMA)
+        self.assertEqual(body["export_state"], "EMPTY")
+        self.assertFalse(body["receipt_available"])
+        self.assertFalse(body["signed"])
+        self.assertIsNone(body["dsse"])
+        self.assertEqual(body["verification"]["state"], "NOT_APPLICABLE")
+        self.assertIn("NO_RECEIPTS", {item["code"] for item in body["limits"]})
+
+    def test_valid_signature_is_exported_only_after_verification(self):
+        receipt, _ = _node("")
+        sig = _TestDSSE.signature("application/vnd.szl.receipt+json", receipt)
+        _, node = _node(sig)
+        body, status = build_receipt_export([node], dsse_module=_TestDSSE, khipu_root=node["digest"])
+        self.assertEqual(status, 200)
+        self.assertEqual(body["export_state"], "SIGNED_VERIFIED")
+        self.assertTrue(body["signed"])
+        self.assertEqual(body["verification"]["state"], "VERIFIED")
+        self.assertTrue(body["dsse"]["payload"])
+        self.assertTrue(body["verify_offline"])
+
+    def test_invalid_signature_never_becomes_signed(self):
+        _, node = _node(base64.b64encode(b"not-a-valid-signature").decode())
+        body, status = build_receipt_export([node], dsse_module=_TestDSSE)
+        self.assertEqual(status, 200)
+        self.assertEqual(body["export_state"], "SIGNATURE_UNVERIFIED")
+        self.assertFalse(body["signed"])
+        self.assertEqual(body["verification"]["state"], "FAILED")
+        self.assertEqual(body["verify_offline"], [])
+        self.assertIn("NO_VERIFIED_SIGNATURE", {item["code"] for item in body["limits"]})
+
+    def test_placeholder_is_not_exposed_as_a_signature(self):
+        _, node = _node("PLACEHOLDER - not a signature")
+        node["dsse"]["signatures"][0]["keyid"] = "PENDING"
+        body, status = build_receipt_export([node], dsse_module=_TestDSSE)
+        self.assertEqual(status, 200)
+        self.assertEqual(body["export_state"], "UNSIGNED")
+        self.assertFalse(body["signed"])
+        self.assertEqual(body["dsse"]["signatures"], [])
+        self.assertEqual(body["verification"]["state"], "UNSIGNED")
+
+    def test_out_of_range_index_is_rejected_without_fallback(self):
+        _, node = _node(base64.b64encode(b"x").decode())
+        body, status = build_receipt_export([node], index=3, dsse_module=_TestDSSE)
+        self.assertEqual(status, 422)
+        self.assertFalse(body["ok"])
+        self.assertEqual(body["export_state"], "INDEX_OUT_OF_RANGE")
+        self.assertEqual(body["requested_index"], 3)
+
+
+if __name__ == "__main__":
+    unittest.main()
+