From 06249ed244df7330c466f524866c7d5a5f67c0cd Mon Sep 17 00:00:00 2001 From: randogoth Date: Sat, 26 Sep 2026 11:01:41 +0300 Subject: [PATCH] feat: add single-file reference client --- README.md | 19 ++ SPEC.md | 2 + smolmail.py | 863 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 884 insertions(+) create mode 100755 smolmail.py diff --git a/README.md b/README.md index 6132d83..5d95c46 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,25 @@ uv run smolmaild.py serve `serve` listens on `127.0.0.1:1961` by default and stores mail in `mail.db`. Use `--host 0.0.0.0` to accept remote connections, and `--invite-token` to close registration. `--help` lists the size, quota, retention and rate limits. +## Reference client + +[smolmail.py](smolmail.py) is a complete client in one file, with the same inline dependencies: + +``` +uv run smolmail.py keygen +uv run smolmail.py trust example.org +uv run smolmail.py register alice@example.org +uv run smolmail.py resolve bob@example.org +echo "hello" | uv run smolmail.py send bob@example.org --subject Hi +uv run smolmail.py fetch +uv run smolmail.py list +uv run smolmail.py read +``` + +`import` adds a contact from a `smol://` address without trusting any server, `contacts` lists known keys and how each was learned, and `rotate` replaces the identity with a signed certificate your contacts accept automatically. + +Mail is stored sealed and opened on demand, so the local database holds no plaintext. Superseded keys are retained after a rotation, because mail already addressed to them is readable with nothing else. + ## Specification [SPEC.md](SPEC.md) — wire format, operations, trust model and conformance. diff --git a/SPEC.md b/SPEC.md index 51f86c2..41ae165 100644 --- a/SPEC.md +++ b/SPEC.md @@ -192,6 +192,8 @@ A server keeps each username's ordered chain of certificates and returns it with A server MUST retain every key ever bound to a username, MUST accept `SEND` addressed to any of them, and MUST return messages addressed to any of them on `FETCH`. Without this, mail from a contact who has not yet seen a rotation is addressed to a superseded key and becomes unreachable. +A client MUST likewise retain every private key it has rotated away from. Sealing is to a specific key (§5.2), so mail addressed to a superseded key is only ever readable with that key's private half. + **Rotation is not revocation.** An attacker holding a stolen identity key can rotate to their own key and the chain will validate. Clients MUST surface rotations in the interface rather than applying them invisibly; out-of-band re-verification is the only defence against a compromised identity key. There is no revocation mechanism in version 1. ## 8. Trust model diff --git a/smolmail.py b/smolmail.py new file mode 100755 index 0000000..83de2eb --- /dev/null +++ b/smolmail.py @@ -0,0 +1,863 @@ +#!/usr/bin/env -S uv run --quiet --script +# /// script +# requires-python = ">=3.11" +# dependencies = ["noiseprotocol>=0.3.1", "pynacl>=1.5"] +# /// +"""Smol Mail reference client. + +Implements SPEC.md version 1 from both ends: one Ed25519 identity, sealed and +signed messages, server key pinning, trust on first use with rotation chains, +and the five operations. + +Mail is stored sealed and opened on demand, so a stolen database is no more +readable than a stolen mailbox on a server. + + uv run smolmail.py keygen + uv run smolmail.py trust example.org + uv run smolmail.py register alice@example.org + uv run smolmail.py send bob@example.org --subject Hello + uv run smolmail.py fetch +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import hmac +import os +import re +import socket +import sqlite3 +import struct +import sys +import time + +import nacl.bindings as sodium +import nacl.exceptions +from nacl.signing import SigningKey, VerifyKey +from noise.connection import NoiseConnection + +NOISE_PROTOCOL = b"Noise_NX_25519_ChaChaPoly_SHA256" +PROLOGUE = b"smolmail/1" +LABEL_AUTH, LABEL_SEAL = b"smolmail/1 auth", b"smolmail/1 seal" +LABEL_MSG, LABEL_ID, LABEL_ROTATE = b"smolmail/1 msg", b"smolmail/1 id", b"smolmail/1 rotate" + +OP_AUTH, OP_RESOLVE, OP_SEND, OP_FETCH, OP_DELETE, OP_REGISTER = range(6) +STATUS = {0: "ok", 1: "malformed", 2: "bad version", 3: "unknown user", + 4: "auth required", 5: "auth failed", 6: "quota exceeded", 7: "too large", + 8: "rate limited", 9: "not permitted", 10: "internal error"} + +MAGIC, VERSION = b"SMOL", 1 +KEY_LEN, ID_LEN, SIG_LEN, CERT_LEN = 32, 16, 64, 136 +PAYLOAD_HEADER = 45 # version 1 + sender 32 + time 8 + body_len 4 +ENVELOPE_HEADER = 69 # magic 4 + version 1 + to 32 + epk 32 +MAX_CHAIN, PAD_TO = 16, 1024 +DEFAULT_PORT, MAX_FRAME, NOISE_PAYLOAD = 1961, 1 << 20, 65535 - 16 +FRONTMATTER_MAX, FRONTMATTER_KEYS = 4096, 64 + + +class SmolError(Exception): + """Anything the user should see as a message rather than a traceback.""" + + +# --- encoding --------------------------------------------------------------- + + +def b32(raw: bytes) -> str: + """RFC 4648 base32, lowercase and unpadded (§3).""" + return base64.b32encode(raw).decode("ascii").rstrip("=").lower() + + +def unb32(text: str) -> bytes: + text = text.strip().upper() + return base64.b32decode(text + "=" * (-len(text) % 8)) + + +def fingerprint(identity: bytes) -> str: + """First 20 characters of the base32 identity, in groups of four (§3).""" + s = b32(identity)[:20] + return " ".join(s[i : i + 4] for i in range(0, 20, 4)) + + +class Reader: + """Fail-closed reader; every parse raises rather than reading past the end.""" + + def __init__(self, buf: bytes) -> None: + self.buf, self.pos = buf, 0 + + def take(self, n: int) -> bytes: + if n < 0 or self.pos + n > len(self.buf): + raise SmolError("truncated response from server") + self.pos += n + return self.buf[self.pos - n : self.pos] + + def u8(self) -> int: + return self.take(1)[0] + + def u16(self) -> int: + return struct.unpack(">H", self.take(2))[0] + + def u32(self) -> int: + return struct.unpack(">I", self.take(4))[0] + + def i64(self) -> int: + return struct.unpack(">q", self.take(8))[0] + + def left(self) -> int: + return len(self.buf) - self.pos + + +ADDRESS = re.compile(r"^(?P[a-z0-9._-]{1,63})@(?P[^/:]+)(?::(?P\d+))?$") + + +class Address: + """A short `user@host` address, or a self-certifying `smol://` one (§3).""" + + def __init__(self, user: str, host: str, port: int, identity: bytes | None = None) -> None: + self.user, self.host, self.port, self.identity = user, host, port, identity + + @classmethod + def parse(cls, text: str) -> "Address": + text, identity = text.strip(), None + if text.startswith("smol://"): + rest = text[len("smol://") :] + if "/" not in rest: + raise SmolError(f"{text}: smol:// address carries no key") + rest, key = rest.rsplit("/", 1) + try: + identity = unb32(key) + except Exception as exc: + raise SmolError(f"{text}: undecodable key: {exc}") from None + if len(identity) != KEY_LEN: + raise SmolError(f"{text}: key is {len(identity)} bytes, expected {KEY_LEN}") + text = rest + m = ADDRESS.match(text) + if not m: + raise SmolError(f"{text!r} is not a valid address") + if m["user"][0] in "._-" or m["user"][-1] in "._-": + raise SmolError(f"{m['user']!r} may not begin or end with a separator") + return cls(m["user"], m["host"], int(m["port"] or DEFAULT_PORT), identity) + + @property + def short(self) -> str: + return f"{self.user}@{self.host}" + ("" if self.port == DEFAULT_PORT else f":{self.port}") + + def uri(self, identity: bytes) -> str: + return f"smol://{self.short}/{b32(identity)}" + + +# --- identity and message cryptography (§2, §5) ----------------------------- + + +def hkdf_sha256(ikm: bytes, salt: bytes, info: bytes, length: int = 32) -> bytes: + """RFC 5869, small enough to carry inline rather than add a dependency for.""" + prk = hmac.new(salt, ikm, hashlib.sha256).digest() + out, block, counter = b"", b"", 1 + while len(out) < length: + block = hmac.new(prk, block + info + bytes([counter]), hashlib.sha256).digest() + out, counter = out + block, counter + 1 + return out[:length] + + +def ed_to_x25519_pub(identity: bytes) -> bytes: + """§2, via libsodium's own conversion as the spec prefers.""" + try: + return sodium.crypto_sign_ed25519_pk_to_curve25519(identity) + except (nacl.exceptions.CryptoError, RuntimeError, ValueError) as exc: + raise SmolError(f"not a valid Ed25519 public key: {exc}") from None + + +class Identity: + """An Ed25519 keypair plus the X25519 keypair derived from it (§2).""" + + def __init__(self, seed: bytes) -> None: + if len(seed) != KEY_LEN: + raise SmolError(f"identity seed must be {KEY_LEN} bytes, got {len(seed)}") + self.seed = seed + self._signing = SigningKey(seed) + self.pk: bytes = bytes(self._signing.verify_key) + self.x_priv: bytes = sodium.crypto_sign_ed25519_sk_to_curve25519( + sodium.crypto_sign_seed_keypair(seed)[1]) + + def sign(self, message: bytes) -> bytes: + return self._signing.sign(message).signature + + +def verify_sig(identity: bytes, signature: bytes, message: bytes) -> bool: + try: + VerifyKey(identity).verify(message, signature) + return True + except (nacl.exceptions.BadSignatureError, ValueError): + return False + + +def agree(x_priv: bytes, x_pub: bytes) -> bytes: + """X25519 with §3.2's checks; libsodium rejects low-order points itself.""" + try: + shared = sodium.crypto_scalarmult(x_priv, x_pub) + except (RuntimeError, nacl.exceptions.CryptoError) as exc: + raise SmolError(f"rejected key agreement: {exc}") from None + if shared == bytes(KEY_LEN): + raise SmolError("rejected all-zero key agreement output") + return shared + + +def message_id(envelope: bytes) -> bytes: + """§5.4, derived from the envelope so no sender can choose it.""" + return hashlib.sha256(LABEL_ID + envelope).digest()[:ID_LEN] + + +def seal(sender: Identity, recipient: bytes, body: bytes, when: int | None = None, + pad: bool = True) -> bytes: + """§5.2 and §5.3. Ephemeral-static, so the nonce is fixed and the sender's + long-term key never takes part in key agreement.""" + when = int(time.time()) if when is None else when + esk = os.urandom(KEY_LEN) + epk = sodium.crypto_scalarmult_base(esk) + key = hkdf_sha256(agree(esk, ed_to_x25519_pub(recipient)), epk + recipient, LABEL_SEAL) + del esk # best effort; CPython offers no way to wipe bytes + + header = bytes([VERSION]) + sender.pk + struct.pack(">qI", when, len(body)) + plaintext = header + body + sender.sign(LABEL_MSG + recipient + epk + header + body) + if pad: + plaintext += bytes(-len(plaintext) % PAD_TO) + aad = MAGIC + bytes([VERSION]) + recipient + epk + return aad + sodium.crypto_aead_chacha20poly1305_ietf_encrypt( + plaintext, aad, bytes(12), key) + + +def unseal(identities: list[Identity], envelope: bytes) -> dict: + """Inverse of seal(); raises unless signature and recipient both check out. + + `identities` may include retired keys: sealing is to a specific key, so mail + addressed to a superseded one needs that key (§7). + """ + if len(envelope) < ENVELOPE_HEADER + 16: + raise SmolError("envelope too short") + if envelope[:4] != MAGIC: + raise SmolError("not a Smol Mail envelope") + if envelope[4] != VERSION: + raise SmolError(f"unsupported envelope version {envelope[4]}") + to, epk, sealed = envelope[5:37], envelope[37:69], envelope[69:] + + me = next((i for i in identities if i.pk == to), None) + if me is None: + raise SmolError(f"addressed to {b32(to)[:16]}…, not one of our keys") + key = hkdf_sha256(agree(me.x_priv, epk), epk + to, LABEL_SEAL) + try: + plaintext = sodium.crypto_aead_chacha20poly1305_ietf_decrypt( + sealed, envelope[:ENVELOPE_HEADER], bytes(12), key) + except nacl.exceptions.CryptoError: + raise SmolError("decryption failed: wrong key or corrupt envelope") from None + + r = Reader(plaintext) + if r.u8() != VERSION: + raise SmolError("unsupported payload version") + sender, when, body_len = r.take(KEY_LEN), r.i64(), r.u32() + if body_len > r.left(): + raise SmolError("payload body length exceeds the payload") + body, signature = r.take(body_len), r.take(SIG_LEN) # trailing bytes are padding + if not verify_sig(sender, signature, + LABEL_MSG + to + epk + plaintext[:PAYLOAD_HEADER] + body): + raise SmolError("signature does not verify") + return {"sender": sender, "time": when, "body": body, "id": message_id(envelope)} + + +# --- body frontmatter (§5.5) ------------------------------------------------ + +FM_KEY = re.compile(r"^[A-Za-z0-9-]{1,64}$") + + +def parse_frontmatter(body: str) -> tuple[dict[str, str], str]: + """§5.5. A flat `Key: value` block, deliberately not YAML. + + Any malformed line invalidates the whole block, which is then returned as + ordinary body text: frontmatter fails closed toward display. + """ + if not body.startswith("---\n"): + return {}, body + lines = body.split("\n") + try: + close = lines.index("---", 1) + except ValueError: + return {}, body + block, rest = lines[1:close], "\n".join(lines[close + 1 :]) + if len(block) > FRONTMATTER_KEYS: + return {}, body + if sum(len(line.encode()) + 1 for line in block) > FRONTMATTER_MAX: + return {}, body + fields: dict[str, str] = {} + for line in block: + key, sep, value = line.partition(":") + if not sep or not FM_KEY.match(key): + return {}, body + fields.setdefault(key, value.strip()) # first occurrence wins + return fields, rest + + +def build_frontmatter(fields: list[tuple[str, str]], body: str) -> str: + """Emit a block only when needed, including to escape a body that genuinely + begins with `---` (§5.5).""" + if not fields and not body.startswith("---\n"): + return body + return "---\n" + "".join(f"{k}: {v}\n" for k, v in fields) + "---\n" + body + + +# --- local state ------------------------------------------------------------ + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS account ( + id INTEGER PRIMARY KEY CHECK (id = 1), username TEXT, host TEXT, port INTEGER); +CREATE TABLE IF NOT EXISTS servers ( + host TEXT PRIMARY KEY, static BLOB NOT NULL, pinned_at INTEGER NOT NULL); +CREATE TABLE IF NOT EXISTS contacts ( + address TEXT PRIMARY KEY, identity BLOB NOT NULL, + verified INTEGER NOT NULL, -- 1 when the key came from a smol:// address + seen_at INTEGER NOT NULL); +-- Private keys rotated away from. Mail sealed to a superseded key is readable +-- only with that key, so these can never be discarded (§7). +CREATE TABLE IF NOT EXISTS retired ( + identity BLOB PRIMARY KEY, seed BLOB NOT NULL, retired_at INTEGER NOT NULL); +-- Envelopes are stored sealed and opened on demand; no plaintext at rest. +CREATE TABLE IF NOT EXISTS inbox ( + id BLOB PRIMARY KEY, envelope BLOB NOT NULL, received_at INTEGER NOT NULL); +CREATE TABLE IF NOT EXISTS sent ( + id BLOB PRIMARY KEY, recipient TEXT NOT NULL, + envelope BLOB NOT NULL, sent_at INTEGER NOT NULL); +""" + + +class Store: + def __init__(self, path: str) -> None: + self.db = sqlite3.connect(path) + self.db.executescript(SCHEMA) + self.db.commit() + + def all(self, sql: str, *args) -> list[tuple]: + return self.db.execute(sql, args).fetchall() + + def one(self, sql: str, *args) -> tuple | None: + return self.db.execute(sql, args).fetchone() + + def run(self, sql: str, *args) -> sqlite3.Cursor: + with self.db: + return self.db.execute(sql, args) + + def account(self) -> Address | None: + row = self.one("SELECT username, host, port FROM account WHERE id = 1") + return Address(row[0], row[1], row[2]) if row and row[0] else None + + def set_account(self, addr: Address) -> None: + self.run("INSERT INTO account (id, username, host, port) VALUES (1, ?1, ?2, ?3) " + "ON CONFLICT (id) DO UPDATE SET username = ?1, host = ?2, port = ?3", + addr.user, addr.host, addr.port) + + def contact(self, address: str) -> tuple[bytes, bool] | None: + row = self.one("SELECT identity, verified FROM contacts WHERE address = ?", address) + return (row[0], bool(row[1])) if row else None + + def save_contact(self, address: str, identity: bytes, verified: bool) -> None: + self.run("INSERT INTO contacts (address, identity, verified, seen_at) VALUES (?,?,?,?) " + "ON CONFLICT (address) DO UPDATE SET identity = ?2, verified = ?3, seen_at = ?4", + address, identity, int(verified), int(time.time())) + + def pin_server(self, host: str, static: bytes) -> None: + self.run("INSERT INTO servers (host, static, pinned_at) VALUES (?,?,?) " + "ON CONFLICT (host) DO UPDATE SET static = ?2, pinned_at = ?3", + host, static, int(time.time())) + + def identities(self, me: Identity) -> list[Identity]: + """The current key plus every key rotated away from.""" + return [me] + [Identity(row[0]) for row in + self.all("SELECT seed FROM retired ORDER BY retired_at")] + + def mail(self, box: str) -> list[tuple]: + if box == "sent": + return self.all("SELECT id, envelope, sent_at, recipient FROM sent " + "ORDER BY sent_at, id") + return self.all("SELECT id, envelope, received_at, NULL FROM inbox " + "ORDER BY received_at, id") + + +# --- transport (§4) --------------------------------------------------------- + + +class Conn: + """A Noise_NX initiator session with pinning layered on top (§4).""" + + def __init__(self, host: str, port: int, pinned: bytes | None) -> None: + try: + self.sock = socket.create_connection((host, port), timeout=30) + except OSError as exc: + raise SmolError(f"cannot reach {host}:{port}: {exc}") from None + noise = NoiseConnection.from_name(NOISE_PROTOCOL) + noise.set_as_initiator() + noise.set_prologue(PROLOGUE) + noise.start_handshake() + first = noise.write_message() + self.sock.sendall(struct.pack(">H", len(first)) + first) + # The library drops the peer's static key once the handshake ends, so + # keep the handshake state to read it afterwards. + state = noise.noise_protocol.handshake_state + noise.read_message(self._recv(struct.unpack(">H", self._recv(2))[0])) + if not noise.handshake_finished: + raise SmolError("handshake did not complete") + + self.server_static = bytes(state.rs.public_bytes) + if pinned is not None and not hmac.compare_digest(pinned, self.server_static): + self.close() + raise SmolError(f"{host} presented a different key than the one pinned\n" + f" pinned: {b32(pinned)}\n" + f" presented: {b32(self.server_static)}") + self.pinned = pinned is not None + self.noise, self.buf = noise, bytearray() + self.handshake_hash: bytes = noise.get_handshake_hash() + + def _recv(self, n: int) -> bytes: + out = b"" + while len(out) < n: + chunk = self.sock.recv(n - len(out)) + if not chunk: + raise SmolError("server closed the connection") + out += chunk + return out + + def call(self, op: int, body: bytes = b"") -> tuple[int, bytes]: + frame = struct.pack(">I", 1 + len(body)) + bytes([op]) + body + if len(frame) > MAX_FRAME + 4: + raise SmolError("request exceeds the maximum frame size") + for off in range(0, len(frame), NOISE_PAYLOAD): + packet = self.noise.encrypt(frame[off : off + NOISE_PAYLOAD]) + self.sock.sendall(struct.pack(">H", len(packet)) + packet) + while len(self.buf) < 5: + self.buf += self.noise.decrypt(self._recv(struct.unpack(">H", self._recv(2))[0])) + (length,) = struct.unpack(">I", self.buf[:4]) + if not 1 <= length <= MAX_FRAME: + raise SmolError(f"server sent a frame of length {length}") + while len(self.buf) < 4 + length: + self.buf += self.noise.decrypt(self._recv(struct.unpack(">H", self._recv(2))[0])) + payload = bytes(self.buf[4 : 4 + length]) + del self.buf[: 4 + length] + return payload[1], payload[2:] # status, body (§6.1) + + def close(self) -> None: + try: + self.sock.close() + except OSError: + pass + + def __enter__(self) -> "Conn": + return self + + def __exit__(self, *exc) -> None: + self.close() + + +def connect(store: Store, addr: Address, require_pin: bool) -> Conn: + """Open a session, enforcing §4's rule about unpinned servers.""" + row = store.one("SELECT static FROM servers WHERE host = ?", addr.host) + pinned = row[0] if row else None + if pinned is None and require_pin: + raise SmolError(f"no pinned key for {addr.host}.\nObtain it from the operator " + f"through a trusted channel, then:\n" + f" smolmail.py trust {addr.host} ") + conn = Conn(addr.host, addr.port, pinned) + if pinned is None: + warn(f"{addr.host} is not pinned; its key is {b32(conn.server_static)}") + return conn + + +def expect_ok(status: int, what: str) -> None: + if status != 0: + raise SmolError(f"{what} failed: {STATUS.get(status, status)} ({status})") + + +# --- operations ------------------------------------------------------------- + + +def do_resolve(conn: Conn, user: str) -> tuple[bytes, list[bytes]]: + """RESOLVE, returning the current key and its rotation chain (§6.1).""" + name = user.encode() + status, body = conn.call(OP_RESOLVE, bytes([len(name)]) + name) + expect_ok(status, f"resolving {user}") + r = Reader(body) + identity = r.take(KEY_LEN) + return identity, [r.take(CERT_LEN) for _ in range(r.u8())] + + +def walk_chain(pinned: bytes, current: bytes, chain: list[bytes]) -> bool: + """§7. Accept a key change only when a signed chain leads from the key we + hold to the one the server now returns.""" + if hmac.compare_digest(pinned, current): + return True + if not chain or len(chain) > MAX_CHAIN: + return False + key, started = pinned, False + for cert in chain: + old, new, when, sig = cert[:32], cert[32:64], cert[64:72], cert[72:] + if not started: + if old != key: + continue # a link predating the key we hold + started = True + elif old != key: + return False # the chain is not continuous + if not verify_sig(old, sig, LABEL_ROTATE + old + new + when): + return False + key = new + return started and hmac.compare_digest(key, current) + + +def trust_key(store: Store, conn: Conn, addr: Address) -> bytes: + """Resolve a contact and apply §8's trust rules.""" + identity, chain = do_resolve(conn, addr.user) + known = store.contact(addr.short) + if known is None: + store.save_contact(addr.short, identity, verified=False) + qualifier = "" if conn.pinned else ", UNVERIFIED server" + print(f"{addr.short} {b32(identity)}\n pinned (trust on first use{qualifier})") + return identity + old, verified = known + if hmac.compare_digest(old, identity): + return identity + if walk_chain(old, identity, chain): + store.save_contact(addr.short, identity, verified) + warn(f"{addr.short} rotated its key; a signed chain confirms it") + print(f" now {b32(identity)}") + return identity + raise SmolError(f"{addr.short} presents a different key with no valid rotation chain.\n" + f" known: {b32(old)}\n offered: {b32(identity)}\n" + f"Verify out of band, then: smolmail.py import {addr.uri(identity)}") + + +def authenticate(conn: Conn, addr: Address, me: Identity) -> None: + """§4 session authentication: sign the handshake hash.""" + name = addr.user.encode() + status, _ = conn.call(OP_AUTH, bytes([len(name)]) + name + me.pk + + me.sign(LABEL_AUTH + conn.handshake_hash)) + expect_ok(status, "authentication") + + +# --- helpers ---------------------------------------------------------------- + + +def warn(message: str) -> None: + print(f"warning: {message}", file=sys.stderr) + + +def load_identity(args: argparse.Namespace) -> Identity: + try: + with open(args.key, "rb") as fh: + return Identity(fh.read()) + except FileNotFoundError: + raise SmolError(f"no identity at {args.key}; run: smolmail.py keygen") from None + + +def write_seed(path: str, seed: bytes) -> None: + """0600 before any bytes land, so the seed is never briefly world-readable.""" + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "wb") as fh: + fh.write(seed) + + +def describe(store: Store, envelope: bytes, identities: list[Identity]) -> dict: + """Open an envelope and split its body into frontmatter and text.""" + opened = unseal(identities, envelope) + fields, text = parse_frontmatter(opened["body"].decode("utf-8", "replace")) + sender = opened["sender"] + known = next((a for a, i, _ in store.all( + "SELECT address, identity, verified FROM contacts") if hmac.compare_digest(i, sender)), None) + opened |= {"fields": fields, "text": text, "subject": fields.get("Subject", ""), + "from": known or f"<{b32(sender)[:20]}…>"} + return opened + + +# --- commands --------------------------------------------------------------- + + +def cmd_keygen(args: argparse.Namespace, store: Store) -> int: + if os.path.exists(args.key) and not args.force: + raise SmolError(f"{args.key} exists; refusing to overwrite (use --force)") + me = Identity(os.urandom(KEY_LEN)) + write_seed(args.key, me.seed) + print(f"identity: {args.key} (back this up; it is the only secret)") + print(f"public key: {b32(me.pk)}\nfingerprint: {fingerprint(me.pk)}") + return 0 + + +def cmd_whoami(args: argparse.Namespace, store: Store) -> int: + me = load_identity(args) + print(f"public key: {b32(me.pk)}\nfingerprint: {fingerprint(me.pk)}") + addr = store.account() + print(f"address: {addr.short}\nuri: {addr.uri(me.pk)}" if addr + else "address: (not registered)") + retired = len(store.all("SELECT 1 FROM retired")) + if retired: + print(f"retired: {retired} superseded key(s), kept to read old mail") + return 0 + + +def cmd_trust(args: argparse.Namespace, store: Store) -> int: + try: + key = unb32(args.key_b32) + except Exception as exc: + raise SmolError(f"undecodable key: {exc}") from None + if len(key) != KEY_LEN: + raise SmolError(f"server key is {len(key)} bytes, expected {KEY_LEN}") + host = args.host.split(":")[0] + row = store.one("SELECT static FROM servers WHERE host = ?", host) + if row and row[0] != key and not args.force: + raise SmolError(f"{host} is already pinned to {b32(row[0])}; use --force to replace") + store.pin_server(host, key) + print(f"pinned {host} {b32(key)}") + return 0 + + +def cmd_register(args: argparse.Namespace, store: Store) -> int: + me, addr = load_identity(args), Address.parse(args.address) + name, token = addr.user.encode(), (args.token or "").encode() + body = (bytes([len(name)]) + name + me.pk + bytes([len(token)]) + token + b"\0") + with connect(store, addr, require_pin=True) as conn: + status, _ = conn.call(OP_REGISTER, body) + expect_ok(status, f"registering {addr.short}") + store.set_account(addr) + print(f"registered {addr.short}\nshare: {addr.uri(me.pk)}") + return 0 + + +def cmd_resolve(args: argparse.Namespace, store: Store) -> int: + addr = Address.parse(args.address) + if addr.identity is not None: + raise SmolError("that address already carries a key; use `import` instead") + with connect(store, addr, require_pin=False) as conn: + trust_key(store, conn, addr) + return 0 + + +def cmd_import(args: argparse.Namespace, store: Store) -> int: + addr = Address.parse(args.uri) + if addr.identity is None: + raise SmolError("import needs a smol:// address carrying a key") + store.save_contact(addr.short, addr.identity, verified=True) + print(f"imported {addr.short} {b32(addr.identity)} (verified)") + print(f"fingerprint: {fingerprint(addr.identity)}") + return 0 + + +def cmd_contacts(args: argparse.Namespace, store: Store) -> int: + rows = store.all("SELECT address, identity, verified FROM contacts ORDER BY address") + if not rows: + print("no contacts") + return 0 + width = max(len(a) for a, _, _ in rows) + for address, identity, verified in rows: + print(f"{address:<{width}} {b32(identity)} {'verified' if verified else 'tofu'}") + return 0 + + +def cmd_send(args: argparse.Namespace, store: Store) -> int: + me, addr = load_identity(args), Address.parse(args.address) + if args.body is not None: + text = args.body + elif sys.stdin.isatty(): + raise SmolError("no message body; pass --body or pipe it on stdin") + else: + text = sys.stdin.read() + + # Prefer a key we already trust; fall back to RESOLVE with trust on first use. + if addr.identity is not None: + recipient = addr.identity + store.save_contact(addr.short, recipient, verified=True) + elif (known := store.contact(addr.short)) is not None: + recipient = known[0] + else: + with connect(store, addr, require_pin=False) as conn: + recipient = trust_key(store, conn, addr) + + fields = [(k, v) for k, v in (("Subject", args.subject), ("In-Reply-To", args.reply_to)) if v] + for raw in args.header or []: + key, sep, value = raw.partition(":") + if not sep or not FM_KEY.match(key.strip()): + raise SmolError(f"{raw!r} is not a valid `Key: value` header") + fields.append((key.strip(), value.strip())) + + body = build_frontmatter(fields, text).encode() + envelope = seal(me, recipient, body, pad=not args.no_pad) + with connect(store, addr, require_pin=False) as conn: + status, payload = conn.call(OP_SEND, envelope) + expect_ok(status, f"sending to {addr.short}") + mid = message_id(envelope) + if payload and payload != mid: + warn("server returned an id we did not derive; it is not authoritative") + # §5.6: the ephemeral is gone, so keep a copy sealed to ourselves. + store.run("INSERT OR IGNORE INTO sent (id, recipient, envelope, sent_at) VALUES (?,?,?,?)", + mid, addr.short, seal(me, me.pk, body, pad=not args.no_pad), int(time.time())) + print(f"sent {mid.hex()} to {addr.short} ({len(envelope)} bytes)") + return 0 + + +def cmd_fetch(args: argparse.Namespace, store: Store) -> int: + me = load_identity(args) + addr = store.account() + if addr is None: + raise SmolError("not registered; run: smolmail.py register ") + identities = store.identities(me) + stored = rejected = total = 0 + + with connect(store, addr, require_pin=True) as conn: + authenticate(conn, addr, me) + while True: + status, body = conn.call(OP_FETCH) + expect_ok(status, "fetching") + r = Reader(body) + count = r.u16() + if count == 0: + break + acked = [] + for _ in range(count): + total += 1 + mid, received_at = r.take(ID_LEN), r.i64() + envelope = r.take(r.u32()) + try: + if message_id(envelope) != mid: + raise SmolError("id does not match the envelope") + unseal(identities, envelope) + except SmolError as exc: + # Left on the server rather than destroyed, so a client-side + # bug cannot lose mail. + warn(f"{mid.hex()}: {exc}; left on server") + rejected += 1 + continue + if store.run("INSERT OR IGNORE INTO inbox (id, envelope, received_at) " + "VALUES (?,?,?)", mid, envelope, received_at).rowcount: + stored += 1 + acked.append(mid) + if not acked or args.keep: + break + status, _ = conn.call(OP_DELETE, struct.pack(">H", len(acked)) + b"".join(acked)) + expect_ok(status, "acknowledging") + + print(f"{total} message(s): {stored} new, {rejected} rejected" + + (" (left on the server)" if args.keep and total else "")) + return 0 + + +def cmd_list(args: argparse.Namespace, store: Store) -> int: + me = load_identity(args) + identities, box = store.identities(me), "sent" if args.sent else "inbox" + rows = store.mail(box) + if not rows: + print(f"{box} is empty") + return 0 + for mid, envelope, when, recipient in rows: + try: + opened = describe(store, envelope, identities) + who = recipient if box == "sent" else opened["from"] + subject = opened["subject"] or "(no subject)" + except SmolError as exc: + who, subject = "?", f"" + stamp = time.strftime("%Y-%m-%d %H:%M", time.localtime(when)) + print(f"{mid.hex()[:8]} {stamp} {who:<28.28} {subject}") + return 0 + + +def cmd_read(args: argparse.Namespace, store: Store) -> int: + me = load_identity(args) + identities, box = store.identities(me), "sent" if args.sent else "inbox" + matches = [r for r in store.mail(box) if r[0].hex().startswith(args.id.lower())] + if not matches: + raise SmolError(f"no message in {box} matching {args.id!r}") + if len(matches) > 1: + raise SmolError(f"{args.id!r} matches {len(matches)} messages; be more specific") + mid, envelope, _, recipient = matches[0] + opened = describe(store, envelope, identities) + print(f"id: {mid.hex()}") + if box == "sent": + print(f"to: {recipient}") + print(f"from: {opened['from']}\nkey: {b32(opened['sender'])}") + print("date: " + time.strftime("%Y-%m-%d %H:%M:%S %z", time.localtime(opened["time"]))) + for key, value in opened["fields"].items(): + print(f"{key.lower() + ':':<9}{value}") + print("signature verified\n") + print(opened["text"], end="" if opened["text"].endswith("\n") else "\n") + return 0 + + +def cmd_rotate(args: argparse.Namespace, store: Store) -> int: + old = load_identity(args) + addr = store.account() + if addr is None: + raise SmolError("not registered; run: smolmail.py register ") + new = Identity(os.urandom(KEY_LEN)) + when = struct.pack(">q", int(time.time())) + cert = old.pk + new.pk + when + old.sign(LABEL_ROTATE + old.pk + new.pk + when) + + name = addr.user.encode() + body = bytes([len(name)]) + name + new.pk + b"\0" + bytes([len(cert)]) + cert + with connect(store, addr, require_pin=True) as conn: + status, _ = conn.call(OP_REGISTER, body) + expect_ok(status, "rotating") + + # Retain the old seed first: mail already sealed to it is otherwise unreadable. + store.run("INSERT OR IGNORE INTO retired (identity, seed, retired_at) VALUES (?,?,?)", + old.pk, old.seed, int(time.time())) + write_seed(args.key, new.seed) + print(f"rotated {addr.short}\nnew key: {b32(new.pk)}") + print(f"fingerprint: {fingerprint(new.pk)}\nshare: {addr.uri(new.pk)}") + print("\nTell your contacts; they will accept the change from the signed chain.") + return 0 + + +# --- CLI -------------------------------------------------------------------- + +COMMANDS = [ + ("keygen", cmd_keygen, "create an identity", + [(("--force",), {"action": "store_true"})]), + ("whoami", cmd_whoami, "show this identity", []), + ("trust", cmd_trust, "pin a server's static key", + [(("host",), {}), (("key_b32",), {"metavar": "key"}), + (("--force",), {"action": "store_true"})]), + ("register", cmd_register, "bind this identity to a username", + [(("address",), {}), (("--token",), {"help": "invite token, if required"})]), + ("resolve", cmd_resolve, "look up and pin a contact's key", [(("address",), {})]), + ("import", cmd_import, "add a contact from a smol:// address", [(("uri",), {})]), + ("contacts", cmd_contacts, "list known keys", []), + ("send", cmd_send, "seal and deliver a message", + [(("address",), {}), (("--subject",), {}), + (("--reply-to",), {"metavar": "ID"}), + (("--body",), {"help": "message text; read from stdin when omitted"}), + (("--header",), {"action": "append", "metavar": "KEY:VALUE", + "help": "extra frontmatter field; repeatable"}), + (("--no-pad",), {"action": "store_true", "help": "do not pad to 1 KiB"})]), + ("fetch", cmd_fetch, "retrieve, verify and acknowledge mail", + [(("--keep",), {"action": "store_true", "help": "do not delete from the server"})]), + ("list", cmd_list, "list stored mail", [(("--sent",), {"action": "store_true"})]), + ("read", cmd_read, "show one message", + [(("id",), {}), (("--sent",), {"action": "store_true"})]), + ("rotate", cmd_rotate, "replace this identity, signing the change", []), +] + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--key", default="identity.key", help="identity seed file") + parser.add_argument("--db", default="smolmail.db", help="local state and mail") + sub = parser.add_subparsers(dest="command", required=True) + for name, func, help_text, arguments in COMMANDS: + child = sub.add_parser(name, help=help_text) + child.set_defaults(func=func) + for flags, options in arguments: + child.add_argument(*flags, **options) + + args = parser.parse_args(argv) + try: + return args.func(args, Store(args.db)) + except SmolError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + except KeyboardInterrupt: + return 130 + + +if __name__ == "__main__": + sys.exit(main())