From ff0aacc9b497671ac68610caddda785d5302e805 Mon Sep 17 00:00:00 2001 From: Harvmaster Date: Mon, 27 Jul 2026 04:24:24 +0000 Subject: [PATCH] Add resource routes and signed writes --- docs/demo-app-transport-migration.md | 267 +++++++++++++++++++++++ src/app.ts | 9 +- src/routes/resources.ts | 302 +++++++++++++++++++++++++++ tests/routes/resources.test.ts | 140 +++++++++++++ 4 files changed, 717 insertions(+), 1 deletion(-) create mode 100644 docs/demo-app-transport-migration.md create mode 100644 src/routes/resources.ts create mode 100644 tests/routes/resources.test.ts diff --git a/docs/demo-app-transport-migration.md b/docs/demo-app-transport-migration.md new file mode 100644 index 0000000..d32b24b --- /dev/null +++ b/docs/demo-app-transport-migration.md @@ -0,0 +1,267 @@ +# Demo App Transport Migration + +This guide describes the client-facing protocol for the route-agnostic transport +architecture. + +## Connection Model + +HTTP, SSE, and WebSocket do not authenticate at the transport layer. Do not send +request-auth headers such as `X-PublicKey`, `X-Signature`, or `X-Timestamp` +for normal route dispatch. + +Authentication is scoped to each resource operation. Every written resource +carries its own `publicKey`, `timestamp`, and `signature`. + +All route input belongs in the request body. Paths are exact route names; there +are no path parameters, query parameters, or alternate HTTP methods. + +## Request Size Limit + +The server accepts at most 1048576 encoded bytes per HTTP request body or +complete WebSocket message by default. Deployments can change this with +`SERVER_MAX_REQUEST_BODY_BYTES`. The count is over the Extended JSON wire text, +so hexadecimal `Uint8Array` values use approximately twice their decoded byte +length. + +HTTP limits use Hono's built-in `bodyLimit` middleware and return `413`. +WebSocket messages do not pass through HTTP middleware; Hono's Node adapter uses +`ws`, whose native `maxPayload` option rejects oversized frames before the +application message handler. Response bodies are not limited. + +## HTTP Requests + +All application HTTP routes use POST. + +Read: + +```http +POST /data/get +Content-Type: application/json + +{ + "resourceId": ["resource-a", "resource-b"] +} +``` + +Write: + +```http +POST /data/write +Content-Type: application/json + +{ + "resources": [ + { + "id": "resource-a", + "publicKey": "...hex...", + "timestamp": 1730000000000, + "signature": "...der-hex...", + "value": "" + } + ] +} +``` + +Write response: + +```json +{ + "resources": [ + { + "id": "resource-a", + "publicKey": "...hex...", + "blob": "", + "timestamp": 1730000001000 + } + ] +} +``` + +If any item has invalid authorization, the whole batch fails. + +## SSE Subscriptions + +The client explicitly requests SSE using `Accept: text/event-stream`: + +```http +POST /data/subscribe +Accept: text/event-stream +Content-Type: application/json + +{ + "resourceId": ["resource-a"] +} +``` + +A subscription without this Accept header receives a `406` JSON error. + +The server keeps the SSE request open while its resource topics remain +subscribed. Subscribing sends no current state; it only enables future +publications. Request current state separately through `/data/get` when needed. + +SSE publication: + +```text +id: 1730000001000 +event: instance-changed +data: {"resourceId":"resource-a"} +``` + +After SSE begins, failures arrive as error events and the server closes that SSE +connection: + +```text +event: error +data: {"statusCode":500,"error":"Internal Server Error"} +``` + +To change an SSE subscription, abort the existing request and open a new +`/data/subscribe` request with the complete desired `resourceId` list. + +## WebSocket + +Connect without a connection-auth message: + +```ts +const ws = new WebSocket("ws://host:port/ws"); +``` + +Every client message uses this strict envelope. Messages may execute +concurrently; the server binds `id` and `body` to each individual dispatch so +responses remain correctly correlated even when they finish out of order: + +```ts +type WsRequest = { + id?: string; + path: string; + body?: unknown; +}; +``` + +Old fields such as `event`, `type`, `url`, `method`, `headers`, +`params`, and `data` are rejected. + +Read: + +```json +{ + "id": "read-1", + "path": "/data/get", + "body": { + "resourceId": ["resource-a", "resource-b"] + } +} +``` + +Write: + +```json +{ + "id": "write-1", + "path": "/data/write", + "body": { + "resources": [ + { + "id": "resource-a", + "publicKey": "...hex...", + "timestamp": 1730000000000, + "signature": "...der-hex...", + "value": "" + } + ] + } +} +``` + +Subscribe: + +```json +{ + "id": "sub-1", + "path": "/data/subscribe", + "body": { + "resourceId": ["resource-a"] + } +} +``` + +The subscribe request sends no immediate message. Its dispatch remains active +until those topics are removed or the connection closes. Request current state +separately through `/data/get` when needed. + +The client should consolidate its local subscribers and send only the topic set +needed by the shared WebSocket. + +Unsubscribe selected resources without closing the socket: + +```json +{ + "id": "unsub-1", + "path": "/data/unsubscribe", + "body": { + "resourceId": ["resource-a"] + } +} +``` + +Unsubscribe is idempotent. The response is a normal correlated response: + +```json +{ + "id": "unsub-1", + "type": "response", + "statusCode": 200, + "body": {} +} +``` + +Normal response: + +```json +{ + "id": "write-1", + "type": "response", + "statusCode": 200, + "body": {} +} +``` + +Error response: + +```json +{ + "id": "write-1", + "type": "error", + "statusCode": 400, + "error": "Validation Error", + "details": [] +} +``` + +Message-level errors do not close the WebSocket. The client decides whether to +retry, alter its subscription, or reconnect. + +Published events retain the shared stream shape: + +```json +{ + "id": "server-event-id", + "type": "instance-changed", + "data": {} +} +``` + +## Signing + +For each written resource, sign this canonical payload: + +```ts +`${timestamp}${resourceId}${canonicalBody(value)}`; +``` + +`value` is the `Uint8Array` payload. `canonicalBody(value)` uses the same +Extended JSON rules as the server: + +```json +"" +``` diff --git a/src/app.ts b/src/app.ts index d14b9bc..bedf6c7 100644 --- a/src/app.ts +++ b/src/app.ts @@ -6,6 +6,7 @@ import { HttpTransportRouter } from './services/transport/http-transport.ts'; import { WsTransportRouter } from './services/transport/ws-transport.ts'; import { ServerHost } from './services/server-host.ts'; import { Logger } from './utils/logger.ts'; +import { DataRoute } from './routes/resources.ts'; /** Application composition root. */ export class App { @@ -24,13 +25,19 @@ export class App { // Domain services are shared across all transports and route modules. const broadcaster = new Broadcaster(debug); - const routes = []; + const routes = [ + // DataRoute owns resource read/write/subscribe logic and maps resource + // ids to broadcaster topics. timestampWindowMs controls write replay protection. + new DataRoute(database, broadcaster, config.auth.timestampWindowMs), + ]; // Route loading is an explicit startup phase, not first-request work. // ApplicationRouter.create validates every path and rejects duplicates // before any client can connect. const router = await ApplicationRouter.create(routes); + // Both transports share one ApplicationRouter. Routes use the shared + // Broadcaster directly, while HTTP and WebSocket remain protocol adapters. const http = new HttpTransportRouter(router, debug); const ws = new WsTransportRouter(router, debug, config.server.maxRequestBodyBytes); const host = new ServerHost(config, debug, [http, ws]); diff --git a/src/routes/resources.ts b/src/routes/resources.ts new file mode 100644 index 0000000..eaf833d --- /dev/null +++ b/src/routes/resources.ts @@ -0,0 +1,302 @@ +import { createHash } from 'node:crypto'; +import { hexToBin, instantiateSecp256k1, type Secp256k1 } from '@bitauth/libauth'; +import { toExtendedJson } from '@xo-cash/utils'; +import { z } from 'zod'; + +import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.ts'; + +import type { BaseBroadcaster } from '../services/broadcaster.ts'; +import { ApplicationError, UnauthorizedError } from '../errors/index.ts'; +import type { Database } from '../services/storage/database.ts'; +import type { RouteDefinition, RouteModule, RouteStream } from './types.ts'; + +/** + * Schema to validate a single write resource. + * + * Each write targets one resource slot identified by (id, publicKey). + * The publicKey owner signs the payload to prove write authority. + */ +const writeResource = z.object({ + id: z.string().min(1), + publicKey: z.string().min(1), + timestamp: z.number().positive(), + signature: z.string().min(1), + value: z.instanceof(Uint8Array), +}); + +/** + * Schema to validate the write body. + * + * The body contains an array of write resources. + */ +const writeBody = z.object({ + resources: z.array(writeResource).min(1), +}); + +/** + * Schema to validate and deduplicate resource IDs supplied in an application body. + * + * The body contains an array of resource IDs. + */ +const resourceIdsSchema = z.object({ + resourceId: z + .array(z.string().min(1)) + .min(1, 'At least one resourceId is required') + .transform((ids) => [...new Set(ids)]), +}); + +type WriteResource = z.infer; + +/** + * Resource data domain routes. + * + * A "resource" is a shared document identified by id. Multiple instances + * (one per publicKey) can coexist within the same resource. Clients read, + * write, subscribe to changes, and unsubscribe over the transport-neutral + * RouteStream API. + */ +export class DataRoute implements RouteModule { + /** + * Promise to instantiate the secp256k1 library + * This is a bit annoying, but keeping a single instance alive makes more sense than instantiating it for each verification. + */ + private readonly secp256k1Promise: Promise = instantiateSecp256k1(); + + constructor( + private readonly database: Database, + private readonly broadcaster: BaseBroadcaster, + private readonly timestampWindowMs: number, + ) {} + + /** Declare exact routes; each handler owns its stream behavior. */ + async getRoutes(): Promise> { + return [ + { + url: '/data/get', + handler: this.getData.bind(this), + }, + { + url: '/data/write', + handler: this.writeData.bind(this), + }, + { + url: '/data/subscribe', + handler: this.subscribeData.bind(this), + }, + { + url: '/data/unsubscribe', + handler: this.unsubscribeData.bind(this), + }, + ]; + } + + /** Reads the requested resources from the database, returning all instances for each resource */ + async getData(stream: RouteStream): Promise { + const resourceIds = this.getResourceIds(stream); + + // Remove duplicates. + const uniqueIds = [...new Set(resourceIds)]; + + // If there are no resource ids, return an empty array. + if (uniqueIds.length === 0) { + await stream.send([]); + + return; + } + + // Read the data from the database. + const rows = await this.database.db + .selectFrom('resource_data') + .select(['resource_id', 'public_key', 'blob', 'timestamp']) + .where('resource_id', 'in', uniqueIds) + .orderBy('timestamp', 'asc') + .execute(); + + // Format the rows into the read resource responses. + const formattedRows = rows.map((row) => ({ + resourceId: row.resource_id, + publicKey: row.public_key, + blob: new Uint8Array(row.blob), + timestamp: row.timestamp, + })); + + await stream.send(formattedRows); + } + + /** + * Authenticated batch write. + * + * Each resource in the batch is verified independently, then persisted and + * broadcast to any subscribers listening on that resource's topic. + */ + async writeData(stream: RouteStream): Promise { + const { resources } = writeBody.parse(stream.body); + + // Authenticate the whole batch before producing any storage side effects. + // A single bad signature rejects the entire write — no partial commits. + await Promise.all(resources.map((resource) => this.verifyWriteResource(resource))); + + const timestamp = Date.now(); + const rows = resources.map((resource) => ({ + resourceId: resource.id, + publicKey: resource.publicKey, + blob: Buffer.from(resource.value), + })); + + // Upsert every row atomically; conflicts update blob and timestamp only. + await this.database.db.transaction().execute(async (trx) => { + for (const { resourceId, publicKey, blob } of rows) { + await trx + .insertInto('resource_data') + .values({ + resource_id: resourceId, + public_key: publicKey, + blob, + timestamp, + }) + .onConflict((oc) => + oc.columns(['resource_id', 'public_key']).doUpdateSet({ + blob, + timestamp, + }), + ) + .execute(); + } + }); + + // Format the rows into the written resource responses. + const written = rows.map(({ resourceId, publicKey, blob }) => ({ + resourceId, + instance: { + publicKey, + blob: new Uint8Array(blob), + timestamp, + }, + })); + + // Notify subscribers on each changed resource. Topic names are scoped per + // resource id so clients only receive events for resources they joined. + for (const { resourceId, instance } of written) { + await this.broadcaster.publish(DataRoute.resourceTopic(resourceId), { + type: 'instance-changed', + data: { resourceId, ...instance }, + }); + } + + // Return the persisted instances so the writer can confirm what was stored. + await stream.send({ + resources: written.map(({ resourceId, instance }) => ({ + id: resourceId, + ...instance, + })), + }); + } + + /** Subscribe this connection to future changes for the requested resources. */ + async subscribeData(stream: RouteStream): Promise { + const resourceIds = this.getResourceIds(stream); + + const topics = resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)); + + // Subscribe registers the topics synchronously, then keeps this route + // active until those topics are removed or the connection closes. + await this.broadcaster.subscribe(stream, topics); + } + + /** + * Leave resource topics without closing the connection. + * + * Only available on bidirectional transports (WebSocket). SSE clients + * unsubscribe implicitly by aborting the HTTP request. + */ + async unsubscribeData(stream: RouteStream): Promise { + const resourceIds = this.getResourceIds(stream); + + // If the stream is not bidirectional, throw an error. + if (!stream.bidirectional) { + throw new ApplicationError(HTTP_STATUS_CODE_NOT_ACCEPTED, 'This route requires an existing bidirectional stream'); + } + + // Unsubscribe from the resource topics. + await this.broadcaster.unsubscribe( + stream, + resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)), + ); + await stream.send({}); + } + + /** Extract resource id list from the decoded request body. */ + private getResourceIds(stream: RouteStream): string[] { + const body = + typeof stream.body === 'object' && stream.body !== null && !Array.isArray(stream.body) ? (stream.body as Record) : {}; + + return resourceIdsSchema.parse(body).resourceId; + } + + /** + * Verify one write's timestamp freshness and secp256k1 signature. + * + * The client must sign the canonical payload built from timestamp, resource + * id, and value — not the raw HTTP/WebSocket envelope. + */ + private async verifyWriteResource(resource: WriteResource): Promise { + this.assertFreshTimestamp(resource.timestamp); + + if (!(await this.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) { + throw new UnauthorizedError('Invalid resource signature'); + } + } + + /** + * Reject writes with stale timestamps to limit replay window. + * + * Both past and future timestamps outside the window are rejected. + */ + private assertFreshTimestamp(timestamp: number): void { + const age = Math.abs(Date.now() - timestamp); + if (age > this.timestampWindowMs) { + throw new UnauthorizedError('Timestamp outside allowed window'); + } + } + + /** + * Verify a secp256k1 signature. + * + * @param publicKeyHex - The public key to verify the signature against. + * @param signatureHex - The signature to verify. + * @param payload - The payload to verify the signature against. + * @returns Whether the signature is valid. + */ + private async verifySignature(publicKeyHex: string, signatureHex: string, payload: string): Promise { + const secp256k1 = await this.secp256k1Promise; + + try { + // Convert the public key and signature to binary. + const publicKey = hexToBin(publicKeyHex); + const signature = hexToBin(signatureHex); + + // Create a SHA-256 hash of the payload. + const messageHash = createHash('sha256').update(payload).digest(); + + // Low-S normalization rejects malleable signature encodings. + return secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash); + } catch { + return false; + } + } + + /** + * Build the transport-independent payload authenticated by each signature. + * + * canonicalBody ensures Uint8Array values hash consistently regardless of + * whether the client sent them over HTTP or WebSocket. + */ + private static canonicalWritePayload(resource: WriteResource): string { + return `${resource.timestamp}${resource.id}${toExtendedJson(resource.value)}`; + } + + /** Broadcaster topic for a single resource's instance-changed events. */ + private static resourceTopic(resourceId: string): string { + return `resource:${resourceId}`; + } +} diff --git a/tests/routes/resources.test.ts b/tests/routes/resources.test.ts new file mode 100644 index 0000000..ec0b7b9 --- /dev/null +++ b/tests/routes/resources.test.ts @@ -0,0 +1,140 @@ +import { describe, expect, it, vi } from "vitest"; + +import { DataRoute } from "../../src/routes/resources.js"; +import { UnauthorizedError } from "../../src/errors/index.js"; +import { type BaseBroadcaster } from "../../src/services/broadcaster.js"; +import { ApplicationRouteStream } from "../../src/services/route-stream.js"; +import { Database } from "../../src/services/storage/database.js"; +import { TestConnection } from "../helpers/test-connection.js"; +import { HTTP_STATUS_CODE_NOT_ACCEPTED } from "../../src/constants.js"; + +function createBroadcasterStub() { + return { + subscribe: vi.fn(), + unsubscribe: vi.fn().mockResolvedValue(undefined), + publish: vi.fn(), + sendEvent: vi.fn(), + } as unknown as BaseBroadcaster; +} + +describe("DataRoute subscriptions", () => { + it("subscribes to future resource changes until removal", async () => { + let resolveRemoved: () => void = () => undefined; + const removed = new Promise((resolve) => { + resolveRemoved = resolve; + }); + const storage = { + db: { + transaction: vi.fn(), + }, + } as unknown as Database; + const broadcaster = createBroadcasterStub(); + vi.mocked(broadcaster.subscribe).mockReturnValue(removed); + const connection = new TestConnection(true, false); + const stream = new ApplicationRouteStream(connection, { + resourceId: ["a", "b"], + }); + const route = new DataRoute(storage, broadcaster, 0); + + const execution = route.subscribeData(stream); + + expect(broadcaster.subscribe).toHaveBeenCalledWith( + stream, + ["resource:a", "resource:b"], + ); + expect(storage.db.transaction).not.toHaveBeenCalled(); + expect(connection.messages).toEqual([]); + await expect( + Promise.race([ + execution.then(() => "settled"), + Promise.resolve("pending"), + ]), + ).resolves.toBe("pending"); + + resolveRemoved(); + await execution; + }); + + it("unsubscribes a bidirectional connection and acknowledges the request", async () => { + const storage = { + db: { + transaction: vi.fn(), + }, + } as unknown as Database; + const broadcaster = createBroadcasterStub(); + const connection = new TestConnection(true, true); + const stream = new ApplicationRouteStream( + connection, + { resourceId: ["a"] }, + "unsubscribe-1", + ); + const route = new DataRoute(storage, broadcaster, 0); + + await route.unsubscribeData(stream); + + expect(broadcaster.unsubscribe).toHaveBeenCalledWith(stream, [ + "resource:a", + ]); + expect(connection.messages).toEqual([ + { + id: "unsubscribe-1", + type: "response", + statusCode: 200, + body: {}, + }, + ]); + }); + + it("rejects selective unsubscribe on a one-way connection", async () => { + const storage = { + db: { + transaction: vi.fn(), + }, + } as unknown as Database; + const broadcaster = createBroadcasterStub(); + const stream = new ApplicationRouteStream(new TestConnection(true, false), { + resourceId: ["a"], + }); + const route = new DataRoute(storage, broadcaster, 0); + + await expect(route.unsubscribeData(stream)).rejects.toMatchObject({ statusCode: HTTP_STATUS_CODE_NOT_ACCEPTED }); + expect(broadcaster.unsubscribe).not.toHaveBeenCalled(); + }); +}); + +describe("DataRoute resource write auth", () => { + it("rejects an invalid resource signature before writing the batch", async () => { + const storage = { + db: { + transaction: vi.fn(), + }, + } as unknown as Database; + + const broadcaster = createBroadcasterStub() + const route = new DataRoute(storage, broadcaster, 0); + + await expect( + route.writeData({ + connection: new TestConnection(true, true), + streaming: true, + bidirectional: true, + send: vi.fn(), + body: { + resources: [ + { + id: "resource-a", + publicKey: "not-a-public-key", + timestamp: Date.now(), + signature: "not-a-signature", + value: new Uint8Array([1, 2, 3]), + }, + ], + }, + } as unknown as ApplicationRouteStream), + ).rejects.toBeInstanceOf(UnauthorizedError); + + expect(storage.db.transaction).not.toHaveBeenCalled() + expect(broadcaster.publish).not.toHaveBeenCalled(); + }); +}); +