diff --git a/DESIGN.md b/DESIGN.md index 9725b38..b919ff4 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -94,26 +94,58 @@ Node API (defined once): `child`, `as`, `get(key?)`, `getMany`, `set(value)` / ` `handle(Request) → Response`): `POST /` with Extended JSON bodies; `scan` streams NDJSON. Unreachable server (fetch failure, 502/503/504) → `StorageUnavailableError`. The server exposes the whole underlying store (primitives use absolute keys). - - `KVStorageComposite`: see 4.1. -- **Change stream (SSE).** `GET /events` on the server streams every change to the wrapped storage - as `event: change` with id `-` (epoch = server instance). The server keeps the last + - `KVStorageComposite`: see 4.2. +- **Change stream (SSE).** `POST /events` with `{ paths }` streams changes to those keys and their + subtrees as `event: change` with id `-` (epoch = server instance). The server keeps the last `historySize` (10 000) change batches. - A client reconnecting with `Last-Event-ID` gets the batches it missed, or `event: resync` if it can't be resumed (another epoch, or too old). - Every catch-up ends with `event: ready`. - Keep-alive comments are sent every 15 s. - - `KVStorageHttp.watch()` subscribes (via the vendored `SSESession`, which reconnects with - backoff), emits `change` events with `origin: 'remote'` and `resync`, and resolves once caught up. + - `KVStorageHttp.subscribe(path = this.key)` / `unsubscribe(path)` manage the subscribed paths. + Each change reconnects the stream (vendored `SSESession`, backoff) with the new list, resuming + from `lastEventId`. The client emits `change` (`origin: 'remote'`) and `resync`, and `subscribe` + resolves once caught up. A 401/403 rejects `subscribe` instead of retrying. - Limitation: a server only sees writes made through its own storage instance, so several server processes over one database need a shared pub/sub (not built). - **Wrappers** (e.g. a future encryption layer) extend `BaseKVStorage` and delegate the primitives. Keys must stay plaintext (or be deterministically encrypted) for prefix scans to work; values can be encrypted freely. -### 4.1 `KVStorageComposite({ source, replicas, retry? })` +### 4.1 Authentication (`utils/request-auth.ts`) -Keeps full local copies of a source (e.g. `KVStorageHttp`) so reads are local and the app works -offline. Replicas are listed fastest first (e.g. `[inMemory, indexedDb]`). The last replica is the +The server requires `auth`: either `{ verifiers, maxSkewMs?, now? }`, or `'none'` (open, for local +development and tests). It can't be left open by omission. + +- **Signed requests.** The client's `RequestSigner` signs `${timestamp}:${path}:${body}` (path = + URL pathname, body = the exact request body; same scheme as Harvmaster/sync-server-example). + Headers: `X-Public-Key` (lowercase hex), `X-Timestamp` (ms), `X-Signature` (hex) and + `X-Signature-Algorithm`. Every request is signed, and so is each SSE (re)connect, with a fresh + timestamp. +- **Verification (401).** The server picks the `SignatureVerifier` registered for the algorithm + (unknown → 401) and checks the timestamp is within `maxSkewMs` (default 60 s). It then rejects a + signature already accepted within that window (replay protection), and verifies the signature + over the received path and raw body. The body is parsed only afterwards. +- **Algorithms are pluggable.** Built in: `Ed25519Signer` / `Ed25519Verifier` (WebCrypto, so no + dependencies; Node 22+ and current browsers; keys persist as PKCS #8). Others, e.g. secp256k1 via + libauth, implement `RequestSigner` (`algorithm`, `publicKey`, `sign`) and `SignatureVerifier` + (`algorithm`, `verify`). The server can accept several algorithms at once. +- **Authorization (403): the public key is the user id.** Every key in `readMany`, `writeMany`, + `removeMany` and `writeIfAbsent`, and every subscription path, must be `` or under + `/`. A scan must stay within `[, 0)`, i.e. the key and its + subtree (escaping guarantees no other key falls in that range, including lookalikes such as + `x`). +- **Scoped events.** Each subscriber only receives changes under its own (authorized) paths, live + and when resuming. Event ids are a global sequence, so a client can infer the server's overall + write volume, though nothing about the contents. +- Proxies must preserve the URL path, because the signature covers it. + +### 4.2 `KVStorageComposite({ source, prefix?, replicas, retry? })` + +Keeps local copies of a source (e.g. `KVStorageHttp`) so reads are local and the app works +offline. With `prefix` (e.g. the user's public key) only that key and its subtree are synced, cached +and subscribed to; keys outside it throw `RangeError`. Without it, the whole source is copied, which +is only appropriate when the client may read all of it. Replicas are listed fastest first (e.g. `[inMemory, indexedDb]`). The last replica is the persistent one: it seeds the others on start and stores the composite's internal keys under `~q/` (hidden from reads): `~q/meta` (`{ lastEventId }`, present once a sync has completed) and the outbox `~q/o/`. @@ -222,8 +254,9 @@ Node API: `get(key?)`, `getMany`, `set(doc)` / `set(key, doc)`, `insert(doc)` ## 8. Open questions -- Composite: replicate a subtree instead of the whole store; conflict policies beyond last writer - wins for offline writes. +- Composite: lazily syncing parts of a large subtree on first read; conflict policies beyond last + writer wins for offline writes. +- Authorization beyond "public key = user id" (sharing a subtree with other keys). - Multi-process servers need a shared change feed (pub/sub) for SSE. - Hard deletes from immutable event logs (privacy requests): compaction vs crypto-shredding. The diff --git a/src/index.ts b/src/index.ts index 4ef085e..966cb34 100644 --- a/src/index.ts +++ b/src/index.ts @@ -16,6 +16,8 @@ export * from './utils/document-update.js' export * from './utils/errors.js' export * from './utils/event-emitter.js' export * from './utils/extended-json.js' +export * from './utils/hex.js' +export * from './utils/request-auth.js' export * from './utils/json.js' export * from './utils/keys.js' export * from './utils/schema.js' diff --git a/src/kv-storage/kv-storage-composite.ts b/src/kv-storage/kv-storage-composite.ts index 3b7175e..c698f80 100644 --- a/src/kv-storage/kv-storage-composite.ts +++ b/src/kv-storage/kv-storage-composite.ts @@ -1,13 +1,20 @@ import { StorageUnavailableError } from '../utils/errors.js' import type { OffCallback } from '../utils/event-emitter.js' import { Json } from '../utils/json.js' -import type { AnySpec, Spec } from '../utils/schema.js' +import { Keys } from '../utils/keys.js' +import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js' import { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.js' import { BaseKVStorage, type KVChange, type KVChangeEvent, type ScanOptions } from './base-kv-storage.js' export interface KVStorageCompositeOptions { /** The source of truth (e.g. `KVStorageHttp`). */ source: BaseKVStorage + /** + * Only this key and its subtree are synced, cached and subscribed to (e.g. the user's public + * key). Keys outside it are rejected. Default `''`: the whole source (only for sources the + * client may read entirely). + */ + prefix?: string /** * Local full copies, fastest first (e.g. `[inMemory, indexedDb]`). The last one should be * persistent: it also stores the offline outbox and sync metadata, and seeds the others on start. @@ -19,8 +26,8 @@ export interface KVStorageCompositeOptions { /** Implemented by sources that can push remote changes (e.g. `KVStorageHttp`). */ interface WatchableSource { - watch(): Promise - unwatch(): Promise + subscribe(path: string): Promise + unsubscribe(path: string): Promise lastEventId: string | undefined } @@ -38,16 +45,22 @@ const OUTBOX_PREFIX = '~q/o/' const OUTBOX_END = '~q/o0' const SYNC_BATCH = 500 +// Types only: children of a KVStorageComposite are KVStorageComposite nodes too (child() creates them with +// Object.create, so they share this class's prototype). +export interface KVStorageComposite { + child>(key: K): KVStorageComposite> +} + /** * Keeps local replicas of a source in sync, so reads (including `keys()`, `entries()` and document * queries) are served locally, and the app keeps working offline. * * - **Start** (lazily, on first use): the persistent replica seeds the faster ones. If it holds a * previous sync, reads are served from it straight away (possibly stale until caught up). - * - **Staying in sync:** remote changes pushed by the source (`watch()`, e.g. SSE) are applied to + * - **Staying in sync:** remote changes pushed by the source (`subscribe(prefix)`, e.g. SSE) are applied to * every replica. On restart the source resumes from the saved `lastEventId`; a full sync (diffing * the source against the replicas) only runs the first time, or when the source says it can't - * resume (`resync`). Without `watch()`, a full sync runs on start. + * resume (`resync`). Without `subscribe()`, a full sync runs on start. * - **Writes** go to the source, then the replicas. If the source is unreachable * (`StorageUnavailableError`), they are applied locally and queued in a persisted outbox, then * replayed in order when it's reachable again (last writer wins on conflict). While any write is @@ -59,6 +72,8 @@ const SYNC_BATCH = 500 */ export class KVStorageComposite extends BaseKVStorage { private readonly source: BaseKVStorage + /** Synced key range: `[prefix, prefix + '0')` is the prefix's own key and its subtree. */ + private readonly scope: { prefix: string; gte?: string; lt?: string } private readonly replicas: BaseKVStorage[] /** Persistent replica: seeds the others and holds the outbox and metadata. */ private readonly persistent: BaseKVStorage @@ -102,6 +117,8 @@ export class KVStorageComposite extends BaseKVStorage< super() if (options.replicas.length === 0) throw new TypeError('KVStorageComposite needs at least one replica') this.source = options.source + const prefix = options.prefix ?? '' + this.scope = prefix === '' ? { prefix } : { prefix, gte: prefix, lt: prefix + Keys.afterSeparator } this.replicas = options.replicas this.persistent = options.replicas.at(-1)! this.backoff = { baseDelay: 1000, maxDelay: 30_000, ...options.retry } @@ -134,21 +151,27 @@ export class KVStorageComposite extends BaseKVStorage< this.state.closed = true clearTimeout(this.state.retryTimer) for (const off of this.offs.splice(0)) off() - if (this.watchable) await (this.source as unknown as WatchableSource).unwatch() + if (this.watchable) await (this.source as unknown as WatchableSource).unsubscribe(this.scope.prefix) } async readMany(keys: string[]): Promise { + this.checkScope(keys) await this.start() return (this.state.hasSnapshot ? this.replicas[0]! : this.source).readMany(keys) } async *scan(options: ScanOptions = {}): AsyncGenerator<[string, unknown]> { await this.start() - if (!this.state.hasSnapshot) return yield* this.source.scan(options) + // Narrow the range to the synced scope (and, against an authenticated server, to what it allows). + const gte = this.scope.gte !== undefined && (options.gte === undefined || options.gte < this.scope.gte) ? this.scope.gte : options.gte + const lt = this.scope.lt !== undefined && (options.lt === undefined || options.lt > this.scope.lt) ? this.scope.lt : options.lt + if (gte !== undefined && lt !== undefined && gte >= lt) return + const scoped = { ...options, gte, lt } + if (!this.state.hasSnapshot) return yield* this.source.scan(scoped) // Skip internal keys before applying the limit. let remaining = options.limit ?? Infinity if (remaining <= 0) return - for await (const entry of this.replicas[0]!.scan({ ...options, limit: undefined })) { + for await (const entry of this.replicas[0]!.scan({ ...scoped, limit: undefined })) { if (entry[0] >= INTERNAL_PREFIX && entry[0] < INTERNAL_END) continue yield entry if (--remaining <= 0) return @@ -156,16 +179,19 @@ export class KVStorageComposite extends BaseKVStorage< } protected async applyWrite(entries: [string, unknown][]): Promise { + this.checkScope(entries.map(([key]) => key)) await this.start() await this.writeThrough({ type: 'write', entries }) } protected async applyRemove(keys: string[]): Promise { + this.checkScope(keys) await this.start() await this.writeThrough({ type: 'remove', keys }) } protected async applyWriteIfAbsent(entries: [string, unknown][]): Promise { + this.checkScope(entries.map(([key]) => key)) await this.start() const written = await this.source.writeIfAbsent(entries) const changes = entries.filter((_, i) => written[i]).map(([key, value]) => ({ key, value })) @@ -176,8 +202,17 @@ export class KVStorageComposite extends BaseKVStorage< // ------------------------------------------------------------------ start + private inScope(key: string): boolean { + const { prefix } = this.scope + return prefix === '' || key === prefix || key.startsWith(prefix + Keys.separator) + } + + private checkScope(keys: string[]): void { + for (const key of keys) if (!this.inScope(key)) throw new RangeError(`"${key}" is outside this composite's prefix "${this.scope.prefix}"`) + } + private get watchable(): boolean { - return typeof (this.source as unknown as Partial).watch === 'function' + return typeof (this.source as unknown as Partial).subscribe === 'function' } private async open(): Promise { @@ -201,8 +236,9 @@ export class KVStorageComposite extends BaseKVStorage< const source = this.source as unknown as WatchableSource source.lastEventId = meta?.lastEventId // Without a resume point, catch up on the stream first so no change after the snapshot is missed. - const watching = source.watch() - if (!meta?.lastEventId) void watching.then(() => this.enqueueApply(() => this.fullSync())) + const watching = source.subscribe(this.scope.prefix) + if (!meta?.lastEventId) void watching.then(() => this.enqueueApply(() => this.fullSync())).catch((error) => console.error(error)) + else void watching.catch((error) => console.error(error)) } else { this.enqueueApply(() => this.fullSync()) } @@ -238,7 +274,9 @@ export class KVStorageComposite extends BaseKVStorage< } private async applyRemote(event: KVChangeEvent): Promise { - const candidates = event.changes.filter(({ key }) => !this.consumeEcho(key) && !this.pending.has(key) && !key.startsWith(INTERNAL_PREFIX)) + const candidates = event.changes.filter( + ({ key }) => this.inScope(key) && !this.consumeEcho(key) && !this.pending.has(key) && !key.startsWith(INTERNAL_PREFIX), + ) // Skip changes the replicas already reflect. const current = await this.replicas[0]!.readMany(candidates.map(({ key }) => key)) const changes = candidates.filter(({ value }, i) => !Json.equal(value, current[i])) @@ -276,8 +314,9 @@ export class KVStorageComposite extends BaseKVStorage< /** Changes that turn the persistent replica into the source (both scanned in key order). */ private async *diff(): AsyncGenerator { - const remote = this.source.scan({ values: true })[Symbol.asyncIterator]() - const local = this.persistent.scan({ values: true })[Symbol.asyncIterator]() + const range = { gte: this.scope.gte, lt: this.scope.lt, values: true } + const remote = this.source.scan(range)[Symbol.asyncIterator]() + const local = this.persistent.scan(range)[Symbol.asyncIterator]() const nextLocal = async () => { for (let next = await local.next(); ; next = await local.next()) if (next.done || !next.value[0].startsWith(INTERNAL_PREFIX)) return next } diff --git a/src/kv-storage/kv-storage-http-server.ts b/src/kv-storage/kv-storage-http-server.ts index 208262a..d785a2b 100644 --- a/src/kv-storage/kv-storage-http-server.ts +++ b/src/kv-storage/kv-storage-http-server.ts @@ -1,7 +1,16 @@ +import { AuthError } from '../utils/errors.js' import { ExtendedJson } from '../utils/extended-json.js' +import { Keys } from '../utils/keys.js' +import { RequestAuth, type RequestAuthOptions, type SignatureVerifier } from '../utils/request-auth.js' import type { BaseKVStorage, KVChange, ScanOptions } from './base-kv-storage.js' export interface KVStorageHttpServerOptions { + /** + * `{ verifiers }`: requests must be signed (see `RequestAuth`), and a caller may only touch keys + * under its own public key (`` and `/...`). + * `'none'`: no authentication, everything is open; for local development and tests only. + */ + auth: 'none' | ({ verifiers: SignatureVerifier[] } & RequestAuthOptions) /** Path prefix to strip, e.g. `/kv`. Default `''`. */ basePath?: string /** Recent change batches kept for clients resuming with `Last-Event-ID`. Default 10 000. */ @@ -15,32 +24,45 @@ interface HistoryEntry { changes: KVChange[] } +interface Subscriber { + /** Keys whose own value and subtree this subscriber receives (`''` = everything). */ + paths: string[] + send: (frame: string) => void +} + +type RequestBody = { keys: string[]; entries: [string, unknown][]; paths: string[] } & ScanOptions + /** * Serves any KV storage over HTTP for `KVStorageHttp`. Framework-agnostic: `handle(Request)` - * returns a `Response`. + * returns a `Response`. Every route is `POST` with an Extended JSON body: * - * - `POST /` with an Extended JSON body (`scan` streams NDJSON). - * - `GET /events`: Server-Sent Events of every change to the storage. Event ids are - * `-` (epoch = this server instance). A client sending `Last-Event-ID` gets the - * changes it missed if they are still in the history; otherwise it gets a `resync` event. - * A client without one gets a `ready` event carrying the current id. + * - `/readMany`, `/writeMany`, `/removeMany`, `/writeIfAbsent`, `/scan` (streams NDJSON). + * - `/events` with `{ paths }`: Server-Sent Events of changes to those keys and their subtrees. + * Event ids are `-` (epoch = this server instance). A client sending + * `Last-Event-ID` gets the matching changes it missed if they are still in the history, + * otherwise a `resync` event; either way a `ready` event ends the catch-up. + * + * With authentication, each request is checked first (401), then every key, scan range and + * subscription path must be the caller's public key or under it (403). * * Only changes made through the wrapped storage instance are seen; several server processes over * one database each see only their own writes. */ export class KVStorageHttpServer { + private readonly auth: RequestAuth | undefined private readonly basePath: string private readonly historySize: number private readonly keepAliveMs: number private readonly epoch = Math.random().toString(36).slice(2, 10) private seq = 0 private readonly history: HistoryEntry[] = [] - private readonly subscribers = new Set<(frame: string) => void>() + private readonly subscribers = new Set() constructor( private readonly storage: BaseKVStorage, - options: KVStorageHttpServerOptions = {}, + options: KVStorageHttpServerOptions, ) { + this.auth = options.auth === 'none' ? undefined : new RequestAuth(options.auth.verifiers, options.auth) this.basePath = options.basePath ?? '' this.historySize = options.historySize ?? 10_000 this.keepAliveMs = options.keepAliveMs ?? 15_000 @@ -49,58 +71,99 @@ export class KVStorageHttpServer { async handle(request: Request): Promise { const path = new URL(request.url).pathname - if (!path.startsWith(this.basePath)) return KVStorageHttpServer.error(404, 'Not found') - const route = `${request.method} ${path.slice(this.basePath.length)}` - if (route === 'GET /events') return this.events(request.headers.get('last-event-id')) - if (request.method !== 'POST') return KVStorageHttpServer.error(404, 'Not found') + if (request.method !== 'POST' || !path.startsWith(this.basePath)) return KVStorageHttpServer.error(404, 'Not found') try { - const body = ExtendedJson.parse(await request.text()) as { keys: string[]; entries: [string, unknown][] } & ScanOptions - switch (route) { - case 'POST /readMany': { + const text = await request.text() + const owner = this.auth ? await this.auth.verify(request.headers, path, text) : undefined + const body = ExtendedJson.parse(text) as RequestBody + switch (path.slice(this.basePath.length)) { + case '/readMany': { + this.authorize(owner, body.keys) const values = await this.storage.readMany(body.keys) const missing = values.flatMap((v, i) => (v === undefined ? [i] : [])) return KVStorageHttpServer.json({ values: values.map((v) => v ?? null), missing }) } - case 'POST /writeMany': + case '/writeMany': + this.authorize( + owner, + body.entries.map(([key]) => key), + ) await this.storage.writeMany(body.entries) return new Response(null, { status: 204 }) - case 'POST /removeMany': + case '/removeMany': + this.authorize(owner, body.keys) await this.storage.removeMany(body.keys) return new Response(null, { status: 204 }) - case 'POST /writeIfAbsent': + case '/writeIfAbsent': + this.authorize( + owner, + body.entries.map(([key]) => key), + ) return KVStorageHttpServer.json(await this.storage.writeIfAbsent(body.entries)) - case 'POST /scan': + case '/scan': + this.authorizeRange(owner, body) return this.scan(body) + case '/events': + if (!Array.isArray(body.paths) || body.paths.length === 0) return KVStorageHttpServer.error(400, 'Subscribe to at least one path') + this.authorize(owner, body.paths) + return this.events(body.paths, request.headers.get('last-event-id')) default: return KVStorageHttpServer.error(404, 'Not found') } } catch (error) { + if (error instanceof AuthError) return KVStorageHttpServer.error(error.status, error.message) return KVStorageHttpServer.error(500, error instanceof Error ? error.message : String(error)) } } + // ------------------------------------------------------------------ authorization + + /** Every key must be the owner's key or under it (no-op without authentication). */ + private authorize(owner: string | undefined, keys: string[]): void { + if (owner === undefined) return + for (const key of keys) { + if (typeof key !== 'string' || !KVStorageHttpServer.within(owner, key)) throw new AuthError(403, `No access to "${key}"`) + } + } + + /** A scan must stay inside `[owner, owner + '0')`: the owner's key and its subtree. */ + private authorizeRange(owner: string | undefined, options: ScanOptions): void { + if (owner === undefined) return + const { gte, lt } = options + if (gte === undefined || lt === undefined || gte < owner || lt > owner + Keys.afterSeparator) throw new AuthError(403, 'Scan outside your keys') + } + + /** Whether `key` is `path` or below it (`''` covers everything). */ + private static within(path: string, key: string): boolean { + return path === '' || key === path || key.startsWith(path + Keys.separator) + } + + // ------------------------------------------------------------------ change stream + private publish(changes: KVChange[]): void { const entry = { seq: ++this.seq, changes } this.history.push(entry) if (this.history.length > this.historySize) this.history.splice(0, this.history.length - this.historySize) - const frame = this.frame('change', entry.seq, changes) - for (const send of this.subscribers) send(frame) + for (const subscriber of this.subscribers) { + const frame = this.changeFrame(entry, subscriber.paths) + if (frame) subscriber.send(frame) + } } - private events(lastEventId: string | null): Response { + private events(paths: string[], lastEventId: string | null): Response { const encoder = new TextEncoder() - let send: ((frame: string) => void) | undefined + let subscriber: Subscriber | undefined let keepAlive: ReturnType | undefined const stream = new ReadableStream({ start: (controller) => { - send = (frame) => controller.enqueue(encoder.encode(frame)) - send(this.catchUpFrames(lastEventId)) - this.subscribers.add(send) - keepAlive = setInterval(() => send!(': keep-alive\n\n'), this.keepAliveMs) + subscriber = { paths, send: (frame) => controller.enqueue(encoder.encode(frame)) } + subscriber.send(this.catchUpFrames(paths, lastEventId)) + this.subscribers.add(subscriber) + keepAlive = setInterval(() => subscriber!.send(': keep-alive\n\n'), this.keepAliveMs) ;(keepAlive as { unref?: () => void }).unref?.() }, cancel: () => { - if (send) this.subscribers.delete(send) + if (subscriber) this.subscribers.delete(subscriber) clearInterval(keepAlive) }, }) @@ -108,10 +171,10 @@ export class KVStorageHttpServer { } /** - * What a (re)connecting client gets before live changes: the changes it missed, or `resync` when - * they can't be replayed; then always `ready` (the client is caught up as of that id). + * What a (re)connecting client gets before live changes: the matching changes it missed, or + * `resync` when they can't be replayed; then always `ready` (caught up as of that id). */ - private catchUpFrames(lastEventId: string | null): string { + private catchUpFrames(paths: string[], lastEventId: string | null): string { const ready = this.frame('ready', this.seq, null) if (lastEventId === null) return ready const [epoch, seqText] = lastEventId.split('-') @@ -119,14 +182,26 @@ export class KVStorageHttpServer { const oldest = this.history[0]?.seq ?? this.seq + 1 // Resumable only if it's our epoch and nothing after `seq` has been dropped from the history. if (epoch !== this.epoch || !Number.isInteger(seq) || seq > this.seq || seq < oldest - 1) return this.frame('resync', this.seq, null) + ready - const missed = this.history.filter((entry) => entry.seq > seq).map((entry) => this.frame('change', entry.seq, entry.changes)) - return missed.join('') + ready + return ( + this.history + .filter((entry) => entry.seq > seq) + .map((entry) => this.changeFrame(entry, paths)) + .join('') + ready + ) + } + + /** A `change` frame with only the changes under `paths`, or `''` if none match. */ + private changeFrame(entry: HistoryEntry, paths: string[]): string { + const changes = entry.changes.filter(({ key }) => paths.some((path) => KVStorageHttpServer.within(path, key))) + return changes.length ? this.frame('change', entry.seq, changes) : '' } private frame(event: string, seq: number, data: unknown): string { return `id: ${this.epoch}-${seq}\nevent: ${event}\ndata: ${ExtendedJson.stringify(data)}\n\n` } + // ------------------------------------------------------------------ scan + private scan(options: ScanOptions): Response { const iterator = this.storage.scan(options)[Symbol.asyncIterator]() const encoder = new TextEncoder() diff --git a/src/kv-storage/kv-storage-http.ts b/src/kv-storage/kv-storage-http.ts index 4e43026..9cfe6cc 100644 --- a/src/kv-storage/kv-storage-http.ts +++ b/src/kv-storage/kv-storage-http.ts @@ -1,39 +1,58 @@ -import { StorageUnavailableError } from '../utils/errors.js' +import { AuthError, StorageUnavailableError } from '../utils/errors.js' import { ExtendedJson } from '../utils/extended-json.js' -import type { AnySpec, Spec } from '../utils/schema.js' +import { RequestAuth, type RequestSigner } from '../utils/request-auth.js' +import type { AnySpec, ChildOf, KeyOf, Spec } from '../utils/schema.js' +import { ExponentialBackoff } from '../utils/sse-session/exponential-backoff.js' import { SSESession } from '../utils/sse-session/sse-session.js' import { BaseKVStorage, type KVChange, type ScanOptions } from './base-kv-storage.js' export interface KVStorageHttpOptions { /** Base URL of a `KVStorageHttpServer`, e.g. `https://host/kv`. */ url: string + /** Signs every request (required by servers with authentication). */ + signer?: RequestSigner headers?: Record fetch?: (url: string, init: RequestInit) => Promise /** Called when the change stream fails (it keeps reconnecting). Default: ignore. */ - onWatchError?: (error: Error) => void + onSubscriptionError?: (error: Error) => void +} + +// Types only: children of a KVStorageHttp are KVStorageHttp nodes too (child() creates them with +// Object.create, so they share this class's prototype). +export interface KVStorageHttp { + child>(key: K): KVStorageHttp> } /** * KV storage client for a `KVStorageHttpServer`. Same API as any other KV storage. * - * `watch()` subscribes to the server's change stream (SSE): changes made by anyone are emitted - * here as `change` events with `origin: 'remote'`, and `resync` is emitted when the server can't - * resume from `lastEventId`. Values travel as Extended JSON (`bigint` and `Uint8Array` survive). + * `subscribe(path)` receives changes under a key (pushed over SSE) as `change` events with + * `origin: 'remote'`; `resync` is emitted when the server can't resume from `lastEventId`. With a + * `signer`, every request is signed, and an authenticated server only allows keys under the + * signer's public key. Values travel as Extended JSON (`bigint` and `Uint8Array` survive). */ export class KVStorageHttp extends BaseKVStorage { private readonly url: string + private readonly signer: RequestSigner | undefined private readonly headers: Record private readonly fetcher: (url: string, init: RequestInit) => Promise - private readonly onWatchError: (error: Error) => void + private readonly onSubscriptionError: (error: Error) => void /** Mutable state lives here: child nodes share this object (see `child()`). */ - private readonly state: { session?: SSESession; lastEventId?: string } = {} + private readonly state: { + paths: Set + session?: SSESession + lastEventId?: string + /** Subscription changes are applied one at a time. */ + updating: Promise + } = { paths: new Set(), updating: Promise.resolve() } constructor(options: KVStorageHttpOptions) { super() this.url = options.url.replace(/\/$/, '') + this.signer = options.signer this.headers = options.headers ?? {} this.fetcher = options.fetch ?? ((url, init) => globalThis.fetch(url, init)) - this.onWatchError = options.onWatchError ?? (() => {}) + this.onSubscriptionError = options.onSubscriptionError ?? (() => {}) } /** Position in the server's change stream; sent as `Last-Event-ID` when (re)connecting. */ @@ -45,6 +64,11 @@ export class KVStorageHttp extends BaseKVStorage { this.state.lastEventId = id } + /** Keys currently subscribed to. */ + get subscriptions(): ReadonlySet { + return this.state.paths + } + async readMany(keys: string[]): Promise { if (keys.length === 0) return [] const { values, missing } = (await this.post('readMany', { keys })) as { values: unknown[]; missing: number[] } @@ -75,39 +99,22 @@ export class KVStorageHttp extends BaseKVStorage { } /** - * Subscribes to the server's change stream. Resolves once connected and caught up (missed - * changes replayed, or `resync` emitted); keeps reconnecting and resuming from `lastEventId` - * until `unwatch()`. While the server is unreachable this stays pending, so don't await it on a - * path that must work offline. + * Receives changes to `path` (default: this node's key) and everything under it. Resolves once + * the stream is connected and caught up (missed changes replayed, or `resync` emitted). While + * the server is unreachable this stays pending, so don't await it on a path that must work offline. */ - async watch(): Promise { - if (this.state.session) return - const session = new SSESession(`${this.url}/events`, { - fetch: this.fetcher, - headers: this.headers, - persistent: true, - onError: this.onWatchError, - onRequest: async (request) => - this.lastEventId ? { ...request, headers: { ...(request.headers as Record), 'Last-Event-ID': this.lastEventId } } : request, - }) - session.on('message', (message) => { - if (message.id) this.lastEventId = message.id - if (message.event === 'change') { - this.emit('change', { changes: ExtendedJson.parse(message.data) as KVChange[], origin: 'remote', id: message.id }) - } else if (message.event === 'resync') { - this.emit('resync', { id: message.id }) - } - }) - this.state.session = session - const caughtUp = session.waitFor('message', (message) => message.event === 'ready') - await session.connect() - await caughtUp + subscribe(path: string = this.key): Promise { + return this.updateSubscriptions(() => this.state.paths.add(path)) } - async unwatch(): Promise { - const session = this.state.session - this.state.session = undefined - await session?.disconnect() + /** Stops receiving changes to `path` (default: this node's key). */ + unsubscribe(path: string = this.key): Promise { + return this.updateSubscriptions(() => this.state.paths.delete(path)) + } + + /** Drops every subscription and closes the stream. */ + close(): Promise { + return this.updateSubscriptions(() => this.state.paths.clear()) } protected async applyWrite(entries: [string, unknown][]): Promise { @@ -122,24 +129,102 @@ export class KVStorageHttp extends BaseKVStorage { return (await this.post('writeIfAbsent', { entries })) as boolean[] } + // ------------------------------------------------------------------ subscriptions + + /** Applies a subscription change, then reconnects the stream with the new path list. */ + private updateSubscriptions(change: () => unknown): Promise { + const run = async () => { + const before = JSON.stringify([...this.state.paths]) + change() + if (JSON.stringify([...this.state.paths]) === before && (this.state.session || this.state.paths.size === 0)) return + await this.state.session?.disconnect() + this.state.session = undefined + if (this.state.paths.size > 0) await this.connect([...this.state.paths]) + } + const result = this.state.updating.then(run) + this.state.updating = result.catch(() => {}) + return result + } + + private async connect(paths: string[]): Promise { + const url = `${this.url}/events` + const body = ExtendedJson.stringify({ paths }) + const session = new SSESession(url, { + method: 'POST', + body, + fetch: this.fetcher, + headers: { ...this.headers, 'content-type': 'application/json' }, + persistent: true, + onError: this.onSubscriptionError, + retry: KVStorageHttp.retryUnlessDenied(), + // Signed on every (re)connect, so the timestamp is always fresh. + onRequest: async (request) => ({ + ...request, + headers: { + ...(request.headers as Record), + ...(this.state.lastEventId ? { 'Last-Event-ID': this.state.lastEventId } : {}), + ...(await this.authHeaders(url, body)), + }, + }), + }) + session.on('message', (message) => { + if (message.id) this.state.lastEventId = message.id + if (message.event === 'change') { + this.emit('change', { changes: ExtendedJson.parse(message.data) as KVChange[], origin: 'remote', id: message.id }) + } else if (message.event === 'resync') { + this.emit('resync', { id: message.id }) + } + }) + this.state.session = session + const caughtUp = session.waitFor('message', (message) => message.event === 'ready') + try { + await session.connect() + } catch (error) { + this.state.session = undefined + throw (error as Error).cause instanceof AuthError ? (error as Error).cause : error + } + await caughtUp + } + + // ------------------------------------------------------------------ requests + private async post(path: string, body: unknown): Promise { const response = await this.request(path, body) return response.status === 204 ? undefined : ExtendedJson.parse(await response.text()) } - private async request(path: string, body: unknown): Promise { + private async request(path: string, payload: unknown): Promise { + const url = `${this.url}/${path}` + const body = ExtendedJson.stringify(payload) let response: Response try { - response = await this.fetcher(`${this.url}/${path}`, { + response = await this.fetcher(url, { method: 'POST', - headers: { ...this.headers, 'content-type': 'application/json' }, - body: ExtendedJson.stringify(body), + headers: { ...this.headers, 'content-type': 'application/json', ...(await this.authHeaders(url, body)) }, + body, }) } catch (error) { throw new StorageUnavailableError(`Cannot reach ${this.url}`, { cause: error }) } if ([502, 503, 504].includes(response.status)) throw new StorageUnavailableError(`HTTP ${response.status} from ${this.url}`) + if (response.status === 401 || response.status === 403) throw new AuthError(response.status, await response.text()) if (!response.ok) throw new Error(`HTTP ${response.status}: ${await response.text()}`) return response } + + /** Retries connecting with backoff, except when the server denies access (401/403): that won't fix itself. */ + private static retryUnlessDenied(): { run(fn: () => Promise): Promise } { + const backoff = new ExponentialBackoff({ baseDelay: 1000, maxDelay: 10_000, maxAttempts: 0, growthRate: 1.3, jitter: 0.3 }) + return { + run: (fn) => + backoff.run(fn, (error, context) => { + const status = /^HTTP error: (40[13])/.exec(error.message)?.[1] + if (status) context.abort(new AuthError(Number(status) as 401 | 403, error.message)) + }), + } + } + + private async authHeaders(url: string, body: string): Promise> { + return this.signer ? RequestAuth.headers(this.signer, new URL(url).pathname, body) : {} + } } diff --git a/src/utils/errors.ts b/src/utils/errors.ts index c92957d..5c15807 100644 --- a/src/utils/errors.ts +++ b/src/utils/errors.ts @@ -25,3 +25,14 @@ export class StorageUnavailableError extends Error { this.name = 'StorageUnavailableError' } } + +/** A request was not authenticated (401) or not allowed for this caller (403). */ +export class AuthError extends Error { + constructor( + readonly status: 401 | 403, + message: string, + ) { + super(message) + this.name = 'AuthError' + } +} diff --git a/src/utils/extended-json.ts b/src/utils/extended-json.ts index 13888e3..f7ada03 100644 --- a/src/utils/extended-json.ts +++ b/src/utils/extended-json.ts @@ -1,6 +1,8 @@ // Adapted from xo-cash-utils `source/extended-json.ts` (branch installable-sse), with local hex // encoding instead of @bitauth/libauth. +import { Hex } from './hex.js' + /** Matches a bigint encoded as ``. */ const BIGINT_PATTERN = /^[+-]?[0-9]+)n>$/u @@ -22,7 +24,7 @@ export class ExtendedJson { /** `JSON.stringify` replacer. */ static replacer(this: void, _key: string, value: unknown): unknown { - if (value instanceof Uint8Array) return `` + if (value instanceof Uint8Array) return `` if (typeof value === 'bigint') return `` return value } @@ -33,19 +35,7 @@ export class ExtendedJson { const bigint = value.match(BIGINT_PATTERN) if (bigint) return BigInt(bigint.groups!.bigint!) const bytes = value.match(UINT8ARRAY_PATTERN) - if (bytes) return ExtendedJson.fromHex(bytes.groups!.hex!) + if (bytes) return Hex.decode(bytes.groups!.hex!) return value } - - private static toHex(bytes: Uint8Array): string { - let hex = '' - for (const byte of bytes) hex += byte.toString(16).padStart(2, '0') - return hex - } - - private static fromHex(hex: string): Uint8Array { - const bytes = new Uint8Array(hex.length / 2) - for (let i = 0; i < bytes.length; i++) bytes[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16) - return bytes - } } diff --git a/src/utils/hex.ts b/src/utils/hex.ts new file mode 100644 index 0000000..5d7737a --- /dev/null +++ b/src/utils/hex.ts @@ -0,0 +1,16 @@ +/** Lowercase hex encoding of bytes. */ +export class Hex { + static encode(bytes: Uint8Array): string { + let hex = '' + for (const byte of bytes) hex += byte.toString(16).padStart(2, '0') + return hex + } + + /** Throws on odd length or non-hex characters. */ + static decode(hex: string): Uint8Array { + if (hex.length % 2 !== 0 || !/^[0-9a-fA-F]*$/.test(hex)) throw new TypeError('Invalid hex string') + const bytes = new Uint8Array(hex.length / 2) + for (let i = 0; i < bytes.length; i++) bytes[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16) + return bytes + } +} diff --git a/src/utils/request-auth.ts b/src/utils/request-auth.ts new file mode 100644 index 0000000..87737be --- /dev/null +++ b/src/utils/request-auth.ts @@ -0,0 +1,150 @@ +import { AuthError } from './errors.js' +import { Hex } from './hex.js' + +/** Signs requests. Implement this to use another algorithm (e.g. secp256k1 via libauth). */ +export interface RequestSigner { + /** Algorithm name sent in `X-Signature-Algorithm`; the server must have a matching verifier. */ + readonly algorithm: string + /** Public key as lowercase hex. Also the owner segment of the caller's keys (`//...`). */ + readonly publicKey: string + sign(payload: Uint8Array): Promise +} + +/** Verifies signatures for one algorithm. */ +export interface SignatureVerifier { + readonly algorithm: string + verify(publicKey: string, payload: Uint8Array, signature: Uint8Array): Promise +} + +export const AUTH_HEADERS = { + publicKey: 'X-Public-Key', + timestamp: 'X-Timestamp', + signature: 'X-Signature', + algorithm: 'X-Signature-Algorithm', +} as const + +export interface RequestAuthOptions { + /** Accepted difference between the request timestamp and the server clock. Default 60 000 ms. */ + maxSkewMs?: number + now?: () => number +} + +/** + * Signed-request authentication. The client signs `${timestamp}:${path}:${body}` (path = URL + * pathname, body = the exact request body) and sends the public key, timestamp, signature (hex) + * and algorithm as headers. The server checks the timestamp is recent, verifies the signature + * with the verifier for that algorithm, and rejects a signature it has already seen within the + * timestamp window (replay protection). + */ +export class RequestAuth { + private readonly verifiers: Map + private readonly maxSkewMs: number + private readonly now: () => number + /** Signatures accepted within the window → when they expire; insertion order ≈ expiry order. */ + private readonly seen = new Map() + + constructor(verifiers: SignatureVerifier[], options: RequestAuthOptions = {}) { + if (verifiers.length === 0) throw new TypeError('RequestAuth needs at least one verifier') + this.verifiers = new Map(verifiers.map((v) => [v.algorithm, v])) + this.maxSkewMs = options.maxSkewMs ?? 60_000 + this.now = options.now ?? Date.now + } + + static payload(timestamp: string, path: string, body: string): Uint8Array { + return new TextEncoder().encode(`${timestamp}:${path}:${body}`) + } + + /** Auth headers for a request to `path` with `body`. */ + static async headers(signer: RequestSigner, path: string, body: string, now: number = Date.now()): Promise> { + const timestamp = String(now) + const signature = await signer.sign(RequestAuth.payload(timestamp, path, body)) + return { + [AUTH_HEADERS.publicKey]: signer.publicKey, + [AUTH_HEADERS.timestamp]: timestamp, + [AUTH_HEADERS.signature]: Hex.encode(signature), + [AUTH_HEADERS.algorithm]: signer.algorithm, + } + } + + /** Returns the caller's public key, or throws `AuthError` (401). */ + async verify(headers: Headers, path: string, body: string): Promise { + const publicKey = headers.get(AUTH_HEADERS.publicKey) + const timestamp = headers.get(AUTH_HEADERS.timestamp) + const signature = headers.get(AUTH_HEADERS.signature) + const algorithm = headers.get(AUTH_HEADERS.algorithm) + if (!publicKey || !timestamp || !signature || !algorithm) throw new AuthError(401, 'Missing authentication headers') + const verifier = this.verifiers.get(algorithm) + if (!verifier) throw new AuthError(401, `Unsupported signature algorithm "${algorithm}"`) + if (!/^[0-9a-f]+$/.test(publicKey)) throw new AuthError(401, 'Public key must be lowercase hex') + const now = this.now() + const time = Number(timestamp) + if (!Number.isInteger(time) || Math.abs(now - time) > this.maxSkewMs) throw new AuthError(401, 'Timestamp outside the accepted window') + this.forgetExpired(now) + if (this.seen.has(signature)) throw new AuthError(401, 'Replayed request') + let valid = false + try { + valid = await verifier.verify(publicKey, RequestAuth.payload(timestamp, path, body), Hex.decode(signature)) + } catch { + valid = false + } + if (!valid) throw new AuthError(401, 'Invalid signature') + this.seen.set(signature, time + this.maxSkewMs) + return publicKey + } + + private forgetExpired(now: number): void { + for (const [signature, expires] of this.seen) { + if (expires >= now) break + this.seen.delete(signature) + } + } +} + +/** Ed25519 signer using WebCrypto (Node 22+ and current browsers; no dependencies). */ +export class Ed25519Signer implements RequestSigner { + readonly algorithm = 'ed25519' + + private constructor( + private readonly privateKey: CryptoKey, + readonly publicKey: string, + ) {} + + /** A new random key pair. */ + static async generate(): Promise { + const pair = (await crypto.subtle.generateKey({ name: 'Ed25519' }, true, ['sign', 'verify'])) as CryptoKeyPair + return new Ed25519Signer(pair.privateKey, Hex.encode(new Uint8Array(await crypto.subtle.exportKey('raw', pair.publicKey)))) + } + + /** Restores a key exported with `exportPrivateKey()` (PKCS #8). */ + static async fromPrivateKey(pkcs8: Uint8Array): Promise { + const privateKey = await crypto.subtle.importKey('pkcs8', pkcs8 as BufferSource, { name: 'Ed25519' }, true, ['sign']) + // WebCrypto can't derive the public key directly; the JWK export of a private key includes it (`x`). + const jwk = await crypto.subtle.exportKey('jwk', privateKey) + const publicKey = Uint8Array.from(atob(jwk.x!.replace(/-/g, '+').replace(/_/g, '/')), (c) => c.charCodeAt(0)) + return new Ed25519Signer(privateKey, Hex.encode(publicKey)) + } + + async exportPrivateKey(): Promise { + return new Uint8Array(await crypto.subtle.exportKey('pkcs8', this.privateKey)) + } + + async sign(payload: Uint8Array): Promise { + return new Uint8Array(await crypto.subtle.sign({ name: 'Ed25519' }, this.privateKey, payload as BufferSource)) + } +} + +/** Ed25519 verifier using WebCrypto. */ +export class Ed25519Verifier implements SignatureVerifier { + readonly algorithm = 'ed25519' + private readonly keys = new Map>() + + async verify(publicKey: string, payload: Uint8Array, signature: Uint8Array): Promise { + let key = this.keys.get(publicKey) + if (!key) { + key = crypto.subtle.importKey('raw', Hex.decode(publicKey) as BufferSource, { name: 'Ed25519' }, false, ['verify']) + this.keys.set(publicKey, key) + if (this.keys.size > 10_000) this.keys.delete(this.keys.keys().next().value!) + } + return crypto.subtle.verify({ name: 'Ed25519' }, await key, signature as BufferSource, payload as BufferSource) + } +} diff --git a/test/kv-storage/kv-storage-http-auth.test.ts b/test/kv-storage/kv-storage-http-auth.test.ts new file mode 100644 index 0000000..79575a8 --- /dev/null +++ b/test/kv-storage/kv-storage-http-auth.test.ts @@ -0,0 +1,95 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { KVChangeEvent } from '../../src/kv-storage/base-kv-storage.js' +import { KVStorageComposite } from '../../src/kv-storage/kv-storage-composite.js' +import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' +import { AuthError } from '../../src/utils/errors.js' +import { Ed25519Signer, Ed25519Verifier } from '../../src/utils/request-auth.js' +import { collect } from '../helpers.js' +import { TestNetwork } from './sync-helpers.js' + +let network: TestNetwork +let alice: Ed25519Signer +let bob: Ed25519Signer + +beforeEach(async () => { + network = new TestNetwork({ auth: { verifiers: [new Ed25519Verifier()] } }) + alice = await Ed25519Signer.generate() + bob = await Ed25519Signer.generate() + await network.store.writeMany([ + [`${alice.publicKey}/profile`, { name: 'Alice' }], + [`${bob.publicKey}/profile`, { name: 'Bob' }], + [`${bob.publicKey}/posts/1`, { title: 'secret' }], + ]) +}) +afterEach(() => network.close()) + +describe('authenticated KVStorageHttpServer', () => { + it('lets a user read, write and list their own keys', async () => { + const mine = network.client(alice).child(alice.publicKey) + expect(await mine.get('profile')).toEqual({ name: 'Alice' }) + await mine.child('posts').set('1', { title: 'hello' }) + expect(await collect(mine.keys())).toEqual(['posts', 'profile']) + expect(await collect(mine.child('posts').entries())).toEqual([['1', { title: 'hello' }]]) + }) + + it("forbids reading, writing or scanning another user's keys", async () => { + const client = network.client(alice) + const theirs = client.child(bob.publicKey) + await expect(theirs.get('profile')).rejects.toMatchObject({ status: 403 }) + await expect(theirs.set('profile', { name: 'hacked' })).rejects.toBeInstanceOf(AuthError) + await expect(theirs.delete('profile')).rejects.toMatchObject({ status: 403 }) + await expect(collect(theirs.keys())).rejects.toMatchObject({ status: 403 }) + await expect(collect(client.keys())).rejects.toMatchObject({ status: 403 }) // root listing + await expect(client.readMany([alice.publicKey + 'x/a'])).rejects.toMatchObject({ status: 403 }) // lookalike prefix + expect(await network.store.readMany([`${bob.publicKey}/profile`])).toEqual([{ name: 'Bob' }]) + }) + + it('rejects unsigned requests', async () => { + await expect(network.client().child(alice.publicKey).get('profile')).rejects.toMatchObject({ status: 401 }) + }) + + it('delivers only subscribed, authorized changes', async () => { + const watcher = network.client(alice) + const received: KVChangeEvent[] = [] + watcher.on('change', (e) => e.origin === 'remote' && received.push(e)) + await watcher.child(alice.publicKey).child('posts').subscribe() + const writer = network.client(alice).child(alice.publicKey) + await network.client(bob).child(bob.publicKey).child('posts').set('2', { title: 'bob only' }) + await writer.set('profile', { name: 'Alice 2' }) // not subscribed + const arrived = watcher.waitFor('change', () => true) + await writer.child('posts').set('1', { title: 'mine' }) + await arrived + expect(received.flatMap((e) => e.changes)).toEqual([{ key: `${alice.publicKey}/posts/1`, value: { title: 'mine' } }]) + }) + + it("refuses subscriptions to another user's keys without retrying forever", async () => { + await expect(network.client(alice).child(bob.publicKey).subscribe()).rejects.toMatchObject({ status: 403 }) + await expect(network.client(alice).subscribe('')).rejects.toMatchObject({ status: 403 }) + }) + + it('replays only matching changes when resuming', async () => { + const watcher = network.client(alice) + const mine = watcher.child(alice.publicKey) + await mine.subscribe() + await mine.unsubscribe() + await network.client(bob).child(bob.publicKey).set('x', 1) + await network.client(alice).child(alice.publicKey).set('y', 2) + const replayed: KVChangeEvent[] = [] + watcher.on('change', (e) => replayed.push(e)) + await mine.subscribe() + expect(replayed.flatMap((e) => e.changes)).toEqual([{ key: `${alice.publicKey}/y`, value: 2 }]) + }) + + it('a composite scoped to the user syncs only their data', async () => { + const db = new KVStorageComposite({ source: network.client(alice), prefix: alice.publicKey, replicas: [new KVStorageInMemory()] }) + await db.synced() + expect(await collect(db.scan({ values: true }))).toEqual([[`${alice.publicKey}/profile`, { name: 'Alice' }]]) + expect(await collect(db.keys())).toEqual([alice.publicKey]) + await expect(db.child(bob.publicKey).get('profile')).rejects.toBeInstanceOf(RangeError) + const pushed = db.waitFor('change', (e) => e.origin === 'remote') + await network.client(bob).child(bob.publicKey).set('ignored', 1) + await network.client(alice).child(alice.publicKey).set('fromElsewhere', 1) + expect((await pushed).changes).toEqual([{ key: `${alice.publicKey}/fromElsewhere`, value: 1 }]) + await db.close() + }) +}) diff --git a/test/kv-storage/kv-storage-http-watch.test.ts b/test/kv-storage/kv-storage-http-watch.test.ts index 889a9e9..31dc777 100644 --- a/test/kv-storage/kv-storage-http-watch.test.ts +++ b/test/kv-storage/kv-storage-http-watch.test.ts @@ -6,11 +6,11 @@ import { TestNetwork } from './sync-helpers.js' let network: TestNetwork afterEach(() => network?.close()) -describe('KVStorageHttp.watch (SSE)', () => { +describe('KVStorageHttp.subscribe (SSE)', () => { it('receives changes made by other clients as remote change events', async () => { network = new TestNetwork() const watcher = network.client() - await watcher.watch() + await watcher.subscribe('') const received = watcher.waitFor('change', (e) => e.origin === 'remote') await network.client().child('users').set('1', { name: 'Ada', n: 1n }) const event = await received @@ -23,16 +23,16 @@ describe('KVStorageHttp.watch (SSE)', () => { network = new TestNetwork() const writer = network.client() const watcher = network.client() - await watcher.watch() + await watcher.subscribe('') const first = watcher.waitFor('change', (e) => e.origin === 'remote') await writer.writeMany([['a', 1]]) await first - await watcher.unwatch() + await watcher.unsubscribe('') await writer.writeMany([['b', 2]]) await writer.writeMany([['c', 3]]) const replayed: KVChangeEvent[] = [] watcher.on('change', (e) => replayed.push(e)) - await watcher.watch() // resolves once caught up, so the missed changes have been emitted + await watcher.subscribe('') // resolves once caught up, so the missed changes have been emitted expect(replayed.map((e) => e.changes)).toEqual([[{ key: 'b', value: 2 }], [{ key: 'c', value: 3 }]]) }) @@ -40,15 +40,15 @@ describe('KVStorageHttp.watch (SSE)', () => { network = new TestNetwork({ historySize: 1 }) const writer = network.client() const watcher = network.client() - await watcher.watch() + await watcher.subscribe('') const seen = watcher.waitFor('change', () => true) await writer.writeMany([['a', 1]]) await seen - await watcher.unwatch() + await watcher.unsubscribe('') await writer.writeMany([['b', 2]]) await writer.writeMany([['c', 3]]) const resync = watcher.waitFor('resync', () => true) - await watcher.watch() + await watcher.subscribe('') expect(await resync).toEqual({ id: expect.stringMatching(/-3$/) }) }) diff --git a/test/kv-storage/kv-storage-http.test.ts b/test/kv-storage/kv-storage-http.test.ts index 2dc295e..50d6410 100644 --- a/test/kv-storage/kv-storage-http.test.ts +++ b/test/kv-storage/kv-storage-http.test.ts @@ -9,7 +9,7 @@ import { runKVStorageConformance } from './kv-storage.conformance.js' import { serve } from './node-server.js' runKVStorageConformance('KVStorageHttp (in-process)', () => { - const server = new KVStorageHttpServer(new KVStorageInMemory(), { basePath: '/kv' }) + const server = new KVStorageHttpServer(new KVStorageInMemory(), { auth: 'none', basePath: '/kv' }) return new KVStorageHttp({ url: 'http://test/kv', fetch: async (input, init) => server.handle(new Request(input, init)) }) }) @@ -22,7 +22,7 @@ describe('over a real socket', () => { }) afterAll(() => server.close()) const fresh = () => { - stores.push(new KVStorageHttpServer(new KVStorageInMemory(), { basePath: `/${stores.length}` })) + stores.push(new KVStorageHttpServer(new KVStorageInMemory(), { auth: 'none', basePath: `/${stores.length}` })) return new KVStorageHttp({ url: `${server.url}/${stores.length - 1}` }) } @@ -47,7 +47,7 @@ describe('over a real socket', () => { describe('server', () => { it('maps errors to 404 / 500', async () => { - const server = new KVStorageHttpServer(new KVStorageInMemory()) + const server = new KVStorageHttpServer(new KVStorageInMemory(), { auth: 'none' }) expect((await server.handle(new Request('http://x/nope', { method: 'POST', body: '{}' }))).status).toBe(404) expect((await server.handle(new Request('http://x/readMany', { method: 'POST', body: '{"keys": 5}' }))).status).toBe(500) }) diff --git a/test/kv-storage/sync-helpers.ts b/test/kv-storage/sync-helpers.ts index e37e1a3..8b3eece 100644 --- a/test/kv-storage/sync-helpers.ts +++ b/test/kv-storage/sync-helpers.ts @@ -1,6 +1,7 @@ import { KVStorageHttp } from '../../src/kv-storage/kv-storage-http.js' import { KVStorageHttpServer, type KVStorageHttpServerOptions } from '../../src/kv-storage/kv-storage-http-server.js' import { KVStorageInMemory } from '../../src/kv-storage/kv-storage-in-memory.js' +import type { RequestSigner } from '../../src/utils/request-auth.js' import { KVStorageCounting } from '../helpers.js' /** @@ -13,13 +14,14 @@ export class TestNetwork { readonly server: KVStorageHttpServer private readonly clients: KVStorageHttp[] = [] - constructor(options: KVStorageHttpServerOptions = {}) { - this.server = new KVStorageHttpServer(this.store, options) + constructor(options: Partial = {}) { + this.server = new KVStorageHttpServer(this.store, { auth: 'none', ...options }) } - client(): KVStorageHttp { + client(signer?: RequestSigner): KVStorageHttp { const client = new KVStorageHttp({ url: 'http://server', + signer, fetch: async (url, init) => { if (!this.online) throw new TypeError('fetch failed') return this.server.handle(new Request(url, init)) @@ -30,6 +32,6 @@ export class TestNetwork { } async close(): Promise { - await Promise.all(this.clients.map((c) => c.unwatch())) + await Promise.all(this.clients.map((c) => c.close())) } } diff --git a/test/utils/request-auth.test.ts b/test/utils/request-auth.test.ts new file mode 100644 index 0000000..532f10d --- /dev/null +++ b/test/utils/request-auth.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, it } from 'vitest' +import { AUTH_HEADERS, Ed25519Signer, Ed25519Verifier, RequestAuth, type RequestSigner, type SignatureVerifier } from '../../src/utils/request-auth.js' +import { AuthError } from '../../src/utils/errors.js' + +const NOW = 1_700_000_000_000 + +async function signed(signer: RequestSigner, path = '/kv/readMany', body = '{"keys":["a"]}', at = NOW) { + return new Headers(await RequestAuth.headers(signer, path, body, at)) +} + +describe('RequestAuth', () => { + it('accepts a correctly signed request and returns the public key', async () => { + const signer = await Ed25519Signer.generate() + const auth = new RequestAuth([new Ed25519Verifier()], { now: () => NOW }) + const headers = await signed(signer) + expect(headers.get(AUTH_HEADERS.algorithm)).toBe('ed25519') + expect(signer.publicKey).toMatch(/^[0-9a-f]{64}$/) + expect(await auth.verify(headers, '/kv/readMany', '{"keys":["a"]}')).toBe(signer.publicKey) + }) + + it('rejects tampered bodies, paths and timestamps', async () => { + const signer = await Ed25519Signer.generate() + const auth = new RequestAuth([new Ed25519Verifier()], { now: () => NOW }) + await expect(auth.verify(await signed(signer), '/kv/readMany', '{"keys":["b"]}')).rejects.toThrow(/Invalid signature/) + await expect(auth.verify(await signed(signer), '/kv/writeMany', '{"keys":["a"]}')).rejects.toThrow(/Invalid signature/) + const headers = await signed(signer) + headers.set(AUTH_HEADERS.timestamp, String(NOW + 1)) + await expect(auth.verify(headers, '/kv/readMany', '{"keys":["a"]}')).rejects.toThrow(/Invalid signature/) + const other = await Ed25519Signer.generate() + const swapped = await signed(signer) + swapped.set(AUTH_HEADERS.publicKey, other.publicKey) + await expect(auth.verify(swapped, '/kv/readMany', '{"keys":["a"]}')).rejects.toThrow(/Invalid signature/) + }) + + it('rejects old or future timestamps, replays, missing headers and unknown algorithms', async () => { + const signer = await Ed25519Signer.generate() + const auth = new RequestAuth([new Ed25519Verifier()], { now: () => NOW, maxSkewMs: 1000 }) + await expect(auth.verify(await signed(signer, '/p', 'b', NOW - 5000), '/p', 'b')).rejects.toThrow(/window/) + await expect(auth.verify(await signed(signer, '/p', 'b', NOW + 5000), '/p', 'b')).rejects.toThrow(/window/) + const once = await signed(signer, '/p', 'b') + await auth.verify(once, '/p', 'b') + await expect(auth.verify(once, '/p', 'b')).rejects.toThrow(/Replayed/) + await expect(auth.verify(new Headers(), '/p', 'b')).rejects.toBeInstanceOf(AuthError) + const unknown = await signed(signer, '/p', 'c') + unknown.set(AUTH_HEADERS.algorithm, 'rot13') + await expect(auth.verify(unknown, '/p', 'c')).rejects.toThrow(/Unsupported signature algorithm "rot13"/) + }) + + it('restores an Ed25519 key from its exported private key', async () => { + const signer = await Ed25519Signer.generate() + const restored = await Ed25519Signer.fromPrivateKey(await signer.exportPrivateKey()) + expect(restored.publicKey).toBe(signer.publicKey) + const auth = new RequestAuth([new Ed25519Verifier()], { now: () => NOW }) + expect(await auth.verify(await signed(restored), '/kv/readMany', '{"keys":["a"]}')).toBe(signer.publicKey) + }) + + it('supports other algorithms through the verifier registry', async () => { + // A toy algorithm (signature = sha256(publicKey + payload)) standing in for e.g. secp256k1 via libauth. + const digest = async (publicKey: string, payload: Uint8Array) => + new Uint8Array(await crypto.subtle.digest('SHA-256', new Uint8Array([...new TextEncoder().encode(publicKey), ...payload]))) + const toySigner: RequestSigner = { algorithm: 'toy', publicKey: 'abcd', sign: (payload) => digest('abcd', payload) } + const toyVerifier: SignatureVerifier = { + algorithm: 'toy', + verify: async (publicKey, payload, signature) => (await digest(publicKey, payload)).every((byte, i) => byte === signature[i]), + } + const auth = new RequestAuth([new Ed25519Verifier(), toyVerifier], { now: () => NOW }) + expect(await auth.verify(await signed(toySigner), '/kv/readMany', '{"keys":["a"]}')).toBe('abcd') + }) +})