From df2c42e024341ae013c28828ba68c7d1934b062e Mon Sep 17 00:00:00 2001 From: Harvmaster Date: Thu, 22 Oct 2026 14:51:46 +0000 Subject: [PATCH] Add experiment encyption clients. Rewrite demos. Reduce implementation code. Add value decoding. Rename to base-client. Type inference and generics --- .../encrypted-client-per-resource.ts | 176 ++++++++++++ demo/experimental/encrypted-client.ts | 127 +++++++++ demo/shared.ts | 26 ++ demo/sse.ts | 33 --- demo/suite.ts | 89 ++++++ demo/tui.ts | 56 ++++ demo/ws.ts | 40 --- src/shared/base-client.ts | 267 ++++++++++++++++++ src/shared/client.ts | 142 ---------- src/sse/client.ts | 159 ++++++++--- src/ws/client.ts | 91 ++++-- 11 files changed, 923 insertions(+), 283 deletions(-) create mode 100644 demo/experimental/encrypted-client-per-resource.ts create mode 100644 demo/experimental/encrypted-client.ts create mode 100644 demo/shared.ts delete mode 100644 demo/sse.ts create mode 100644 demo/suite.ts create mode 100644 demo/tui.ts delete mode 100644 demo/ws.ts create mode 100644 src/shared/base-client.ts delete mode 100644 src/shared/client.ts diff --git a/demo/experimental/encrypted-client-per-resource.ts b/demo/experimental/encrypted-client-per-resource.ts new file mode 100644 index 0000000..179598e --- /dev/null +++ b/demo/experimental/encrypted-client-per-resource.ts @@ -0,0 +1,176 @@ +import { subtle, getRandomValues } from "node:crypto"; +import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils"; +import { sha256, binToBase64, binToHex } from "@bitauth/libauth"; + +import { BaseSyncClient, type Resource, type ResourcesRecord } from "../../src/shared/base-client.js"; + +type ResourceKey = { + resourceId: string; + keyMaterial: Uint8Array; +}; + +/** + * Encrypts each resource with its own key, where the resource ID itself is the secret. + * + * The caller-facing resource ID is hashed into a 32 byte AES key, and that key is hashed again + * to get the resource ID sent to the server. The server never sees the AES key or the original + * resource ID, so the resource ID must be unguessable (high entropy) for this to be secure. + */ +export class EncryptedClientPerResource extends BaseSyncClient { + /** Maps the hashed (server-side) resource ID to the original resource ID and its key */ + private readonly resourceKeys = new Map(); + + constructor(private readonly client: BaseSyncClient) { + super(); + + this.client.on('resource-instance-updated', async (message) => { + // Ignore updates for resources we don't hold the key for + const resourceKey = this.resourceKeys.get(message.resourceId); + if (!resourceKey) return; + + try { + this.emit('resource-instance-updated', { + ...message, + resourceId: resourceKey.resourceId, + value: await EncryptedClientPerResource.decrypt(resourceKey.keyMaterial, message.value) as T, + }); + } catch (error) { + this.emit('error', error instanceof Error ? error : new Error(String(error))); + } + }); + } + + /** + * Derives the AES key and server-side resource ID from a resource ID, and remembers the mapping. + * @param resourceId - The resource ID (the secret) + * @returns The hashed resource ID to send to the server + */ + private hashResourceId(resourceId: string): string { + const keyMaterial = sha256.hash(new TextEncoder().encode(resourceId)); + const hashedResourceId = binToHex(sha256.hash(keyMaterial)); + + this.resourceKeys.set(hashedResourceId, { resourceId, keyMaterial }); + + return hashedResourceId; + } + + /** + * Gets the original resource ID and key for a hashed resource ID. + * @param hashedResourceId - The hashed (server-side) resource ID + * @returns The original resource ID and its key material + */ + private getResourceKey(hashedResourceId: string): ResourceKey { + const resourceKey = this.resourceKeys.get(hashedResourceId); + + if (!resourceKey) { + throw new Error(`No key found for resource ${hashedResourceId}`); + } + + return resourceKey; + } + + /** + * Decrypts every instance of every resource in a record, and maps the hashed resource IDs back to the original resource IDs. + * @param resources - The encrypted resources record, keyed by hashed resource ID + * @returns The resources record with decrypted values, keyed by original resource ID + */ + private async decryptResources(resources: ResourcesRecord): Promise> { + const entries = await Promise.all(Object.entries(resources).map(async ([ hashedResourceId, resource ]) => { + const { resourceId, keyMaterial } = this.getResourceKey(hashedResourceId); + + const instances: Resource[] = await Promise.all(resource.instances.map(async (instance) => ({ + ...instance, + resourceId, + value: await EncryptedClientPerResource.decrypt(keyMaterial, instance.value) as T, + }))); + + return [ resourceId, { ...resource, instances } ] as const; + })); + + return Object.fromEntries(entries); + } + + public async connect(): Promise { + return this.client.connect(); + } + + public async disconnect(): Promise { + return this.client.disconnect(); + } + + public async readMany(resourceIds: string[]): Promise> { + const hashedResourceIds = resourceIds.map((resourceId) => this.hashResourceId(resourceId)); + + const encryptedResources = await this.client.readMany(hashedResourceIds); + + return this.decryptResources(encryptedResources); + } + + public async writeMany(resources: Record): Promise> { + const encryptedEntries = await Promise.all(Object.entries(resources).map(async ([ resourceId, value ]) => { + const hashedResourceId = this.hashResourceId(resourceId); + const { keyMaterial } = this.getResourceKey(hashedResourceId); + + return [ hashedResourceId, await EncryptedClientPerResource.encrypt(keyMaterial, value) ] as const; + })); + + const response = await this.client.writeMany(Object.fromEntries(encryptedEntries)); + + return this.decryptResources(response); + } + + public async subscribe(resourceId: string): Promise { + return this.client.subscribe(this.hashResourceId(resourceId)); + } + + public async unsubscribe(resourceId: string): Promise { + return this.client.unsubscribe(this.hashResourceId(resourceId)); + } + + /** + * Imports raw key material as an AES-GCM key + * @param keyMaterial - The 32 byte key material + * @returns The AES-GCM CryptoKey + */ + private static importKey(keyMaterial: Uint8Array) { + return subtle.importKey('raw', Buffer.from(keyMaterial), { name: 'AES-GCM' }, false, ['encrypt', 'decrypt']); + } + + /** + * Encrypts a value using AES-GCM with a random 12 byte IV + * @param keyMaterial - The encryption key material + * @param value - The value to encrypt + * @returns The IV and ciphertext as a base64 string + */ + public static async encrypt(keyMaterial: Uint8Array, value: unknown): Promise { + const key = await this.importKey(keyMaterial); + const iv = getRandomValues(new Uint8Array(12)); + + // Encode the value as extended JSON and then to UTF-8 bytes + const encodedValue = new TextEncoder().encode(toExtendedJson(value)); + + const encrypted = await subtle.encrypt({ name: 'AES-GCM', iv }, key, encodedValue); + + // Prepend the IV so it can be recovered when decrypting + return binToBase64(Buffer.concat([ iv, new Uint8Array(encrypted) ])); + } + + /** + * Decrypts a value encrypted by `encrypt` + * @param keyMaterial - The encryption key material + * @param value - The IV and ciphertext as a base64 string + * @returns The decrypted value + */ + public static async decrypt(keyMaterial: Uint8Array, value: string): Promise { + const key = await this.importKey(keyMaterial); + + // Decode the base64 first: the IV is 12 bytes, not 12 base64 characters + const bytes = Buffer.from(value, 'base64'); + const iv = bytes.subarray(0, 12); + const encryptedValue = bytes.subarray(12); + + const decrypted = await subtle.decrypt({ name: 'AES-GCM', iv }, key, encryptedValue); + + return fromExtendedJson(new TextDecoder().decode(decrypted)); + } +} diff --git a/demo/experimental/encrypted-client.ts b/demo/experimental/encrypted-client.ts new file mode 100644 index 0000000..bbcfb6e --- /dev/null +++ b/demo/experimental/encrypted-client.ts @@ -0,0 +1,127 @@ +import { subtle, getRandomValues } from "node:crypto"; +import { fromExtendedJson, toExtendedJson } from "@xo-cash/utils"; +import { sha256, binToBase64 } from "@bitauth/libauth"; + +import { BaseSyncClient, type ResourcesRecord } from "../../src/shared/base-client.js"; + +/** + * Encrypts every resource with a single shared key. + */ +export class EncryptedClient extends BaseSyncClient { + private readonly encryptionKey: Uint8Array; + + constructor(private readonly client: BaseSyncClient, encryptionKey: Uint8Array | string) { + super(); + + // Hash the encryption key into 32 bytes if it is a string + this.encryptionKey = typeof encryptionKey === 'string' ? sha256.hash(new TextEncoder().encode(encryptionKey)) : encryptionKey; + + this.client.on('resource-instance-updated', async (message) => { + try { + this.emit('resource-instance-updated', { + ...message, + value: await EncryptedClient.decrypt(this.encryptionKey, message.value) as T, + }); + } catch (error) { + this.emit('error', error instanceof Error ? error : new Error(String(error))); + } + }); + } + + /** + * Decrypts every instance of every resource in a record. + * @param resources - The encrypted resources record + * @returns The resources record with decrypted values + */ + private async decryptResources(resources: ResourcesRecord): Promise> { + const entries = await Promise.all(Object.entries(resources).map(async ([ resourceId, resource ]) => { + const instances = await Promise.all(resource.instances.map(async (instance) => ({ + ...instance, + value: await EncryptedClient.decrypt(this.encryptionKey, instance.value) as T, + }))); + + return [ resourceId, { ...resource, instances } ] as const; + })); + + return Object.fromEntries(entries); + } + + public async connect(): Promise { + return this.client.connect(); + } + + public async disconnect(): Promise { + return this.client.disconnect(); + } + + public async readMany(resourceIds: string[]): Promise> { + const encryptedResources = await this.client.readMany(resourceIds); + + return this.decryptResources(encryptedResources); + } + + public async writeMany(resources: Record): Promise> { + const encryptedEntries = await Promise.all(Object.entries(resources).map(async ([ resourceId, value ]) => { + return [ resourceId, await EncryptedClient.encrypt(this.encryptionKey, value) ] as const; + })); + + const response = await this.client.writeMany(Object.fromEntries(encryptedEntries)); + + return this.decryptResources(response); + } + + public async subscribe(resourceId: string): Promise { + return this.client.subscribe(resourceId); + } + + public async unsubscribe(resourceId: string): Promise { + return this.client.unsubscribe(resourceId); + } + + /** + * Imports raw key material as an AES-GCM key + * @param keyMaterial - The 32 byte key material + * @returns The AES-GCM CryptoKey + */ + private static importKey(keyMaterial: Uint8Array) { + return subtle.importKey('raw', Buffer.from(keyMaterial), { name: 'AES-GCM' }, false, ['encrypt', 'decrypt']); + } + + /** + * Encrypts a value using AES-GCM with a random 12 byte IV + * @param keyMaterial - The encryption key material + * @param value - The value to encrypt + * @returns The IV and ciphertext as a base64 string + */ + public static async encrypt(keyMaterial: Uint8Array, value: unknown): Promise { + const key = await this.importKey(keyMaterial); + const iv = getRandomValues(new Uint8Array(12)); + + // Encode the value as extended JSON and then to UTF-8 bytes + const encodedValue = new TextEncoder().encode(toExtendedJson(value)); + + const encrypted = await subtle.encrypt({ name: 'AES-GCM', iv }, key, encodedValue); + + // Prepend the IV so it can be recovered when decrypting + return binToBase64(Buffer.concat([ iv, new Uint8Array(encrypted) ])); + } + + /** + * Decrypts a value encrypted by `encrypt` + * @param keyMaterial - The encryption key material + * @param value - The IV and ciphertext as a base64 string + * @returns The decrypted value + */ + public static async decrypt(keyMaterial: Uint8Array, value: string): Promise { + const key = await this.importKey(keyMaterial); + + // Decode the base64 first: the IV is 12 bytes, not 12 base64 characters + const bytes = Buffer.from(value, 'base64'); + const iv = bytes.subarray(0, 12); + const encryptedValue = bytes.subarray(12); + + const decrypted = await subtle.decrypt({ name: 'AES-GCM', iv }, key, encryptedValue); + + return fromExtendedJson(new TextDecoder().decode(decrypted)); + } +} diff --git a/demo/shared.ts b/demo/shared.ts new file mode 100644 index 0000000..4b34032 --- /dev/null +++ b/demo/shared.ts @@ -0,0 +1,26 @@ +import type { BaseSyncClient, ResourcesRecord } from "../src/shared/base-client.js"; + +export const writeMany = async (client: BaseSyncClient<{ message: string }>): Promise> => { + // Create a group of resources with changes + const resources = Object.fromEntries(Array.from({ length: 10 }, (_, index) => [ + `resource-${index}`, + { + message: `Hello, world! this is the minimal sync client using ${client.constructor.name} ${index}`, + }, + ])); + + // Write the resources to the server + const res = await client.writeMany(resources); + + return res; +} + +export const readMany = async (client: BaseSyncClient<{ message: string }>,): Promise> => { + // Create a list of resource IDs + const resources = Array.from({ length: 10 }, (_, index) => `resource-${index}`); + + // Read the resources from the server + const res = await client.readMany(resources); + + return res; +} diff --git a/demo/sse.ts b/demo/sse.ts deleted file mode 100644 index 3f9b565..0000000 --- a/demo/sse.ts +++ /dev/null @@ -1,33 +0,0 @@ -import { PrivateKey } from "@xo-cash/primitives"; - -import { SSEClient } from '../src/sse/client.js'; - -const testSSE = async () => { - const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2'); - const client = new SSEClient('https://v2.sync.xo.harvmaster.com', privateKey); - - client.on('message', (message) => { - console.log(message); - }); - - // Connect to the sync server - await client.connect(); - - // Subscribe to the resource - await client.subscribe('test'); - - // Write a new value to the resource - await client.write('test', {'message': 'Hello, world! this is the minimal sync client using SSE'}); - - // Read the current value of the resource - const data = await client.read('test'); - console.log(data); - - // If the user presses Ctrl+C, disconnect the client and exit the program - process.on('SIGINT', async () => { - await client.disconnect(); - process.exit(0); - }); -} - -testSSE(); diff --git a/demo/suite.ts b/demo/suite.ts new file mode 100644 index 0000000..302de78 --- /dev/null +++ b/demo/suite.ts @@ -0,0 +1,89 @@ +import { PrivateKey } from "@xo-cash/primitives"; + +// Core clients +import type { BaseSyncClient } from "../src/shared/base-client.js"; +import { SSEClient } from '../src/sse/client.js'; +import { WsClient } from "../src/ws/client.js"; + +// Experimental clients +import { EncryptedClientPerResource } from "./experimental/encrypted-client-per-resource.js"; +import { EncryptedClient } from "./experimental/encrypted-client.js"; + +// Shared functions +import { writeMany, readMany } from "./shared.js"; + +const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2'); + +const clients = { + sse: new SSEClient<{ message: string }>('https://dev.v2.sync.xo.harvmaster.com', privateKey), + ws: new WsClient<{ message: string }>('wss://dev.v2.sync.xo.harvmaster.com', privateKey), + encrypted: new EncryptedClient<{ message: string }>(new SSEClient('https://dev.v2.sync.xo.harvmaster.com', privateKey), 'secret'), + encryptedPerResource: new EncryptedClientPerResource<{ message: string }>(new SSEClient('https://dev.v2.sync.xo.harvmaster.com', privateKey)), +}; + +// Test the suite for a given client +const testSuite = async (client: BaseSyncClient<{ message: string }>) => { + client.on('resource-instance-updated', (message) => { + console.log(message); + }); + + // Connect to the sync server + await client.connect(); + + // Subscribe to the resource + await client.subscribe('test'); + + // Write a new value to the resource + await client.write('test', {'message': 'Hello, world! this is the minimal sync client using ' + client.constructor.name}); + + // Read the current value of the resource + const data = await client.read('test'); + console.log(data.map((resource) => resource.value.message)); + + // Use the shared functions to write and read many resources + const writtenResources = await writeMany(client); + console.dir(writtenResources, { depth: null }); + + const readResources = await readMany(client); + console.dir(readResources, { depth: null }); + + // If the user presses Ctrl+C, disconnect the client and exit the program + process.on('SIGINT', async () => { + await client.disconnect(); + process.exit(0); + }); +} + +// Get the clients to test +const selectClients = (params: string[]): BaseSyncClient<{ message: string }>[] => { + // If no params are provided, test all clients + const names = params.length > 0 ? params : Object.keys(clients); + + // Find values that are unknown to the clients object + const unknown = names.filter((n) => !(n in clients)); + if (unknown.length > 0) { + console.error(`Invalid client type(s): ${unknown.join(', ')}. Expected: ${Object.keys(clients).join(', ')}`); + process.exit(1); + } + + // Return the clients that are known to the clients object + return names.map((n) => clients[n as keyof typeof clients]); +} + +// Get the params +const params = process.argv.slice(2); + +// Get the clients to test +const clientsToTest = selectClients(params); + +// Test the clients +for (const client of clientsToTest) { + await testSuite(client); +} + +// Wait for the user to press Ctrl+C +await new Promise((resolve) => process.on('SIGINT', resolve)); + +// Exit the program +console.log('Exiting...'); +process.exit(0); \ No newline at end of file diff --git a/demo/tui.ts b/demo/tui.ts new file mode 100644 index 0000000..8f774e2 --- /dev/null +++ b/demo/tui.ts @@ -0,0 +1,56 @@ +import { PrivateKey } from "@xo-cash/primitives"; + +import type { BaseSyncClient } from "../src/shared/base-client.js"; + +import { SSEClient } from "../src/sse/client.js"; +import { EncryptedClientPerResource } from "./experimental/encrypted-client-per-resource.js"; + +const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2'); + +// Define some formatting helpers +const formatResourceId = (resourceId: string) => { + return `\x1b[32m${resourceId}\x1b[0m`; +} +const formatMessage = (message: string) => { + return `\x1b[34m${message}\x1b[0m`; +} +const resetFormatting = () => { + return `\x1b[0m`; +} + +const startTUI = async (client: BaseSyncClient) => { + // Listen and log out the resource change + client.on('resource-instance-updated', (message) => { + // Print the message with colours + console.log(`${formatResourceId(message.resourceId)}:${formatResourceId(message.publicKey)} updated: ${formatMessage(message.value)}${resetFormatting()}`); + }); + + // Connect to the sync server and subscribe to the resource + await client.connect(); + await client.subscribe('encrypted-test'); + + // Start the TUI loop + while (true) { + // Print the prompt + process.stdout.write('> '); + + // Wait for the user to enter a command + const data = await new Promise((resolve) => process.stdin.once('data', resolve)); + + // If the user entered a command, write it to the resource + if (data) { + const message = data.toString().trim(); + await client.write('encrypted-test', message); + } + } +} + +const clients = { + 'sse': new SSEClient('https://dev.v2.sync.xo.harvmaster.com', privateKey), + 'encrypted': new EncryptedClientPerResource(new SSEClient('https://dev.v2.sync.xo.harvmaster.com', privateKey)), +} + +const params = process.argv.slice(2); +const client = params[0] === 'false' ? clients.sse : clients.encrypted; + +startTUI(client); \ No newline at end of file diff --git a/demo/ws.ts b/demo/ws.ts deleted file mode 100644 index ed5b7c1..0000000 --- a/demo/ws.ts +++ /dev/null @@ -1,40 +0,0 @@ -import { PrivateKey } from "@xo-cash/primitives"; - -import { WsClient } from '../src/ws/client.js'; - -const testWS = async () => { - const privateKey = PrivateKey.fromString('c440ce6ac5b63ae7bc4af9891b80b7a88ccdeebba5e43246885e9ec56f912ac2'); - const client = new WsClient('https://v2.sync.xo.harvmaster.com', privateKey, { - // onMessage: (message) => { - // console.log(toExtendedJson(message)); - // }, - onError: (error) => { - console.error(error); - }, - }); - - client.on('message', (message) => { - console.log(message); - }); - - // Connect to the sync server - await client.connect(); - - // Subscribe to the resource - client.subscribe('test'); - - // Write a new value to the resource - await client.write('test', {'message': 'Hello, world! this is the minimal sync client using WS'}); - - // Read the current value of the resource - const data = await client.read('test'); - console.log(data); - - // If the user presses Ctrl+C, disconnect the client and exit the program - process.on('SIGINT', async () => { - await client.disconnect(); - process.exit(0); - }); -} - -testWS(); diff --git a/src/shared/base-client.ts b/src/shared/base-client.ts new file mode 100644 index 0000000..0631348 --- /dev/null +++ b/src/shared/base-client.ts @@ -0,0 +1,267 @@ +import { binToHex, flattenBinArray, secp256k1, sha256 } from "@bitauth/libauth"; + +import { toExtendedJson, EventEmitter, fromExtendedJson } from "@xo-cash/utils"; +import type { WriteRequest } from "../types.js"; +import z from "zod"; + +export type AuthenticatedRequestHeaders = { + 'X-Public-Key': string; + 'X-Timestamp': string; + 'X-Signature': string; +}; + +export type SignedPayload = { publicKey: string; signature: string }; + +export type XOSyncClientEventMap = { + 'connection-updated': 'connected' | 'disconnected'; + 'resource-instance-updated': Resource & { + resourceId: string; + }; + 'subscriptions-updated': Set; + 'error': Error; +}; + +export type Resource = z.infer & { + value: PayloadType; +}; + +export type ResourcesRecord = Record[] }>; + +export abstract class BaseSyncClient extends EventEmitter> { + + /** + * Connects to the sync server + */ + abstract connect(): Promise; + + /** + * Disconnects from the sync server + */ + abstract disconnect(): Promise; + + /** + * Sign and write one resource + * @param resourceId - The resource ID to write to + * @param value - The value to write + * @returns The Resource Object with the decoded value + */ + async write(resourceId: string, value: T): Promise[]> { + // Write the resource to the server + const response = await this.writeMany({ [resourceId]: value }); + + // Get the resource from the response + const resource = response[resourceId]; + + // If the resource is not found, throw an error + if (!resource) { + throw new Error(`Resource ${resourceId} not found`); + } + + // Return the resource instances + return resource.instances; + } + + /** + * Writes multiple values to resources + * @param resources - The resources to write to + * @returns The resources with the written values + */ + abstract writeMany(resources: Record): Promise>; + + /** + * Reads a value from a resource + * @param resourceId - The resource ID to read from + * @returns The resource with the read value + */ + async read(resourceId: string): Promise[]> { + // Read the resource from the server + const response = await this.readMany([ resourceId ]); + + // Get the resource from the response + const resource = response[resourceId]; + + // If the resource is not found, throw an error + if (!resource) { + throw new Error(`Resource ${resourceId} not found`); + } + + // Return the resource instances + return resource.instances; + } + + /** + * Reads multiple values from resources + * @param resourceIds - The resource IDs to read from + * @returns The resources with the read values + */ + abstract readMany(resourceIds: string[]): Promise>; + + /** + * Subscribes to a resource + * @param resourceId - The resource ID to subscribe to + */ + abstract subscribe(resourceId: string): Promise; + + /** + * Unsubscribes from a resource + * @param resourceId - The resource ID to unsubscribe from + */ + abstract unsubscribe(resourceId: string): Promise; + + /** + * Derives the resource private key from the private key and resource ID + * @param privateKey - The private key to derive the resource private key from + * @param resourceId - The resource ID to derive the resource private key from + * @returns The resource private key + */ + static deriveResourcePrivateKey(privateKey: Uint8Array, resourceId: string): Uint8Array { + // Convert the resource ID to bytes + const resourceIdBytes = new TextEncoder().encode(resourceId); + + // Hash the private key and resource ID + return sha256.hash(flattenBinArray([ privateKey, resourceIdBytes ])); + } + + /** + * Signs a payload with the private key and returns the public key and signature + * @param privateKey - The private key to sign the payload with + * @param payload - The payload to sign + * @returns The public key and signature + */ + static signPayload(privateKey: Uint8Array, payload: string): SignedPayload { + // Convert the payload to Binary and hash it + const payloadHash = sha256.hash(new TextEncoder().encode(payload)); + + // Derive the public key + const publicKey = secp256k1.derivePublicKeyCompressed(privateKey); + + // If the public key is a string, throw an error + if (typeof publicKey === 'string') { + throw new Error('Failed to derive public key'); + } + + // Sign the payload + const signature = secp256k1.signMessageHashDER(privateKey, payloadHash); + + // If the signature is a string, throw an error + if (typeof signature === 'string') { + throw new Error('Failed to sign message'); + } + + return { + publicKey: binToHex(publicKey), + signature: binToHex(signature), + }; + } + + /** + * Signs a write request with the private key and returns the write request + * @param privateKey - The private key to sign the write request with + * @param resourceId - The resource ID to write to + * @param value - The value to write + * @returns The write request + */ + static signWriteRequest(privateKey: Uint8Array, resourceId: string, value: Uint8Array): WriteRequest { + // Derive the resource private key + const derivedKey = this.deriveResourcePrivateKey(privateKey, resourceId); + + // Create the payload + const timestamp = Date.now(); + const payload = `${timestamp}:${resourceId}:${toExtendedJson(value)}`; + + // Sign the payload + const { publicKey, signature } = this.signPayload(derivedKey, payload); + + return { + id: resourceId, + value, + publicKey, + timestamp, + signature, + }; + } + + /** + * Signs a request with the private key and returns the public key, timestamp, and signature + * @param privateKey - The private key to sign the request with + * @param path - The path of the request + * @param body - The body of the request + * @returns The public key, timestamp, and signature + */ + static signRequest(privateKey: Uint8Array, path: string, body: unknown): AuthenticatedRequestHeaders { + // Create the payload + const timestamp = Date.now(); + + // Convert the body to a string if it isn't already a string + const bodyStr = typeof body === 'string' ? body : toExtendedJson(body); + + // Create the payload as `Timestamp:Path:Body` + const payload = `${timestamp}:${path}:${bodyStr}`; + + // Sign the payload + const { publicKey, signature } = this.signPayload(privateKey, payload); + + return { + 'X-Public-Key': publicKey, + 'X-Timestamp': timestamp.toString(), + 'X-Signature': signature, + }; + } + + /** + * Decodes the 'blob' field of the Resource Object from the server from a read/write + * @param response - The ResourceResponse Object from the server from a read/write + * @returns The Resource Object with the decoded value + */ + static decodeResourcesResponse(response: z.infer): ResourcesRecord { + return Object.entries(response.resources).reduce>((acc, [resourceId, resource]) => { + // Get the instances + const instances = resource.instances; + + // Decode the instances + const decodedInstances = instances.map((instance) => { + return BaseSyncClient.decodeBlobValue(instance); + }); + + // Add the decoded instances to the accumulator + acc[resourceId] = { + ...resource, + instances: decodedInstances, + }; + + // Return the accumulator + return acc; + }, {}); + } + + static decodeBlobValue(instance: z.infer): Resource { + const decodedBlob = new TextDecoder().decode(instance.blob); + const value = fromExtendedJson(decodedBlob) as T; + + return { + ...instance, + value, + }; + } + + static resourceInstanceSchema = z.object({ + resourceId: z.string(), + blob: z.instanceof(Uint8Array), + timestamp: z.number(), + publicKey: z.string(), + signature: z.string(), + }); + + static rawResourcesSchema = z.object({ + instances: z.array(BaseSyncClient.resourceInstanceSchema), + }); + + static resourcesResponseSchema = z.object({ + resources: z.record( + // Resource ID + z.string(), + // Resource instances + BaseSyncClient.rawResourcesSchema, + ), + }); +} diff --git a/src/shared/client.ts b/src/shared/client.ts deleted file mode 100644 index a194ef5..0000000 --- a/src/shared/client.ts +++ /dev/null @@ -1,142 +0,0 @@ -import { binToHex, flattenBinArray, secp256k1, sha256 } from "@bitauth/libauth"; - -import { toExtendedJson, EventEmitter } from "@xo-cash/utils"; -import type { WriteRequest } from "../types.js"; -import z from "zod"; - -export type AuthenticatedRequestHeaders = { - 'X-Public-Key': string; - 'X-Timestamp': string; - 'X-Signature': string; -}; - -export type SignedPayload = { publicKey: string; signature: string }; - -export type XOSyncClientEventMap = { - 'connection-updated': 'connected' | 'disconnected'; - 'message': { - blob: unknown; - timestamp: number; - publicKey: string; - signature: string; - }; - 'subscriptions-updated': Set; - 'error': Error; -}; - -export abstract class SyncClient extends EventEmitter { - - abstract connect(): Promise; - abstract disconnect(): Promise; - abstract write(resourceId: string, value: Record): Promise; - abstract read(resourceId: string): Promise; - abstract subscribe(resourceId: string): Promise; - abstract unsubscribe(resourceId: string): Promise; - - /** - * Derives the resource private key from the private key and resource ID - * @param privateKey - The private key to derive the resource private key from - * @param resourceId - The resource ID to derive the resource private key from - * @returns The resource private key - */ - static deriveResourcePrivateKey(privateKey: Uint8Array, resourceId: string): Uint8Array { - // Convert the resource ID to bytes - const resourceIdBytes = new TextEncoder().encode(resourceId); - - // Hash the private key and resource ID - return sha256.hash(flattenBinArray([ privateKey, resourceIdBytes ])); - } - - /** - * Signs a payload with the private key and returns the public key and signature - * @param privateKey - The private key to sign the payload with - * @param payload - The payload to sign - * @returns The public key and signature - */ - static signPayload(privateKey: Uint8Array, payload: string): SignedPayload { - // Convert the payload to Binary and hash it - const payloadHash = sha256.hash(new TextEncoder().encode(payload)); - - // Derive the public key - const publicKey = secp256k1.derivePublicKeyCompressed(privateKey); - - // If the public key is a string, throw an error - if (typeof publicKey === 'string') { - throw new Error('Failed to derive public key'); - } - - // Sign the payload - const signature = secp256k1.signMessageHashDER(privateKey, payloadHash); - - // If the signature is a string, throw an error - if (typeof signature === 'string') { - throw new Error('Failed to sign message'); - } - - return { - publicKey: binToHex(publicKey), - signature: binToHex(signature), - }; - } - - /** - * Signs a write request with the private key and returns the write request - * @param privateKey - The private key to sign the write request with - * @param resourceId - The resource ID to write to - * @param value - The value to write - * @returns The write request - */ - static signWriteRequest(privateKey: Uint8Array, resourceId: string, value: Uint8Array): WriteRequest { - // Derive the resource private key - const derivedKey = this.deriveResourcePrivateKey(privateKey, resourceId); - - // Create the payload - const timestamp = Date.now(); - const payload = `${timestamp}:${resourceId}:${toExtendedJson(value)}`; - - // Sign the payload - const { publicKey, signature } = this.signPayload(derivedKey, payload); - - return { - id: resourceId, - value, - publicKey, - timestamp, - signature, - }; - } - - /** - * Signs a request with the private key and returns the public key, timestamp, and signature - * @param privateKey - The private key to sign the request with - * @param path - The path of the request - * @param body - The body of the request - * @returns The public key, timestamp, and signature - */ - static signRequest(privateKey: Uint8Array, path: string, body: unknown): AuthenticatedRequestHeaders { - // Create the payload - const timestamp = Date.now(); - - // Convert the body to a string if it isn't already a string - const bodyStr = typeof body === 'string' ? body : toExtendedJson(body); - - // Create the payload as `Timestamp:Path:Body` - const payload = `${timestamp}:${path}:${bodyStr}`; - - // Sign the payload - const { publicKey, signature } = this.signPayload(privateKey, payload); - - return { - 'X-Public-Key': publicKey, - 'X-Timestamp': timestamp.toString(), - 'X-Signature': signature, - }; - } - - static messageSchema = z.object({ - blob: z.unknown(), - timestamp: z.number(), - publicKey: z.string(), - signature: z.string(), - }); -} diff --git a/src/sse/client.ts b/src/sse/client.ts index 906b641..e7e873d 100644 --- a/src/sse/client.ts +++ b/src/sse/client.ts @@ -1,9 +1,9 @@ import { SSESession, toExtendedJson, fromExtendedJson } from '@xo-cash/utils'; import { PrivateKey } from '@xo-cash/primitives' -import { SyncClient } from '../shared/client.js'; +import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.js'; -export class SSEClient extends SyncClient { +export class SSEClient extends BaseSyncClient { private readonly url: string; public sseSession: SSESession | undefined; private readonly subscriptions: Set = new Set(); @@ -15,6 +15,11 @@ export class SSEClient extends SyncClient { return `${this.url}/data/subscribe`; } + /** + * Creates a new SSE client + * @param url - The URL of the server + * @param privateKey - The private key to use for authentication + */ constructor(url: string, privateKey: PrivateKey) { super(); @@ -29,6 +34,9 @@ export class SSEClient extends SyncClient { }, 25); } + /** + * Connects to the server + */ async connect() { // The server rejects a subscribe request with no resource IDs, so there is nothing to connect to yet. // subscribe() connects once there is a subscription. read() and write() are plain fetches and don't need a session. @@ -41,10 +49,16 @@ export class SSEClient extends SyncClient { await this.sseSession.connect(); } + /** + * Disconnects from the server + */ async disconnect() { await this.sseSession?.disconnect(); } + /** + * Reconnects to the server + */ async reconnect() { // Close the current session. Its subscription list is fixed in the request body, so it can't be reused this.sseSession?.close(); @@ -54,6 +68,10 @@ export class SSEClient extends SyncClient { await this.connect(); } + /** + * Creates a new SSE session + * @returns The SSE session + */ private createSession() { // Build the request body const body = toExtendedJson({ resourceId: [ ...this.subscriptions ] }); @@ -66,7 +84,7 @@ export class SSEClient extends SyncClient { const path = url.pathname; // Sign the request body - const authHeaders = SyncClient.signRequest(this.privateKey.toBytes(), path, request.body); + const authHeaders = BaseSyncClient.signRequest(this.privateKey.toBytes(), path, request.body); // Initialize the request headers if they don't exist request.headers ??= {}; @@ -95,12 +113,15 @@ export class SSEClient extends SyncClient { // Create a listener to re-emit the messages session.on('message', (message) => { - const parsedMessage = SyncClient.messageSchema.safeParse(fromExtendedJson(message.data)); + const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(fromExtendedJson(message.data)); if (!parsedMessage.success) { return; } - this.emit('message', parsedMessage.data); + this.emit('resource-instance-updated', { + ...parsedMessage.data, + value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType, + }); }); // Create a listener to re-emit the errors @@ -112,69 +133,70 @@ export class SSEClient extends SyncClient { return session; } - // Send a POST /data/write request to the server - async write(resourceId: string, value: Record) { - const url = `${this.url}/data/write`; - - const valueStr = toExtendedJson(value); - const valueBytes = this.textEncoder.encode(valueStr); - const resource = SyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes); - - const body = toExtendedJson({ resources: [resource] }); - - const res = await fetch(url, { - method: 'POST', - - // Headers arent required unless we have payment service running - headers: { - 'Content-Type': 'application/json', - ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/write', body), - }, - - body, + /** + * Send a POST /data/write request to the server + * @param resources - The resources to write + * @returns The Resource Object with the decoded value + */ + async writeMany(resources: Record): Promise> { + // Sign the resource write requests + const signedResources = Object.entries(resources).map(([resourceId, value]) => { + const valueStr = toExtendedJson(value); + const valueBytes = this.textEncoder.encode(valueStr); + return BaseSyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes); }); + // Send the resource write requests to the server + const res = await this.sendRequest('/data/write', { resources: signedResources }); + + // If the response is not ok, throw an error if (!res.ok) { const error = await res.json(); - throw new Error(`Failed to write resource [${resourceId}] (${res.statusText}): ${JSON.stringify(error)}`); + throw new Error(`Failed to write resources [${Object.keys(resources).join(', ')}] (${res.statusText}): ${JSON.stringify(error)}`); } - const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[]; + // Decode the response and return the resources + const decodedResources = await this.decodeResponse(res); - return text; + // Return the decoded resources + return decodedResources; } - // Send a POST /data/read request to the server - async read(resourceId: string) { - const url = `${this.url}/data/get`; - const body = toExtendedJson({ resourceId: [ resourceId ] }); - - const res = await fetch(url, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - ...SyncClient.signRequest(this.privateKey.toBytes(), '/data/get', body), - }, - body, - }); + /** + * Send a POST /data/read request to the server + * @param resourceIds - The resource IDs to read + * @returns The resources with the read values + */ + async readMany(resourceIds: string[]): Promise> { + // Send the resource read requests to the server + const res = await this.sendRequest('/data/get', { resourceId: resourceIds }); + // If the response is not ok, throw an error if (!res.ok) { - throw new Error(`Failed to read resource: ${res.statusText}`); + throw new Error(`Failed to read resources [${resourceIds.join(', ')}] (${res.statusText})`); } - const text = fromExtendedJson(await res.text()) as { blob: ArrayBuffer, signature: string, publicKey: string, timestamp: number }[]; + // Decode the response and return the resources + const decodedResources = await this.decodeResponse(res); - return text; + // Return the decoded resources + return decodedResources; } - // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list + /** + * Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list + * @param resourceId - The resource ID to add to the subscription list + */ async subscribe(resourceId: string) { this.subscriptions.add(resourceId); this.emit('subscriptions-updated', this.subscriptions); } - // Destroy the current SSE Session, then re-create it with the additional resource ID in the subscription list + /** + * Destroy the current SSE Session, then re-create it without the resource ID in the subscription list + * @param resourceId - The resource ID to remove from the subscription list + */ async unsubscribe(resourceId: string) { this.subscriptions.delete(resourceId); @@ -184,4 +206,49 @@ export class SSEClient extends SyncClient { async handleOnConnectionUpdated(value: 'connected' | 'disconnected') { this.emit('connection-updated', value); } + + /** + * Sends a request to the server + * @param path - The path to send the request to + * @param body - The body of the request + * @returns The response from the server + */ + private async sendRequest(path: string, body: unknown): Promise { + // Build the URL + const url = `${this.url}${path}`; + + // Format the body as Extended JSON + const formattedBody = toExtendedJson(body); + + // Send the request to the server + const res = await fetch(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + ...BaseSyncClient.signRequest(this.privateKey.toBytes(), path, body), + }, + body: formattedBody, + }); + + // Return the response + return res; + } + + /** + * Decodes the blobs on the response body to the type T + * @param response - The response to decode + * @returns The decoded resources + */ + private async decodeResponse(response: Response): Promise> { + // Decode the response from Extended JSON + const text = fromExtendedJson(await response.text()) + + // Parse the response as a record of resources + const resources = BaseSyncClient.resourcesResponseSchema.parse(text); + + // Decode the blobs and convert them from extended json to the type T + const decodedResources = BaseSyncClient.decodeResourcesResponse(resources); + + return decodedResources; + } } diff --git a/src/ws/client.ts b/src/ws/client.ts index a54c7bd..c4fec1f 100644 --- a/src/ws/client.ts +++ b/src/ws/client.ts @@ -2,7 +2,7 @@ import { ExponentialBackoff, fromExtendedJson, toExtendedJson } from '@xo-cash/u import type { PrivateKey } from '@xo-cash/primitives'; import { WsMessageSchema, WsSuccessResponseSchema, WsErrorResponseSchema, type WsMessage } from './types.js'; -import { SyncClient } from '../shared/client.js'; +import { BaseSyncClient, type ResourcesRecord } from '../shared/base-client.js'; /** A request sent over the sync server's WebSocket connection. */ type WsRequest = { @@ -38,13 +38,14 @@ export type WsClientOptions = { * * Heartbeats and request timeouts are not implemented. */ -export class WsClient extends SyncClient { +export class WsClient extends BaseSyncClient { private readonly url: string; private readonly privateKey: PrivateKey; private readonly pendingRequests = new Map(); private readonly messageListeners = new Set<(message: WsMessage) => void>(); private readonly errorListeners = new Set<(error: Error) => void>(); - + private readonly textEncoder = new TextEncoder(); + /** Resource IDs to (re)subscribe to whenever a socket opens. */ private readonly subscriptions = new Set(); @@ -167,19 +168,37 @@ export class WsClient extends SyncClient { return attempt; } - /** Read every stored instance of one resource. */ - async read(resourceId: string): Promise { - return this.request('/data/get', { resourceId: [resourceId] }); + async readMany(resourceIds: string[]): Promise> { + // Read the resources from the server + const response = await this.request('/data/get', { resourceId: resourceIds }); + + // Parse the response as a record of resources + const resources = BaseSyncClient.resourcesResponseSchema.parse(response); + + // Decode the resources and convert the blobs from Uint8Arrays to the type T + const decodedResources = BaseSyncClient.decodeResourcesResponse(resources); + + return decodedResources; } - /** - * Sign and write one resource, using the same wire format as the SSE client. - */ - async write(resourceId: string, value: Record): Promise { - const valueBytes = new TextEncoder().encode(toExtendedJson(value)); - const resource = SyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes); + async writeMany(resourcesToWrite: Record): Promise> { + // Sign the resources + const signedResources = Object.entries(resourcesToWrite).map(([resourceId, value]) => { + const valueStr = toExtendedJson(value); + const valueBytes = this.textEncoder.encode(valueStr); + return BaseSyncClient.signWriteRequest(this.privateKey.toBytes(), resourceId, valueBytes); + }); - return this.request('/data/write', { resources: [resource] }); + // Write the resources to the server + const response = await this.request('/data/write', { resources: signedResources }); + + // Parse the response as a record of resources + const resources = BaseSyncClient.resourcesResponseSchema.parse(response); + + // Decode the resources and convert the blobs from Uint8Arrays to the type T + const decodedResources = BaseSyncClient.decodeResourcesResponse(resources); + + return decodedResources; } /** @@ -188,6 +207,8 @@ export class WsClient extends SyncClient { * The server never replies to a successful subscribe (the route stays pending until the topic is * removed), so this resolves once the frame is sent. A failed subscribe comes back as an `error` * message that matches no pending request, which handleMessage() reports through emitError(). + * + * @param resourceId - The resource ID to subscribe to */ async subscribe(resourceId: string): Promise { await this.send('/data/subscribe', { resourceId: [resourceId] }); @@ -200,6 +221,7 @@ export class WsClient extends SyncClient { /** * Unsubscribe from one resource topic. + * @param resourceId - The resource ID to unsubscribe from */ async unsubscribe(resourceId: string): Promise { this.subscriptions.delete(resourceId); @@ -225,7 +247,12 @@ export class WsClient extends SyncClient { }); } - /** Connect if needed, then sign and send one request envelope using the shared Extended JSON codec. */ + /** + * Connect if needed, then sign and send one request envelope using the shared Extended JSON codec. + * @param path - The path of the request + * @param body - The body of the request + * @param id - The ID of the request + */ private async send(path: string, body: unknown, id: string = crypto.randomUUID()): Promise { const socket = await this.getWebSocket(); @@ -235,13 +262,17 @@ export class WsClient extends SyncClient { } // Signed after connecting: the server rejects signatures older than its timestamp window (5 minutes by default). - const headers = SyncClient.signRequest(this.privateKey.toBytes(), path, toExtendedJson(body)); + const headers = BaseSyncClient.signRequest(this.privateKey.toBytes(), path, toExtendedJson(body)); const request: WsRequest = { id, path, body, headers }; socket.send(toExtendedJson(request)); } - /** Resolve with the socket once it opens, or reject if the handshake fails. */ + /** + * Resolve with the socket once it opens, or reject if the handshake fails. + * @param socket - The WebSocket to wait for + * @returns The WebSocket + */ private static waitForOpen(socket: WebSocket): Promise { return new Promise((resolve, reject) => { socket.onopen = () => resolve(socket); @@ -249,7 +280,10 @@ export class WsClient extends SyncClient { }); } - /** Decode and route one frame received from the sync server. */ + /** + * Decode and route one frame received from the sync server. + * @param raw - The raw message received from the sync server + */ private handleMessage(raw: string): void { const { success, data: message } = WsMessageSchema.safeParse(fromExtendedJson(raw)); if (!success) { @@ -275,12 +309,15 @@ export class WsClient extends SyncClient { } if (message.type === 'instance-changed') { - const parsedMessage = SyncClient.messageSchema.safeParse(message.data); + const parsedMessage = BaseSyncClient.resourceInstanceSchema.safeParse(message.data); if (!parsedMessage.success) { return; } - this.emit('message', parsedMessage.data); + this.emit('resource-instance-updated', { + ...parsedMessage.data, + value: fromExtendedJson(new TextDecoder().decode(parsedMessage.data.blob)) as PayloadType, + }); } return; @@ -311,14 +348,20 @@ export class WsClient extends SyncClient { } } - // Helper to emit an error to all error listeners + /** + * Helper to emit an error to all error listeners + * @param error - The error to emit + */ private emitError(error: Error): void { for (const listener of this.errorListeners) { listener(error); } } - // Helper for when the client disconnects + /** + * Helper for when the client disconnects + * @param error - The error to reject the pending requests with + */ private rejectPendingRequests(error: Error): void { for (const pending of this.pendingRequests.values()) { pending.reject(error); @@ -327,7 +370,11 @@ export class WsClient extends SyncClient { this.pendingRequests.clear(); } - // Helper to convert an HTTP URL to a WebSocket URL + /** + * Helper to convert an HTTP URL to a WebSocket URL + * @param httpUrl - The HTTP URL to convert + * @returns The WebSocket URL + */ static httpToWsUrl(httpUrl: string): string { const url = new URL(httpUrl);