Spaces:
Running
Running
File size: 4,666 Bytes
01a4407 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | 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()
|