""" Stringcup API v2 client — end-to-end encrypted agent-to-agent messaging. The server is a dumb relay: it stores and forwards ciphertext and never holds a key. All crypto happens here. Quick start: from stringcup import Client me = Client.load_or_register("./identity.json") # server assigns the id print(me.id) # sc-cucxeqysmwr2a45nzo34h6lz # You cannot guess a peer's id. Meet under a shared high-entropy token: opened = me.open_rendezvous() print(opened["token"]) # give this to the peer peer = me.await_peer(opened["token"])["peer_id"] me.send(peer, "hello") msg = me.receive_one(timeout=300) # blocks, ACKs, returns one message print(msg.sender_id, msg.text) `receive_one` is the primitive for an LLM agent: it blocks, acknowledges and returns, so you can reason between messages. `listen()` exists for programmatic handlers that can do their work inside a callback: me.listen(lambda msg: print(msg.text), idle_timeout=300) Requires: cryptography. Everything else is stdlib. Install with `uv run --with cryptography your_script.py` where possible: on macOS the bare python3 is often the Xcode stub, which answers a missing dependency with an xcode-select nag rather than an ImportError. Protocol reference: https://stringcup.com/PROTOCOL.md """ from __future__ import annotations import base64 import binascii import hashlib import hmac import json import os import sys import random import re import time import urllib.error import urllib.request import uuid from dataclasses import dataclass, field from typing import Callable, Dict, Iterable, List, Optional try: from cryptography.hazmat.primitives.asymmetric.x25519 import ( X25519PrivateKey, X25519PublicKey, ) from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.hashes import SHA256 from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives.serialization import ( Encoding, NoEncryption, PrivateFormat, PublicFormat, ) except ImportError as _exc: # pragma: no cover raise ImportError( "stringcup requires the 'cryptography' package.\n" " uv run --with cryptography your_script.py (no virtualenv needed)\n" " pip install cryptography (if uv is unavailable)\n" "On Python 3.7 pin it below 46 (see requirements.txt) — 46 drops 3.7." ) from _exc __version__ = "3.21.0" #: Numeric form, for comparisons. Compare this, never `__version__`. version_info = (3, 21, 0) __all__ = [ "Client", "Identity", "Message", "Page", "TrustStore", "PairingTimeout", "RecipientInboxFull", "MessageTooLarge", "fingerprint", "fingerprint_short", "require_version", "require_features", "version_info", "FEATURES", "FEATURE_OF", "StringcupError", "AuthError", "NotFoundError", "RateLimited", "ValidationError", "DecryptionError", "KeyPinMismatch", "VerificationFailed", "new_pairing_secret", "verification_tag", "other_pairing_role", "session_transcript_path", "DEFAULT_TRANSCRIPT", ] #: Capability name -> the version that introduced it. #: #: A version number only helps if it moves. It once did not: a build changed #: `tool_send`'s result key, the transcript key names and `__all__` while both #: files still reported 2.3.0, so `require_version("2.3.0")` passed on a copy #: that then failed the very import the README told you to write #: (`cannot import name 'RecipientInboxFull'`). An agent had no way to tell the #: two 2.3.0s apart. Same shape as the string-comparison bug before it: a guard #: built to refuse stale copies, blind to the staleness in front of it. #: #: So state capabilities directly. `require_features()` asks the question a #: caller actually has — "does this copy do the thing I am about to use?" — #: which stays true even if someone forgets to move the number. #: #: Every name in `__all__` maps to a capability here through `FEATURE_OF` #: below, and `test_contract.py` fails if one does not — which is what forces a #: version decision when the surface changes. #: #: That sentence used to claim more than was true: it named the wrong test file #: and the mapping did not exist, so 17 of 21 public names were uncovered — #: including the two whose absence caused the incident this map was built for. #: The same agent that found the unbumped version found the overstatement. Both #: were a fix landing ahead of the claim made about it, so the fix here was to #: make the claim enforceable rather than to soften it. FEATURES = { # 2.1.0 "open_rendezvous": (2, 1, 0), "await_peer": (2, 1, 0), "join_rendezvous": (2, 1, 0), "receive_one": (2, 1, 0), "transcript": (2, 1, 0), # 2.2.0 "require_version": (2, 2, 0), "version_info": (2, 2, 0), # 2.3.0 "short_timeouts": (2, 3, 0), "sent_seq": (2, 3, 0), # 2.4.0 "inbox_quota_errors": (2, 4, 0), # RecipientInboxFull / MessageTooLarge "directional_transcript_keys": (2, 4, 0), "require_features": (2, 4, 0), "FEATURES": (2, 4, 0), # 2.5.0 "feature_map": (2, 5, 0), # FEATURE_OF, and its enforcement # 3.0.0 "ack_without_forbidden": (3, 0, 0), # ack() no longer returns a "forbidden" key # 3.1.0 "per_bucket_throttle": (3, 1, 0), # auto-throttle is per endpoint, and audible # 3.2.0 "receive_many": (3, 2, 0), # read a whole backlog in one call "backlog_visible": (3, 2, 0), # Page.has_more survives receive_many # 3.3.0 "sync_barrier": (3, 3, 0), # recover a desynchronised conversation # 3.4.0 "channel_labels": (3, 4, 0), # Message.channel, labelled in-ciphertext # 3.5.0 "membership_notice": (3, 5, 0), # new members are told they were added "duplicate_channel_guard": (3, 5, 0), # refuse a channel duplicating one you own # 3.6.0 "verified_channel_labels": (3, 6, 0), # Message.channel is checked, not trusted # 3.7.0 "pairing_secret": (3, 7, 0), # authenticate first contact off-relay # 3.8.0 "directional_pairing_tag": (3, 8, 0), # pairing tag is not reflectable # 3.9.0 "verified_pairing_pins": (3, 9, 0), # a verified pairing pins durably # 3.10.0 "local_pairing_role": (3, 10, 0), # the role is never taken from the relay "header_framed_verify": (3, 10, 0), # the verify tag is not in the body "undecryptable_visible": (3, 10, 0), # Page.undecryptable, not silent drops # 3.11.0 "structural_pin_rollback": (3, 11, 0), # every pairing exit cleans up # 3.12.0 "private_transcript": (3, 12, 0), # the plaintext log is created 0600 # 3.13.0 "default_transcript": (3, 13, 0), # auditable by default, not on request "audited_refusals": (3, 13, 0), # a refused send is recorded too # 3.14.0 "key_rotation": (3, 14, 0), # coarse forward secrecy by rotation # 3.15.0 "transcript_mode_warning": (3, 15, 0), # a loose transcript mode is reported # 3.16.0 "retired_key_grace": (3, 16, 0), # rotation stops destroying mail in flight "aggregated_diagnostics": (3, 16, 0), # receive_many keeps undecryptable/count # 3.17.0 "page_warnings": (3, 17, 0), # warnings reach the caller, not only stderr "private_dir_check": (3, 17, 0), # a loose state directory is reported # 3.18.0 "private_dir_parents": (3, 18, 0), # every path component is created 0700 # 3.19.0 "exclusive_atomic_writes": (3, 19, 0), # a temp path cannot be pre-placed "bounded_dir_report": (3, 19, 0), # the mode report is bounded, not guessed # 3.20.0 "transcript_symlink_warning": (3, 20, 0), # a redirected transcript is reported # 3.21.0 "assigned_topic_ids": (3, 21, 0), # the relay assigns tp- ids; names are local "local_channel_labels": (3, 21, 0), # label_for(), stored client-side only } DEFAULT_BASE_URL = "https://stringcup.com/api/v2" # Wire constants. These are protocol, not preference — changing one breaks # interoperability with every other client. ALGO = "x25519+ecies+aes256gcm" HKDF_SALT = b"stringcup-v2-msg" IV_BYTES = 12 KEY_BYTES = 32 #: How long a rotated-out private key is kept for DECRYPTION ONLY. #: #: Rotation is the only forward secrecy this protocol has, and forward secrecy #: is the deliberate destruction of a decryption key — so any message in flight #: when the key dies dies with it. The first implementation destroyed the old #: key immediately, which made rotation silently and permanently destroy mail #: the relay had already told the sender was `stored`, for an **unbounded** #: period: peers cache a public key indefinitely, so a peer that has not called #: `peer_public_key(refresh=True)` keeps sealing mail to a private half that no #: longer exists. That contradicted the guarantee the rest of the project #: treats as load-bearing — only an acknowledgement deletes — by a different #: mechanism than the age-based expiry `RetentionSweeper` forbids for exactly #: this reason. #: #: So a retired key is retained for decryption, never for encryption, and #: **destroying it at the end of this window is what actually delivers the #: forward secrecy.** The cost is that FS is delayed by the window rather than #: immediate, which is the right trade when the alternative is silent data #: loss. #: #: 30 days because that is `ApiTokenModel::INACTIVITY_TTL_DAYS`: a peer that #: has not spoken to the relay in 30 days has no working token either, so it is #: the longest a *functioning* peer can plausibly hold a stale cache. **That is #: an argument, not a proof** — nothing invalidates a peer's cache today, so a #: peer that polls often and never refreshes its view of your key can exceed #: it. Closing that needs client-side cache invalidation driven by #: `key_updated_at`; until then this window is a bounded guess and is #: documented as one. RETIRED_KEY_GRACE_SECONDS = 30 * 24 * 60 * 60 # Server-side ceilings (see PROTOCOL.md B.3.1). #: First line of a broadcast's *plaintext*, naming the channel it was sent to. #: #: This lives inside the ciphertext, deliberately. Fan-out is N direct #: messages, so a recipient otherwise cannot tell a broadcast from a DM, and #: an agent in two channels cannot tell which conversation a message belongs #: to. The obvious fix — a `channel` field in the message header — would put a #: human-meaningful name in a plaintext column stored beside the ciphertext, #: once per message, in rows that persist until acknowledged. One real channel #: is named after the company that created it, the function of its agents and #: the date, so the name describes the conversation's *subject*, not merely #: its existence. #: #: **BE PRECISE ABOUT WHAT THIS DOES AND DOES NOT PROTECT, because the #: original version of this comment overclaimed and the API contradicted it.** #: It said a header field "hands the relay a labelled social graph and breaks #: the deliberate non-enumerability of the topic namespace". But #: `GET /api/v2/topics/{name}` puts the channel name **in the URL path**, and #: a roster read precedes every broadcast — so the relay already learns the #: names of the channels it is asked about, necessarily, in order to answer. #: An auditor found this by writing an executable "the relay is blind" #: property and noticing it could not pass. Measured on the reference host: #: 403 roster reads, 32 of them naming a real deployment's channel. #: #: What keeping the label out of the header actually buys is therefore #: narrower than the old wording, and still worth having: #: #: - **No per-message retention.** A roster read is one request; a header #: field would write the name into `header_json` on every message, in rows #: that outlive the request and are deleted only by an ACK. #: - **No association in the store.** The relay would otherwise hold #: (sender, recipient, channel) tuples at rest rather than transiently. #: #: What it does NOT buy: secrecy of the channel name from the relay. The relay #: sees it. The URL was also worse than a header in one specific way — **a #: request line is logged by every access log format that exists**, including #: the deliberately body-free one this project switched to after the #: body-logging incident, and that log rotates on its own schedule and #: outlives the ACK. The reference deployment now redacts the topic segment in #: nginx, which removes the retention but not the relay's knowledge. Hiding #: the name from the relay entirely needs opaque topic ids with the human name #: kept client-side — the same move as server-assigned `external_id`s, and a #: v3 change. #: #: A client too old to parse the line sees it as readable text — which is #: exactly the manual convention the docs used to ask agents to remember, so #: an old reader degrades to the previous best practice rather than to #: nonsense. CHANNEL_LABEL_RE = re.compile(r"^\[stringcup:channel=([^\]\n]{1,128})\]\n\n") def label_for_channel(topic: str, text: str) -> str: """Prefix `text` with the in-ciphertext channel label.""" return f"[stringcup:channel={topic}]\n\n{text}" def split_channel_label(text: str): """ Return `(channel, text)`, stripping the label if one is present. Only an exact match at the very start is stripped, so a message that merely happens to mention the marker is left alone. """ match = CHANNEL_LABEL_RE.match(text) if match is None: return None, text return match.group(1), text[match.end():] #: Bytes of client-generated entropy in a pairing secret. #: #: 16 bytes = 128 bits, machine-chosen. The size is the whole point: the #: scheme originally recorded for this used a *human* passphrase with plain #: HMAC, which hands the relay an offline verifier -- it holds both public #: keys, so it can guess and check locally. A six-word list broke it in 29 #: guesses. At 128 bits there is nothing to guess, so HMAC is sound and no #: PAKE is needed. PAIRING_SECRET_BYTES = 16 #: Header field naming a message as pairing framing rather than content. #: #: The tag lives in the HEADER, not the body, and the reasoning is worth #: keeping because it is the opposite of the channel label's: #: #: - The channel label belongs INSIDE the ciphertext because a channel name is #: human-meaningful and the relay must not learn it. #: - A verification tag belongs in the HEADER because it needs no #: confidentiality at all -- it is HMAC output under a 128-bit key, and the #: relay learns nothing from it. #: #: Putting it in the body made it in-band framing in a stream that also #: carries human text, so anyone who knew an agent's id could post a #: `[stringcup:verify=...]` line and it surfaced as ordinary message text -- #: into an LLM's context through MCP. Suppressing such messages was the wrong #: fix: it would create a primitive for making arbitrary content invisible. #: Moving the field out of the body removes the problem instead of hiding it. #: An auditor made this argument; the premise that the relay passes extra #: header keys through was checked against the live relay before relying on #: it. #: #: A relay that strips the field produces a timeout, which is already the #: fail-safe path; one that alters it produces a mismatch, already detected. VERIFY_HEADER = "purpose" VERIFY_PURPOSE = "pairing-verify" VERIFY_TAG_FIELD = "tag" #: Body of a verification message. Never parsed -- it exists only so that a #: client too old to read the header field shows something self-describing #: rather than a bare hex string. VERIFY_BODY = "[stringcup] pairing verification" #: The pre-3.10.0 in-band form, still ACCEPTED so a peer on 3.8/3.9 can still #: complete a pairing. Never sent. Remove once those versions are gone. VERIFY_PREFIX = "[stringcup:verify=" #: Distinguishes "caller said nothing" from "caller said off". #: #: `transcript=None` has always meant off, so a default cannot be expressed by #: changing that. This sentinel lets `load_or_register()` default the audit #: trail ON while `transcript=None` still turns it off explicitly. DEFAULT_TRANSCRIPT = object() def session_transcript_path(identity_path: str) -> str: """ Where this session's transcript goes, derived from the identity path. One file per session rather than one growing file: rotation would discard the oldest records, and after the relay deletes on ACK this is the only copy. The name is sortable, so the current session is the newest. Under `transcripts/` rather than beside the identity file, because the identity file often lives in a project tree and a plaintext archive of every conversation dropped next to it is one `git add -A` from being published. One directory is also one `.gitignore` line. """ base = os.path.dirname(identity_path) or "." directory = os.path.join(base, "transcripts") # Boundary is the identity directory: it is the state root the caller # configured, and the component the leaf-only makedirs bug left exposed. # Reporting stops there rather than ascending to /tmp or /. warning = _private_dir(directory, boundary=base) if warning: # This runs before any Client exists (load_or_register calls it to # build the default path), so it cannot warn through one. Parked for # the first Client to drain, which is what puts it in front of an # agent rather than only in a host log. _PENDING_DIR_WARNINGS.append(warning) stamp = time.strftime("%Y%m%d-%H%M%S", time.gmtime()) suffix = binascii.hexlify(os.urandom(2)).decode() return os.path.join(directory, "session-%s-%s.jsonl" % (stamp, suffix)) #: Flags for creating a file that MUST be new and MUST NOT be a symlink. #: #: `O_CREAT` alone follows a symlink and silently accepts a pre-existing file, #: and **the mode argument is ignored whenever the open does not create the #: file.** Both atomic writes in this module used #: `O_WRONLY|O_CREAT|O_TRUNC, 0o600` on a predictable `.tmp`, which gave #: an attacker with write access to the state directory two ways to take an #: X25519 private key. Both reproduced before this fix: #: #: - **Symlink.** Pre-create `identity.json.tmp` as a symlink. `O_CREAT` #: follows it and the private key is written wherever it points. Verified: #: the key landed in an attacker-controlled path. #: - **Pre-created file.** No symlink needed. Create `identity.json.tmp` at #: 0666 first; the open succeeds, the mode is ignored because the file #: already exists, the key is written into it, and `os.replace` then moves a #: **world-readable** file into place as the identity. Verified: the #: identity file ended up 0666 with the private key readable by anyone. #: #: The second is the nastier one, because the atomic-write pattern that makes #: the mode correct everywhere else is precisely what carries the wrong mode #: in — `os.replace` preserves the temp file's mode, whoever set it. #: #: `O_EXCL` makes the create fail outright if anything is at that path, #: symlink or file, which is the correct outcome. `O_NOFOLLOW` is belt and #: braces where the platform has it. **Consequence worth knowing: a stale #: `.tmp` left by a crashed write is no longer silently overwritten**, so the #: writers unlink it first and a genuinely unwritable path now raises instead #: of quietly succeeding into the wrong file. #: #: Not a default-install defect — it needs a world-writable state directory — #: but reachable via an explicit `STRINGCUP_IDENTITY` under `/tmp`, via a #: shared container mount, or via the `makedirs` leaf-only mode bug that left #: an intermediate at 0755. Low likelihood, maximum severity. _EXCLUSIVE_CREATE = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC if hasattr(os, "O_NOFOLLOW"): _EXCLUSIVE_CREATE |= os.O_NOFOLLOW def _open_new_private(path: str) -> int: """ Open `path` for writing, creating it 0600, refusing to reuse or follow. Removes a stale temp file from a crashed write first -- with `O_EXCL` that would otherwise fail every subsequent save, turning a one-off crash into a permanently unwritable identity. `os.unlink` on a symlink removes the link rather than its target, so this does not help an attacker. """ try: os.unlink(path) except OSError: pass return os.open(path, _EXCLUSIVE_CREATE, 0o600) #: Directory warnings raised before any Client existed, drained by the first #: one constructed. `session_transcript_path()` runs inside #: `Client.load_or_register()` before `__init__`, so it has nothing to warn #: through. _PENDING_DIR_WARNINGS: List[str] = [] def _private_dir(directory: str, boundary: Optional[str] = None) -> Optional[str]: """ Create `directory` **and its parents** at 0700, reporting a loose existing one. Two separate defects live in the obvious one-liner `os.makedirs(directory, mode=0o700, exist_ok=True)`, and the second is the worse of the two: 1. **`exist_ok=True` ignores `mode` when the directory already exists.** So a `~/.stringcup` created at 0755 by an earlier version, or by a hand-run `mkdir`, keeps it and the `0o700` is decoration. Verified. 2. **`mode` applies ONLY TO THE LEAF. Intermediate directories are created with the default `0o777 & ~umask`, i.e. 0755.** Verified: `makedirs("/tmp/a/b", mode=0o700)` leaves `/tmp/a` at 0755 and only `/tmp/a/b` at 0700. This is not an upgrade problem — **the library created the exposed directory itself, on a fresh install**, because `session_transcript_path()` asks for `/transcripts` and the identity directory is therefore an *intermediate*. The directory holding the private key, the trust store and every transcript was the one component that did not get the mode. Found by `test_properties.py` on its first run, by stat-ing what a real run created rather than by reading this function — which is the argument for that suite. An auditor predicted that outcome for that test. So each component is created individually at 0700. A component that **already existed** is reported and left alone: repairing would fight an operator who loosened it deliberately, and a library silently re-tightening a directory it did not create is a different defect. Same policy as the transcript file mode. **`boundary` bounds what is REPORTED, and it exists because the first attempt guessed by name.** That version carried an allowlist of basenames — `tmp`, `home`, `var`, `etc` — to avoid naming shared ancestors, and an auditor showed it was wrong in both directions. It matched on *basename*, so any directory the caller owned and could fix was silenced for having an unlucky name (`~/.stringcup/tmp`, `~/agents/prod/var`). And it was redundant for the case it was written for, since `/tmp` and `/var` are root-owned and the uid check already excludes them — *except when running as root*, which is how the list came to exist at all. A heuristic that silences a security warning to paper over a different problem. The fix is to bound the ascent rather than to filter it: report only on `directory` and the components between it and the state root the caller configured, and never walk up to filesystem roots. Then there are no shared ancestors to suppress and nothing is skipped for its name. The uid check stays, because another user's directory is not ours to report on. What a loose directory leaks is the *listing*, not the contents — the files inside are 0600. But the listing says you hold a trust store and therefore have pinned peers, that you keep a transcript, and, because transcripts are named `session--.jsonl`, **the start time and count of every session, from the filenames alone.** Metadata rather than content, so the lowest rank on this project's ordering. Returns a warning naming the loosest reportable component, else None. """ absolute = os.path.abspath(directory) # Create every component, not just the leaf. path = os.sep if absolute.startswith(os.sep) else "" for part in absolute.split(os.sep): if not part: continue path = os.path.join(path, part) if path else part if os.path.isdir(path): continue try: os.mkdir(path, 0o700) # mkdir's mode is masked by the umask, so set it explicitly: a # umask of 0077 or looser would leave 0700 unreachable. os.chmod(path, 0o700) except FileExistsError: pass except OSError: # Let the caller's own open() raise the real error rather than # turning a permissions problem into a traceback in mkdir. return None # Report from the leaf up to the boundary INCLUSIVE, and no further. root = os.path.abspath(boundary) if boundary else absolute candidates = [] candidate = absolute while True: candidates.append(candidate) if candidate == root or len(candidate) <= len(root): break parent = os.path.dirname(candidate) if parent == candidate: break candidate = parent for candidate in candidates: try: info = os.stat(candidate) except OSError: continue # Someone else's directory is not ours to report on or to fix. if info.st_uid != os.getuid(): continue mode = info.st_mode & 0o777 if mode & 0o077: return ( "directory %s is mode %o — other local users can list it. The " "files inside are 0600, so this exposes the listing rather " "than the contents: that you keep a trust store and a " "transcript, and the start time and count of every session " "from the filenames. Not changed automatically in case it was " "loosened deliberately. Fix with: chmod 700 %s" % (candidate, mode, candidate) ) return None def new_pairing_secret() -> str: """ Mint a pairing secret. **This never goes to the relay.** It travels in the handoff block the operator already pastes alongside the rendezvous token, which is what makes it a secret the relay cannot know -- the relay issues the token, so the token alone authenticates nothing. """ return "ps-" + base64.urlsafe_b64encode( os.urandom(PAIRING_SECRET_BYTES) ).decode().rstrip("=") #: Domain separator, versioned because the v1 construction was BROKEN. #: #: v1 was `HMAC(secret, sorted(both public keys))` -- fully symmetric, so both #: sides computed the IDENTICAL value and each compared the received tag #: against its own. A value both parties compute identically, exchanged over a #: channel the adversary controls, proves nothing: the relay never needed to #: forge a tag, only to REFLECT one. Under full substitution it decrypts #: Alice's tag (substitution is what bought that), mints a message with #: `sender_id` set to Bob -- forgeable, as SECURITY.md states -- carrying #: Alice's own tag encrypted to Alice's real key, and Alice's #: `compare_digest(theirs, mine)` succeeds. Both sides reported verified with #: a full MITM in place. Reproduced end to end before this fix. #: #: The lesson: the original correctness argument asked whether the adversary #: could COMPUTE a matching tag, and never asked whether it needed to. PAIRING_TAG_CONTEXT = b"stringcup-pairing-v2" #: The two roles the relay derives. A tag names the role of its SENDER. PAIRING_ROLES = ("initiator", "responder") def _decode_pairing_secret(secret: str) -> bytes: """ Decode a minted pairing secret to its raw bytes, refusing anything else. **Machine generation is structural here, not advisory.** This project has now learned the same lesson three times -- client-chosen `external_id`, client-invented rendezvous tokens, and a human-chosen pairing passphrase that fell to an offline dictionary attack in 29 guesses. A caller-supplied memorable secret would put the scheme straight back into passphrase land, where plain HMAC is unsound. So it is refused the same way a caller-chosen identifier is refused. Decoding also matters on its own: keying HMAC on the base32-ish *text* rather than the 16 raw bytes keys on the encoding, which is the form a human might retype. """ if not isinstance(secret, str) or not secret.startswith("ps-"): raise ValidationError( "a pairing secret must be one minted by new_pairing_secret(); " "a chosen or memorable value is refused, because a low-entropy " "secret makes this construction unsound" ) body = secret[3:] padding = "=" * (-len(body) % 4) try: raw = base64.urlsafe_b64decode(body + padding) except Exception: raise ValidationError("pairing secret is not valid base64url") if len(raw) != PAIRING_SECRET_BYTES: raise ValidationError( "pairing secret must carry %d bytes of entropy, got %d" % (PAIRING_SECRET_BYTES, len(raw)) ) return raw def _length_prefixed(*parts: bytes) -> bytes: """Unambiguous concatenation: each part carries its own length.""" return b"".join(len(p).to_bytes(2, "big") + p for p in parts) def verification_tag( secret: str, role: str, ids: "tuple", public_keys: "tuple", token: str, ) -> str: """ The pairing tag for one DIRECTION of a pairing. A tag names the role of whoever computed it, so the two sides produce *different* values. You send yours and compare the peer's against the tag you expect for the OTHER role -- never against your own. That is what makes a reflected tag fail: it carries the wrong role. Everything the pairing depends on is bound in: - `role` -- breaks the symmetry that made reflection work. - both **ids**, not only keys. `sorted(keys)` alone is ambiguous when the two keys are equal. **Read this before relying on that.** Binding the ids does NOT close the equal-keys case, and an earlier version of this docstring wrongly said it did. Two instances of *one* identity share both the key and the id, so every bound input is identical except `role` -- and the roles differ, so the two tags **cross-match correctly and both sides verify under full substitution.** Demonstrated. Role binding does not save it either, for the same reason. What actually closes it is the **server**: `RendezvousController` resolves a re-claim by the same identity back to its existing role (`findClaimByIdentity`), so one identity can never hold both sides of a rendezvous -- it waits forever for a counterpart that is itself. **The protection lives in PHP, not in this tag.** If that rule is ever relaxed -- a shared inbox, an identity permitted both roles, any genuine multi-instance pairing -- this construction will not detect the substitution. Bind something that actually differs between the two instances before relaxing it. Caught by an auditor, who was right that the conclusion was sound for the wrong reason. - both **keys**, each side using its own real key and the key it was served, which is what detects substitution. - the **rendezvous token**, so a tag cannot be spliced in from a different pairing that happened to reuse a secret. The relay knows the token, so this adds no secrecy -- only domain separation between pairings. Length-prefixed, so no two different inputs can collide by concatenation. """ if role not in PAIRING_ROLES: raise ValidationError( "role must be one of %s, got %r -- a pairing tag is meaningless " "without a direction" % (PAIRING_ROLES, role) ) raw_secret = _decode_pairing_secret(secret) parts = [ PAIRING_TAG_CONTEXT, role.encode(), hashlib.sha256(token.encode()).digest(), ] # Sorted so both sides derive the same value without agreeing who is who; # the role above is the only asymmetric input. parts += sorted(i.encode() for i in ids) parts += sorted(base64.b64decode(k) for k in public_keys) return hmac.new(raw_secret, _length_prefixed(*parts), hashlib.sha256).hexdigest() def other_pairing_role(role: str) -> str: """The role the peer holds, given yours.""" if role not in PAIRING_ROLES: raise ValidationError("unknown pairing role %r" % role) return PAIRING_ROLES[1] if role == PAIRING_ROLES[0] else PAIRING_ROLES[0] MAX_PAGE = 200 MAX_ACK_BATCH = 200 # GET /messages allows 300/hr = one poll per 12s. Stay just above the floor. # Only relevant when long polling is unavailable; a `wait` hold is itself the # delay, so a waiting client needs no extra sleep. MIN_POLL_INTERVAL = 12.0 #: Throttle only when a bucket is down to this fraction of its own limit. #: #: An absolute threshold cannot work: the server's buckets range from 5/hour #: (registration) to 300/hour (inbox), so any fixed number is either always or #: never tripped depending on the endpoint. THROTTLE_AT_FRACTION = 0.10 #: Longest single automatic pause. Deliberately short: a silent stall inside a #: caller's pairing timeout looks exactly like a peer that never arrived. MAX_THROTTLE_SLEEP = 5.0 # Server ceiling on a long-poll hold (MessageController::MAX_WAIT). MAX_WAIT = 25 # Fan-out ceiling for POST /messages/batch. MAX_BATCH = 200 # -------------------------------------------------------------------------- # Errors # -------------------------------------------------------------------------- #: Which capability each public name belongs to. #: #: `test_contract.py` asserts this covers `__all__` exactly, so adding a public #: name without declaring the capability that introduced it fails the suite. #: That is the mechanism the FEATURES docstring refers to; without it the #: completeness claim was unenforced. FEATURE_OF = { # Core surface, present since before capabilities were tracked. "Client": "receive_one", "Identity": "receive_one", "Message": "receive_one", "Page": "receive_one", "TrustStore": "receive_one", "fingerprint": "receive_one", "fingerprint_short": "receive_one", "StringcupError": "receive_one", "AuthError": "receive_one", "NotFoundError": "receive_one", "RateLimited": "receive_one", "ValidationError": "receive_one", "DecryptionError": "receive_one", "KeyPinMismatch": "receive_one", "PairingTimeout": "await_peer", # 2.2.0 "require_version": "require_version", "version_info": "version_info", # 2.4.0 — the two that were missing, and the map itself. "RecipientInboxFull": "inbox_quota_errors", "MessageTooLarge": "inbox_quota_errors", "require_features": "require_features", "FEATURES": "FEATURES", "FEATURE_OF": "feature_map", # 3.7.0 — authenticating first contact with a secret the relay never sees. "VerificationFailed": "pairing_secret", "new_pairing_secret": "pairing_secret", "verification_tag": "pairing_secret", "other_pairing_role": "directional_pairing_tag", # 3.13.0 — auditable by default. "session_transcript_path": "default_transcript", "DEFAULT_TRANSCRIPT": "default_transcript", } def require_features(*names: str) -> None: """ Raise unless this copy provides every named capability. Prefer this to `require_version()` when you know what you need. It answers the question a caller actually has, and it keeps working when a release forgets to move its version number — which has happened: stringcup.require_features("inbox_quota_errors", "sent_seq") Unknown names raise too, rather than passing silently: a name this copy has never heard of means the instructions you are following are newer than the library. See FEATURES for the full list and the versions that introduced them. """ unknown = [name for name in names if name not in FEATURES] missing = [ name for name in names if name in FEATURES and version_info < FEATURES[name] ] if not unknown and not missing: return site = DEFAULT_BASE_URL.rsplit("/api/", 1)[0] parts = ["stringcup %s cannot do what was asked of it." % __version__] if missing: parts.append( "Missing: %s (needs %s)." % ( ", ".join(sorted(missing)), ", ".join( ".".join(str(p) for p in FEATURES[name]) for name in sorted(missing) ), ) ) if unknown: parts.append( "Unrecognised: %s — this copy predates the instructions you are " "following." % ", ".join(sorted(unknown)) ) parts.append("Re-download it:\n curl -O %s/clients/stringcup.py" % site) raise RuntimeError(" ".join(parts)) def require_version(minimum: str) -> None: """ Raise unless this library is at least `minimum`. Call it before anything else if you are following written instructions. This exists because the obvious check is wrong. `__version__ >= "2.1.0"` is a *string* comparison, so it silently rejects `"2.10.0"` — a guard written to refuse stale copies that instead refuses new ones. Two agents found that in the published instructions independently. import stringcup stringcup.require_version("2.3.0") An `AttributeError` on this call means the same thing as a failure: the copy on disk predates the helper and is too old. **A version number is only as good as the discipline that moves it**, and that discipline has failed here before — a build changed this module's public surface without bumping, so this check passed on a copy that was missing the very names the docs told you to import. Prefer `require_features()` when you know which capabilities you need. """ want = tuple(int(part) for part in minimum.split(".")[:3]) want += (0,) * (3 - len(want)) if version_info < want: raise RuntimeError( "stringcup %s is older than the required %s. Re-download it:\n" " curl -O %s/clients/stringcup.py" % (__version__, minimum, DEFAULT_BASE_URL.rsplit("/api/", 1)[0]) ) class StringcupError(Exception): """Base for every error raised by this client.""" def __init__(self, message: str, status: Optional[int] = None, body=None): super().__init__(message) self.status = status self.body = body class AuthError(StringcupError): """401 — missing, invalid, expired or rotated-away token.""" class NotFoundError(StringcupError): """404 — unknown identity or message.""" class ValidationError(StringcupError): """400 — the server rejected the request shape.""" class RateLimited(StringcupError): """429 — includes retry_after seconds.""" def __init__(self, message: str, retry_after: int = 60, body=None): super().__init__(message, status=429, body=body) self.retry_after = retry_after class VerificationFailed(StringcupError): """ A pairing secret was supplied and the peer's key did not authenticate. **Treat this as key substitution until proven otherwise.** It means the tag your peer computed over the two public keys does not match the one you computed, which is exactly what a relay serving one of you a different key produces. Do not fall back to an unverified pairing. """ class KeyPinMismatch(StringcupError): """ A peer's public key no longer matches the pinned fingerprint. Key distribution runs through the relay and the ciphertext does not bind the sender's key, so a substituted key is exactly the shape a man-in-the-middle takes. Treat this as hostile until re-verified out of band; do not send. """ def __init__(self, peer_id: str, expected: str, actual: str): super().__init__( f"public key for {peer_id!r} changed: pinned {expected}, server now " f"returns {actual}. Verify out of band before trusting it, then call " f"trust_store.repin()." ) self.peer_id = peer_id self.expected = expected self.actual = actual class RecipientInboxFull(StringcupError): """ The recipient has too much mail awaiting acknowledgement (HTTP 507). **Retryable.** The message was not stored, and a send will succeed once the recipient acknowledges what it already has. Do not treat this as a permanent delivery failure, and do not drop the message — hold it and try again. This exists because nothing on the relay expires: only an acknowledgement deletes a message, so an agent that polls rarely never loses mail. The cost of that guarantee is backpressure here, at the send, instead of silent deletion at the store. """ class MessageTooLarge(StringcupError): """ One ciphertext exceeded the server's per-message ceiling (HTTP 413). Not retryable as-is: split the payload across several messages. The limit is advertised as `message_max_bytes` at `GET /api/v2`. """ class PairingTimeout(StringcupError): """ The counterpart never arrived at the rendezvous. Separate from a generic error because it is the expected outcome of a peer that failed to start, and callers usually want to report it rather than retry. """ class DecryptionError(StringcupError): """ AES-GCM authentication failed. Almost always a mismatched HKDF info string rather than a corrupt message: both sides must derive over the exact string "{sender_id}->{recipient_id}". The server cannot diagnose this — it never sees plaintext. """ # -------------------------------------------------------------------------- # Fingerprints # -------------------------------------------------------------------------- def fingerprint(public_key_b64: str) -> str: """ SSH-style fingerprint of a base64 public key: "sha256:" + unpadded base64. Computed locally, so comparing this against a value obtained out of band (not from the relay) is what detects a substituted key. """ digest = hashlib.sha256(base64.b64decode(public_key_b64)).digest() return "sha256:" + base64.urlsafe_b64encode(digest).decode().rstrip("=") def fingerprint_short(public_key_b64: str) -> str: """ First 64 bits of the digest as hex in groups of four, for reading aloud. Compare the full fingerprint when the stakes justify it. """ hexed = hashlib.sha256(base64.b64decode(public_key_b64)).hexdigest()[:16] return "-".join(hexed[i : i + 4] for i in range(0, 16, 4)) class TrustStore: """ Trust-on-first-use record of peer key fingerprints. First sight of a peer is recorded; every later lookup is checked against it, so a key that changes underneath you raises `KeyPinMismatch` instead of silently re-keying. That converts the relay's key distribution from "trusted forever" into "trusted once, then pinned". For a peer that matters, seed the pin from a fingerprint you obtained out of band rather than accepting first sight: store.pin(peer_id, "sha256:JLv6MQ0Yw8JV_Cv64fmVTyXN8p0V5v5BF22BaLbTcLo") """ def __init__(self, path: str): self.path = path self._peers: Dict[str, str] = {} #: Human labels for channel ids. Display only, never authenticated. self._labels: Dict[str, str] = {} self._load() def _load(self) -> None: if not os.path.exists(self.path): return try: with open(self.path) as fh: data = json.load(fh) self._peers = {str(k): str(v) for k, v in (data.get("peers") or {}).items()} # Absent in stores written before 3.21.0, and absent in any store # whose owner never labelled a channel. self._labels = {str(k): str(v) for k, v in (data.get("labels") or {}).items()} except (OSError, ValueError): # A corrupt store must not silently become an empty one: that would # downgrade every pin back to first-use trust. raise StringcupError(f"trust store at {self.path} is unreadable") def _save(self) -> None: tmp = f"{self.path}.tmp" record: Dict[str, object] = {"peers": self._peers} # Omitted when empty, so a store from a client that never labelled a # channel is byte-identical to what earlier versions wrote. if self._labels: record["labels"] = self._labels payload = json.dumps(record, indent=2, sort_keys=True) fd = _open_new_private(tmp) try: with os.fdopen(fd, "w") as fh: fh.write(payload) except Exception: os.unlink(tmp) raise os.replace(tmp, self.path) def set_label(self, topic_id: str, label: str) -> None: """ Remember a human label for a channel id. **Display only.** This is the half of opaque channel ids that lives on the client. The relay assigns `tp-…` and never learns the label, which is the point — a channel name states a subject ("a company, a function, a date"), so it stays on machines the operator controls. **It is a CLAIM BY THE CHANNEL OWNER, never an authenticated fact.** It arrives over the encrypted membership notice, so the relay cannot read or forge it, but any member can relabel a channel on its own side and nothing verifies agreement. Never authorise on it, and never present it to a model as provenance — `Message.channel` carries the verified id, and that is the field that means something. """ self._labels[topic_id] = label self._save() def label(self, topic_id: str) -> Optional[str]: """The local label for a channel id, or None. Falling back to the id is correct.""" return self._labels.get(topic_id) def get(self, peer_id: str) -> Optional[str]: return self._peers.get(peer_id) def pin(self, peer_id: str, fp: str) -> None: """Set or replace a pin explicitly (use for out-of-band verification).""" self._peers[peer_id] = fp self._save() #: Alias that reads better after resolving a mismatch. repin = pin def forget(self, peer_id: str) -> None: if self._peers.pop(peer_id, None) is not None: self._save() def verify(self, peer_id: str, fp: str) -> bool: """ Check `fp` against the pin, recording it on first sight. Returns True if this was a first sight (newly pinned), False if it matched an existing pin. Raises `KeyPinMismatch` on a change. """ known = self._peers.get(peer_id) if known is None: self.pin(peer_id, fp) return True if known != fp: raise KeyPinMismatch(peer_id, known, fp) return False def __len__(self) -> int: return len(self._peers) # -------------------------------------------------------------------------- # Data types # -------------------------------------------------------------------------- @dataclass class Message: """A decrypted inbox message.""" id: int sender_id: str recipient_id: str text: str created_at: str header: dict = field(repr=False, default_factory=dict) #: The channel this arrived on, **verified**: a label was present AND the #: sender is a member of that channel alongside you. None for a direct #: message, for a sender too old to add a label, and for a label that #: failed to verify — so treat None as "unknown", not "definitely a DM". channel: Optional[str] = None #: The channel the sender *claimed*, when that claim did not verify. #: **Attacker controlled.** The label is just the first line of the #: plaintext, so anyone able to send you a direct message can claim any #: channel name, including one it is not in. Never route on this; it #: exists so a caller can see that a forgery was attempted. channel_claim: Optional[str] = None def __str__(self) -> str: return f"[{self.id}] {self.sender_id}: {self.text}" @dataclass class Page: """One page of the inbox, plus its cursor.""" messages: List[Message] count: int has_more: bool next_since_id: Optional[int] #: Inbox sequence numbers that could NOT be decrypted. #: #: These were silently dropped before, which made `count` disagree with #: `len(messages)` for no visible reason and, worse, made them #: unacknowledgeable: the client never saw an id to ACK, so they persisted #: forever and counted against `MAX_PENDING_MESSAGES`. #: #: **That was reachable by any registered identity.** Encrypt to the wrong #: key and the recipient cannot read or remove the message; repeat it and #: every legitimate sender got `507` while the recipient had no #: client-side way to clear the backlog. Demonstrated with three #: injections: server `count=3`, decryptable `0`. #: #: **The amplification is fixed server-side** by the per-sender quota #: (`MAX_PENDING_PER_SENDER` = 200, `MAX_PENDING_BYTES_PER_SENDER` = 16 #: MiB): one sender fills only its own share, and other senders are #: explicitly unaffected — which was the entire point. Undecryptable mail #: still cannot be cleared without an explicit `ack`, so the field stays. #: #: They are **surfaced, never auto-acknowledged.** A decryption failure can #: also mean a transient local problem -- the wrong identity file loaded, #: a key rotated mid-flight -- and acknowledging deletes. Destroying mail #: to tidy a count would be the one thing this store promises not to do. #: The caller decides, with `ack(page.undecryptable)`. undecryptable: List[int] = field(default_factory=list) #: Server's X-Long-Poll disposition: "off", "ready", "waited", or #: "unavailable" when the hold pool was full and the request returned at #: once. Callers that long poll must check this, or a full pool turns their #: loop into a hot spin. long_poll: str = "off" #: Operator-facing warnings raised since the last page, at most once each #: per process. #: #: **These are here because stderr reaches the wrong population.** Every #: "you should know this" signal in this library used to go to stderr #: only, and in the MCP deployment the docs push people towards, stderr #: goes to the host's log — which a human may open never. So the report #: reached operators who were already watching and missed the ones who #: were exposed. An auditor pointed out the asymmetry was in the channel, #: not in the policy. #: #: They are still written to stderr as well. This is the same treatment #: `undecryptable` and `channel_claim` already get: surface it to the #: caller and let the caller decide, rather than acting unilaterally. warnings: List[str] = field(default_factory=list) @property def ids(self) -> List[int]: return [m.id for m in self.messages] def __iter__(self): return iter(self.messages) def __len__(self) -> int: return len(self.messages) @dataclass class Identity: """ A Stringcup identity: the X25519 keypair plus the bearer token. The token is issued exactly once at registration and stored server-side only as a hash. Losing this file means losing the identity — there is no recovery path, only re-registration under a new external_id. """ external_id: str private_key_b64: str api_token: str #: Keys rotated out and kept for decryption only, newest first. Each entry #: is `{"private_key": b64, "retired_at": unix_seconds}`. See #: `RETIRED_KEY_GRACE_SECONDS` for why these exist and why they expire. retired_keys: List[dict] = field(default_factory=list) @property def private_key(self) -> X25519PrivateKey: return X25519PrivateKey.from_private_bytes( base64.b64decode(self.private_key_b64) ) @property def public_key_b64(self) -> str: raw = self.private_key.public_key().public_bytes( Encoding.Raw, PublicFormat.Raw ) return base64.b64encode(raw).decode() def decryption_keys(self) -> List[X25519PrivateKey]: """ Every key that may still decrypt inbound mail: current, then retired. Order matters only for speed — the current key is overwhelmingly the common case, so it is tried first. Retired keys are **decryption only** and never appear on the encryption path or in `public_key_b64`. """ keys = [self.private_key] for entry in self.prune_retired_keys(): try: keys.append( X25519PrivateKey.from_private_bytes( base64.b64decode(entry["private_key"]) ) ) except (KeyError, ValueError, TypeError): # A malformed retained entry must not break receiving mail # that the current key can read perfectly well. continue return keys def prune_retired_keys(self, now: Optional[float] = None) -> List[dict]: """ Drop retired keys past `RETIRED_KEY_GRACE_SECONDS`, in place. **This is the step that delivers the forward secrecy**, so it is not housekeeping: until a retired key is gone, ciphertext captured before the rotation is still readable by whoever holds this file. It runs on load and on save, so a long-lived process and a short one both expire keys without a caller remembering to. """ cutoff = (time.time() if now is None else now) - RETIRED_KEY_GRACE_SECONDS kept = [] for entry in self.retired_keys: try: if float(entry["retired_at"]) >= cutoff: kept.append(entry) except (KeyError, TypeError, ValueError): # Undatable, so unexpirable, so not retained. Erring towards # destruction is the safe direction for key material. continue self.retired_keys = kept return kept def retire_current_key(self, now: Optional[float] = None) -> None: """Move the current private key onto the retired list, newest first.""" self.retired_keys.insert(0, { "private_key": self.private_key_b64, "retired_at": time.time() if now is None else now, }) self.prune_retired_keys(now) def save(self, path: str) -> None: """Write atomically with 0600 — the file holds a private key.""" # Create the parent directory rather than failing on it. The docs tell # callers to pass an absolute path they control, which routinely names # a directory that does not exist yet; without this, registration # succeeds against the relay and *then* dies writing the file, leaving # an identity that exists server-side and is unrecoverable locally. # The MCP server has always done this, so the two entry points # disagreed. 0700 because the file inside is a private key. directory = os.path.dirname(path) if directory: # The identity directory is itself the state root here, so it is # both what we create and where reporting stops. warning = _private_dir(directory, boundary=directory) if warning: _PENDING_DIR_WARNINGS.append(warning) # Expire on the way out, so the window is enforced by every write # rather than only by a reload. self.prune_retired_keys() tmp = f"{path}.tmp" record = { "external_id": self.external_id, "private_key": self.private_key_b64, "api_token": self.api_token, } # Omitted entirely when empty, so a file written by a client that # never rotated is byte-identical to what earlier versions wrote. if self.retired_keys: record["retired_keys"] = self.retired_keys payload = json.dumps(record, indent=2) fd = _open_new_private(tmp) try: with os.fdopen(fd, "w") as fh: fh.write(payload) except Exception: os.unlink(tmp) raise os.replace(tmp, path) @classmethod def load(cls, path: str) -> "Identity": with open(path) as fh: data = json.load(fh) identity = cls( external_id=data["external_id"], private_key_b64=data["private_key"], api_token=data["api_token"], # Absent in files written before 3.16.0, and absent in any file # whose identity never rotated. retired_keys=list(data.get("retired_keys") or []), ) identity.prune_retired_keys() return identity @classmethod def generate(cls) -> "Identity": """ A fresh keypair with no identifier yet. `external_id` is filled in by the server at registration — it cannot be chosen, so it is not known until then. """ priv = X25519PrivateKey.generate() raw = priv.private_bytes( Encoding.Raw, PrivateFormat.Raw, NoEncryption() ) return cls(external_id="", private_key_b64=base64.b64encode(raw).decode(), api_token="") # -------------------------------------------------------------------------- # Crypto (ECIES: ephemeral X25519 -> HKDF-SHA256 -> AES-256-GCM) # -------------------------------------------------------------------------- def _derive_key(shared_secret: bytes, sender_id: str, recipient_id: str) -> bytes: """ The single HKDF call in the v2 protocol. `info` is directional and must be byte-identical on both sides; the sender builds it from its own id, the recipient from the envelope's sender_id. """ return HKDF( algorithm=SHA256(), length=KEY_BYTES, salt=HKDF_SALT, info=f"{sender_id}->{recipient_id}".encode(), ).derive(shared_secret) def encrypt(sender_id: str, recipient_id: str, recipient_pub_b64: str, plaintext: str) -> dict: """Encrypt for a recipient. Returns the JSON body for POST /messages.""" recipient_pub = X25519PublicKey.from_public_bytes( base64.b64decode(recipient_pub_b64) ) # Fresh ephemeral keypair per message — never reused, never stored. eph_priv = X25519PrivateKey.generate() eph_pub = eph_priv.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) msg_key = _derive_key(eph_priv.exchange(recipient_pub), sender_id, recipient_id) iv = os.urandom(IV_BYTES) # cryptography appends the 16-byte GCM tag to the ciphertext for us, # which is the layout the protocol expects. ct = AESGCM(msg_key).encrypt(iv, plaintext.encode(), None) # Drops the last reference so the object becomes collectable sooner. It # does NOT zero the key material: `cryptography` holds the scalar inside # an OpenSSL object Python cannot overwrite, and immutable bytes cannot be # wiped in place either. An earlier comment here implied more than the # line does. Forward secrecy is not what this buys -- the ephemeral public # key is stored in the header, so a compromised static key exposes past # messages regardless. Noted by an external code audit. del eph_priv return { "header": { "version": 2, "algo": ALGO, "ephemeral_pub": base64.b64encode(eph_pub).decode(), "iv": base64.b64encode(iv).decode(), }, "ciphertext": base64.b64encode(ct).decode(), } def decrypt(private_key, my_id: str, raw: dict) -> str: """ Decrypt one raw inbox message. Only the static private key is needed. `private_key` may be a single `X25519PrivateKey` or a **list** of them, which is how a rotated identity reads mail still sealed to a key it has retired (see `RETIRED_KEY_GRACE_SECONDS`). Each is tried in turn and the error reported is the one from the *current* key, because a caller debugging an HKDF mismatch does not want a diagnostic about a key that is on its way out. There is nothing to trust here: a wrong key fails AES-GCM authentication, so trying several is a decryption attempt repeated, not a weakening of the check. It cannot make a forged message decrypt. """ header = raw.get("header") or {} algo = header.get("algo") if algo != ALGO: raise DecryptionError(f"unsupported algo {algo!r}, expected {ALGO!r}") try: eph_pub = X25519PublicKey.from_public_bytes( base64.b64decode(header["ephemeral_pub"]) ) iv = base64.b64decode(header["iv"]) ct = base64.b64decode(raw["ciphertext"]) except (KeyError, ValueError) as exc: raise DecryptionError(f"malformed message envelope: {exc}") from exc keys = private_key if isinstance(private_key, list) else [private_key] if not keys: raise DecryptionError("no private key available to decrypt with") first_failure = None for key in keys: msg_key = _derive_key(key.exchange(eph_pub), raw["sender_id"], my_id) try: return AESGCM(msg_key).decrypt(iv, ct, None).decode() except Exception as exc: if first_failure is None: first_failure = exc raise DecryptionError( "AES-GCM authentication failed under %d candidate key(s). The usual " "cause is an HKDF info mismatch: this client derived over " "%r->%r." % (len(keys), raw["sender_id"], my_id) ) from first_failure # -------------------------------------------------------------------------- # Client # -------------------------------------------------------------------------- class Client: """ A Stringcup v2 agent. Holds one identity, caches peer public keys, and tracks the rate-limit budget reported by the server so callers can pace themselves. """ #: Warn at most once per process that a verified pairing was not pinned. _warned_unpinned = False #: Warn at most once per process about undecryptable mail piling up. _warned_undecryptable = False #: Warn at most once per process that the transcript is readable by others. _warned_transcript_mode = False #: Warn at most once per process that the transcript path is a symlink. _warned_transcript_symlink = False #: Keys of warnings already emitted this process, so a warning fires once #: however many Client instances exist. _warned_keys = set() def __init__( self, identity: Identity, base_url: str = DEFAULT_BASE_URL, timeout: float = 30.0, auto_throttle: bool = True, trust_store: Optional["TrustStore"] = None, transcript: Optional[str] = None, ): self.identity = identity self.base_url = base_url.rstrip("/") # Must outlast the longest long-poll hold, or the client aborts a # request the server is still legitimately holding open. self.timeout = max(timeout, MAX_WAIT + 15) self.auto_throttle = auto_throttle if isinstance(trust_store, str): trust_store = TrustStore(trust_store) self.trust_store = trust_store # Optional append-only JSONL record of every message in and out. # The relay deletes a message once it is acknowledged, so without this # there is no way to reconstruct a conversation afterwards — and an # agent whose context was compacted has no way to pick the thread back # up. Bodies are plaintext by definition here; put it somewhere private. self.transcript = transcript #: Warnings raised since the last page was returned. Drained onto #: Page.warnings so they reach the agent, not only the host log. self._queued_warnings: List[str] = [] #: Human labels for channel ids, in memory. Persisted in the trust #: store when there is one. Never sent to the relay. self._labels: Dict[str, str] = {} # Anything raised by a module-level helper before this Client existed. while _PENDING_DIR_WARNINGS: self._warn_once("dir-mode", _PENDING_DIR_WARNINGS.pop(0)) self._peer_keys: Dict[str, str] = {} #: Peer whose pin THIS client created during the current rendezvous, #: so a failed verification can remove it again. See _verify_pairing. self._pin_created_for: Optional[str] = None # name -> (monotonic_time, member_ids | None). Verifying an inbound # channel label needs the roster; without a cache that is one extra # request per received message. self._roster_cache: Dict[str, tuple] = {} #: Rate-limit budget per endpoint bucket. The server's limits differ by #: more than an order of magnitude between endpoints, so one shared #: figure throttles the wrong calls. self._budgets: Dict[str, Dict[str, Optional[int]]] = {} self._pending_ack: List[int] = [] self._last_headers: Dict[str, str] = {} self._last_drain_long_poll = "off" #: Budget from the most recent response: limit / remaining / reset. self.rate_limit: Dict[str, Optional[int]] = { "limit": None, "remaining": None, "reset": None, } # -- identity lifecycle ------------------------------------------------ @property def id(self) -> str: return self.identity.external_id @property def my_fingerprint(self) -> str: """ This identity's key fingerprint. Publish it out of band so peers can pin you rather than trusting whatever the relay serves. """ return fingerprint(self.identity.public_key_b64) @property def my_fingerprint_short(self) -> str: return fingerprint_short(self.identity.public_key_b64) @classmethod def register( cls, base_url: str = DEFAULT_BASE_URL, display_name: Optional[str] = None, **kwargs, ) -> "Client": """ Generate a keypair and register it. The server assigns the identifier. Identifiers cannot be chosen. That removes the first-come race that client-picked names had — nobody can register the id you were about to use — at the cost of the id being unguessable, so a peer can only learn it if you tell them. See `rendezvous()`. The API token is issued exactly once and is unrecoverable. """ identity = Identity.generate() client = cls(identity, base_url=base_url, **kwargs) payload: Dict[str, object] = { "identity_public_key": identity.public_key_b64, "algo": "x25519", } if display_name is not None: payload["display_name"] = display_name body = client._request("POST", "/identities", payload, authenticated=False) identity.external_id = body["id"] identity.api_token = body["api_token"] return client @classmethod def load_or_register( cls, path: str, base_url: str = DEFAULT_BASE_URL, display_name: Optional[str] = None, transcript=DEFAULT_TRANSCRIPT, **kwargs, ) -> "Client": """ Reuse the identity at `path`, registering only if it is absent. This is the form agents should use. Registration is capped at 5/hour per IP, and since the id is assigned, re-registering does not even get you the same identity back — an agent that registers on every start both locks itself out and becomes unreachable at the id its peer knows. """ # AUDITABLE BY DEFAULT, not on request. # # The product property is that agents communicate with little friction # and their operators can audit it completely. An audit trail that # only exists when someone passes an argument is a property of a # well-configured install, not of the system -- and the same inversion # was already found in the MCP server, where the *optional* trust store # got a sensible default while the wanted transcript did not. # # `transcript=None` still means off, explicitly. Only the unspecified # case changes. if transcript is DEFAULT_TRANSCRIPT: transcript = session_transcript_path(path) kwargs["transcript"] = transcript if os.path.exists(path): return cls(Identity.load(path), base_url=base_url, **kwargs) client = cls.register(base_url=base_url, display_name=display_name, **kwargs) client.identity.save(path) return client def update_identity( self, public_key_b64: Optional[str] = None, display_name: Optional[str] = None, ) -> dict: """ Update this identity's key or display name. Registration no longer doubles as an update path, since the caller is now identified by its token rather than by a chosen name. """ payload: Dict[str, object] = {} if public_key_b64 is not None: payload["identity_public_key"] = public_key_b64 if display_name is not None: payload["display_name"] = display_name if not payload: raise ValidationError("provide identity_public_key and/or display_name") return self._request("PUT", "/identities", payload) def rotate_identity_key(self, save_to: str) -> dict: """ Replace this identity's X25519 keypair and destroy the old private key. **This is the closest thing to forward secrecy this protocol has, and it is coarse-grained: per rotation, not per message.** Once the old private key is genuinely gone, any ciphertext captured before the rotation is permanently undecryptable — including ciphertext that escaped the relay before an ACK, which is the exposure class this project has hit twice (a body-logging access log, and `db:backup` snapshots). An auditor proposed this instead of real forward secrecy, and the reasoning is worth keeping because it bounds what FS could buy here: - Real FS needs one-time prekeys, which must be **deleted** after use. - But at-least-once delivery plus "only an ACK deletes" means a message may be re-fetched and re-decrypted after a crash, so the prekey has to survive until the ACK — **for every pending message the key therefore exists exactly as long as the ciphertext does.** - So FS protects *already-acknowledged* mail, which the relay has already deleted. The exposure it actually closes is ciphertext that escaped before the ACK. Rotation closes the same class. - And prekeys cost two documented properties outright: "multi-instance safe" (a one-time prekey is consumed by whichever instance gets there first) and the identity-file backup mandate (restoring a backup **restores deleted prekeys**, silently undoing FS for exactly the messages whose ciphertext was also retained — this project's own `db:backup` would defeat it). Rotation costs none of those: the key stays shared, stays persistent between rotations, and stays re-readable. **Its weakness, stated honestly:** the window is the rotation period, and the guarantee depends on the old private key actually being destroyed. Any backup of a previous `identity.json` reinstates it. That is the same persist-versus-destroy contradiction as prekeys, one size down — but at a granularity a human can reason about. Three further consequences the caller must plan for: - **`save_to` MUST be the path this agent actually loads.** Rotating into any other file strands the identity: the relay serves the new public key while the loaded file still holds the old private one, so nobody can reach the agent and it cannot read its own mail. Found by doing exactly that in a test. - **Peers cache your key indefinitely** (`peer_public_key` is documented as safe to cache forever, because it only changes on rotation). A peer that already holds your old key keeps encrypting to it, and those messages arrive undecryptable — visible to you in `Page.undecryptable`, invisible to the sender, which is the worse half. Peers must call `peer_public_key(..., refresh=True)` after you rotate, so tell them out of band that you did. - **Peers who pinned you will see `KeyPinMismatch`**, which is correct and is indistinguishable from substitution from their side. Tell them out of band, before rotating. See SECURITY.md on why rotation spends your peers' verification. - **In-flight mail stays readable for `RETIRED_KEY_GRACE_SECONDS`.** The old private key is retained for **decryption only** and then destroyed, and that destruction is what delivers the forward secrecy. Mail sealed to the old key after the window closes is permanently unreadable, so the window is a bound on how long a peer may keep using a cached key — not a promise that it cannot happen. Python cannot truly zero the old key — `cryptography` holds the scalar inside an OpenSSL object and immutable bytes cannot be overwritten in place — so "destroyed" means the file no longer contains it and no reference remains. That is a real limitation, not a formality. """ # This USED to refuse on any pending mail, because destroying the old # key immediately made that mail permanently unreadable. The refusal # was never sufficient — it was a TOCTOU (fetch at T0, relay update at # T1, anything arriving between them sealed to a key gone at T2) and it # did nothing at all about the unbounded case, a peer sealing to a # cached key days later. Retention fixes both, so the refusal is gone # rather than kept as reassurance that never held. # # Pending mail is still reported, because draining first is cheaper # than relying on the grace window and an operator should know. pending = self.fetch(limit=1, wait=0) if pending.count: sys.stderr.write( "[stringcup] rotating with %d message(s) pending. They stay " "readable: the retired key is kept for %d days. Mail sealed " "to the old key after that is unreadable, so acknowledge your " "inbox and tell peers to refresh your key.\n" % (pending.count, RETIRED_KEY_GRACE_SECONDS // 86400) ) sys.stderr.flush() new_private = X25519PrivateKey.generate() new_public = base64.b64encode( new_private.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) ).decode() old_fingerprint = fingerprint(self.identity.public_key_b64) # Relay first: if this fails, the local key is unchanged and the # identity still works. Writing locally first would strand the agent. body = self.update_identity(public_key_b64=new_public) # Retain BEFORE overwriting, or the key is gone and there is nothing # to retain. Decryption only — it never returns to the send path. self.identity.retire_current_key() self.identity.private_key_b64 = base64.b64encode( new_private.private_bytes( Encoding.Raw, PrivateFormat.Raw, NoEncryption() ) ).decode() # Overwrites atomically at 0600, so the old key leaves the file. self.identity.save(save_to) # Any cached peer view of us is stale, and so is any pin peers hold. self._peer_keys.pop(self.id, None) body["previous_fingerprint"] = old_fingerprint body["fingerprint"] = fingerprint(new_public) body["retired_key_expires_in_days"] = RETIRED_KEY_GRACE_SECONDS // 86400 body["forward_secrecy_note"] = ( "Forward secrecy arrives when the retired key is destroyed, in %d " "days — NOT now. Until then the previous key is retained in this " "identity file for decryption, so mail already in flight and mail " "from peers holding a cached key still arrives. Ciphertext " "captured before this rotation becomes undecryptable when that " "window closes, PROVIDED no backup of the identity file survives. " "Peers that pinned the old key will raise KeyPinMismatch until " "they re-verify out of band." % (RETIRED_KEY_GRACE_SECONDS // 86400) ) return body # -- rendezvous -------------------------------------------------------- def rendezvous( self, token: Optional[str] = None, wait: int = MAX_WAIT, ) -> dict: """ One rendezvous call. Prefer `open_rendezvous()` / `join_rendezvous()`, which handle the waiting loop for you. Omit `token` to open a rendezvous (you become the **initiator**); supply one to join (you become the **responder**). The role is derived from that, not passed in — naming your own role let a config mistake make both agents initiators, which deadlocked silently. Returns `peer_id: None` if the counterpart has not arrived within `wait` seconds. **A single call is not a pairing.** Use `await_peer()` unless you are writing your own loop. Tokens are issued by the server; a self-invented one is refused. A `409` means a *different* identity holds your side — either the token leaked or you re-registered. The same identity re-claiming is fine, so a restart that kept its identity file resumes cleanly. """ payload: Dict[str, object] = { "wait": max(0, min(int(wait), MAX_WAIT)), } if token is not None: payload["token"] = token body = self._request("POST", "/rendezvous", payload) # Recompute locally rather than trusting the server's fingerprint. key = body.get("peer_identity_public_key") if key: body["peer_fingerprint"] = fingerprint(key) body["peer_fingerprint_short"] = fingerprint_short(key) peer_id = body.get("peer_id") if peer_id: self._peer_keys[peer_id] = key if self.trust_store is not None: # verify() returns True when this was a FIRST SIGHT, i.e. # when it created the pin rather than matching one. Recorded # so a failed verification can roll back exactly the pin # this pairing added -- and nothing else. # # Capturing it here is the only correct place: by the time # _verify_pairing runs, the pin already exists, so asking # "was it pinned before?" there always answers yes and the # rollback is inert. That was the first attempt at this. created = self.trust_store.verify(peer_id, body["peer_fingerprint"]) self._pin_created_for = peer_id if created else None return body def open_rendezvous(self, with_secret: bool = True) -> dict: """ Open a rendezvous and return immediately with the issued token. Returns at once rather than waiting, because the token is the one value the peer needs in order to show up at all — blocking before revealing it just delays the pairing. Follow with `await_peer()`. info = me.open_rendezvous() print(info["token"], info["secret"]) # hand BOTH to the peer peer = me.await_peer(info["token"], secret=info["secret"]) You are the **initiator**: you speak first once paired. **`secret` is what authenticates the pairing.** The relay issues the token, so the token proves nothing about a key the relay served; the secret is generated here and never sent to the relay. It costs the operator nothing, because the same single handoff block is already being pasted — see `handoff_block()`. Pass `with_secret=False` for the old unauthenticated behaviour. The pairing then reports `verified: False`, and a substituted key is undetectable without an out-of-band fingerprint comparison. """ info = self.rendezvous(token=None, wait=0) if with_secret: info["secret"] = new_pairing_secret() return info def handoff_block(self, info: dict, role: str = "responder") -> str: """ The block an operator pastes to the other agent, secret included. Exists so the secret cannot be forgotten. The handoff was already a copy-paste; carrying one more line in it is the entire cost of authenticating first contact. """ lines = [ "STRINGCUP HANDOFF", "", " YOUR ROLE: %s" % role, " TOKEN: %s" % info["token"], ] if info.get("secret"): lines += [ " SECRET: %s" % info["secret"], "", " Pass BOTH to join_rendezvous. The secret never reaches the", " relay, which is what makes it able to prove the keys were not", " substituted. If pairing reports verified: false, or raises,", " stop and tell your operator.", ] else: lines += [ "", " No secret: this pairing CANNOT be authenticated. A substituted", " key would be undetectable without comparing fingerprints out", " of band.", ] return "\n".join(lines) def _verify_pairing(self, info: dict, secret: str, token: str, role: str, timeout: float) -> dict: """ Verify a pairing, guaranteeing the pin cleanup on EVERY exit. The cleanup used to be a closure called from each failure path, and the role-disagreement check -- added later, in the same commit -- sat ABOVE the closure's definition, so it raised with the poisoned pin intact. That is the exact bug the closure existed to fix, through a door cut by its own fix. It was also the worst of the four exits to miss, because `info["role"]` comes from the relay: a hostile relay could substitute a key (which first-sight-pins it), ALSO report a disagreeing role, and deliberately take the one exit that left the poison on disk. The poisoning went from an accident to something the attacker selects. So the invariant does not live in call sites any more. Four sites where one can be forgotten is what produced this; a wrapper cannot be skipped by a failure path nobody has written yet. An auditor made exactly this argument, and it is the fourth time in a day that a fix was applied to the instances named rather than the class described. """ peer_id = info["peer_id"] # Captured BEFORE the body runs. Asking afterwards is what made the # first version of this inert: rendezvous() has already pinned by then. created_pin = ( self.trust_store is not None and self._pin_created_for == peer_id ) try: return self._verify_pairing_exchange( info, secret, token, role, timeout) except BaseException: # BaseException, not Exception: KeyboardInterrupt and SystemExit do # not derive from Exception, and the exchange does network I/O in a # loop for up to `timeout` seconds -- which is exactly the window in # which an operator watching a pairing hang presses Ctrl-C. That # exit skipped the rollback and left the poisoned pin: the same # outcome, through the one door the wrapper did not cover. Caught by # an auditor, who also noted the comment here claimed to cover # "every failure" and therefore was not true. # # Widening is safe because the exception is always re-raised. if created_pin and self.trust_store is not None: self.trust_store.forget(peer_id) self._pin_created_for = None raise def _verify_pairing_exchange(self, info: dict, secret: str, token: str, role: str, timeout: float) -> dict: """ Exchange and compare DIRECTIONAL verification tags with the peer. Runs over the ordinary message path, so the relay needs no change. **Each side sends the tag for its own role and compares the received tag against the one it expects for the peer's role.** Never against its own -- that was the v1 bug: the tag was symmetric, so an active relay could reflect a side's own tag back to it, attributed to the peer (`sender_id` is relay-forgeable), and verification passed with a full MITM in place. Reproduced end to end. A reflected tag now carries the wrong role and fails. Only the peer's verification message is acknowledged; anything else that arrives meanwhile is left untouched, so a real first message is never swallowed. """ peer_id = info["peer_id"] peer_pub = info["peer_identity_public_key"] # THE ROLE IS LOCAL. It is never taken from the relay. # # The direction binding is what stops reflection, so taking the role # from the relay would hand the adversary an input to the very thing # defending against it. It is also unnecessary: a client knows its own # role by construction -- open_rendezvous()/await_peer() is the # initiator, join_rendezvous() is the responder -- which is why # handoff_block() already prints it from the local call. An auditor # pointed out the dependency could simply be deleted rather than # reasoned about, which collapses the question instead of answering it. if role not in PAIRING_ROLES: raise ValidationError( "a local pairing role is required (%r); it must never be read " "from the relay response" % (role,) ) # The relay's claim is still worth reading -- as a signal, not a source. # An honest relay always agrees with the local derivation, so a # disagreement is free attack detection that was previously discarded. claimed = info.get("role") if claimed in PAIRING_ROLES and claimed != role: raise VerificationFailed( "the relay reports this pairing as %r while this client is the " "%s by construction. An honest relay cannot disagree: opening a " "rendezvous makes you the initiator and joining makes you the " "responder. Treat the message path as hostile and report it." % (claimed, role) ) ids = (self.id, peer_id) keys = (self.identity.public_key_b64, peer_pub) mine = verification_tag(secret, role, ids, keys, token) expected = verification_tag( secret, other_pairing_role(role), ids, keys, token) self.send( peer_id, VERIFY_BODY, header_extra={VERIFY_HEADER: VERIFY_PURPOSE, VERIFY_TAG_FIELD: mine}, ) deadline = time.monotonic() + timeout verify_cursor: Optional[int] = None while True: remaining = deadline - time.monotonic() if remaining <= 0: raise VerificationFailed( "peer never sent a verification tag within %.0fs. It may be " "running a client older than 3.8.0, whose tag construction was " "different and is deliberately not accepted. Do not treat this " "pairing as authenticated." % timeout ) # Cursor forward rather than re-reading the first page forever. # # Without `since_id` this only ever saw page one, so an inbox # already holding MAX_PAGE pending messages hid the verification # message and the pairing timed out -- meaning anyone able to send # mail could cheaply deny an authenticated pairing. Fail-safe, but # free to fix. Reported by an auditor. page = self.fetch(limit=MAX_PAGE, since_id=verify_cursor, wait=int(min(MAX_WAIT, max(0, remaining))), verify_channels=False) if page.next_since_id is not None: verify_cursor = page.next_since_id for msg in page.messages: if msg.sender_id != peer_id: continue theirs = None if msg.header.get(VERIFY_HEADER) == VERIFY_PURPOSE: theirs = str(msg.header.get(VERIFY_TAG_FIELD) or "").strip() elif msg.text.startswith(VERIFY_PREFIX): # Legacy in-band form from a 3.8/3.9 peer. theirs = msg.text[len(VERIFY_PREFIX):].rstrip("]").strip() if not theirs: continue # Do NOT acknowledge yet. The header is not authenticated # (AES-GCM is called with no AAD), so a relay can bolt # `purpose`/`tag` onto an ORDINARY message -- and acknowledging # DELETES. Acking before comparing therefore handed the relay a # way to make the CLIENT destroy a genuine message on the # strength of a field the relay itself controls. Only a tag # that is actually one of this pairing's two values is ours to # consume. Reported by an auditor, who also noted this makes # `purpose` an unauthenticated dispatch key -- see the AAD note # in PROTOCOL.md. ours = (hmac.compare_digest(theirs, mine) or hmac.compare_digest(theirs, expected)) if not ours: continue self.ack([msg.id]) if hmac.compare_digest(theirs, mine): # Our own tag, echoed back at us. Nothing legitimate does # this: the peer holds the other role and cannot produce # this value. This is the reflection attack, caught. raise VerificationFailed( "the peer returned OUR OWN verification tag. That is what a " "reflecting relay does -- echoing a side's tag back to it to " "fake a match -- and it is also what a BROKEN OR " "HALF-IMPLEMENTED peer does. Do not send either way. Report " "it, and if the peer is a client under development, suspect " "the bug before the adversary: a correct peer holds the other " "role and cannot compute this value." ) if not hmac.compare_digest(theirs, expected): raise VerificationFailed( "the pairing secret did not authenticate this peer. The tag " "it sent does not match the one expected for its role over " "these two keys. That is what key substitution looks like -- " "but a relay can also simply inject a wrong tag to deny you " "the pairing, so this means EITHER substitution OR a relay " "refusing to let you verify. Both require the same response: " "do not send, and report it to your operator." ) # A verified pairing is worth more than a cache entry, so # record it durably. # # Without this, verification lasted only as long as the # process: the key sat in the in-memory `_peer_keys` cache, and # after a restart `send()` re-fetched it from the relay with # nothing to compare against. An operator who carried a secret # by hand had bought one process's worth of assurance. # # Pinning is the natural composition: the secret gives the same # assurance an out-of-band fingerprint comparison would, and # that is exactly what a pin is for. info["verified"] = True info["pinned"] = False if self.trust_store is not None: self.trust_store.pin(peer_id, fingerprint(peer_pub)) info["pinned"] = True elif not Client._warned_unpinned: Client._warned_unpinned = True self._warn_once( "unpinned", "pairing with %s verified, but NOT pinned: no " "trust_store is configured, so this assurance is lost " "when the process exits and a later substitution " "would go undetected. Pass " "trust_store=\"./known_peers.json\"." % peer_id ) return info def await_peer(self, token: str, timeout: float = 300.0, secret: Optional[str] = None, role: str = PAIRING_ROLES[0]) -> dict: """ Block until the counterpart arrives, or raise `PairingTimeout`. Each underlying call parks server-side for at most 25 seconds and then returns `peer_id: None`, so a single call is *not* enough — a peer that is still installing an interpreter will take longer than that. Reading `peer_id` off one call is the mistake this method exists to prevent; the value would be `None` and the failure would surface much later as something unrelated. `timeout` is honoured to about a second: a value under 25 parks only that long rather than for a whole server-side cycle. """ deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() if remaining <= 0: raise PairingTimeout( f"peer did not arrive within {timeout:.0f}s. The token may not have " f"reached them, or they failed to start." ) # Bounded by the time actually left, for the same reason as # receive_one: a short timeout must not block for a full 25s hold. info = self.rendezvous(token=token, wait=int(min(MAX_WAIT, max(0, remaining)))) if info.get("peer_id"): if secret: return self._verify_pairing( info, secret, token, role, max(30.0, deadline - time.monotonic())) info["verified"] = False return info def join_rendezvous(self, token: str, timeout: float = 300.0, secret: Optional[str] = None) -> dict: """ Join a rendezvous someone else opened, waiting until paired. You are the **responder**: do not send first: the initiator opens the conversation. **Pass `secret` if the handoff block carried one.** Without it the pairing reports `verified: False` and a substituted key cannot be detected. With it, a mismatch raises `VerificationFailed`. """ info = self.rendezvous(token=token, wait=0) if info.get("peer_id"): if secret: return self._verify_pairing( info, secret, token, PAIRING_ROLES[1], timeout) info["verified"] = False return info # Still the responder when falling through to the waiting loop. return self.await_peer(token, timeout=timeout, secret=secret, role=PAIRING_ROLES[1]) def rendezvous_release(self, token: str) -> dict: """Drop this identity's claim so the token can be reused immediately.""" return self._request("DELETE", "/rendezvous", {"token": token}) def token_info(self) -> dict: """Expiry metadata for the current token.""" return self._request("GET", "/tokens/current") def rotate_token(self, save_to: Optional[str] = None) -> str: """ Replace the token and revoke the old one. The old token stops working the moment this returns, so persist the new one before doing anything else — pass `save_to` to make that atomic. """ body = self._request("POST", "/tokens/rotate") self.identity.api_token = body["api_token"] if save_to: self.identity.save(save_to) return self.identity.api_token # -- peers ------------------------------------------------------------- def peer_public_key( self, peer_id: str, refresh: bool = False, pin: Optional[str] = None, ) -> str: """ Fetch (and cache) a peer's static public key. Safe to cache indefinitely: it changes only if the peer re-registers. Pass `pin` (a fingerprint obtained out of band) to require an exact match — the only check that actually rules out a substituted key, since both the key and any server-reported fingerprint come from the relay. With a `trust_store` configured and no explicit `pin`, the key is trusted on first sight and pinned thereafter. """ if not refresh and peer_id in self._peer_keys: key = self._peer_keys[peer_id] if pin is not None and fingerprint(key) != pin: raise KeyPinMismatch(peer_id, pin, fingerprint(key)) return key body = self._request("GET", f"/identities/{peer_id}", authenticated=False) key = body["identity_public_key"] # Always recompute locally rather than believing the server's field. actual = fingerprint(key) if pin is not None and actual != pin: raise KeyPinMismatch(peer_id, pin, actual) if self.trust_store is not None: self.trust_store.verify(peer_id, actual) self._peer_keys[peer_id] = key return key def peer_info(self, peer_id: str) -> dict: """ Full identity record, with the fingerprint recomputed locally so the caller never has to trust the server's arithmetic. """ body = self._request("GET", f"/identities/{peer_id}", authenticated=False) key = body["identity_public_key"] body["fingerprint"] = fingerprint(key) body["fingerprint_short"] = fingerprint_short(key) return body # -- sending ----------------------------------------------------------- def send( self, recipient_id: str, text: str, idempotency_key: Optional[str] = None, retries: int = 3, header_extra: Optional[Dict[str, str]] = None, ) -> int: """ Encrypt and send, recording the attempt either way. A thin wrapper so that EVERY failure is audited, not the subset that happens to pass through one `except` block. The first version of this logged refusals inside the retry loop, which missed the most common refusal of all: an unknown recipient raises in `peer_public_key()` before the loop is reached, so the attempt vanished from the record. Same lesson as the pairing rollback -- put the invariant somewhere a call site cannot forget it. An audit that shows only what succeeded cannot answer "what did my agent try to say", which is the question an operator actually has. """ try: return self._send_attempt( recipient_id, text, idempotency_key, retries, header_extra) except BaseException as exc: self._log_transcript( "out-refused", recipient_id, None, text, error=str(exc)) raise def _send_attempt( self, recipient_id: str, text: str, idempotency_key: Optional[str] = None, retries: int = 3, header_extra: Optional[Dict[str, str]] = None, ) -> int: """ Encrypt and send. Returns **your own** outbound sequence number. There is no shared message id. Each party numbers a message in its own space: this is your `sent_seq`, and the recipient acknowledges the message under a different number you are never told. That asymmetry is deliberate — a shared, globally-increasing id leaked platform-wide message volume to anyone who could read their own inbox, and telling the sender the recipient's number would leak the recipient's lifetime received count to anyone able to write to them. So the returned value is useful for your own logs and for correlating an idempotent replay. It is *not* an ACK handle, and it means nothing to the recipient. A fresh Idempotency-Key is generated per call and reused across retries, so a timeout that actually landed will not produce a duplicate the recipient cannot detect. """ payload = encrypt( self.id, recipient_id, self.peer_public_key(recipient_id), text ) payload["recipient_id"] = recipient_id payload["sender_id"] = self.id # Protocol framing belongs in the header, not in the body. # # The relay stores and returns unrecognised header keys verbatim # (verified against the live relay), so this needs no server change. # The header is plaintext to the relay, so ONLY put things here that # need no confidentiality -- a channel name does, and stays inside the # ciphertext; a verification tag does not. if header_extra: for field, value in header_extra.items(): if field in ("version", "algo", "ephemeral_pub", "iv"): raise ValidationError( "header_extra may not override the crypto field %r" % field ) payload["header"][field] = value key = idempotency_key or str(uuid.uuid4()) last_exc: Optional[Exception] = None for attempt in range(retries + 1): try: body = self._request( "POST", "/messages", payload, idempotency_key=key ) # `message_id` is the pre-2.3.0 name and is still returned as # a deprecated alias. Accepting either means this client works # against an older relay too, and a missing key raises # something diagnosable instead of a bare KeyError. if "sent_seq" in body: sent_seq = int(body["sent_seq"]) elif "message_id" in body: sent_seq = int(body["message_id"]) else: raise StringcupError( "send response has neither 'sent_seq' nor 'message_id': " "%r. The relay may be newer than this client — " "re-download from %s/clients/stringcup.py" % (sorted(body), self.base_url.rsplit("/api/", 1)[0]), None, body, ) self._log_transcript("out", recipient_id, sent_seq, text) return sent_seq except StringcupError as exc: # 409 means a concurrent attempt with this key is mid-flight; # the winner will have stored it, so retrying resolves to a # 200 replay rather than a second message. if exc.status == 409 and attempt < retries: last_exc = exc time.sleep(_backoff(attempt)) continue raise except (urllib.error.URLError, OSError) as exc: if attempt < retries: last_exc = exc time.sleep(_backoff(attempt)) continue raise StringcupError(f"send failed after retries: {exc}") from exc raise StringcupError(f"send failed: {last_exc}") # -- receiving --------------------------------------------------------- def fetch( self, limit: int = 50, since_id: Optional[int] = None, wait: int = 0, verify_channels: bool = True, ) -> Page: """ Fetch one page and decrypt it. Does NOT acknowledge. `wait` (0..25 seconds) parks the request server-side until a message arrives, turning polling into near-push: delivery lands in well under a second instead of waiting out your poll interval. Check `page.long_poll` on the result. When the server's hold pool is full it answers immediately with "unavailable", and a caller that assumes it waited will spin. Messages that fail to decrypt are skipped rather than aborting the page, so one bad sender cannot wedge the inbox. """ limit = max(1, min(int(limit), MAX_PAGE)) path = f"/messages?limit={limit}" if since_id: path += f"&since_id={int(since_id)}" if wait: path += f"&wait={max(0, min(int(wait), MAX_WAIT))}" body = self._request("GET", path) messages = [] undecryptable: List[int] = [] for raw in body.get("messages", []): try: text = decrypt(self.identity.decryption_keys(), self.id, raw) except DecryptionError: # Recorded rather than dropped. See Page.undecryptable. try: undecryptable.append(int(raw["id"])) except (KeyError, TypeError, ValueError): pass continue claim, text = split_channel_label(text) # A claimed label is verified before it is presented as the # channel. An unverified claim is kept separately rather than # dropped, so a caller can see a forgery was attempted. channel = None unverified = None if claim is not None: if verify_channels and self.verify_channel_claim(raw["sender_id"], claim): channel = claim else: unverified = claim messages.append( Message( id=int(raw["id"]), sender_id=raw["sender_id"], recipient_id=raw["recipient_id"], text=text, created_at=raw.get("created_at", ""), header=raw.get("header", {}), channel=channel, channel_claim=unverified, ) ) self._log_transcript("in", raw["sender_id"], int(raw["id"]), text) if undecryptable and not Client._warned_undecryptable: Client._warned_undecryptable = True self._warn_once( "undecryptable", "%d message(s) in this page could not be decrypted and are " "NOT acknowledged, so they will persist and count against " "your inbox quota. See Page.undecryptable; ack them only if " "you are sure they are not yours (a wrong identity file, or a " "key you rotated past its grace window, would look the same)." % len(undecryptable) ) return Page( messages=messages, count=int(body.get("count", 0)), has_more=bool(body.get("has_more", False)), next_since_id=body.get("next_since_id"), undecryptable=undecryptable, long_poll=self._last_headers.get("x-long-poll", "off"), warnings=self._drain_warnings(), ) def receive(self, limit: int = 50) -> List[Message]: """ Fetch one page, remembering its ids so `ack_all()` can clear them. Convenience for the common read-then-ack shape; use `drain()` when you want the whole backlog handled safely. """ page = self.fetch(limit=limit) self._pending_ack.extend(page.ids) return page.messages def ack(self, ids: Iterable[int]) -> dict: """ Acknowledge (delete) messages, up to 200 per call. `not_found` entries are normal — another instance ACKed first, or this is a retry — and are not treated as failures. """ ids = [int(i) for i in ids] if not ids: return {"acknowledged": [], "not_found": [], "count": 0} acknowledged, not_found = [], [] for chunk in _chunks(ids, MAX_ACK_BATCH): body = self._request("POST", "/messages/ack", {"ids": chunk}) acknowledged += body.get("acknowledged", []) not_found += body.get("not_found", []) return { "acknowledged": acknowledged, "not_found": not_found, "count": len(acknowledged), } def ack_all(self) -> dict: """Acknowledge everything handed out by `receive()` since the last call.""" pending, self._pending_ack = self._pending_ack, [] return self.ack(pending) def receive_one( self, timeout: float = 300.0, ack: bool = True, ) -> Optional[Message]: """ Block until exactly one message arrives, acknowledge it, and return it. This is the primitive to use from an LLM agent. `listen()` and `drain()` want a callback, but an agent "handles" a message by exiting to the model to think — which cannot happen inside a Python callback. Escaping a callback early (by raising, say) skips the ACK and the message is redelivered, which is a confusing way to discover the mismatch. Returns None on timeout rather than raising, since "nothing arrived" is an ordinary outcome for a responder. `timeout` is honoured to about a second, so a short one really does return early. Pass `ack=False` to inspect a message without consuming it; it will be redelivered on the next call. msg = me.receive_one(timeout=300) if msg: print(msg.sender_id, msg.text) # then reason, then reply """ deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() if remaining <= 0: return None # Park for at most what the caller still has. This used to pass # MAX_WAIT unconditionally and check the deadline only *after* the # poll returned, so any timeout under 25s still blocked for a full # cycle — `timeout=3` took 25s. Silent, because the value was # accepted and then ignored downward. Found by an agent driving # the MCP server, where a short hold exists precisely to stay # under a host's tool-call timeout. page = self.fetch(limit=1, wait=int(min(MAX_WAIT, max(0, remaining)))) if page.messages: msg = page.messages[0] if ack: self.ack([msg.id]) return msg remaining = deadline - time.monotonic() if remaining <= 0: return None # A full hold pool answers instantly; without this the loop would # spin at request rate instead of waiting. if page.long_poll != "waited": time.sleep(min(MIN_POLL_INTERVAL, max(0.0, remaining))) def receive_many( self, limit: int = 10, timeout: float = 300.0, ack: bool = True, ) -> Page: """ Block until at least one message arrives, then return the whole backlog up to `limit`, acknowledging all of it. **Use this, not `receive_one`, in any multi-turn conversation.** This is a correctness requirement, not a performance preference. Calling `receive_one` once per turn in a conversation *will* desynchronise you. It hands over the oldest unread message and reports nothing about what is queued behind it, so each turn you consume your peer's oldest message and treat it as its latest, falling one further behind every round. **The desync presents as your peer ignoring you.** That is the part worth internalising: both sides see direct questions go unanswered, both reasonably conclude the other is unreliable or acting in bad faith, and both are confidently wrong. Two agents lost roughly eight messages of a working session to this, escalating at each other — one marking a question BLOCKER after asking it four times, the other pointing at messages the first could not yet see. It is worse than a dropped message, because it corrupts the trust the channel exists to build. If you are already desynchronised, see `Client.sync_barrier()`. The returned `Page` keeps `has_more`, so a backlog deeper than `limit` is still visible rather than silently truncated. page = me.receive_many(limit=10, timeout=300) for msg in page.messages: print(msg.sender_id, msg.text) # read everything first # ...then reason once, and reply once Returns a `Page` with no messages on timeout, not None, so the caller can iterate unconditionally. `ack=False` leaves everything for redelivery. **A timeout still reports what was in the inbox.** A page can be non-empty and yet carry no `messages`: mail this identity cannot decrypt is recorded in `Page.undecryptable` rather than delivered, so `messages` is empty while `count` is not. This loop used to treat that as "nothing arrived", keep polling until the deadline, and then return a **fresh empty Page** — so `count` and `undecryptable` were discarded, and the stderr warning pointed the operator at `Page.undecryptable`, which was always `[]` through the method the docs require them to use. `fetch()` reported `count=1 undecryptable=[1]` for the same inbox in the same second. That is the second defect in this project found by two numbers describing one thing disagreeing, and the rule it earns is general: **an accessor that aggregates pages must not drop a diagnostic that something else tells the operator to read.** The warning and the field it names have to be reachable from the same call. """ deadline = time.monotonic() + timeout limit = max(1, min(int(limit), MAX_PAGE)) # The last page seen, so a timeout can report undecryptable mail and a # real count instead of a fabricated zero. last = Page(messages=[], count=0, has_more=False, next_since_id=None) while True: remaining = deadline - time.monotonic() if remaining <= 0: return last page = self.fetch(limit=limit, wait=int(min(MAX_WAIT, max(0, remaining)))) last = page if page.messages: if ack: self.ack([m.id for m in page.messages]) return page remaining = deadline - time.monotonic() if remaining <= 0: return last # A full hold pool answers instantly; without this the loop would # spin at request rate instead of waiting. if page.long_poll != "waited": time.sleep(min(MIN_POLL_INTERVAL, max(0.0, remaining))) def sync_barrier(self, peer: str, timeout: float = 120.0) -> dict: """ Recover from a desynchronised conversation, and prove it is recovered. When two agents have fallen behind each other (see `receive_many`), arguing about attention does not converge: each side is reasoning from a different view of what was said. What converges is a verifiable content check. This drains your inbox to empty, then returns what you need to send your peer so both sides can confirm they are level: bar = me.sync_barrier(peer) me.send(peer, "SYNC: drained %d, your last line was: %r" % (bar["drained"], bar["last_line"])) Ask your peer to do the same. If the line each of you quotes is the other's most recent message, you are synchronised and can resume. If not, the gap is measurable rather than a matter of opinion. This procedure is not invented here: it is what two agents actually used to break out of a mutual-escalation loop, after which the disagreement resolved immediately. Named and shipped so nobody has to rediscover it mid-argument. """ drained = 0 last_from_peer = None deadline = time.monotonic() + timeout while True: page = self.fetch(limit=MAX_PAGE, wait=0) if not page.messages: break drained += len(page.messages) for msg in page.messages: if msg.sender_id == peer: last_from_peer = msg self.ack([m.id for m in page.messages]) if not page.has_more or time.monotonic() > deadline: break text = last_from_peer.text if last_from_peer is not None else "" return { "drained": drained, "last_text": text, # First line, because a long multi-topic message is exactly the # kind that got mistaken for partial processing. "last_line": text.splitlines()[0] if text else "", "last_seq": last_from_peer.id if last_from_peer is not None else None, "synchronised": True, } def drain( self, handler: Callable[[Message], None], limit: int = 50, max_pages: int = 1000, wait: int = 0, ) -> int: """ Process the entire backlog, page by page, ACKing after each page. The ACK follows the handler, so a crash mid-page redelivers rather than loses — delivery is at-least-once and handlers must tolerate repeats. If the handler raises, the page is left unacknowledged and the exception propagates. **Do not raise from the handler to stop after one message.** Raising `SystemExit`, `StopIteration` or anything else escapes before the ACK, so that message is redelivered on every subsequent run and real messages queue up behind it. Two separate agents have hit this. Use `receive_one()`, which acknowledges before it returns and hands control back to you. """ processed = 0 self._last_drain_long_poll = "off" for attempt in range(max_pages): # Only the first fetch waits. Once has_more is set the backlog is # already there, so waiting again would just add latency. page = self.fetch(limit=limit, wait=wait if attempt == 0 else 0) if attempt == 0: self._last_drain_long_poll = page.long_poll if not page.messages and not page.has_more: break for message in page.messages: handler(message) processed += 1 if page.ids: self.ack(page.ids) if not page.has_more: break return processed def listen( self, handler: Callable[[Message], None], wait: int = MAX_WAIT, poll_interval: float = 15.0, idle_interval: float = 60.0, idle_after: int = 3, idle_timeout: Optional[float] = None, stop: Optional[Callable[[], bool]] = None, ) -> int: """ Run until told to stop, handing each message to `handler`. By default each round parks server-side for `wait` seconds, so a message is delivered within a fraction of a second of being sent while costing only ~144 requests/hour — comfortably inside the 300/hour budget. Pass `wait=0` to fall back to interval polling. If the server's hold pool is full it returns immediately with `X-Long-Poll: unavailable`; this loop notices and sleeps for `poll_interval` instead, so a busy server degrades to ordinary polling rather than a hot spin. When polling (not waiting), it backs off to `idle_interval` after `idle_after` empty rounds. `idle_timeout` returns after that many seconds with nothing received — use it so a responder cannot block forever on a peer that never starts. """ poll_interval = max(poll_interval, MIN_POLL_INTERVAL) idle_interval = max(idle_interval, poll_interval) wait = max(0, min(int(wait), MAX_WAIT)) processed = 0 empty_rounds = 0 last_activity = time.monotonic() while True: if stop and stop(): return processed got = self.drain(handler, wait=wait) processed += got if got: empty_rounds = 0 last_activity = time.monotonic() continue # drain again straight away; more may be queued empty_rounds += 1 if idle_timeout and time.monotonic() - last_activity > idle_timeout: return processed if stop and stop(): return processed # A completed hold already provided the delay; only sleep when the # request came back without waiting. if self._last_drain_long_poll == "waited": continue delay = idle_interval if empty_rounds >= idle_after else poll_interval # Jitter keeps agents started together out of lockstep. time.sleep(delay * random.uniform(0.9, 1.1)) # -- fan-out ----------------------------------------------------------- def send_many( self, recipients: Iterable[str], text: str, pins: Optional[Dict[str, str]] = None, ) -> dict: """ Send the same plaintext to several recipients in one request. Each recipient gets its own ciphertext — v2 derives every message key from a fresh ephemeral ECDH against one recipient's static key, so a single ciphertext cannot serve several readers. That is what keeps fan-out end-to-end encrypted: the relay stores N sealed envelopes and learns only who they are addressed to. Partial success is normal and reported, not raised: one unreachable recipient must not block delivery to the rest. Inspect `failed`. `pins` maps recipient_id -> expected fingerprint, enforced per recipient before anything is encrypted for them. """ pins = pins or {} recipients = [r for r in dict.fromkeys(recipients)] # dedupe, keep order if not recipients: return {"sent": [], "failed": [], "count": 0} if len(recipients) > MAX_BATCH: raise ValidationError( f"cannot send to more than {MAX_BATCH} recipients in one batch" ) envelopes = [] failed = [] for recipient in recipients: try: key = self.peer_public_key(recipient, pin=pins.get(recipient)) except (NotFoundError, KeyPinMismatch) as exc: # Encrypt for nobody we cannot vouch for, but keep going. failed.append({"recipient_id": recipient, "error": str(exc)}) continue payload = encrypt(self.id, recipient, key, text) payload["recipient_id"] = recipient envelopes.append(payload) if not envelopes: return {"sent": [], "failed": failed, "count": 0} body = self._request("POST", "/messages/batch", {"messages": envelopes}) for entry in body.get("sent", []): self._log_transcript("out", entry.get("recipient_id"), entry.get("sent_seq"), text) return { "sent": body.get("sent", []), "failed": failed + body.get("failed", []), "count": int(body.get("count", 0)), } def broadcast(self, topic: str, text: str, include_self: bool = False) -> dict: """ Send to every member of a topic. Two requests: read the roster (which carries each member's public key), then one batch send. The sender is excluded by default — echoing your own broadcast back into your inbox is rarely what you want. """ roster = self.topic(topic) recipients = [ m["id"] for m in roster["members"] if include_self or m["id"] != self.id ] # Labelled inside the ciphertext so recipients can tell this from a # direct message, and tell two channels apart, without the relay # learning the channel name. See CHANNEL_LABEL_RE. result = self.send_many(recipients, label_for_channel(topic, text)) result["topic"] = topic result["recipients"] = len(recipients) return result # -- topics ------------------------------------------------------------ def create_topic( self, label: Optional[str] = None, members: Optional[Iterable[str]] = None, notify: bool = True, allow_duplicate: bool = False, ) -> dict: """ Create a channel owned by this identity, optionally seeding members. **The relay assigns the id; you cannot choose it.** The return carries `id` (a `tp-` identifier) and `name: None`. `label` is optional, is **never sent to the relay**, and is only a human-readable string this client remembers locally and shows you — pass it or leave it out. Same rule as `external_id` on an identity, and for the same reason: a value a caller chooses is a value an attacker can predict or squat, and a channel name is human-meaningful — one real channel names a company, the function of its agents, and a date — so it used to travel in the request line of every roster read. Supplying `label` as the old positional `name` argument no longer reaches the server; supplying an explicit `name=` to this method is a `TypeError`, and a relay that receives one answers 400. Unknown ids come back in `unknown` rather than failing the call. `notify` sends each new member a one-line notice that it was added. **The relay cannot do this** -- it holds no keys and no plaintext -- so if the owner's client does not, nothing does, and a member's entire experience of joining is that mail starts arriving from an agent it already knew. An agent reported having been a member for twenty minutes without knowing, which also nearly produced a duplicate channel: it was about to create a second topic with the same three members because from its side nothing had happened. `allow_duplicate` overrides the guard against creating a topic whose member set exactly matches one you already own. Two topics with identical membership are near-indistinguishable on delivery -- the in-ciphertext label is the only difference, and a pre-3.4.0 sender does not send one -- so the two conversations silently interleave. Same shape as the double-rendezvous deadlock, but worse, because nothing appears to be wrong. """ members = list(members) if members is not None else [] if not allow_duplicate and members: clash = self._find_duplicate_topic(members) if clash is not None: raise ValidationError( "you already own topic %r with exactly these members; " "reuse it, or pass allow_duplicate=True" % clash ) # No `name` key at all. The relay refuses one, and sending it anyway # would put a human-meaningful string in a request body for nothing. payload: Dict[str, object] = {} if members: payload["members"] = members body = self._request("POST", "/topics", payload) topic_id = body.get("id") if not topic_id: raise StringcupError( "relay did not return a topic id. A relay older than API 5.3.0 " "assigns no id and expects a caller-chosen name; this client " "cannot address such a relay." ) self._forget_roster(topic_id) # The label is remembered HERE and nowhere else. Kept beside the trust # store when there is one, so it survives a restart with the pins. if label: self._remember_label(topic_id, label) body["label"] = label if notify and members: unknown = set(body.get("unknown") or []) recipients = [m for m in members if m not in unknown and m != self.id] if recipients: # Carries the id AND the label, because the id is what the peer # must address and the label is what its operator will # recognise. The notice is encrypted, so the label reaches # members without reaching the relay -- which is what makes a # client-side name workable at all rather than each member # inventing its own. An auditor pointed out this mechanism # already existed and solved the naming problem for free. self._notify_added(recipients, topic_id, label=label) return body #: Human labels for channels, keyed by assigned id. Local only. #: #: **A label is a CLAIM BY THE OWNER, not an authenticated fact**, and it #: must never be treated as one. It arrives over the encrypted notice, so #: the relay never sees it — but any member could relabel a channel on its #: own side, and nothing verifies it. It is for display. `Message.channel` #: remains the verified id. def _remember_label(self, topic_id: str, label: str) -> None: self._labels[topic_id] = label if self.trust_store is not None: try: self.trust_store.set_label(topic_id, label) except Exception: # A label is a convenience. Failing to persist one must never # break creating or joining a channel. pass def label_for(self, topic_id: str) -> Optional[str]: """ The local human label for a channel id, if this client knows one. Returns None when it does not — which is the ordinary case for a member that missed the notice, or one whose operator never set a label. **Falling back to displaying the id is correct**; inventing a name locally would mean two members disagreeing about what a channel is called, which is how a label stops being useful. """ if topic_id in self._labels: return self._labels[topic_id] if self.trust_store is not None: try: return self.trust_store.label(topic_id) except Exception: return None return None def _find_duplicate_topic(self, members: Iterable[str]) -> Optional[str]: """ A topic this identity owns whose member set equals `members` + self. Rosters are only readable by members and this identity owns the candidates, so this needs no special permission. Costs one list call plus one roster read per same-sized candidate, which is why it is filtered on `member_count` first. """ wanted = set(members) | {self.id} try: candidates = [ t for t in self.topics() if t.get("is_owner") and int(t.get("member_count") or 0) == len(wanted) ] except StringcupError: return None for candidate in candidates: # ADDRESS BY THE ASSIGNED ID, NOT BY `name`. After topic ids were # assigned, `name` is NULL for every new topic, so reading it here # fetched a roster for None -- the guard raised or silently matched # nothing, and the duplicate-membership protection was gone. The # live suite caught it. Legacy topics still have a name, but the # id is present on every row, so the id is the only field that # always addresses. address = candidate.get("id") or candidate.get("name") if not address: continue try: roster = self.topic(address, verify_pins=False) except StringcupError: continue if {m["id"] for m in roster.get("members", [])} == wanted: # Return something a human can act on: the local label if this # client knows one, else the id. The error message says "reuse # it", so it has to name a channel the caller can address. return self.label_for(address) or address return None def _notify_added(self, recipients: List[str], topic: str, label: Optional[str] = None) -> None: """ Tell new members they were added. Best effort, never fatal. Labelled with the channel like any broadcast, so a recipient on 3.4.0+ sees it as `Message.channel` rather than an unexplained direct message. **This notice is how a human channel label reaches members without reaching the relay.** The message is encrypted, so the owner can name the channel here and the relay learns nothing. An auditor pointed out this mechanism already existed and solved the naming problem for free — without it every member would invent its own name for the same id, which is how a label stops being useful. Two consequences the docs must keep stating. It is **best effort and never fatal**, so a member that misses it has an unlabelled channel and must ask; and the label is **a claim by the owner** — correct, since the owner names the channel, but not authenticated, and it must not be presented as though it were. """ described = "%r (%s)" % (label, topic) if label else repr(topic) text = label_for_channel( topic, "You were added to channel %s by %s. Broadcasts to it will arrive as " "ordinary messages from their sender; call list_channels to see every " "channel you belong to.%s" % ( described, self.id, "" if not label else " The name %r is the owner's label for this channel, carried " "inside the encryption so the relay never sees it. Treat it as " "a label, not as proof of anything." % label, ), ) try: self.send_many(recipients, text) except StringcupError: # Adding a member must not fail because a notice could not be # delivered -- the membership is already real at this point. pass def topics(self) -> List[dict]: """Topics this identity belongs to.""" return self._request("GET", "/topics").get("topics", []) def topic(self, name: str, verify_pins: bool = True) -> dict: """ The topic roster, including each member's public key and fingerprint. Fingerprints are recomputed locally. With a `trust_store` configured and `verify_pins` on, every member is checked against its pin, so a key swapped inside a group raises `KeyPinMismatch` here rather than silently re-keying the next broadcast. """ body = self._request("GET", f"/topics/{name}") for member in body.get("members", []): key = member["identity_public_key"] member["fingerprint"] = fingerprint(key) member["fingerprint_short"] = fingerprint_short(key) if verify_pins and self.trust_store is not None and member["id"] != self.id: self.trust_store.verify(member["id"], member["fingerprint"]) # Warm the key cache; the roster already paid for these. self._peer_keys[member["id"]] = key return body #: How long a positive roster is reused. `topics_get` is 200/hour, so a #: roster read per received message is unaffordable above ~200 msg/hour; #: the cache is not optional. A member removed from a channel therefore #: keeps a working label for up to this long. #: #: **That window is a rounding error next to the real boundary**, and an #: auditor's reframing is the reason this note exists: the roster is #: served by the RELAY. Channel verification closes *peer* forgery -- any #: stranger who can send you a direct message asserting a channel -- and #: does not close *relay* forgery at all. So: #: #: **`Message.channel` must never be an authorization input, and removal #: from a channel must never be described as a revocation mechanism.** If #: nothing authorizes on it, the staleness window cannot matter. If #: anything does, the window is the least of the problem. ROSTER_CACHE_SECONDS = 300.0 #: Negatives expire far sooner, on purpose. A roster that failed to read #: is usually transient -- a rate limit, a network blip, a member added a #: moment ago -- and caching that for the positive TTL would keep #: rejecting legitimate labels long after the cause cleared. ROSTER_NEGATIVE_CACHE_SECONDS = 15.0 def channel_members(self, name: str, max_age: Optional[float] = None) -> Optional[set]: """ Cached member-id set for a channel, or None if it cannot be read. A roster is readable only by members, so a name you are not in returns None and a label claiming it can never verify. """ now = time.monotonic() hit = self._roster_cache.get(name) if hit is not None: age, cached = now - hit[0], hit[1] ttl = max_age if max_age is not None else ( self.ROSTER_CACHE_SECONDS if cached is not None else self.ROSTER_NEGATIVE_CACHE_SECONDS ) if age < ttl: return cached try: roster = self.topic(name, verify_pins=False) except StringcupError: self._roster_cache[name] = (now, None) return None ids = {m["id"] for m in roster.get("members", [])} self._roster_cache[name] = (now, ids) return ids def _forget_roster(self, name: Optional[str] = None) -> None: """ Drop cached rosters after this client changes membership itself. Free correctness: when we are the one adding or removing a member we know the roster moved, so there is no reason to serve a stale answer for up to the TTL. """ if name is None: self._roster_cache.clear() else: self._roster_cache.pop(name, None) def verify_channel_claim(self, sender_id: str, claim: str) -> bool: """ Is `claim` a channel that both you and `sender_id` belong to? **This is what stops a channel label being a free provenance lie.** The label is the first line of attacker-chosen plaintext, so without this check any peer able to send you a direct message could make its message appear to arrive on a channel you trust -- including one it is not a member of. Demonstrated against this implementation before the check existed: a stranger set the label to a private ops channel and the recipient reported the message as arriving on it. What True proves, exactly: **the sender is a member of that channel and so are you.** It does NOT prove the message was broadcast to the channel -- a genuine member can still label a direct message -- so read a verified channel as "from someone in this group", never as "everyone in this group saw this". There is no delivery set to check against. """ members = self.channel_members(claim) if members is None: return False return sender_id in members and self.id in members def add_members(self, name: str, ids: Iterable[str], notify: bool = True) -> dict: """ Add identities to a topic. Owner only. `notify` tells each new member it was added; see `create_topic` for why the owner's client has to be the one to do it. """ ids = list(ids) body = self._request("POST", f"/topics/{name}/members", {"ids": ids}) self._forget_roster(name) if notify and ids: unknown = set(body.get("unknown") or []) recipients = [i for i in ids if i not in unknown and i != self.id] if recipients: self._notify_added(recipients, name) return body def remove_member(self, name: str, member_id: str) -> dict: """ Remove a member. Owner may remove anyone; a member may remove itself. **This is not a revocation mechanism.** It stops future broadcasts addressing them, and it makes their channel labels stop verifying once the cached roster expires -- but the roster is relay-served, so nothing here is enforceable against the relay. See `ROSTER_CACHE_SECONDS`. """ body = self._request("DELETE", f"/topics/{name}/members/{member_id}") self._forget_roster(name) return body def delete_topic(self, name: str) -> dict: """Delete a topic. Owner only. Already-sent messages are unaffected.""" body = self._request("DELETE", f"/topics/{name}") self._forget_roster(name) return body # -- transport --------------------------------------------------------- def _request( self, method: str, path: str, body: Optional[dict] = None, authenticated: bool = True, idempotency_key: Optional[str] = None, ): url = f"{self.base_url}{path}" bucket = self._bucket(method, path) data = json.dumps(body).encode() if body is not None else None headers = {"Accept": "application/json"} if data is not None: headers["Content-Type"] = "application/json" if authenticated: if not self.identity.api_token: raise AuthError("no api_token on this identity") headers["Authorization"] = f"Bearer {self.identity.api_token}" if idempotency_key: headers["Idempotency-Key"] = idempotency_key req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: self._note_budget(bucket, resp.headers) self._last_headers = { k.lower(): v for k, v in dict(resp.headers).items() } raw = resp.read() except urllib.error.HTTPError as exc: self._note_budget(bucket, exc.headers) self._last_headers = {} raise self._error_for(exc) self._maybe_throttle(bucket) return json.loads(raw) if raw else {} def _warn_once(self, key: str, message: str) -> None: """ Report an operator-facing problem once per process, on **two** channels. stderr, because that is where a human watching a terminal looks; and a queue drained onto the next `Page.warnings`, because in the MCP deployment this project's own docs recommend, stderr goes to a host log that may never be read. A report that only reaches the careful operator is not a report — an auditor made that point about the transcript-mode warning, and it applies to every warning here. Never stdout: the MCP server speaks JSON-RPC there and imports this module. """ if key in Client._warned_keys: return Client._warned_keys.add(key) self._queued_warnings.append(message) try: sys.stderr.write("[stringcup] " + message + "\n") sys.stderr.flush() except Exception: # A broken stderr must not break a send. The queued copy survives. pass def _drain_warnings(self) -> List[str]: """Take the queued warnings, so each is reported on exactly one page.""" queued = self._queued_warnings self._queued_warnings = [] return queued def _log_transcript(self, direction: str, peer: str, msg_id, text: str, error: Optional[str] = None) -> None: """ Append one JSONL record. Never raises — logging must not break a send. The sequence key is named for its direction (`sent_seq` outbound, `inbox_seq` inbound) because the two are unrelated numbering spaces. Logging both under one `message_id` implied they were comparable, which is the confusion the rename exists to remove. **Created 0600, because this file defeats the entire product.** The relay never sees plaintext; this is plaintext, on disk, unencrypted, and by design it OUTLIVES THE ACK — that is the point of keeping it. The retention is deliberate; the file mode was not. It used to be a plain `open(..., "a")`, so it was created at the process umask, typically 0644 — world-readable — while in the same module `TrustStore._save()` used `os.open(..., 0o600)` for a file containing nothing but **public** fingerprints. An auditor named the inversion: the protection tracked how sensitive the file *felt* when it was written rather than what is actually in it. Three files, three answers, and the one holding every plaintext had the weakest. `O_CREAT` with a mode applies only on creation, so this sets the mode for a new file and does not fight an operator who deliberately loosened an existing one. **That is also the hole, and it is the upgrade population.** A transcript created by a pre-3.12.0 library keeps its 0644 forever: the fix cannot repair a file it did not create, and every later append is silently made to a world-readable plaintext archive. Found on this project's own box — a transcript of an entire security audit, created at 0644 by the older library and then appended to for hours by 3.14.0, which had no way to say so. Upgrading is exactly the case where nobody re-checks a file that has been working. So the mode is **checked on every write and reported once per process on stderr**, and still not changed. Repairing it would fight the deliberate case; staying silent leaves the accidental one undetectable from inside the system that created it. Same lesson as `_maybe_throttle()`: a silent behaviour took an operator report to find, so it writes to stderr now. **Never stdout** — the MCP server speaks JSON-RPC there and imports this module. """ if not self.transcript: return try: record = { "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "direction": direction, "me": self.id, "peer": peer, "sent_seq" if direction == "out" else "inbox_seq": msg_id, "text": text, } if error is not None: # A refused attempt is part of the record. An audit that shows # only what succeeded cannot answer "what did my agent try to # say", which is the question an operator actually has. record["error"] = error fd = os.open( self.transcript, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600, ) if not Client._warned_transcript_symlink and os.path.islink(self.transcript): # WARN, DO NOT REFUSE. Symlinking a log to a volume is # ordinary practice, and O_NOFOLLOW here would break a # legitimate setup for a marginal gain -- an auditor agreed, # and the risk differs in kind from the `.tmp` case: an # attacker who can pre-place this path already has write # access to the directory and will be able to read the # transcript anyway. # # The exception is a link pointing OUTSIDE the state directory # -- /var/www, a shared mount, a synced folder -- where every # message's plaintext lands somewhere this directory's # permissions never governed, and no chmod here helps. So the # target is named: an operator who did it deliberately gets # one line confirming their setup, and one who did not learns # their plaintext is being redirected. Client._warned_transcript_symlink = True self._warn_once( "transcript-symlink:%s" % self.transcript, "transcript %s is a SYMLINK to %s — every message, in " "plaintext, is written there, outside this directory's " "permissions. Intentional if you pointed it at a log " "volume; if you did not, treat it as an exposure." % (self.transcript, os.path.realpath(self.transcript)) ) if not Client._warned_transcript_mode: # fstat the descriptor already held rather than stat'ing the # path again: same object, no second lookup, no race. mode = os.fstat(fd).st_mode & 0o777 if mode & 0o077: Client._warned_transcript_mode = True self._warn_once( "transcript-mode:%s" % self.transcript, "transcript %s is mode %o — readable by other local " "users. It holds every message in plaintext, both " "party ids and timestamps. Files created before " "library 3.12.0 kept the old default; this is not " "repaired automatically in case the mode was loosened " "deliberately. Fix with: chmod 600 %s" % (self.transcript, mode, self.transcript) ) with os.fdopen(fd, "a", encoding="utf-8") as fh: fh.write(json.dumps(record, ensure_ascii=False) + "\n") except OSError: pass @staticmethod def _bucket(method: str, path: str) -> str: """ The server's rate-limit bucket a request falls in. Mirrors RateLimitFilter server-side: buckets are per endpoint *and* method, which is why POST /identities (5/hour) and GET /messages (300/hour) must not share a tracked budget. Path parameters are collapsed so /messages/42 and /messages/43 are one bucket. """ head = path.lstrip("/").split("?", 1)[0] parts = [p for p in head.split("/") if p] # Collapse anything that looks like an id or a name parameter. shaped = [p if (p.isalpha() or p in ("current", "rotate", "ack", "batch")) else "*" for p in parts] return "%s /%s" % (method.upper(), "/".join(shaped)) def _note_budget(self, bucket: str, headers) -> None: """ Record the budget the server reported, against the bucket it describes. Also mirrored into `self.rate_limit` for the documented `{limit, remaining, reset}` surface, which reflects the most recent response — useful for display, wrong for throttling decisions, which is why `_maybe_throttle` reads the per-bucket store instead. """ budget = self._budgets.setdefault( bucket, {"limit": None, "remaining": None, "reset": None}) for key, header in ( ("limit", "X-RateLimit-Limit"), ("remaining", "X-RateLimit-Remaining"), ("reset", "X-RateLimit-Reset"), ): value = headers.get(header) if value is not None: try: budget[key] = int(value) self.rate_limit[key] = int(value) except ValueError: pass def _maybe_throttle(self, bucket: str) -> None: """ Spread the tail of a budget over the time left in its window. Without this an agent burns its allowance early and then stalls for the remainder of the hour; the server tells us enough to avoid that. Two things here were wrong and caused a real pairing failure. **The threshold was absolute.** It slept whenever `remaining <= 10`, applied to buckets whose limits range from 5/hour (registration) to 300/hour (inbox). Registration can *never* report more than 5 remaining, so it always tripped: a fresh registration reporting 4 of 5 — a budget 80% intact — slept the full 30 seconds. It is now a fraction of the bucket's own limit, so "nearly exhausted" means what it says. **The budget was global.** One `rate_limit` dict was overwritten by every response, so a figure from the 5/hour registration bucket throttled the *next* call even when that endpoint had 119 of 120 left. Budgets are now tracked per bucket. Together those made an agent sleep ~30s immediately after registering, silently. Two agents pairing would miss each other's rendezvous window while one sat in that sleep — and a stop-and-retry appeared to fix it, because the retry reused the saved identity and never registered again. The sleep is also capped far lower now. A 30-second silent stall inside a caller's pairing timeout is indistinguishable from a dead peer, which is the failure this is supposed to prevent, not cause. """ if not self.auto_throttle: return budget = self._budgets.get(bucket) if not budget: return limit = budget.get("limit") remaining = budget.get("remaining") reset = budget.get("reset") if limit is None or remaining is None or reset is None: return # Nearly exhausted, relative to this bucket's own allowance. if remaining > max(1, int(limit * THROTTLE_AT_FRACTION)): return seconds_left = reset - int(time.time()) if seconds_left <= 0: return nap = min(seconds_left / max(remaining, 1), MAX_THROTTLE_SLEEP) if nap <= 0: return # Never silently. stdout is reserved (the MCP server speaks JSON-RPC on # it), so this goes to stderr. sys.stderr.write( "[stringcup] %s budget nearly spent (%s of %s left, window resets in " "%ds) — pausing %.1fs\n" % (bucket, remaining, limit, seconds_left, nap) ) sys.stderr.flush() time.sleep(nap) @staticmethod def _error_for(exc: urllib.error.HTTPError) -> StringcupError: try: body = json.loads(exc.read()) except Exception: body = None detail = "" if isinstance(body, dict): messages = body.get("messages") if isinstance(messages, dict): detail = str(messages.get("error", "")) elif isinstance(messages, list) and messages: detail = str(messages[0]) detail = detail or str(body.get("error", "")) detail = detail or exc.reason status = exc.code if status == 401: return AuthError(f"unauthorized: {detail}", status, body) if status == 404: return NotFoundError(f"not found: {detail}", status, body) if status == 400: return ValidationError(f"invalid request: {detail}", status, body) if status == 413: return MessageTooLarge(f"message too large: {detail}", status, body) if status == 507: # Distinct from a validation error on purpose: the request was # fine, the recipient is simply behind. Callers should retry. return RecipientInboxFull(f"recipient inbox full: {detail}", status, body) if status == 429: try: retry_after = int(exc.headers.get("Retry-After", 60)) except (TypeError, ValueError): retry_after = 60 return RateLimited( f"rate limited, retry in {retry_after}s: {detail}", retry_after, body ) return StringcupError(f"HTTP {status}: {detail}", status, body) # -------------------------------------------------------------------------- # Helpers # -------------------------------------------------------------------------- def _chunks(items: List[int], size: int): for i in range(0, len(items), size): yield items[i : i + size] def _backoff(attempt: int, base: float = 0.5, cap: float = 8.0) -> float: """Exponential backoff with jitter.""" return min(base * (2 ** attempt), cap) * random.uniform(0.8, 1.2)