#!/usr/bin/env python3 """Reference implementation of TCE (Trust Canonical Encoding). This file is the executable form of docs/PROTOCOL.md. It exists for two reasons: 1. To generate the frozen test vectors in testdata/vectors/. 2. To be an implementation written independently of the Go code, so that "the Go implementation is correct" means "it agrees with a second implementation", not "it agrees with itself". It is deliberately written in plain Python with no dependencies beyond `cryptography` for Ed25519, so that it can be read as a specification. Run: python3 tools/reference/tce_reference.py --emit testdata/vectors """ from __future__ import annotations import argparse import hashlib import json import re import sys from dataclasses import dataclass from pathlib import Path from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey # -------------------------------------------------------------------------- # Framing constants # -------------------------------------------------------------------------- MAGIC = b"trust.n1ko.dev/tce/1\x00" # 21 bytes PROTOCOL_VERSION = 1 # Object tags. TAG_IDENTITY = 0x01 TAG_CLAIM = 0x02 TAG_REVOCATION = 0x03 TAG_APPROVAL_REQUEST = 0x04 TAG_APPROVAL_RESPONSE = 0x05 TAG_AUTH_ASSERTION = 0x06 TAG_DELEGATION = 0x07 TAG_KEY_ROTATION = 0x08 TAG_KEY_ROTATION_CONF = 0x09 # Value tags. Separate tags for false and true remove any question of how a # boolean is encoded. V_NULL = 0x00 V_FALSE = 0x01 V_TRUE = 0x02 V_STRING = 0x03 V_NUMBER = 0x04 V_BYTES = 0x05 # reserved, rejected in v1 V_ARRAY = 0x06 # reserved, rejected in v1 V_MAP = 0x07 # reserved, rejected in v1 RESERVED_VALUE_TAGS = {V_BYTES, V_ARRAY, V_MAP} # Address / key encoding. ADDR_VERSION = 0 PUBKEY_LEN = 32 NONCE_LEN = 16 HASH_LEN = 32 SIG_LEN = 64 # Decisions. DECISION_DENY = 0 DECISION_ALLOW = 1 # Limits (see docs/PROTOCOL.md section 6). MAX_UVARINT_BYTES = 10 MAX_CLAIM_KEY = 128 MAX_STRING_VALUE = 512 MAX_NUMBER = 52 MAX_ACTION = 128 MAX_MESSAGE = 256 MAX_REASON = 256 MAX_ALIAS = 64 MAX_SCOPE = 32 MAX_AUDIENCE = 128 MAX_MAP_ENTRIES = 32 MAX_NUMBER_INT_DIGITS = 32 MAX_NUMBER_FRAC_DIGITS = 18 MAX_NUMBER_TOKEN = 64 MAX_TCE_IDENTITY = 1024 MAX_TCE_CLAIM = 4096 MAX_TCE_DELEGATION = 2048 MAX_TCE_KEY_ROTATION = 1024 MAX_TCE_REVOCATION = 1024 MAX_TCE_APPROVAL_REQUEST = 8192 MAX_TCE_APPROVAL_RESPONSE = 1024 MAX_TCE_AUTH_ASSERTION = 1024 # Timestamps. MIN_TIMESTAMP = 1_000_000_000 # 2001-09-09T01:46:40Z MAX_TIMESTAMP = 4_102_444_800 # 2100-01-01T00:00:00Z MAX_APPROVAL_LIFETIME = 60 # seconds AUDIENCE = "trust.n1ko.dev" class EncodingError(ValueError): """Raised when a value has no canonical encoding.""" # -------------------------------------------------------------------------- # Primitives # -------------------------------------------------------------------------- def uvarint(n: int) -> bytes: """LEB128 unsigned varint, canonical (shortest) form.""" if n < 0 or n > 0xFFFF_FFFF_FFFF_FFFF: raise EncodingError(f"uvarint out of range: {n}") out = bytearray() while True: b = n & 0x7F n >>= 7 if n: out.append(b | 0x80) else: out.append(b) return bytes(out) def read_uvarint(buf: bytes, off: int) -> tuple[int, int]: """Decode a canonical uvarint. Rejects non-minimal and overlong forms.""" n = 0 shift = 0 start = off while True: if off >= len(buf): raise EncodingError("uvarint: truncated") if off - start >= MAX_UVARINT_BYTES: raise EncodingError("uvarint: too long") b = buf[off] off += 1 n |= (b & 0x7F) << shift if b & 0x80 == 0: # Canonical form: a multi-byte encoding must not end in 0x00, # because that is a longer spelling of a shorter value. if off - start > 1 and b == 0x00: raise EncodingError("uvarint: non-minimal encoding") if n > 0xFFFF_FFFF_FFFF_FFFF: raise EncodingError("uvarint: overflow") return n, off shift += 7 def enc_bytes(b: bytes) -> bytes: """Length-prefixed byte string.""" return uvarint(len(b)) + b def validate_utf8(s: str, *, field: str, maxlen: int) -> bytes: """Encode a string, enforcing the protocol's string rules.""" raw = s.encode("utf-8", errors="strict") if len(raw) > maxlen: raise EncodingError(f"{field}: too long ({len(raw)} > {maxlen})") # Re-decode to reject surrogates and other invalid sequences that Python # would otherwise let through. if raw.decode("utf-8", errors="strict") != s: raise EncodingError(f"{field}: not valid UTF-8") for ch in s: cp = ord(ch) if cp <= 0x1F or cp == 0x7F or 0x80 <= cp <= 0x9F: raise EncodingError(f"{field}: control character U+{cp:04X}") if 0xD800 <= cp <= 0xDFFF: raise EncodingError(f"{field}: surrogate U+{cp:04X}") return raw def enc_string(s: str, *, field: str, maxlen: int) -> bytes: return enc_bytes(validate_utf8(s, field=field, maxlen=maxlen)) def enc_identity(pubkey: bytes) -> bytes: """Identity field: address version, then the length-prefixed raw key. The raw key is encoded rather than the bech32m address text, because the address is a presentation format. Signing the key directly means a future change to the textual encoding cannot invalidate existing signatures. """ if len(pubkey) != PUBKEY_LEN: raise EncodingError("identity: public key must be 32 bytes") return uvarint(ADDR_VERSION) + enc_bytes(pubkey) def enc_timestamp(ts: int, *, field: str, allow_zero: bool = False) -> bytes: if allow_zero and ts == 0: return uvarint(0) if not isinstance(ts, int) or isinstance(ts, bool): raise EncodingError(f"{field}: must be an integer") if ts < MIN_TIMESTAMP or ts > MAX_TIMESTAMP: raise EncodingError(f"{field}: out of range: {ts}") return uvarint(ts) # -------------------------------------------------------------------------- # Number canonicalization # -------------------------------------------------------------------------- _NUMBER_RE = re.compile(r"^-?(0|[1-9][0-9]*)(\.[0-9]+)?([eE][+-]?[0-9]+)?$") def canonical_number(token: str) -> str: """Reduce a JSON number token to its unique canonical decimal form. Numbers are carried as decimal text rather than as a binary float because JSON numbers are arbitrary precision: 1, 1.0 and 1e0 are the same value written three ways, and converting through a float would both lose precision and make the signed bytes depend on the implementation's rounding. Canonical text has exactly one spelling per value. Canonical form: an optional minus sign, digits with no leading zero, an optional fractional part with no trailing zero, and never an exponent. Negative zero is not representable; it canonicalizes to "0". """ if not isinstance(token, str): raise EncodingError("number: must be given as a decimal token") if len(token) > MAX_NUMBER_TOKEN: raise EncodingError(f"number: token too long ({len(token)})") m = _NUMBER_RE.match(token) if not m: raise EncodingError(f"number: not a valid JSON number: {token!r}") int_part, frac_part, exp_part = m.group(1), m.group(2), m.group(3) negative = token.startswith("-") frac_digits = frac_part[1:] if frac_part else "" exp = int(exp_part[1:]) if exp_part else 0 if exp_part and len(exp_part.lstrip("eE+-")) > 4: raise EncodingError("number: exponent has too many digits") # value = sign * mantissa * 10^scale mantissa = int(int_part + frac_digits) scale = exp - len(frac_digits) if mantissa == 0: return "0" # Remove factors of ten that only exist to pad the fraction. while scale < 0 and mantissa % 10 == 0: mantissa //= 10 scale += 1 digits = str(mantissa) if scale >= 0: int_digits = digits + "0" * scale frac_out = "" else: point = len(digits) + scale if point <= 0: int_digits = "0" frac_out = "0" * (-point) + digits else: int_digits = digits[:point] frac_out = digits[point:] if len(int_digits.lstrip("0") or "0") > MAX_NUMBER_INT_DIGITS: raise EncodingError("number: too many integer digits") if len(frac_out) > MAX_NUMBER_FRAC_DIGITS: raise EncodingError("number: too many fractional digits") out = ("-" if negative else "") + int_digits if frac_out: out += "." + frac_out if len(out) > MAX_NUMBER: raise EncodingError("number: canonical form too long") return out # -------------------------------------------------------------------------- # Values and maps # -------------------------------------------------------------------------- @dataclass(frozen=True) class Num: """A JSON number carried as its exact decimal token.""" token: str def enc_value(v) -> bytes: if v is None: return bytes([V_NULL]) if v is True: return bytes([V_TRUE]) if v is False: return bytes([V_FALSE]) if isinstance(v, Num): return bytes([V_NUMBER]) + enc_bytes(canonical_number(v.token).encode("ascii")) if isinstance(v, int): return bytes([V_NUMBER]) + enc_bytes(canonical_number(str(v)).encode("ascii")) if isinstance(v, str): return bytes([V_STRING]) + enc_string(v, field="string value", maxlen=MAX_STRING_VALUE) if isinstance(v, float): raise EncodingError( "number: binary floats are not accepted; pass the decimal token via Num()" ) raise EncodingError(f"value: unsupported type {type(v).__name__}") _KEY_RE = re.compile(r"^[a-z][a-z0-9]*([._-][a-z0-9]+)*$") def validate_key(k: str) -> bytes: """Validate a map key. The charset is restricted so that keys are unambiguous and compare bytewise. This is a lexical rule only: the protocol attaches no meaning to any key, and no implementation may branch on a key's content. """ raw = validate_utf8(k, field="key", maxlen=MAX_CLAIM_KEY) if not _KEY_RE.match(k): raise EncodingError(f"key: must match {_KEY_RE.pattern}: {k!r}") return raw def enc_map(m: dict, *, min_entries: int = 0) -> bytes: if len(m) < min_entries: raise EncodingError(f"map: needs at least {min_entries} entries") if len(m) > MAX_MAP_ENTRIES: raise EncodingError(f"map: too many entries ({len(m)})") encoded = [] seen = set() for k, v in m.items(): raw = validate_key(k) if raw in seen: raise EncodingError(f"map: duplicate key {k!r}") seen.add(raw) encoded.append((raw, v)) # Sort by the raw key bytes, unsigned bytewise ascending. encoded.sort(key=lambda kv: kv[0]) out = uvarint(len(encoded)) for raw, v in encoded: out += enc_bytes(raw) + enc_value(v) return out # -------------------------------------------------------------------------- # Objects # -------------------------------------------------------------------------- def _check_size(blob: bytes, limit: int, what: str) -> bytes: if len(blob) > limit: raise EncodingError(f"{what}: encoding too large ({len(blob)} > {limit})") return blob def tce_identity(*, pubkey: bytes, alias: str, created_at: int) -> bytes: """IdentityRegistration. The alias is included here, and only here. An identity signing its own display label makes that self-assertion tamper-evident. It remains non-authoritative: it is not unique, not verified, and never consulted when checking any other object's signature (INV-7). """ body = ( MAGIC + bytes([TAG_IDENTITY]) + uvarint(PROTOCOL_VERSION) + enc_identity(pubkey) + enc_string(alias, field="alias", maxlen=MAX_ALIAS) + enc_timestamp(created_at, field="created_at") ) return _check_size(body, MAX_TCE_IDENTITY, "identity") def tce_claim( *, issuer: bytes, subject: bytes, claims: dict, created_at: int, expires_at: int, serial: int, nonce: bytes, ) -> bytes: """Claim: issuer asserts key/value statements about subject.""" if len(nonce) != NONCE_LEN: raise EncodingError("nonce: must be 16 bytes") if expires_at != 0 and expires_at <= created_at: raise EncodingError("expires_at: must be 0 or after created_at") body = ( MAGIC + bytes([TAG_CLAIM]) + uvarint(PROTOCOL_VERSION) + enc_identity(issuer) + enc_identity(subject) + enc_map(claims, min_entries=1) + enc_timestamp(created_at, field="created_at") + enc_timestamp(expires_at, field="expires_at", allow_zero=True) + uvarint(serial) + enc_bytes(nonce) ) return _check_size(body, MAX_TCE_CLAIM, "claim") def tce_delegation( *, granter: bytes, grantee: bytes, predicates: dict, max_depth: int, created_at: int, expires_at: int, serial: int, nonce: bytes, ) -> bytes: """DelegationClaim: granter lets grantee speak for listed predicates.""" if len(nonce) != NONCE_LEN: raise EncodingError("nonce: must be 16 bytes") if expires_at != 0 and expires_at <= created_at: raise EncodingError("expires_at: must be 0 or after created_at") if not 1 <= len(predicates) <= 32: raise EncodingError("predicates: must hold between 1 and 32 entries") for k, v in predicates.items(): if v is not True: raise EncodingError("predicates: every value must be true") body = ( MAGIC + bytes([TAG_DELEGATION]) + uvarint(PROTOCOL_VERSION) + enc_identity(granter) + enc_identity(grantee) + enc_map(predicates, min_entries=1) + uvarint(max_depth) + enc_timestamp(created_at, field="created_at") + enc_timestamp(expires_at, field="expires_at", allow_zero=True) + uvarint(serial) + enc_bytes(nonce) ) return _check_size(body, MAX_TCE_DELEGATION, "delegation") def tce_key_rotation_request( *, successor: bytes, predecessor: bytes, created_at: int, expires_at: int, ) -> bytes: """KeyRotationRequest: the incoming key claims succession.""" if expires_at <= created_at: raise EncodingError("expires_at: must be after created_at") if expires_at - created_at > 60: raise EncodingError("expires_at: lifetime exceeds 60 seconds") body = ( MAGIC + bytes([TAG_KEY_ROTATION]) + uvarint(PROTOCOL_VERSION) + enc_identity(successor) + enc_identity(predecessor) + enc_timestamp(created_at, field="created_at") + enc_timestamp(expires_at, field="expires_at") ) return _check_size(body, MAX_TCE_KEY_ROTATION, "key_rotation_request") def tce_key_rotation_confirm( *, rotation_hash: bytes, created_at: int, nonce: bytes, ) -> bytes: """KeyRotationConfirm: the predecessor consents to one exact request.""" if len(nonce) != NONCE_LEN: raise EncodingError("nonce: must be 16 bytes") body = ( MAGIC + bytes([TAG_KEY_ROTATION_CONF]) + uvarint(PROTOCOL_VERSION) + enc_bytes(rotation_hash) + enc_timestamp(created_at, field="created_at") + enc_bytes(nonce) ) return _check_size(body, MAX_TCE_KEY_ROTATION, "key_rotation_confirm") def tce_revocation( *, issuer: bytes, claim_id: bytes, reason: str, created_at: int, nonce: bytes ) -> bytes: """Revocation: the issuer of a claim withdraws it.""" if len(claim_id) != HASH_LEN: raise EncodingError("claim_id: must be 32 bytes") if len(nonce) != NONCE_LEN: raise EncodingError("nonce: must be 16 bytes") body = ( MAGIC + bytes([TAG_REVOCATION]) + uvarint(PROTOCOL_VERSION) + enc_identity(issuer) + enc_bytes(claim_id) + enc_string(reason, field="reason", maxlen=MAX_REASON) + enc_timestamp(created_at, field="created_at") + enc_bytes(nonce) ) return _check_size(body, MAX_TCE_REVOCATION, "revocation") def tce_approval_request( *, sender: bytes, recipient: bytes, action: str, payload: dict, message: str, created_at: int, expires_at: int, nonce: bytes, ) -> bytes: """ApprovalRequest: sender asks recipient to approve an opaque action.""" if len(nonce) != NONCE_LEN: raise EncodingError("nonce: must be 16 bytes") if expires_at <= created_at: raise EncodingError("expires_at: must be after created_at") if expires_at - created_at > MAX_APPROVAL_LIFETIME: raise EncodingError( f"expires_at: lifetime exceeds {MAX_APPROVAL_LIFETIME}s" ) body = ( MAGIC + bytes([TAG_APPROVAL_REQUEST]) + uvarint(PROTOCOL_VERSION) + enc_identity(sender) + enc_identity(recipient) + enc_string(action, field="action", maxlen=MAX_ACTION) + enc_map(payload) + enc_string(message, field="message", maxlen=MAX_MESSAGE) + enc_timestamp(created_at, field="created_at") + enc_timestamp(expires_at, field="expires_at") + enc_bytes(nonce) ) return _check_size(body, MAX_TCE_APPROVAL_REQUEST, "approval request") def tce_approval_response( *, request_hash: bytes, responder: bytes, decision: int, created_at: int, nonce: bytes ) -> bytes: """ApprovalResponse: recipient's signed decision on one exact request. request_hash commits to the entire canonical request, not to a sender-chosen label, which is what makes a signed decision impossible to move to a different request (INV-4). """ if len(request_hash) != HASH_LEN: raise EncodingError("request_hash: must be 32 bytes") if len(nonce) != NONCE_LEN: raise EncodingError("nonce: must be 16 bytes") if decision not in (DECISION_DENY, DECISION_ALLOW): raise EncodingError(f"decision: unknown value {decision}") body = ( MAGIC + bytes([TAG_APPROVAL_RESPONSE]) + uvarint(PROTOCOL_VERSION) + enc_bytes(request_hash) + enc_identity(responder) + uvarint(decision) + enc_timestamp(created_at, field="created_at") + enc_bytes(nonce) ) return _check_size(body, MAX_TCE_APPROVAL_RESPONSE, "approval response") def tce_auth_assertion( *, identity: bytes, challenge: bytes, scope: str, audience: str, created_at: int ) -> bytes: """AuthAssertion: proof of key possession for a server-issued challenge. The audience is signed so that a challenge answered for one server cannot be replayed to another. Claims and approvals carry no audience because they are global statements meant to be portable between relays. """ if len(challenge) != HASH_LEN: raise EncodingError("challenge: must be 32 bytes") body = ( MAGIC + bytes([TAG_AUTH_ASSERTION]) + uvarint(PROTOCOL_VERSION) + enc_identity(identity) + enc_bytes(challenge) + enc_string(scope, field="scope", maxlen=MAX_SCOPE) + enc_string(audience, field="audience", maxlen=MAX_AUDIENCE) + enc_timestamp(created_at, field="created_at") ) return _check_size(body, MAX_TCE_AUTH_ASSERTION, "auth assertion") def object_id(tce: bytes) -> bytes: """Content address of an object: SHA-256 over its canonical bytes.""" return hashlib.sha256(tce).digest() # -------------------------------------------------------------------------- # bech32m (BIP-350) for rendering addresses # -------------------------------------------------------------------------- _CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l" _BECH32M_CONST = 0x2BC830A3 def _polymod(values): gen = [0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3] chk = 1 for v in values: b = chk >> 25 chk = (chk & 0x1FFFFFF) << 5 ^ v for i in range(5): chk ^= gen[i] if ((b >> i) & 1) else 0 return chk def _hrp_expand(hrp): return [ord(x) >> 5 for x in hrp] + [0] + [ord(x) & 31 for x in hrp] def _convertbits(data, frombits, tobits, pad=True): acc = bits = 0 ret = [] maxv = (1 << tobits) - 1 for value in data: acc = (acc << frombits) | value bits += frombits while bits >= tobits: bits -= tobits ret.append((acc >> bits) & maxv) if pad and bits: ret.append((acc << (tobits - bits)) & maxv) return ret def bech32m_address(pubkey: bytes) -> str: payload = bytes([ADDR_VERSION]) + pubkey data = _convertbits(payload, 8, 5) pm = _polymod(_hrp_expand("trust") + data + [0] * 6) ^ _BECH32M_CONST checksum = [(pm >> 5 * (5 - i)) & 31 for i in range(6)] return "trust1" + "".join(_CHARSET[d] for d in data + checksum) # -------------------------------------------------------------------------- # Vector generation # -------------------------------------------------------------------------- @dataclass class Party: name: str seed: bytes @property def key(self) -> Ed25519PrivateKey: return Ed25519PrivateKey.from_private_bytes(self.seed) @property def pub(self) -> bytes: return self.key.public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw ) @property def address(self) -> str: return bech32m_address(self.pub) def sign(self, msg: bytes) -> bytes: return self.key.sign(msg) NIKOCRAFT = Party("NikoCraft", bytes([0x01] * 32)) NIKO = Party("Niko", bytes([0x02] * 32)) N1 = bytes.fromhex("000102030405060708090a0b0c0d0e0f") N2 = bytes.fromhex("101112131415161718191a1b1c1d1e1f") N3 = bytes.fromhex("202122232425262728292a2b2c2d2e2f") N4 = bytes.fromhex("303132333435363738393a3b3c3d3e3f") T0 = 1_700_000_000 def build_vectors() -> dict: vectors = [] def add(name, description, tce, signer: Party | None, extra=None): oid = object_id(tce) entry = { "name": name, "description": description, "tce_hex": tce.hex(), "tce_len": len(tce), "object_id_hex": oid.hex(), } if signer is not None: entry["signer_pubkey_hex"] = signer.pub.hex() entry["signer_address"] = signer.address entry["signature_hex"] = signer.sign(tce).hex() if extra: entry.update(extra) vectors.append(entry) return entry # 1. Identity registration. ident = tce_identity(pubkey=NIKOCRAFT.pub, alias="NikoCraft", created_at=T0) add( "identity/nikocraft", "Self-asserted identity registration. The alias is signed here and " "nowhere else.", ident, NIKOCRAFT, { "json": { "type": "identity", "version": 1, "identity": NIKOCRAFT.address, "alias": "NikoCraft", "created_at": T0, } }, ) ident_niko = tce_identity(pubkey=NIKO.pub, alias="Niko", created_at=T0) add( "identity/niko", "Second identity used as the subject and approver in later vectors.", ident_niko, NIKO, { "json": { "type": "identity", "version": 1, "identity": NIKO.address, "alias": "Niko", "created_at": T0, } }, ) # 2. Boolean claim. claim_bool = tce_claim( issuer=NIKOCRAFT.pub, subject=NIKO.pub, claims={"example.flag": True}, created_at=T0, expires_at=T0 + 86400, serial=1, nonce=N1, ) claim_bool_entry = add( "claim/boolean", "Single boolean statement. The server attaches no meaning to the key.", claim_bool, NIKOCRAFT, { "json": { "type": "claim", "version": 1, "issuer": NIKOCRAFT.address, "subject": NIKO.address, "claims": {"example.flag": True}, "created_at": T0, "expires_at": T0 + 86400, "serial": 1, "nonce": N1.hex(), } }, ) # 3. Multi-key claim exercising every value type and map ordering. claim_multi = tce_claim( issuer=NIKOCRAFT.pub, subject=NIKO.pub, claims={ "z.last": None, "example.rank": "[VIP]", "a.first": False, "example.level": Num("42"), "example.ratio": Num("1.5"), }, created_at=T0, expires_at=0, serial=2, nonce=N2, ) add( "claim/all-value-types", "Every v1 value type in one claim, supplied out of order to exercise " "key sorting. expires_at 0 means the claim does not expire.", claim_multi, NIKOCRAFT, { "note": "input key order was z.last, example.rank, a.first, " "example.level, example.ratio; canonical order is a.first, " "example.level, example.ratio, example.rank, z.last", "json": { "type": "claim", "version": 1, "issuer": NIKOCRAFT.address, "subject": NIKO.address, "claims": { "a.first": False, "example.level": 42, "example.ratio": 1.5, "example.rank": "[VIP]", "z.last": None, }, "created_at": T0, "expires_at": 0, "serial": 2, "nonce": N2.hex(), }, }, ) # 4. Revocation of the boolean claim. rev = tce_revocation( issuer=NIKOCRAFT.pub, claim_id=bytes.fromhex(claim_bool_entry["object_id_hex"]), reason="superseded", created_at=T0 + 100, nonce=N3, ) add( "revocation/boolean-claim", "Signed withdrawal of claim/boolean by its issuer.", rev, NIKOCRAFT, { "json": { "type": "revocation", "version": 1, "issuer": NIKOCRAFT.address, "claim_id": claim_bool_entry["object_id_hex"], "reason": "superseded", "created_at": T0 + 100, "nonce": N3.hex(), } }, ) # 5. Approval request. req = tce_approval_request( sender=NIKOCRAFT.pub, recipient=NIKO.pub, action="example.ban", payload={"target": "Steve"}, message="Ban Steve", created_at=T0, expires_at=T0 + 30, nonce=N4, ) req_entry = add( "approval_request/ban", "Opaque action with an opaque payload. The relay never interprets " "either.", req, NIKOCRAFT, { "json": { "type": "approval_request", "version": 1, "sender": NIKOCRAFT.address, "recipient": NIKO.address, "action": "example.ban", "payload": {"target": "Steve"}, "message": "Ban Steve", "created_at": T0, "expires_at": T0 + 30, "nonce": N4.hex(), }, "request_id_hex": object_id(req).hex(), }, ) # 6. Approval response binding that exact request. resp = tce_approval_response( request_hash=bytes.fromhex(req_entry["object_id_hex"]), responder=NIKO.pub, decision=DECISION_ALLOW, created_at=T0 + 10, nonce=N1, ) add( "approval_response/allow", "Allow decision bound to the exact canonical request bytes.", resp, NIKO, { "json": { "type": "approval_response", "version": 1, "request_hash": req_entry["object_id_hex"], "responder": NIKO.address, "decision": "allow", "created_at": T0 + 10, "nonce": N1.hex(), } }, ) resp_deny = tce_approval_response( request_hash=bytes.fromhex(req_entry["object_id_hex"]), responder=NIKO.pub, decision=DECISION_DENY, created_at=T0 + 10, nonce=N1, ) add( "approval_response/deny", "Deny decision. Differs from allow in exactly one byte, so a verifier " "that ignores the decision field is detectable.", resp_deny, NIKO, { "json": { "type": "approval_response", "version": 1, "request_hash": req_entry["object_id_hex"], "responder": NIKO.address, "decision": "deny", "created_at": T0 + 10, "nonce": N1.hex(), } }, ) # 7. Auth assertion. challenge = bytes.fromhex( "a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8f90" ) auth = tce_auth_assertion( identity=NIKO.pub, challenge=challenge, scope="ws", audience=AUDIENCE, created_at=T0, ) add( "auth_assertion/ws", "Proof of key possession, bound to one server-issued challenge and to " "the audience, so it cannot be replayed to another server.", auth, NIKO, { "json": { "type": "auth_assertion", "version": 1, "identity": NIKO.address, "challenge": challenge.hex(), "scope": "ws", "audience": AUDIENCE, "created_at": T0, } }, ) # 9. Delegation: NikoCraft grants Niko the right to speak. deleg = tce_delegation( granter=NIKOCRAFT.pub, grantee=NIKO.pub, predicates={"example.flag": True}, max_depth=0, created_at=T0, expires_at=T0 + 86_400, serial=1, nonce=N3, ) add( "delegation/minimal", "A grant of voice: granter lets grantee issue claims covering the " "listed predicate. max_depth 0 forbids re-delegation.", deleg, NIKOCRAFT, { "json": { "type": "delegation", "version": 1, "granter": NIKOCRAFT.address, "grantee": NIKO.address, "predicates": {"example.flag": True}, "max_depth": 0, "created_at": T0, "expires_at": T0 + 86_400, "serial": 1, "nonce": N3.hex(), } }, ) # 10. Delegation with several predicates and re-delegation depth. deleg2 = tce_delegation( granter=NIKOCRAFT.pub, grantee=NIKO.pub, predicates={"a.first": True, "b.second": True, "c.third": True}, max_depth=2, created_at=T0, expires_at=0, serial=2, nonce=N4, ) add( "delegation/multi-predicate", "Three exact predicates (map ordering exercised), two permitted " "re-delegation hops below the grantee, no expiry.", deleg2, NIKOCRAFT, { "json": { "type": "delegation", "version": 1, "granter": NIKOCRAFT.address, "grantee": NIKO.address, "predicates": {"a.first": True, "b.second": True, "c.third": True}, "max_depth": 2, "created_at": T0, "expires_at": 0, "serial": 2, "nonce": N4.hex(), } }, ) # 11. Key rotation pair: Niko succeeds... the parties rotate between # themselves for vector purposes. The confirm binds to the request by ID. rot_req = tce_key_rotation_request( successor=NIKO.pub, predecessor=NIKOCRAFT.pub, created_at=T0, expires_at=T0 + 30, ) rot_req_id = object_id(rot_req) add( "key_rotation/request", "The successor claims succession from the predecessor; meaningless " "without the confirm.", rot_req, NIKO, { "json": { "type": "key_rotation_request", "version": 1, "successor": NIKO.address, "predecessor": NIKOCRAFT.address, "created_at": T0, "expires_at": T0 + 30, "object_id_hex": rot_req_id.hex(), } }, ) rot_conf = tce_key_rotation_confirm( rotation_hash=rot_req_id, created_at=T0 + 10, nonce=N1, ) add( "key_rotation/confirm", "The predecessor's consent, bound to the request by content ID.", rot_conf, NIKOCRAFT, { "json": { "type": "key_rotation_confirm", "version": 1, "rotation_hash": rot_req_id.hex(), "created_at": T0 + 10, "nonce": N1.hex(), "request_object_id_hex": rot_req_id.hex(), } }, ) return { "format": "trust.n1ko.dev TCE test vectors", "tce_version": 1, "magic_hex": MAGIC.hex(), "magic_ascii": "trust.n1ko.dev/tce/1\\x00", "parties": { p.name: { "seed_hex": p.seed.hex(), "pubkey_hex": p.pub.hex(), "address": p.address, } for p in (NIKOCRAFT, NIKO) }, "vectors": vectors, "number_canonicalization": build_number_vectors(), "rejects": build_reject_vectors(), } def build_number_vectors() -> dict: accept = {} for token in [ "0", "-0", "0.0", "0e10", "1", "1.0", "1e0", "1.000", "10", "1e1", "100", "1e2", "-1", "-1.0", "1.5", "1.50", "0.1", ".1e1" if False else "1e-1", "0.001", "1e-3", "12345678901234567890", "1.23e2", "123", "-0.5", "1e18", "0.000000000000000001", "-1e-18", "999999999999999999999999", ]: accept[token] = canonical_number(token) reject = {} for token in [ "", "+1", "01", "1.", ".5", "1e", "1e+", "--1", "1.2.3", "0x10", "Infinity", "NaN", "1_000", " 1", "1 ", "1e99999", "1" + "0" * 40, "0." + "0" * 17 + "1" + "1", ]: try: canonical_number(token) reject[token] = "ERROR: accepted" except EncodingError as e: reject[token] = str(e) return {"accept": accept, "reject": reject} def build_reject_vectors() -> list: """Inputs that a conforming decoder must refuse.""" out = [] def rej(name, hex_bytes, reason): out.append({"name": name, "tce_hex": hex_bytes, "reason": reason}) good = tce_claim( issuer=NIKOCRAFT.pub, subject=NIKO.pub, claims={"example.flag": True}, created_at=T0, expires_at=T0 + 86400, serial=1, nonce=N1, ) rej("empty", "", "no magic") rej("magic_truncated", MAGIC[:-1].hex(), "magic is incomplete") rej( "magic_wrong_version", (b"trust.n1ko.dev/tce/2\x00" + good[len(MAGIC):]).hex(), "framing version is not 1", ) rej( "unknown_object_tag", (MAGIC + bytes([0x7F]) + good[len(MAGIC) + 1:]).hex(), "object tag 0x7f is not defined", ) rej( "object_tag_zero", (MAGIC + bytes([0x00]) + good[len(MAGIC) + 1:]).hex(), "object tag 0x00 is permanently reserved", ) rej("truncated_body", good[:-1].hex(), "input ends inside the nonce") rej("trailing_byte", (good + b"\x00").hex(), "trailing data after the object") rej( "non_minimal_uvarint", (MAGIC + bytes([TAG_CLAIM]) + b"\x81\x00" + good[len(MAGIC) + 2:]).hex(), "version encoded as a non-minimal uvarint", ) out.append( { "name": "unsorted_map_keys", "reason": "map entries not in ascending bytewise key order", "note": "constructed by swapping the two entries of a two-key claim", } ) out.append( { "name": "duplicate_map_key", "reason": "the same key appears twice in one map", } ) out.append( { "name": "reserved_value_tag", "reason": "value tags 0x05, 0x06 and 0x07 are reserved and must be " "rejected in v1 rather than skipped", } ) return out def annotate(tce: bytes, fields: list[tuple[str, int]]) -> str: """Produce a byte-by-byte breakdown for the specification.""" lines = [] off = 0 for label, length in fields: chunk = tce[off : off + length] lines.append(f" {off:4d} {chunk.hex():<40} {label}") off += length if off != len(tce): lines.append(f" {off:4d} {tce[off:].hex():<40} UNACCOUNTED") return "\n".join(lines) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--emit", type=Path, help="directory to write vectors into") ap.add_argument("--annotate", action="store_true", help="print a byte breakdown") args = ap.parse_args() data = build_vectors() if args.annotate: claim = bytes.fromhex(data["vectors"][2]["tce_hex"]) print("claim/boolean byte breakdown:") print( annotate( claim, [ ("magic 'trust.n1ko.dev/tce/1\\x00'", 21), ("object tag 0x02 (Claim)", 1), ("version uvarint(1)", 1), ("issuer: addr version uvarint(0)", 1), ("issuer: key length uvarint(32)", 1), ("issuer: public key", 32), ("subject: addr version uvarint(0)", 1), ("subject: key length uvarint(32)", 1), ("subject: public key", 32), ("claims: entry count uvarint(1)", 1), ("claims[0]: key length uvarint(12)", 1), ("claims[0]: key 'example.flag'", 12), ("claims[0]: value tag 0x02 (true)", 1), ("created_at uvarint(1700000000)", 5), ("expires_at uvarint(1700086400)", 5), ("serial uvarint(1)", 1), ("nonce length uvarint(16)", 1), ("nonce", 16), ], ) ) print() if args.emit: args.emit.mkdir(parents=True, exist_ok=True) path = args.emit / "tce_vectors.json" path.write_text(json.dumps(data, indent=2, sort_keys=False) + "\n") print(f"wrote {path} ({len(data['vectors'])} object vectors)") else: json.dump(data, sys.stdout, indent=2) print() return 0 if __name__ == "__main__": raise SystemExit(main())