9d5908104e
Known-answer tests for every part of the protocol that produces bytes: key derivation, sealed envelopes with their intermediate values, accept tokens, session and registration frames, rotation certificates, frontmatter parsing, and the inputs a conformant implementation must refuse. vectors.py derives every value by calling smolmail.py, so the vectors cannot drift from the code, and verifies each one from the opposite side before emitting it. --check regenerates and diffs against the committed files. Verified independently: a checker written against SPEC.md and libsodium alone, without importing the reference, reproduces all 81 values. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
711 lines
34 KiB
Python
711 lines
34 KiB
Python
#!/usr/bin/env -S uv run --quiet --script
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = ["noiseprotocol>=0.3.1", "pynacl>=1.5"]
|
|
# ///
|
|
# Copyright 2026 randogoth
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
"""Test vectors for SPEC.md 1.2, generated from the reference implementation.
|
|
|
|
Every value is produced by calling into smolmail.py rather than restated here,
|
|
so the vectors cannot drift from the code they document. Each one is also
|
|
verified from the opposite side before it is emitted: signatures are checked,
|
|
envelopes are unsealed, and the rejection cases are confirmed to actually
|
|
raise. `--check` repeats the generation and compares it against the committed
|
|
files, which is what makes them a regression test rather than a snapshot.
|
|
|
|
uv run vectors.py # write vectors.json and VECTORS.md
|
|
uv run vectors.py --check # verify the committed files still reproduce
|
|
|
|
Determinism: seal() draws its ephemeral from os.urandom and its timestamp from
|
|
the clock (§5.2), so both are pinned here. Nothing else in the protocol is
|
|
randomised.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import calendar
|
|
import contextlib
|
|
import difflib
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
|
|
import nacl.bindings as sodium
|
|
|
|
import smolmail as sm
|
|
|
|
# Obvious test data: two masters that no one could mistake for a real secret.
|
|
ALICE_MASTER = bytes(range(0x00, 0x20))
|
|
BOB_MASTER = bytes(range(0x20, 0x40))
|
|
SERVER_STATIC = bytes(range(0x60, 0x80))
|
|
HANDSHAKE_HASH = bytes(range(0x80, 0xA0))
|
|
|
|
WHEN = calendar.timegm((2026, 1, 1, 0, 0, 0))
|
|
FUTURE = calendar.timegm((2100, 1, 1, 0, 0, 0)) # beyond §5.3's 86400s skew
|
|
|
|
HOST = "example.org"
|
|
|
|
# §4's low-order blacklist. Every one of these must be refused as an ephemeral
|
|
# public key; the generator asserts that the reference actually refuses them.
|
|
LOW_ORDER = [
|
|
"0000000000000000000000000000000000000000000000000000000000000000",
|
|
"0100000000000000000000000000000000000000000000000000000000000000",
|
|
"e0eb7a7c3b41b8ae1656e3faf19fc46ada098deb9c32b1fd866205165f49b800",
|
|
"5f9c95bca3508c24b1d0b1559c83ef5b04445cc4581c8e86d8224eddd09f1157",
|
|
"ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f",
|
|
"edffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f",
|
|
"eeffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f",
|
|
]
|
|
|
|
|
|
def h(raw: bytes) -> str:
|
|
return raw.hex()
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def ephemeral(esk: bytes):
|
|
"""Pin seal()'s per-message ephemeral so an envelope reproduces byte for byte."""
|
|
real = os.urandom
|
|
sm.os.urandom = lambda n: esk if n == sm.KEY_LEN else real(n)
|
|
try:
|
|
yield sodium.crypto_scalarmult_base(esk)
|
|
finally:
|
|
sm.os.urandom = real
|
|
|
|
|
|
def plaintext_of(sender: sm.Identity, to: bytes, body: bytes, when: int,
|
|
epk: bytes, pad: bool) -> tuple[bytes, bytes, bytes]:
|
|
"""§5.3's payload, returned with the header and the signed preimage.
|
|
|
|
seal() builds this internally and keeps none of it, but an implementer
|
|
debugging a signature needs the exact bytes that went under the key. The
|
|
emitted envelope is asserted equal to seal()'s, which keeps this honest.
|
|
"""
|
|
header = bytes([sm.VERSION]) + sender.pk + struct.pack(">qI", when, len(body))
|
|
signed = sm.LABEL_MSG + to + epk + header + body
|
|
payload = header + body + sender.sign(signed)
|
|
if pad:
|
|
payload += bytes(-len(payload) % sm.PAD_TO)
|
|
return header, signed, payload
|
|
|
|
|
|
def encrypt(payload: bytes, to: bytes, esk: bytes) -> bytes:
|
|
"""§5.2's sealing step alone, so a vector can carry a deliberately broken
|
|
payload that still passes the AEAD."""
|
|
epk = sodium.crypto_scalarmult_base(esk)
|
|
key = sm.hkdf_sha256(sm.agree(esk, sm.ed_to_x25519_pub(to)), epk + to, sm.LABEL_SEAL)
|
|
aad = sm.MAGIC + bytes([sm.VERSION]) + to + epk
|
|
return aad + sodium.crypto_aead_chacha20poly1305_ietf_encrypt(payload, aad, bytes(12), key)
|
|
|
|
|
|
def raises(fn) -> str:
|
|
"""Run a case that must fail and return the reference's own message."""
|
|
try:
|
|
fn()
|
|
except (sm.SmolError, Exception) as exc: # noqa: BLE001 - any refusal counts
|
|
return f"{type(exc).__name__}: {exc}"
|
|
raise AssertionError("expected a rejection, got none")
|
|
|
|
|
|
# --- §2 identity ------------------------------------------------------------
|
|
|
|
def identity_vectors() -> list[dict]:
|
|
out = []
|
|
for name, master, index in [("alice", ALICE_MASTER, 0), ("alice", ALICE_MASTER, 1),
|
|
("bob", BOB_MASTER, 0)]:
|
|
seed = sm.identity_seed(master, index)
|
|
who = sm.Identity(seed)
|
|
out.append({
|
|
"name": name,
|
|
"master": h(master),
|
|
"rotation_index": index,
|
|
"seed": h(seed),
|
|
"identity_key": h(who.pk),
|
|
"x25519_private": h(who.x_priv),
|
|
"x25519_public": h(sm.ed_to_x25519_pub(who.pk)),
|
|
"base32": sm.b32(who.pk),
|
|
"fingerprint": sm.fingerprint(who.pk),
|
|
"uri": f"smol://{name}@{HOST}/{sm.b32(who.pk)}",
|
|
})
|
|
# The X25519 keypair must agree with itself, or the §2 map is wrong.
|
|
assert sodium.crypto_scalarmult_base(who.x_priv) == sm.ed_to_x25519_pub(who.pk)
|
|
return out
|
|
|
|
|
|
def accept_vectors(alice: sm.Account, bob: sm.Account) -> dict:
|
|
token = alice.token_for(bob.me.pk)
|
|
assert token == hmac.new(sm.accept_key(ALICE_MASTER), bob.me.pk, hashlib.sha256).digest()
|
|
return {
|
|
"note": "alice's accept key, and the token she issues to bob. The accept "
|
|
"key is derived without the rotation index, so it and every token "
|
|
"from it survive rotation.",
|
|
"alice_accept_key": h(sm.accept_key(ALICE_MASTER)),
|
|
"bob_accept_key": h(sm.accept_key(BOB_MASTER)),
|
|
"alice_token_for_bob": h(token),
|
|
"alice_token_for_bob_base32": sm.b32(token),
|
|
"bob_token_for_alice_0": h(bob.token_for(alice.keys[0].pk)),
|
|
"alice_token_for_bob_rotated": h(alice.token_for(
|
|
sm.Identity(sm.identity_seed(BOB_MASTER, 1)).pk)),
|
|
}
|
|
|
|
|
|
# --- §5 messages ------------------------------------------------------------
|
|
|
|
def envelope_vectors(alice: sm.Account, bob: sm.Account) -> list[dict]:
|
|
reply_to = f"smol://alice@{HOST}/{sm.b32(alice.keys[0].pk)}"
|
|
accept = sm.b32(alice.token_for(bob.me.pk))
|
|
cases = [
|
|
{"name": "empty-body", "from": ("bob", 0), "to": ("alice", 0), "pad": False,
|
|
"body": b"",
|
|
"note": "body_len 0. The payload is 109 bytes and the envelope 194."},
|
|
{"name": "unpadded", "from": ("bob", 0), "to": ("alice", 0), "pad": False,
|
|
"body": b"hello alice",
|
|
"note": "The minimal form: 69 + 45 + 11 + 64 + 16 bytes. Start here."},
|
|
{"name": "padded", "from": ("bob", 0), "to": ("alice", 0), "pad": True,
|
|
"body": b"hello alice",
|
|
"note": "The same message with §5.3 padding to a 1024-byte multiple. A "
|
|
"receiver must ignore the trailing zeros and trust body_len."},
|
|
{"name": "utf8", "from": ("bob", 0), "to": ("alice", 0), "pad": False,
|
|
"body": "Grüße aus dem Mesh — 🛰\n".encode(),
|
|
"note": "body_len counts bytes, not characters: 29 bytes, 22 characters."},
|
|
{"name": "frontmatter", "from": ("alice", 0), "to": ("bob", 0), "pad": False,
|
|
"body": sm.build_frontmatter(
|
|
[("Subject", "the thing"), ("Reply-To", reply_to), ("Accept", accept)],
|
|
"Here is a token so your mail skips my requests tier.\n").encode(),
|
|
"note": "§5.5 frontmatter carrying the reserved Reply-To (§5.7) and Accept "
|
|
"(§5.8) fields. Both are only meaningful because the signature "
|
|
"covers them."},
|
|
{"name": "superseded-key", "from": ("bob", 0), "to": ("alice", 0), "pad": False,
|
|
"body": b"sent before you rotated\n",
|
|
"note": "Sealed to alice's index-0 key while she is at index 1 (§7). She "
|
|
"opens it with the retained key, which is re-derivable from the "
|
|
"master alone."},
|
|
{"name": "rotated-key", "from": ("bob", 0), "to": ("alice", 1), "pad": False,
|
|
"body": b"sent after you rotated\n",
|
|
"note": "The same correspondents after alice's rotation. Only `to` changes; "
|
|
"her accept token does not."},
|
|
]
|
|
accounts = {"alice": alice, "bob": bob}
|
|
out = []
|
|
for n, case in enumerate(cases):
|
|
esk = bytes([0x40 + n]) * 32
|
|
sender = accounts[case["from"][0]].keys[case["from"][1]]
|
|
recipient = accounts[case["to"][0]].keys[case["to"][1]].pk
|
|
body = case["body"]
|
|
with ephemeral(esk) as epk:
|
|
env = sm.seal(sender, recipient, body, when=WHEN, pad=case["pad"])
|
|
header, signed, payload = plaintext_of(sender, recipient, body, WHEN, epk, case["pad"])
|
|
# plaintext_of/encrypt duplicate seal()'s layout; pin them to it.
|
|
assert encrypt(payload, recipient, esk) == env, case["name"]
|
|
|
|
opened = sm.unseal(accounts[case["to"][0]].keys, env)
|
|
assert opened["body"] == body and opened["sender"] == sender.pk
|
|
assert opened["time"] == WHEN and opened["id"] == sm.message_id(env)
|
|
|
|
shared = sm.agree(esk, sm.ed_to_x25519_pub(recipient))
|
|
token = accounts[case["to"][0]].token_for(sender.pk)
|
|
out.append({
|
|
"name": case["name"],
|
|
"note": case["note"],
|
|
"sender": f"{case['from'][0]}[{case['from'][1]}]",
|
|
"recipient": f"{case['to'][0]}[{case['to'][1]}]",
|
|
"body": h(body),
|
|
"body_len": len(body),
|
|
"time": WHEN,
|
|
"padded": case["pad"],
|
|
"esk": h(esk),
|
|
"epk": h(epk),
|
|
"x25519_shared": h(shared),
|
|
"seal_key": h(sm.hkdf_sha256(shared, epk + recipient, sm.LABEL_SEAL)),
|
|
"aad": h(sm.MAGIC + bytes([sm.VERSION]) + recipient + epk),
|
|
"payload_header": h(header),
|
|
"signed_preimage": h(signed),
|
|
"signature": h(payload[sm.PAYLOAD_HEADER + len(body):][:sm.SIG_LEN]),
|
|
"payload": h(payload),
|
|
"payload_len": len(payload),
|
|
"envelope": h(env),
|
|
"envelope_len": len(env),
|
|
"envelope_sha256": h(hashlib.sha256(env).digest()),
|
|
"message_id": h(sm.message_id(env)),
|
|
"accept_mac": h(sm.accept_mac(token, sm.message_id(env))),
|
|
"send_body": h(bytes([sm.TOKEN_LEN]) + sm.accept_mac(token, sm.message_id(env)) + env),
|
|
})
|
|
return out
|
|
|
|
|
|
# --- §4, §6.1 and §7 wire bytes ---------------------------------------------
|
|
|
|
def session_vectors(alice: sm.Account) -> dict:
|
|
me = alice.keys[0]
|
|
name = b"alice"
|
|
auth_sig = me.sign(sm.LABEL_AUTH + HANDSHAKE_HASH)
|
|
assert sm.verify_sig(me.pk, auth_sig, sm.LABEL_AUTH + HANDSHAKE_HASH)
|
|
token = alice.token_for(sm.Identity(sm.identity_seed(BOB_MASTER, 0)).pk)
|
|
auth_frame = (bytes([len(name)]) + name + me.pk + auth_sig
|
|
+ bytes([1]) + struct.pack(">H", 1) + token)
|
|
|
|
reg_preimage = sm.register_signed(SERVER_STATIC, "alice", me.pk)
|
|
reg_sig = me.sign(reg_preimage)
|
|
assert sm.verify_sig(me.pk, reg_sig, reg_preimage)
|
|
reg_frame = (bytes([len(name)]) + name + me.pk + reg_sig + b"\0" + b"\0")
|
|
return {
|
|
"note": "The AUTH signature covers the Noise handshake hash, so it cannot "
|
|
"be replayed to another session or server. The REGISTER signature "
|
|
"covers the server's static key for the same reason: without it, "
|
|
"any party could bind any public key to any username.",
|
|
"handshake_hash": h(HANDSHAKE_HASH),
|
|
"server_static": h(SERVER_STATIC),
|
|
"auth_signed_preimage": h(sm.LABEL_AUTH + HANDSHAKE_HASH),
|
|
"auth_signature": h(auth_sig),
|
|
"auth_frame": h(auth_frame),
|
|
"auth_frame_note": "username_len=5 || 'alice' || identity || signature || "
|
|
"sync=1 || count=1 || alice's token for bob",
|
|
"register_signed_preimage": h(reg_preimage),
|
|
"register_signature": h(reg_sig),
|
|
"register_frame": h(reg_frame),
|
|
"register_frame_note": "token_len=0 and cert_len=0: a first registration, "
|
|
"no invite token and no rotation certificate",
|
|
}
|
|
|
|
|
|
def rotation_vectors(alice: sm.Account) -> dict:
|
|
old, new = alice.keys[0], alice.keys[1]
|
|
when = struct.pack(">q", WHEN)
|
|
signed = sm.LABEL_ROTATE + b"alice" + old.pk + new.pk + when
|
|
cert = old.pk + new.pk + when + old.sign(signed) + new.sign(signed)
|
|
assert len(cert) == sm.CERT_LEN
|
|
|
|
# A chain that leads from the pinned key to the offered one is accepted; the
|
|
# same certificate checked under another username is not (§7).
|
|
assert sm.walk_chain("alice", old.pk, new.pk, [cert])
|
|
assert not sm.walk_chain("mallory", old.pk, new.pk, [cert])
|
|
assert not sm.walk_chain("alice", old.pk, new.pk, [])
|
|
forged = old.pk + new.pk + when + old.sign(signed) + bytes(sm.SIG_LEN)
|
|
assert not sm.walk_chain("alice", old.pk, new.pk, [forged])
|
|
return {
|
|
"note": "A certificate is 200 bytes and carries no username: the name is "
|
|
"covered by both signatures but not transported, so a certificate "
|
|
"cannot be replayed against another username bound to the same key.",
|
|
"username": "alice",
|
|
"old_key": h(old.pk),
|
|
"new_key": h(new.pk),
|
|
"time": WHEN,
|
|
"signed_preimage": h(signed),
|
|
"sig_old": h(old.sign(signed)),
|
|
"sig_new": h(new.sign(signed)),
|
|
"certificate": h(cert),
|
|
"resolve_body": h(new.pk + bytes([1]) + cert),
|
|
"walk_chain": [
|
|
{"case": "pinned=old, offered=new, chain=[cert], username=alice",
|
|
"accepted": True},
|
|
{"case": "same certificate checked as username=mallory",
|
|
"accepted": False,
|
|
"why": "both signatures cover the username"},
|
|
{"case": "pinned=old, offered=new, empty chain", "accepted": False,
|
|
"why": "a key change needs a chain; absent means ask the user"},
|
|
{"case": "sig_new zeroed", "accepted": False,
|
|
"why": "the new key must sign too, or the old key alone could hand "
|
|
"the username to a key nobody controls"},
|
|
],
|
|
}
|
|
|
|
|
|
# --- §3 addressing ----------------------------------------------------------
|
|
|
|
def addressing_vectors(alice: sm.Account) -> dict:
|
|
def username_ok(name: str) -> bool:
|
|
"""§3 through the real parser: a username is valid when `name@host` parses
|
|
and gives that exact username back."""
|
|
try:
|
|
return sm.Address.parse(f"{name}@{HOST}").user == name
|
|
except sm.SmolError:
|
|
return False
|
|
|
|
names = ["a", "a0", "alice", "al-ice", "a.b-c_d", "x" * 63,
|
|
"", "-alice", "alice-", ".alice", "al..ice", "a._b", "a-_b",
|
|
"Alice", "x" * 64, "al ice", "alice@", "ali/ce"]
|
|
usernames = [{"username": n, "valid": username_ok(n)} for n in names]
|
|
assert [u["valid"] for u in usernames] == [True] * 6 + [False] * 12
|
|
|
|
addresses = []
|
|
for text in [f"alice@{HOST}", f"alice@{HOST}:1961", f"alice@{HOST}:2000",
|
|
f"smol://alice@{HOST}/{sm.b32(alice.keys[0].pk)}"]:
|
|
a = sm.Address.parse(text)
|
|
addresses.append({"input": text, "user": a.user, "host": a.host,
|
|
"port": a.port, "short": a.short,
|
|
"identity": h(a.identity) if a.identity else None})
|
|
bad = {text: raises(lambda t=text: sm.Address.parse(t))
|
|
for text in [f"smol://alice@{HOST}", f"smol://alice@{HOST}/{sm.b32(bytes(31))}",
|
|
f"smol://alice@{HOST}/not-base32!", f"-alice@{HOST}"]}
|
|
return {
|
|
"note": "The short form is resolved by the server; the smol:// form carries "
|
|
"the key and needs no server to be trusted. 1961 is the default "
|
|
"port and is omitted from the short form.",
|
|
"usernames": usernames,
|
|
"addresses": addresses,
|
|
"rejected": bad,
|
|
"base32": {"identity": h(alice.keys[0].pk), "encoded": sm.b32(alice.keys[0].pk),
|
|
"fingerprint": sm.fingerprint(alice.keys[0].pk),
|
|
"fingerprint_note": "the first 20 base32 characters in groups of "
|
|
"four: 100 bits, so a collision costs ~2^100"},
|
|
}
|
|
|
|
|
|
# --- §5.5 frontmatter -------------------------------------------------------
|
|
|
|
def frontmatter_vectors() -> list[dict]:
|
|
cases = [
|
|
("no-block", "hello\n", "no leading '---\\n', so the whole body is text"),
|
|
("simple", "---\nSubject: hi\n---\nbody\n", "one field, then the text"),
|
|
("empty-block", "---\n---\n---\nliteral dashes\n",
|
|
"an empty block escapes a body that genuinely starts with '---'"),
|
|
("case-and-duplicates", "---\nSubject: first\nsubject: second\nX-Mood: calm\n---\nb",
|
|
"keys are lowercased for comparison and the first occurrence wins"),
|
|
("value-trimmed", "---\nSubject: spaced out \n---\n",
|
|
"the value is the rest of the line, trimmed"),
|
|
("colon-in-value", "---\nSubject: re: the thing\n---\n",
|
|
"only the first colon separates"),
|
|
("no-colon", "---\nSubject hi\n---\nbody\n",
|
|
"a malformed line invalidates the whole block, which becomes text"),
|
|
("bad-key", "---\nSub ject: hi\n---\nbody\n", "key charset is [A-Za-z0-9-]"),
|
|
("unterminated", "---\nSubject: hi\nbody\n", "no closing '---'"),
|
|
("too-many-keys", "---\n" + "".join(f"K{i}: v\n" for i in range(65)) + "---\nb",
|
|
"over 64 keys"),
|
|
("too-large", "---\nK: " + "v" * 4100 + "\n---\nb", "over 4 KiB"),
|
|
("injected-field", "---\nSubject: hi\nAccept: " + "a" * 8 + "\n---\nb",
|
|
"what a line break in a sender-supplied value produces: a second field, "
|
|
"signed and indistinguishable from an intended one. A client building "
|
|
"frontmatter MUST refuse a value containing a line break."),
|
|
]
|
|
out = []
|
|
for name, body, note in cases:
|
|
fields, rest = sm.parse_frontmatter(body)
|
|
out.append({"name": name, "note": note, "body": body,
|
|
"fields": fields, "text": rest,
|
|
"block_accepted": bool(fields) or rest != body})
|
|
assert out[2]["text"] == "---\nliteral dashes\n"
|
|
assert out[3]["fields"]["subject"] == "first"
|
|
assert all(not o["block_accepted"] for o in out if o["name"] in
|
|
{"no-colon", "bad-key", "unterminated", "too-many-keys", "too-large"})
|
|
return out
|
|
|
|
|
|
# --- rejections -------------------------------------------------------------
|
|
|
|
def reject_vectors(alice: sm.Account, bob: sm.Account) -> dict:
|
|
"""Inputs an implementation MUST refuse. These are the vectors that catch a
|
|
client which decrypts without verifying, or trusts `to` without checking it."""
|
|
to = alice.keys[0].pk
|
|
esk = bytes([0xA1]) * 32
|
|
with ephemeral(esk) as epk:
|
|
valid = sm.seal(bob.me, to, b"valid\n", when=WHEN, pad=False)
|
|
_, _, payload = plaintext_of(bob.me, to, b"valid\n", WHEN, epk, False)
|
|
assert sm.unseal(alice.keys, valid)["body"] == b"valid\n"
|
|
|
|
def mutate(env: bytes, at: int, raw: bytes) -> bytes:
|
|
return env[:at] + raw + env[at + len(raw):]
|
|
|
|
def repack(new_payload: bytes) -> bytes:
|
|
"""Re-seal a tampered payload so it passes the AEAD and fails deeper in."""
|
|
return encrypt(new_payload, to, esk)
|
|
|
|
bad_sig = repack(payload[:-sm.SIG_LEN] + bytes(sm.SIG_LEN))
|
|
wrong_to = repack(plaintext_of(bob.me, bob.me.pk, b"valid\n", WHEN, epk, False)[2])
|
|
long_body = repack(payload[:sm.PAYLOAD_HEADER - 4] + struct.pack(">I", 1 << 20)
|
|
+ payload[sm.PAYLOAD_HEADER:])
|
|
_, _, future = plaintext_of(bob.me, to, b"later\n", FUTURE, epk, False)
|
|
|
|
envelopes = [
|
|
("bad-magic", mutate(valid, 0, b"XMOL"), "magic is not 'SMOL' (§5.1)"),
|
|
("bad-envelope-version", mutate(valid, 4, b"\x02"),
|
|
"unknown envelope version; the sealed-box construction may differ"),
|
|
("truncated", valid[:sm.ENVELOPE_HEADER],
|
|
"shorter than the 69-byte header plus a 16-byte tag"),
|
|
("tag-flipped", valid[:-1] + bytes([valid[-1] ^ 0x01]),
|
|
"the Poly1305 tag no longer authenticates"),
|
|
("aad-tampered", mutate(valid, 40, bytes([valid[40] ^ 0x01])),
|
|
"epk is covered by the AAD, so altering it breaks the tag rather than "
|
|
"silently re-addressing the message"),
|
|
("not-for-us", repack(payload)[:5] + bob.me.pk + valid[37:],
|
|
"`to` names a key we do not hold; a client MUST check it against its own"),
|
|
("bad-signature", bad_sig,
|
|
"decrypts cleanly and the signature is zeroed. An implementation that "
|
|
"skips §5.3 verification will accept forged mail here."),
|
|
("signature-bound-to-other-recipient", wrong_to,
|
|
"a payload signed for bob, resealed to alice. The signature covers `to`, "
|
|
"so re-addressing a recorded message must not verify."),
|
|
("body-len-overflows", long_body, "body_len exceeds the payload (§5.3)"),
|
|
("payload-version", repack(b"\x02" + payload[1:]), "unknown payload version"),
|
|
("dated-in-the-future", repack(future),
|
|
f"time is {FUTURE}, more than 86400s ahead of any clock before 2100 (§5.3)"),
|
|
("low-order-epk", mutate(valid, 37, bytes.fromhex(LOW_ORDER[2])),
|
|
"a low-order ephemeral key, which §2 requires be refused"),
|
|
]
|
|
out = []
|
|
for name, env, why in envelopes:
|
|
out.append({"name": name, "why": why, "envelope": h(env),
|
|
"envelope_len": len(env),
|
|
"rejected_with": raises(lambda e=env: sm.unseal(alice.keys, e))})
|
|
|
|
low = [{"x25519_public": p,
|
|
"rejected_with": raises(lambda p=p: sm.agree(alice.keys[0].x_priv,
|
|
bytes.fromhex(p)))}
|
|
for p in LOW_ORDER]
|
|
keys = [{"identity_key": k,
|
|
"rejected_with": raises(lambda k=k: sm.ed_to_x25519_pub(bytes.fromhex(k)))}
|
|
for k in ["0000000000000000000000000000000000000000000000000000000000000000",
|
|
"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
|
|
"0100000000000000000000000000000000000000000000000000000000000000"]]
|
|
return {
|
|
"note": "Inputs a conformant implementation refuses. The envelope cases are "
|
|
"mutations of one valid message, listed with the reference's own "
|
|
"error, so a client can be tested against its own failure path.",
|
|
"valid_baseline": h(valid),
|
|
"envelopes": out,
|
|
"low_order_x25519": low,
|
|
"low_order_note": "§2 and §4: a received ephemeral public key that is a "
|
|
"low-order point MUST be refused, and an all-zero X25519 "
|
|
"output MUST be refused. libsodium reports the first as "
|
|
"the second, since the agreement output is zero.",
|
|
"invalid_identity_keys": keys,
|
|
"invalid_identity_note": "Keys that do not decode as Ed25519 points, or are "
|
|
"of small order, so §2's map to X25519 has no "
|
|
"valid result.",
|
|
}
|
|
|
|
|
|
# --- assembly ---------------------------------------------------------------
|
|
|
|
def build() -> dict:
|
|
alice = sm.Account(ALICE_MASTER, 1) # rotated once, so §7 has both keys
|
|
bob = sm.Account(BOB_MASTER, 0)
|
|
return {
|
|
"spec": "smolmail/1.2",
|
|
"generated_by": "vectors.py, from smolmail.py",
|
|
"note": "All values are hex unless the field name says otherwise. `alice[n]` "
|
|
"and `bob[n]` name a master and a rotation index from `identity`. "
|
|
"Masters are test data: never use them.",
|
|
"labels": {name: getattr(sm, name).decode() for name in
|
|
["LABEL_IDENTITY", "LABEL_ACCEPT", "LABEL_SEAL", "LABEL_MSG",
|
|
"LABEL_ID", "LABEL_MAC", "LABEL_AUTH", "LABEL_ROTATE",
|
|
"LABEL_REGISTER", "PROLOGUE"]},
|
|
"constants": {"default_port": sm.DEFAULT_PORT, "magic": sm.MAGIC.decode(),
|
|
"version": sm.VERSION, "envelope_header": sm.ENVELOPE_HEADER,
|
|
"payload_header": sm.PAYLOAD_HEADER, "pad_to": sm.PAD_TO,
|
|
"max_chain": sm.MAX_CHAIN, "max_skew": sm.MAX_SKEW,
|
|
"noise_protocol": sm.NOISE_PROTOCOL.decode()},
|
|
"identity": identity_vectors(),
|
|
"accept": accept_vectors(alice, bob),
|
|
"addressing": addressing_vectors(alice),
|
|
"envelopes": envelope_vectors(alice, bob),
|
|
"session": session_vectors(alice),
|
|
"rotation": rotation_vectors(alice),
|
|
"frontmatter": frontmatter_vectors(),
|
|
"rejects": reject_vectors(alice, bob),
|
|
}
|
|
|
|
|
|
def wrap(value: str, width: int = 76) -> str:
|
|
return "\n".join(value[i:i + width] for i in range(0, len(value), width))
|
|
|
|
|
|
def field(key: str, value, limit: int = 160) -> str:
|
|
"""Long hex goes in a fenced block; everything short stays on one line."""
|
|
if isinstance(value, str) and len(value) > limit:
|
|
return f"\n`{key}`\n\n```\n{wrap(value)}\n```\n"
|
|
if isinstance(value, bool):
|
|
value = "true" if value else "false"
|
|
return f"- `{key}`: `{value}`\n" if value != "" else f"- `{key}`: (empty)\n"
|
|
|
|
|
|
def markdown(v: dict) -> str:
|
|
L = v["labels"]
|
|
md = [
|
|
"<!-- SPDX-License-Identifier: CC-BY-SA-4.0 -->",
|
|
"",
|
|
f"# smolmail Test Vectors, version {v['spec'].split('/')[1]}",
|
|
"",
|
|
"Known-answer tests for [SPEC.md](SPEC.md), generated from the reference "
|
|
"implementation by [vectors.py](vectors.py). [vectors.json](vectors.json) "
|
|
"carries the same values in a form a test harness can load; this file is "
|
|
"for reading alongside the specification.",
|
|
"",
|
|
"Work through them in order. Derivation first (§2), because every later "
|
|
"vector depends on it, then an envelope you can seal and unseal (§5), then "
|
|
"the wire frames (§4, §6.1, §7). Finish with the rejections: they are the "
|
|
"ones that catch an implementation which decrypts without verifying.",
|
|
"",
|
|
"Hex throughout, unless a field name says otherwise. The two masters are "
|
|
"test data and the private keys here are public. Never use them for "
|
|
"anything.",
|
|
"",
|
|
"## Labels (§11)",
|
|
"",
|
|
"Every hash, signature and MAC is domain-separated by one of these, "
|
|
"prefixed to the input as raw bytes with no separator or length:",
|
|
"",
|
|
]
|
|
md += [f"- `{k}` = `{val}`" for k, val in L.items()]
|
|
md += ["", "## §2 Identity", "",
|
|
"`seed_n = HKDF-SHA256(master, salt = \"\", info = "
|
|
f"\"{L['LABEL_IDENTITY']}\" || n)`, with `n` a big-endian u32. The "
|
|
"Ed25519 public key of that seed *is* the identity. The X25519 keypair "
|
|
"comes from the same key via libsodium's conversions, and the accept key "
|
|
"is derived without the index so it survives rotation.", ""]
|
|
for who in v["identity"]:
|
|
md.append(f"### {who['name']}, rotation index {who['rotation_index']}")
|
|
md.append("")
|
|
for key in ["master", "rotation_index", "seed", "identity_key",
|
|
"x25519_private", "x25519_public", "base32", "fingerprint", "uri"]:
|
|
md.append(field(key, who[key]).rstrip("\n"))
|
|
md.append("")
|
|
md += ["### Accept keys and tokens (§5.8)", "",
|
|
v["accept"]["note"], ""]
|
|
md += [field(k, val).rstrip("\n") for k, val in v["accept"].items() if k != "note"]
|
|
md += ["",
|
|
"`alice_token_for_bob_rotated` is the token alice would issue to bob's "
|
|
"index-1 key. It differs from the one bob already holds, which is why "
|
|
"§5.8 says a token is derived from the correspondent's identity *as it "
|
|
"stood when they were accepted*: bob's rotation does not invalidate the "
|
|
"token he was given.", ""]
|
|
|
|
a = v["addressing"]
|
|
md += ["## §3 Addressing", "", a["note"], "", "### Usernames", "",
|
|
"| username | valid |", "| --- | --- |"]
|
|
md += [f"| `{u['username'] or '(empty)'}` | {'yes' if u['valid'] else 'no'} |"
|
|
for u in a["usernames"]]
|
|
md += ["", "### Addresses", "",
|
|
"| input | user | host | port | identity |", "| --- | --- | --- | --- | --- |"]
|
|
for ad in a["addresses"]:
|
|
ident = f"`{ad['identity'][:16]}…`" if ad["identity"] else "resolved"
|
|
md.append(f"| `{ad['input']}` | `{ad['user']}` | `{ad['host']}` | "
|
|
f"{ad['port']} | {ident} |")
|
|
md += ["", "Refused:", ""]
|
|
md += [f"- `{text}` — {why}" for text, why in a["rejected"].items()]
|
|
md += ["", "### Fingerprint", ""]
|
|
md += [field(k, val).rstrip("\n") for k, val in a["base32"].items()]
|
|
|
|
md += ["", "## §5 Messages", "",
|
|
"Each vector pins the ephemeral `esk` that §5.2 would draw at random and "
|
|
"the `time` that §5.3 would take from the clock. Nothing else in the "
|
|
"protocol is randomised, so with those two fixed an envelope reproduces "
|
|
"byte for byte.", "",
|
|
"Build one in this order: agree on `x25519_shared`, derive `seal_key`, "
|
|
"assemble `payload_header`, sign `signed_preimage`, concatenate the "
|
|
"payload, then encrypt it under `aad` with an all-zero nonce. If your "
|
|
"`seal_key` matches and your envelope does not, the fault is in the "
|
|
"payload layout; if `x25519_shared` already differs, it is in §2's map.",
|
|
"",
|
|
"`accept_mac` is what a sender holding the recipient's token attaches to "
|
|
"`SEND` (§5.8); `message_id` is what the server returns. The full payload "
|
|
"plaintext and the whole `SEND` request body are in vectors.json as "
|
|
"`payload` and `send_body`.", ""]
|
|
for e in v["envelopes"]:
|
|
md += [f"### {e['name']}", "", e["note"], ""]
|
|
for key in ["sender", "recipient", "body", "body_len", "time", "padded",
|
|
"esk", "epk", "x25519_shared", "seal_key", "aad",
|
|
"payload_header", "signed_preimage", "signature",
|
|
"payload_len", "envelope_len", "envelope", "envelope_sha256",
|
|
"message_id", "accept_mac"]:
|
|
md.append(field(key, e[key]).rstrip("\n"))
|
|
md.append("")
|
|
|
|
s = v["session"]
|
|
md += ["## §4 and §6.1 Session frames", "", s["note"], ""]
|
|
for key in ["handshake_hash", "server_static", "auth_signed_preimage",
|
|
"auth_signature", "auth_frame"]:
|
|
md.append(field(key, s[key]).rstrip("\n"))
|
|
md += ["", s["auth_frame_note"], ""]
|
|
for key in ["register_signed_preimage", "register_signature", "register_frame"]:
|
|
md.append(field(key, s[key]).rstrip("\n"))
|
|
md += ["", s["register_frame_note"], ""]
|
|
|
|
r = v["rotation"]
|
|
md += ["## §7 Key rotation", "", r["note"], ""]
|
|
for key in ["username", "old_key", "new_key", "time", "signed_preimage",
|
|
"sig_old", "sig_new", "certificate", "resolve_body"]:
|
|
md.append(field(key, r[key]).rstrip("\n"))
|
|
md += ["", "Chain walking:", ""]
|
|
md += [f"- {c['case']} → **{'accept' if c['accepted'] else 'refuse'}**"
|
|
+ (f" ({c['why']})" if c.get("why") else "") for c in r["walk_chain"]]
|
|
|
|
md += ["", "## §5.5 Frontmatter", "",
|
|
"A flat `Key: value` block, not YAML. A malformed line invalidates the "
|
|
"whole block, which is then shown as ordinary body text: frontmatter "
|
|
"fails closed toward display, never toward silent discard.", "",
|
|
"| case | input | fields | text | note |", "| --- | --- | --- | --- | --- |"]
|
|
def clip(text: str, keep: int = 48) -> str:
|
|
return repr(text) if len(text) <= keep else repr(text[:keep]) + "…"
|
|
|
|
for f in v["frontmatter"]:
|
|
md.append(f"| `{f['name']}` | `{clip(f['body'])}` | `{f['fields']}` | "
|
|
f"`{clip(f['text'])}` | {f['note']} |")
|
|
md += ["", "Long inputs are clipped here; vectors.json has them whole."]
|
|
|
|
rj = v["rejects"]
|
|
md += ["", "## Rejections", "", rj["note"], "",
|
|
"Each envelope is a mutation of this valid one:", "",
|
|
field("valid_baseline", rj["valid_baseline"]).rstrip("\n"), ""]
|
|
for case in rj["envelopes"]:
|
|
md += [f"### {case['name']}", "", case["why"], "",
|
|
field("envelope", case["envelope"]).rstrip("\n"),
|
|
field("rejected_with", case["rejected_with"]).rstrip("\n"), ""]
|
|
md += ["### Low-order X25519 keys", "", rj["low_order_note"], ""]
|
|
md += [f"- `{p['x25519_public']}`" for p in rj["low_order_x25519"]]
|
|
md += ["", "### Invalid identity keys", "", rj["invalid_identity_note"], ""]
|
|
md += [f"- `{k['identity_key']}` — {k['rejected_with']}"
|
|
for k in rj["invalid_identity_keys"]]
|
|
md += ["", "## License", "",
|
|
"These vectors are licensed under "
|
|
"[CC BY-SA 4.0](LICENSES/CC-BY-SA-4.0.txt), with the specification.", ""]
|
|
return re.sub(r"\n{3,}", "\n\n", "\n".join(md))
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
check = "--check" in argv
|
|
built = build()
|
|
rendered = markdown(built)
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
files = {"vectors.json": json.dumps(built, indent=2) + "\n",
|
|
"VECTORS.md": rendered}
|
|
if not check:
|
|
for name, text in files.items():
|
|
with open(os.path.join(here, name), "w", encoding="utf-8") as fh:
|
|
fh.write(text)
|
|
print(f"wrote {name} ({len(text)} bytes)")
|
|
counts = {k: len(built[k]) for k in ["identity", "envelopes", "frontmatter"]}
|
|
print(f"{counts['identity']} identities, {counts['envelopes']} envelopes, "
|
|
f"{len(built['rejects']['envelopes'])} rejections, "
|
|
f"{counts['frontmatter']} frontmatter cases: all verified")
|
|
return 0
|
|
|
|
failed = False
|
|
for name, text in files.items():
|
|
path = os.path.join(here, name)
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
have = fh.read()
|
|
except FileNotFoundError:
|
|
print(f"{name}: missing")
|
|
failed = True
|
|
continue
|
|
if have == text:
|
|
print(f"{name}: ok")
|
|
continue
|
|
failed = True
|
|
print(f"{name}: differs from what smolmail.py now produces")
|
|
for line in list(difflib.unified_diff(
|
|
have.splitlines(), text.splitlines(),
|
|
fromfile=f"{name} (committed)", tofile=f"{name} (regenerated)",
|
|
lineterm=""))[:40]:
|
|
print(f" {line}")
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|