From 1ca9648c095d373fa4b08cec395ff139fc96f40e Mon Sep 17 00:00:00 2001 From: Harvmaster Date: Mon, 31 Aug 2026 12:28:18 +0000 Subject: [PATCH] Added auth and request storage --- package.json | 2 +- source/auth/accounts.ts | 49 +++++++ source/auth/auth.ts | 110 ++++++++++++++++ source/constants.ts | 5 + source/index.ts | 28 +++- source/routes/account.ts | 105 +++++++++++++++ source/routes/resources.ts | 122 ++++++------------ source/routes/types.ts | 7 + source/services/config.ts | 8 +- source/services/route-stream.ts | 15 ++- source/services/router.ts | 58 ++++++++- .../storage/migrations/001-resources.ts | 12 ++ source/services/storage/tables.ts | 35 +++++ source/services/transport/http-transport.ts | 2 + source/services/transport/request-headers.ts | 38 ++++++ source/services/transport/ws-transport.ts | 3 + test/routes/resources.test.ts | 62 +++++++-- test/services/broadcaster.test.ts | 2 +- test/services/router.test.ts | 21 ++- .../transports/http-transport.test.ts | 38 ++++++ test/services/transports/ws-transport.test.ts | 29 ++++- 21 files changed, 634 insertions(+), 117 deletions(-) create mode 100644 source/auth/accounts.ts create mode 100644 source/auth/auth.ts create mode 100644 source/routes/account.ts create mode 100644 source/services/transport/request-headers.ts diff --git a/package.json b/package.json index 23433e5..233b721 100644 --- a/package.json +++ b/package.json @@ -17,7 +17,7 @@ "scripts": { "analyze": "tsdown --clean --no-fixed-extension --sourcemap source/index.ts && esbuild-analyzer dist/", "build": "tsdown --clean --sourcemap source/index.ts", - "dev": "tsx watch source/app.ts", + "dev": "tsx watch source/index.ts", "docs": "typedoc --hideGenerator --categorizeByGroup", "format": "prettier --write . && eslint --fix", "spellcheck": "cspell 'source/**' 'test/**' 'playground/**'", diff --git a/source/auth/accounts.ts b/source/auth/accounts.ts new file mode 100644 index 0000000..987e223 --- /dev/null +++ b/source/auth/accounts.ts @@ -0,0 +1,49 @@ +/** + * This is a shim for payments that isnt really going to be too reflective of the real world payment system. + * + * The reason im doing it in such an overly simplified way is because an actual payment system is COMPLEX. + * + * Double entry accounting, transaction objects, idempotent requests, etc... They are a LOT to implement. + * + * Im certain that we can implement this more complex system into this project, its just going to be a lot of work on the actual payment handling. + * The route side can actually remain pretty simple because it can be wrapped into a function call. + * + * This implementation will just offer a `getBalance(publicKey: string): Promise` and `setBalance(publicKey: string, amount: number): Promise` + * + * Routes will just do a setBalance(pulicKey, await getBalance(publicKey) - amount) where amount is the size of the data being written. + */ +export class Accounts { + private accounts: Map = new Map(); + + async getBalance(publicKey: string): Promise { + if (!this.accounts.has(publicKey)) { + this.accounts.set(publicKey, 1_000_000_000_000); + } + + return this.accounts.get(publicKey)!; + } + + async setBalance(publicKey: string, amount: number): Promise { + this.accounts.set(publicKey, amount); + + return amount; + } + + async deductBalance(publicKey: string, amount: number): Promise { + const balance = await this.getBalance(publicKey); + + if (balance < amount) { + throw new Error('Insufficient balance'); + } + + const balanceAfterDeduction = balance - amount; + + return await this.setBalance(publicKey, balanceAfterDeduction); + } + + async hasSufficientBalance(publicKey: string, amount: number): Promise { + return true; + const balance = await this.getBalance(publicKey); + return balance >= amount; + } +} diff --git a/source/auth/auth.ts b/source/auth/auth.ts new file mode 100644 index 0000000..abb944f --- /dev/null +++ b/source/auth/auth.ts @@ -0,0 +1,110 @@ +import type { Database } from '../services/storage/database'; + +import { hexToBin, instantiateSecp256k1, type Secp256k1, sha256 } from '@bitauth/libauth'; +import { UnauthorizedError } from '../errors/unauthorized-error'; + +export type AuthSecp256k1RequiredDeps = { + database: Database; +} + +export type AuthSecp256k1OptionalDeps = { + secp256k1: Secp256k1; +} +export type AuthSecp256k1Deps = AuthSecp256k1RequiredDeps & Partial; + +export type AuthSecp256k1Options = { + timestampWindowMs: number; +} + +export class AuthSecp256k1 { + /** + * Create a new instance of AuthSecp256k1 + * @returns A new instance of AuthSecp256k1 + */ + static async create(inputDeps: AuthSecp256k1Deps, options: AuthSecp256k1Options): Promise { + const deps = { + secp256k1: await instantiateSecp256k1(), + ...inputDeps, + } + + return new AuthSecp256k1(deps, options); + } + + /** + * TODO: Consider adding a Record where each key is the signature to guarantee signatures arent being processed concurrently. + */ + + /** + * @param deps - The dependencies to use + * @param options - The options to use + */ + private constructor(private readonly deps: Required, private readonly options: AuthSecp256k1Options) {} + + /** + * Verify a 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 + */ + async verifySignature(publicKeyHex: string, signatureHex: string, payload: string): Promise { + return true; + // Convert the public key and signature to binary + const publicKey = hexToBin(publicKeyHex); + const signature = hexToBin(signatureHex); + + // Convert the payload to bytes and compute the sha256 hash + const payloadBytes = new TextEncoder().encode(payload); + const messageHash = sha256.hash(payloadBytes); + + // Verify the signature + const verified = await this.deps.secp256k1.verifySignatureDERLowS(signature, publicKey, messageHash); + + // If the signature is not valid, throw an unauthorized error + if (!verified) { + throw new UnauthorizedError('Invalid signature'); + } + + // Return the verified signature + return verified; + } + + async verifyUniqueRequest(signature: string): Promise { + if (!signature) { + throw new UnauthorizedError('Signature is required'); + } + + console.log('Verifying unique request', signature); + + // Check if the signature has been used before + const request = await this.deps.database.db. + selectFrom('authed_requests') + .selectAll() + .where('signature', '=', signature) + .executeTakeFirst(); + + if (request) { + throw new UnauthorizedError('Request already used'); + } + + // Add the signature to the requests table + await this.deps.database.db.insertInto('authed_requests').values({ signature }).execute(); + + // Return true if the signature is valid + return true; + } + + /** + * Assert that a timestamp is within a allowed window + * @param timestamp - The timestamp to check + * @param windowMs - The window in milliseconds + * + * @throws An {@link UnauthorizedError} if the timestamp is outside the allowed window + */ + assertTimestampFreshness(timestamp: number, windowMs = this.options.timestampWindowMs): void { + const age = Math.abs(Date.now() - timestamp); + if (age > windowMs) { + throw new UnauthorizedError('Timestamp outside allowed window'); + } + } +} \ No newline at end of file diff --git a/source/constants.ts b/source/constants.ts index 37b4606..fa4f8cc 100644 --- a/source/constants.ts +++ b/source/constants.ts @@ -22,3 +22,8 @@ export const HTTP_STATUS_CODE_BAD_REQUEST = 400; * HTTP status code for "Not Acceptable" error. */ export const HTTP_STATUS_CODE_NOT_ACCEPTED = 406; + +/** + * Not implemented response status code. + */ +export const HTTP_STATUS_CODE_NOT_IMPLEMENTED = 501; diff --git a/source/index.ts b/source/index.ts index 2ff7ba8..47d0165 100644 --- a/source/index.ts +++ b/source/index.ts @@ -1,12 +1,16 @@ import { Config } from './services/config.ts'; import { Database, MigrationService } from './services/storage/index.ts'; +import { AuthSecp256k1 } from './auth/auth.ts'; import { Broadcaster } from './services/broadcaster.ts'; import { ApplicationRouter } from './services/router.ts'; 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'; +import { AccountRoute } from './routes/account.ts'; +import { Accounts } from './auth/accounts.ts'; /** Application composition root. */ export class App { @@ -23,18 +27,23 @@ export class App { const migrations = new MigrationService(database, debug); await migrations.migrateToLatest(); + // Create an Auth instance that can be passed in for signature validation + const auth = await AuthSecp256k1.create({ database }, { timestampWindowMs: config.auth.timestampWindowMs }); + // Domain services are shared across all transports and route modules. const broadcaster = new Broadcaster(debug); + const accounts = new Accounts(); 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(database, broadcaster, auth, accounts), + // new AccountRoute(database, auth), ]; // 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); + const router = await ApplicationRouter.create({ auth }, routes); // Both transports share one ApplicationRouter. Routes use the shared // Broadcaster directly, while HTTP and WebSocket remain protocol adapters. @@ -43,7 +52,13 @@ export class App { const ws = new WsTransportRouter(router, debug, config.server.maxRequestBodyBytes); const host = new ServerHost(config, debug, [ http, ws ]); - return new App(host, database); + // Create the app instance + const app = new App(host, database); + + // Start the unique request cleanup interval + app.startUniqueRequestCleanup(config.auth.uniqueRequestCleanupIntervalMs, config.auth.timestampWindowMs); + + return app; } private stopPromise: Promise | undefined; @@ -69,6 +84,13 @@ export class App { })(); await this.stopPromise; } + + startUniqueRequestCleanup(cleanupIntervalMs: number, timestampWindowMs: number): void { + // Every 10 seconds, we will cleanup the requests table + setInterval(async () => { + await this.database.db.deleteFrom('authed_requests').where('timestamp', '<', Date.now() - timestampWindowMs).execute(); + }, cleanupIntervalMs); + } } const app = await App.create(); diff --git a/source/routes/account.ts b/source/routes/account.ts new file mode 100644 index 0000000..a29ca39 --- /dev/null +++ b/source/routes/account.ts @@ -0,0 +1,105 @@ +import { z } from "zod"; + +import type { AuthSecp256k1 } from '../auth/auth.ts'; +import type { Database } from "../services/storage/database.ts"; +import type { RouteDefinition, RouteStream } from './types.ts'; +import type { Accounts } from '../auth/accounts.ts'; + +import { HTTP_STATUS_CODE_NOT_IMPLEMENTED, HTTP_STATUS_CODE_SUCCESS } from "../constants.ts"; +import { UnauthorizedError } from '../errors/index.ts'; + +const accountSchema = z.object({ + publicKey: z.string(), + signature: z.string(), + timestamp: z.coerce.number(), +}); + +const setBalanceSchema = z.object({ + amount: z.number(), +}); + +export class AccountRoute { + + constructor(private readonly database: Database, private readonly auth: AuthSecp256k1, private readonly accounts: Accounts) {} + + async getRoutes(): Promise> { + return [ + { + url: '/account', + handler: this.getAccount.bind(this), + }, + { + url: '/account/balance', + handler: this.getAccount.bind(this), + }, + { + url: '/account/setbalance', + handler: this.setBalance.bind(this), + }, + // This one may not make sense. It could be a large overhead for something that most wont use? + // Maybe make this an optional endpoint or one that we don't support, but the client can try to hit or something? + { + url: '/account/ledger', + handler: this.getLedger.bind(this), + } + ]; + } + + async getAccount(stream: RouteStream): Promise { + // Get the public key, signature and timestamp from the headers + const { publicKey, signature, timestamp } = accountSchema.parse(stream.headers); + + // Create the canonical payload for the signature verification + const payload = `${stream.path}:${timestamp}`; + + // Verify the signature + const verified = await this.auth.verifySignature(publicKey, signature, payload); + + if (!verified) { + throw new UnauthorizedError('Invalid signature'); + } + + // Get the account's balance from the database + const balance = await this.accounts.getBalance(publicKey); + + // Send the balance to the client + stream.send({ + statusCode: HTTP_STATUS_CODE_SUCCESS, + body: balance, + }); + } + + async setBalance(stream: RouteStream): Promise { + const { publicKey, signature, timestamp } = accountSchema.parse(stream.headers); + const { amount } = setBalanceSchema.parse(stream.body); + + // Create the canonical payload for the signature verification + const payload = `${stream.path}:${timestamp}`; + + // Verify the signature + const verified = await this.auth.verifySignature(publicKey, signature, payload); + + if (!verified) { + throw new UnauthorizedError('Invalid signature'); + } + + // Get the account's balance from the database + const balance = await this.accounts.getBalance(publicKey); + + // Set the balance to the database + await this.accounts.setBalance(publicKey, amount); + + // Send the balance to the client + stream.send({ + statusCode: HTTP_STATUS_CODE_SUCCESS, + body: balance, + }); + } + + async getLedger(stream: RouteStream): Promise { + stream.send({ + statusCode: HTTP_STATUS_CODE_NOT_IMPLEMENTED, + body: 'Not implemented', + }); + } +} \ No newline at end of file diff --git a/source/routes/resources.ts b/source/routes/resources.ts index ac7a250..d070d6d 100644 --- a/source/routes/resources.ts +++ b/source/routes/resources.ts @@ -1,14 +1,16 @@ -import { createHash } from 'node:crypto'; -import { hexToBin, instantiateSecp256k1, type Secp256k1 } from '@bitauth/libauth'; + +// NOTE: Replace this with libauth sha256 import { toExtendedJson } from '@xo-cash/utils'; import { z } from 'zod'; import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.ts'; +import type{ AuthSecp256k1 } from '../auth/auth.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'; +import type { Accounts } from '../auth/accounts.ts'; /** * Schema to validate a single write resource. @@ -56,16 +58,11 @@ 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 auth: AuthSecp256k1, + private readonly accounts: Accounts, ) {} /** Declare exact routes; each handler owns its stream behavior. */ @@ -92,7 +89,7 @@ export class DataRoute implements RouteModule { /** Reads the requested resources from the database, returning all instances for each resource */ async getData(stream: RouteStream): Promise { - const resourceIds = this.getResourceIds(stream); + const resourceIds = resourceIdsSchema.parse(stream.body).resourceId; // Remove duplicates. const uniqueIds = [ ...new Set(resourceIds) ]; @@ -107,7 +104,7 @@ export class DataRoute implements RouteModule { // Read the data from the database. const rows = await this.database.db .selectFrom('resource_data') - .select([ 'resource_id', 'public_key', 'blob', 'timestamp' ]) + .select([ 'resource_id', 'public_key', 'blob', 'timestamp', 'signature' ]) .where('resource_id', 'in', uniqueIds) .orderBy('timestamp', 'asc') .execute(); @@ -118,6 +115,7 @@ export class DataRoute implements RouteModule { publicKey: row.public_key, blob: new Uint8Array(row.blob), timestamp: row.timestamp, + signature: row.signature, })); await stream.send(formattedRows); @@ -132,6 +130,10 @@ export class DataRoute implements RouteModule { async writeData(stream: RouteStream): Promise { const { resources } = writeBody.parse(stream.body); + // temporarily disable the signature verification for testing + // const publicKey = stream.headers?.['x-public-key'] + const publicKey = 'public-key'; + // 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))); @@ -141,11 +143,22 @@ export class DataRoute implements RouteModule { resourceId: resource.id, publicKey: resource.publicKey, blob: Buffer.from(resource.value), + signature: resource.signature, })); + // Get the total size of the bytes being written + const totalSize = rows.reduce((acc, row) => acc + row.blob.length, 0); + + if (!await this.accounts.hasSufficientBalance(publicKey, totalSize)) { + throw new ApplicationError(203, 'Insufficient balance'); + } + + // Set the balance of the public key + await this.accounts.deductBalance(publicKey, totalSize); + // 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) { + for (const { resourceId, publicKey, blob, signature } of rows) { await trx .insertInto('resource_data') .values({ @@ -153,6 +166,7 @@ export class DataRoute implements RouteModule { public_key: publicKey, blob, timestamp, + signature, }) .onConflict((oc) => oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({ @@ -164,36 +178,38 @@ export class DataRoute implements RouteModule { }); // Format the rows into the written resource responses. - const written = rows.map(({ resourceId, publicKey, blob }) => ({ - resourceId, + const written = rows.map(({ resourceId, publicKey, blob, signature }) => ({ instance: { + resourceId, publicKey, blob: new Uint8Array(blob), timestamp, + signature, }, })); // 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), { + for (const { instance } of written) { + await this.broadcaster.publish(DataRoute.resourceTopic(instance.resourceId), { type: 'instance-changed', - data: { resourceId, ...instance }, + data: instance, }); } // Return the persisted instances so the writer can confirm what was stored. await stream.send({ - resources: written.map(({ resourceId, instance }) => ({ - id: resourceId, + resources: written.map(({ instance }) => ({ + id: instance.resourceId, ...instance, })), + balance: await this.accounts.getBalance(publicKey), }); } /** Subscribe this connection to future changes for the requested resources. */ async subscribeData(stream: RouteStream): Promise { - const resourceIds = this.getResourceIds(stream); + const resourceIds = resourceIdsSchema.parse(stream.body).resourceId; const topics = resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)); @@ -209,7 +225,7 @@ export class DataRoute implements RouteModule { * unsubscribe implicitly by aborting the HTTP request. */ async unsubscribeData(stream: RouteStream): Promise { - const resourceIds = this.getResourceIds(stream); + const resourceIds = resourceIdsSchema.parse(stream.body).resourceId; // If the stream is not bidirectional, throw an error. if (!stream.bidirectional) { @@ -224,14 +240,6 @@ export class DataRoute implements RouteModule { 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. * @@ -239,62 +247,16 @@ export class DataRoute implements RouteModule { * id, and value — not the raw HTTP/WebSocket envelope. */ private async verifyWriteResource(resource: WriteResource): Promise { - this.assertFreshTimestamp(resource.timestamp); + this.auth.assertTimestampFreshness(resource.timestamp); - if (!(await this.verifySignature(resource.publicKey, resource.signature, DataRoute.canonicalWritePayload(resource)))) { + // Compile the signature payload as `Timestamp:ID:Value` + const signaturePayload = `${resource.timestamp}:${resource.id}:${toExtendedJson(resource.value)}`; + + if (!(await this.auth.verifySignature(resource.publicKey, resource.signature, signaturePayload))) { 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/source/routes/types.ts b/source/routes/types.ts index 2d2746e..8888ab1 100644 --- a/source/routes/types.ts +++ b/source/routes/types.ts @@ -1,5 +1,8 @@ import type { BaseStream } from '../services/stream/base-stream.js'; +/** Canonical request headers exposed to transport-neutral route handlers. */ +export type RequestHeaders = Readonly>; + export type RouteSendOptions = { /** Defaults to `response`; any other value sends an application event. */ @@ -20,7 +23,11 @@ export interface RouteStream { /** Connection shared by every request on the same transport session. */ readonly connection: BaseStream; + /** Canonical route path selected for this request. */ + readonly path: string; + readonly body: unknown; + readonly headers: RequestHeaders; readonly streaming: boolean; readonly bidirectional: boolean; diff --git a/source/services/config.ts b/source/services/config.ts index 19bdd54..d13c079 100644 --- a/source/services/config.ts +++ b/source/services/config.ts @@ -32,7 +32,7 @@ const configSchema = z.object({ .object({ origin: z.string().default('*'), methods: z.array(z.string()).default([ 'GET', 'POST', 'PUT', 'DELETE', 'OPTIONS' ]), - allowedHeaders: z.array(z.string()).default([ 'Content-Type', 'cache-control', 'X-Timestamp', 'X-PublicKey', 'X-Signature' ]), + allowedHeaders: z.array(z.string()).default([ 'Content-Type', 'cache-control', 'X-Timestamp', 'X-Public-Key', 'X-Signature' ]), }) .partial() .prefault({}), @@ -48,6 +48,11 @@ const configSchema = z.object({ .int() .positive() .default(5 * 60 * 1000), + uniqueRequestCleanupIntervalMs: z.coerce + .number() + .int() + .positive() + .default(10 * 60 * 1000), }) .prefault({}), }); @@ -83,6 +88,7 @@ export class Config { }, auth: { timestampWindowMs: process.env.AUTH_TIMESTAMP_WINDOW_MS ? Number(process.env.AUTH_TIMESTAMP_WINDOW_MS) : undefined, + uniqueRequestCleanupIntervalMs: process.env.AUTH_UNIQUE_REQUEST_CLEANUP_INTERVAL_MS ? Number(process.env.AUTH_UNIQUE_REQUEST_CLEANUP_INTERVAL_MS) : undefined, }, }); } diff --git a/source/services/route-stream.ts b/source/services/route-stream.ts index e3b1b6c..156133b 100644 --- a/source/services/route-stream.ts +++ b/source/services/route-stream.ts @@ -1,7 +1,10 @@ -import type { RouteSendOptions, RouteStream } from '../routes/types.ts'; +import type { RequestHeaders, RouteSendOptions, RouteStream } from '../routes/types.ts'; import type { BaseStream } from './stream/base-stream.ts'; import { HTTP_STATUS_CODE_NO_CONTENT, HTTP_STATUS_CODE_SUCCESS } from '../constants.ts'; +/** Shared immutable value used when a request supplies no headers. */ +const EMPTY_REQUEST_HEADERS: RequestHeaders = Object.freeze({}); + /** * Binds one application request to a connection-level stream. * @@ -10,16 +13,24 @@ import { HTTP_STATUS_CODE_NO_CONTENT, HTTP_STATUS_CODE_SUCCESS } from '../consta * connection-level services such as the broadcaster. */ export class ApplicationRouteStream implements RouteStream { + readonly headers: RequestHeaders; + /** * @param connection - Shared transport stream backing this request. * @param body - Transport-decoded application payload for the route handler. + * @param path - Canonical application route selected for this request. * @param requestId - Optional correlation ID for multiplexed transports. + * @param headers - Transport-normalized request headers for this dispatch. */ constructor( readonly connection: BaseStream, readonly body: unknown, + readonly path: string, private readonly requestId?: string, - ) {} + headers: RequestHeaders = EMPTY_REQUEST_HEADERS, + ) { + this.headers = headers === EMPTY_REQUEST_HEADERS ? headers : Object.freeze({ ...headers }); + } /** Whether the underlying connection can deliver server-pushed events. */ get streaming(): boolean { diff --git a/source/services/router.ts b/source/services/router.ts index 71c4d50..3ce81bc 100644 --- a/source/services/router.ts +++ b/source/services/router.ts @@ -1,7 +1,11 @@ -import type { RouteDefinition, RouteModule } from '../routes/types.ts'; -import { ApplicationError } from '../errors/index.ts'; +import { z } from 'zod'; +import { toExtendedJson } from '@xo-cash/utils'; + +import type { RequestHeaders, RouteDefinition, RouteModule } from '../routes/types.ts'; +import { ApplicationError, UnauthorizedError } from '../errors/index.ts'; import { ApplicationRouteStream } from './route-stream.ts'; import type { BaseStream } from './stream/base-stream.ts'; +import type { AuthSecp256k1 } from '../auth/auth.ts'; /** Canonical request produced by every transport adapter. */ export type ApplicationRequest = { @@ -12,14 +16,32 @@ export type ApplicationRequest = { /** Transport-decoded application payload. */ body?: unknown; + /** Transport-normalized request headers, keyed by lowercase name. */ + headers?: RequestHeaders; + /** Optional correlation ID supplied by a multiplexed transport. */ requestId?: string; }; +export type ApplicationRouterDependencies = { + /** Authentication service. */ + auth: AuthSecp256k1; +}; + +const accountSchema = z.object({ + 'x-public-key': z.string(), + 'x-signature': z.string(), + 'x-timestamp': z.coerce.number(), +}).transform((data) => ({ + publicKey: data['x-public-key'], + signature: data['x-signature'], + timestamp: data['x-timestamp'], +})); + /** Exact-match application routing shared by every wire transport. */ export class ApplicationRouter { /** @param routes - Validated route table keyed by exact path. */ - private constructor(private readonly routes: ReadonlyMap) {} + private constructor(private readonly deps: ApplicationRouterDependencies, private readonly routes: ReadonlyMap) {} /** * Load and validate the complete route table before accepting traffic. @@ -27,7 +49,7 @@ export class ApplicationRouter { * @param routeModules - Route modules whose handlers will be registered. * @returns A ready-to-dispatch router instance. */ - static async create(routeModules: RouteModule[]): Promise { + static async create(deps: ApplicationRouterDependencies, routeModules: RouteModule[]): Promise { const routes = new Map(); // Collect routes from every module and reject duplicates at startup. @@ -42,7 +64,7 @@ export class ApplicationRouter { } } - return new ApplicationRouter(routes); + return new ApplicationRouter(deps, routes); } /** @@ -52,12 +74,36 @@ export class ApplicationRouter { * @param connection - Shared connection stream for this transport session. */ async dispatch(request: ApplicationRequest, connection: BaseStream): Promise { + const { publicKey, signature, timestamp } = accountSchema.parse(request.headers); + // Authenticate the headers on the request. (TODO: Remove the defaults, just here for testing) + // const publicKey = request.headers?.['x-public-key'] || 'public-key'; + // const signature = request.headers?.['x-signature'] || 'signature'; + // const timestamp = request.headers?.['x-timestamp'] || Date.now(); + + // Make sure the request signature is valid and hasnt been used before + await this.deps.auth.verifyUniqueRequest(signature); + + // Ensure that the headers are present. + if (!publicKey || !signature || !timestamp) { + throw new UnauthorizedError('Missing authentication headers'); + } + + // Compile the signature payload as `Path:Timestamp:Body` + const signaturePayload = `${timestamp}:${request.path}:${toExtendedJson(request.body)}`; + + // Verify the signature of the request. + const verified = await this.deps.auth.verifySignature(publicKey, signature, signaturePayload); + if (!verified) { + throw new UnauthorizedError('Invalid signature'); + } + + // Get the route from the routes map. const route = this.routes.get(request.path); if (!route) { throw new ApplicationError(404, `No route found for ${request.path}`); } - const stream = new ApplicationRouteStream(connection, request.body, request.requestId); + const stream = new ApplicationRouteStream(connection, request.body, request.path, request.requestId, request.headers); await route.handler(stream); } diff --git a/source/services/storage/migrations/001-resources.ts b/source/services/storage/migrations/001-resources.ts index 27dcf73..9967790 100644 --- a/source/services/storage/migrations/001-resources.ts +++ b/source/services/storage/migrations/001-resources.ts @@ -23,8 +23,19 @@ export const up = async (db: Kysely): Promise => { .addColumn('public_key', 'text', (col) => col.notNull()) .addColumn('blob', 'blob', (col) => col.notNull()) .addColumn('timestamp', 'integer', (col) => col.notNull().defaultTo(millisecondTime)) + .addColumn('signature', 'text', (col) => col.notNull()) .addPrimaryKeyConstraint('pk_resource_data', [ 'resource_id', 'public_key' ]) .execute(); + + // Table for authed requests + // We will store the signature and the timestamp of the request, and we will clear out rows that are older than our msTimeout for our auth + await db.schema + .createTable('authed_requests') + .ifNotExists() + .addColumn('signature', 'text', (col) => col.notNull()) + .addColumn('timestamp', 'integer', (col) => col.notNull().defaultTo(millisecondTime)) + .addPrimaryKeyConstraint('pk_authed_requests', [ 'signature' ]) + .execute(); }; /** @@ -35,4 +46,5 @@ export const up = async (db: Kysely): Promise => { export const down = async (db: Kysely): Promise => { await db.schema.dropTable('resource_data').ifExists() .execute(); + await db.schema.dropTable('authed_requests').ifExists().execute(); }; diff --git a/source/services/storage/tables.ts b/source/services/storage/tables.ts index d937c28..5241240 100644 --- a/source/services/storage/tables.ts +++ b/source/services/storage/tables.ts @@ -22,9 +22,44 @@ export interface ResourceDataTable { /** Millisecond timestamp of the last write. */ timestamp: Timestamp; + + /** Signature of the write. */ + signature: string; } +export interface AuthedRequestsTable { + /** Signature of the request. */ + signature: string; + + /** Millisecond timestamp of the request. */ + timestamp: Timestamp; +} + +// export interface PaymentsTable { +// /** Unique identifier for the payment. */ +// payment_id: string; + +// /** Public key of the account in the transaction */ +// public_key: string; + +// /** Amount of the payment. This can be positive or negative.*/ +// amount: number; + +// /** Timestamp of the payment. */ +// timestamp: Timestamp; + +// /** Signature of the payment. */ +// signature: string; + +// /** Hash of the message that was signed. */ +// message_hash: string; + +// /** Resource ID of the payment. */ +// resource_id: string; +// } + /** Complete Kysely schema mapping for the sync server database. */ export interface DatabaseTables { resource_data: ResourceDataTable; + authed_requests: AuthedRequestsTable; } diff --git a/source/services/transport/http-transport.ts b/source/services/transport/http-transport.ts index ec8b2a6..29f99c7 100644 --- a/source/services/transport/http-transport.ts +++ b/source/services/transport/http-transport.ts @@ -11,6 +11,7 @@ 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 { normalizeRequestHeaders } from './request-headers.ts'; /** Hono context key where decoded Extended JSON bodies are stored. */ const PARSED_BODY_KEY = 'parsedBody'; @@ -110,6 +111,7 @@ export class HttpTransportRouter implements TransportRouter { return { path: context.req.path, + headers: normalizeRequestHeaders(context.req.header()), ...(body === undefined ? {} : { body }), }; } diff --git a/source/services/transport/request-headers.ts b/source/services/transport/request-headers.ts new file mode 100644 index 0000000..9a65db2 --- /dev/null +++ b/source/services/transport/request-headers.ts @@ -0,0 +1,38 @@ +import type { RequestHeaders } from '../../routes/types.ts'; +import { ApplicationError } from '../../errors/index.ts'; +import { HTTP_STATUS_CODE_BAD_REQUEST } from '../../constants.ts'; + +/** RFC 9110 field-name token grammar. */ +const HEADER_NAME_PATTERN = /^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$/; + +/** + * 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. + */ +export const 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)); +}; diff --git a/source/services/transport/ws-transport.ts b/source/services/transport/ws-transport.ts index 544d467..b8201f1 100644 --- a/source/services/transport/ws-transport.ts +++ b/source/services/transport/ws-transport.ts @@ -13,6 +13,7 @@ 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 { normalizeRequestHeaders } from './request-headers.ts'; /** Default WebSocket upgrade path for application messages. */ const WS_ROUTE = '/ws'; @@ -24,6 +25,7 @@ const wsRequestSchema = z .optional(), path: z.string().min(1), body: z.unknown().optional(), + headers: z.record(z.string(), z.string()).optional(), }) .strict(); @@ -208,6 +210,7 @@ export class WsTransportRouter implements UpgradeTransportRouter { path: envelope.path, ...(envelope.id === undefined ? {} : { requestId: envelope.id }), ...(envelope.body === undefined ? {} : { body: envelope.body }), + ...(envelope.headers === undefined ? {} : { headers: normalizeRequestHeaders(envelope.headers) }), }; } diff --git a/test/routes/resources.test.ts b/test/routes/resources.test.ts index 65a47b1..06d2b1e 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 type { AuthSecp256k1 } from '../../source/auth/auth.ts'; +import { Accounts } from '../../source/auth/accounts.ts'; const createBroadcasterStub = (): BaseBroadcaster => { return { @@ -17,6 +19,31 @@ const createBroadcasterStub = (): BaseBroadcaster => { } as unknown as BaseBroadcaster; }; +const createAuthStub = (): AuthSecp256k1 => { + return { + verifySignature: vi.fn().mockImplementation(() => { + return true; + }), + } as unknown as AuthSecp256k1; +}; + +const createAccountsStub = (): Accounts => { + return { + getBalance: vi.fn().mockImplementation(() => { + return 0; + }), + setBalance: vi.fn().mockImplementation(() => { + return; + }), + deductBalance: vi.fn().mockImplementation(() => { + return; + }), + hasSufficientBalance: vi.fn().mockImplementation(() => { + return true; + }), + } as unknown as Accounts; +}; + describe('DataRoute subscriptions', (): void => { it('subscribes to future resource changes until removal', async (): Promise => { let resolveRemoved: () => void = () => undefined; @@ -31,10 +58,14 @@ describe('DataRoute subscriptions', (): void => { 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 stream = new ApplicationRouteStream( + connection, + { + resourceId: [ 'a', 'b' ], + }, + '/data/subscribe', + ); + const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); const execution = route.subscribeData(stream); @@ -55,8 +86,8 @@ 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' ] }, '/data/unsubscribe', 'unsubscribe-1'); + const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); await route.unsubscribeData(stream); @@ -78,10 +109,14 @@ describe('DataRoute subscriptions', (): void => { }, } as unknown as Database; const broadcaster = createBroadcasterStub(); - const stream = new ApplicationRouteStream(new TestConnection(true, false), { - resourceId: [ 'a' ], - }); - const route = new DataRoute(storage, broadcaster, 0); + const stream = new ApplicationRouteStream( + new TestConnection(true, false), + { + resourceId: [ 'a' ], + }, + '/data/unsubscribe', + ); + const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); await expect(route.unsubscribeData(stream)).rejects.toMatchObject({ statusCode: HTTP_STATUS_CODE_NOT_ACCEPTED }); expect(broadcaster.unsubscribe).not.toHaveBeenCalled(); @@ -97,7 +132,7 @@ describe('DataRoute resource write auth', (): void => { } as unknown as Database; const broadcaster = createBroadcasterStub(); - const route = new DataRoute(storage, broadcaster, 0); + const route = new DataRoute(storage, broadcaster, createAuthStub(), createAccountsStub(), 0); await expect(route.writeData({ connection: new TestConnection(true, true), @@ -115,6 +150,11 @@ describe('DataRoute resource write auth', (): void => { }, ], }, + headers: { + publicKey: 'public-key', + signature: 'signature', + timestamp: Date.now(), + }, } as unknown as ApplicationRouteStream)).rejects.toBeInstanceOf(UnauthorizedError); expect(storage.db.transaction).not.toHaveBeenCalled(); diff --git a/test/services/broadcaster.test.ts b/test/services/broadcaster.test.ts index 3acc018..d3a56e7 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, '/test'); }; const expectPending = async (promise: Promise): Promise => { diff --git a/test/services/router.test.ts b/test/services/router.test.ts index 20c1826..b9f2256 100644 --- a/test/services/router.test.ts +++ b/test/services/router.test.ts @@ -33,13 +33,15 @@ describe('ApplicationRouter dispatch', (): void => { url: '/echo', handler: async (stream): Promise => { expect(stream.connection).toBe(connection); + expect(stream.path).toBe('/echo'); + expect(stream.headers).toEqual({ 'x-request-token': 'route-1' }); await stream.send(stream.body); }, }, ]), ]); - await router.dispatch({ path: '/echo', body: { value: 1 }, requestId: 'request-1' }, connection); + await router.dispatch({ path: '/echo', body: { value: 1 }, requestId: 'request-1', headers: { 'x-request-token': 'route-1' } }, connection); expect(connection.messages).toEqual([ { @@ -61,16 +63,23 @@ describe('ApplicationRouter dispatch', (): void => { url: '/delayed', handler: async (stream): Promise => { const key = (stream.body as { key: string }).key; + const token = stream.headers['x-request-token']; await new Promise((resolve) => completions.set(key, resolve)); - await stream.send({ key }); + await stream.send({ key, token }); }, }, ]), ]); const connection = new TestConnection(true, true); - const first = router.dispatch({ path: '/delayed', body: { key: 'A' }, requestId: 'A' }, connection); - const second = router.dispatch({ path: '/delayed', body: { key: 'B' }, requestId: 'B' }, connection); + const first = router.dispatch( + { path: '/delayed', body: { key: 'A' }, requestId: 'A', headers: { 'x-request-token': 'token-a' } }, + connection, + ); + const second = router.dispatch( + { path: '/delayed', body: { key: 'B' }, requestId: 'B', headers: { 'x-request-token': 'token-b' } }, + connection, + ); completions.get('B')?.(); await second; @@ -82,13 +91,13 @@ describe('ApplicationRouter dispatch', (): void => { id: 'B', type: 'response', statusCode: 200, - body: { key: 'B' }, + body: { key: 'B', token: 'token-b' }, }, { id: 'A', type: 'response', statusCode: 200, - body: { key: 'A' }, + body: { key: 'A', token: 'token-a' }, }, ]); }); diff --git a/test/services/transports/http-transport.test.ts b/test/services/transports/http-transport.test.ts index 095d266..b5f670c 100644 --- a/test/services/transports/http-transport.test.ts +++ b/test/services/transports/http-transport.test.ts @@ -70,6 +70,23 @@ describe('HttpTransportRouter', (): void => { expect(await response.text()).toBe(''); }); + it('passes normalized HTTP request headers to the route stream', async (): Promise => { + const app = await createApp([ + { + url: '/headers', + handler: async (stream): Promise => stream.send({ path: stream.path, token: stream.headers['x-request-token'] }), + }, + ]); + + const response = await app.request('/headers', { + method: 'POST', + headers: { 'X-Request-Token': 'http-token' }, + }); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ path: '/headers', token: 'http-token' }); + }); + it('returns normalized errors for non-streaming requests', async (): Promise => { const app = await createApp([]); @@ -149,6 +166,27 @@ describe('HttpTransportRouter', (): void => { expect(events).toContain('data: {"ok":true}'); }); + it('passes normalized HTTP request headers to an SSE route stream', async (): Promise => { + const app = await createApp([ + { + url: '/headers', + handler: (stream): Promise => stream.send({ token: stream.headers['x-request-token'] }), + }, + ]); + + const response = await app.request('/headers', { + method: 'POST', + headers: { + accept: 'text/event-stream', + 'X-Request-Token': 'sse-token', + }, + }); + const events = await response.text(); + + expect(response.status).toBe(200); + expect(events).toContain('data: {"token":"sse-token"}'); + }); + it("keeps SSE open until the route's subscription promise resolves", async (): Promise => { let removeSubscription: () => Promise = async () => undefined; let markSubscribed: () => void = () => undefined; diff --git a/test/services/transports/ws-transport.test.ts b/test/services/transports/ws-transport.test.ts index 662316f..b8b9512 100644 --- a/test/services/transports/ws-transport.test.ts +++ b/test/services/transports/ws-transport.test.ts @@ -13,19 +13,36 @@ describe('WebSocket request decoding', (): void => { id: 'request-1', path: '/data/write', body: { value: new Uint8Array([ 1, 2, 3 ]) }, + headers: { 'X-Request-Token': 'ws-token' }, }))).resolves.toEqual({ requestId: 'request-1', path: '/data/write', body: { value: new Uint8Array([ 1, 2, 3 ]) }, + headers: { 'x-request-token': 'ws-token' }, }); }); - it.each([ '{}', '{"path":42}', '{"path":"/data/get","id":1}', '{"path":"/data/get","method":"POST"}' ])( - 'rejects an invalid envelope: %s', - async (payload) => { - await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError); - }, - ); + it('allows the optional headers object to be omitted', async (): Promise => { + await expect(WsTransportRouter.decodeWebSocketRequest('{"path":"/data/get"}')).resolves.toEqual({ path: '/data/get' }); + }); + + it.each([ + '{}', + '{"path":42}', + '{"path":"/data/get","id":1}', + '{"path":"/data/get","method":"POST"}', + '{"path":"/data/get","headers":{"x-request-token":1}}', + ])('rejects an invalid envelope: %s', async (payload) => { + await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toBeInstanceOf(z.ZodError); + }); + + it.each([ + '{"path":"/data/get","headers":{"bad header":"value"}}', + '{"path":"/data/get","headers":{"x-request-token":"first","X-Request-Token":"second"}}', + '{"path":"/data/get","headers":{"x-request-token":"first\\r\\nsecond"}}', + ])('rejects malformed request headers: %s', async (payload): Promise => { + await expect(WsTransportRouter.decodeWebSocketRequest(payload)).rejects.toMatchObject({ statusCode: 400 }); + }); it('rejects malformed JSON', async (): Promise => { await expect(WsTransportRouter.decodeWebSocketRequest('{')).rejects.toMatchObject({