diff --git a/package.json b/package.json index 233b721..13a3ecd 100644 --- a/package.json +++ b/package.json @@ -63,7 +63,7 @@ "@types/debug": "^4.1.13", "@types/node": "^25.9.3", "@types/ws": "^8.18.1", - "@vitest/coverage-v8": "^4.1.9", + "@vitest/coverage-v8": "^5.0.0", "@viz-kit/esbuild-analyzer": "^1.0.0", "@xo-cash/eslint-config": "1.0.2", "cspell": "^10.0.1", @@ -74,6 +74,6 @@ "typedoc-plugin-coverage": "^4.0.2", "typescript": "^6.0.3", "typescript-eslint": "^8.65.0", - "vitest": "^4.1.9" + "vitest": "^5.0.0" } } diff --git a/source/auth/auth.ts b/source/auth/auth.ts index 929a243..5ef6402 100644 --- a/source/auth/auth.ts +++ b/source/auth/auth.ts @@ -23,7 +23,7 @@ export class AuthSecp256k1 { */ static async create(inputDeps: AuthSecp256k1Deps, options: AuthSecp256k1Options): Promise { const deps = { - secp256k1: await instantiateSecp256k1(), + secp256k1: inputDeps.secp256k1 || await instantiateSecp256k1(), ...inputDeps, }; diff --git a/source/index.ts b/source/index.ts index cf09105..b61e363 100644 --- a/source/index.ts +++ b/source/index.ts @@ -32,7 +32,13 @@ export class App { 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), + new DataRoute({ + auth: auth, + database: database, + broadcaster: broadcaster, + }, { + timestampWindowMs: config.auth.timestampWindowMs, + }), ]; // Route loading is an explicit startup phase, not first-request work. diff --git a/source/routes/resources.ts b/source/routes/resources.ts index ac7a250..fbfce4d 100644 --- a/source/routes/resources.ts +++ b/source/routes/resources.ts @@ -1,5 +1,3 @@ -import { createHash } from 'node:crypto'; -import { hexToBin, instantiateSecp256k1, type Secp256k1 } from '@bitauth/libauth'; import { toExtendedJson } from '@xo-cash/utils'; import { z } from 'zod'; @@ -9,6 +7,7 @@ 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'; +import { AuthSecp256k1 } from '../auth/auth.ts'; /** * Schema to validate a single write resource. @@ -47,6 +46,16 @@ const resourceIdsSchema = z.object({ type WriteResource = z.infer; +export type DataRouteDeps = { + auth: AuthSecp256k1; + database: Database; + broadcaster: BaseBroadcaster; +} + +export type DataRouteOptions = { + timestampWindowMs: number; +} + /** * Resource data domain routes. * @@ -56,16 +65,9 @@ type WriteResource = z.infer; * 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, + private readonly deps: DataRouteDeps, + private readonly options: DataRouteOptions, ) {} /** Declare exact routes; each handler owns its stream behavior. */ @@ -105,7 +107,7 @@ export class DataRoute implements RouteModule { } // Read the data from the database. - const rows = await this.database.db + const rows = await this.deps.database.db .selectFrom('resource_data') .select([ 'resource_id', 'public_key', 'blob', 'timestamp' ]) .where('resource_id', 'in', uniqueIds) @@ -140,17 +142,19 @@ export class DataRoute implements RouteModule { const rows = resources.map((resource) => ({ resourceId: resource.id, publicKey: resource.publicKey, + signature: resource.signature, 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 this.deps.database.db.transaction().execute(async (trx) => { + for (const { resourceId, publicKey, blob, signature } of rows) { await trx .insertInto('resource_data') .values({ resource_id: resourceId, public_key: publicKey, + signature: signature, blob, timestamp, }) @@ -176,7 +180,7 @@ export class DataRoute implements RouteModule { // 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), { + await this.deps.broadcaster.publish(DataRoute.resourceTopic(resourceId), { type: 'instance-changed', data: { resourceId, ...instance }, }); @@ -199,7 +203,7 @@ export class DataRoute implements RouteModule { // 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); + await this.deps.broadcaster.subscribe(stream, topics); } /** @@ -217,7 +221,7 @@ export class DataRoute implements RouteModule { } // Unsubscribe from the resource topics. - await this.broadcaster.unsubscribe( + await this.deps.broadcaster.unsubscribe( stream, resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)), ); @@ -241,7 +245,7 @@ export class DataRoute implements RouteModule { private async verifyWriteResource(resource: WriteResource): Promise { this.assertFreshTimestamp(resource.timestamp); - if (!(await this.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) { + if (!(await this.deps.auth.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) { throw new UnauthorizedError('Invalid resource signature'); } } @@ -253,38 +257,11 @@ export class DataRoute implements RouteModule { */ private assertFreshTimestamp(timestamp: number): void { const age = Math.abs(Date.now() - timestamp); - if (age > this.timestampWindowMs) { + if (age > this.options.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. * @@ -292,7 +269,7 @@ export class DataRoute implements RouteModule { * whether the client sent them over HTTP or WebSocket. */ private static canonicalWritePayload(resource: WriteResource): string { - return `${resource.timestamp}${resource.id}${toExtendedJson(resource.value)}`; + return `${resource.timestamp}:${resource.id}:${toExtendedJson(resource.value)}`; } /** Broadcaster topic for a single resource's instance-changed events. */ diff --git a/source/services/transport/base-transport.ts b/source/services/transport/base-transport.ts new file mode 100644 index 0000000..8525189 --- /dev/null +++ b/source/services/transport/base-transport.ts @@ -0,0 +1,59 @@ +import { Hono } from "hono"; +import { RequestHeaders } from "../../routes/types"; +import { ApplicationError } from "../../errors"; +import { HTTP_STATUS_CODE_BAD_REQUEST } from "../../constants"; + +/** RFC 9110 field-name token grammar. */ +const HEADER_NAME_PATTERN = /^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$/; + +/** Hono variables populated by transport-boundary middleware. */ +export type AppEnv = { + Variables: { + + /** Decoded Extended JSON request body, when present. */ + parsedBody?: unknown; + + /** Raw JSON text preserved for signature verification. */ + rawJsonBody?: string; + }; +}; + +export abstract class BaseTransport { + /** Attach wire endpoints and middleware to the shared Hono application. */ + abstract register(app: Hono): void + + /** Close transport-owned long-lived connections during server shutdown. */ + abstract stop(): Promise + + /** + * Validate and normalize request headers at a transport boundary. + * + * Lowercase names give HTTP and WebSocket routes identical lookup semantics. + * Case-insensitive duplicates are rejected instead of selecting an ambiguous + * authentication value. Header values may not contain line breaks. + */ + public static normalizeRequestHeaders(headers: Readonly>): RequestHeaders { + const normalizedEntries: Array<[string, string]> = []; + const names = new Set(); + + for (const [ name, value ] of Object.entries(headers)) { + if (!HEADER_NAME_PATTERN.test(name)) { + throw new ApplicationError(HTTP_STATUS_CODE_BAD_REQUEST, 'Invalid request header name'); + } + + if (value.includes('\r') || value.includes('\n')) { + throw new ApplicationError(HTTP_STATUS_CODE_BAD_REQUEST, 'Invalid request header value'); + } + + const normalizedName = name.toLowerCase(); + if (names.has(normalizedName)) { + throw new ApplicationError(HTTP_STATUS_CODE_BAD_REQUEST, 'Duplicate request header name'); + } + + names.add(normalizedName); + normalizedEntries.push([ normalizedName, value ]); + } + + return Object.freeze(Object.fromEntries(normalizedEntries)); + }; +} \ No newline at end of file diff --git a/source/services/transport/http-transport.ts b/source/services/transport/http-transport.ts index ec8b2a6..af71d4b 100644 --- a/source/services/transport/http-transport.ts +++ b/source/services/transport/http-transport.ts @@ -5,12 +5,12 @@ import { toExtendedJson, fromExtendedJson } from '@xo-cash/utils'; import type { Logger } from '../../utils/logger.ts'; import type { ApplicationRequest, ApplicationRouter } from '../router.ts'; import type { StreamResponse } from '../stream/base-stream.ts'; -import type { AppEnv, TransportRouter } from './transport-router.ts'; import { ApplicationError, normalizePublicError } from '../../errors/index.ts'; import { HTTP_STATUS_CODE_BAD_REQUEST, HTTP_STATUS_CODE_NO_CONTENT } from '../../constants.ts'; import { HonoSSEStream } from '../stream/hono-sse-stream.ts'; import { HttpRequestStream } from '../stream/http-request-stream.ts'; +import { type AppEnv, BaseTransport } from './base-transport.ts'; /** Hono context key where decoded Extended JSON bodies are stored. */ const PARSED_BODY_KEY = 'parsedBody'; @@ -21,7 +21,7 @@ const PARSED_BODY_KEY = 'parsedBody'; * Normal HTTP and SSE both enter the same application router with different * connection-stream capabilities. */ -export class HttpTransportRouter implements TransportRouter { +export class HttpTransportRouter extends BaseTransport { private readonly debug: Logger; /** SSE connections retained until their final subscription or peer closes. */ @@ -35,6 +35,8 @@ export class HttpTransportRouter implements TransportRouter { private readonly router: ApplicationRouter, debug: Logger, ) { + super(); + this.debug = debug.extend('http-transport'); } @@ -111,6 +113,7 @@ export class HttpTransportRouter implements TransportRouter { return { path: context.req.path, ...(body === undefined ? {} : { body }), + headers: context.req.header(), }; } diff --git a/source/services/transport/ws-transport.ts b/source/services/transport/ws-transport.ts index 544d467..0b79864 100644 --- a/source/services/transport/ws-transport.ts +++ b/source/services/transport/ws-transport.ts @@ -8,11 +8,11 @@ import { z } from 'zod'; import type { Logger } from '../../utils/logger.ts'; import type { ApplicationRequest, ApplicationRouter } from '../router.ts'; -import type { AppEnv, UpgradeTransportRouter } from './transport-router.ts'; import { ApplicationError, normalizePublicError } from '../../errors/index.ts'; import { HTTP_STATUS_CODE_BAD_REQUEST } from '../../constants.ts'; import { WSStream } from '../stream/ws-stream.ts'; +import { type AppEnv, BaseTransport } from './base-transport.ts'; /** Default WebSocket upgrade path for application messages. */ const WS_ROUTE = '/ws'; @@ -33,7 +33,7 @@ const wsRequestSchema = z * One WSStream is shared by every message on a socket, allowing subscribe and * unsubscribe requests to operate on the same broadcaster registration. */ -export class WsTransportRouter implements UpgradeTransportRouter { +export class WsTransportRouter extends BaseTransport { /** * Native Node WebSocket server used by Hono's Node adapter. * @@ -42,9 +42,6 @@ export class WsTransportRouter implements UpgradeTransportRouter { */ private readonly wsServer: WebSocketServer; - /** Upgrade server wired into the Node HTTP listener by ServerHost. */ - readonly websocketServer: WebSocketServerLike; - private readonly debug: Logger; /** @@ -59,6 +56,8 @@ export class WsTransportRouter implements UpgradeTransportRouter { private readonly maxRequestBodyBytes: number, private readonly url: string = WS_ROUTE, ) { + super(); + this.debug = debug.extend('ws-transport'); // Hono's Node WebSocket helper delegates frame handling to `ws`; unlike @@ -68,7 +67,6 @@ export class WsTransportRouter implements UpgradeTransportRouter { noServer: true, maxPayload: this.maxRequestBodyBytes, }); - this.websocketServer = this.wsServer as unknown as WebSocketServerLike; } /** diff --git a/test/helpers/controlled-request.ts b/test/helpers/controlled-request.ts new file mode 100644 index 0000000..4d9b09c --- /dev/null +++ b/test/helpers/controlled-request.ts @@ -0,0 +1,68 @@ +import { ApplicationRouter } from "../../source/services/router"; +import { BaseStream } from "../../source/services/stream/base-stream"; + +export type ControlledRequestDeps = { + router: ApplicationRouter; + connection: BaseStream; +} + +export type ControlledRequestOptions = { + path: string; + requestId: string; + body?: unknown; + headers?: Record; + includeAuthHeaders?: boolean; + timeout?: number; +} + +export type ControlledRequestParams = ControlledRequestOptions & ControlledRequestDeps; + +export type ControlledRequest = { + request: Promise; + started: Promise; + release: () => void; +} + +export const createControlledRequest = ({ router, connection, ...requestOptions }: ControlledRequestParams): ControlledRequest => { + // Create promises with resolvers to signal the request has started and released + const { promise: started, resolve: signalStarted } = Promise.withResolvers(); + const { promise: released, resolve: release } = Promise.withResolvers(); + + // The request options (sets the includeAuthHeaders to true by default) + const options = { + includeAuthHeaders: true, + ...requestOptions + } + + // The test auth headers + const authHeaders = { + 'x-public-key': 'public-key', + 'x-signature': 'signature', + 'x-timestamp': '1000', + } + + // If the request should include the test auth headers (default is true) + if (options.includeAuthHeaders) { + options.headers = { + ...options.headers, + ...authHeaders, + }; + } + + // Add the signalStarted and released promises to the request body + options.body = { + ...(options.body ?? {}), + signalStarted, + released, + } + + // Dispatch the request + const request = router.dispatch(options, connection); + + // Return the request, started, and release promises + return { + request, + started, + release, + }; +}; diff --git a/test/helpers/misc.ts b/test/helpers/misc.ts new file mode 100644 index 0000000..927782e --- /dev/null +++ b/test/helpers/misc.ts @@ -0,0 +1,29 @@ +import { vi } from "vitest"; +import type { AuthSecp256k1 } from "../../source/auth/auth"; +import type { RouteDefinition, RouteModule } from "../../source/routes/types"; + +/** + * Convert an array of route definitions to a route module by returning an object with a getRoutes method that returns the routes. + * @param routes - The route definitions to convert. + * @returns A route module. + */ +export const toRoutes = (routes: RouteDefinition[]): RouteModule => { + return { + getRoutes: async () => routes, + }; +}; + + +export const mockAuth = { + verifySignature: vi.fn().mockResolvedValue(true), + verifyUniqueRequest: vi.fn().mockResolvedValue(true), + assertTimestampFreshness: vi.fn().mockResolvedValue(true), +} as unknown as AuthSecp256k1; + +/** + * Create a mock of the AuthSecp256k1 service + * @returns A mock of the AuthSecp256k1 service + */ +export const createMockAuth = (): AuthSecp256k1 => { + return mockAuth; +}; diff --git a/test/routes/resources.test.ts b/test/routes/resources.test.ts index 65a47b1..c6e7558 100644 --- a/test/routes/resources.test.ts +++ b/test/routes/resources.test.ts @@ -7,6 +7,8 @@ import { ApplicationRouteStream } from '../../source/services/route-stream.ts'; import type { Database } from '../../source/services/storage/database.ts'; import { TestConnection } from '../helpers/test-connection.ts'; import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../../source/constants.ts'; +import { createMockAuth } from '../helpers/misc.ts'; +import { AuthSecp256k1 } from '../../source/auth/auth.ts'; const createBroadcasterStub = (): BaseBroadcaster => { return { @@ -33,8 +35,14 @@ describe('DataRoute subscriptions', (): void => { const connection = new TestConnection(true, false); const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a', 'b' ], + }, 'subscribe-1'); + const route = new DataRoute({ + auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }), + database: storage, + broadcaster: broadcaster, + }, { + timestampWindowMs: 0, }); - const route = new DataRoute(storage, broadcaster, 0); const execution = route.subscribeData(stream); @@ -55,8 +63,14 @@ describe('DataRoute subscriptions', (): void => { } 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); + const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a' ] }, '/unsubscribe', 'unsubscribe-1'); + const route = new DataRoute({ + auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }), + database: storage, + broadcaster: broadcaster, + }, { + timestampWindowMs: 0, + }); await route.unsubscribeData(stream); @@ -78,10 +92,17 @@ describe('DataRoute subscriptions', (): void => { }, } as unknown as Database; const broadcaster = createBroadcasterStub(); - const stream = new ApplicationRouteStream(new TestConnection(true, false), { + const connection = new TestConnection(true, false); + const stream = new ApplicationRouteStream(connection, { resourceId: [ 'a' ], + }, '/unsubscribe', 'unsubscribe-1'); + const route = new DataRoute({ + auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }), + database: storage, + broadcaster: broadcaster, + }, { + timestampWindowMs: 0, }); - 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(); @@ -97,7 +118,13 @@ describe('DataRoute resource write auth', (): void => { } as unknown as Database; const broadcaster = createBroadcasterStub(); - const route = new DataRoute(storage, broadcaster, 0); + const route = new DataRoute({ + auth: await AuthSecp256k1.create({ database: storage }, { timestampWindowMs: 0 }), + database: storage, + broadcaster: broadcaster, + }, { + timestampWindowMs: 0, + }); await expect(route.writeData({ connection: new TestConnection(true, true), diff --git a/test/services/broadcaster.test.ts b/test/services/broadcaster.test.ts index 3acc018..07380e9 100644 --- a/test/services/broadcaster.test.ts +++ b/test/services/broadcaster.test.ts @@ -10,7 +10,7 @@ const createBroadcaster = (): Broadcaster => { }; const routeStream = (connection: TestConnection): ApplicationRouteStream => { - return new ApplicationRouteStream(connection, undefined); + return new ApplicationRouteStream(connection, undefined, '/items', 'items-1'); }; const expectPending = async (promise: Promise): Promise => { diff --git a/test/services/router.test.ts b/test/services/router.test.ts index 318170b..ca8cdef 100644 --- a/test/services/router.test.ts +++ b/test/services/router.test.ts @@ -1,52 +1,27 @@ -import { describe, expect, it, vi } from 'vitest'; +import { describe, expect, it } from 'vitest'; -import type { RouteDefinition, RouteModule } from '../../source/routes/types.ts'; +// Source import { ApplicationRouter } from '../../source/services/router.ts'; + +// Helpers +import { createMockAuth, toRoutes } from '../helpers/misc.ts'; import { TestConnection } from '../helpers/test-connection.ts'; - -import type { AuthSecp256k1 } from '../../source/auth/auth.ts'; - -/** - * A controlled request is a request that is controlled by the test. - * It is used to control the request flow and ensure that the request is completed in the correct order. - */ -type ControlledRequest = { - request: Promise; - started: Promise; - release: () => void; -}; +import { createControlledRequest } from '../helpers/controlled-request.ts'; /** * A mock of the AuthSecp256k1 service */ -const auth = { - verifySignature: vi.fn().mockResolvedValue(true), - verifyUniqueRequest: vi.fn().mockResolvedValue(true), - assertTimestampFreshness: vi.fn().mockResolvedValue(true), -} as unknown as AuthSecp256k1; - -/** - * A helper function to create a route module with the given routes - * @param routes - The routes to create the module with - * @returns The created route module - */ -const moduleWith = (routes: RouteDefinition[]): RouteModule => { - return { - async getRoutes(): Promise { - return routes; - }, - }; -}; +const auth = createMockAuth(); describe('ApplicationRouter initialization', (): void => { it('rejects duplicate exact paths during startup', async (): Promise => { const route = { url: '/echo', handler: (): void => undefined }; - await expect(ApplicationRouter.create({ auth }, [ moduleWith([ route ]), moduleWith([ route ]) ])).rejects.toThrow('Duplicate application route: /echo'); + await expect(ApplicationRouter.create({ auth }, [ toRoutes([ route ]), toRoutes([ route ]) ])).rejects.toThrow('Duplicate application route: /echo'); }); it.each([ 'echo', '/', '/items/:id', '/items?active=true', '/items#active' ])('rejects the invalid route path %s', async (url): Promise => { - await expect(ApplicationRouter.create({ auth }, [ moduleWith([{ url, handler: (): void => undefined }]) ])).rejects.toThrow('Invalid application route'); + await expect(ApplicationRouter.create({ auth }, [ toRoutes([{ url, handler: (): void => undefined }]) ])).rejects.toThrow('Invalid application route'); }); }); @@ -54,7 +29,7 @@ describe('ApplicationRouter dispatch', (): void => { it('binds the connection, body, and request ID to one route stream', async (): Promise => { const connection = new TestConnection(false, false); const router = await ApplicationRouter.create({ auth }, [ - moduleWith([ + toRoutes([ { url: '/echo', handler: async (stream): Promise => { @@ -94,7 +69,7 @@ describe('ApplicationRouter dispatch', (): void => { it('preserves correlation when concurrent requests finish out of order', async (): Promise => { const router = await ApplicationRouter.create({ auth }, [ - moduleWith([ + toRoutes([ { url: '/delayed', handler: async (stream): Promise => { @@ -115,38 +90,8 @@ describe('ApplicationRouter dispatch', (): void => { const connection = new TestConnection(false, false); - const createControlledRequest = (key: string): ControlledRequest => { - const { promise: started, resolve: signalStarted } = Promise.withResolvers(); - - const { promise: released, resolve: release } = Promise.withResolvers(); - - const request = router.dispatch( - { - path: '/delayed', - body: { - key, - signalStarted, - released, - }, - requestId: key, - headers: { - 'x-public-key': 'public-key', - 'x-signature': 'signature', - 'x-timestamp': '1000', - }, - }, - connection, - ); - - return { - request, - started, - release, - }; - }; - - const first = createControlledRequest('A'); - const second = createControlledRequest('B'); + const first = createControlledRequest({ router, connection, path: '/delayed', requestId: 'A', body: { key: 'A' } }); + const second = createControlledRequest({ router, connection, path: '/delayed', requestId: 'B', body: { key: 'B' } }); expect(connection.messages).toEqual([]); @@ -181,12 +126,12 @@ describe('ApplicationRouter dispatch', (): void => { body: { key: 'A' }, }, ]); - }); + }, 1000); it('propagates route failures without infrastructure-specific cleanup', async (): Promise => { const error = new Error('route failed'); const router = await ApplicationRouter.create({ auth }, [ - moduleWith([ + toRoutes([ { url: '/failure', handler: (): void => { diff --git a/test/services/transports/http-transport.test.ts b/test/services/transports/http-transport.test.ts index 095d266..f2c4976 100644 --- a/test/services/transports/http-transport.test.ts +++ b/test/services/transports/http-transport.test.ts @@ -10,6 +10,22 @@ import type { AppEnv } from '../../../source/services/transport/transport-router import { fromExtendedJson, toExtendedJson } from '@xo-cash/utils'; import { Logger } from '../../../source/utils/logger.ts'; import { ServerHost } from '../../../source/services/server-host.ts'; +import { AuthSecp256k1 } from '../../../source/auth/auth.ts'; + +/** + * A mock of the AuthSecp256k1 service + */ +const auth = { + verifySignature: vi.fn().mockResolvedValue(true), + verifyUniqueRequest: vi.fn().mockResolvedValue(true), + assertTimestampFreshness: vi.fn().mockResolvedValue(true), +} as unknown as AuthSecp256k1; + +const mockAuthHeaders = { + 'x-public-key': 'public-key', + 'x-signature': 'signature', + 'x-timestamp': '1000', +}; const createApp = async ( routes: RouteDefinition[] | ((broadcaster: Broadcaster) => RouteDefinition[]), @@ -18,7 +34,9 @@ const createApp = async ( const debug = new Logger('http-transport-test'); const broadcaster = new Broadcaster(debug); const resolvedRoutes = typeof routes === 'function' ? routes(broadcaster) : routes; - const router = await ApplicationRouter.create([ + const router = await ApplicationRouter.create({ + auth, + }, [ { async getRoutes(): Promise { return resolvedRoutes; @@ -40,18 +58,20 @@ describe('HttpTransportRouter', (): void => { it('runs normal HTTP through a non-streaming route stream', async (): Promise => { const app = await createApp([ { - url: '/echo', + url: '/echowtf', handler: async (stream): Promise => stream.send(stream.body), }, ]); const value = new Uint8Array([ 1, 2, 3 ]); - const response = await app.request('/echo', { + const request = new Request('http://localhost/echowtf', { method: 'POST', - headers: { 'content-type': 'application/json' }, + headers: { 'content-type': 'application/json', ...mockAuthHeaders }, body: toExtendedJson({ value }), }); + const response = await app.request(request); + expect(response.status).toBe(200); expect(fromExtendedJson(await response.text())).toEqual({ value }); }); @@ -64,7 +84,7 @@ describe('HttpTransportRouter', (): void => { }, ]); - const response = await app.request('/nothing', { method: 'POST' }); + const response = await app.request('/nothing', { method: 'POST', headers: { ...mockAuthHeaders } }); expect(response.status).toBe(204); expect(await response.text()).toBe(''); @@ -73,7 +93,7 @@ describe('HttpTransportRouter', (): void => { it('returns normalized errors for non-streaming requests', async (): Promise => { const app = await createApp([]); - const missing = await app.request('/missing', { method: 'POST' }); + const missing = await app.request('/missing', { method: 'POST', headers: { ...mockAuthHeaders } }); expect(missing.status).toBe(404); expect(await missing.json()).toEqual({ statusCode: 404, @@ -82,7 +102,7 @@ describe('HttpTransportRouter', (): void => { const invalid = await app.request('/missing', { method: 'POST', - headers: { 'content-type': 'application/json' }, + headers: { 'content-type': 'application/json', ...mockAuthHeaders }, body: '{', }); expect(invalid.status).toBe(400); @@ -102,7 +122,7 @@ describe('HttpTransportRouter', (): void => { }, ]); - const response = await app.request('/items/subscribe', { method: 'POST' }); + const response = await app.request('/items/subscribe', { method: 'POST', headers: { ...mockAuthHeaders } }); expect(response.status).toBe(406); expect(await response.json()).toMatchObject({ statusCode: 406 }); @@ -120,7 +140,7 @@ describe('HttpTransportRouter', (): void => { const response = await app.request('/items/subscribe', { method: 'POST', - headers: { accept: 'text/event-stream' }, + headers: { accept: 'text/event-stream', ...mockAuthHeaders }, }); const events = await response.text(); @@ -140,7 +160,7 @@ describe('HttpTransportRouter', (): void => { const response = await app.request('/echo', { method: 'POST', - headers: { accept: 'text/event-stream' }, + headers: { accept: 'text/event-stream', ...mockAuthHeaders }, }); const events = await response.text(); @@ -171,7 +191,7 @@ describe('HttpTransportRouter', (): void => { const response = await app.request('/items/subscribe', { method: 'POST', - headers: { accept: 'text/event-stream' }, + headers: { accept: 'text/event-stream', ...mockAuthHeaders }, }); const body = response.text(); const completed = vi.fn(); @@ -203,7 +223,7 @@ describe('HttpTransportRouter', (): void => { const response = await app.request('/items/unsubscribe', { method: 'POST', - headers: { accept: 'text/event-stream' }, + headers: { accept: 'text/event-stream', ...mockAuthHeaders }, }); const events = await response.text(); @@ -224,7 +244,7 @@ describe('HttpTransportRouter', (): void => { const response = await app.request('/echo', { method: 'POST', - headers: { 'content-type': 'application/json' }, + headers: { 'content-type': 'application/json', ...mockAuthHeaders }, body: JSON.stringify({ value: 'x'.repeat(64) }), }); diff --git a/test/services/transports/ws-transport.test.ts b/test/services/transports/ws-transport.test.ts index 662316f..2e575ad 100644 --- a/test/services/transports/ws-transport.test.ts +++ b/test/services/transports/ws-transport.test.ts @@ -1,11 +1,21 @@ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { z } from 'zod'; -import type { WebSocketServer } from 'ws'; import { ApplicationRouter } from '../../../source/services/router.ts'; import { WsTransportRouter } from '../../../source/services/transport/ws-transport.ts'; import { Logger } from '../../../source/utils/logger.ts'; import { toExtendedJson } from '@xo-cash/utils'; +import { AuthSecp256k1 } from '../../../source/auth/auth.ts'; +import { toRoutes } from '../../helpers/misc.ts'; + +/** + * A mock of the AuthSecp256k1 service + */ +const auth = { + verifySignature: vi.fn().mockResolvedValue(true), + verifyUniqueRequest: vi.fn().mockResolvedValue(true), + assertTimestampFreshness: vi.fn().mockResolvedValue(true), +} as unknown as AuthSecp256k1; describe('WebSocket request decoding', (): void => { it('decodes the minimal route-agnostic envelope and Extended JSON body', async (): Promise => { @@ -38,11 +48,21 @@ describe('WebSocket request decoding', (): void => { describe('WsTransportRouter payload limits', (): void => { it("configures Hono's ws server with the requested maxPayload", async () => { const debug = new Logger('ws-transport-test'); - const router = await ApplicationRouter.create([]); + const router = await ApplicationRouter.create({ + auth, + }, [ + toRoutes([ + { + url: '/data/write', + handler: async (stream): Promise => { + await stream.send({}); + }, + }, + ]), + ]); const transport = new WsTransportRouter(router, debug, 1024); - const wsServer = transport.websocketServer as unknown as WebSocketServer; - expect(wsServer.options.maxPayload).toBe(1024); + expect(transport['wsServer'].options.maxPayload).toBe(1024); await transport.stop(); }); }); diff --git a/test/subscription-flow.test.ts b/test/subscription-flow.test.ts index d585d11..28df429 100644 --- a/test/subscription-flow.test.ts +++ b/test/subscription-flow.test.ts @@ -1,35 +1,38 @@ import { describe, expect, it, vi } from 'vitest'; -import type { RouteDefinition, RouteModule } from '../source/routes/types.ts'; import { ApplicationRouter } from '../source/services/router.ts'; import { Broadcaster } from '../source/services/broadcaster.ts'; import { Logger } from '../source/utils/logger.ts'; import { TestConnection } from './helpers/test-connection.ts'; -const moduleWith = (routes: RouteDefinition[]): RouteModule => { - return { - async getRoutes(): Promise { - return routes; - }, - }; -}; +import { createControlledRequest } from './helpers/controlled-request.ts'; +import { createMockAuth, toRoutes } from './helpers/misc.ts'; -const expectPending = async (promise: Promise): Promise => { - const settled = vi.fn(); - void promise.then(settled); - await Promise.resolve(); - expect(settled).not.toHaveBeenCalled(); -}; +/** + * A mock of the AuthSecp256k1 service + */ +const auth = createMockAuth(); describe('long-lived subscription dispatch', (): void => { it('allows duplicate subscribe and unsubscribe requests while the original request waits', async (): Promise => { const broadcaster = new Broadcaster(new Logger('subscription-flow-test')); - const router = await ApplicationRouter.create([ - moduleWith([ + const router = await ApplicationRouter.create({ + auth, + }, + [ + toRoutes([ { url: '/items/subscribe', handler: async (stream): Promise => { + const { signalStarted, released } = stream.body as { + signalStarted: () => void; + released: Promise; + }; + + signalStarted(); + await broadcaster.subscribe(stream, [ 'items' ]); + await released; }, }, { @@ -43,17 +46,24 @@ describe('long-lived subscription dispatch', (): void => { ]); const connection = new TestConnection(true, true); - const original = router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-1' }, connection); + const original = createControlledRequest({ router, connection, path: '/items/subscribe', requestId: 'subscribe-1' }); await vi.waitFor(() => expect(connection.closeCallbacks).toHaveLength(1)); - await expectPending(original); + await original.started; // This request uses a different ApplicationRouteStream over the same // connection. Since the topic already exists, its dispatch completes. - await router.dispatch({ path: '/items/subscribe', requestId: 'subscribe-2' }, connection); - await expectPending(original); + const second = createControlledRequest({ router, connection, path: '/items/subscribe', requestId: 'subscribe-2' }); + second.release(); + await second.request; - await router.dispatch({ path: '/items/unsubscribe', requestId: 'unsubscribe-1' }, connection); - await original; + // Unsubscribe the original request stream + const third = createControlledRequest({ router, connection, path: '/items/unsubscribe', requestId: 'unsubscribe-1' }); + third.release(); + await third.request; + + // Release the original request stream + original.release(); + await original.request; expect(connection.messages).toEqual([ {