diff --git a/DESIGN.md b/DESIGN.md index 2958383..e50d9d1 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -128,20 +128,19 @@ no own value. A node's own value and its children are independent keys. 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. +- **Signed requests.** The client's `RequestSigner` signs `${timestamp}:${nonce}:${path}:${body}` + (nonce = 16 random bytes as hex, fresh per request; path = URL pathname; body = the exact request + body). Based on Harvmaster/sync-server-example, plus the nonce. Headers: `X-Public-Key` + (lowercase hex), `X-Timestamp` (ms), `X-Nonce`, `X-Signature` (hex) and `X-Signature-Algorithm`. + Every request is signed, and so is each SSE (re)connect, with a fresh timestamp and nonce. - **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. -- **Unique timestamps.** Ed25519 signatures are deterministic, so identical requests in the same - millisecond would carry the same signature and the second would be rejected as a replay. - `RequestAuth.headers` makes timestamps strictly increase per public key within a process. Two - processes (e.g. browser tabs) with the same key can still collide, rarely; a nonce in the - payload would fix that but changes the signing scheme. + (unknown → 401), checks the nonce format (16–128 characters of `[0-9A-Za-z_-]`) and that the + timestamp is within `maxSkewMs` (default 60 s). It then rejects a nonce the same public key + already used within that window (replay protection), and verifies the signature over the + received timestamp, nonce, path and raw body. The body is parsed only afterwards. +- **Why a nonce.** Ed25519 signatures are deterministic, so without it two identical requests + from one key at the same millisecond (two tabs or devices subscribing to the same path, parallel + identical reads) would be indistinguishable, and the second rejected as a replay. - **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` @@ -178,7 +177,9 @@ is only appropriate when the client may read all of it. only where they differ from what the first replica holds. No write tagging or echo tracking is needed: echoes of this client's writes change nothing. - **`sync()`**: replay the queue in order (send, then re-apply locally) → merge-scan the source - against the last replica (both key-ordered), applying the difference in batches of 500 → + against the last replica (both key-ordered), applying the difference in batches of 500, except + for keys with local writes the source hasn't confirmed yet (in flight or queued; counted per + key, and reloaded from a persistent queue on start), which would otherwise briefly revert → replay again (writes made meanwhile) → mark `synced`. It runs on start, on `resync` (the stream reconnected), and `retryMs` after a failure while offline. Called while a sync is running, it schedules one more after it (shared by all callers meanwhile), so changes after the running diff --git a/README.md b/README.md index a98ee43..71d769c 100644 --- a/README.md +++ b/README.md @@ -239,10 +239,11 @@ key a value is stored under is authenticated too, so the server can't move value ### Signed requests -Every request is signed with the user's key: `X-Public-Key`, `X-Timestamp`, `X-Signature` and -`X-Signature-Algorithm` headers, over `${timestamp}:${path}:${body}`. The server checks the -signature and a 60 s timestamp window (with replay protection), then that every key is the public -key or under it. Ed25519 is built in (WebCrypto); other algorithms plug in by implementing +Every request is signed with the user's key: `X-Public-Key`, `X-Timestamp`, `X-Nonce` (random +per request), `X-Signature` and `X-Signature-Algorithm` headers, over +`${timestamp}:${nonce}:${path}:${body}`. The server checks the signature, a 60 s timestamp +window and that the nonce wasn't used before (replay protection), then that every key is the +public key or under it. Ed25519 is built in (WebCrypto); other algorithms plug in by implementing `RequestSigner` / `SignatureVerifier`. ### Live updates without the composite diff --git a/src/kv-storage/kv-storage-composite.ts b/src/kv-storage/kv-storage-composite.ts index 69b305c..637f42c 100644 --- a/src/kv-storage/kv-storage-composite.ts +++ b/src/kv-storage/kv-storage-composite.ts @@ -51,7 +51,8 @@ export interface KVStorageComposite { * client's own writes therefore changes nothing. * - A write the source can't take (`StorageUnavailableError`) is applied locally and stored in * the `queue` store; later writes queue behind it, so order is kept. - * - `sync()` replays the queue, then copies what changed on the source into the replicas. It runs + * - `sync()` replays the queue, then copies what changed on the source into the replicas, except + * keys with local writes the source hasn't confirmed yet (in flight or queued). It runs * on start, when the source reconnects (`resync`), and after a retry delay while offline. * - Reads wait for the first sync (until one succeeds, they go to the source). Whether a sync has * completed is kept in the `queue` store, so with a persistent one a restart reads offline. @@ -77,9 +78,11 @@ export class KVStorageComposite extends BaseKVStorage< queued: number /** Remote changes are applied one after another. */ applying: Promise + /** Keys with local writes the source hasn't confirmed yet (in flight or queued) → count. */ + pending: Map retry?: ReturnType offs: OffCallback[] - } = { synced: false, queued: 0, applying: Promise.resolve(), offs: [] } + } = { synced: false, queued: 0, applying: Promise.resolve(), pending: new Map(), offs: [] } constructor(options: KVStorageCompositeOptions) { super() @@ -140,20 +143,25 @@ export class KVStorageComposite extends BaseKVStorage< const keys = changes.map(({ key }) => key) this.checkRange(keys) await this.start() - // Applied locally first, so the source relaying this write back finds nothing new. + // Applied locally first, so the source relaying this write back finds nothing new. Until the + // source confirms it, a sync must not replace it with the source's (older) value. const previous = await this.replicas[0]!.readMany(keys) + this.markPending(keys, 1) await this.apply(changes) // While syncing or with writes queued, queue this one too so the source sees writes in order. - if (this.state.queued === 0 && !this.state.syncing) { - try { - return await this.source.writeMany(changes) - } catch (error) { - if (!(error instanceof StorageUnavailableError)) { - await this.apply(keys.map((key, i) => ({ key, value: previous[i] }))) - throw error - } - } + if (this.state.queued > 0 || this.state.syncing) return this.enqueue(changes) + try { + await this.source.writeMany(changes) + } catch (error) { + if (error instanceof StorageUnavailableError) return this.enqueue(changes) + this.markPending(keys, -1) + await this.apply(keys.map((key, i) => ({ key, value: previous[i] }))) + throw error } + this.markPending(keys, -1) + } + + private async enqueue(changes: KVChange[]): Promise { await this.writes.set(this.ids.generate(), changes) this.state.queued++ this.retryLater() @@ -167,7 +175,13 @@ export class KVStorageComposite extends BaseKVStorage< private async open(): Promise { this.state.synced = (await this.queue.get('synced')) === true - for await (const _ of this.writes.keys()) this.state.queued++ + for await (const [, changes] of this.writes.entries()) { + this.state.queued++ + this.markPending( + (changes as KVChange[]).map(({ key }) => key), + 1, + ) + } await this.seed() this.state.offs.push( this.source.on('change', (event) => { @@ -192,7 +206,10 @@ export class KVStorageComposite extends BaseKVStorage< try { await this.replay() let batch: KVChange[] = [] - for await (const change of this.diff()) if (batch.push(change) >= 500) await this.apply(batch.splice(0), true) + for await (const change of this.diff()) { + if (this.state.pending.has(change.key)) continue // a newer local write, not yet sent + if (batch.push(change) >= 500) await this.apply(batch.splice(0), true) + } await this.apply(batch, true) await this.replay() // writes made during the diff: send them, and re-apply them over it await this.queue.set('synced', true) @@ -224,6 +241,10 @@ export class KVStorageComposite extends BaseKVStorage< } await this.writes.delete(next[0]) this.state.queued-- + this.markPending( + changes.map(({ key }) => key), + -1, + ) } } @@ -275,6 +296,14 @@ export class KVStorageComposite extends BaseKVStorage< await flush() } + private markPending(keys: string[], delta: 1 | -1): void { + for (const key of keys) { + const count = (this.state.pending.get(key) ?? 0) + delta + if (count > 0) this.state.pending.set(key, count) + else this.state.pending.delete(key) + } + } + private retryLater(): void { clearTimeout(this.state.retry) this.state.retry = setTimeout(() => void this.sync(), this.retryMs) diff --git a/src/utils/request-auth.ts b/src/utils/request-auth.ts index c94f39c..3fb89a5 100644 --- a/src/utils/request-auth.ts +++ b/src/utils/request-auth.ts @@ -19,6 +19,7 @@ export interface SignatureVerifier { export const AUTH_HEADERS = { publicKey: 'X-Public-Key', timestamp: 'X-Timestamp', + nonce: 'X-Nonce', signature: 'X-Signature', algorithm: 'X-Signature-Algorithm', } as const @@ -30,17 +31,18 @@ export interface RequestAuthOptions { } /** - * 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). + * Signed-request authentication. The client signs `${timestamp}:${nonce}:${path}:${body}` (nonce = + * random per request, path = URL pathname, body = the exact request body) and sends the public + * key, timestamp, nonce, 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 + * nonce the same key already used within the timestamp window (replay protection). The nonce also + * keeps identical requests made at the same moment distinct (Ed25519 signatures are deterministic). */ 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. */ + /** `:` accepted within the window → when it expires; insertion order ≈ expiry order. */ private readonly seen = new Map() constructor(verifiers: SignatureVerifier[], options: RequestAuthOptions = {}) { @@ -50,27 +52,19 @@ export class RequestAuth { this.now = options.now ?? Date.now } - static payload(timestamp: string, path: string, body: string): Uint8Array { - return new TextEncoder().encode(`${timestamp}:${path}:${body}`) + static payload(timestamp: string, nonce: string, path: string, body: string): Uint8Array { + return new TextEncoder().encode(`${timestamp}:${nonce}:${path}:${body}`) } - /** Last timestamp used per public key in this process (see `headers`). */ - private static readonly lastTimestamps = new Map() - - /** - * Auth headers for a request to `path` with `body`. Timestamps per public key strictly increase - * within a process: Ed25519 signatures are deterministic, so two identical requests in the same - * millisecond would otherwise carry the same signature, and the second would be rejected as a - * replay. - */ + /** Auth headers for a request to `path` with `body`. */ static async headers(signer: RequestSigner, path: string, body: string, now: number = Date.now()): Promise> { - const time = Math.max(now, (RequestAuth.lastTimestamps.get(signer.publicKey) ?? 0) + 1) - RequestAuth.lastTimestamps.set(signer.publicKey, time) - const timestamp = String(time) - const signature = await signer.sign(RequestAuth.payload(timestamp, path, body)) + const timestamp = String(now) + const nonce = Hex.encode(crypto.getRandomValues(new Uint8Array(16))) + const signature = await signer.sign(RequestAuth.payload(timestamp, nonce, path, body)) return { [AUTH_HEADERS.publicKey]: signer.publicKey, [AUTH_HEADERS.timestamp]: timestamp, + [AUTH_HEADERS.nonce]: nonce, [AUTH_HEADERS.signature]: Hex.encode(signature), [AUTH_HEADERS.algorithm]: signer.algorithm, } @@ -80,32 +74,35 @@ export class RequestAuth { async verify(headers: Headers, path: string, body: string): Promise { const publicKey = headers.get(AUTH_HEADERS.publicKey) const timestamp = headers.get(AUTH_HEADERS.timestamp) + const nonce = headers.get(AUTH_HEADERS.nonce) 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') + if (!publicKey || !timestamp || !nonce || !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') + if (!/^[0-9A-Za-z_-]{16,128}$/.test(nonce)) throw new AuthError(401, 'Nonce must be 16-128 characters of [0-9A-Za-z_-]') 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') + const used = `${publicKey}:${nonce}` + if (this.seen.has(used)) throw new AuthError(401, 'Replayed request') let valid = false try { - valid = await verifier.verify(publicKey, RequestAuth.payload(timestamp, path, body), Hex.decode(signature)) + valid = await verifier.verify(publicKey, RequestAuth.payload(timestamp, nonce, path, body), Hex.decode(signature)) } catch { valid = false } if (!valid) throw new AuthError(401, 'Invalid signature') - this.seen.set(signature, time + this.maxSkewMs) + this.seen.set(used, time + this.maxSkewMs) return publicKey } private forgetExpired(now: number): void { - for (const [signature, expires] of this.seen) { + for (const [used, expires] of this.seen) { if (expires >= now) break - this.seen.delete(signature) + this.seen.delete(used) } } } diff --git a/test/kv-storage/kv-storage-composite.test.ts b/test/kv-storage/kv-storage-composite.test.ts index 09d64eb..4a6b55e 100644 --- a/test/kv-storage/kv-storage-composite.test.ts +++ b/test/kv-storage/kv-storage-composite.test.ts @@ -257,4 +257,29 @@ describe('KVStorageComposite sync', () => { ['b', 2], ]) }) + + it('a sync started while a write is on its way to the source does not undo it', async () => { + let release!: () => void + const gate = new Promise((r) => (release = r)) + const inner = new KVStorageInMemory() + const source = new KVStorageCounting(inner) + const writeMany = source.writeMany.bind(source) + source.writeMany = async (changes) => { + await gate // a slow request + return writeMany(changes) + } + const replica = new KVStorageInMemory() + const db = new KVStorageComposite({ source, replicas: [replica] }) + composites.push(db) + await db.sync() + const writing = db.set('a', 1) + await vi.waitFor(async () => expect(await replica.readMany(['a'])).toEqual([1])) // applied locally, not yet on the source + const syncing = db.sync() + await new Promise((r) => setTimeout(r, 10)) + expect(await replica.readMany(['a'])).toEqual([1]) // the sync skips the unconfirmed key + release() + await Promise.all([writing, syncing]) + expect(await replica.readMany(['a'])).toEqual([1]) + expect(await inner.readMany(['a'])).toEqual([1]) + }) }) diff --git a/test/utils/request-auth.test.ts b/test/utils/request-auth.test.ts index 706f07c..8c4099f 100644 --- a/test/utils/request-auth.test.ts +++ b/test/utils/request-auth.test.ts @@ -18,38 +18,48 @@ describe('RequestAuth', () => { expect(await auth.verify(headers, '/kv/readMany', '{"keys":["a"]}')).toBe(signer.publicKey) }) - it('rejects tampered bodies, paths and timestamps', async () => { + it('rejects tampered bodies, paths, timestamps and nonces', 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(Number(headers.get(AUTH_HEADERS.timestamp)) + 1)) + headers.set(AUTH_HEADERS.timestamp, String(NOW + 1)) await expect(auth.verify(headers, '/kv/readMany', '{"keys":["a"]}')).rejects.toThrow(/Invalid signature/) + const renonced = await signed(signer) + renonced.set(AUTH_HEADERS.nonce, 'f'.repeat(32)) + await expect(auth.verify(renonced, '/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('accepts identical requests made in the same millisecond (timestamps per key increase)', async () => { + it('accepts identical requests made at the same moment (each has its own nonce)', async () => { const signer = await Ed25519Signer.generate() const sameKey = await Ed25519Signer.fromPrivateKey(await signer.exportPrivateKey()) const auth = new RequestAuth([new Ed25519Verifier()], { now: () => NOW }) const requests = await Promise.all([signed(signer), signed(signer), signed(sameKey)]) - expect(new Set(requests.map((h) => h.get(AUTH_HEADERS.signature))).size).toBe(3) + expect(new Set(requests.map((h) => h.get(AUTH_HEADERS.timestamp)))).toEqual(new Set([String(NOW)])) + expect(new Set(requests.map((h) => h.get(AUTH_HEADERS.nonce))).size).toBe(3) for (const headers of requests) expect(await auth.verify(headers, '/kv/readMany', '{"keys":["a"]}')).toBe(signer.publicKey) }) 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(await Ed25519Signer.generate(), '/p', 'b', NOW - 5000), '/p', 'b')).rejects.toThrow(/window/) - await expect(auth.verify(await signed(await Ed25519Signer.generate(), '/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/) + 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 noNonce = await signed(signer, '/p', 'd') + noNonce.delete(AUTH_HEADERS.nonce) + await expect(auth.verify(noNonce, '/p', 'd')).rejects.toThrow(/Missing/) + const shortNonce = await signed(signer, '/p', 'e') + shortNonce.set(AUTH_HEADERS.nonce, 'abc') + await expect(auth.verify(shortNonce, '/p', 'e')).rejects.toThrow(/Nonce/) 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"/)