Merge branch '4-add-broadcaster' into 5-add-http-and-sse

source/auth/auth.ts

@@ -0,0 +1,122 @@
+import type { Database } from '../services/storage/database.ts';
+
+import { hexToBin, instantiateSecp256k1, type Secp256k1, sha256 } from '@bitauth/libauth';
+import { UnauthorizedError } from '../errors/unauthorized-error.ts';
+
+export type AuthSecp256k1RequiredDeps = {
+    database: Database;
+};
+
+export type AuthSecp256k1OptionalDeps = {
+    secp256k1: Secp256k1;
+};
+export type AuthSecp256k1Deps = AuthSecp256k1RequiredDeps & Partial<AuthSecp256k1OptionalDeps>;
+
+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<AuthSecp256k1> {
+        const deps = {
+            secp256k1: await instantiateSecp256k1(),
+            ...inputDeps,
+        };
+
+        return new AuthSecp256k1(deps, options);
+    }
+
+    /**
+     * TODO: Consider adding a Record<string, Mutex> where each key is the signature to guarantee signatures arent being processed concurrently.
+     */
+    readonly #deps: Required<AuthSecp256k1Deps>;
+    readonly #options: AuthSecp256k1Options;
+
+    /**
+     * @param deps - The dependencies to use
+     * @param options - The options to use
+     */
+    private constructor(deps: Required<AuthSecp256k1Deps>, options: AuthSecp256k1Options) {
+        this.#deps = deps;
+        this.#options = options;
+    }
+
+    /**
+     * 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<boolean> {
+        // 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 = 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;
+    }
+
+    /**
+     * Verify a signature is unique and add it to the database if it is
+     *
+     * @param signature - The signature to verify
+     * @returns Whether the signature is unique
+     */
+    async verifyUniqueRequest(signature: string): Promise<boolean> {
+        // If the signature is not provided, throw an unauthorized error
+        if (!signature) {
+            throw new UnauthorizedError('Signature is required');
+        }
+
+        // Check if the signature has been used before
+        const request = await this.#deps.database.db.selectFrom('authed_requests').selectAll()
+.where('signature', '=', signature)
+.executeTakeFirst();
+
+        // If the signature has been used before, throw an unauthorized error
+        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 {
+        // Subtract the timestamp from the current time to get the age in milliseconds
+        const age = Math.abs(Date.now() - timestamp);
+
+        // If the timestamp is outside the allowed window, throw an unauthorized error
+        if (age > windowMs) {
+            throw new UnauthorizedError('Timestamp outside allowed window');
+        }
+    }
+}

source/index.ts

@@ -1,10 +1,12 @@
 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 { ServerHost } from './services/server-host.ts';
 import { Logger } from './utils/logger.ts';
+import type { RouteModule } from './routes/types.ts';
 
 /** Application composition root. */
 export class App {
@@ -21,14 +23,17 @@ 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 routes = [];
+        const routes: RouteModule[] = [];
 
         // 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);
 
         const http = new HttpTransportRouter(router, debug);
         const host = new ServerHost(config, debug, [ http ]);
@@ -59,6 +64,16 @@ 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();

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<Record<string, string>>;
+
 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;
 

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,
             },
         });
     }

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 {

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,38 @@ 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<string, RouteDefinition>) {}
+    private constructor(
+        private readonly deps: ApplicationRouterDependencies,
+        private readonly routes: ReadonlyMap<string, RouteDefinition>,
+    ) {}
 
     /**
      * Load and validate the complete route table before accepting traffic.
@@ -27,7 +55,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<ApplicationRouter> {
+    static async create(deps: ApplicationRouterDependencies, routeModules: RouteModule[]): Promise<ApplicationRouter> {
         const routes = new Map<string, RouteDefinition>();
 
         // Collect routes from every module and reject duplicates at startup.
@@ -42,7 +70,7 @@ export class ApplicationRouter {
             }
         }
 
-        return new ApplicationRouter(routes);
+        return new ApplicationRouter(deps, routes);
     }
 
     /**
@@ -52,12 +80,33 @@ export class ApplicationRouter {
      * @param connection - Shared connection stream for this transport session.
      */
     async dispatch(request: ApplicationRequest, connection: BaseStream): Promise<void> {
+        // Authenticate the headers on the request.
+        const { publicKey, signature, timestamp } = accountSchema.parse(request.headers);
+
+        // 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);
     }
 

source/services/storage/migrations/001-resources.ts

@@ -23,8 +23,19 @@ export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
         .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,7 @@ export const up = async (db: Kysely<DatabaseTables>): Promise<void> => {
 export const down = async (db: Kysely<DatabaseTables>): Promise<void> => {
     await db.schema.dropTable('resource_data').ifExists()
 .execute();
+
+    await db.schema.dropTable('authed_requests').ifExists()
+.execute();
 };

source/services/storage/tables.ts

@@ -22,9 +22,22 @@ 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;
 }
 
 /** Complete Kysely schema mapping for the sync server database. */
 export interface DatabaseTables {
     resource_data: ResourceDataTable;
+    authed_requests: AuthedRequestsTable;
 }

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<Record<string, string>>): RequestHeaders => {
+    const normalizedEntries: Array<[string, string]> = [];
+    const names = new Set<string>();
+
+    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));
+};