25 Commits

Author SHA1 Message Date
egutierrez 7b7226f069 Merge quick/mobile-parity: binding mobile ampliado (wallet BIP39, ListMyRooms, Directory) para paridad del cliente Android 2026-06-18 23:52:21 +02:00
agent 53f8a4a3d6 feat(mobile): reconstruir+ampliar binding gomobile para paridad con la web
El wrapper mobile/unibus.go se habia perdido del repo (solo quedaba compilado
en el .aar del 5 jun). Se reconstruye y amplia con:

- Wallet BIP39 determinista: NewMnemonic, ValidateMnemonic, DeriveAndSaveIdentity.
  Deriva la MISMA identidad que uniweb (web/src/wallet/derive.ts): PBKDF2-BIP39 ->
  HKDF-SHA256(info unibus-sign-v1 / unibus-kex-v1) -> Ed25519 + X25519. Test de
  paridad contra el vector de oro (mnemonica abandon...about -> sign_pub
  34302746...b3c8) garantiza misma cuenta web<->movil byte a byte.
- Selector de salas: Session.ListMyRooms() -> JSON [{id,subject,mode,role}].
- Nombres legibles: Session.Directory() + Client.Directory()/EndpointID() nuevos
  en pkg/client (GET /directory firmado).
- HasIdentity/SignPubAt para el onboarding.

Aditivo; build/vet/test del modulo verdes (incluido TestDeriveParityWithWeb).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-18 23:38:14 +02:00
egutierrez 4dea99a524 chore: auto-commit (1 archivos)
- build/

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-06-14 23:55:18 +02:00
egutierrez 07978fc697 Merge branch 'issue/room-history-endpoint'
Server owns the JetStream stream of persisted rooms + GET /rooms/{id}/history so
clients without JetStream (uniweb) can read the backlog over plain HTTP.
2026-06-14 19:47:05 +02:00
egutierrez bf47511f2a docs(unibus): bump to 0.16.0; document server stream ownership + history endpoint
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 19:46:56 +02:00
egutierrez 73fd89f0b9 feat(membershipd): open JetStream for the embedded node + wire it into the server
The control plane previously opened a privileged JetStream client only when
clustered or running --store kv (needJS). It now also opens one for a standalone
single-node embedded deployment (openJS = needJS || embedded), because the
embedded NATS always ships JetStream and the server needs it to own persisted
rooms' durable streams (ensure on create + serve GET /rooms/{id}/history). An
external NATS without a cluster/KV feature is unchanged (no JetStream; history
degrades to empty).

The internal service identity is generated under the same broadened condition so
the in-process JetStream connection authenticates under enforce. After NewServer
the js context is wired via SetJetStream with the control-plane KV replication
factor, so a persisted room's history is as available as its metadata.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 19:46:56 +02:00
egutierrez e71063b16e feat(membership): server owns persisted rooms' stream + GET /rooms/{id}/history
The durable JetStream stream of a persisted (ModeMatrix) room was created only
by the Go client's first publish/subscribe. A client that speaks only core NATS
(the browser client uniweb, which has no JetStream) therefore never created it,
so its messages were captured nowhere and lost on reload. Move stream ownership
to the control plane and expose the backlog over plain HTTP.

- handleCreateRoom ensures the room's stream (idempotent CreateOrUpdateStream)
  BEFORE writing the room row, so the subject is captured from the first message
  whoever publishes it. Done before the store write so a stream failure leaves no
  orphan room. Skipped when no JetStream is wired (room still works, no history).
- New member-only GET /rooms/{id}/history?limit=N (default 200, hard cap 1000):
  reads the stream server-side via the modern jetstream API (Stream.Info +
  GetMsg by sequence, no consumer) and returns the last N frames oldest->newest
  as {"messages":[<base64-std of the marshaled frame>]}. The server never
  decrypts — it relays the E2E ciphertext bytes the stream already holds.
  Existence is checked first (404), then membership (403); enforce rejects an
  unsigned request with 401 before the handler runs.
- Lazy backfill: the history endpoint ensures the stream of a pre-existing
  persisted room, so it starts capturing from now on. Messages sent before the
  stream existed were never captured and are unrecoverable.
- The stream config (streamConfigForRoom) mirrors pkg/client/persist.go
  byte-for-byte plus Replicas (matched to the control-plane KV replication). It
  is copied rather than imported because pkg/client imports pkg/membership and
  the reverse would be an import cycle; the source of truth is documented in a
  comment.
- Server gains SetJetStream(js, replicas) to wire the privileged JetStream
  context and the room-stream replication factor.

Tests (history_test.go): golden (3 frames round-trip in order, decodable),
core-NATS capture (the central fix), handleCreateRoom creates the stream, limit,
empty room ([] not null), 401 unsigned, 403 non-member, 404 missing room.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 19:46:45 +02:00
egutierrez 3fdbb54353 Merge branch 'quick/directory-route-path'
fix: directory route registered as /directory (Caddy strips /api) — was 404 in prod
2026-06-14 16:05:06 +02:00
egutierrez 8c3ddaa294 fix(membership): register directory route as /directory, not /api/directory
Caddy strips /api via `handle_path /api/*` before forwarding to membershipd,
so the SPA's GET /api/directory arrives as GET /directory. The route was
registered with the /api prefix, so the stripped request hit no route and
returned 404 in production: the directory never resolved and uniweb fell back
to short ids. Every other control-plane route is registered without the prefix;
this aligns directory with them.

The unit test passed despite the bug because it requested /api/directory, the
same wrong path as the registration. Corrected the request paths to /directory
so the test now exercises the real production path (verified: reverting the
registration to /api/directory now makes TestDirectoryGolden fail with 404).

Bump 0.15.0 -> 0.15.1.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 16:05:00 +02:00
egutierrez e48b092135 Merge branch 'issue/names-bot-provisioning'
Integra GET /api/directory (endpoint->handle resolution) y el provisioning
one-command de bots (membershipd bot add).
2026-06-14 15:39:55 +02:00
egutierrez 0b39e86ed6 docs(unibus): bump to 0.15.0; document directory + bot provisioning
Add the 'Directorio de nombres (endpoint -> handle)' and 'Provisioning de bots /
unibots' sections with an end-to-end snippet, and a capability growth log entry
for v0.15.0.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 15:32:00 +02:00
egutierrez 669bad52af feat(membershipd): one-command bot provisioning (bot add)
Add `membershipd bot add --handle <name> --out <path> [--role] [--store]` to
provision a bus identity for an automated process in a single step: mint a fresh
Ed25519+X25519 identity (cs.GenerateIdentity, the same derivation worker/chat
use), register its signing key in the allowlist, and write the credentials to a
0600 file. The file is the canonical identity format read by client.LoadIdentity,
so a worker/clientcheck binary pointed at --out connects as the new user with no
extra conversion. Shares the sqlite/kv store plumbing with `user add`.

New exported pkg/client.WriteNewIdentity writes an identity in that format but
refuses to overwrite an existing file (never silently clobber private keys).

provisionBot ordering guarantees no half-provisioned bot: refuse an existing
--out before touching the store, register (an already-registered key is a clear
error, not a panic), then write credentials. Tests cover the golden path
(register + 0600 file + LoadIdentity round-trip), default role, the
already-registered error path (no file written), and the out-exists error path
(no orphan user).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 15:30:17 +02:00
egutierrez 2ba40701b2 feat(membership): add GET /api/directory for endpoint->handle resolution
Authenticated bus users (member or admin) can now map a sender's endpoint id
back to a readable handle. The endpoint is derived server-side from each user's
sign_pub with frame.EndpointID (base64url(sha256(signPub)), unpadded), matching
the bus's own construction byte-for-byte. Only active users are listed; under
enforce the existing auth middleware rejects an unauthenticated caller with 401.

Tests cover the golden path (two users -> 200 with handles + endpoints), the
auth contract (unsigned -> 401), revoked-user exclusion, and endpoint parity
against the cross-language vector from cmd/busvectors.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 15:27:38 +02:00
egutierrez 363aa97def Merge branch 'quick/proxy-ready' 2026-06-14 13:49:24 +02:00
egutierrez e3f40913bc chore(deploy): version the same-origin Caddy config for uniweb
Capture the reverse-proxy vhost that fronts the browser-native uniweb
client on magnus (chat-<hash>.organic-machine.com): the SPA at /, the
signed control plane under /api (prefix stripped so request signatures
verify), and the NATS-over-WebSocket data plane under /nats. One origin
means no CORS and keeps the cluster node IPs hidden behind the proxy.

Self-contained fragment (includes the shared security_headers snippet) so
it validates with `caddy validate` on its own; the other vhosts on magnus
carry basic-auth secrets and are intentionally left out of git. Documents
the matching membershipd flags this config requires (--cors-origins with
the same-origin host, --trusted-proxies naming the Caddy node).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 12:43:15 +02:00
egutierrez 0b96c114b6 feat(membership): trust reverse-proxy forwarded client IP for rate limit
The per-IP rate limiter keys on the transport RemoteAddr. Behind the
same-origin Caddy proxy that fronts the control plane, every request
arrives with the proxy's single IP, which collapses the limiter into one
bucket shared by the whole world — a flood from one client throttles all
of them.

Add an opt-in `--trusted-proxies` flag (comma-separated IPs/CIDRs). When
the immediate peer is one of the named proxies, clientIP now believes its
X-Forwarded-For (read right-to-left, skipping trusted hops) or X-Real-IP
and keys on the real client. A direct, non-trusted peer's forwarding
headers are ignored entirely, so this opens no quota-fanning hole: an
attacker connecting straight to the public :8470 cannot spoof a key. The
zero value (no flag) preserves the prior RemoteAddr-only behavior exactly.

Covered by ratelimit_proxy_test.go: trusted vs untrusted peers, XFF
right-to-left precedence, client-prepended forgery, X-Real-IP fallback,
and rejection of malformed proxy entries.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-14 12:29:57 +02:00
egutierrez 294905984c fix(membership): allow X-Unibus-* auth headers in CORS preflight
A browser signs every control-plane request with X-Unibus-Pub/Ts/Nonce/Sig
(busauth.signedHeaders). The CORS Allow-Headers only listed Content-Type and
Authorization, so the browser's preflight rejected the real request and the SPA
failed with 'Failed to fetch' on the first authenticated call (listRooms). Add the
four X-Unibus-* headers to Access-Control-Allow-Headers.

This was invisible to the Node smoke (fetch in Node does no CORS preflight); only a
real browser surfaced it. Verified live: enmanuel logs into uniweb against the
cluster and lists rooms. Regression test asserts the header is present.
2026-06-14 12:12:20 +02:00
egutierrez feb917fc6a feat(cluster): deploy browser WebSocket + CORS to the 3-node cluster
Roll the --ws-port + --cors-origins flags (issue uniweb/0001) out to the unibus
cluster so the browser-native uniweb client can reach the data plane (nats.ws)
and the control plane (CORS) on every node. The WS reuses the data-plane TLS
(wss://) and the same origin allowlist.

Per-node WS port override (WS_PORT_<NAME>): magnus runs unibus_admin on
127.0.0.1:8480, so the bus WS binds 8485 there to avoid a crash-loop; homer and
datardos keep 8480. deploy-cluster.sh also gains DEPLOY_ONLY=<name> for rolling
one node at a time. Rolled out and verified 2026-06-13: all three nodes healthy,
WS reachable, CORS 204, cluster quorum (R3) intact throughout.
2026-06-13 23:23:52 +02:00
egutierrez c0216de766 feat(membershipd): --ws-port wires the embedded NATS WebSocket listener
Phase 0 left the WebsocketConfig field unwired; add --ws-port so membershipd can
actually expose the browser data-plane transport. It reuses the data-plane TLS
(wss:// when TLS is on, ws:// for a loopback dev stack) and the same
--cors-origins allowlist that gates the control plane, so one flag pair opens
both planes to the browser-native uniweb client (issue uniweb/0001).
2026-06-13 23:05:33 +02:00
egutierrez 0088fb946b feat(busvectors): add nkey + signed control-request vectors
Extend the cross-language vectors with the NATS user nkey derived from the
Ed25519 public key, and a signed control-plane request (CanonicalRequest +
Ed25519 signature). These let the TypeScript busauth port verify it authenticates
on both planes exactly like the Go client (issue uniweb/0001, Phase 1).
2026-06-13 22:49:20 +02:00
egutierrez e058b324f4 Merge branch 'quick/0001-ws-cors-prep' 2026-06-13 22:21:51 +02:00
egutierrez a5086ecd18 chore: bump unibus to 0.14.0 (browser-native client prep, Phase 0) 2026-06-13 22:21:51 +02:00
egutierrez 8a51c5cc1f feat(busvectors): deterministic cross-language test vectors
Add cmd/busvectors, a generator that emits stable JSON test vectors for the bus
protocol and its E2E crypto (endpoint id, Ed25519 sign, ChaCha20-Poly1305 AEAD
with a fixed nonce, sealed-box of a room key, and canonical Frame wire bytes +
SigningBytes). It uses the same registry crypto (functions/cybersecurity) the
bus uses, so the vectors are the contract the TypeScript port must match
byte-for-byte (issue uniweb/0001, Phase 0).

Regenerate with: go run ./cmd/busvectors > ../uniweb/web/src/bus/testdata/vectors.json
2026-06-13 22:21:32 +02:00
egutierrez ec8d34aaa1 feat(membership): opt-in CORS allowlist for the browser-native client
Add Server.AllowedOrigins and an applyCORS step at the top of ServeHTTP so a
browser SPA (uniweb) can call the control plane cross-origin: an allow-listed
Origin gets the Access-Control-Allow-* headers, and a preflight (OPTIONS) is
answered 204 before the rate limiter and auth ever run. A disallowed or missing
origin gets no headers (preflight 403), so the browser blocks the request.

Wire it through membershipd's --cors-origins flag (comma list, reusing
splitRoutes as a generic parser). Empty allowlist = CORS off, no headers
emitted, behavior identical to before: native Go/Kotlin clients send no Origin
and are unaffected. Opt-in per deployment (issue uniweb/0001, Phase 0).

Tests: preflight allow/deny, header on the real response, CORS-off default, and
no-Origin native client unaffected.
2026-06-13 22:17:44 +02:00
egutierrez 36f4ba0eaf feat(embeddednats): optional WebSocket listener for browser clients
Add WebsocketConfig to ServerConfig so the embedded nats-server can expose an
additional WebSocket port (nats.ws) alongside the TCP data plane. This lets a
browser SPA speak the NATS protocol directly, the way native TCP peers (Go,
Kotlin/android) already do — the first enabler for uniweb becoming a
browser-native client with no Go gateway (issue uniweb/0001, Phase 0).

The client authenticator applies to WebSocket connections too, so this adds a
transport, not a trust bypass. Plain ws:// is used only without TLS (loopback
dev); a certificate yields wss://. An empty AllowedOrigins enforces same-origin.
Nil WebsocketConfig keeps the server TCP-only, so existing single-node and
cluster deployments are unchanged.

Tests: WebSocket listener opens and completes the upgrade handshake (101); no
listener opens when WebsocketConfig is nil.
2026-06-13 22:11:39 +02:00
25 changed files with 2754 additions and 15 deletions
+94 -1
View File
@@ -2,7 +2,7 @@
name: unibus name: unibus
lang: go lang: go
domain: infra domain: infra
version: 0.13.0 version: 0.16.0
description: "Bus de mensajería unificado sobre NATS+JetStream con cifrado E2E por room (megolm/olm reducido): service de membresía/claves, librería cliente y peers demo." description: "Bus de mensajería unificado sobre NATS+JetStream con cifrado E2E por room (megolm/olm reducido): service de membresía/claves, librería cliente y peers demo."
tags: [service, messaging, nats, e2e] tags: [service, messaging, nats, e2e]
uses_functions: uses_functions:
@@ -158,6 +158,62 @@ Para apuntar a un NATS externo en producción: `--nats-url nats://host:4222` en
`cybersecurity` del registry compila limpio con `CGO_ENABLED=0`. NO requiere `cybersecurity` del registry compila limpio con `CGO_ENABLED=0`. NO requiere
`fts5` ni `gcc`. `fts5` ni `gcc`.
## Directorio de nombres (endpoint → handle)
Cada frame del bus lleva el **endpoint id** del remitente
(`base64url(sha256(signPub))`, sin padding — `frame.EndpointID`), no un nombre
legible. Para que un cliente muestre nombres en vez de hashes, el control-plane
expone la ruta del directorio. La SPA la llama como `GET /api/directory`, pero
Caddy hace `handle_path /api/*` y **stripea `/api`** antes de reenviar a
`membershipd`, así que el servidor la registra (como todas las rutas del
control-plane) SIN el prefijo: `GET /directory`:
- **Auth:** el mismo middleware de firma que el resto del control-plane
(cabeceras `X-Unibus-Pub/Ts/Nonce/Sig` sobre `CanonicalRequest`). NO es
admin-only: cualquier usuario activo del bus (member o admin) puede leerlo. En
modo `enforce`, una request sin firmar recibe 401 antes de llegar al handler.
- **Respuesta** `{ "members": [ { "sign_pub", "endpoint", "handle", "role" } ] }`,
solo usuarios `status=active`. El `endpoint` lo computa el servidor desde el
`sign_pub` con la misma derivación que el bus, así que casa byte a byte con el
sender id que el cliente ya tiene en cada mensaje.
- CORS: cubierto por la allowlist `--cors-origins` existente (mismas cabeceras
que el resto de rutas, sin caso especial).
## Provisioning de bots / unibots
Dar de alta una identidad para un proceso automatizado es **un solo comando**.
Antes había que derivar un keypair a mano y pasar el `sign_pub` a `user add`;
ahora `bot add` lo hace todo: mintea una identidad de bus fresca (Ed25519 +
X25519, la misma derivación `cs.GenerateIdentity` que usan `worker`/`chat`),
registra su `sign_pub` en el allowlist con `handle` y `role`, y escribe las
credenciales a un fichero 0600 que el proceso lee para conectar.
```bash
# 1. Provisionar el bot (store sqlite local; usa --store kv contra un cluster vivo).
membershipd bot add --handle notifier --out ./local_files/notifier.id
# provisioned bot "notifier" role=member
# sign_pub: 97d5a903...b1d4
# endpoint: HU85l2onjrK4EoTLoBfJVkGEXMw9LAjNEjPWiDS8YwM
# credentials: ./local_files/notifier.id (0600)
# 2. El proceso arranca como ese usuario leyendo el --out (formato canónico
# pkg/client.LoadIdentity, sin conversión): el worker demo lo consume directo.
worker --id-file ./local_files/notifier.id --nats-url nats://127.0.0.1:4250 \
--ctrl-url http://127.0.0.1:8470
# 3. (opcional) Verlo en el directorio / en user list.
membershipd user list
```
Las credenciales (`--out`) quedan en el fichero indicado, con permisos 0600. Es
el secreto del bot: contiene las claves privadas, trátalo como una clave SSH
(ver Gotcha "Identidad = secreto crítico"). `bot add` rehúsa sobrescribir un
`--out` existente, y registra al usuario ANTES de escribir el fichero, de modo
que un fallo nunca deja un bot a medias.
Flags: `--handle` y `--out` obligatorios; `--role admin|member` (default member);
`--store sqlite|kv` y el resto de flags de conexión idénticos a `user add`.
## Convención de subjects ## Convención de subjects
``` ```
@@ -169,6 +225,43 @@ agent.<nombre>.{in,out} inbox/outbox de agente LLM (agent.scout.in)
## Capability growth log ## Capability growth log
- v0.16.0 (2026-06-14) — feat: el server asegura el stream JetStream de las rooms
persist + `GET /rooms/{id}/history` para que clientes sin JetStream (uniweb) lean
el histórico. (1) `handleCreateRoom` crea (idempotente, `CreateOrUpdateStream`) el
stream durable `UNIBUS_<roomID>` de una room persist ANTES de responder, así su
subject se captura desde el minuto cero venga el mensaje de un cliente Go o de un
cliente browser que solo habla core NATS (antes el stream lo creaba solo el cliente
Go, así que los mensajes de uniweb se perdían). (2) Nuevo endpoint member-only
`GET /rooms/{id}/history?limit=N` (default 200, cap 1000): lee el stream
server-side y devuelve `{messages:[<base64-std del frame marshalado>]}` en orden
oldest→newest; el server jamás descifra (relay del ciphertext E2E). Backfill de
rooms persist existentes: lazy-ensure del stream en el endpoint (empiezan a
capturar desde ahora; los mensajes previos al stream no son recuperables). El
control plane abre ahora su propio contexto JetStream también en single-node
embebido. Todo aditivo; build/vet/test verdes.
- v0.15.1 (2026-06-14) — fix: la ruta del directorio se registraba con prefijo /api y Caddy lo stripeaba (404 en prod); corregida a /directory.
- v0.15.0 (2026-06-14) — nombres legibles + provisioning de bots de un comando.
(1) Nuevo `GET /api/directory` en el control-plane: cualquier usuario activo del
bus (member o admin), autenticado con la misma firma Ed25519 que el resto de
rutas, resuelve endpoint id → handle. Devuelve `{members:[{sign_pub, endpoint,
handle, role}]}` solo de usuarios activos; el endpoint lo deriva el servidor con
`frame.EndpointID`, casando byte a byte con el sender id de cada frame (paridad
verificada contra el vector de `cmd/busvectors`). (2) Nuevo `membershipd bot add
--handle <name> --out <path> [--role] [--store]`: mintea identidad, la registra en
el allowlist y escribe credenciales 0600 en formato `client.LoadIdentity`, de modo
que un proceso (worker/clientcheck) conecta como ese usuario sin pasos manuales.
Nuevo helper exportado `pkg/client.WriteNewIdentity` (no sobrescribe ficheros
existentes). Todo aditivo; build/vet/test verdes.
- v0.14.0 (2026-06-13) — prep para el cliente browser-nativo `uniweb` (issue
uniweb/0001, Fase 0), todo aditivo y opt-in: (1) el nats-server embebido puede
exponer un listener WebSocket (`WebsocketConfig`) para que un navegador hable el
protocolo NATS via `nats.ws`, igual que los peers TCP nativos; el authenticator
nkey aplica también al WebSocket. (2) El control-plane (`membershipd`) gana una
allowlist CORS opt-in (`--cors-origins`) para aceptar llamadas cross-origin del
navegador; vacía = CORS off, sin cambios para clientes nativos. (3) `cmd/busvectors`
genera vectores de test deterministas (endpoint id, firma Ed25519, AEAD
ChaCha20-Poly1305, sealed-box, wire del Frame) como contrato de paridad para el
port TypeScript. Peers Go/Kotlin existentes sin cambios; build/vet/test verdes.
- v0.13.0 (2026-06-13) — el frontend web se separa a su propia app `uniweb` - v0.13.0 (2026-06-13) — el frontend web se separa a su propia app `uniweb`
(`projects/message_bus/apps/uniweb`, sub-repo Gitea propio). unibus deja de (`projects/message_bus/apps/uniweb`, sub-repo Gitea propio). unibus deja de
contener la SPA (`web/`) y el gateway web (`cmd/webgw`): ahora es estrictamente contener la SPA (`web/`) y el gateway web (`cmd/webgw`): ahora es estrictamente
BIN
View File
Binary file not shown.
+281
View File
@@ -0,0 +1,281 @@
// Command busvectors emits deterministic cross-language test vectors for the bus
// protocol and its end-to-end crypto. The browser-native client (uniweb) ports the
// protocol to TypeScript; these vectors are the contract that proves the port is
// byte-for-byte compatible with this Go reference implementation (issue
// uniweb/0001, Phase 0).
//
// Every input is fixed (hardcoded key material and messages) so the output is
// stable across runs and can be committed as a golden file. The crypto primitives
// are the SAME registry functions the bus uses (functions/cybersecurity), so the
// vectors exercise the real path, not a test-only reimplementation.
//
// Coverage:
// - endpoint_id : EndpointID(signPub) = base64url(sha256(signPub))
// - sign : Ed25519 signature over a fixed message (deterministic)
// - aead : ChaCha20-Poly1305 seal with a FIXED nonce (deterministic, so
// the TS port must reproduce the same ciphertext AND open it)
// - keybox : sealed-box (X25519) of a room key for a recipient; the TS port
// must OPEN it (the ephemeral sender key is random, so only the
// open direction is a stable vector — the TS->Go seal direction
// is covered by the live E2E test in Phase 3)
// - frame : canonical JSON wire bytes of a Frame, and its SigningBytes
//
// Usage:
//
// go run ./cmd/busvectors > ../uniweb/web/src/bus/testdata/vectors.json
package main
import (
"crypto/ed25519"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"os"
cs "fn-registry/functions/cybersecurity"
"github.com/enmanuel/unibus/pkg/busauth"
"github.com/enmanuel/unibus/pkg/frame"
"github.com/enmanuel/unibus/pkg/membership"
"golang.org/x/crypto/chacha20poly1305"
"golang.org/x/crypto/curve25519"
)
// Fixed key material. The bytes are arbitrary but stable: the point is a golden
// file, not secrecy (these are test vectors, never real identities).
var (
signSeed = mustHex("000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f")
kexPriv = mustHex("202122232425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f")
recipientKexPriv = mustHex("404142434445464748494a4b4c4d4e4f505152535455565758595a5b5c5d5e5f")
aeadKey = mustHex("606162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f")
aeadNonce = mustHex("808182838485868788898a8b") // 12 bytes (ChaCha20-Poly1305 IETF)
roomKey = mustHex("a0a1a2a3a4a5a6a7a8a9aaabacadaeafb0b1b2b3b4b5b6b7b8b9babbbcbdbebf")
signMessage = []byte("unibus parity vector message")
aeadAAD = []byte("unibus-room-42")
aeadPlaintext = []byte("hello from the bus")
)
// vectors is the JSON document consumed by the TypeScript parity tests. Every field
// is hex except the frame wire bytes, which are base64 (the frame is JSON, so the
// TS side compares the exact UTF-8 bytes).
type vectors struct {
Note string `json:"note"`
Endpoint endpointVector `json:"endpoint_id"`
Nkey nkeyVector `json:"nkey"`
Sign signVector `json:"sign"`
AEAD aeadVector `json:"aead"`
KeyBox keyboxVector `json:"keybox"`
Frame frameVector `json:"frame"`
CtrlReq controlReqVector `json:"control_request"`
}
type endpointVector struct {
SignPubHex string `json:"sign_pub_hex"`
EndpointID string `json:"endpoint_id"` // base64url(sha256(sign_pub)), unpadded
}
type nkeyVector struct {
SignPubHex string `json:"sign_pub_hex"`
NkeyPublic string `json:"nkey_public"` // NATS user nkey ("U...") from the Ed25519 pubkey
}
type controlReqVector struct {
Method string `json:"method"`
Path string `json:"path"`
Ts string `json:"ts"`
Nonce string `json:"nonce"`
BodyHex string `json:"body_hex"` // raw request body (empty for GET)
CanonicalHex string `json:"canonical_hex"` // bytes that get signed
SigHex string `json:"sig_hex"` // Ed25519 over canonical, by the signer below
SignPrivHex string `json:"sign_priv_hex"`
}
type signVector struct {
SignPrivHex string `json:"sign_priv_hex"`
SignPubHex string `json:"sign_pub_hex"`
MessageHex string `json:"message_hex"`
SigHex string `json:"sig_hex"`
}
type aeadVector struct {
KeyHex string `json:"key_hex"`
NonceHex string `json:"nonce_hex"`
AADHex string `json:"aad_hex"`
PlaintextHex string `json:"plaintext_hex"`
CiphertextHex string `json:"ciphertext_hex"` // includes the 16-byte Poly1305 tag
}
type keyboxVector struct {
RecipientKexPubHex string `json:"recipient_kex_pub_hex"`
RecipientKexPrivHex string `json:"recipient_kex_priv_hex"`
SecretHex string `json:"secret_hex"`
SealedHex string `json:"sealed_hex"`
}
type frameVector struct {
// The source fields, so the TS side can build the same Frame and compare.
Type int `json:"type"`
Subject string `json:"subject"`
Sender string `json:"sender"`
MsgID string `json:"msg_id"`
Epoch int `json:"epoch"`
NonceHex string `json:"nonce_hex"`
PayloadHex string `json:"payload_hex"`
WireB64 string `json:"wire_b64"` // base64(Marshal()) — full frame incl. sig
SigningB64 string `json:"signing_bytes_b64"` // base64(SigningBytes()) — what gets signed
SigHex string `json:"sig_hex"` // Ed25519 over SigningBytes
}
func main() {
if err := run(os.Stdout); err != nil {
fmt.Fprintln(os.Stderr, "busvectors:", err)
os.Exit(1)
}
}
func run(out *os.File) error {
// Identity from the fixed seed: Go's ed25519 private key layout is seed||pub, the
// same 64-byte layout cs.Identity and the TS wallet use.
signPriv := ed25519.NewKeyFromSeed(signSeed)
signPub := signPriv.Public().(ed25519.PublicKey)
// X25519 public keys from the fixed private scalars (curve25519 clamps internally,
// matching @noble/curves x25519.getPublicKey).
kexPub, err := curve25519.X25519(kexPriv, curve25519.Basepoint)
if err != nil {
return fmt.Errorf("kex pub: %w", err)
}
recipientKexPub, err := curve25519.X25519(recipientKexPriv, curve25519.Basepoint)
if err != nil {
return fmt.Errorf("recipient kex pub: %w", err)
}
// AEAD with a FIXED nonce so the vector is deterministic. This is the same cipher
// (ChaCha20-Poly1305 IETF, 12-byte nonce) that cs.SealAEAD uses; we set the nonce
// explicitly only to make the vector reproducible. OpenAEAD verifies round-trip.
aead, err := chacha20poly1305.New(aeadKey)
if err != nil {
return fmt.Errorf("aead cipher: %w", err)
}
ciphertext := aead.Seal(nil, aeadNonce, aeadPlaintext, aeadAAD)
if _, err := cs.OpenAEAD(aeadKey, aeadNonce, ciphertext, aeadAAD); err != nil {
return fmt.Errorf("aead self-check: %w", err)
}
// Sealed box of the room key for the recipient. The sender's ephemeral key is
// random (anonymous sealed box), so SealedHex changes per run; the stable, useful
// assertion for the TS port is that OpenKeyBox recovers the secret, which we
// self-check here. The TS test opens SealedHex and compares to SecretHex.
sealed, err := cs.SealKeyBox(recipientKexPub, roomKey)
if err != nil {
return fmt.Errorf("seal keybox: %w", err)
}
if got, err := cs.OpenKeyBox(recipientKexPub, recipientKexPriv, sealed); err != nil || hex.EncodeToString(got) != hex.EncodeToString(roomKey) {
return fmt.Errorf("keybox self-check failed: %v", err)
}
// A representative encrypted-room frame, signed end-to-end.
f := frame.Frame{
Type: frame.PUB,
Subject: "room.parity",
Sender: frame.EndpointID(signPub),
MsgID: "01HZY0VECTORFIXEDULID0001",
Epoch: 1,
Nonce: aeadNonce,
Payload: ciphertext,
}
f.Sig = ed25519.Sign(signPriv, f.SigningBytes())
wire, err := f.Marshal()
if err != nil {
return fmt.Errorf("marshal frame: %w", err)
}
// NATS user nkey derived from the Ed25519 public key (the browser must produce
// the same "U..." string to authenticate on the data plane).
nkeyPub, err := busauth.NkeyPublicFromSignPub(signPub)
if err != nil {
return fmt.Errorf("nkey public: %w", err)
}
// A signed control-plane request vector: the browser signs CanonicalRequest the
// same way to authenticate every HTTP call to membershipd. A POST with a body
// exercises the sha256(body) term.
const ctrlMethod = "POST"
const ctrlPath = "/rooms"
const ctrlTs = "1700000000"
const ctrlNonce = "Zm9vYmFyMTIzNDU2Nzg5MA=="
ctrlBody := []byte(`{"subject":"room.parity"}`)
canonical := membership.CanonicalRequest(ctrlMethod, ctrlPath, ctrlTs, ctrlNonce, ctrlBody)
ctrlSig := ed25519.Sign(signPriv, canonical)
v := vectors{
Note: "Deterministic cross-language vectors for the unibus protocol. Generated by " +
"cmd/busvectors in the unibus repo; regenerate with `go run ./cmd/busvectors`. " +
"sealed_hex varies per run (anonymous sealed box); assert via OpenKeyBox.",
Endpoint: endpointVector{
SignPubHex: hex.EncodeToString(signPub),
EndpointID: frame.EndpointID(signPub),
},
Nkey: nkeyVector{
SignPubHex: hex.EncodeToString(signPub),
NkeyPublic: nkeyPub,
},
Sign: signVector{
SignPrivHex: hex.EncodeToString(signPriv),
SignPubHex: hex.EncodeToString(signPub),
MessageHex: hex.EncodeToString(signMessage),
SigHex: hex.EncodeToString(ed25519.Sign(signPriv, signMessage)),
},
AEAD: aeadVector{
KeyHex: hex.EncodeToString(aeadKey),
NonceHex: hex.EncodeToString(aeadNonce),
AADHex: hex.EncodeToString(aeadAAD),
PlaintextHex: hex.EncodeToString(aeadPlaintext),
CiphertextHex: hex.EncodeToString(ciphertext),
},
KeyBox: keyboxVector{
RecipientKexPubHex: hex.EncodeToString(recipientKexPub),
RecipientKexPrivHex: hex.EncodeToString(recipientKexPriv),
SecretHex: hex.EncodeToString(roomKey),
SealedHex: hex.EncodeToString(sealed),
},
Frame: frameVector{
Type: int(f.Type),
Subject: f.Subject,
Sender: f.Sender,
MsgID: f.MsgID,
Epoch: f.Epoch,
NonceHex: hex.EncodeToString(f.Nonce),
PayloadHex: hex.EncodeToString(f.Payload),
WireB64: base64.StdEncoding.EncodeToString(wire),
SigningB64: base64.StdEncoding.EncodeToString(f.SigningBytes()),
SigHex: hex.EncodeToString(f.Sig),
},
CtrlReq: controlReqVector{
Method: ctrlMethod,
Path: ctrlPath,
Ts: ctrlTs,
Nonce: ctrlNonce,
BodyHex: hex.EncodeToString(ctrlBody),
CanonicalHex: hex.EncodeToString(canonical),
SigHex: hex.EncodeToString(ctrlSig),
SignPrivHex: hex.EncodeToString(signPriv),
},
// kexPub is unused in a vector field today but derived above to validate the
// scalar; reference it so the intent is documented.
}
_ = kexPub
enc := json.NewEncoder(out)
enc.SetIndent("", " ")
return enc.Encode(v)
}
func mustHex(s string) []byte {
b, err := hex.DecodeString(s)
if err != nil {
panic("busvectors: bad fixed hex: " + s)
}
return b
}
+159
View File
@@ -0,0 +1,159 @@
package main
import (
"encoding/hex"
"errors"
"flag"
"fmt"
"os"
cs "fn-registry/functions/cybersecurity"
"github.com/enmanuel/unibus/pkg/client"
"github.com/enmanuel/unibus/pkg/frame"
"github.com/enmanuel/unibus/pkg/membership"
)
// runBotCLI implements `membershipd bot add ...`, one-command provisioning of a
// bus identity for an automated process. Where `user add` requires the operator
// to derive a keypair by hand and pass the public key, `bot add` mints the
// identity, registers its signing key in the allowlist, AND writes the bot's
// credentials to a 0600 file the process reads to connect — no manual key
// derivation, no second step. It shares the SQLite/KV store plumbing with the
// user CLI, so `--store kv` provisions against a live cluster the same way.
//
// Like the user CLI it never returns: it exits non-zero on error so it composes
// in shell scripts and systemd ExecStartPre hooks.
func runBotCLI(args []string) {
if len(args) == 0 {
botUsage()
os.Exit(2)
}
sub, rest := args[0], args[1:]
switch sub {
case "add":
botAdd(rest)
case "-h", "--help", "help":
botUsage()
os.Exit(0)
default:
fmt.Fprintf(os.Stderr, "membershipd bot: unknown subcommand %q\n\n", sub)
botUsage()
os.Exit(2)
}
}
func botUsage() {
fmt.Fprint(os.Stderr, `usage: membershipd bot add [flags]
Provision a bus identity for an automated process (a "unibot") in one command:
mint a fresh Ed25519+X25519 identity, register its signing key in the allowlist,
and write the credentials to a 0600 file the process loads to connect.
required flags:
--handle <name> human-readable name for the bot (shown in the directory)
--out <path> where to write the bot credentials (refused if it exists)
optional flags:
--role <role> admin or member (default member)
--store <kind> sqlite (local DB, default) | kv (the live cluster's allowlist)
--db <path> SQLite database path (--store sqlite; default ./local_files/unibus.db)
--store kv flags (defaults assume an on-node invocation):
--nats-url <url> cluster NATS (default nats://127.0.0.1:4250)
--internal-id-file <path> persisted internal service identity (default /opt/unibus/secrets/internal.id)
--ca <path> CA cert pinning the data-plane TLS (default /opt/unibus/tls/ca.crt)
--kv-replicas <n> KV replication factor, match the cluster (default 3)
examples:
membershipd bot add --handle notifier --out ./local_files/notifier.id
membershipd bot add --store kv --handle relay --role member --out /opt/unibus/secrets/relay.id
The --out file is the canonical identity format read by the worker/clientcheck
clients (pkg/client.LoadIdentity), so the provisioned bot connects with no extra
conversion: point the process at it (e.g. worker --id-file <path>) and it joins
the bus as this user.
`)
}
func botAdd(args []string) {
fs := flag.NewFlagSet("bot add", flag.ExitOnError)
handle := fs.String("handle", "", "human-readable bot name (required)")
role := fs.String("role", membership.RoleMember, "role: admin or member")
out := fs.String("out", "", "path to write the bot credentials, 0600 (required)")
dbPath := fs.String("db", defaultDBPath, "SQLite database path")
kf := registerKVFlags(fs)
_ = fs.Parse(args)
if *handle == "" || *out == "" {
fmt.Fprintln(os.Stderr, "membershipd bot add: --handle and --out are required")
os.Exit(2)
}
store, kv, closeStore := resolveStore("bot add", kf, *dbPath)
defer closeStore()
signPubHex, endpoint, err := provisionBot(store, *handle, *role, *out)
if err != nil {
fmt.Fprintf(os.Stderr, "membershipd bot add: %v\n", err)
os.Exit(1)
}
fmt.Printf("provisioned bot %q role=%s\n", *handle, *role)
fmt.Printf(" sign_pub: %s\n", signPubHex)
fmt.Printf(" endpoint: %s\n", endpoint)
fmt.Printf(" credentials: %s (0600)\n", *out)
if kv != nil {
reportKVReplication(kv.js)
}
}
// provisionBot mints a fresh bus identity and provisions it. It is the generating
// half; provisionBotWithIdentity does the registration + persistence so a test can
// inject a known identity (e.g. to exercise the already-registered error path).
func provisionBot(store membership.Store, handle, role, out string) (signPubHex, endpoint string, err error) {
id, err := cs.GenerateIdentity()
if err != nil {
return "", "", fmt.Errorf("generate bot identity: %w", err)
}
return provisionBotWithIdentity(store, id, handle, role, out)
}
// provisionBotWithIdentity registers id's signing key under handle/role and writes
// id's credentials to out. It returns the lowercase-hex signing key and the
// derived endpoint id.
//
// Ordering is deliberate so a failure never leaves a half-provisioned bot:
// 1. refuse if out already exists, BEFORE the store is touched (no orphan user);
// 2. register the user — an already-registered key is a clear error, not a panic;
// 3. only then write the 0600 credentials file.
//
// A write failure after a successful register is reported with the registered key
// so the operator can revoke it; this is the one residual non-atomic seam (a
// local admin command, acceptable per KISS).
func provisionBotWithIdentity(store membership.Store, id cs.Identity, handle, role, out string) (signPubHex, endpoint string, err error) {
if handle == "" || out == "" {
return "", "", fmt.Errorf("handle and out are required")
}
if role == "" {
role = membership.RoleMember
}
if _, statErr := os.Stat(out); statErr == nil {
return "", "", fmt.Errorf("out file %q already exists; refusing to overwrite bot credentials", out)
} else if !os.IsNotExist(statErr) {
return "", "", fmt.Errorf("stat out %q: %w", out, statErr)
}
signPubHex = hex.EncodeToString(id.SignPub)
endpoint = frame.EndpointID(id.SignPub)
if err := store.AddUser(signPubHex, handle, role); err != nil {
if errors.Is(err, membership.ErrUserExists) {
return "", "", fmt.Errorf("sign_pub %s already registered; revoke it first to replace", signPubHex)
}
return "", "", fmt.Errorf("register bot user: %w", err)
}
if err := client.WriteNewIdentity(out, id); err != nil {
return "", "", fmt.Errorf("write bot credentials to %q (user %s WAS registered — revoke it to retry): %w", out, signPubHex, err)
}
return signPubHex, endpoint, nil
}
+149
View File
@@ -0,0 +1,149 @@
package main
import (
"encoding/hex"
"os"
"path/filepath"
"testing"
cs "fn-registry/functions/cybersecurity"
"github.com/enmanuel/unibus/pkg/client"
"github.com/enmanuel/unibus/pkg/frame"
"github.com/enmanuel/unibus/pkg/membership"
)
// openTestStore opens a fresh SQLite membership store in a temp dir.
func openTestStore(t *testing.T) membership.Store {
t.Helper()
store, err := membership.Open(filepath.Join(t.TempDir(), "unibus.db"))
if err != nil {
t.Fatalf("open store: %v", err)
}
t.Cleanup(func() { store.Close() })
return store
}
// TestProvisionBotGolden is the happy path: provisioning a bot registers it in the
// allowlist with the right handle and role, AND writes a 0600 credentials file
// that LoadIdentity reconstructs into the same identity — so a worker/clientcheck
// binary pointed at the file connects as exactly this user with no extra step.
func TestProvisionBotGolden(t *testing.T) {
store := openTestStore(t)
out := filepath.Join(t.TempDir(), "notifier.id")
signPubHex, endpoint, err := provisionBot(store, "notifier", membership.RoleMember, out)
if err != nil {
t.Fatalf("provisionBot: %v", err)
}
// Registered in the allowlist with the right handle/role/status.
u, err := store.GetUser(signPubHex)
if err != nil {
t.Fatalf("get provisioned user: %v", err)
}
if u.Handle != "notifier" || u.Role != membership.RoleMember || u.Status != membership.StatusActive {
t.Fatalf("provisioned user row wrong: %+v", u)
}
// And it shows up in user list (the `user list` surface).
users, err := store.ListUsers()
if err != nil {
t.Fatalf("list users: %v", err)
}
found := false
for _, x := range users {
if x.SignPub == signPubHex {
found = true
}
}
if !found {
t.Fatalf("provisioned bot missing from user list: %+v", users)
}
// Credentials file exists, is 0600, and round-trips through LoadIdentity to the
// same signing key + endpoint (no-friction contract).
info, err := os.Stat(out)
if err != nil {
t.Fatalf("stat out file: %v", err)
}
if perm := info.Mode().Perm(); perm != 0o600 {
t.Fatalf("out file perms = %o, want 600", perm)
}
id, err := client.LoadIdentity(out)
if err != nil {
t.Fatalf("LoadIdentity(out): %v", err)
}
if got := hex.EncodeToString(id.SignPub); got != signPubHex {
t.Fatalf("loaded sign_pub %q != provisioned %q", got, signPubHex)
}
if got := frame.EndpointID(id.SignPub); got != endpoint {
t.Fatalf("loaded endpoint %q != reported %q", got, endpoint)
}
}
// TestProvisionBotDefaultRole: an empty role defaults to member.
func TestProvisionBotDefaultRole(t *testing.T) {
store := openTestStore(t)
out := filepath.Join(t.TempDir(), "bot.id")
signPubHex, _, err := provisionBot(store, "defrole", "", out)
if err != nil {
t.Fatalf("provisionBot: %v", err)
}
u, err := store.GetUser(signPubHex)
if err != nil {
t.Fatalf("get user: %v", err)
}
if u.Role != membership.RoleMember {
t.Fatalf("empty role should default to member, got %q", u.Role)
}
}
// TestProvisionBotSignPubAlreadyRegistered is the error path: provisioning an
// identity whose signing key is already in the allowlist fails with a clear error
// (not a panic) AND does not write a credentials file (no half-provisioned bot).
func TestProvisionBotSignPubAlreadyRegistered(t *testing.T) {
store := openTestStore(t)
// Pre-register a key, then try to provision a bot with that SAME identity.
id, err := cs.GenerateIdentity()
if err != nil {
t.Fatalf("generate identity: %v", err)
}
signPubHex := hex.EncodeToString(id.SignPub)
if err := store.AddUser(signPubHex, "preexisting", membership.RoleMember); err != nil {
t.Fatalf("pre-register: %v", err)
}
out := filepath.Join(t.TempDir(), "dup.id")
_, _, err = provisionBotWithIdentity(store, id, "dupbot", membership.RoleMember, out)
if err == nil {
t.Fatalf("provisioning an already-registered key should error")
}
if _, statErr := os.Stat(out); !os.IsNotExist(statErr) {
t.Fatalf("credentials file must NOT be written on a duplicate-key failure (stat err = %v)", statErr)
}
}
// TestProvisionBotOutExists is the other error path: an existing --out file is
// refused BEFORE the store is mutated, so the run leaves no orphan user behind.
func TestProvisionBotOutExists(t *testing.T) {
store := openTestStore(t)
out := filepath.Join(t.TempDir(), "taken.id")
if err := os.WriteFile(out, []byte("preexisting credentials"), 0o600); err != nil {
t.Fatalf("seed out file: %v", err)
}
_, _, err := provisionBot(store, "clobber", membership.RoleMember, out)
if err == nil {
t.Fatalf("provisioning over an existing out file should error")
}
// The store must be untouched: no user was registered.
users, err := store.ListUsers()
if err != nil {
t.Fatalf("list users: %v", err)
}
if len(users) != 0 {
t.Fatalf("no user should be registered when out exists, got %+v", users)
}
}
+69 -3
View File
@@ -13,6 +13,7 @@ import (
"net/http" "net/http"
"os" "os"
"os/signal" "os/signal"
"strings"
"syscall" "syscall"
"time" "time"
@@ -46,6 +47,14 @@ func main() {
runMigrateCLI(os.Args[2:]) runMigrateCLI(os.Args[2:])
return return
} }
// `membershipd bot add ...` provisions a bus identity for an automated process
// in one command (mint identity + register + write 0600 credentials). It shares
// the same trusted-host model and store plumbing as the user CLI, so it is
// dispatched here before the server flag set parses os.Args.
if len(os.Args) > 1 && os.Args[1] == "bot" {
runBotCLI(os.Args[2:])
return
}
var ( var (
bind = flag.String("bind", "127.0.0.1", "network interface to bind the HTTP API and the embedded NATS to; use 0.0.0.0 to accept LAN/remote peers") bind = flag.String("bind", "127.0.0.1", "network interface to bind the HTTP API and the embedded NATS to; use 0.0.0.0 to accept LAN/remote peers")
@@ -54,8 +63,11 @@ func main() {
dbPath = flag.String("db", "./local_files/unibus.db", "SQLite database path") dbPath = flag.String("db", "./local_files/unibus.db", "SQLite database path")
storeDir = flag.String("store-dir", "./local_files/blobs", "blob store directory") storeDir = flag.String("store-dir", "./local_files/blobs", "blob store directory")
natsPort = flag.Int("nats-port", 4250, "embedded NATS listen port (when --nats-url empty)") natsPort = flag.Int("nats-port", 4250, "embedded NATS listen port (when --nats-url empty)")
wsPort = flag.Int("ws-port", 0, "WebSocket listen port for browser clients (nats.ws); 0 = disabled. Enables the browser-native uniweb client (issue uniweb/0001)")
natsStore = flag.String("nats-store", "./local_files/jetstream", "embedded JetStream store dir") natsStore = flag.String("nats-store", "./local_files/jetstream", "embedded JetStream store dir")
busAuth = flag.String("bus-auth", "off", "control-plane auth rollout: off|soft|enforce (feature flag bus-auth)") busAuth = flag.String("bus-auth", "off", "control-plane auth rollout: off|soft|enforce (feature flag bus-auth)")
corsOrigins = flag.String("cors-origins", "", "comma-separated CORS allowlist of browser origins permitted to call the control plane (e.g. http://localhost:5173,https://chat.example.com); empty = CORS off. Enables the browser-native uniweb client (issue uniweb/0001)")
trustedProxies = flag.String("trusted-proxies", "", "comma-separated IPs/CIDRs of reverse proxies whose X-Forwarded-For/X-Real-IP is trusted for the per-IP rate limit; empty = trust the direct connection only. Set to the same-origin proxy's address (e.g. the Caddy node) so the rate limit stays per-client behind the proxy")
tlsCert = flag.String("tls-cert", "", "PATH to the NATS server certificate (deploy/tls/server.crt); enables TLS on the embedded data plane") tlsCert = flag.String("tls-cert", "", "PATH to the NATS server certificate (deploy/tls/server.crt); enables TLS on the embedded data plane")
tlsKey = flag.String("tls-key", "", "path to the NATS server private key (deploy/tls/server.key); required with --tls-cert") tlsKey = flag.String("tls-key", "", "path to the NATS server private key (deploy/tls/server.key); required with --tls-cert")
// Cluster (issue 0003a): empty --cluster-name keeps the server standalone. // Cluster (issue 0003a): empty --cluster-name keeps the server standalone.
@@ -138,6 +150,16 @@ func main() {
decentralized := *storeBackend == "kv" decentralized := *storeBackend == "kv"
needJS := clustered || decentralized needJS := clustered || decentralized
enforce := authMode == membership.AuthEnforce enforce := authMode == membership.AuthEnforce
embedded := *natsURL == ""
// The control plane also needs a privileged JetStream client to OWN the durable
// per-room streams of persisted rooms (ensure the stream on room creation so the
// subject is captured from the first message — even from a JetStream-less browser
// client — and read it back for GET /rooms/{id}/history). The embedded NATS
// always ships JetStream, so open the client whenever we run embedded, even for a
// standalone SQLite node. For an EXTERNAL NATS we only reach for JetStream when a
// cluster/KV feature explicitly requires it (unchanged), so an operator-managed
// external deployment without those features behaves exactly as before.
openJS := needJS || embedded
// Internal service identity (issue 0006a): when the embedded data plane enforces // Internal service identity (issue 0006a): when the embedded data plane enforces
// auth, membershipd must still connect to its OWN server to manage JetStream. // auth, membershipd must still connect to its OWN server to manage JetStream.
@@ -147,7 +169,7 @@ func main() {
// the server is embedded), so a standalone or non-enforce node is unchanged. // the server is embedded), so a standalone or non-enforce node is unchanged.
var internalID cs.Identity var internalID cs.Identity
var internalPubHex string var internalPubHex string
if needJS && enforce && *natsURL == "" { if openJS && enforce && embedded {
if *internalIDFile != "" { if *internalIDFile != "" {
// Persisted identity: load it, generating + writing it (0600) on first // Persisted identity: load it, generating + writing it (0600) on first
// start. A stable internal key is what `user add --store kv` presents to // start. A stable internal key is what `user add --store kv` presents to
@@ -267,6 +289,24 @@ func main() {
cfg.TLS = tlsCfg cfg.TLS = tlsCfg
log.Printf("NATS TLS: ON (%s)", *tlsCert) log.Printf("NATS TLS: ON (%s)", *tlsCert)
} }
if *wsPort > 0 {
// Expose a WebSocket listener so browser clients (uniweb via nats.ws) reach
// the data plane directly. It reuses the data-plane TLS (wss:// when TLS is
// on, ws:// for a loopback dev stack) and the same browser-origin allowlist
// as the control-plane CORS, so opening the data plane to the browser and
// opening the control plane to it are governed by one --cors-origins list.
scheme := "ws"
if cfg.TLS != nil {
scheme = "wss"
}
cfg.Websocket = &embeddednats.WebsocketConfig{
Host: *bind,
Port: *wsPort,
TLS: cfg.TLS,
AllowedOrigins: splitRoutes(*corsOrigins),
}
log.Printf("NATS WebSocket: ON (%s://%s:%d)", scheme, *bind, *wsPort)
}
ns, err = embeddednats.StartServer(cfg) ns, err = embeddednats.StartServer(cfg)
if err != nil { if err != nil {
log.Fatalf("start embedded nats: %v", err) log.Fatalf("start embedded nats: %v", err)
@@ -286,9 +326,9 @@ func main() {
// only client that can connect in this window (the holder still denies everyone // only client that can connect in this window (the holder still denies everyone
// else; the internal identity bypasses the store). // else; the internal identity bypasses the store).
var js jetstream.JetStream var js jetstream.JetStream
if needJS { if openJS {
var internalNC *nats.Conn var internalNC *nats.Conn
if *natsURL == "" { if embedded {
internalNC, js, err = connectInternalJS(ns, internalID, enforce) internalNC, js, err = connectInternalJS(ns, internalID, enforce)
} else { } else {
internalNC, js, err = connectExternalJS(natsClientURL, *caFile) internalNC, js, err = connectExternalJS(natsClientURL, *caFile)
@@ -310,6 +350,14 @@ func main() {
} }
srv := membership.NewServer(store, blobs, authMode) srv := membership.NewServer(store, blobs, authMode)
// Wire the privileged JetStream context so the control plane owns persisted
// rooms' durable streams (ensure on create + serve GET /rooms/{id}/history). The
// stream replication factor matches the control-plane KV replication so a room's
// history is as available as its metadata. js is nil only for an external NATS
// without a cluster/KV feature, where history degrades to empty (see openJS).
if js != nil {
srv.SetJetStream(js, *kvReplicas)
}
// On a public (non-loopback) bind, disable cleartext rooms: the embedded NATS // On a public (non-loopback) bind, disable cleartext rooms: the embedded NATS
// has no per-subject ACL, so cleartext content would be readable by any // has no per-subject ACL, so cleartext content would be readable by any
// registered peer. Forcing E2E keeps message content confidential regardless // registered peer. Forcing E2E keeps message content confidential regardless
@@ -329,6 +377,24 @@ func main() {
Cluster: clustered, Cluster: clustered,
Store: *storeBackend, Store: *storeBackend,
} }
// CORS allowlist for the browser-native client (uniweb). splitRoutes is reused
// as a generic comma-list parser (trim + drop empties). Empty flag => empty
// slice => CORS stays off, identical to the pre-flag behavior.
if origins := splitRoutes(*corsOrigins); len(origins) > 0 {
srv.AllowedOrigins = origins
log.Printf("CORS: allowing %d browser origin(s): %s", len(origins), strings.Join(origins, ", "))
}
// Trusted reverse proxies for the per-IP rate limit. Behind the same-origin
// Caddy proxy every request arrives with the proxy's IP, which would collapse
// the per-IP rate limit into one bucket for the whole world; naming the proxy
// here lets the limiter believe its X-Forwarded-For and key on the real client
// instead. Empty flag => trust the direct connection only (unchanged behavior).
if proxies := splitRoutes(*trustedProxies); len(proxies) > 0 {
if err := srv.SetTrustedProxies(proxies); err != nil {
log.Fatalf("invalid --trusted-proxies: %v", err)
}
log.Printf("rate limit: trusting forwarded client IP from proxies: %s", strings.Join(proxies, ", "))
}
// Replicated anti-replay (issue 0006a, audit 0008 N3): a clustered node MUST // Replicated anti-replay (issue 0006a, audit 0008 N3): a clustered node MUST
// share its nonce store across the cluster, or a request accepted on one node // share its nonce store across the cluster, or a request accepted on one node
+84
View File
@@ -0,0 +1,84 @@
# Same-origin reverse proxy for the browser-native uniweb chat client.
#
# This is the self-contained fragment that exposes uniweb on magnus
# (organic-machine.com). It is merged into magnus's /etc/caddy/Caddyfile, which
# also hosts unrelated services; only this service's blocks are versioned here
# (the other vhosts carry basic-auth secrets that do not belong in git). The live
# file imports the shared (security_headers) snippet that is duplicated below so
# this fragment validates on its own.
#
# One origin fronts the whole app so the SPA and the bus share an origin: no CORS,
# and the unibus cluster node IPs stay hidden behind this proxy. Caddy obtains and
# renews the Let's Encrypt certificate automatically (the *.organic-machine.com
# wildcard A record points here).
#
# / -> the static SPA (uniweb web/dist) with a single-page-app fallback
# /api/* -> the signed HTTPS control plane (membershipd :8470), prefix stripped
# /nats -> the NATS-over-WebSocket data plane (:8485 magnus / :8480 peers)
#
# Upstreams speak TLS with the bus's own self-signed CA, so Caddy skips upstream
# verification (the hop is still encrypted). The control plane signs requests over
# the UNPREFIXED path, so /api MUST be stripped (handle_path) or signatures fail.
#
# The membershipd nodes must run with the same-origin host in --cors-origins (so
# the NATS WebSocket Origin check accepts it) and with --trusted-proxies naming
# this Caddy node (127.0.0.1,::1,135.125.201.30) so the per-IP rate limit keys on
# the real client behind the proxy instead of collapsing to the proxy's one IP.
(security_headers) {
header {
Strict-Transport-Security "max-age=31536000"
X-Content-Type-Options "nosniff"
X-Frame-Options "DENY"
Referrer-Policy "no-referrer"
-Server
}
}
chat-c200aa64c3125ce8b5f068e0.organic-machine.com {
import security_headers
# Control plane: strip /api so /api/rooms reaches membershipd as /rooms (the
# path the client signs). Prefer the local node; lb_try_duration retries the
# next node within the request on a dial error (safe: a refused connection sent
# no bytes, so even a POST cannot double-apply), and fail_duration plus the
# active /healthz check take a down node out of rotation.
handle_path /api/* {
reverse_proxy https://127.0.0.1:8470 https://141.94.69.66:8470 https://51.91.100.142:8470 {
transport http {
tls_insecure_skip_verify
}
lb_policy first
lb_try_duration 5s
lb_try_interval 250ms
fail_duration 10s
health_uri /healthz
health_interval 10s
health_timeout 5s
}
}
# Data plane: NATS over WebSocket. Strip /nats so the upgrade reaches the ws
# listener at its root. Caddy proxies the WebSocket upgrade natively. The ws
# listener speaks TLS on :8485 (magnus; :8480 is taken by unibus_admin there)
# and :8480 on the peers. Passive failover only (an HTTP health probe would be
# rejected by the NATS ws endpoint).
handle_path /nats* {
reverse_proxy https://127.0.0.1:8485 https://141.94.69.66:8480 https://51.91.100.142:8480 {
transport http {
tls_insecure_skip_verify
}
lb_policy first
lb_try_duration 5s
lb_try_interval 250ms
fail_duration 30s
}
}
# SPA: static files with a client-side-routing fallback to index.html.
handle {
root * /opt/uniweb/dist
try_files {path} /index.html
file_server
}
}
+15
View File
@@ -69,6 +69,12 @@ routes_for() {
echo "==> [2/3] stage each node (REMOTE_DIR=$REMOTE_DIR)" echo "==> [2/3] stage each node (REMOTE_DIR=$REMOTE_DIR)"
for row in "${CLUSTER_NODES[@]}"; do for row in "${CLUSTER_NODES[@]}"; do
read -r name ssh _pub _wg <<<"$row" read -r name ssh _pub _wg <<<"$row"
# Rolling deploy: DEPLOY_ONLY=<name> stages just that node, so a new binary can be
# rolled out one node at a time (the other nodes keep the cluster quorum). Empty =
# stage every node (the original behavior).
if [[ -n "${DEPLOY_ONLY:-}" && "$name" != "$DEPLOY_ONLY" ]]; then
continue
fi
target="${SSH_USER}@${ssh}" target="${SSH_USER}@${ssh}"
nodedir="out/${name}" nodedir="out/${name}"
if [[ ! -d "$nodedir" ]]; then if [[ ! -d "$nodedir" ]]; then
@@ -79,6 +85,13 @@ for row in "${CLUSTER_NODES[@]}"; do
echo "-- node ${name} (ssh ${ssh}) routes=${routes}" echo "-- node ${name} (ssh ${ssh}) routes=${routes}"
# Resolve this node's WebSocket port. magnus runs unibus_admin on 127.0.0.1:8480,
# so the bus WS cannot bind 0.0.0.0:8480 there (it crash-loops). A per-node
# override (WS_PORT_<NAME> in nodes.env) lets magnus use a free port while the
# rest share the default — keeping the deploy reproducible (issue uniweb/0001).
node_ws_var="WS_PORT_${name^^}"
node_ws="${!node_ws_var:-$WS_PORT}"
# Generate this node's cluster.env locally, then ship it. # Generate this node's cluster.env locally, then ship it.
envfile="build/cluster-${name}.env" envfile="build/cluster-${name}.env"
mkdir -p build mkdir -p build
@@ -90,6 +103,8 @@ KV_REPLICAS=${KV_REPLICAS}
HTTP_PORT=${HTTP_PORT} HTTP_PORT=${HTTP_PORT}
NATS_CLIENT_PORT=${NATS_CLIENT_PORT} NATS_CLIENT_PORT=${NATS_CLIENT_PORT}
NATS_ROUTE_PORT=${NATS_ROUTE_PORT} NATS_ROUTE_PORT=${NATS_ROUTE_PORT}
WS_PORT=${node_ws}
CORS_ORIGINS=${CORS_ORIGINS}
ROUTES=${routes} ROUTES=${routes}
CLUSTER_PASS_FILE=${REMOTE_DIR}/secrets/cluster.pass CLUSTER_PASS_FILE=${REMOTE_DIR}/secrets/cluster.pass
TLS_CERT=${REMOTE_DIR}/tls/server-${name}.crt TLS_CERT=${REMOTE_DIR}/tls/server-${name}.crt
+3 -1
View File
@@ -35,7 +35,9 @@ ExecStart=/opt/unibus/membershipd \
--route-tls-ca ${ROUTE_TLS_CA} \ --route-tls-ca ${ROUTE_TLS_CA} \
--internal-id-file ${INTERNAL_ID_FILE} \ --internal-id-file ${INTERNAL_ID_FILE} \
--store kv \ --store kv \
--kv-replicas ${KV_REPLICAS} --kv-replicas ${KV_REPLICAS} \
--ws-port ${WS_PORT} \
--cors-origins ${CORS_ORIGINS}
# Restart=always (NOT on-failure): a clean SIGTERM exits success, and on-failure # Restart=always (NOT on-failure): a clean SIGTERM exits success, and on-failure
# would then NOT restart, leaving the node silently dead (see function_tags.md). # would then NOT restart, leaving the node silently dead (see function_tags.md).
Restart=always Restart=always
+13
View File
@@ -23,6 +23,19 @@ NATS_CLIENT_PORT=4250
NATS_ROUTE_PORT=6250 NATS_ROUTE_PORT=6250
HTTP_PORT=8470 HTTP_PORT=8470
# Browser data-plane: WebSocket listener so the browser-native uniweb client
# (nats.ws) reaches NATS, and the CORS allowlist for its calls to the control
# plane. WS reuses the data-plane TLS, so it serves wss:// (the cluster runs with
# TLS). CORS_ORIGINS is a comma-separated list of allowed browser origins (no
# spaces). Issue uniweb/0001. The node's firewall must allow WS_PORT.
WS_PORT=8480
# Per-node WS port override (WS_PORT_<NAME>). magnus runs unibus_admin on
# 127.0.0.1:8480, so the bus WebSocket cannot bind 0.0.0.0:8480 there — it would
# crash-loop. magnus therefore serves the browser WS on 8485; homer and datardos
# keep 8480 (no admin panel). Verified during the 2026-06-13 rollout.
WS_PORT_MAGNUS=8485
CORS_ORIGINS="http://localhost:5173"
# Remote install layout and SSH login user. # Remote install layout and SSH login user.
REMOTE_DIR="/opt/unibus" REMOTE_DIR="/opt/unibus"
SSH_USER="root" SSH_USER="root"
+2 -1
View File
@@ -10,6 +10,8 @@ require (
github.com/nats-io/nats.go v1.49.0 github.com/nats-io/nats.go v1.49.0
github.com/nats-io/nkeys v0.4.15 github.com/nats-io/nkeys v0.4.15
github.com/oklog/ulid/v2 v2.1.0 github.com/oklog/ulid/v2 v2.1.0
github.com/tyler-smith/go-bip39 v1.1.0
golang.org/x/crypto v0.51.0
golang.org/x/time v0.15.0 golang.org/x/time v0.15.0
modernc.org/sqlite v1.47.0 modernc.org/sqlite v1.47.0
) )
@@ -26,7 +28,6 @@ require (
github.com/nats-io/nuid v1.0.1 // indirect github.com/nats-io/nuid v1.0.1 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
golang.org/x/crypto v0.51.0 // indirect
golang.org/x/mobile v0.0.0-20260602190626-68735029466e // indirect golang.org/x/mobile v0.0.0-20260602190626-68735029466e // indirect
golang.org/x/mod v0.36.0 // indirect golang.org/x/mod v0.36.0 // indirect
golang.org/x/sync v0.20.0 // indirect golang.org/x/sync v0.20.0 // indirect
+8
View File
@@ -35,18 +35,26 @@ github.com/oklog/ulid/v2 v2.1.0/go.mod h1:rcEKHmBBKfef9DhnvX7y1HZBYxjXb0cP5ExxNs
github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o= github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/tyler-smith/go-bip39 v1.1.0 h1:5eUemwrMargf3BSLRRCalXT93Ns6pQJIjYQN2nyfOP8=
github.com/tyler-smith/go-bip39 v1.1.0/go.mod h1:gUYDtqQw1JS3ZJ8UWVcGTGqqr6YIN3CWg+kkNaLt55U=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI=
golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8=
golang.org/x/mobile v0.0.0-20260602190626-68735029466e h1:YxPXu/HWDTcSSrzSX+sCltsfcNCa/ZYVG43oslMouNU= golang.org/x/mobile v0.0.0-20260602190626-68735029466e h1:YxPXu/HWDTcSSrzSX+sCltsfcNCa/ZYVG43oslMouNU=
golang.org/x/mobile v0.0.0-20260602190626-68735029466e/go.mod h1:ltIbhcRzKgwHa4ZxKJeiv0nyzcXUUYCqMyO0Y+vPmXw= golang.org/x/mobile v0.0.0-20260602190626-68735029466e/go.mod h1:ltIbhcRzKgwHa4ZxKJeiv0nyzcXUUYCqMyO0Y+vPmXw=
golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4= golang.org/x/mod v0.36.0 h1:JJjpVx6myfUsUdAzZuOSTTmRE0PfZeNWzzvKrP7amb4=
golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ= golang.org/x/mod v0.36.0/go.mod h1:moc6ELqsWcOw5Ef3xVprK5ul/MvtVvkIXLziUOICjUQ=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ= golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ=
golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8= golang.org/x/tools v0.45.0 h1:18qN3FAooORvApf5XjCXgsuayZOEtXf6JK18I3+ONa8=
+49
View File
@@ -0,0 +1,49 @@
package mobile
import (
"encoding/hex"
"testing"
)
// TestDeriveParityWithWeb pins the Go wallet derivation to the TypeScript one
// (web/src/wallet/derive.ts). The canonical BIP39 test mnemonic must derive to this
// exact Ed25519 sign_pub — the same value the uniweb client showed for this phrase.
// If this fails, web and mobile would derive different identities from the same seed
// and the "same account on both devices" guarantee breaks.
func TestDeriveParityWithWeb(t *testing.T) {
const mnemonic = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
const wantSignPub = "34302746268e7370d35940e1bcef8c0b1c13a857ea6209e6ecc6e9b3af06b3c8"
id, err := deriveIdentity(mnemonic)
if err != nil {
t.Fatalf("deriveIdentity: %v", err)
}
if got := hex.EncodeToString(id.SignPub); got != wantSignPub {
t.Fatalf("sign_pub mismatch:\n got %s\n want %s", got, wantSignPub)
}
if len(id.SignPriv) != 64 || len(id.KexPub) != 32 || len(id.KexPriv) != 32 {
t.Fatalf("bad key lengths: signPriv=%d kexPub=%d kexPriv=%d",
len(id.SignPriv), len(id.KexPub), len(id.KexPriv))
}
// sign_priv is Go's ed25519 layout: seed || pub, so its tail must equal sign_pub.
if got := hex.EncodeToString(id.SignPriv[32:]); got != wantSignPub {
t.Fatalf("sign_priv tail != sign_pub:\n got %s\n want %s", got, wantSignPub)
}
}
// TestMnemonicRoundTrip checks a freshly generated phrase validates and derives.
func TestMnemonicRoundTrip(t *testing.T) {
m, err := NewMnemonic()
if err != nil {
t.Fatalf("NewMnemonic: %v", err)
}
if !ValidateMnemonic(m) {
t.Fatalf("generated mnemonic failed validation: %q", m)
}
if _, err := deriveIdentity(m); err != nil {
t.Fatalf("deriveIdentity(fresh): %v", err)
}
if ValidateMnemonic("not a real mnemonic at all please") {
t.Fatal("garbage phrase validated as a mnemonic")
}
}
+266
View File
@@ -0,0 +1,266 @@
// Package mobile is the gomobile binding surface for the unibus client. It wraps
// pkg/client behind a flat API (strings, []byte, JSON, a small interface) that
// gobind can export to Kotlin/Swift, so an Android or iOS app speaks the real bus
// protocol — NATS data plane, signed HTTP control plane and end-to-end crypto —
// with no reimplementation of either the protocol or the cryptography.
//
// The wrapper is intentionally thin: every method delegates to pkg/client and only
// reshapes types into something gobind understands (gomobile cannot export slices
// of structs, so list results come back as JSON the caller decodes).
package mobile
import (
"crypto/ed25519"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"os"
bip39 "github.com/tyler-smith/go-bip39"
"golang.org/x/crypto/curve25519"
"golang.org/x/crypto/hkdf"
cs "fn-registry/functions/cybersecurity"
"github.com/enmanuel/unibus/pkg/client"
"github.com/enmanuel/unibus/pkg/frame"
"github.com/enmanuel/unibus/pkg/room"
)
// HKDF domain-separation labels. These MUST stay identical to the uniweb wallet
// derivation (web/src/wallet/derive.ts) so the same BIP39 mnemonic yields the same
// identity (same sign_pub) on web and mobile — that shared identity is what lets a
// user reach the same rooms from either device. Changing a label forks the wallet.
const (
infoSign = "unibus-sign-v1"
infoKex = "unibus-kex-v1"
)
// FrameListener receives decrypted messages from a subscribed room. OnFrame is
// invoked on a NATS delivery goroutine; the app must hop to its UI thread before
// touching UI state.
type FrameListener interface {
OnFrame(roomID, sender, msgID, text string)
}
// NewMnemonic returns a fresh 12-word BIP39 recovery phrase (128 bits of entropy
// from the system CSPRNG). Show it to the user exactly once and never persist it:
// it is the only way to recover the identity on another device.
func NewMnemonic() (string, error) {
ent, err := bip39.NewEntropy(128)
if err != nil {
return "", err
}
return bip39.NewMnemonic(ent)
}
// ValidateMnemonic reports whether m is a valid 12-word BIP39 phrase (every word in
// the wordlist and a correct checksum). A phrase that fails this must not be used
// to derive an identity.
func ValidateMnemonic(m string) bool {
return bip39.IsMnemonicValid(m)
}
// hkdfExpand derives 32 bytes from ikm under an info label with an empty salt,
// matching the web's hkdf(sha256, seed, undefined, info, 32).
func hkdfExpand(ikm []byte, info string) ([]byte, error) {
r := hkdf.New(sha256.New, ikm, nil, []byte(info))
out := make([]byte, 32)
if _, err := io.ReadFull(r, out); err != nil {
return nil, err
}
return out, nil
}
// deriveIdentity reproduces the uniweb wallet derivation byte for byte:
//
// seed = BIP39_seed(mnemonic) (PBKDF2, salt "mnemonic", 64 bytes)
// signSeed = HKDF-SHA256(seed, salt="", info=sign, 32)
// kexSeed = HKDF-SHA256(seed, salt="", info=kex, 32)
// Ed25519 signing key from signSeed (priv = seed||pub, Go's layout)
// X25519 key-exchange key from kexSeed
//
// The result is exactly a cs.Identity (sign_pub 32, sign_priv 64, kex_pub 32,
// kex_priv 32), so the bus accepts the derived keys as a first-class peer.
func deriveIdentity(mnemonic string) (cs.Identity, error) {
if !bip39.IsMnemonicValid(mnemonic) {
return cs.Identity{}, fmt.Errorf("invalid mnemonic")
}
seed := bip39.NewSeed(mnemonic, "") // 64-byte BIP39 seed (salt "mnemonic", no passphrase)
signSeed, err := hkdfExpand(seed, infoSign)
if err != nil {
return cs.Identity{}, err
}
kexSeed, err := hkdfExpand(seed, infoKex)
if err != nil {
return cs.Identity{}, err
}
signPriv := ed25519.NewKeyFromSeed(signSeed) // 64 bytes = signSeed || sign_pub
kexPub, err := curve25519.X25519(kexSeed, curve25519.Basepoint)
if err != nil {
return cs.Identity{}, err
}
return cs.Identity{
SignPub: append([]byte(nil), signPriv[32:]...),
SignPriv: append([]byte(nil), signPriv...),
KexPub: kexPub,
KexPriv: append([]byte(nil), kexSeed...),
}, nil
}
// DeriveAndSaveIdentity derives the deterministic identity from a BIP39 mnemonic
// and writes it to path, overwriting any previous identity on this device — which
// is exactly what recovering on a new device (or after a reset) needs. It returns
// the sign_pub hex so the UI can show which identity the phrase reconstructs. The
// mnemonic itself is never stored; only the derived keypair is persisted (0600).
func DeriveAndSaveIdentity(path, mnemonic string) (string, error) {
id, err := deriveIdentity(mnemonic)
if err != nil {
return "", err
}
// WriteNewIdentity refuses to overwrite an existing file; recover/re-create must
// replace the device's identity, so drop the old one first.
_ = os.Remove(path)
if err := client.WriteNewIdentity(path, id); err != nil {
return "", err
}
return hex.EncodeToString(id.SignPub), nil
}
// HasIdentity reports whether a saved identity already exists at path, so the UI can
// decide between the onboarding screen and going straight to connect.
func HasIdentity(path string) bool {
_, err := os.Stat(path)
return err == nil
}
// SignPubAt returns the sign_pub hex of the identity saved at path (for showing the
// current account), or an error if none is saved.
func SignPubAt(path string) (string, error) {
id, err := client.LoadIdentity(path)
if err != nil {
return "", err
}
return hex.EncodeToString(id.SignPub), nil
}
// GenerateIdentity creates a fresh random identity at path if none exists yet.
// Retained for callers that want a throwaway local identity instead of a
// recoverable BIP39 wallet.
func GenerateIdentity(path string) error {
if _, err := os.Stat(path); err == nil {
return nil
}
id, err := cs.GenerateIdentity()
if err != nil {
return err
}
return client.WriteNewIdentity(path, id)
}
// Session is a connected unibus peer. Create it with NewSession and release it with
// Close when the app stops.
type Session struct {
c *client.Client
endpointID string
}
// NewSession loads the identity at idPath and connects to the bus. natsURL is the
// data plane (e.g. nats://host:4250) and ctrlURL is the control-plane HTTP endpoint
// (e.g. http://host:8470).
func NewSession(idPath, natsURL, ctrlURL string) (*Session, error) {
id, err := client.LoadIdentity(idPath)
if err != nil {
return nil, err
}
c, err := client.New(natsURL, ctrlURL, id)
if err != nil {
return nil, err
}
return &Session{c: c, endpointID: frame.EndpointID(id.SignPub)}, nil
}
// Close disconnects the peer from the bus.
func (s *Session) Close() error { return s.c.Close() }
// EndpointID returns this peer's stable endpoint id (the sender stamped on frames).
func (s *Session) EndpointID() string { return s.endpointID }
// CreateRoom opens a room on the given subject. mode is "matrix" for the encrypted,
// persisted and signed policy, or "nats" for plain cleartext. Returns the room id
// used by Join, Publish and Subscribe.
func (s *Session) CreateRoom(subject, mode string) (string, error) {
p := room.ModeNATS
if mode == "matrix" {
p = room.ModeMatrix
}
return s.c.CreateRoom(subject, p)
}
// Join prepares the session to publish to and receive from a room, fetching the
// room key when the room is encrypted.
func (s *Session) Join(roomID string) error { return s.c.Join(roomID) }
// Publish sends a UTF-8 text message to the room.
func (s *Session) Publish(roomID, text string) error { return s.c.Publish(roomID, []byte(text)) }
// Subscribe streams decrypted messages of the room to the listener until the
// session is closed. For persisted (matrix) rooms this also replays the room's
// history (JetStream DeliverAll) before live messages, so opening a room shows past
// messages — the same behaviour as the web client.
func (s *Session) Subscribe(roomID string, l FrameListener) error {
_, err := s.c.Subscribe(roomID, func(f frame.Frame, plaintext []byte) {
l.OnFrame(roomID, f.Sender, f.MsgID, string(plaintext))
})
return err
}
// ListMyRooms returns every room this peer is a member of, as a JSON array of
// objects {id, subject, mode, role} where mode is "matrix" or "nats". Returned as
// JSON because gomobile cannot export a slice of structs.
func (s *Session) ListMyRooms() (string, error) {
rooms, err := s.c.ListMyRooms()
if err != nil {
return "", err
}
type roomJSON struct {
ID string `json:"id"`
Subject string `json:"subject"`
Mode string `json:"mode"`
Role string `json:"role"`
}
out := make([]roomJSON, 0, len(rooms))
for _, r := range rooms {
mode := "nats"
if r.Policy.Encrypt {
mode = "matrix"
}
out = append(out, roomJSON{ID: r.RoomID, Subject: r.Subject, Mode: mode, Role: r.Role})
}
b, err := json.Marshal(out)
return string(b), err
}
// Directory returns the cluster member directory as a JSON array of objects
// {sign_pub, endpoint, handle, role}, so the UI can map a frame's endpoint id to a
// readable handle. Returns "[]" semantics via JSON; callers degrade to short ids if
// a handle is missing.
func (s *Session) Directory() (string, error) {
entries, err := s.c.Directory()
if err != nil {
return "", err
}
type dirJSON struct {
SignPub string `json:"sign_pub"`
Endpoint string `json:"endpoint"`
Handle string `json:"handle"`
Role string `json:"role"`
}
out := make([]dirJSON, 0, len(entries))
for _, e := range entries {
out = append(out, dirJSON{SignPub: e.SignPub, Endpoint: e.Endpoint, Handle: e.Handle, Role: e.Role})
}
b, err := json.Marshal(out)
return string(b), err
}
+50
View File
@@ -0,0 +1,50 @@
package client
// This file exposes two read helpers the mobile binding needs but that the core
// client did not surface yet: the peer's own endpoint id, and the cluster member
// directory (endpoint id -> human handle). Both are additive and read-only.
// EndpointID returns this peer's stable endpoint id, base64url(sha256(signPub)),
// the value the bus stamps as the sender of every frame this peer publishes.
func (c *Client) EndpointID() string { return c.endpoint }
// DirectoryEntry maps a member's stable endpoint id to a human handle, as served
// by the control plane's GET /directory. A client uses it to render readable names
// instead of raw endpoint ids on incoming frames.
type DirectoryEntry struct {
SignPub string // 64-hex Ed25519 public key
Endpoint string // base64url-nopad, == EndpointID(signPub)
Handle string
Role string
}
type directoryMemberWire struct {
SignPub string `json:"sign_pub"`
Endpoint string `json:"endpoint"`
Handle string `json:"handle"`
Role string `json:"role"`
}
type directoryResp struct {
Members []directoryMemberWire `json:"members"`
}
// Directory fetches the cluster-wide member directory (endpoint id -> handle). Any
// active user may read it; the request is signed like every other control-plane
// call. Returns the active members only (the server filters to status=active).
func (c *Client) Directory() ([]DirectoryEntry, error) {
var resp directoryResp
if err := c.doJSON("GET", "/directory", nil, &resp); err != nil {
return nil, err
}
out := make([]DirectoryEntry, 0, len(resp.Members))
for _, m := range resp.Members {
out = append(out, DirectoryEntry{
SignPub: m.SignPub,
Endpoint: m.Endpoint,
Handle: m.Handle,
Role: m.Role,
})
}
return out, nil
}
+16
View File
@@ -75,6 +75,22 @@ func LoadOrCreateIdentity(path string) (cs.Identity, error) {
return id, nil return id, nil
} }
// WriteNewIdentity writes id to path in the canonical identity-file format read
// by LoadIdentity, but REFUSES to overwrite an existing file: provisioning a new
// identity must never silently clobber another process's private keys. The file
// is created 0600 (it holds private keys). It is the write half of one-command
// bot provisioning (`membershipd bot add --out <path>`): the freshly minted
// identity it writes is exactly what LoadIdentity reconstructs, so a bot binary
// (worker/clientcheck) consumes the credentials with no extra conversion step.
func WriteNewIdentity(path string, id cs.Identity) error {
if _, err := os.Stat(path); err == nil {
return fmt.Errorf("client: identity file %q already exists; refusing to overwrite", path)
} else if !os.IsNotExist(err) {
return fmt.Errorf("client: stat identity %q: %w", path, err)
}
return saveIdentity(path, id)
}
func saveIdentity(path string, id cs.Identity) error { func saveIdentity(path string, id cs.Identity) error {
if dir := filepath.Dir(path); dir != "" { if dir := filepath.Dir(path); dir != "" {
if err := os.MkdirAll(dir, 0o755); err != nil { if err := os.MkdirAll(dir, 0o755); err != nil {
+59
View File
@@ -79,6 +79,42 @@ type ServerConfig struct {
// availability (issue 0003a). Nil keeps the server standalone (the legacy // availability (issue 0003a). Nil keeps the server standalone (the legacy
// single-node behavior). // single-node behavior).
Cluster *ClusterConfig Cluster *ClusterConfig
// Websocket, when non-nil, opens an ADDITIONAL WebSocket listener on the
// embedded nats-server so browser clients (nats.ws) can reach the data plane
// directly, the same way native TCP peers (Go, Kotlin) do (issue uniweb/0001).
// Native TCP clients are unaffected: the WebSocket listener is a separate port
// layered on top of the existing TCP listener, and the client authenticator
// (Auth) applies to both. Nil keeps the server TCP-only (legacy behavior).
Websocket *WebsocketConfig
}
// WebsocketConfig configures the embedded nats-server's WebSocket listener so a
// browser can speak the NATS protocol over ws://. A browser cannot open a raw TCP
// socket, so this is the only way the SPA reaches the data plane without a Go
// gateway in between.
//
// Security: off loopback a browser requires wss:// (TLS) — set TLS with a
// certificate the browser trusts. NoTLS plain ws:// is acceptable only for a
// loopback dev stack. The WebSocket upgrade also enforces an Origin allowlist
// (browser same-origin policy); AllowedOrigins must list the SPA's origins or the
// browser handshake is refused.
type WebsocketConfig struct {
// Host is the bind interface for the WebSocket listener; "" lets nats-server
// pick its default. Use "127.0.0.1" to keep it loopback-only in dev.
Host string
// Port is the WebSocket listen port (e.g. 8480). Required (non-zero) for the
// listener to open.
Port int
// NoTLS serves plain ws:// instead of wss://. Loopback/dev only: browsers refuse
// ws:// to a non-loopback origin. Ignored when TLS is set (TLS implies wss://).
NoTLS bool
// TLS, when set, serves wss:// with this certificate. Required for any browser
// origin that is not loopback.
TLS *tls.Config
// AllowedOrigins is the allowlist of browser Origin headers permitted to upgrade
// the WebSocket. Empty = same-origin only (nats-server SameOrigin). Never use a
// wildcard in production; list the exact SPA origins.
AllowedOrigins []string
} }
// Start is a thin backward-compatible wrapper: embedded JetStream server on the // Start is a thin backward-compatible wrapper: embedded JetStream server on the
@@ -170,6 +206,29 @@ func StartServer(cfg ServerConfig) (*server.Server, error) {
opts.TLS = true opts.TLS = true
} }
if cfg.Websocket != nil {
// Layer a WebSocket listener on top of the TCP data plane so browser
// clients (nats.ws) can connect. The client authenticator (opts.*Auth above)
// applies to WebSocket connections too, so a browser still has to pass the
// nkey + allowlist check; this only adds a transport, not a trust bypass.
ws := server.WebsocketOpts{
Host: cfg.Websocket.Host,
Port: cfg.Websocket.Port,
AllowedOrigins: cfg.Websocket.AllowedOrigins,
}
if cfg.Websocket.TLS != nil {
ws.TLSConfig = cfg.Websocket.TLS
} else {
// No certificate: plain ws:// (loopback/dev only). Browsers refuse this
// off-loopback, which is the intended guard rail.
ws.NoTLS = true
}
// Empty AllowedOrigins means "same-origin only": tell nats-server to enforce
// it rather than defaulting to accept-any-origin.
ws.SameOrigin = len(cfg.Websocket.AllowedOrigins) == 0
opts.Websocket = ws
}
if cfg.Cluster != nil { if cfg.Cluster != nil {
if err := applyClusterOpts(opts, cfg.Cluster); err != nil { if err := applyClusterOpts(opts, cfg.Cluster); err != nil {
return nil, err return nil, err
+108
View File
@@ -0,0 +1,108 @@
package embeddednats_test
import (
"fmt"
"net"
"net/http"
"strings"
"testing"
"time"
"github.com/enmanuel/unibus/pkg/embeddednats"
)
// wsFreePort returns an OS-assigned free TCP port on loopback. Kept local to this
// file so the WebSocket tests do not depend on the cluster test helpers.
func wsFreePort(t *testing.T) int {
t.Helper()
l, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("reserve free port: %v", err)
}
defer l.Close()
return l.Addr().(*net.TCPAddr).Port
}
// TestWebsocketListenerOpens verifies that when a ServerConfig carries a
// WebsocketConfig the embedded nats-server opens the additional WebSocket port and
// accepts a connection there, while the regular TCP client port keeps working. A
// browser cannot speak raw TCP, so this WebSocket listener is the only path the SPA
// has to the data plane (issue uniweb/0001).
func TestWebsocketListenerOpens(t *testing.T) {
clientPort := wsFreePort(t)
wsPort := wsFreePort(t)
ns, err := embeddednats.StartServer(embeddednats.ServerConfig{
StoreDir: t.TempDir(),
Host: "127.0.0.1",
Port: clientPort,
Websocket: &embeddednats.WebsocketConfig{
Host: "127.0.0.1",
Port: wsPort,
NoTLS: true, // loopback dev: plain ws://
},
})
if err != nil {
t.Fatalf("StartServer with websocket: %v", err)
}
t.Cleanup(func() { ns.Shutdown(); ns.WaitForShutdown() })
// The WebSocket listener must accept a TCP connection on its dedicated port.
addr := fmt.Sprintf("127.0.0.1:%d", wsPort)
conn, err := net.DialTimeout("tcp", addr, 2*time.Second)
if err != nil {
t.Fatalf("websocket port %d not accepting connections: %v", wsPort, err)
}
conn.Close()
// And it must speak the WebSocket upgrade handshake: a GET with the upgrade
// headers should get a 101 Switching Protocols (nats-server's ws endpoint),
// proving it is a real WebSocket listener, not just an open socket.
req, err := http.NewRequest(http.MethodGet, "http://"+addr+"/", nil)
if err != nil {
t.Fatalf("build upgrade request: %v", err)
}
req.Header.Set("Upgrade", "websocket")
req.Header.Set("Connection", "Upgrade")
req.Header.Set("Sec-WebSocket-Version", "13")
req.Header.Set("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
client := &http.Client{Timeout: 2 * time.Second}
resp, err := client.Do(req)
if err != nil {
t.Fatalf("websocket upgrade request: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusSwitchingProtocols {
t.Fatalf("websocket upgrade: got status %d, want 101 Switching Protocols", resp.StatusCode)
}
}
// TestNoWebsocketByDefault verifies the listener stays TCP-only when WebsocketConfig
// is nil: opening the browser transport must be an explicit opt-in so existing
// single-node and cluster deployments are unchanged.
func TestNoWebsocketByDefault(t *testing.T) {
clientPort := wsFreePort(t)
// Reserve a port, then free it, so we can assert nothing is listening there.
maybeWSPort := wsFreePort(t)
ns, err := embeddednats.StartServer(embeddednats.ServerConfig{
StoreDir: t.TempDir(),
Host: "127.0.0.1",
Port: clientPort,
// Websocket intentionally nil.
})
if err != nil {
t.Fatalf("StartServer: %v", err)
}
t.Cleanup(func() { ns.Shutdown(); ns.WaitForShutdown() })
conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", maybeWSPort), 300*time.Millisecond)
if err == nil {
conn.Close()
t.Fatalf("a listener is unexpectedly open on %d with no WebsocketConfig", maybeWSPort)
}
if !strings.Contains(err.Error(), "refused") && !strings.Contains(err.Error(), "timeout") {
t.Logf("dial error (acceptable, port closed): %v", err)
}
}
+157
View File
@@ -0,0 +1,157 @@
package membership_test
import (
"net/http"
"net/http/httptest"
"path/filepath"
"strings"
"testing"
"github.com/enmanuel/unibus/pkg/blobstore"
"github.com/enmanuel/unibus/pkg/membership"
)
// newCORSServer builds a control-plane server with the given CORS allowlist over a
// throwaway store, and returns a live httptest server. /healthz is auth-exempt, so
// the CORS tests can exercise the cross-origin pipeline without signing requests.
func newCORSServer(t *testing.T, origins ...string) *httptest.Server {
t.Helper()
dir := t.TempDir()
store, err := membership.Open(filepath.Join(dir, "unibus.db"))
if err != nil {
t.Fatalf("store: %v", err)
}
t.Cleanup(func() { store.Close() })
blobs, _ := blobstore.New(filepath.Join(dir, "blobs"))
srv := membership.NewServer(store, blobs, membership.AuthOff)
srv.AllowedOrigins = origins
ts := httptest.NewServer(srv)
t.Cleanup(ts.Close)
return ts
}
// TestCORSPreflightAllowedOrigin: a preflight (OPTIONS) from an allow-listed origin
// is answered 204 with the Access-Control headers, and never reaches auth. This is
// what lets the browser-native uniweb client call the control plane (issue
// uniweb/0001).
func TestCORSPreflightAllowedOrigin(t *testing.T) {
const origin = "http://localhost:5173"
ts := newCORSServer(t, origin)
req, _ := http.NewRequest(http.MethodOptions, ts.URL+"/rooms", nil)
req.Header.Set("Origin", origin)
req.Header.Set("Access-Control-Request-Method", "POST")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("preflight: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusNoContent {
t.Fatalf("preflight status = %d, want 204", resp.StatusCode)
}
if got := resp.Header.Get("Access-Control-Allow-Origin"); got != origin {
t.Fatalf("Allow-Origin = %q, want %q", got, origin)
}
if got := resp.Header.Get("Access-Control-Allow-Methods"); got == "" {
t.Fatalf("Allow-Methods missing on preflight")
}
// The control-plane request-auth headers a browser signs every request with must
// be allow-listed, or the browser's preflight blocks the real request (the bug a
// live browser surfaced: listRooms failed with "Failed to fetch").
if got := resp.Header.Get("Access-Control-Allow-Headers"); !strings.Contains(got, "X-Unibus-Sig") {
t.Fatalf("Allow-Headers must include the X-Unibus-* auth headers, got %q", got)
}
}
// TestCORSPreflightDisallowedOrigin: a preflight from an origin NOT in the allowlist
// gets 403 and no Access-Control headers, so the browser blocks the real request.
func TestCORSPreflightDisallowedOrigin(t *testing.T) {
ts := newCORSServer(t, "http://localhost:5173")
req, _ := http.NewRequest(http.MethodOptions, ts.URL+"/rooms", nil)
req.Header.Set("Origin", "https://evil.example.com")
req.Header.Set("Access-Control-Request-Method", "POST")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("preflight: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusForbidden {
t.Fatalf("disallowed preflight status = %d, want 403", resp.StatusCode)
}
if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "" {
t.Fatalf("Allow-Origin leaked for disallowed origin: %q", got)
}
}
// TestCORSActualRequestCarriesHeader: a real GET from an allow-listed origin is
// served normally AND carries the Allow-Origin header so the browser accepts the
// response.
func TestCORSActualRequestCarriesHeader(t *testing.T) {
const origin = "http://localhost:5173"
ts := newCORSServer(t, origin)
req, _ := http.NewRequest(http.MethodGet, ts.URL+"/healthz", nil)
req.Header.Set("Origin", origin)
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("get: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("healthz status = %d, want 200", resp.StatusCode)
}
if got := resp.Header.Get("Access-Control-Allow-Origin"); got != origin {
t.Fatalf("Allow-Origin = %q, want %q", got, origin)
}
}
// TestCORSDisabledByDefault: with an empty allowlist no Access-Control header is
// ever emitted (CORS off) and requests behave exactly as before. This guards the
// opt-in invariant: untouched deployments are unaffected.
func TestCORSDisabledByDefault(t *testing.T) {
ts := newCORSServer(t) // no origins
req, _ := http.NewRequest(http.MethodGet, ts.URL+"/healthz", nil)
req.Header.Set("Origin", "http://localhost:5173")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("get: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("healthz status = %d, want 200", resp.StatusCode)
}
if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "" {
t.Fatalf("Allow-Origin emitted with CORS off: %q", got)
}
}
// TestCORSNativeClientUnaffected: a request with no Origin header (a native Go/Kotlin
// client) is processed normally and gets no CORS headers, even when an allowlist is
// configured.
func TestCORSNativeClientUnaffected(t *testing.T) {
ts := newCORSServer(t, "http://localhost:5173")
resp, err := http.Get(ts.URL + "/healthz") // no Origin header
if err != nil {
t.Fatalf("get: %v", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
t.Fatalf("healthz status = %d, want 200", resp.StatusCode)
}
if got := resp.Header.Get("Access-Control-Allow-Origin"); got != "" {
t.Fatalf("Allow-Origin set for a no-Origin native client: %q", got)
}
}
+155
View File
@@ -0,0 +1,155 @@
package membership
import (
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"net/http"
"testing"
cs "fn-registry/functions/cybersecurity"
"github.com/enmanuel/unibus/pkg/frame"
)
// directory signs a GET /directory as id and decodes the response envelope. The
// path has no /api prefix: Caddy strips /api before forwarding to membershipd, so
// the route is registered (and hit here) as /directory, matching production.
func directory(t *testing.T, h *authHarness, id cs.Identity, n int) (int, directoryResp) {
t.Helper()
code, body := signedJSON(t, h, "GET", "/directory", nil, id, n)
var resp directoryResp
if code == http.StatusOK {
if err := json.Unmarshal([]byte(body), &resp); err != nil {
t.Fatalf("decode directory: %v (%s)", err, body)
}
}
return code, resp
}
// findMember returns the directory entry for a signing key (case-insensitive).
func findMember(members []directoryMember, signPub string) (directoryMember, bool) {
want := normalizeSignPub(signPub)
for _, m := range members {
if normalizeSignPub(m.SignPub) == want {
return m, true
}
}
return directoryMember{}, false
}
// TestDirectoryGolden is the happy path: an authenticated bus user (here the seed
// admin alice, plus a registered member bob) reads the directory and gets every
// active user's handle, role, and an endpoint derived server-side from the
// sign_pub with the bus's own construction (frame.EndpointID). Two users in ->
// 200 with both handles and correct endpoints.
func TestDirectoryGolden(t *testing.T) {
h := newAuthHarness(t, AuthEnforce)
bob, _ := cs.GenerateIdentity()
register(t, h, bob, "bob") // role member
bobPub := hex.EncodeToString(bob.SignPub)
code, resp := directory(t, h, h.alice, 1)
if code != http.StatusOK {
t.Fatalf("directory should be 200 for an authenticated user, got %d", code)
}
aliceRow, ok := findMember(resp.Members, h.alicePub)
if !ok {
t.Fatalf("seed admin alice missing from directory: %+v", resp.Members)
}
if aliceRow.Handle != "alice" || aliceRow.Role != RoleAdmin {
t.Fatalf("alice row wrong: %+v", aliceRow)
}
if want := frame.EndpointID(h.alice.SignPub); aliceRow.Endpoint != want {
t.Fatalf("alice endpoint = %q, want %q", aliceRow.Endpoint, want)
}
bobRow, ok := findMember(resp.Members, bobPub)
if !ok {
t.Fatalf("registered member bob missing from directory: %+v", resp.Members)
}
if bobRow.Handle != "bob" || bobRow.Role != RoleMember {
t.Fatalf("bob row wrong: %+v", bobRow)
}
if want := frame.EndpointID(bob.SignPub); bobRow.Endpoint != want {
t.Fatalf("bob endpoint = %q, want %q", bobRow.Endpoint, want)
}
}
// TestDirectoryUnauthenticatedRejected is the auth contract: under enforce an
// unsigned GET /directory is rejected with 401 by the middleware, before the
// handler ever runs — the directory is not public.
func TestDirectoryUnauthenticatedRejected(t *testing.T) {
h := newAuthHarness(t, AuthEnforce)
req, _ := http.NewRequest("GET", h.ts.URL+"/directory", nil)
code, _ := do(t, req)
if code != http.StatusUnauthorized {
t.Fatalf("unsigned directory request under enforce should be 401, got %d", code)
}
}
// TestDirectoryExcludesRevoked: a revoked user must not appear in the directory
// (status=active filter), while active users still do.
func TestDirectoryExcludesRevoked(t *testing.T) {
h := newAuthHarness(t, AuthEnforce)
gone, _ := cs.GenerateIdentity()
register(t, h, gone, "gone")
gonePub := hex.EncodeToString(gone.SignPub)
if err := h.store.RevokeUser(gonePub); err != nil {
t.Fatalf("revoke gone: %v", err)
}
code, resp := directory(t, h, h.alice, 1)
if code != http.StatusOK {
t.Fatalf("directory should be 200, got %d", code)
}
if _, ok := findMember(resp.Members, gonePub); ok {
t.Fatalf("revoked user must not appear in directory: %+v", resp.Members)
}
if _, ok := findMember(resp.Members, h.alicePub); !ok {
t.Fatalf("active admin alice should still appear: %+v", resp.Members)
}
}
// TestDirectoryEndpointParity pins the server-side endpoint derivation to the
// cross-language parity vector emitted by cmd/busvectors (and consumed by the
// uniweb crypto.ts endpointID test): for a FIXED sign_pub the directory must
// return the exact base64url(sha256(signPub)) endpoint, byte-for-byte. The
// expected value is recomputed here independently of frame.EndpointID so the test
// fails if the handler ever diverges from the canonical construction.
func TestDirectoryEndpointParity(t *testing.T) {
// Vector from cmd/busvectors (seed 000102..1f -> Ed25519 public key).
const vectorSignPub = "03a107bff3ce10be1d70dd18e74bc09967e4d6309ba50d5f1ddc8664125531b8"
const vectorEndpoint = "Vkdap1RjR0wChd9dvyvKtz2mUTWIOem3dIGy6rEHcIw"
// Independent recomputation: base64url(sha256(raw signPub bytes)), unpadded.
raw, err := hex.DecodeString(vectorSignPub)
if err != nil {
t.Fatalf("decode vector sign_pub: %v", err)
}
sum := sha256.Sum256(raw)
if got := base64.RawURLEncoding.EncodeToString(sum[:]); got != vectorEndpoint {
t.Fatalf("vector self-check: recomputed endpoint %q != pinned %q", got, vectorEndpoint)
}
h := newAuthHarness(t, AuthEnforce)
if err := h.store.AddUser(vectorSignPub, "vectorbot", RoleMember); err != nil {
t.Fatalf("add vector user: %v", err)
}
code, resp := directory(t, h, h.alice, 1)
if code != http.StatusOK {
t.Fatalf("directory should be 200, got %d", code)
}
row, ok := findMember(resp.Members, vectorSignPub)
if !ok {
t.Fatalf("vector user missing from directory: %+v", resp.Members)
}
if row.Endpoint != vectorEndpoint {
t.Fatalf("endpoint parity broken: directory returned %q, want %q", row.Endpoint, vectorEndpoint)
}
}
+198
View File
@@ -0,0 +1,198 @@
package membership
// Server-side durable history for persisted rooms (room.ModeMatrix / Persist).
//
// A persisted room's messages ride a file-backed JetStream stream named
// "UNIBUS_<roomID>" (roomStreamName, identical to pkg/client.streamName). Until
// now that stream was created only by the Go client's first publish/subscribe; a
// client that speaks only core NATS (the browser client uniweb, which has no
// JetStream) therefore never created it, so its messages were captured nowhere and
// vanished on reload. This file moves stream ownership to the server: the control
// plane ensures the stream when a persisted room is created (so capture starts at
// minute zero whoever publishes) and exposes GET /rooms/{id}/history so a
// JetStream-less client can read the backlog over plain HTTP.
//
// The server never decrypts: each stored message is the E2E frame exactly as it
// was published (ciphertext for an encrypted room). The history endpoint returns
// those bytes verbatim (base64-encoded for JSON safety), so end-to-end encryption
// is preserved — the server only relays the bytes it already holds.
import (
"context"
"encoding/base64"
"errors"
"fmt"
"net/http"
"strconv"
"time"
"github.com/nats-io/nats.go/jetstream"
)
const (
// defaultHistoryLimit is the number of most-recent messages returned when the
// caller does not specify ?limit.
defaultHistoryLimit = 200
// maxHistoryLimit is the hard ceiling on a single history response, so a caller
// cannot ask the server to buffer an unbounded backlog into one JSON payload.
maxHistoryLimit = 1000
// historyOpTimeout bounds each JetStream operation the history path performs
// (stream lookup/ensure, info, per-message get) so a stalled data plane cannot
// hang a control-plane request indefinitely.
historyOpTimeout = 5 * time.Second
)
// historyResp is the GET /rooms/{id}/history response envelope. messages is the
// ordered (oldest→newest) list of the room's most recent frames, each the base64
// (standard encoding) of the marshaled, still-encrypted frame as it was published.
// The key is a stable contract consumed by the browser client; do not rename it.
type historyResp struct {
Messages []string `json:"messages"`
}
// streamConfigForRoom builds the JetStream stream config for a persisted room.
//
// It MUST stay byte-for-byte compatible with pkg/client/persist.go's ensureStream
// (the original owner of this format): same name derivation (roomStreamName ==
// pkg/client.streamName), same single subject, LimitsPolicy retention, file
// storage. pkg/client is the source of truth for the format; we copy it here
// rather than import it because pkg/client imports pkg/membership and importing it
// back would be a cycle. The only addition is Replicas, matched to the cluster's
// control-plane replication so a persisted room's history is as available as its
// metadata (1 standalone, up to 3 in an HA cluster). CreateOrUpdateStream treats a
// matching config as a no-op, so the client's later ensureStream is harmless.
func streamConfigForRoom(roomID, subject string, replicas int) jetstream.StreamConfig {
if replicas < 1 {
replicas = 1
}
return jetstream.StreamConfig{
Name: roomStreamName(roomID),
Subjects: []string{subject},
Retention: jetstream.LimitsPolicy,
Storage: jetstream.FileStorage,
Replicas: replicas,
}
}
// ensureRoomStream idempotently creates (or no-ops on) the durable stream that
// captures a persisted room's subject. CreateOrUpdateStream returns the existing
// stream unchanged when the config matches, so this is safe to call on every room
// creation and on every history read (lazy backfill of pre-existing rooms).
func ensureRoomStream(ctx context.Context, js jetstream.JetStream, roomID, subject string, replicas int) error {
if _, err := js.CreateOrUpdateStream(ctx, streamConfigForRoom(roomID, subject, replicas)); err != nil {
return fmt.Errorf("membership: ensure stream for room %s: %w", roomID, err)
}
return nil
}
// readRoomHistory returns the last `limit` messages of a room's durable stream in
// chronological order (oldest→newest), each base64-encoded (standard encoding). A
// stream that does not exist yet, or that holds no messages, yields an empty slice
// (not an error): a freshly created or never-used room simply has no history. It
// reads by sequence via the stream MSG.GET API rather than binding a consumer, so
// it has no side effects on any peer's durable ack position. A gap in the sequence
// range (a purged/deleted message) is skipped rather than failing the whole read,
// so the result length is bounded by `limit` but may be smaller.
func readRoomHistory(ctx context.Context, js jetstream.JetStream, roomID string, limit int) ([]string, error) {
out := []string{}
stream, err := js.Stream(ctx, roomStreamName(roomID))
if err != nil {
if errors.Is(err, jetstream.ErrStreamNotFound) {
return out, nil
}
return nil, fmt.Errorf("membership: lookup stream for room %s: %w", roomID, err)
}
si, err := stream.Info(ctx)
if err != nil {
return nil, fmt.Errorf("membership: stream info for room %s: %w", roomID, err)
}
first, last := si.State.FirstSeq, si.State.LastSeq
if si.State.Msgs == 0 || last == 0 {
return out, nil
}
// Window of the last `limit` sequence numbers, clamped to the first stored seq.
// last >= limit guards the unsigned subtraction against underflow.
start := first
if last >= uint64(limit) {
if cand := last - uint64(limit) + 1; cand > start {
start = cand
}
}
for seq := start; seq <= last; seq++ {
raw, err := stream.GetMsg(ctx, seq)
if err != nil {
// A purged/deleted sequence leaves a gap; skip it rather than abort.
continue
}
out = append(out, base64.StdEncoding.EncodeToString(raw.Data))
}
return out, nil
}
// parseHistoryLimit reads the ?limit query value, applying the default when it is
// absent and clamping out-of-range / malformed values to [1, maxHistoryLimit].
func parseHistoryLimit(q string) int {
if q == "" {
return defaultHistoryLimit
}
n, err := strconv.Atoi(q)
if err != nil || n <= 0 {
return defaultHistoryLimit
}
if n > maxHistoryLimit {
return maxHistoryLimit
}
return n
}
// handleRoomHistory serves GET /rooms/{id}/history: the last ?limit (default 200,
// hard cap 1000) messages of a persisted room, oldest→newest, each the base64 of
// the still-encrypted frame as published. The server never decrypts — it relays
// the ciphertext bytes the stream already holds, preserving E2E.
//
// Authorization mirrors the sibling room reads (/key, /members): the request must
// be a member of the room (requireMember; allowed under AuthOff/dev where no signer
// is verified). A missing room is 404; a non-member is 403; an unsigned request
// under enforce is rejected with 401 by the auth middleware before this runs.
//
// For a persisted room the stream is ensured first (lazy backfill): a room created
// before the server managed streams begins capturing from now on. Messages sent
// before the stream existed were never captured and are unrecoverable — only
// messages from stream creation onward appear here.
func (s *Server) handleRoomHistory(w http.ResponseWriter, r *http.Request) {
roomID := r.PathValue("id")
// Existence first so a missing room is a clean 404 (the documented contract),
// distinct from a 403 for an existing room the caller is not a member of.
info, err := s.store.GetRoom(roomID)
if err != nil {
writeErr(w, http.StatusNotFound, "room not found")
return
}
if _, ok := s.requireMember(w, r, roomID); !ok {
return
}
limit := parseHistoryLimit(r.URL.Query().Get("limit"))
// No JetStream wired (e.g. an external-NATS deployment without a cluster/KV
// feature): there is no durable stream to read, so report an empty history
// rather than 500 — a client degrades to "no backlog" gracefully.
if s.js == nil {
writeJSON(w, http.StatusOK, historyResp{Messages: []string{}})
return
}
ctx, cancel := context.WithTimeout(r.Context(), historyOpTimeout)
defer cancel()
if info.Persist {
if err := ensureRoomStream(ctx, s.js, roomID, info.Subject, s.streamReplicas); err != nil {
writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
return
}
}
msgs, err := readRoomHistory(ctx, s.js, roomID, limit)
if err != nil {
writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
return
}
writeJSON(w, http.StatusOK, historyResp{Messages: msgs})
}
+400
View File
@@ -0,0 +1,400 @@
package membership
import (
"context"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"path/filepath"
"testing"
"time"
cs "fn-registry/functions/cybersecurity"
"github.com/enmanuel/unibus/pkg/blobstore"
"github.com/enmanuel/unibus/pkg/embeddednats"
"github.com/enmanuel/unibus/pkg/frame"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
)
// historyHarness is an enforce-mode control plane wired to a real embedded NATS
// JetStream, so the history path exercises the production code: the server ensures
// and reads actual durable streams. alice is a seeded admin (and any room's owner),
// bob is a registered user added as a room member, and carol is a registered user
// that is NOT a member of the test room (to exercise the 403 path).
type historyHarness struct {
ts *httptest.Server
store Store
js jetstream.JetStream
nc *nats.Conn
alice cs.Identity // admin + room owner
bob cs.Identity // room member
carol cs.Identity // registered, non-member
}
func newHistoryHarness(t *testing.T) *historyHarness {
t.Helper()
dir := t.TempDir()
ns, err := embeddednats.StartServer(embeddednats.ServerConfig{
StoreDir: filepath.Join(dir, "jetstream"),
Host: "127.0.0.1",
Port: kvFreePort(t),
})
if err != nil {
t.Fatalf("embedded nats: %v", err)
}
nc, err := nats.Connect(ns.ClientURL())
if err != nil {
ns.Shutdown()
t.Fatalf("nats connect: %v", err)
}
js, err := jetstream.New(nc)
if err != nil {
nc.Close()
ns.Shutdown()
t.Fatalf("jetstream: %v", err)
}
store, err := Open(filepath.Join(dir, "unibus.db"))
if err != nil {
nc.Close()
ns.Shutdown()
t.Fatalf("open store: %v", err)
}
blobs, err := blobstore.New(filepath.Join(dir, "blobs"))
if err != nil {
store.Close()
nc.Close()
ns.Shutdown()
t.Fatalf("open blobs: %v", err)
}
mustID := func(name string) cs.Identity {
id, err := cs.GenerateIdentity()
if err != nil {
t.Fatalf("identity %s: %v", name, err)
}
return id
}
alice, bob, carol := mustID("alice"), mustID("bob"), mustID("carol")
if err := store.AddUser(hex.EncodeToString(alice.SignPub), "alice", RoleAdmin); err != nil {
t.Fatalf("seed admin: %v", err)
}
for _, u := range []struct {
id cs.Identity
handle string
}{{bob, "bob"}, {carol, "carol"}} {
if err := store.AddUser(hex.EncodeToString(u.id.SignPub), u.handle, RoleMember); err != nil {
t.Fatalf("register %s: %v", u.handle, err)
}
}
srv := NewServer(store, blobs, AuthEnforce)
srv.SetJetStream(js, 1)
ts := httptest.NewServer(srv)
t.Cleanup(func() {
ts.Close()
store.Close()
nc.Close()
ns.Shutdown()
ns.WaitForShutdown()
})
return &historyHarness{ts: ts, store: store, js: js, nc: nc, alice: alice, bob: bob, carol: carol}
}
// seedPersistRoom creates a persisted (Matrix-policy) room directly in the store
// with alice as owner and bob as a member, returning its id and subject. It does
// NOT create the stream — that is left to the code under test (handleCreateRoom or
// the lazy ensure in the history endpoint), which is exactly what we want to verify.
func (h *historyHarness) seedPersistRoom(t *testing.T) (roomID, subject string) {
t.Helper()
roomID = newULID()
subject = "unibus.room." + roomID
aliceEp := frame.EndpointID(h.alice.SignPub)
info := RoomInfo{RoomID: roomID, Subject: subject, OwnerEndpoint: aliceEp, Encrypt: true, Persist: true}
if err := h.store.CreateRoom(info, h.alice.SignPub, h.alice.KexPub, []byte("alice-sealed")); err != nil {
t.Fatalf("seed room: %v", err)
}
bobEp := frame.EndpointID(h.bob.SignPub)
bobM := Member{Endpoint: bobEp, Role: RoleMember, SignPub: h.bob.SignPub, KexPub: h.bob.KexPub}
if err := h.store.AddMember(roomID, bobM, 0, []byte("bob-sealed")); err != nil {
t.Fatalf("add member bob: %v", err)
}
return roomID, subject
}
// makeFrame builds a marshaled PUB frame whose payload identifies it, so a test can
// assert exact bytes and ordering after a round trip through the stream + endpoint.
func makeFrame(t *testing.T, subject, sender string, i int) []byte {
t.Helper()
f := frame.Frame{
Type: frame.PUB,
Subject: subject,
Sender: sender,
MsgID: fmt.Sprintf("msg-%02d", i),
Payload: []byte(fmt.Sprintf("ciphertext-%02d", i)),
}
b, err := f.Marshal()
if err != nil {
t.Fatalf("marshal frame %d: %v", i, err)
}
return b
}
// getHistory signs a GET /rooms/{id}/history request as id and returns the status,
// the raw body, and the decoded envelope. query is the raw query string (e.g.
// "limit=2") or "". The signed path includes the query because the server verifies
// the signature over r.URL.RequestURI(), which carries it.
func (h *historyHarness) getHistory(t *testing.T, id cs.Identity, roomID, query string, n int) (int, string, historyResp) {
t.Helper()
path := "/rooms/" + roomID + "/history"
if query != "" {
path += "?" + query
}
req := signedReq(t, h.ts.URL, "GET", path, nil, id, time.Now().Unix(), nonceN(n))
code, body := do(t, req)
var out historyResp
if code == 200 {
if err := json.Unmarshal([]byte(body), &out); err != nil {
t.Fatalf("decode history: %v (%s)", err, body)
}
}
return code, body, out
}
// TestCreateRoomEnsuresStream verifies handleCreateRoom creates the durable stream
// for a persisted room before responding, so capture starts at room creation.
func TestCreateRoomEnsuresStream(t *testing.T) {
h := newHistoryHarness(t)
aliceEp := frame.EndpointID(h.alice.SignPub)
reqBody := createRoomReq{
Subject: "unibus.room.created",
Policy: policyJSON{Encrypt: true, Persist: true},
Owner: endpointJSON{Endpoint: aliceEp, SignPub: h.alice.SignPub, KexPub: h.alice.KexPub},
SealedKeySelf: []byte("alice-sealed"),
}
body, _ := json.Marshal(reqBody)
req := signedReq(t, h.ts.URL, "POST", "/rooms", body, h.alice, time.Now().Unix(), nonceN(1))
code, respBody := do(t, req)
if code != 201 {
t.Fatalf("create room: want 201, got %d (%s)", code, respBody)
}
var cr createRoomResp
if err := json.Unmarshal([]byte(respBody), &cr); err != nil {
t.Fatalf("decode create resp: %v (%s)", err, respBody)
}
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
if _, err := h.js.Stream(ctx, roomStreamName(cr.RoomID)); err != nil {
t.Fatalf("stream for created persist room should exist: %v", err)
}
}
// TestRoomHistoryGolden is the golden path: three frames published to a persisted
// room's stream come back from the endpoint base64-encoded, in chronological order,
// and decode to the exact frames that were published.
func TestRoomHistoryGolden(t *testing.T) {
h := newHistoryHarness(t)
roomID, subject := h.seedPersistRoom(t)
bobEp := frame.EndpointID(h.bob.SignPub)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := ensureRoomStream(ctx, h.js, roomID, subject, 1); err != nil {
t.Fatalf("ensure stream: %v", err)
}
want := make([][]byte, 3)
for i := 0; i < 3; i++ {
want[i] = makeFrame(t, subject, bobEp, i)
// js.Publish waits for the stream ack, so the message is durably stored before
// the next iteration — no sleeps, deterministic ordering.
if _, err := h.js.Publish(ctx, subject, want[i]); err != nil {
t.Fatalf("publish %d: %v", i, err)
}
}
code, raw, hr := h.getHistory(t, h.bob, roomID, "", 10)
if code != 200 {
t.Fatalf("history: want 200, got %d (%s)", code, raw)
}
if len(hr.Messages) != 3 {
t.Fatalf("want 3 messages, got %d (%s)", len(hr.Messages), raw)
}
for i, m := range hr.Messages {
decoded, err := base64.StdEncoding.DecodeString(m)
if err != nil {
t.Fatalf("message %d not valid base64: %v", i, err)
}
if string(decoded) != string(want[i]) {
t.Fatalf("message %d bytes mismatch (order or content)", i)
}
f, err := frame.Unmarshal(decoded)
if err != nil {
t.Fatalf("message %d does not decode to a frame: %v", i, err)
}
if f.MsgID != fmt.Sprintf("msg-%02d", i) {
t.Fatalf("message %d: want MsgID msg-%02d, got %q", i, i, f.MsgID)
}
}
}
// TestRoomHistoryCapturesCoreNATSPublish proves the central fix: a message
// published over PLAIN core NATS (as the JetStream-less browser client uniweb does)
// is captured by the server-owned stream and served by the endpoint. Without the
// server ensuring the stream, this message would be captured nowhere.
func TestRoomHistoryCapturesCoreNATSPublish(t *testing.T) {
h := newHistoryHarness(t)
roomID, subject := h.seedPersistRoom(t)
bobEp := frame.EndpointID(h.bob.SignPub)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := ensureRoomStream(ctx, h.js, roomID, subject, 1); err != nil {
t.Fatalf("ensure stream: %v", err)
}
sent := makeFrame(t, subject, bobEp, 7)
if err := h.nc.Publish(subject, sent); err != nil {
t.Fatalf("core publish: %v", err)
}
if err := h.nc.Flush(); err != nil {
t.Fatalf("flush: %v", err)
}
// Core NATS publish has no stream ack; poll the stream until the message lands.
h.waitMsgs(t, roomID, 1)
code, raw, hr := h.getHistory(t, h.bob, roomID, "", 11)
if code != 200 {
t.Fatalf("history: want 200, got %d (%s)", code, raw)
}
if len(hr.Messages) != 1 {
t.Fatalf("want 1 captured message, got %d (%s)", len(hr.Messages), raw)
}
decoded, err := base64.StdEncoding.DecodeString(hr.Messages[0])
if err != nil || string(decoded) != string(sent) {
t.Fatalf("captured core-NATS message round-trip mismatch (err=%v)", err)
}
}
// TestRoomHistoryLimit verifies ?limit caps the response to the most recent N
// messages, oldest→newest within the window.
func TestRoomHistoryLimit(t *testing.T) {
h := newHistoryHarness(t)
roomID, subject := h.seedPersistRoom(t)
bobEp := frame.EndpointID(h.bob.SignPub)
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := ensureRoomStream(ctx, h.js, roomID, subject, 1); err != nil {
t.Fatalf("ensure stream: %v", err)
}
for i := 0; i < 5; i++ {
if _, err := h.js.Publish(ctx, subject, makeFrame(t, subject, bobEp, i)); err != nil {
t.Fatalf("publish %d: %v", i, err)
}
}
code, raw, hr := h.getHistory(t, h.bob, roomID, "limit=2", 12)
if code != 200 {
t.Fatalf("history: want 200, got %d (%s)", code, raw)
}
if len(hr.Messages) != 2 {
t.Fatalf("limit=2 over 5 messages: want 2, got %d", len(hr.Messages))
}
// The window is the last two messages (indices 3 and 4), in order.
for off, m := range hr.Messages {
decoded, _ := base64.StdEncoding.DecodeString(m)
f, err := frame.Unmarshal(decoded)
if err != nil {
t.Fatalf("limited message %d does not decode: %v", off, err)
}
want := fmt.Sprintf("msg-%02d", off+3)
if f.MsgID != want {
t.Fatalf("limited message %d: want MsgID %s, got %q", off, want, f.MsgID)
}
}
}
// TestRoomHistoryEmptyRoom verifies a persisted room with no messages returns an
// empty (non-null) array, lazily ensuring the stream on the way.
func TestRoomHistoryEmptyRoom(t *testing.T) {
h := newHistoryHarness(t)
roomID, _ := h.seedPersistRoom(t)
code, raw, hr := h.getHistory(t, h.bob, roomID, "", 13)
if code != 200 {
t.Fatalf("history: want 200, got %d (%s)", code, raw)
}
if hr.Messages == nil {
t.Fatalf("empty room must return [] not null (%s)", raw)
}
if len(hr.Messages) != 0 {
t.Fatalf("empty room: want 0 messages, got %d", len(hr.Messages))
}
// The lazy ensure should have created the stream even though no message exists.
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
if _, err := h.js.Stream(ctx, roomStreamName(roomID)); err != nil {
t.Fatalf("lazy ensure should have created the stream: %v", err)
}
}
// TestRoomHistoryUnauthenticated verifies an unsigned request is rejected with 401
// under enforce, before the handler runs.
func TestRoomHistoryUnauthenticated(t *testing.T) {
h := newHistoryHarness(t)
roomID, _ := h.seedPersistRoom(t)
// No signing headers: plain GET against the enforce-mode control plane.
req, err := http.NewRequest("GET", h.ts.URL+"/rooms/"+roomID+"/history", nil)
if err != nil {
t.Fatalf("new request: %v", err)
}
code, body := do(t, req)
if code != 401 {
t.Fatalf("unauthenticated history: want 401, got %d (%s)", code, body)
}
}
// TestRoomHistoryNonMember verifies a registered user who is NOT a member of the
// room is rejected with 403.
func TestRoomHistoryNonMember(t *testing.T) {
h := newHistoryHarness(t)
roomID, _ := h.seedPersistRoom(t)
code, body, _ := h.getHistory(t, h.carol, roomID, "", 14)
if code != 403 {
t.Fatalf("non-member history: want 403, got %d (%s)", code, body)
}
}
// TestRoomHistoryRoomNotFound verifies a request for a non-existent room is a 404,
// distinct from the 403 a non-member of an existing room gets.
func TestRoomHistoryRoomNotFound(t *testing.T) {
h := newHistoryHarness(t)
code, body, _ := h.getHistory(t, h.alice, newULID(), "", 15)
if code != 404 {
t.Fatalf("missing room history: want 404, got %d (%s)", code, body)
}
}
// waitMsgs polls the room's stream until it holds at least want messages or a short
// deadline elapses, so a core-NATS publish (which carries no stream ack) is observed
// deterministically without a fixed sleep.
func (h *historyHarness) waitMsgs(t *testing.T, roomID string, want uint64) {
t.Helper()
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
st, err := h.js.Stream(ctx, roomStreamName(roomID))
if err == nil {
si, ierr := st.Info(ctx)
if ierr == nil && si.State.Msgs >= want {
cancel()
return
}
}
cancel()
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("stream for room %s never reached %d message(s)", roomID, want)
}
+111 -8
View File
@@ -1,8 +1,10 @@
package membership package membership
import ( import (
"fmt"
"net" "net"
"net/http" "net/http"
"strings"
"sync" "sync"
"time" "time"
@@ -78,16 +80,117 @@ func (l *ipRateLimiter) reapLocked(now time.Time) {
} }
} }
// clientIP extracts the source IP of an HTTP request, stripping the port. It // clientIP extracts the rate-limit key for a request: the source IP, with the
// trusts the transport's RemoteAddr only (no X-Forwarded-For parsing): a public // port stripped. By default it trusts the transport's RemoteAddr ONLY (no
// deployment terminates TLS at this process or behind a proxy that the operator // X-Forwarded-For parsing): honoring an attacker-supplied header would let a
// controls, and honoring an attacker-supplied header would let a single IP fan // single IP fan its quota across forged identities. When the operator runs the
// its quota across forged identities. If parsing fails the whole RemoteAddr is // control plane behind a reverse proxy they control (the same-origin Caddy
// used as the key (still a stable per-connection bucket). // deployment), SetTrustedProxies names that proxy's address(es); only then, and
func clientIP(r *http.Request) string { // only when the immediate peer is one of them, is the forwarded client IP
// believed. This keeps the per-IP rate limit meaningful behind the proxy, where
// every request would otherwise share the proxy's single IP. If parsing fails the
// whole RemoteAddr is used as the key (still a stable per-connection bucket).
func (s *Server) clientIP(r *http.Request) string {
host, _, err := net.SplitHostPort(r.RemoteAddr) host, _, err := net.SplitHostPort(r.RemoteAddr)
if err != nil { if err != nil {
return r.RemoteAddr host = r.RemoteAddr
}
if !s.trustedProxies.has(host) {
return host
}
if fwd := forwardedClientIP(r, s.trustedProxies); fwd != "" {
return fwd
} }
return host return host
} }
// forwardedClientIP returns the real client IP a trusted proxy reported, or "" if
// none is present. X-Forwarded-For is read RIGHT-TO-LEFT: the rightmost entry is
// the one our immediate (trusted) proxy appended and therefore cannot be spoofed
// by the client, which can only prepend entries to the left. Trusted-proxy hops
// are skipped so a chain of proxies we own resolves to the first address none of
// them owns — the actual external client. X-Real-IP is a single-value fallback for
// proxies that set it instead. A non-trusted immediate peer never reaches here, so
// a direct attacker's forged header is ignored entirely.
func forwardedClientIP(r *http.Request, trusted trustedProxyMatcher) string {
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
parts := strings.Split(xff, ",")
for i := len(parts) - 1; i >= 0; i-- {
ip := strings.TrimSpace(parts[i])
if ip == "" || trusted.has(ip) {
continue
}
if net.ParseIP(ip) != nil {
return ip
}
}
}
if xrip := strings.TrimSpace(r.Header.Get("X-Real-IP")); xrip != "" {
if net.ParseIP(xrip) != nil {
return xrip
}
}
return ""
}
// trustedProxyMatcher is the set of reverse-proxy addresses whose forwarding
// headers may be honored. The zero value (nil) matches nothing, so the default
// behavior is RemoteAddr-only.
type trustedProxyMatcher []*net.IPNet
// SetTrustedProxies configures the proxies whose X-Forwarded-For / X-Real-IP this
// server trusts for the per-IP rate limit. Each entry is an IP (treated as a /32
// or /128) or a CIDR. It returns an error on the first unparseable entry and
// leaves the previous configuration unchanged. Passing no entries clears the set.
func (s *Server) SetTrustedProxies(entries []string) error {
m, err := parseTrustedProxies(entries)
if err != nil {
return err
}
s.trustedProxies = m
return nil
}
// parseTrustedProxies turns a list of IPs/CIDRs into a matcher. A bare IP becomes
// a host route (/32 for IPv4, /128 for IPv6); blanks are skipped.
func parseTrustedProxies(entries []string) (trustedProxyMatcher, error) {
var m trustedProxyMatcher
for _, e := range entries {
e = strings.TrimSpace(e)
if e == "" {
continue
}
if _, ipnet, err := net.ParseCIDR(e); err == nil {
m = append(m, ipnet)
continue
}
ip := net.ParseIP(e)
if ip == nil {
return nil, fmt.Errorf("trusted proxy %q is not an IP or CIDR", e)
}
bits := 32
if ip.To4() == nil {
bits = 128
}
m = append(m, &net.IPNet{IP: ip, Mask: net.CIDRMask(bits, bits)})
}
return m, nil
}
// has reports whether host (an IP string with no port) falls inside any trusted
// range. A nil matcher and an unparseable host both report false.
func (m trustedProxyMatcher) has(host string) bool {
if len(m) == 0 {
return false
}
ip := net.ParseIP(host)
if ip == nil {
return false
}
for _, n := range m {
if n.Contains(ip) {
return true
}
}
return false
}
+113
View File
@@ -0,0 +1,113 @@
package membership
import (
"net/http"
"testing"
)
// TestClientIPTrustedProxy covers the rate-limit key extraction behind a reverse
// proxy: forwarding headers are believed ONLY when the immediate peer is a
// configured trusted proxy, and never otherwise. This is what keeps the per-IP
// rate limit per-client once the control plane runs behind the same-origin Caddy
// proxy, without opening a quota-fanning hole for a direct attacker.
func TestClientIPTrustedProxy(t *testing.T) {
const caddy = "135.125.201.30"
cases := []struct {
name string
proxies []string
remote string
xff string
xRealIP string
want string
}{
{
name: "no trusted proxies ignores XFF",
remote: "203.0.113.7:5000",
xff: "1.2.3.4",
want: "203.0.113.7",
},
{
name: "trusted proxy honors XFF client",
proxies: []string{caddy},
remote: caddy + ":4451",
xff: "198.51.100.23",
want: "198.51.100.23",
},
{
name: "loopback proxy honors XFF (magnus-local hop)",
proxies: []string{"127.0.0.1/32", "::1/128"},
remote: "127.0.0.1:33344",
xff: "198.51.100.99",
want: "198.51.100.99",
},
{
name: "untrusted peer cannot spoof XFF",
proxies: []string{caddy},
remote: "203.0.113.7:5000",
xff: "10.0.0.1",
want: "203.0.113.7",
},
{
name: "XFF read right-to-left, trusted hops skipped",
proxies: []string{caddy},
remote: caddy + ":4451",
xff: "198.51.100.23, " + caddy,
want: "198.51.100.23",
},
{
name: "client-prepended forgery is skipped, real appended wins",
proxies: []string{caddy},
remote: caddy + ":4451",
xff: "9.9.9.9, 198.51.100.23",
want: "198.51.100.23",
},
{
name: "X-Real-IP fallback when no XFF",
proxies: []string{caddy},
remote: caddy + ":4451",
xRealIP: "198.51.100.77",
want: "198.51.100.77",
},
{
name: "trusted peer but no forwarding header falls back to peer",
proxies: []string{caddy},
remote: caddy + ":4451",
want: caddy,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
s := &Server{}
if len(tc.proxies) > 0 {
if err := s.SetTrustedProxies(tc.proxies); err != nil {
t.Fatalf("SetTrustedProxies(%v): %v", tc.proxies, err)
}
}
r, _ := http.NewRequest(http.MethodGet, "/rooms", nil)
r.RemoteAddr = tc.remote
if tc.xff != "" {
r.Header.Set("X-Forwarded-For", tc.xff)
}
if tc.xRealIP != "" {
r.Header.Set("X-Real-IP", tc.xRealIP)
}
if got := s.clientIP(r); got != tc.want {
t.Fatalf("clientIP = %q, want %q", got, tc.want)
}
})
}
}
// TestParseTrustedProxiesRejectsGarbage proves a malformed entry is a hard error
// (the command turns it into a startup failure) rather than a silently ignored
// misconfiguration that would leave the rate limit collapsed behind the proxy.
func TestParseTrustedProxiesRejectsGarbage(t *testing.T) {
if _, err := parseTrustedProxies([]string{"not-an-ip"}); err == nil {
t.Fatal("expected error for non-IP/CIDR entry, got nil")
}
if _, err := parseTrustedProxies([]string{"10.0.0.0/8", "127.0.0.1"}); err != nil {
t.Fatalf("valid entries rejected: %v", err)
}
}
+195 -1
View File
@@ -3,6 +3,7 @@ package membership
import ( import (
"bytes" "bytes"
"context" "context"
"encoding/hex"
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
@@ -87,6 +88,42 @@ type Server struct {
// posture a secure cluster requires (audit 0008 N1). It is set by the command; // posture a secure cluster requires (audit 0008 N1). It is set by the command;
// the zero value (all false) reflects an unsecured dev node. // the zero value (all false) reflects an unsecured dev node.
Posture Posture Posture Posture
// AllowedOrigins is the CORS allowlist of browser Origin headers permitted to
// call the control plane cross-origin. It exists so a browser-native client
// (uniweb) can talk to membershipd directly, the way the Go/Kotlin clients
// already do over a non-browser transport (issue uniweb/0001). Native clients
// send no Origin header and are unaffected. The zero value (empty) keeps CORS
// OFF — no Access-Control headers are emitted and the server behaves exactly as
// before — so this is opt-in per deployment. Entries are matched exactly (scheme
// + host + port); never use "*" with credentials. Set by the command from a flag.
AllowedOrigins []string
// trustedProxies names the reverse proxies whose forwarding headers
// (X-Forwarded-For / X-Real-IP) the rate limiter is allowed to believe. It
// exists for the same-origin deployment where a single proxy (Caddy) fronts
// the control plane: without it every proxied request would share the proxy's
// one IP and collapse the per-IP rate limit into a single bucket for the whole
// world. Only when the immediate peer is one of these addresses is the
// forwarded client IP trusted; the zero value (nil) trusts nobody, preserving
// the RemoteAddr-only behavior that predates the flag. Set by the command via
// SetTrustedProxies. See clientIP.
trustedProxies trustedProxyMatcher
// js is the privileged JetStream context the server uses to own the durable
// per-room streams of persisted rooms: it ensures a room's stream on creation
// so the room's subject is captured from the first message — even from a
// JetStream-less browser client (uniweb) that speaks only core NATS — and reads
// it back for GET /rooms/{id}/history. It is wired by the command via
// SetJetStream whenever a JetStream-capable data plane is available (always for
// the embedded server). nil leaves history empty and stream-ensure a no-op,
// preserving the pre-feature behavior for a deployment without JetStream.
js jetstream.JetStream
// streamReplicas is the replication factor for the room streams the server
// creates, matched to the cluster's control-plane (KV) replication — 1 for a
// standalone node, up to 3 in an HA cluster — so a persisted room's history is
// as available as its metadata. Used only when js != nil. See SetJetStream.
streamReplicas int
} }
// Posture describes the security posture a membershipd node runs with. It is // Posture describes the security posture a membershipd node runs with. It is
@@ -121,6 +158,19 @@ func NewServer(store Store, blobs blobstore.Store, authMode AuthMode) *Server {
return s return s
} }
// SetJetStream wires the privileged JetStream context (and the room-stream
// replication factor) the server uses to ensure and read the durable streams of
// persisted rooms. replicas below 1 is clamped to 1. It must be called once at
// startup, before the server begins serving; leaving it unset keeps history empty
// and stream-ensure a no-op, the behavior for a deployment without JetStream.
func (s *Server) SetJetStream(js jetstream.JetStream, replicas int) {
if replicas < 1 {
replicas = 1
}
s.js = js
s.streamReplicas = replicas
}
// UseReplicatedNonces switches the server's anti-replay store from the // UseReplicatedNonces switches the server's anti-replay store from the
// per-process in-memory cache to a JetStream KV bucket shared across the cluster // per-process in-memory cache to a JetStream KV bucket shared across the cluster
// (issue 0003e). It MUST be called on every node of a multi-node deployment: // (issue 0003e). It MUST be called on every node of a multi-node deployment:
@@ -143,10 +193,19 @@ func (s *Server) UseReplicatedNonces(js jetstream.JetStream, replicas int) error
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
now := time.Now() now := time.Now()
// CORS runs before everything else so a browser preflight never pays the rate
// limit or auth cost. When the request carries an allowed Origin we echo the
// Access-Control headers; a preflight (OPTIONS) is answered here and short-
// circuits the pipeline. With an empty allowlist this is a no-op, so non-browser
// clients and untouched deployments behave exactly as before (issue uniweb/0001).
if s.applyCORS(w, r) {
return // preflight handled
}
// Per-IP rate limit runs first, ahead of auth and body reads, so a flood is // Per-IP rate limit runs first, ahead of auth and body reads, so a flood is
// shed at the cheapest possible point. The health probe is exempt so liveness // shed at the cheapest possible point. The health probe is exempt so liveness
// checks are never throttled. // checks are never throttled.
if !isAuthExempt(r) && !s.limiter.allow(clientIP(r), now) { if !isAuthExempt(r) && !s.limiter.allow(s.clientIP(r), now) {
writeErr(w, http.StatusTooManyRequests, "rate limit exceeded") writeErr(w, http.StatusTooManyRequests, "rate limit exceeded")
return return
} }
@@ -221,6 +280,57 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
s.mux.ServeHTTP(w, r.WithContext(withSigner(r.Context(), res.endpoint, res.pubHex))) s.mux.ServeHTTP(w, r.WithContext(withSigner(r.Context(), res.endpoint, res.pubHex)))
} }
// applyCORS handles cross-origin requests for the control plane. When the request
// carries an Origin in the allowlist it sets the Access-Control-Allow-* response
// headers so the browser accepts the eventual response; when the request is a CORS
// preflight (OPTIONS) it writes the preflight reply and returns true so ServeHTTP
// short-circuits before the rate limiter and auth ever run. It returns false for
// every non-preflight request — including same-origin and native clients that send
// no Origin header — leaving the normal pipeline to run unchanged. With an empty
// AllowedOrigins it never sets a header (CORS is off): the opt-in default.
func (s *Server) applyCORS(w http.ResponseWriter, r *http.Request) (preflight bool) {
origin := r.Header.Get("Origin")
allowed := origin != "" && s.originAllowed(origin)
if allowed {
h := w.Header()
h.Set("Access-Control-Allow-Origin", origin)
// Vary: Origin so a cache never serves an allow-listed response to another
// origin. Add (not Set) to preserve any Vary the handler may add later.
h.Add("Vary", "Origin")
h.Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
// Allow the control-plane request-auth headers a browser client signs every
// request with (busauth.signedHeaders), or the browser's CORS preflight blocks
// the real request. Content-Type/Authorization stay for JSON bodies.
h.Set("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Unibus-Pub, X-Unibus-Ts, X-Unibus-Nonce, X-Unibus-Sig")
h.Set("Access-Control-Max-Age", "600")
}
if r.Method == http.MethodOptions {
// Answer the preflight here so it never reaches the rate limiter or auth. An
// allowed origin gets 204 with the headers above; a disallowed or missing
// origin gets 403 with no Access-Control headers, so the browser blocks the
// real cross-origin request.
if allowed {
w.WriteHeader(http.StatusNoContent)
} else {
w.WriteHeader(http.StatusForbidden)
}
return true
}
return false
}
// originAllowed reports whether origin is in the CORS allowlist. Matching is exact
// (scheme + host + port): a browser Origin is an opaque string, so an exact compare
// is both correct and the safest policy (no wildcard, no suffix tricks).
func (s *Server) originAllowed(origin string) bool {
for _, o := range s.AllowedOrigins {
if o == origin {
return true
}
}
return false
}
// isBodyTooLarge reports whether err is the sentinel returned by MaxBytesReader // isBodyTooLarge reports whether err is the sentinel returned by MaxBytesReader
// when the body exceeds its limit, so the middleware can map it to 413. // when the body exceeds its limit, so the middleware can map it to 413.
func isBodyTooLarge(err error) bool { func isBodyTooLarge(err error) bool {
@@ -321,6 +431,13 @@ func (s *Server) routes() {
s.mux.HandleFunc("POST /rooms/{id}/invite", s.handleInvite) s.mux.HandleFunc("POST /rooms/{id}/invite", s.handleInvite)
s.mux.HandleFunc("GET /rooms/{id}/key", s.handleGetKey) s.mux.HandleFunc("GET /rooms/{id}/key", s.handleGetKey)
s.mux.HandleFunc("GET /rooms/{id}/members", s.handleListMembers) s.mux.HandleFunc("GET /rooms/{id}/members", s.handleListMembers)
// Durable message history for a persisted room, read server-side from the room's
// JetStream stream so a client without JetStream (the browser client uniweb) can
// load the backlog over plain HTTP. Member-only, like /key and /members.
// Registered without the /api prefix like every other control-plane route: Caddy
// strips /api via handle_path /api/* before forwarding, so the SPA's
// GET /api/rooms/{id}/history arrives here as GET /rooms/{id}/history.
s.mux.HandleFunc("GET /rooms/{id}/history", s.handleRoomHistory)
s.mux.HandleFunc("GET /members/{endpoint}/rooms", s.handleListMemberRooms) s.mux.HandleFunc("GET /members/{endpoint}/rooms", s.handleListMemberRooms)
s.mux.HandleFunc("POST /rooms/{id}/rekey", s.handleRekey) s.mux.HandleFunc("POST /rooms/{id}/rekey", s.handleRekey)
s.mux.HandleFunc("GET /rooms/{id}", s.handleGetRoom) s.mux.HandleFunc("GET /rooms/{id}", s.handleGetRoom)
@@ -333,6 +450,15 @@ func (s *Server) routes() {
s.mux.HandleFunc("GET /users", s.handleListUsers) s.mux.HandleFunc("GET /users", s.handleListUsers)
s.mux.HandleFunc("POST /users", s.handleAddUser) s.mux.HandleFunc("POST /users", s.handleAddUser)
s.mux.HandleFunc("POST /users/{signpub}/revoke", s.handleRevokeUser) s.mux.HandleFunc("POST /users/{signpub}/revoke", s.handleRevokeUser)
// Member directory — any authenticated bus user (member or admin) may map an
// endpoint id back to its human handle, so clients can render readable sender
// names instead of raw endpoint hashes. Unlike /users it is NOT admin-only and
// returns only active users; under enforce the auth middleware already rejects
// an unauthenticated caller with 401 before this handler runs (uniweb/0002).
// Registered without the /api prefix like every other control-plane route:
// Caddy strips /api via handle_path /api/* before forwarding to membershipd,
// so the SPA's GET /api/directory arrives here as GET /directory.
s.mux.HandleFunc("GET /directory", s.handleDirectory)
} }
// ---- wire types ----------------------------------------------------------- // ---- wire types -----------------------------------------------------------
@@ -431,6 +557,24 @@ type addUserReq struct {
Role string `json:"role"` Role string `json:"role"`
} }
// directoryMember is one entry of the GET /directory response: enough for a
// client to map a message's endpoint id (which the bus stamps on every frame)
// back to a readable handle. endpoint is derived server-side from sign_pub with
// the SAME construction the bus uses (frame.EndpointID = base64url(sha256(signPub)),
// unpadded), so it matches the sender id a client already has byte-for-byte.
type directoryMember struct {
SignPub string `json:"sign_pub"`
Endpoint string `json:"endpoint"`
Handle string `json:"handle"`
Role string `json:"role"`
}
// directoryResp is the GET /directory response envelope. The members key is a
// stable contract consumed by the browser client; do not rename it.
type directoryResp struct {
Members []directoryMember `json:"members"`
}
// ---- helpers -------------------------------------------------------------- // ---- helpers --------------------------------------------------------------
func writeJSON(w http.ResponseWriter, code int, v any) { func writeJSON(w http.ResponseWriter, code int, v any) {
@@ -523,6 +667,21 @@ func (s *Server) handleCreateRoom(w http.ResponseWriter, r *http.Request) {
SignMsgs: req.Policy.SignMsgs, SignMsgs: req.Policy.SignMsgs,
OwnerEndpoint: req.Owner.Endpoint, OwnerEndpoint: req.Owner.Endpoint,
} }
// Own the durable stream for a persisted room (issue room-history): ensure it
// BEFORE the room row is written so the subject is captured from the very first
// message whoever publishes it — a Go client OR a JetStream-less browser client.
// Done first so a stream failure aborts cleanly with no orphan room row (the
// rare orphan empty stream it can leave is harmless and idempotently reused).
// Skipped when no JetStream is wired: the room still works, just without history.
if info.Persist && s.js != nil {
ctx, cancel := context.WithTimeout(r.Context(), historyOpTimeout)
err := ensureRoomStream(ctx, s.js, roomID, info.Subject, s.streamReplicas)
cancel()
if err != nil {
writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
return
}
}
if err := s.store.CreateRoom(info, req.Owner.SignPub, req.Owner.KexPub, req.SealedKeySelf); err != nil { if err := s.store.CreateRoom(info, req.Owner.SignPub, req.Owner.KexPub, req.SealedKeySelf); err != nil {
writeServerErr(w, r, http.StatusInternalServerError, "internal error", err) writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
return return
@@ -776,6 +935,41 @@ func (s *Server) handleListUsers(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, out) writeJSON(w, http.StatusOK, out)
} }
// handleDirectory returns the active bus user directory so a client can resolve a
// sender's endpoint id to a readable handle. Unlike handleListUsers it is NOT
// admin-only: every authenticated bus user may read it (the auth middleware has
// already verified the caller is an active user under enforce, and rejected an
// unauthenticated one with 401). Only active users are listed, and each endpoint
// is computed server-side from the user's sign_pub with frame.EndpointID — the
// exact derivation the bus stamps on every frame, so the returned endpoint matches
// the sender id a client already holds. A user with a malformed sign_pub (which
// the add path rejects, so this is defensive) is skipped rather than failing the
// whole listing.
func (s *Server) handleDirectory(w http.ResponseWriter, r *http.Request) {
users, err := s.store.ListUsers()
if err != nil {
writeServerErr(w, r, http.StatusInternalServerError, "internal error", err)
return
}
out := make([]directoryMember, 0, len(users))
for _, u := range users {
if u.Status != StatusActive {
continue
}
signPub, err := hex.DecodeString(u.SignPub)
if err != nil || len(signPub) != 32 {
continue
}
out = append(out, directoryMember{
SignPub: u.SignPub,
Endpoint: frame.EndpointID(signPub),
Handle: u.Handle,
Role: u.Role,
})
}
writeJSON(w, http.StatusOK, directoryResp{Members: out})
}
// handleAddUser registers a new bus user from an admin-supplied Ed25519 signing // handleAddUser registers a new bus user from an admin-supplied Ed25519 signing
// key. It mirrors the `membershipd user add` CLI: the key must be 64-hex, the // key. It mirrors the `membershipd user add` CLI: the key must be 64-hex, the
// role must be admin or member (empty defaults to member), and re-adding an // role must be admin or member (empty defaults to member), and re-adding an