Niko/niko_trust
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
niko_trust/tools/reference/tce_reference.py
Niko Marmeladkov 20cc52c3a5 feat: network layer — PoW, checkpoint chain, gossip, light node, WS, delegation, rotation, BFT 
- BLAKE3 keyed proof-of-work on object storage and auth challenges,
  with frozen vectors cross-checked against an independent Python
  reference implementing the single-block hash it needs.
- Sparse Merkle trie over object IDs: order-independent roots,
  inclusion and absence proofs (internal/smt).
- Signed checkpoint chain per relay: transport key amendment to INV-1,
  /v1/checkpoint/* and inclusion/absence proof endpoints, restart-safe
  epoch continuity (internal/checkpoint).
- Head gossip with TOFU pinning and equivocation detection; light node
  (cmd/lightnode) that stores no history: quorum of pinned relays,
  every served object proven against the agreed root, LRU disk cache.
- WebSocket streaming on relay and light node (coder/websocket):
  scoped channels mirroring REST, raw envelopes verified client-side;
  light node marks streamed objects unproven until checkpoint coverage.
- Protocol v1 additions: DelegationClaim tag 0x07 with deterministic
  chain resolution in verify.Graph, KeyRotationRequest/Confirm tags
  0x08/0x09 with hash-bound two-sided consent and Policy.RotationMaxAge;
  spec sections, frozen vectors appended byte-identically, Python
  reference extended.
- Optional permissioned BFT finality over gossip (internal/bft):
  prevote/precommit with quorum certificates verifiable offline.
- Quick wins: Policy.TrustedIssuers, per-type stored metrics,
  batch fetch, lexicographic lists with stable cursor pagination.
- Security review of the network layer (docs/SECURITY-REVIEW.md) with
  findings F-01..F-09; hub send/close race and unstable pagination
  fixed under review.

12 packages green, vet/gofmt clean, protocol fuzzing stable.
2026-08-25 20:38:40 +03:00

1246 lines
38 KiB
Python
Raw Blame History

#!/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())
Reference in a new issue View git blame Copy permalink