diff --git a/src/client/page-app.jsx b/src/client/page-app.jsx index 30fefa4..1497943 100644 --- a/src/client/page-app.jsx +++ b/src/client/page-app.jsx @@ -1,5 +1,8 @@ import './page-app.css' +/** + * @returns {preact.JSX.Element} app wrapper component + */ export function App() { return ( <> diff --git a/src/common/async/aborts.js b/src/common/async/aborts.js index 8b1c4f8..d5de502 100644 --- a/src/common/async/aborts.js +++ b/src/common/async/aborts.js @@ -32,7 +32,7 @@ export function timeoutSignal(ms) { * Create an abort signal that aborts if any of the passed signals aborts. * Better than managing them manually because we do proper cleanup of non-triggered aborts. * - * @param {...AbortSignal} signals - the signals to combine + * @param {...(AbortSignal | undefined)} signals - the signals to combine * @returns {AbortSignal} the combined signal */ export function combineSignals(...signals) { @@ -41,6 +41,9 @@ export function combineSignals(...signals) { const cleanups = [] for (const signal of signals) { + if (!signal) + continue + if (signal.aborted) { controller.abort(signal.reason) return controller.signal diff --git a/src/common/async/blocking-atom.js b/src/common/async/blocking-atom.js index c7728b6..8fe05ae 100644 --- a/src/common/async/blocking-atom.js +++ b/src/common/async/blocking-atom.js @@ -47,7 +47,6 @@ export class BlockingAtom { } signal?.throwIfAborted() - return undefined } } diff --git a/src/common/async/blocking-queue.js b/src/common/async/blocking-queue.js index 68a0b73..1568c58 100644 --- a/src/common/async/blocking-queue.js +++ b/src/common/async/blocking-queue.js @@ -25,6 +25,27 @@ export class BlockingQueue { this.#maxsize = maxsize ? maxsize : undefined } + /** @returns {number} how deep is the queue? */ + get depth() { + return this.#items.length + } + + /** + * place one or more items on the queue, to be picked up by awaiters. + * + * @param {...T} elements the items to place on the queue. + */ + prequeue(...elements) { + for (const el of elements.reverse()) { + if (this.#maxsize && this.#items.length >= this.#maxsize) { + throw Error('out of room') + } + + this.#items.unshift(el) + this.#sema.free() + } + } + /** * place one or more items on the queue, to be picked up by awaiters. * diff --git a/src/common/async/gate.js b/src/common/breaker.js similarity index 100% rename from src/common/async/gate.js rename to src/common/breaker.js diff --git a/src/common/crypto/jwks.js b/src/common/crypto/jwks.js new file mode 100644 index 0000000..0e802f3 --- /dev/null +++ b/src/common/crypto/jwks.js @@ -0,0 +1,147 @@ +/** @module common/crypto */ + +import * as jose from 'jose' +import { z } from 'zod/v4' + +const signAlgo = { name: 'ES256' } + +const jwkBaseSchema = z.object({ + 'alg': z.string().optional(), + 'ext': z.boolean().optional(), + 'key_ops': z.array(z.string()).optional(), + 'kid': z.string().optional(), + 'use': z.string().optional(), + 'x5c': z.array(z.string()).optional(), + 'x5t#S256': z.string().optional(), + 'x5t': z.string().optional(), + 'x5u': z.string().optional(), +}) + +const jwkOkpPublicSchema = z.object({ + ...jwkBaseSchema.shape, + crv: z.string(), + x: z.string(), +}) + +const jwkOkpPrivateSchema = z.object({ + ...jwkOkpPublicSchema.shape, + d: z.string(), +}) + +const jwkEcPublicSchema = z.object({ + ...jwkBaseSchema.shape, + crv: z.string(), + x: z.string(), + y: z.string(), +}) + +const jwkEcPrivateSchema = z.object({ + ...jwkEcPublicSchema.shape, + d: z.string(), +}) + +const jwkRSAPublicSchema = z.object({ + ...jwkBaseSchema.shape, + e: z.string(), + n: z.string(), +}) + +const jwkRSAPrivateSchema = z.object({ + ...jwkRSAPublicSchema.shape, + d: z.string(), + dp: z.string(), + qp: z.string(), + p: z.string(), + q: z.string(), + qi: z.string(), +}) + +const jwkOctSchema = z.object({ + ...jwkBaseSchema.shape, + k: z.string(), +}) + +/** + * a zod schema describing a JWK from jose + * EC, OKP, RSA and oct key types are supported + * + * @see https://www.rfc-editor.org/rfc/rfc7517 + * @see https://github.com/panva/jose/blob/main/src/types.d.ts#L2 + * @type {z.ZodType} + */ +export const jwkSchema = z.union([ + jwkBaseSchema, + jwkOkpPublicSchema, + jwkOkpPrivateSchema, + jwkEcPublicSchema, + jwkEcPrivateSchema, + jwkRSAPublicSchema, + jwkRSAPrivateSchema, + jwkOctSchema, +]) + +/** + * zod transform from JWK to CryptoKey + * + * @type {z.ZodTransform} + */ +export const jwkImport = z.transform(async (val, ctx) => { + try { + if (typeof val === 'object' && val !== null) { + const key = await jose.importJWK(val, signAlgo.name) + if (key instanceof CryptoKey) { + return key + } + + ctx.issues.push({ + code: 'custom', + message: 'symmetric keys unsupported', + input: val, + }) + } + else { + ctx.issues.push({ + code: 'custom', + message: 'not a valid JWK object', + input: val, + }) + } + } + catch (e) { + ctx.issues.push({ + code: 'custom', + message: `could not import JWK object: ${e}`, + input: val, + }) + } + + return z.NEVER +}) + +/** + * zod transform from exportable CryptoKey to JWK + * + * @type {z.ZodTransform} + */ +export const jwkExport = z.transform(async (val, ctx) => { + try { + if (val.extractable) { + return await jose.exportJWK(val) + } + + ctx.issues.push({ + code: 'custom', + message: 'non-extractable key!', + input: val, + }) + } + catch (e) { + ctx.issues.push({ + code: 'custom', + message: `could not export JWK object: ${e}`, + input: val, + }) + } + + return z.NEVER +}) diff --git a/src/common/crypto/signing.js b/src/common/crypto/signing.js index 1a37d18..ea0ca34 100644 --- a/src/common/crypto/signing.js +++ b/src/common/crypto/signing.js @@ -3,12 +3,10 @@ import * as jose from 'jose' import { z } from 'zod/v4' -const signAlgo = { name: 'ES256' } - /** - * @typedef JWTToken + * @typedef {object} JWTToken * @property {string} token the still-encoded JWT, for later verification - * @augments jose.JWTPayload + * @property {jose.JWTPayload} payload the decoded JWT payload, for later verification * * A JWTToken is both the decoded payload and the token itself, for later processing. */ @@ -22,7 +20,7 @@ const signAlgo = { name: 'ES256' } export const jwtSchema = z.jwt({ abort: true }).transform((token, ctx) => { try { const payload = jose.decodeJwt(token) - return { ...payload, token } + return { payload, token } } catch (e) { ctx.issues.push({ @@ -35,72 +33,6 @@ export const jwtSchema = z.jwt({ abort: true }).transform((token, ctx) => { } }) -/** - * schema describing a transform from JWK to CryptoKey - * - * @type {z.ZodTransform} - */ -export const jwkImport = z.transform(async (val, ctx) => { - try { - if (typeof val === 'object' && val !== null) { - const key = await jose.importJWK(val, signAlgo.name) - if (key instanceof CryptoKey) { - return key - } - - ctx.issues.push({ - code: 'custom', - message: 'symmetric keys unsupported', - input: val, - }) - } - else { - ctx.issues.push({ - code: 'custom', - message: 'not a valid JWK object', - input: val, - }) - } - } - catch (e) { - ctx.issues.push({ - code: 'custom', - message: `could not import JWK object: ${e}`, - input: val, - }) - } - - return z.NEVER -}) - -/** - * schema describing a transform from exportable CryptoKey to JWK - * - * @type {z.ZodTransform} - */ -export const jwkExport = z.transform(async (val, ctx) => { - try { - if (val.extractable) { - return await jose.exportJWK(val) - } - - ctx.issues.push({ - code: 'custom', - message: 'non-extractable key!', - input: val, - }) - } - catch (e) { - ctx.issues.push({ - code: 'custom', - message: `could not export JWK object: ${e}`, - input: val, - }) - } - - return z.NEVER -}) - /** * generate a fingerprint for the given crypto key * diff --git a/src/common/errors.js b/src/common/errors.js new file mode 100644 index 0000000..80c2323 --- /dev/null +++ b/src/common/errors.js @@ -0,0 +1,71 @@ +/** @module common */ + +import { prettifyError, ZodError } from 'zod/v4' + +/** + * @private + * @type {Record} + */ +const StatusCodes = { + 400: 'Bad Request', + 403: 'Forbidden', + 404: 'Not Found', + 408: 'Request Timeout', + 409: 'Conflict', + 429: 'Too Many Requests', + 499: 'Client Closed Request', + 500: 'Internal Server Error', +} + +/** + * Common base class for Websocket Errors + * + * @augments Error + */ +export class ProtocolError extends Error { + + /** + * @param {string} message a "details" message representing this error + * @param {number} status the HTTP status code representing this error + * @param {Error} [cause] a previous error we're wrapping + */ + constructor(message, status, cause) { + const statusText = StatusCodes[status] || 'Unknown' + super(`${status} ${statusText}: ${message}`) + + this.name = this.constructor.name + this.status = status + this.cause = cause + } + +} + +/** + * @param {Error} e the error to check + * @param {number} [status] the protocol status to check, or any protocol error if undefined. + * @returns {boolean} true if the given error is a protocol error with the given status. + */ +export function isProtocolError(e, status) { + return (e instanceof ProtocolError && e.status == status) +} + +/** + * Normalizes the given error into a protocol error; passes through input that is already + * protocol errors. + * + * @param {unknown} error some caught error + * @returns {ProtocolError} an apporpriate protocol error + */ +export function normalizeProtocolError(error) { + if (error instanceof ProtocolError) + return error + + if (error instanceof ZodError) + return new ProtocolError(prettifyError(error), 400, error) + + if (error instanceof Error) + return new ProtocolError(error.message, 500, error) + + // fallback, unknown + return new ProtocolError(`Error! ${error}`, 500) +} diff --git a/src/common/protocol.js b/src/common/protocol.js new file mode 100644 index 0000000..f338ce6 --- /dev/null +++ b/src/common/protocol.js @@ -0,0 +1,46 @@ +/** @module common/protocol */ + +import { z } from 'zod/v4' +import { jwkSchema } from '@/common/crypto/jwks.js' +import { Brand } from '@/common/schema/brand.js' + +export const IdentID = new Brand('ident') +export const RealmID = new Brand('realm') + +/** zod schema for `preauth.authn` message */ +export const preauthAuthnMessageSchema = z.object({ + msg: z.literal('preauth.authn'), + pubkey: jwkSchema, +}) + +/** zod schema for any `preauth` messages */ +export const preauthMessageSchema = z.discriminatedUnion('msg', [ + preauthAuthnMessageSchema, +]) + +//// + +/** zod schema for `realm.status` message */ +export const realmStatusMessageSchema = z.object({ + msg: z.literal('realm.status'), +}) + +/** zod schema for `realm.status` response */ +export const realmStatusResponseSchema = z.object({ + msg: z.literal('realm.status'), + realm: RealmID.schema, + identities: z.array(IdentID.schema), +}) + +/** zod schema for `realm.broadcast` message */ +export const realmBroadcastMessageSchema = z.object({ + msg: z.literal('realm.broadcast'), + recipients: z.array(IdentID.schema), + payload: z.any(), +}) + +/** zod schema for any `realm` messages */ +export const realmMessageSchema = z.discriminatedUnion('msg', [ + realmStatusMessageSchema, + realmBroadcastMessageSchema, +]) diff --git a/src/common/schema/brand.js b/src/common/schema/brand.js new file mode 100644 index 0000000..2b90825 --- /dev/null +++ b/src/common/schema/brand.js @@ -0,0 +1,69 @@ +/** @module common/schema */ + +import { nanoid } from 'nanoid' +import { z } from 'zod/v4' + +/** + * A brand creates identifiers that are typesafe by construction, + * and shouldn't be able to be passed to the wrong resource type. + */ +export class Brand { + + /** + * @private + * @typedef {z.core.$ZodBranded} BrandSchema + * @typedef {z.infer} BrandId + */ + + /** @type {string} */ + #prefix + + /** @type {number} */ + #length + + /** @type {BrandSchema} */ + #schema + + /** + * @param {string} prefix string prefix for the identifier, eg. "ident", "acct") + * @param {number} [length] the length of the random identifier part + */ + constructor(prefix, length = 16) { + const brand = prefix.replace(/[.*+?^${}()|[\]\\]/g, '\\$&') // escape regex chars + const pattern = new RegExp(`^${brand}-[A-Za-z0-9_-]{${length.toString()}}$`) + + this.#prefix = brand + this.#length = length + this.#schema = z.string().regex(pattern).brand(Symbol(brand)) + } + + get schema() { + return this.#schema + } + + /** @returns {BrandId} a generated identifier */ + generate() { + const id = nanoid(this.#length) + return this.#schema.parse(`${this.#prefix}-${id}`) + } + + /** + * @param {unknown} input the input to validate + * @returns {BrandId} a valid identifier + * @throws {z.ZodError} if the input could not be parsed + */ + parse(input) { + return this.#schema.parse(input) + } + + /** + * @param {unknown} input the input to validate + * @returns {boolean} true if a valid identifier, false otherwise + */ + validate(input) { + return input != null + && typeof input === 'string' + && this.#schema.safeParse(input).success + } + +} diff --git a/src/common/socket.js b/src/common/socket.js new file mode 100644 index 0000000..54bfc44 --- /dev/null +++ b/src/common/socket.js @@ -0,0 +1,179 @@ +/** @module common/socket */ + +import { combineSignals } from '@/common/async/aborts.js' +import { BlockingAtom } from '@/common/async/blocking-atom.js' +import { BlockingQueue } from '@/common/async/blocking-queue.js' +import { Breaker } from '@/common/breaker.js' +import { ProtocolError } from '@/common/errors.js' + +/** + * Given a websocket, wait and take a single message off and return it. + * + * @example + * const ws = new WebSocket("wss://example.com/stream") + * const timeout = timeoutSignal(5000) + * + * try { + * const msg = await takeSocket(ws, timeout.signal) + * doWhatever(msg) + * } finally { + * timeout.cleanup() + * if (ws.readyState !== ws.CLOSED) + * ws.close(); + * } + * + * @param {WebSocket} ws the socket to read + * @param {AbortSignal} [signal] an abort signal to cancel the block + * @returns {Promise} the message off the socket + */ +export async function takeSocket(ws, signal) { + signal?.throwIfAborted() + + const atom = new BlockingAtom() + const breaker = new Breaker() + + const error = new AbortController() + const multisignal = combineSignals(error.signal, signal) + + const onMessage = breaker.tripThen(m => atom.set(m.data)) + const onError = breaker.tripThen(e => error.abort(e)) + const onClose = breaker.tripThen(() => error.abort('closed')) + + try { + ws.addEventListener('message', onMessage) + ws.addEventListener('error', onError) + ws.addEventListener('close', onClose) + + const data = await atom.get(multisignal) + if (!data) { + throw new ProtocolError('socket read aborted', 408, multisignal?.reason) + } + + return data + } + finally { + ws.removeEventListener('message', onMessage) + ws.removeEventListener('error', onError) + ws.removeEventListener('close', onClose) + } +} + +/** + * stream configuration options + * + * @typedef ConfigProps + * @property {number} maxDepth max depth of the queue, after which messages are dropped + * @property {AbortSignal} [signal] an abort signal to cancel the block + */ + +/** @type {ConfigProps} */ +export const STREAM_CONFIG_DEFAULT = { maxDepth: 1000 } + +// symbols for iteration protocol +const yield$ = Symbol('yield$') +const error$ = Symbol('error$') +const end$ = Symbol('end$') + +/** + * Given a websocket, stream messages off the socket as an async generator. + * + * @example + * const ws = new WebSocket("wss://example.com/stream") + * const timeout = timeoutSignal(5000) + * + * try { + * // signal will fire in 5s, so we'll take as many messages as we get until then + * for await (const msg of streamSocket(ws, { signal: timeout.signal })) { + * doWhatever(msg) + * } + * } finally { + * timeout.cleanup() + * if (ws.readyState !== ws.CLOSED) + * ws.close(); + * } + * + * + * + * @param {WebSocket} ws the socket to read + * @param {Partial} [config_] stream configuration to merge into defaults + * @yields {unknown} messages from the socket + */ +export async function* streamSocket(ws, config_) { + const { signal, ...config } = { ...STREAM_CONFIG_DEFAULT, ...config_ } + signal?.throwIfAborted() + + // await incoming messages without blocking + /** @type {BlockingQueue>} */ + const queue = new BlockingQueue(config.maxDepth) + + // if true, we're ignoring incoming messages until we drop the queue + let inBackoffMode = false + const backoffThresh = Math.floor(config.maxDepth * 0.9) + + // we don't want to keep processing after we've been closed + const breaker = new Breaker() + + // define callback functions (need to be able to reference for removing them) + + /** @type {function(MessageEvent): void} */ + const onMessage = breaker.untilTripped((m) => { + if (inBackoffMode) { + console.warn('ignoring incoming message due to backpressure protection!') + return + } + + queue.enqueue([yield$, m.data]) + inBackoffMode = queue.depth > backoffThresh + if (inBackoffMode) { + console.warn('message stream will start dropping messages due to backpressure!') + } + }) + + /** @type {function(Event): void} */ + // todo: why are we getting this on client shutdown instead of onClose? + const onError = breaker.tripThen((e) => { + if (e.message === 'Unexpected EOF') { + queue.enqueue([end$]) + } + else { + queue.enqueue([error$, e]) + } + }) + + /** @type {function(): void} */ + const onClose = breaker.tripThen(() => { + queue.enqueue([end$]) + }) + + // finally get into our loop + try { + ws.addEventListener('message', onMessage) + ws.addEventListener('error', onError) + ws.addEventListener('close', onClose) + + while (true) { + signal?.throwIfAborted() + + const [event, value] = await queue.dequeue(signal) + if (queue.depth < backoffThresh) { + console.log('message stream will stop dropping messages due to eased backpressure') + inBackoffMode = false + } + + switch (event) { + case yield$: + yield value + continue + case error$: + throw value + case end$: + return + } + } + } + finally { + ws.removeEventListener('message', onMessage) + ws.removeEventListener('error', onError) + ws.removeEventListener('close', onClose) + } +} diff --git a/src/common.js b/src/types.js similarity index 73% rename from src/common.js rename to src/types.js index aaa54cf..3e8d82a 100644 --- a/src/common.js +++ b/src/types.js @@ -1,15 +1,13 @@ -/** - * useful types and utilities that don't belong elsewhere. - * - * @module common - */ +/** @module common */ /** - * @typedef {function(...*): void} Callback * A callback function, with arbitrary arguments; use {Parameters} to extract them. + * + * @typedef {function(...*): void} Callback */ /** - * @typedef {function(): void} VoidCallback * A callback function with no arguments. + * + * @typedef {function(): void} VoidCallback */