Niko Marmeladkov
9d66003689
- server: relay storing signed objects (PUT/GET), per-IP rate limiting, per-subject quota (1000), one-response-per-request, pagination, /v1/healthz /v1/readyz /v1/metrics - verify: signature-verifying trust evaluator; every object is checked via env.Verify(), approvals via VerifyApprovalResponse, revocations via VerifyRevocationOf; k-of-n approval quorum - docs: TRUST-MODEL.md and API.md describing issuer-anchored signatures and the endpoint/status-code contract - tests: server, verify, and ratelimit packages
1044 lines
32 KiB
Python
1044 lines
32 KiB
Python
#!/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
|
|
|
|
# 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_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_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,
|
|
}
|
|
},
|
|
)
|
|
|
|
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())
|