From f94158fbefcc1eedad5994761b0a45498c331bee Mon Sep 17 00:00:00 2001 From: Jonathan Raphaelson Date: Fri, 20 Jun 2025 16:02:01 -0600 Subject: [PATCH] revamped socket protocol handling --- eslint.config.js | 2 + readme.org | 4 +- src/client/components/peer-list.tsx | 5 +- src/client/realm/connection.ts | 445 +++++++++----------- src/client/realm/context.tsx | 12 +- src/client/realm/peer.ts | 25 ++ src/client/realm/types.ts | 17 + src/common/protocol.ts | 20 +- src/common/protocol/messages-preauth.ts | 22 - src/common/protocol/messages-realm.ts | 65 --- src/common/protocol/messages.ts | 88 +++- src/common/protocol/schema.ts | 70 +++ src/common/socket.ts | 5 - src/common/types.ts | 1 + src/server/index.ts | 10 +- src/server/routes-error.ts | 1 - src/server/routes-socket/handler-preauth.ts | 30 +- src/server/routes-socket/handler-realm.ts | 60 ++- src/server/routes-socket/handler.ts | 2 - 19 files changed, 470 insertions(+), 414 deletions(-) create mode 100644 src/client/realm/peer.ts create mode 100644 src/client/realm/types.ts delete mode 100644 src/common/protocol/messages-preauth.ts delete mode 100644 src/common/protocol/messages-realm.ts create mode 100644 src/common/protocol/schema.ts diff --git a/eslint.config.js b/eslint.config.js index 52c3cfe..d29e91f 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -50,6 +50,8 @@ export default tseslint.config( '@typescript-eslint/no-unused-vars': ['warn', {varsIgnorePattern: '(?:^_)'}], '@typescript-eslint/no-unnecessary-condition': 'off', '@typescript-eslint/restrict-template-expressions': 'off', + '@typescript-eslint/no-unnecessary-type-parameters': 'off', + '@typescript-eslint/no-unnecessary-type-constraint': 'off', 'tsdoc/syntax': 'warn', }, }, diff --git a/readme.org b/readme.org index 3fef397..6a17551 100644 --- a/readme.org +++ b/readme.org @@ -22,8 +22,8 @@ An offline first RSS & Podcast PWA Application Run the pwa+server with: #+BEGIN_SRC bash - $ pnpm install - $ pnpm run dev # lots of stuff concurrently with wireit + $ npm install + $ npm run dev # lots of stuff concurrently with wireit #+END_SRC - Common diff --git a/src/client/components/peer-list.tsx b/src/client/components/peer-list.tsx index 3c5f22f..24bbab6 100644 --- a/src/client/components/peer-list.tsx +++ b/src/client/components/peer-list.tsx @@ -1,6 +1,7 @@ import {useEffect, useState} from 'preact/hooks' -import {PeerState, RealmConnection} from '#client/realm/connection' +import {RealmConnection} from '#client/realm/connection' +import {PeerState} from '#client/realm/types' import {IdentID} from '#common/protocol' export const PeerList: preact.FunctionComponent<{webrtcManager: RealmConnection}> = (props) => { @@ -13,7 +14,7 @@ export const PeerList: preact.FunctionComponent<{webrtcManager: RealmConnection} const updatePeers = () => { queueMicrotask(() => { const states = webrtcManager.getPeerStates() - console.log('updating peers', states) + console.debug('updating peers', states) setPeers(states) }) } diff --git a/src/client/realm/connection.ts b/src/client/realm/connection.ts index d1f705d..5a633db 100644 --- a/src/client/realm/connection.ts +++ b/src/client/realm/connection.ts @@ -1,52 +1,43 @@ import WebSocket from 'isomorphic-ws' import {nanoid} from 'nanoid' -import SimplePeer from 'simple-peer' +import SimplePeer, {SimplePeerData} from 'simple-peer' import {z} from 'zod/v4' import {generateSignableJwt, jwkExport} from '#common/crypto/jwks' import {normalizeError, normalizeProtocolError, ProtocolError} from '#common/errors' import { IdentID, - parseJson, - PreauthRegisterMessage, - RealmBroadcastMessage, - realmFromServerMessageSchema, - RealmID, - RealmRtcPeerWelcomeMessage, - realmRtcPeerWelcomeMessageSchema, - RealmRtcSignalMessage, + PreauthRegisterRequest, + preauthRespSchema, + realmRtcPeerJoinedEventSchema, + realmRtcPeerLeftEventSchema, + realmRtcPingRequestSchema, + RealmRtcPongResponse, + realmRtcPongResponseSchema, + RealmRtcSignalEvent, + realmRtcSignalEventSchema, } from '#common/protocol' -import {sendSocket, streamSocketJson, takeSocketJson} from '#common/socket' +import {streamSocketJson, takeSocketJson} from '#common/socket' +import {RealmPeer} from './peer' +import {ConnectionIdentity, PeerState} from './types' -/** the state of a specific peer */ -export interface PeerState { - /** if the peer connection is active */ - connected: boolean - - /** if the peer connection has been destroyed */ - destroyed: boolean - - /** the peer's address (ip and port) */ - address: ReturnType -} - -/** identity info for connecting to a realm */ -export interface RealmIdentity { - realmid: RealmID - identid: IdentID - keypair: CryptoKeyPair -} +const realmRtcMessagesSchema = z.union([ + realmRtcSignalEventSchema, + realmRtcPeerJoinedEventSchema, + realmRtcPeerLeftEventSchema, + realmRtcPongResponseSchema, +]) /** manages websocket and webrtc connections for a realm */ export class RealmConnection extends EventTarget { #url: string - #identity: RealmIdentity + #identity: ConnectionIdentity #socket: WebSocket - #peers: Map + #peers: Map #nonces: Map - constructor(url: string, identity: RealmIdentity) { + constructor(url: string, identity: ConnectionIdentity) { super() this.#url = url @@ -61,170 +52,136 @@ export class RealmConnection extends EventTarget { this.#socket.onerror = this.#handleSocketError } - #handleSocketOpen: WebSocket['onopen'] = async () => { - if (this.#socket == undefined) throw new Error('socket open handler called with no socket?') - - try { - console.debug('realm connection, socket loop open') - this.#dispatchCustomEvent('wsopen') - - // do the auth dance - // TODO: this should be a state machine - - const pubkey = await jwkExport.parseAsync(this.#identity.keypair.publicKey) - this.#socket.send( - await this.#signJwt({msg: 'preauth.register', pubkey} as PreauthRegisterMessage), - ) - - // the next message should be a welcome message - // this will throw (authenticated) otherwise - - let welcome: RealmRtcPeerWelcomeMessage - try { - welcome = await takeSocketJson(this.#socket, realmRtcPeerWelcomeMessageSchema) - } catch (exc) { - const err = normalizeError(exc) - throw new ProtocolError('failure on authentication', 401, {cause: err}) - } - - this.#dispatchCustomEvent('wsauth', {welcome}) - - // we initiate connections outbound when starting up - // we never initiate connections to other peers otherwise - - for (const peerid of welcome.peers) { - if (peerid === this.#identity.identid) continue - this.connectToPeer(peerid, true) - } - - // then continue looping over messages on the socket and handle them + get connected() { + return this.#socket.readyState === this.#socket.OPEN + } - for await (const data of streamSocketJson(this.#socket)) { - const parse = realmFromServerMessageSchema.safeParse(data) + send(identid: IdentID, data: T) { + const peer = this.#peers.get(identid) + if (!peer?.connected) throw new Error(`Not connected to peer: ${identid}`) - // not a known rtc specific message, so we don't capture it - if (!parse.success) { - this.#dispatchCustomEvent('wsdata', {data}) - continue - } + peer.send(JSON.stringify(data)) + } - // otherwise, we handle it - switch (parse.data.msg) { - case 'realm.rtc.peer-joined': - // new peers connect to us, we don't connect to them - this.#dispatchCustomEvent('peerjoined', {identid: parse.data.identid}) - continue - - case 'realm.rtc.peer-left': - // peer is gone, disconnect - this.disconnectPeer(parse.data.identid) - this.#dispatchCustomEvent('peerleft', {identid: parse.data.identid}) - continue - - case 'realm.rtc.signal': { - // some other peer is trying to send us some rtc data - // only connect if _they_ are the initiator - our initiating conns were above - let peer = this.#peers.get(parse.data.sender) - if (!peer && parse.data.initiator) { - peer = this.connectToPeer(parse.data.sender, false) - } - - // may not have a connection yet if we're waiting for them to answer - if (peer) { - peer.signal(parse.data.payload) - } - - continue - } - } - } - } catch (exc) { - const err = normalizeProtocolError(exc) + sendRaw(identid: IdentID, data: SimplePeerData) { + const peer = this.#peers.get(identid) + if (!peer?.connected) throw new Error(`Not connected to peer: ${identid}`) - console.error('realm connection, socket loop error', err) - this.#dispatchCustomEvent('wserror', {error: err}) - } finally { - console.debug('realm connection, socket loop ended') - this.destroy() - } + peer.send(data) } - #handleSocketError: WebSocket['onerror'] = (exc) => { - this.#dispatchCustomEvent('wserror', {error: normalizeProtocolError(exc)}) - this.destroy() + broadcast(data: unknown, self = false) { + this.#peers.forEach((peer, identid) => { + if (self || identid !== this.#identity.identid) peer.send(JSON.stringify(data)) + }) } - #handleSocketClose: WebSocket['onclose'] = () => { - this.#dispatchCustomEvent('wsclose') - this.destroy() - } + getPeerStates(): Record { + const states: Record = {} + for (const [identid, peer] of this.#peers) { + states[identid] = { + identid, + address: peer.address(), + connected: peer.connected, + destroyed: peer.destroyed, + } + } - get connected() { - return this.#socket.readyState === this.#socket.OPEN + return states } destroy() { - console.debug('realm connection #destroy!') + console.debug('realm connection destroy!') if (this.connected) this.#socket.close() - for (const peer of this.#peers.values()) peer.destroy() this.#peers.clear() this.#nonces.clear() } - #dispatchCustomEvent(type: string, detail?: object) { - this.dispatchEvent(new CustomEvent(type, {detail})) - } + // do the auth dance + // TODO: this should be a state machine + #connectionAuthenticate = async () => { + try { + const pubkey = await jwkExport.parseAsync(this.#identity.keypair.publicKey) + await this.#socketSignedWrite({ + typ: 'req', + msg: 'preauth.register', + dat: {pubkey}, + }) - /** generates and signs a JWT scoped to this identity/realm containing the given payload */ - async #signJwt(payload: object): Promise { - return await generateSignableJwt({ - aud: this.#identity.realmid, - iss: this.#identity.identid, - payload, - }).sign(this.#identity.keypair.privateKey) - } + const resp = await takeSocketJson(this.#socket, preauthRespSchema) + this.#dispatchCustomEvent('wsauth', resp) - connectToPeer(remoteid: IdentID, initiator: boolean): SimplePeer.Instance { - let peer = this.#peers.get(remoteid) - if (peer) { - console.log(`already connected to ${remoteid}`) - return peer + // we initiate connections outbound when starting up + // we never initiate connections to other peers otherwise + for (const peerid of resp.dat.activePeers) { + if (peerid === this.#identity.identid) continue + this.#connectPeer(peerid, true) + } + } catch (exc) { + const err = normalizeError(exc) + throw new ProtocolError('failure on authentication', 401, {cause: err}) } + } - peer = new RealmConnectionPeer(this, nanoid(), this.#identity.identid, remoteid, initiator) - - peer.on('connect', () => { - console.log(`connected to ${remoteid}`) + // handle some message coming in on the _websocket_ + #connectionMessage = (data: unknown) => { + const parse = realmRtcMessagesSchema.safeParse(data) + if (!parse.success) { + // publish non-handled data to the listeners + this.#dispatchCustomEvent('wsdata', {data}) + return + } - this.#dispatchCustomEvent('peeropen', {remoteid}) - }) + switch (parse.data.msg) { + case 'realm.rtc.pong': + console.debug('got a pong response', parse) + return - peer.on('close', () => { - console.log(`Disconnected from ${remoteid}`) + case 'realm.rtc.signal': { + // only connect if _they_ are the initiator - our initiating conns were above + let peer = this.#peers.get(parse.data.dat.localid) + if (!peer && parse.data.dat.initiator) { + peer = this.#connectPeer(parse.data.dat.localid, false) + } - this.#peers.delete(remoteid) - this.#nonces.delete(remoteid) - this.#dispatchCustomEvent('peerclose', {remoteid}) - }) + // may not have a connection yet if we're waiting for them to answer + peer?.signal(parse.data.dat.payload) + return + } - peer.on('error', (err) => { - console.error(`Error with peer ${remoteid}:`, err) + case 'realm.rtc.peer-joined': + // new peers connect to us, we don't connect to them + this.#dispatchCustomEvent('peerjoined', {identid: parse.data.dat.identid}) + return - this.#dispatchCustomEvent('peererror', {remoteid, error: err}) - }) + case 'realm.rtc.peer-left': + // peer is gone, disconnect + this.#disconnectPeer(parse.data.dat.identid) + this.#dispatchCustomEvent('peerleft', {identid: parse.data.dat.identid}) + return + } + } - peer.on('message', (data: unknown) => { - this.#dispatchCustomEvent('peerdata', {remoteid, data}) - }) + #connectPeer(remoteid: IdentID, initiator: boolean): RealmPeer { + let peer = this.#peers.get(remoteid) + if (!peer) { + peer = new RealmPeer(remoteid, nanoid(), initiator) + peer.on('connect', this.#handlePeerConnect.bind(this, peer)) + peer.on('close', this.#handlePeerClose.bind(this, peer)) + peer.on('error', this.#handlePeerError.bind(this, peer)) + peer.on('data', this.#handlePeerData.bind(this, peer)) + peer.on('signal', this.#handlePeerSignal.bind(this, peer)) + + this.#peers.set(remoteid, peer) + } - this.#peers.set(remoteid, peer) return peer } - disconnectPeer(identid: IdentID) { + #disconnectPeer(identid: IdentID) { const peer = this.#peers.get(identid) if (peer) { peer.destroy() @@ -233,111 +190,105 @@ export class RealmConnection extends EventTarget { } } - sendToPeer(identid: IdentID, data: unknown) { - const peer = this.#peers.get(identid) - if (peer && peer.connected) { - peer.send(JSON.stringify(data)) - } else { - throw new Error(`Not connected to peer: ${identid}`) - } - } + // handle socket open is the main loop for the sendSocket + #handleSocketOpen: WebSocket['onopen'] = async () => { + if (this.#socket == undefined) throw new Error('socket open handler called with no socket?') - sendToServer(data: unknown) { - sendSocket(this.#socket, data) - } + console.debug('realm connection, socket loop open') + this.#dispatchCustomEvent('wsopen') - broadcast(data: unknown) { - const message = JSON.stringify(data) - for (const [_, peer] of this.#peers) { - if (peer.connected) { - peer.send(message) + try { + await this.#connectionAuthenticate() + for await (const data of streamSocketJson(this.#socket)) { + this.#connectionMessage(data) } + } catch (exc) { + const err = normalizeProtocolError(exc) + + console.error('realm connection, socket loop error', err) + this.#dispatchCustomEvent('wserror', {error: err}) + } finally { + this.destroy() } } - broadcastViaServer(data: unknown) { - const resp: RealmBroadcastMessage = { - msg: 'realm.broadcast', - payload: data, - recipients: false, - } + #handleSocketError: WebSocket['onerror'] = (exc) => { + this.#dispatchCustomEvent('wserror', {error: normalizeProtocolError(exc)}) + this.destroy() + } - sendSocket(this.#socket, resp) + #handleSocketClose: WebSocket['onclose'] = () => { + this.#dispatchCustomEvent('wsclose') + this.destroy() } - getPeerStates(): Record { - const states: Record = {} - for (const [peerId, peer] of this.#peers) { - states[peerId] = { - address: peer.address(), - connected: peer.connected, - destroyed: peer.destroyed, - } + #handlePeerSignal = (peer: RealmPeer, payload: SimplePeer.SignalData) => { + const msg: RealmRtcSignalEvent = { + typ: 'evt', + msg: 'realm.rtc.signal', + dat: { + initiator: peer.initiator, + payload: JSON.stringify(payload), + localid: this.#identity.identid, + remoteid: peer.identid, + }, } - return states + this.#socket.send(JSON.stringify(msg)) } -} -const peerPingSchema = z.object({type: z.literal('ping'), timestamp: z.number()}) - -/** a single webrtc peer connection within a realm */ -export class RealmConnectionPeer extends SimplePeer { - #connection: RealmConnection - - initiator: boolean - localid: IdentID - remoteid: IdentID - nonce: string - - constructor( - connection: RealmConnection, - nonce: string, - localid: IdentID, - remoteid: IdentID, - initiator: boolean, - ) { - super({ - initiator, - config: { - iceServers: [ - {urls: 'stun:stun.l.google.com:19302'}, - {urls: 'stun:stun1.l.google.com:19302'}, - ], - }, - }) - - this.initiator = initiator - this.localid = localid - this.remoteid = remoteid - this.nonce = nonce + #handlePeerConnect = (peer: RealmPeer) => { + console.debug(`connected to ${peer.identid}`) + this.#dispatchCustomEvent('peeropen', {identid: peer.identid}) + } - this.#connection = connection + #handlePeerClose = (peer: RealmPeer) => { + console.debug(`disconnected from ${peer.identid}`) - this.on('signal', this.#handlePeerSignal) - this.on('data', this.#handlePeerData) + this.#peers.delete(peer.identid) + this.#nonces.delete(peer.identid) + this.#dispatchCustomEvent('peerclose', {identid: peer.identid}) } - #handlePeerSignal = (e: SimplePeer.SignalData) => { - this.#connection.sendToServer({ - msg: 'realm.rtc.signal', - initiator: this.initiator, - sender: this.localid, - recipient: this.remoteid, - payload: JSON.stringify(e), - } satisfies RealmRtcSignalMessage) + #handlePeerError = (peer: RealmPeer, err: Error) => { + console.error(`Error with peer ${peer.identid}:`, err) + this.#dispatchCustomEvent('peererror', {identid: peer.identid, error: err}) } - #handlePeerData = (chunk: string) => { - try { - const ping = parseJson.pipe(peerPingSchema).safeParse(chunk) - if (ping.success) { - this.send(JSON.stringify({type: 'pong', timestamp: ping.data.timestamp})) - } else { - this.emit('message', chunk) - } - } catch (err) { - console.error('Failed to parse message:', err) + #handlePeerData = (peer: RealmPeer, chunk: string | Uint8Array) => { + const data = typeof chunk === 'string' ? chunk : new TextDecoder().decode(chunk) + const parsed = realmRtcPingRequestSchema.safeParse(chunk) + if (parsed.success) { + // reply to pings with pongs + this.send(peer.identid, { + typ: 'res', + msg: 'realm.rtc.pong', + seq: parsed.data.seq, + }) + } else { + this.#dispatchCustomEvent('peerdata', {identid: peer.identid, data}) } } + + // helpers that are just too damn long + + #dispatchCustomEvent(type: string, detail?: object) { + this.dispatchEvent(new CustomEvent(type, {detail})) + } + + // typed helpers + // @example + // this.#socketWrite({ + // ... // <- get type errors here + // }) + + #socketSignedWrite = async (payload: T) => { + const token = await generateSignableJwt({ + aud: this.#identity.realmid, + iss: this.#identity.identid, + payload, + }).sign(this.#identity.keypair.privateKey) + + this.#socket.send(token) + } } diff --git a/src/client/realm/context.tsx b/src/client/realm/context.tsx index 8427399..6acaabc 100644 --- a/src/client/realm/context.tsx +++ b/src/client/realm/context.tsx @@ -1,12 +1,13 @@ import {createContext} from 'preact' import {useCallback, useEffect, useState} from 'preact/hooks' -import {RealmConnection, RealmIdentity} from '#client/realm/connection' +import {RealmConnection} from '#client/realm/connection' +import {ConnectionIdentity} from './types' interface RealmConnectionContext { realm?: RealmConnection - identity?: RealmIdentity - setIdentity: (ident: RealmIdentity) => void + identity?: ConnectionIdentity + setIdentity: (ident: ConnectionIdentity) => void } export const RealmConnectionContext = createContext(null) @@ -15,7 +16,7 @@ export const RealmConnectionProvider: preact.FunctionComponent<{ url: string children: preact.ComponentChildren }> = (props) => { - const [identity$, setIdentity$] = useState() + const [identity$, setIdentity$] = useState() const [connection$, setConnection$] = useState() const connect = useCallback(() => { @@ -29,9 +30,8 @@ export const RealmConnectionProvider: preact.FunctionComponent<{ connection$?.destroy() }, [connection$]) + // connect on mount, or identity change useEffect(() => { - console.log('use effect in provider', identity$) - // connect on mount, or identity change if (identity$) connect() // disconnect on unsubscribe diff --git a/src/client/realm/peer.ts b/src/client/realm/peer.ts new file mode 100644 index 0000000..52ba3ce --- /dev/null +++ b/src/client/realm/peer.ts @@ -0,0 +1,25 @@ +import {IdentID} from '#common/protocol.js' +import SimplePeer from 'simple-peer' + +/** a single webrtc peer connection within a realm */ +export class RealmPeer extends SimplePeer { + identid: IdentID + nonce: string + initiator: boolean + + constructor(identid: IdentID, nonce: string, initiator: boolean) { + super({ + initiator, + config: { + iceServers: [ + {urls: 'stun:stun.l.google.com:19302'}, + {urls: 'stun:stun1.l.google.com:19302'}, + ], + }, + }) + + this.identid = identid + this.nonce = nonce + this.initiator = initiator + } +} diff --git a/src/client/realm/types.ts b/src/client/realm/types.ts new file mode 100644 index 0000000..01b3e85 --- /dev/null +++ b/src/client/realm/types.ts @@ -0,0 +1,17 @@ +import {IdentID, RealmID} from '#common/protocol' +import SimplePeer from 'simple-peer' + +/** identity info for connecting to a realm */ +export interface ConnectionIdentity { + realmid: RealmID + identid: IdentID + keypair: CryptoKeyPair +} + +/** the state of a specific peer */ +export interface PeerState { + identid: IdentID + connected: boolean + destroyed: boolean + address: ReturnType +} diff --git a/src/common/protocol.ts b/src/common/protocol.ts index e9753ff..010636f 100644 --- a/src/common/protocol.ts +++ b/src/common/protocol.ts @@ -1,9 +1,9 @@ import {z} from 'zod/v4' +import {ProtocolError} from './errors' +import {errorMessageSchema} from './protocol/schema' export * from './protocol/brands' export * from './protocol/messages' -export * from './protocol/messages-preauth' -export * from './protocol/messages-realm' /** a zod transformer for parsing json */ export const parseJson = z.transform((input, ctx) => { @@ -19,3 +19,19 @@ export const parseJson = z.transform((input, ctx) => { return z.NEVER } }) + +export function makeError( + error: ProtocolError, + detail: string, + seq?: number, +): z.infer { + return { + typ: 'err', + msg: error.message, + seq, + dat: { + code: error.status, + detail, + }, + } +} diff --git a/src/common/protocol/messages-preauth.ts b/src/common/protocol/messages-preauth.ts deleted file mode 100644 index 8c2a0f4..0000000 --- a/src/common/protocol/messages-preauth.ts +++ /dev/null @@ -1,22 +0,0 @@ -import {jwkSchema} from '#common/crypto/jwks' -import {z} from 'zod/v4' - -/** zod schema for `preauth.authn` message */ -export const preauthRegisterMessageSchema = z.object({ - msg: z.literal('preauth.register'), - pubkey: jwkSchema, -}) -export type PreauthRegisterMessage = z.infer - -/** zod schema for `preauth.authn` message */ -export const preauthAuthnMessageSchema = z.object({ - msg: z.literal('preauth.authn'), -}) -export type PreauthAuthnMessage = z.infer - -/** zod schema for any `preauth` messages */ -export const preauthMessageSchema = z.discriminatedUnion('msg', [ - preauthRegisterMessageSchema, - preauthAuthnMessageSchema, -]) -export type PreauthMessage = z.infer diff --git a/src/common/protocol/messages-realm.ts b/src/common/protocol/messages-realm.ts deleted file mode 100644 index 547927a..0000000 --- a/src/common/protocol/messages-realm.ts +++ /dev/null @@ -1,65 +0,0 @@ -import {z} from 'zod/v4' - -import {IdentBrand} from './brands' -import {responseOkSchema} from './messages' - -/** - * zod schema for `realm.broadcast` message - * - * recipients = true, include self in broadcast - * recipients = false, exclude self in broadcast (default) - * recipients = [], use these exact recipients - */ -export const realmBroadcastMessageSchema = z.object({ - msg: z.literal('realm.broadcast'), - payload: z.any(), - recipients: z.union([z.boolean(), z.array(IdentBrand.schema)]).default(false), -}) - -export type RealmBroadcastMessage = z.infer - -// rtc messages - -export const realmRtcSignalMessageSchema = z.object({ - msg: z.literal('realm.rtc.signal'), - payload: z.string(), - sender: IdentBrand.schema, - recipient: IdentBrand.schema, - initiator: z.boolean(), -}) - -export const realmRtcPeerWelcomeMessageSchema = responseOkSchema.extend({ - msg: z.literal('realm.rtc.peer-welcome'), - peers: z.array(IdentBrand.schema), -}) - -export const realmRtcPeerJoinedMessageSchema = responseOkSchema.extend({ - msg: z.literal('realm.rtc.peer-joined'), - identid: IdentBrand.schema, -}) - -export const realmRtcPeerLeftMessageSchema = responseOkSchema.extend({ - msg: z.literal('realm.rtc.peer-left'), - identid: IdentBrand.schema, -}) - -export type RealmRtcSignalMessage = z.infer -export type RealmRtcPeerWelcomeMessage = z.infer -export type RealmRtcPeerJoinedMessage = z.infer -export type RealmRtcPeerLeftMessage = z.infer - -/// useful unions - -export const realmToServerMessageSchema = z.discriminatedUnion('msg', [ - realmBroadcastMessageSchema, - realmRtcSignalMessageSchema, -]) - -export const realmFromServerMessageSchema = z.discriminatedUnion('msg', [ - realmRtcPeerJoinedMessageSchema, - realmRtcPeerLeftMessageSchema, - realmRtcSignalMessageSchema, -]) - -export type RealmToServerMessage = z.infer -export type RealmFromServerMessage = z.infer diff --git a/src/common/protocol/messages.ts b/src/common/protocol/messages.ts index 611a21a..3429cea 100644 --- a/src/common/protocol/messages.ts +++ b/src/common/protocol/messages.ts @@ -1,15 +1,77 @@ +import {jwkSchema} from '#common/crypto/jwks' import {z} from 'zod/v4' -/** zod schema for `ok` message */ -export const responseOkSchema = z.object({ - ok: z.literal(true), -}) -export type ResponseOk = z.infer - -/** zod schema for `error` message */ -export const responseErrorSchema = z.object({ - ok: z.literal(false), - status: z.number(), - message: z.string(), -}) -export type ErrorResponse = z.infer +import {IdentBrand} from './brands' +import { + makeEmptyRequestSchema, + makeEmptyResponseSchema, + makeEventSchema, + makeRequestSchema, + makeResponseSchema, +} from './schema' + +/// preauth + +export const preauthRegisterReqSchema = makeRequestSchema( + 'preauth.register', + z.object({pubkey: jwkSchema}), +) + +export const preauthAuthnReqSchema = makeEmptyRequestSchema('preauth.authn') + +export const preauthRespSchema = makeResponseSchema( + 'preauth.authn', + z.object({ + activePeers: z.array(IdentBrand.schema), + }), +) + +export const preauthReqSchema = z.union([preauthAuthnReqSchema, preauthRegisterReqSchema]) + +export type PreauthRegisterRequest = z.infer +export type PreauthAuthnRequest = z.infer +export type PreauthResponse = z.infer + +/// realms + +export const realmBroadcastEventSchema = makeEventSchema( + 'realm.broadcast', + z.object({ + payload: z.unknown(), + recipients: z.union([z.boolean(), z.array(IdentBrand.schema)]).default(false), + }), +) + +export const realmRtcPingRequestSchema = makeEmptyRequestSchema('realm.rtc.ping') +export const realmRtcPongResponseSchema = makeEmptyResponseSchema('realm.rtc.pong') +export const realmRtcPingPongMessageSchema = z.union([ + realmRtcPingRequestSchema, + realmRtcPongResponseSchema, +]) + +export const realmRtcSignalEventSchema = makeEventSchema( + 'realm.rtc.signal', + z.object({ + payload: z.string(), + initiator: z.boolean(), + localid: IdentBrand.schema, + remoteid: IdentBrand.schema, + }), +) + +export const realmRtcPeerJoinedEventSchema = makeEventSchema( + 'realm.rtc.peer-joined', + z.object({identid: IdentBrand.schema}), +) + +export const realmRtcPeerLeftEventSchema = makeEventSchema( + 'realm.rtc.peer-left', + z.object({identid: IdentBrand.schema}), +) + +export type RealmBroadcastEvent = z.infer +export type RealmRtcPingRequest = z.infer +export type RealmRtcPongResponse = z.infer +export type RealmRtcSignalEvent = z.infer +export type RealmRtcPeerJoinedEvent = z.infer +export type RealmRtcPeerLeftEvent = z.infer diff --git a/src/common/protocol/schema.ts b/src/common/protocol/schema.ts new file mode 100644 index 0000000..d8f8483 --- /dev/null +++ b/src/common/protocol/schema.ts @@ -0,0 +1,70 @@ +import {z} from 'zod/v4' + +const errorSchema = z.object({code: z.number(), detail: z.string().optional()}) + +export const eventMessageSchema = z.object({ + typ: z.literal('evt'), + msg: z.string(), + dat: z.unknown(), +}) + +export const requestMessageSchema = z.object({ + typ: z.literal('req'), + msg: z.string(), + seq: z.number().optional(), + dat: z.unknown(), +}) + +export const responseMessageSchema = z.object({ + typ: z.literal('res'), + msg: z.string(), + seq: z.number().optional(), + dat: z.unknown(), +}) + +export const errorMessageSchema = z.object({ + typ: z.literal('err'), + msg: z.string(), + seq: z.number().optional(), + dat: errorSchema, +}) + +export const messageSchema = z.discriminatedUnion('typ', [ + eventMessageSchema, + requestMessageSchema, + responseMessageSchema, + errorMessageSchema, +]) + +export const makeEventSchema = (name: N, schema: Z) => { + return eventMessageSchema.extend({ + msg: z.literal(name), + dat: schema, + }) +} + +export const makeEmptyEventSchema = (name: N) => { + return eventMessageSchema.extend({msg: z.literal(name)}).omit({dat: true}) +} + +export const makeRequestSchema = (name: N, schema: Z) => { + return requestMessageSchema.extend({ + msg: z.literal(name), + dat: schema, + }) +} + +export const makeEmptyRequestSchema = (name: N) => { + return requestMessageSchema.extend({msg: z.literal(name)}).omit({dat: true}) +} + +export const makeResponseSchema = (name: N, schema: Z) => { + return responseMessageSchema.extend({ + msg: z.literal(name), + dat: schema, + }) +} + +export const makeEmptyResponseSchema = (name: N) => { + return responseMessageSchema.extend({msg: z.literal(name)}).omit({dat: true}) +} diff --git a/src/common/socket.ts b/src/common/socket.ts index e456e93..0a6dd3c 100644 --- a/src/common/socket.ts +++ b/src/common/socket.ts @@ -9,11 +9,6 @@ import {z} from 'zod/v4' import {parseJson} from './protocol' -/** send some data in json format down the wire */ -export function sendSocket(ws: WebSocket, data: unknown): void { - ws.send(JSON.stringify(data)) -} - /** * given a websocket, wait and take a single message off and return it * diff --git a/src/common/types.ts b/src/common/types.ts index e69de29..0eed3d8 100644 --- a/src/common/types.ts +++ b/src/common/types.ts @@ -0,0 +1 @@ +export type Replace = Omit & Record diff --git a/src/server/index.ts b/src/server/index.ts index 21b041c..128b97f 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -32,13 +32,9 @@ export function buildServer(root: string): http.Server { // WebSocket handling const wss = new WebSocketServer({server, path: '/stream'}) wss.on('connection', (ws) => { - socketHandler(ws) - .catch((e: unknown) => { - console.error('uncaught error from websocket', e) - }) - .finally(() => { - console.log('socket handler complete') - }) + socketHandler(ws).catch((e: unknown) => { + console.error('uncaught error from websocket', e) + }) }) return server diff --git a/src/server/routes-error.ts b/src/server/routes-error.ts index a324169..baed4ac 100644 --- a/src/server/routes-error.ts +++ b/src/server/routes-error.ts @@ -1,6 +1,5 @@ import * as express from 'express' export const notFoundHandler: express.RequestHandler = (req, res) => { - console.log(req.url) res.status(404).send('wut') } diff --git a/src/server/routes-socket/handler-preauth.ts b/src/server/routes-socket/handler-preauth.ts index 463f279..5b7def9 100644 --- a/src/server/routes-socket/handler-preauth.ts +++ b/src/server/routes-socket/handler-preauth.ts @@ -4,7 +4,14 @@ import {combineSignals, timeoutSignal} from '#common/async/aborts' import {jwkImport} from '#common/crypto/jwks' import {jwtPayload, verifyJwtToken} from '#common/crypto/jwts' import {normalizeError, ProtocolError} from '#common/errors' -import {IdentBrand, IdentID, preauthMessageSchema, RealmBrand, RealmID} from '#common/protocol' +import { + IdentBrand, + IdentID, + preauthReqSchema, + PreauthResponse, + RealmBrand, + RealmID, +} from '#common/protocol' import {takeSocket} from '#common/socket' import * as realms from './state' @@ -23,19 +30,23 @@ export async function preauthHandler( try { const data = await takeSocket(ws, combinedSignal) + const jwt = await jwtPayload(preauthReqSchema).parseAsync(data) // if any of the parsing fails, it'll throw a zod error - const jwt = await jwtPayload(preauthMessageSchema).parseAsync(data) const identid = IdentBrand.parse(jwt.claims.iss) const realmid = RealmBrand.parse(jwt.claims.aud) // if we're registering, make sure the realm exists if (jwt.payload.msg === 'preauth.register') { - const registrantkey = await jwkImport.parseAsync(jwt.payload.pubkey) + const registrantkey = await jwkImport.parseAsync(jwt.payload.dat.pubkey) realms.ensureRegisteredRealm(realmid, identid, registrantkey) } - return await authenticatePreauth(realmid, identid, jwt.token) + const auth = await authenticatePreauth(realmid, identid, jwt.token) + const msg = preauthResponse(auth, jwt.payload.seq) + ws.send(JSON.stringify(msg)) + + return auth } finally { timeout.cancel() } @@ -59,3 +70,14 @@ async function authenticatePreauth( throw new ProtocolError('jwt verification failed', 401, {cause: err}) } } + +function preauthResponse(auth: realms.AuthenticatedIdentity, seq?: number): PreauthResponse { + return { + typ: 'res', + msg: 'preauth.authn', + dat: { + activePeers: Array.from(auth.realm.sockets.keys()), + }, + seq, + } +} diff --git a/src/server/routes-socket/handler-realm.ts b/src/server/routes-socket/handler-realm.ts index 4c7d667..1010656 100644 --- a/src/server/routes-socket/handler-realm.ts +++ b/src/server/routes-socket/handler-realm.ts @@ -1,10 +1,17 @@ import {WebSocket} from 'isomorphic-ws' +import {z} from 'zod/v4' import {normalizeProtocolError, ProtocolError} from '#common/errors' import * as protocol from '#common/protocol' -import {sendSocket, streamSocket} from '#common/socket' +import {streamSocket} from '#common/socket' import * as realm from '#server/routes-socket/state' +// what can the server handle? +const incomingMessageSchema = z.union([ + protocol.realmBroadcastEventSchema, + protocol.realmRtcSignalEventSchema, +]) + /** * ance we've retrieved authentication details, we go into the main realm loop. * read messages as they come in and dispatch actions. @@ -15,21 +22,19 @@ export async function realmHandler( signal?: AbortSignal, ) { realmBroadcast(auth, buildRtcPeerJoined(auth)) - sendSocket(ws, buildRtcPeerWelcome(auth)) - - const parser = protocol.parseJson.pipe(protocol.realmToServerMessageSchema) try { - for await (const data of streamSocket(ws, {signal})) { + const incomingParser = protocol.parseJson.pipe(incomingMessageSchema) + for await (const msg of streamSocket(ws, {signal})) { try { - const msg = await parser.parseAsync(data) - switch (msg.msg) { + const data = await incomingParser.parseAsync(msg) + switch (data.msg) { case 'realm.broadcast': - realmBroadcast(auth, msg.payload, msg.recipients) + realmBroadcast(auth, data.dat, data.dat.recipients) continue case 'realm.rtc.signal': - realmBroadcast(auth, msg, [msg.recipient]) + realmBroadcast(auth, data, [data.dat.remoteid]) continue default: @@ -41,47 +46,29 @@ export async function realmHandler( if (error.status >= 500) throw error if (ws.readyState === ws.OPEN) { - sendSocket(ws, buildRealmError(error)) + const err = protocol.makeError(error, 'error in realm loop') + ws.send(JSON.stringify(err)) } } } } finally { - console.log('client left!', auth) realmBroadcast(auth, buildRtcPeerLeft(auth)) } } -function buildRtcPeerWelcome( - auth: realm.AuthenticatedIdentity, -): protocol.RealmRtcPeerWelcomeMessage { +function buildRtcPeerJoined(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerJoinedEvent { return { - ok: true, - msg: 'realm.rtc.peer-welcome', - peers: Array.from(auth.realm.sockets.keys()), - } -} - -function buildRtcPeerJoined(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerJoinedMessage { - return { - ok: true, + typ: 'evt', msg: 'realm.rtc.peer-joined', - identid: auth.identid, + dat: {identid: auth.identid}, } } -function buildRtcPeerLeft(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerLeftMessage { +function buildRtcPeerLeft(auth: realm.AuthenticatedIdentity): protocol.RealmRtcPeerLeftEvent { return { - ok: true, + typ: 'evt', msg: 'realm.rtc.peer-left', - identid: auth.identid, - } -} - -function buildRealmError(error: ProtocolError): protocol.ErrorResponse { - return { - ok: false, - message: error.message, - status: error.status, + dat: {identid: auth.identid}, } } @@ -100,6 +87,7 @@ function realmBroadcast( ) { const echo = recipients === true || Array.isArray(recipients) const recips = Array.isArray(recipients) ? recipients : Array.from(auth.realm.identities.keys()) + const json = JSON.stringify(payload) for (const recip of recips) { if (!echo && recip === auth.identid) continue @@ -107,7 +95,7 @@ function realmBroadcast( const sockets = auth.realm.sockets.get(recip) if (sockets) { for (const socket of sockets) { - sendSocket(socket, payload) + socket.send(json) } } } diff --git a/src/server/routes-socket/handler.ts b/src/server/routes-socket/handler.ts index 25f6af4..a0bce7f 100644 --- a/src/server/routes-socket/handler.ts +++ b/src/server/routes-socket/handler.ts @@ -9,8 +9,6 @@ import {attachSocket, detachSocket} from './state' /** when the socket connects, we drive our protocol through handlers */ export async function socketHandler(ws: WebSocket) { - console.log('WebSocket connection established') - try { const auth = await preauthHandler(ws) try { -- 2.51.2