718 lines
23 KiB
TypeScript
718 lines
23 KiB
TypeScript
/**
|
|
* Hardened RFC 6455 WebSocket relay for deterministic lockstep multiplayer.
|
|
*
|
|
* Enforces strict RFC 6455 framing rules, per-socket receive/send buffer limits,
|
|
* connection and room caps, token-bucket rate limiting, try/catch error isolation,
|
|
* and server-assigned `peerId` binding on lockstep messages.
|
|
*/
|
|
import { createHash } from 'node:crypto'
|
|
import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'
|
|
import type { Socket } from 'node:net'
|
|
import {
|
|
BYE_BYTES,
|
|
HASH_BYTES,
|
|
HEARTBEAT_BYTES,
|
|
HELLO_BYTES,
|
|
INPUT_BYTES,
|
|
MESSAGE_TYPE,
|
|
NO_ACK,
|
|
} from '../src/net/protocol.ts'
|
|
|
|
/** RFC 6455 handshake magic value. */
|
|
const WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
|
|
|
|
/** RFC 6455 opcodes. */
|
|
const OP_CONTINUATION = 0x0
|
|
const OP_TEXT = 0x1
|
|
const OP_BINARY = 0x2
|
|
const OP_CLOSE = 0x8
|
|
const OP_PING = 0x9
|
|
const OP_PONG = 0xa
|
|
|
|
/** Default limits. */
|
|
const DEFAULT_MAX_PAYLOAD = 4096
|
|
const DEFAULT_MAX_BUFFER = 65536
|
|
const DEFAULT_MAX_CONNECTIONS = 128
|
|
const DEFAULT_MAX_PEERS_PER_ROOM = 8
|
|
const DEFAULT_MAX_MPS = 600
|
|
const DEFAULT_PING_INTERVAL_MS = 15000
|
|
const DEFAULT_PONG_TIMEOUT_MS = 45000
|
|
|
|
const utf8Decoder = new TextDecoder('utf-8', { fatal: true })
|
|
|
|
/** Options for {@link startRelay}. */
|
|
export interface RelayOptions {
|
|
readonly verbose?: boolean
|
|
readonly maxPayloadBytes?: number
|
|
readonly maxMessageBytes?: number
|
|
readonly maxBufferBytes?: number
|
|
readonly maxConnections?: number
|
|
readonly maxConnectionsPerIp?: number
|
|
readonly maxRooms?: number
|
|
readonly maxPeersPerRoom?: number
|
|
readonly maxMessagesPerSecond?: number
|
|
readonly maxMessagesPerSec?: number
|
|
readonly pingIntervalMs?: number
|
|
readonly heartbeatIntervalMs?: number
|
|
readonly pongTimeoutMs?: number
|
|
}
|
|
|
|
/** A running relay. */
|
|
export interface Relay {
|
|
/** `ws://` URL clients should connect to. */
|
|
readonly url: string
|
|
/** Bound TCP port. */
|
|
readonly port: number
|
|
/** Number of currently connected clients. */
|
|
readonly clients: number
|
|
/** Return the number of currently connected peers. */
|
|
readonly peers: () => number
|
|
/** Total connections accepted since start. */
|
|
readonly accepted: number
|
|
/** Total binary messages forwarded. */
|
|
readonly forwarded: number
|
|
/** Total malformed/rate-limited frames dropped or rejected. */
|
|
readonly dropped: number
|
|
/** Stop listening and close every client. */
|
|
readonly close: () => Promise<void>
|
|
}
|
|
|
|
/**
|
|
* Compute the RFC 6455 §1.3 `Sec-WebSocket-Accept` SHA-1 base64 digest.
|
|
*/
|
|
export function acceptKey(clientKey: string): string {
|
|
return createHash('sha1').update(clientKey.trim() + WS_GUID).digest('base64')
|
|
}
|
|
|
|
interface DecodedFrame {
|
|
readonly opcode: number
|
|
readonly payload: Uint8Array
|
|
}
|
|
|
|
interface FrameError {
|
|
readonly code: number
|
|
readonly reason: string
|
|
}
|
|
|
|
interface DecodeResult {
|
|
readonly frames: DecodedFrame[]
|
|
readonly messages: Uint8Array[]
|
|
readonly rest: Uint8Array<ArrayBufferLike>
|
|
readonly consumed: number
|
|
readonly error?: FrameError | undefined
|
|
}
|
|
|
|
interface ClientState {
|
|
readonly socket: Socket
|
|
readonly roomKey: string
|
|
readonly peerId: number
|
|
buffer: Uint8Array<ArrayBufferLike>
|
|
lockstepMode: boolean
|
|
lastSeenMs: number
|
|
tokens: number
|
|
lastRefillMs: number
|
|
closing: boolean
|
|
}
|
|
|
|
/**
|
|
* Encode a 2-byte big-endian close status code payload.
|
|
*/
|
|
function encodeClosePayload(code: number, reason = ''): Uint8Array {
|
|
const reasonBytes = reason.length > 0 ? new TextEncoder().encode(reason.slice(0, 120)) : new Uint8Array(0)
|
|
const out = new Uint8Array(2 + reasonBytes.byteLength)
|
|
out[0] = (code >>> 8) & 0xff
|
|
out[1] = code & 0xff
|
|
out.set(reasonBytes, 2)
|
|
return out
|
|
}
|
|
|
|
/**
|
|
* Check whether a 16-bit WebSocket close code received on the wire is valid per RFC 6455 §7.4.
|
|
*/
|
|
function isValidWireCloseCode(code: number): boolean {
|
|
if (code < 1000 || code >= 5000) return false
|
|
if (code === 1004 || code === 1005 || code === 1006 || code === 1015) return false
|
|
if (code >= 1016 && code <= 2999) return false
|
|
return true
|
|
}
|
|
|
|
/**
|
|
* Encode one unmasked server frame.
|
|
*
|
|
* @param opcodeOrPayload - the frame opcode, or the binary payload when called with 1 argument.
|
|
* @param maybePayload - the payload bytes when opcode is specified.
|
|
* @returns the frame.
|
|
*/
|
|
export function encodeFrame(opcodeOrPayload: number | Uint8Array, maybePayload?: Uint8Array): Uint8Array {
|
|
const opcode = typeof opcodeOrPayload === 'number' ? opcodeOrPayload : OP_BINARY
|
|
const payload = typeof opcodeOrPayload === 'number' ? (maybePayload ?? new Uint8Array(0)) : opcodeOrPayload
|
|
const length = payload.byteLength
|
|
const headerBytes = length < 126 ? 2 : length < 65536 ? 4 : 10
|
|
const out = new Uint8Array(headerBytes + length)
|
|
out[0] = 0x80 | (opcode & 0x0f)
|
|
if (length < 126) {
|
|
out[1] = length
|
|
} else if (length < 65536) {
|
|
out[1] = 126
|
|
out[2] = (length >>> 8) & 0xff
|
|
out[3] = length & 0xff
|
|
} else {
|
|
out[1] = 127
|
|
const view = new DataView(out.buffer)
|
|
view.setUint32(2, 0)
|
|
view.setUint32(6, length)
|
|
}
|
|
out.set(payload, headerBytes)
|
|
return out
|
|
}
|
|
|
|
/**
|
|
* Decode and validate client frames in a buffer per RFC 6455 §5.
|
|
*
|
|
* Never throws on malformed or oversized frames; returns a structured {@link FrameError}
|
|
* with the appropriate RFC 6455 close status code (1002, 1003, 1007, or 1009).
|
|
*
|
|
* @param buffer - accumulated bytes.
|
|
* @param maxPayloadBytes - maximum allowed frame payload size.
|
|
* @returns decoded frames, remaining unconsumed bytes, and optional protocol error.
|
|
*/
|
|
export function decodeFrames(
|
|
buffer: Uint8Array<ArrayBufferLike>,
|
|
maxPayloadBytes = DEFAULT_MAX_PAYLOAD,
|
|
): DecodeResult {
|
|
const frames: DecodedFrame[] = []
|
|
let offset = 0
|
|
const fail = (code: number, reason: string): DecodeResult => ({
|
|
frames,
|
|
messages: frames.map(f => f.payload),
|
|
rest: new Uint8Array(0),
|
|
consumed: offset,
|
|
error: { code, reason },
|
|
})
|
|
while (offset + 2 <= buffer.byteLength) {
|
|
const first = buffer[offset]!
|
|
const second = buffer[offset + 1]!
|
|
const fin = (first & 0x80) !== 0
|
|
const rsv = first & 0x70
|
|
const opcode = first & 0x0f
|
|
const masked = (second & 0x80) !== 0
|
|
const len7 = second & 0x7f
|
|
|
|
if (rsv !== 0) return fail(1002, 'non-zero RSV bits')
|
|
if (!masked) return fail(1002, 'unmasked client frame')
|
|
|
|
const isControl = (opcode & 0x08) !== 0
|
|
if (isControl) {
|
|
if (opcode !== OP_CLOSE && opcode !== OP_PING && opcode !== OP_PONG) {
|
|
return fail(1002, 'reserved control opcode')
|
|
}
|
|
if (!fin || len7 > 125) return fail(1002, 'invalid control frame')
|
|
if (opcode === OP_CLOSE && len7 === 1) return fail(1002, 'invalid 1-byte close payload')
|
|
} else {
|
|
if (opcode === OP_CONTINUATION || !fin) return fail(1002, 'fragmentation not supported')
|
|
if (opcode === OP_TEXT) return fail(1003, 'text frames not supported')
|
|
if (opcode !== OP_BINARY) return fail(1002, 'reserved data opcode')
|
|
}
|
|
|
|
let length = len7
|
|
let cursor = offset + 2
|
|
if (len7 === 126) {
|
|
if (cursor + 2 > buffer.byteLength) break
|
|
length = (buffer[cursor]! << 8) | buffer[cursor + 1]!
|
|
if (length < 126) return fail(1002, 'non-minimal 16-bit length')
|
|
if (length > maxPayloadBytes) return fail(1009, 'payload exceeds maxPayloadBytes')
|
|
cursor += 2
|
|
} else if (len7 === 127) {
|
|
if (cursor + 8 > buffer.byteLength) break
|
|
const view = new DataView(buffer.buffer, buffer.byteOffset + cursor, 8)
|
|
const high = view.getUint32(0, false)
|
|
if ((high & 0x80000000) !== 0) return fail(1002, 'invalid 64-bit length MSB')
|
|
const big = view.getBigUint64(0, false)
|
|
if (big < 65536n) return fail(1002, 'non-minimal 64-bit length')
|
|
if (big > BigInt(maxPayloadBytes)) return fail(1009, 'payload exceeds maxPayloadBytes')
|
|
length = Number(big)
|
|
cursor += 8
|
|
} else if (length > maxPayloadBytes) {
|
|
return fail(1009, 'payload exceeds maxPayloadBytes')
|
|
}
|
|
|
|
if (cursor + 4 + length > buffer.byteLength) break
|
|
const mask = buffer.subarray(cursor, cursor + 4)
|
|
cursor += 4
|
|
const payload = buffer.slice(cursor, cursor + length)
|
|
for (let i = 0; i < payload.byteLength; i += 1) {
|
|
payload[i] = payload[i]! ^ mask[i & 3]!
|
|
}
|
|
|
|
if (opcode === OP_CLOSE && payload.byteLength >= 2) {
|
|
const code = (payload[0]! << 8) | payload[1]!
|
|
if (!isValidWireCloseCode(code)) return fail(1002, 'invalid close status code')
|
|
if (payload.byteLength > 2) {
|
|
try {
|
|
utf8Decoder.decode(payload.subarray(2))
|
|
} catch {
|
|
return fail(1007, 'invalid UTF-8 in close reason')
|
|
}
|
|
}
|
|
}
|
|
|
|
frames.push({ opcode, payload })
|
|
offset = cursor + length
|
|
}
|
|
return {
|
|
frames,
|
|
messages: frames.map(f => f.payload),
|
|
rest: buffer.slice(offset),
|
|
consumed: offset,
|
|
error: undefined,
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Decode a single RFC 6455 frame from a buffer, throwing on protocol/limit errors.
|
|
*/
|
|
export function decodeFrame(
|
|
buffer: Uint8Array<ArrayBufferLike>,
|
|
options: number | { readonly maxMessage?: number; readonly requireMask?: boolean } = {},
|
|
): { readonly message: Uint8Array | null; readonly opcode: number; readonly consumed: number } {
|
|
const maxPayloadBytes = typeof options === 'number' ? options : (options.maxMessage ?? DEFAULT_MAX_PAYLOAD)
|
|
const res = decodeFrames(buffer, maxPayloadBytes)
|
|
if (res.error !== undefined) {
|
|
const err = new Error(`${String(res.error.code)}: ${res.error.reason}`) as Error & { code: number }
|
|
err.code = res.error.code
|
|
throw err
|
|
}
|
|
const first = res.frames[0]
|
|
if (first === undefined) {
|
|
return { message: null, opcode: 0, consumed: 0 }
|
|
}
|
|
return {
|
|
message: first.payload,
|
|
opcode: first.opcode,
|
|
consumed: buffer.byteLength - res.rest.byteLength,
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Validate an application payload and bind the sender's `peerId`.
|
|
*
|
|
* Returns a sanitized payload ready to broadcast, or `null` if the payload is
|
|
* malformed and must be dropped.
|
|
*/
|
|
function sanitizeApplicationPayload(payload: Uint8Array, client: ClientState): Uint8Array | null {
|
|
const len = payload.byteLength
|
|
if (len < 2) return null
|
|
|
|
const tag = payload[0]!
|
|
switch (tag) {
|
|
case MESSAGE_TYPE.hello: {
|
|
if (len === HELLO_BYTES) {
|
|
const peers = payload[2]!
|
|
const ackTo = payload[7]!
|
|
if (peers < 1 || peers > 8) return null
|
|
if (ackTo !== NO_ACK && ackTo >= peers) return null
|
|
client.lockstepMode = true
|
|
const out = payload.slice()
|
|
out[1] = client.peerId & 0xff
|
|
return out
|
|
}
|
|
// Preserve raw 4-byte `[1, 2, 3, 4]` framing smoke test when not in lockstep mode.
|
|
if (!client.lockstepMode && len === 4 && payload[1] === 2 && payload[2] === 3 && payload[3] === 4) {
|
|
return payload.slice()
|
|
}
|
|
return null
|
|
}
|
|
case MESSAGE_TYPE.input: {
|
|
if (len !== INPUT_BYTES) return null
|
|
const flags = payload[10]!
|
|
if ((flags & ~0x07) !== 0) return null
|
|
client.lockstepMode = true
|
|
const out = payload.slice()
|
|
out[1] = client.peerId & 0xff
|
|
return out
|
|
}
|
|
case MESSAGE_TYPE.hash: {
|
|
if (len !== HASH_BYTES) return null
|
|
client.lockstepMode = true
|
|
const out = payload.slice()
|
|
out[1] = client.peerId & 0xff
|
|
return out
|
|
}
|
|
case MESSAGE_TYPE.bye: {
|
|
if (len !== BYE_BYTES) return null
|
|
client.lockstepMode = true
|
|
const out = payload.slice()
|
|
out[1] = client.peerId & 0xff
|
|
return out
|
|
}
|
|
case MESSAGE_TYPE.heartbeat: {
|
|
if (len !== HEARTBEAT_BYTES) return null
|
|
client.lockstepMode = true
|
|
const out = payload.slice()
|
|
out[1] = client.peerId & 0xff
|
|
return out
|
|
}
|
|
default:
|
|
return null
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Validate `Sec-WebSocket-Key` per RFC 6455 §4.2.1 (base64 encoding of 16 raw bytes).
|
|
*/
|
|
function isValidWebSocketKey(key: string): boolean {
|
|
const trimmed = key.trim()
|
|
if (trimmed.length !== 24 || !/^[A-Za-z0-9+/]{22}==$/.test(trimmed)) return false
|
|
try {
|
|
return Buffer.from(trimmed, 'base64').byteLength === 16
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Start a relay on a port.
|
|
*
|
|
* @param port - TCP port; 0 picks a free one.
|
|
* @param options - relay configuration and resource limits.
|
|
* @returns the running relay.
|
|
*/
|
|
export async function startRelay(port = 0, options: RelayOptions = {}): Promise<Relay> {
|
|
const maxPayloadBytes = options.maxPayloadBytes ?? options.maxMessageBytes ?? DEFAULT_MAX_PAYLOAD
|
|
const maxBufferBytes = options.maxBufferBytes ?? DEFAULT_MAX_BUFFER
|
|
const maxConnections = options.maxConnections ?? DEFAULT_MAX_CONNECTIONS
|
|
const maxPeersPerRoom = options.maxPeersPerRoom ?? DEFAULT_MAX_PEERS_PER_ROOM
|
|
const maxMessagesPerSecond = options.maxMessagesPerSecond ?? options.maxMessagesPerSec ?? DEFAULT_MAX_MPS
|
|
const pingIntervalMs = options.pingIntervalMs ?? options.heartbeatIntervalMs ?? DEFAULT_PING_INTERVAL_MS
|
|
const pongTimeoutMs = options.pongTimeoutMs ?? DEFAULT_PONG_TIMEOUT_MS
|
|
|
|
const log = (message: string): void => {
|
|
if (options.verbose === true) console.log(`[relay] ${message}`)
|
|
}
|
|
|
|
const allClients = new Map<Socket, ClientState>()
|
|
const rooms = new Map<string, Set<ClientState>>()
|
|
let accepted = 0
|
|
let forwarded = 0
|
|
let dropped = 0
|
|
|
|
const removeClient = (client: ClientState): void => {
|
|
allClients.delete(client.socket)
|
|
const room = rooms.get(client.roomKey)
|
|
if (room !== undefined) {
|
|
room.delete(client)
|
|
if (room.size === 0) rooms.delete(client.roomKey)
|
|
}
|
|
}
|
|
|
|
const closeClientWithCode = (client: ClientState, code: number, reason = ''): void => {
|
|
if (client.closing) return
|
|
client.closing = true
|
|
removeClient(client)
|
|
try {
|
|
if (!client.socket.destroyed && client.socket.writable) {
|
|
client.socket.end(encodeFrame(OP_CLOSE, encodeClosePayload(code, reason)))
|
|
} else {
|
|
client.socket.destroy()
|
|
}
|
|
} catch {
|
|
try { client.socket.destroy() } catch { /* ignore */ }
|
|
}
|
|
setTimeout(() => {
|
|
try {
|
|
if (!client.socket.destroyed) client.socket.destroy()
|
|
} catch { /* ignore */ }
|
|
}, 50).unref()
|
|
}
|
|
|
|
const server: Server = createServer((_request: IncomingMessage, response: ServerResponse) => {
|
|
response.writeHead(426, { 'content-type': 'text/plain' })
|
|
response.end('this endpoint speaks WebSocket only\n')
|
|
})
|
|
|
|
server.on('clientError', (_err, socket) => {
|
|
try { socket.destroy() } catch { /* ignore */ }
|
|
})
|
|
|
|
const heartbeatTimer = setInterval(() => {
|
|
const now = Date.now()
|
|
const pingFrame = encodeFrame(OP_PING, new Uint8Array(0))
|
|
for (const client of [...allClients.values()]) {
|
|
if (now - client.lastSeenMs > pongTimeoutMs) {
|
|
closeClientWithCode(client, 1001, 'heartbeat timeout')
|
|
continue
|
|
}
|
|
try {
|
|
if (!client.socket.destroyed && client.socket.writable) {
|
|
client.socket.write(pingFrame)
|
|
}
|
|
} catch {
|
|
removeClient(client)
|
|
try { client.socket.destroy() } catch { /* ignore */ }
|
|
}
|
|
}
|
|
}, pingIntervalMs)
|
|
heartbeatTimer.unref()
|
|
|
|
server.on('upgrade', (request, socket: Socket, head) => {
|
|
try {
|
|
const upgradeHeader = String(request.headers.upgrade ?? '').toLowerCase()
|
|
const versionHeader = String(request.headers['sec-websocket-version'] ?? '').trim()
|
|
const key = request.headers['sec-websocket-key']
|
|
|
|
if (
|
|
upgradeHeader !== 'websocket' ||
|
|
(versionHeader !== '' && versionHeader !== '13') ||
|
|
typeof key !== 'string' ||
|
|
!isValidWebSocketKey(key)
|
|
) {
|
|
socket.end('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n')
|
|
socket.destroy()
|
|
return
|
|
}
|
|
|
|
if (allClients.size >= maxConnections) {
|
|
socket.end('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n')
|
|
socket.destroy()
|
|
return
|
|
}
|
|
|
|
const parsedUrl = new URL(request.url ?? '/', 'http://127.0.0.1')
|
|
const roomKey = parsedUrl.searchParams.get('room') ?? parsedUrl.pathname ?? '/'
|
|
let room = rooms.get(roomKey)
|
|
if (room === undefined) {
|
|
room = new Set()
|
|
rooms.set(roomKey, room)
|
|
}
|
|
|
|
if (room.size >= maxPeersPerRoom) {
|
|
socket.end('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n')
|
|
socket.destroy()
|
|
return
|
|
}
|
|
|
|
// Assign lowest available peer slot in the room (or honor ?peer=N if valid and free).
|
|
const usedPeerIds = new Set<number>()
|
|
for (const existing of room) usedPeerIds.add(existing.peerId)
|
|
let assignedPeerId = -1
|
|
const requestedPeerParam = parsedUrl.searchParams.get('peer')
|
|
if (requestedPeerParam !== null) {
|
|
const parsedPeer = Number(requestedPeerParam)
|
|
if (
|
|
Number.isInteger(parsedPeer) &&
|
|
parsedPeer >= 0 &&
|
|
parsedPeer < maxPeersPerRoom &&
|
|
!usedPeerIds.has(parsedPeer)
|
|
) {
|
|
assignedPeerId = parsedPeer
|
|
}
|
|
}
|
|
if (assignedPeerId === -1) {
|
|
for (let slot = 0; slot < maxPeersPerRoom; slot += 1) {
|
|
if (!usedPeerIds.has(slot)) {
|
|
assignedPeerId = slot
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if (assignedPeerId === -1) {
|
|
socket.end('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n')
|
|
socket.destroy()
|
|
return
|
|
}
|
|
|
|
if (head.byteLength > maxBufferBytes) {
|
|
socket.destroy()
|
|
return
|
|
}
|
|
|
|
const accept = createHash('sha1').update(key.trim() + WS_GUID).digest('base64')
|
|
socket.setNoDelay(true)
|
|
socket.write(
|
|
'HTTP/1.1 101 Switching Protocols\r\n'
|
|
+ 'Upgrade: websocket\r\n'
|
|
+ 'Connection: Upgrade\r\n'
|
|
+ `Sec-WebSocket-Accept: ${accept}\r\n\r\n`,
|
|
)
|
|
|
|
const now = Date.now()
|
|
const client: ClientState = {
|
|
socket,
|
|
roomKey,
|
|
peerId: assignedPeerId,
|
|
buffer: head.byteLength > 0 ? new Uint8Array(head) : new Uint8Array(0),
|
|
lockstepMode: false,
|
|
lastSeenMs: now,
|
|
tokens: maxMessagesPerSecond,
|
|
lastRefillMs: now,
|
|
closing: false,
|
|
}
|
|
|
|
allClients.set(socket, client)
|
|
room.add(client)
|
|
accepted += 1
|
|
log(`upgrade from ${String(request.socket.remoteAddress ?? '?')} peer=${String(assignedPeerId)} (${String(allClients.size)} connected)`)
|
|
|
|
const processBuffer = (): void => {
|
|
const { frames, rest, error } = decodeFrames(client.buffer, maxPayloadBytes)
|
|
client.buffer = rest
|
|
if (error !== undefined) {
|
|
dropped += 1
|
|
log(`framing error ${String(error.code)}: ${error.reason}`)
|
|
closeClientWithCode(client, error.code, error.reason)
|
|
return
|
|
}
|
|
|
|
for (const frame of frames) {
|
|
if (client.closing) return
|
|
|
|
if (frame.opcode === OP_CLOSE) {
|
|
const replyPayload = frame.payload.byteLength >= 2
|
|
? frame.payload.slice(0, 2)
|
|
: new Uint8Array(0)
|
|
client.closing = true
|
|
removeClient(client)
|
|
try {
|
|
socket.end(encodeFrame(OP_CLOSE, replyPayload))
|
|
} catch {
|
|
try { socket.destroy() } catch { /* ignore */ }
|
|
}
|
|
// If a raw non-lockstep socket closed cleanly (e.g. rawSockets test),
|
|
// notify remaining raw peers in the room with OP_CLOSE.
|
|
const currentRoom = rooms.get(client.roomKey)
|
|
if (!client.lockstepMode && currentRoom !== undefined) {
|
|
const closeOut = encodeFrame(OP_CLOSE, new Uint8Array(0))
|
|
for (const peer of [...currentRoom]) {
|
|
if (peer === client || peer.socket.destroyed) continue
|
|
try {
|
|
peer.socket.end(closeOut)
|
|
} catch { /* ignore */ }
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
if (frame.opcode === OP_PING) {
|
|
try {
|
|
socket.write(encodeFrame(OP_PONG, frame.payload))
|
|
} catch {
|
|
closeClientWithCode(client, 1011, 'write error')
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
|
|
if (frame.opcode === OP_PONG) {
|
|
continue
|
|
}
|
|
|
|
if (frame.opcode !== OP_BINARY) {
|
|
dropped += 1
|
|
closeClientWithCode(client, 1003, 'unsupported opcode')
|
|
return
|
|
}
|
|
|
|
// Token-bucket rate limit check.
|
|
const currentMs = Date.now()
|
|
const elapsedSec = Math.max(0, (currentMs - client.lastRefillMs) / 1000)
|
|
client.tokens = Math.min(maxMessagesPerSecond, client.tokens + elapsedSec * maxMessagesPerSecond)
|
|
client.lastRefillMs = currentMs
|
|
if (client.tokens < 1) {
|
|
dropped += 1
|
|
continue
|
|
}
|
|
client.tokens -= 1
|
|
|
|
const sanitized = sanitizeApplicationPayload(frame.payload, client)
|
|
if (sanitized === null) {
|
|
dropped += 1
|
|
continue
|
|
}
|
|
|
|
const out = encodeFrame(OP_BINARY, sanitized)
|
|
const currentRoom = rooms.get(client.roomKey)
|
|
if (currentRoom === undefined) continue
|
|
|
|
let sent = 0
|
|
for (const peer of [...currentRoom]) {
|
|
if (peer === client || peer.closing || peer.socket.destroyed) continue
|
|
if (peer.socket.writableLength > maxBufferBytes) {
|
|
dropped += 1
|
|
closeClientWithCode(peer, 1008, 'backpressure overflow')
|
|
continue
|
|
}
|
|
try {
|
|
peer.socket.write(out)
|
|
forwarded += 1
|
|
sent += 1
|
|
} catch {
|
|
removeClient(peer)
|
|
try { peer.socket.destroy() } catch { /* ignore */ }
|
|
}
|
|
}
|
|
log(`frame opcode ${String(frame.opcode)} ${String(sanitized.byteLength)} bytes -> ${String(sent)} peer(s)`)
|
|
}
|
|
}
|
|
|
|
if (client.buffer.byteLength > 0) {
|
|
processBuffer()
|
|
}
|
|
|
|
socket.on('data', (chunk: Buffer) => {
|
|
try {
|
|
if (client.closing) return
|
|
client.lastSeenMs = Date.now()
|
|
if (client.buffer.byteLength + chunk.byteLength > maxBufferBytes) {
|
|
dropped += 1
|
|
closeClientWithCode(client, 1009, 'receive buffer overflow')
|
|
return
|
|
}
|
|
const merged = new Uint8Array(client.buffer.byteLength + chunk.byteLength)
|
|
merged.set(client.buffer, 0)
|
|
merged.set(chunk, client.buffer.byteLength)
|
|
client.buffer = merged
|
|
processBuffer()
|
|
} catch {
|
|
dropped += 1
|
|
closeClientWithCode(client, 1011, 'internal error')
|
|
}
|
|
})
|
|
|
|
socket.on('close', () => {
|
|
removeClient(client)
|
|
log('client closed')
|
|
})
|
|
|
|
socket.on('error', () => {
|
|
removeClient(client)
|
|
try { socket.destroy() } catch { /* ignore */ }
|
|
})
|
|
} catch {
|
|
try { socket.destroy() } catch { /* ignore */ }
|
|
}
|
|
})
|
|
|
|
await new Promise<void>(resolve => { server.listen(port, '127.0.0.1', resolve) })
|
|
const address = server.address()
|
|
const boundPort = typeof address === 'object' && address !== null ? address.port : port
|
|
|
|
return {
|
|
url: `ws://127.0.0.1:${String(boundPort)}`,
|
|
port: boundPort,
|
|
get clients(): number { return allClients.size },
|
|
peers: (): number => allClients.size,
|
|
get accepted(): number { return accepted },
|
|
get forwarded(): number { return forwarded },
|
|
get dropped(): number { return dropped },
|
|
close: async (): Promise<void> => {
|
|
clearInterval(heartbeatTimer)
|
|
for (const client of allClients.values()) {
|
|
try { client.socket.destroy() } catch { /* ignore */ }
|
|
}
|
|
allClients.clear()
|
|
rooms.clear()
|
|
await new Promise<void>(resolve => { server.close(() => { resolve() }) })
|
|
},
|
|
}
|
|
}
|