diff --git a/.github/workflows/publish_extra.yml b/.github/workflows/publish_extra.yml index 980eb1da55..863c850266 100644 --- a/.github/workflows/publish_extra.yml +++ b/.github/workflows/publish_extra.yml @@ -48,6 +48,11 @@ jobs: npm ci --omit=dev npm test + # Runs the real entrypoint-extra.sh with stub services, so the supervision + # behaviour is checked without building the image first. + - name: Run entrypoint supervision tests + run: bash docker/test_entrypoint_extra.sh + - name: Set up Docker Buildx uses: docker/setup-buildx-action@v3 diff --git a/docker/entrypoint-extra.sh b/docker/entrypoint-extra.sh index b32ad2dd94..84b337a8a1 100644 --- a/docker/entrypoint-extra.sh +++ b/docker/entrypoint-extra.sh @@ -4,18 +4,51 @@ set -e # Windmill Extra Services Entrypoint # Starts LSP, Multiplayer, and Debugger services based on environment variables -# Track PIDs for cleanup +# Track PIDs for cleanup, with a parallel array of names so a dead service can be +# named in the logs rather than reported as a bare PID. PIDS=() +NAMES=() -cleanup() { +# How long a service gets to honour SIGTERM before it is killed outright. The +# exit path below signals itself, so nothing else bounds it there; a wedged +# service would otherwise hold the container open half-dead forever. +SHUTDOWN_GRACE_SECS="${SHUTDOWN_GRACE_SECS:-10}" + +stop_services() { echo "[entrypoint] Shutting down services..." for pid in "${PIDS[@]}"; do if kill -0 "$pid" 2>/dev/null; then kill "$pid" 2>/dev/null || true fi done + + local i running pid + for ((i = 0; i < SHUTDOWN_GRACE_SECS * 10; i++)); do + running=0 + for pid in "${PIDS[@]}"; do + if kill -0 "$pid" 2>/dev/null; then + running=1 + fi + done + if [ "$running" -eq 0 ]; then + break + fi + sleep 0.1 + done + + for pid in "${PIDS[@]}"; do + if kill -0 "$pid" 2>/dev/null; then + echo "[entrypoint] WARNING: PID $pid did not stop within ${SHUTDOWN_GRACE_SECS}s, killing it" >&2 + kill -9 "$pid" 2>/dev/null || true + fi + done + wait echo "[entrypoint] All services stopped" +} + +cleanup() { + stop_services exit 0 } @@ -116,6 +149,7 @@ if [ "${ENABLE_LSP:-true}" = "true" ]; then cd /pyls PORT=${LSP_PORT:-3001} python3 pyls_launcher.py & PIDS+=($!) + NAMES+=("LSP") echo "[entrypoint] LSP started (PID: ${PIDS[-1]})" fi @@ -125,6 +159,7 @@ if [ "${ENABLE_MULTIPLAYER:-true}" = "true" ]; then cd /multiplayer PORT=${MULTIPLAYER_PORT:-3002} HOST=${HOST:-0.0.0.0} node server.mjs & PIDS+=($!) + NAMES+=("Multiplayer") echo "[entrypoint] Multiplayer started (PID: ${PIDS[-1]})" fi @@ -144,6 +179,7 @@ if [ "${ENABLE_DEBUGGER:-true}" = "true" ]; then bun run dap_debug_service.ts $DEBUGGER_ARGS & PIDS+=($!) + NAMES+=("Debugger") echo "[entrypoint] Debugger started (PID: ${PIDS[-1]})" fi @@ -153,6 +189,7 @@ if [ "${ENABLE_GATEWAY:-true}" = "true" ]; then cd /multiplayer PORT=${GATEWAY_PORT:-3000} node gateway.mjs & PIDS+=($!) + NAMES+=("Gateway") echo "[entrypoint] Gateway started (PID: ${PIDS[-1]})" fi @@ -165,15 +202,27 @@ fi echo "[entrypoint] All enabled services started. Waiting..." -# Wait for any process to exit -wait -n "${PIDS[@]}" 2>/dev/null || true +# Wait for any process to exit. `|| status=$?` also keeps `set -e` from aborting +# here, which would skip the reporting and shutdown below. +status=0 +wait -n "${PIDS[@]}" 2>/dev/null || status=$? -# If one process exits, check which one and report +# Nothing here restarts a dead service, and the health checks in front of this +# container probe a single one, so a dead sibling is invisible and stays dead. +# Name it, stop the rest and exit non-zero, leaving the orchestrator to replace +# the container. for i in "${!PIDS[@]}"; do if ! kill -0 "${PIDS[$i]}" 2>/dev/null; then - echo "[entrypoint] Service (PID: ${PIDS[$i]}) has exited" + echo "[entrypoint] ERROR: ${NAMES[$i]} (PID: ${PIDS[$i]}) has exited" >&2 fi done -# Keep running and wait for remaining processes -wait +stop_services + +# Carry the dead service's own status where there is one, but never exit 0: that +# would read as a successful, deliberate shutdown. +if [ "$status" -eq 0 ]; then + status=1 +fi +echo "[entrypoint] Exiting with status $status so the container is restarted" >&2 +exit "$status" diff --git a/docker/test_entrypoint_extra.sh b/docker/test_entrypoint_extra.sh new file mode 100755 index 0000000000..e0b84da366 --- /dev/null +++ b/docker/test_entrypoint_extra.sh @@ -0,0 +1,246 @@ +#!/usr/bin/env bash +# +# Supervision tests for docker/entrypoint-extra.sh. +# +# The real script is run unmodified inside a container, with the four services it +# starts replaced by stubs, so what is under test is the shipped file and the +# shipped bash. The stubs are recognised by the argv the entrypoint uses +# (pyls_launcher.py, server.mjs, dap_debug_service.ts, gateway.mjs), which is +# also what keeps the test honest: change how a service is started and the +# corresponding case here stops matching. +# +# Usage: +# bash docker/test_entrypoint_extra.sh +# +# The image defaults to the base of DockerfileExtra's chain (debian:trixie-slim, +# via windmill-ee-slim), so the bash built-ins behave as they do in the real +# image. Override with ENTRYPOINT_TEST_IMAGE to run it against another one, e.g. +# ENTRYPOINT_TEST_IMAGE=windmill-extra:test bash docker/test_entrypoint_extra.sh + +set -euo pipefail + +IMAGE="${ENTRYPOINT_TEST_IMAGE:-debian:trixie-slim}" +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +ENTRYPOINT="$SCRIPT_DIR/entrypoint-extra.sh" + +if ! command -v docker >/dev/null 2>&1; then + echo "docker is required to run this test" >&2 + exit 1 +fi + +WORKDIR="$(mktemp -d)" +trap 'rm -rf "$WORKDIR"' EXIT + +# --- the pieces that run inside the container ----------------------------- + +cat > "$WORKDIR/stub" <<'STUB' +#!/bin/bash +# Stands in for python3 / node / bun. Works out which service it was started as +# from its arguments, then either stays up until it is signalled (recording that +# it was stopped) or exits straight away, to play the crashed service. +name=unknown +for arg in "$@"; do + case "$arg" in + pyls_launcher.py) name=lsp ;; + server.mjs) name=multiplayer ;; + dap_debug_service.ts) name=debugger ;; + gateway.mjs) name=gateway ;; + esac +done +echo "stub[$name] started" + +if [ "$name" = "${STUB_IGNORE_TERM:-}" ]; then + # Plays a service that is wedged or slow to honour SIGTERM. + trap 'echo "stub[$name] ignoring SIGTERM"' TERM +else + trap 'echo "stub[$name] got SIGTERM"; touch "/out/$name.stopped"; exit 0' TERM +fi +touch "/out/$name.ready" + +if [ "$name" = "${STUB_DIE_AS:-}" ]; then + # Only die once every stub in $STUB_WAIT_FOR has its trap installed. Without + # this the entrypoint can reach its shutdown while a later stub is still + # starting, and that stub takes SIGTERM's default action instead of running + # its handler — a race in the test, not in the entrypoint. + for _ in $(seq 1 600); do + ready=1 + for other in ${STUB_WAIT_FOR:-}; do + if [ ! -f "/out/$other.ready" ]; then ready=0; fi + done + if [ "$ready" -eq 1 ]; then break; fi + sleep 0.1 + done + echo "stub[$name] exiting with ${STUB_DIE_CODE:-1}" + exit "${STUB_DIE_CODE:-1}" +fi +# `sleep & wait` rather than a bare `sleep`, so the trap runs as soon as the +# signal arrives instead of after the sleep returns. The loop matters: bash +# resumes after an interrupted `wait`, so without it even the trap that only +# logs would fall off the end of the script and exit. +while true; do + sleep 3000 & + wait $! +done +STUB + +cat > "$WORKDIR/harness.sh" <<'HARNESS' +#!/bin/bash +# Runs inside the container: installs the stubs, then drives the real +# /entrypoint.sh through each scenario. +set -uo pipefail + +mkdir -p /pyls /multiplayer /debugger /out +for tool in python3 node bun; do + cp /harness/stub "/usr/local/bin/$tool" + chmod +x "/usr/local/bin/$tool" +done + +export ENABLE_LSP=true ENABLE_MULTIPLAYER=true ENABLE_DEBUGGER=true ENABLE_GATEWAY=true +# Every stub the dying one must see started before it exits. +export STUB_WAIT_FOR="lsp multiplayer debugger gateway" + +failures=0 +check() { + local what="$1" expected="$2" actual="$3" + if [ "$expected" = "$actual" ]; then + echo " ok: $what is $expected" + else + echo " FAIL: $what: expected $expected, got $actual" + failures=$((failures + 1)) + fi +} + +check_log() { + local what="$1" pattern="$2" + if grep -qE "$pattern" /out/log; then + echo " ok: $what" + else + echo " FAIL: $what (no line matching /$pattern/)" + failures=$((failures + 1)) + fi +} + +stopped() { + [ -f "/out/$1.stopped" ] && echo yes || echo no +} + +# `wait` on a backgrounded entrypoint, with a watchdog that SIGKILLs it rather +# than letting a wedged shutdown block the run. Sets $wait_status. +wait_bounded() { + local pid="$1" secs="${2:-30}" watchdog + (sleep "$secs"; kill -KILL "$pid" 2>/dev/null) & + watchdog=$! + wait "$pid" + wait_status=$? + kill "$watchdog" 2>/dev/null + wait "$watchdog" 2>/dev/null +} + +# Block until the entrypoint has written $1, so no scenario acts on a script that +# has not reached the state it is about to be tested in. +await_log() { + local pattern="$1" + for _ in $(seq 1 600); do + grep -qE "$pattern" /out/log && return 0 + sleep 0.1 + done + echo " FAIL: timed out waiting for /$pattern/ in the entrypoint log" + failures=$((failures + 1)) + return 1 +} + +echo "== a service that exits takes the container down with it" +rm -rf /out && mkdir -p /out +STUB_DIE_AS=multiplayer STUB_DIE_CODE=3 timeout -k 5 60 bash /entrypoint.sh > /out/log 2>&1 +check "exit status" 3 "$?" +cat /out/log +check_log "the dead service is named in the log" 'ERROR: Multiplayer \(PID: [0-9]+\) has exited' +# The survivors were signalled and ran their own shutdown, rather than being left +# running or killed outright. +for name in lsp debugger gateway; do + check "$name stopped cleanly" yes "$(stopped "$name")" +done + +echo +echo "== a service that exits 0 is still a failure for the container" +rm -rf /out && mkdir -p /out +STUB_DIE_AS=gateway STUB_DIE_CODE=0 timeout -k 5 60 bash /entrypoint.sh > /out/log 2>&1 +check "exit status" 1 "$?" +check_log "the dead service is named in the log" 'ERROR: Gateway \(PID: [0-9]+\) has exited' + +echo +echo "== a service that ignores SIGTERM cannot hold the container open" +rm -rf /out && mkdir -p /out +STUB_DIE_AS=multiplayer STUB_DIE_CODE=3 STUB_IGNORE_TERM=lsp SHUTDOWN_GRACE_SECS=2 \ + timeout -k 5 60 bash /entrypoint.sh > /out/log 2>&1 +check "exit status" 3 "$?" +check_log "the wedged service is killed after the grace period" \ + 'WARNING: PID [0-9]+ did not stop within 2s, killing it' +check_log "the shutdown still completes" 'All services stopped' +# The ones that do honour SIGTERM still stop the polite way. +for name in debugger gateway; do + check "$name stopped cleanly" yes "$(stopped "$name")" +done + +echo +echo "== SIGTERM is still a clean shutdown" +rm -rf /out && mkdir -p /out +bash /entrypoint.sh > /out/log 2>&1 & +entrypoint_pid=$! +if await_log 'All enabled services started'; then + kill -TERM "$entrypoint_pid" + wait_bounded "$entrypoint_pid" + check "exit status" 0 "$wait_status" + for name in lsp multiplayer debugger gateway; do + check "$name stopped cleanly" yes "$(stopped "$name")" + done +else + kill -KILL "$entrypoint_pid" 2>/dev/null +fi + +echo +echo "== with no services enabled the entrypoint sleeps instead of exiting" +rm -rf /out && mkdir -p /out +ENABLE_LSP=false ENABLE_MULTIPLAYER=false ENABLE_DEBUGGER=false ENABLE_GATEWAY=false \ + bash /entrypoint.sh > /out/log 2>&1 & +entrypoint_pid=$! +if await_log 'Sleeping indefinitely'; then + # Staying up is the correct behaviour here, so there is nothing to wait for + # but the passage of time. + sleep 2 + check "still running" yes "$(kill -0 "$entrypoint_pid" 2>/dev/null && echo yes || echo no)" +fi +# SIGKILL, not SIGTERM: this branch blocks in a foreground `sleep infinity`, so +# bash defers the trap until it returns. That is pre-existing behaviour and is +# not what this test is about. +kill -KILL "$entrypoint_pid" 2>/dev/null +wait_bounded "$entrypoint_pid" + +echo +if [ "$failures" -eq 0 ]; then + echo "all entrypoint supervision checks passed" +else + echo "$failures check(s) failed" +fi +exit "$failures" +HARNESS + +echo "Running entrypoint supervision tests in $IMAGE" + +# --entrypoint: the windmill-extra image's own ENTRYPOINT is ["/entrypoint.sh"] +# (docker/DockerfileExtra), which would swallow the command and start the real +# services instead of the harness. Harmless for a plain base image. +DOCKER_RUN=(docker run --rm --init + --entrypoint bash + -v "$ENTRYPOINT:/entrypoint.sh:ro" + -v "$WORKDIR:/harness:ro" + "$IMAGE" /harness/harness.sh) + +# Bound the whole run where the host has a `timeout` (GNU coreutils; stock macOS +# ships none), so a stalled image pull cannot hang a CI job. Every scenario +# inside the container is bounded on its own regardless. +if command -v timeout >/dev/null 2>&1; then + timeout "${ENTRYPOINT_TEST_TIMEOUT:-900}" "${DOCKER_RUN[@]}" +else + "${DOCKER_RUN[@]}" +fi diff --git a/multiplayer/server.mjs b/multiplayer/server.mjs index ee3ef60277..87b20bd66e 100644 --- a/multiplayer/server.mjs +++ b/multiplayer/server.mjs @@ -168,6 +168,13 @@ const getYDoc = (docname) => { return doc } +/** + * Every error message is peer-controlled: `applyAwarenessUpdate` runs + * `JSON.parse` on the peer's bytes and V8 quotes the offending input back. Only + * printable ASCII survives, so nothing in it can forge or flood a log line. + */ +const describeError = (error) => String(error?.message ?? error).replace(/[^\x20-\x7e]+/g, ' ').slice(0, 200) + const send = (conn, message) => { if (conn.readyState === 1) { // WebSocket.OPEN conn.send(message, err => { if (err) console.error(err) }) @@ -176,6 +183,7 @@ const send = (conn, message) => { const setupWSConnection = (conn, req, docName, bufferedMessages = []) => { const doc = getYDoc(docName) + const clientIp = req.socket.remoteAddress // Initialize awareness if (!doc.awareness) { @@ -189,22 +197,36 @@ const setupWSConnection = (conn, req, docName, bufferedMessages = []) => { doc.conns.add(conn) const messageHandler = (message) => { - const data = new Uint8Array(message) - const decoder = decoding.createDecoder(data) - const messageType = decoding.readVarUint(decoder) + // A frame that arrived before we decided to close this connection (or that a + // replay below already made moot) must not still be applied to the document. + if (conn.readyState !== 1) return // WebSocket.OPEN - switch (messageType) { - case messageSync: - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageSync) - syncProtocol.readSyncMessage(decoder, encoder, doc, null) - if (encoding.length(encoder) > 1) { - send(conn, encoding.toUint8Array(encoder)) + try { + const data = new Uint8Array(message) + const decoder = decoding.createDecoder(data) + const messageType = decoding.readVarUint(decoder) + + switch (messageType) { + case messageSync: { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + syncProtocol.readSyncMessage(decoder, encoder, doc, null) + if (encoding.length(encoder) > 1) { + send(conn, encoding.toUint8Array(encoder)) + } + break } - break - case messageAwareness: - awarenessProtocol.applyAwarenessUpdate(awareness, decoding.readVarUint8Array(decoder), conn) - break + case messageAwareness: + awarenessProtocol.applyAwarenessUpdate(awareness, decoding.readVarUint8Array(decoder), conn) + break + } + } catch (error) { + // The decoders throw on unparseable input and `ws` re-emits a listener's + // exception on the process, so one bad frame from one peer would end the + // server for every document and client. Drop only the offender, with RFC + // 6455's 1007; the frame itself is untrusted and is never logged. + console.warn(`[${new Date().toISOString()}] MALFORMED MESSAGE: doc="${docName}" from=${clientIp} error="${describeError(error)}"`) + conn.close(1007, 'Invalid message') } } conn.on('message', messageHandler) @@ -259,7 +281,12 @@ const setupWSConnection = (conn, req, docName, bufferedMessages = []) => { }) // Replay, in order, the messages that arrived while the token was being - // verified, now that the handlers above are in place. + // verified, now that the handlers above are in place. The log line is the only + // outside evidence that a frame took this path rather than the live handler, + // and marks a client that beat the JWKS fetch on a slow-starting instance. + if (bufferedMessages.length > 0) { + console.log(`[${new Date().toISOString()}] REPLAY: doc="${docName}" from=${clientIp} messages=${bufferedMessages.length}`) + } bufferedMessages.forEach(messageHandler) } @@ -287,6 +314,14 @@ const server = http.createServer((req, res) => { const wss = new WebSocketServer({ server }) +// `ws` forwards the HTTP server's errors here, and those are fatal: a failed +// listen leaves nothing to serve, and exiting 0 would read as a clean shutdown. +// `process.exitCode` rather than `process.exit()`, which truncates this line. +wss.on('error', (error) => { + console.error(`[${new Date().toISOString()}] WEBSOCKET SERVER ERROR: ${describeError(error)}`) + process.exitCode = 1 +}) + wss.on('connection', async (ws, req) => { let docName = req.url?.slice(1).split('?')[0] || 'unknown' @@ -297,6 +332,14 @@ wss.on('connection', async (ws, req) => { const clientIp = req.socket.remoteAddress + // A frame `ws` cannot parse at the protocol level fails in its Receiver, never + // reaching the handler below, and an unhandled 'error' on an EventEmitter ends + // the process. Attached before authentication, since the pre-auth window is + // exposed too; `ws` has already closed the connection by the time this runs. + ws.on('error', (error) => { + console.warn(`[${new Date().toISOString()}] SOCKET ERROR: doc="${docName}" from=${clientIp} error="${describeError(error)}"`) + }) + // Handle ping test — respond and close immediately if (docName === '__ping__') { console.log(`[${new Date().toISOString()}] WS ping from=${clientIp}`) diff --git a/multiplayer/test/cold_start.test.mjs b/multiplayer/test/cold_start.test.mjs index 4c1d37f92d..e2ac68efcb 100644 --- a/multiplayer/test/cold_start.test.mjs +++ b/multiplayer/test/cold_start.test.mjs @@ -15,13 +15,21 @@ import assert from 'node:assert/strict' import test from 'node:test' import crypto from 'node:crypto' import { setTimeout as delay } from 'node:timers/promises' -import { WebSocket } from 'ws' import * as Y from 'yjs' import * as syncProtocol from 'y-protocols/sync' import * as encoding from 'lib0/encoding' import * as decoding from 'lib0/decoding' import { mintToken, startJwksServer, startMultiplayerServer, waitFor } from './helpers.mjs' +import { + hasSyncType, + messageKind, + openClient, + syncStep1Message, + syncStep2, + syncUpdate, + updateMessage +} from './protocol.mjs' const WORKSPACE = 'test_workspace' const DOC_PATH = `${WORKSPACE}/f/foo/bar` @@ -33,53 +41,6 @@ const FLIGHT_MARGIN_MS = 250 // server demonstrably has no key, so anything it receives is in the pre-auth window. const KEY_LOADED = 'Successfully loaded Ed25519 public key' -const messageSync = 0 -const syncStep2 = 1 - -/** `messageSync` + sync step 1 for an empty document, as y-websocket sends on open. */ -function syncStep1Message() { - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageSync) - syncProtocol.writeSyncStep1(encoder, new Y.Doc()) - return encoding.toUint8Array(encoder) -} - -/** `messageSync` + a Yjs update inserting `text` into the 'content' text type. */ -function updateMessage(text) { - const doc = new Y.Doc() - doc.getText('content').insert(0, text) - const encoder = encoding.createEncoder() - encoding.writeVarUint(encoder, messageSync) - syncProtocol.writeUpdate(encoder, Y.encodeStateAsUpdate(doc)) - return encoding.toUint8Array(encoder) -} - -/** Read the (messageType, syncType) pair that prefixes a sync message. */ -function messageKind(data) { - const decoder = decoding.createDecoder(data) - const messageType = decoding.readVarUint(decoder) - if (messageType !== messageSync) return { messageType } - return { messageType, syncType: decoding.readVarUint(decoder) } -} - -/** Open a socket and collect every frame it receives, plus its close code. */ -function openClient(url, { onOpen } = {}) { - const ws = new WebSocket(url) - const received = [] - const client = { ws, received, closeCode: undefined } - ws.on('message', (data) => received.push(new Uint8Array(data))) - ws.on('close', (code) => { - client.closeCode = code - }) - ws.on('error', () => {}) - if (onOpen) ws.on('open', () => onOpen(ws)) - return client -} - -function hasKind(client, syncType) { - return client.received.some((data) => messageKind(data).syncType === syncType) -} - test('cold start: a sync step 1 sent while the key is still being fetched is answered', { timeout: 60000 }, async (t) => { const jwks = await startJwksServer({ hold: true }) const server = await startMultiplayerServer({ WINDMILL_BASE_URL: jwks.baseUrl }) @@ -112,11 +73,11 @@ test('cold start: a sync step 1 sent while the key is still being fetched is ans await delay(FLIGHT_MARGIN_MS) jwks.release() - await waitFor(() => hasKind(cold, syncStep2), { + await waitFor(() => hasSyncType(cold, syncStep2), { message: 'the server to answer the cold client with sync step 2' }) // The echo of our own update proves it was applied to the server-side doc. - await waitFor(() => hasKind(cold, 2), { + await waitFor(() => hasSyncType(cold, syncUpdate), { message: 'the server to broadcast back the update sent during the cold window' }) cold.ws.close() @@ -127,7 +88,7 @@ test('cold start: a sync step 1 sent while the key is still being fetched is ans }) t.after(() => warm.ws.close()) - await waitFor(() => hasKind(warm, syncStep2), { + await waitFor(() => hasSyncType(warm, syncStep2), { message: 'the server to answer the second client with sync step 2' }) @@ -196,7 +157,7 @@ test('cold start: a peer that floods before authenticating is closed and the ser }) t.after(() => healthy.ws.close()) - await waitFor(() => hasKind(healthy, syncStep2), { + await waitFor(() => hasSyncType(healthy, syncStep2), { message: 'the server to still answer a well-behaved client with sync step 2' }) }) diff --git a/multiplayer/test/helpers.mjs b/multiplayer/test/helpers.mjs index e998cd4961..371123d864 100644 --- a/multiplayer/test/helpers.mjs +++ b/multiplayer/test/helpers.mjs @@ -172,6 +172,12 @@ export async function startMultiplayerServer(env = {}, attemptsLeft = 5) { get output() { return output }, + /** How the child died, or null while it is still running. */ + get exitStatus() { + if (child.exitCode !== null) return `exited with code ${child.exitCode}` + if (child.signalCode !== null) return `killed by ${child.signalCode}` + return null + }, async close() { if (child.exitCode !== null) return const exited = new Promise((resolve) => child.once('exit', resolve)) @@ -181,6 +187,52 @@ export async function startMultiplayerServer(env = {}, attemptsLeft = 5) { } } +/** + * Start server.mjs and wait for it to exit, for the cases where it is expected + * to fail rather than come up. Returns how it died and everything it printed. + */ +export async function runMultiplayerServerUntilExit(env = {}, { timeoutMs = 15000 } = {}) { + const child = spawn(process.execPath, [SERVER_PATH], { + env: { + ...process.env, + REQUIRE_SIGNED_MULTIPLAYER_REQUESTS: 'true', + BASE_INTERNAL_URL: '', + HOST: '127.0.0.1', + ...env + }, + stdio: ['ignore', 'pipe', 'pipe'] + }) + + let output = '' + child.stdout.on('data', (chunk) => { output += chunk.toString() }) + child.stderr.on('data', (chunk) => { output += chunk.toString() }) + + // Record the rescue rather than inferring it from the exit signal: `signal` is + // null for every child exit on Windows, so a caller checking it could not tell + // a server that exited on its own from one this had to kill. + let killedByTimeout = false + const killer = setTimeout(() => { + // Claim the rescue only if there was something to rescue. The child may have + // exited moments ago with 'close' still pending on the stdio drain — the very + // window this helper waits for — and killing a corpse is not an intervention. + if (child.exitCode !== null || child.signalCode !== null) return + killedByTimeout = child.kill('SIGKILL') + }, timeoutMs) + + try { + // 'close' rather than 'exit': 'exit' fires when the child terminates, which + // can be before its stdio pipes have been drained, and the caller reads + // `output`. A child that never starts emits neither, only 'error'. + const [code, signal] = await new Promise((resolve, reject) => { + child.once('close', (exitCode, exitSignal) => resolve([exitCode, exitSignal])) + child.once('error', reject) + }) + return { code, signal, output, killedByTimeout } + } finally { + clearTimeout(killer) + } +} + /** Poll `predicate` until it is true, or throw after `timeoutMs`. */ export async function waitFor(predicate, { timeoutMs = 15000, intervalMs = 10, message = 'condition' } = {}) { const deadline = Date.now() + timeoutMs diff --git a/multiplayer/test/malformed_frame.test.mjs b/multiplayer/test/malformed_frame.test.mjs new file mode 100644 index 0000000000..8ba340de37 --- /dev/null +++ b/multiplayer/test/malformed_frame.test.mjs @@ -0,0 +1,231 @@ +/** + * Regression tests for malformed frames sent by an authenticated peer. + * + * The 'message' handler in `setupWSConnection` decodes whatever arrives with + * `decoding.readVarUint`, `syncProtocol.readSyncMessage` and + * `awarenessProtocol.applyAwarenessUpdate`. All three throw on input they cannot + * parse, and `ws` emits the listener's exception on the process. With no + * `uncaughtException` handler installed, one such frame used to take the whole + * multiplayer server down — every other document and every other client with it. + * + * The payloads below are the ones that reproduce it: 600 KiB of zeros decodes as + * sync step 1 with an empty state vector, which makes `Y.encodeStateAsUpdate` + * throw "Unexpected end of array"; the truncated frames run off the end of the + * buffer inside the decoder. + */ + +import assert from 'node:assert/strict' +import test from 'node:test' +import { setTimeout as delay } from 'node:timers/promises' +import * as encoding from 'lib0/encoding' + +import { mintToken, startJwksServer, startMultiplayerServer, waitFor } from './helpers.mjs' +import { + hasSyncType, + messageAwareness, + messageSync, + openClient, + syncedDoc, + syncStep1, + syncStep1Message, + syncStep2, + updateMessage +} from './protocol.mjs' + +const WORKSPACE = 'test_workspace' +const DOC_PATH = `${WORKSPACE}/f/foo/bar` +// server.mjs closes a connection whose frame it could not decode with 1007 +// ("invalid frame payload data"). +const INVALID_PAYLOAD = 1007 +// Logged by server.mjs for every frame it refused; asserted on so the tests +// cannot pass on a connection that was closed for some unrelated reason. +const REFUSED = 'MALFORMED MESSAGE' +// Logged by server.mjs once the JWKS fetch resolves. While it is absent the +// server demonstrably has no key, so anything it received is in the pre-auth window. +const KEY_LOADED = 'Successfully loaded Ed25519 public key' +// Grace for a frame the client has already written to be delivered over loopback +// and read by the (otherwise idle) server, before the key is released. Only the +// already-written bytes have to land in this time, and the REPLAY assertion +// below fails loudly if they did not, rather than quietly testing the wrong path. +const FLIGHT_MARGIN_MS = 500 +// Logged by server.mjs when, and only when, it replays messages that were +// buffered before authentication. +const REPLAYED = 'REPLAY:' +// A control character a peer would use to forge log lines or drive a terminal. +const ESC = '\x1b' + +/** + * 600 KiB of zeros: `messageSync`, then sync step 1 with a zero-length state + * vector, which `Y.encodeStateAsUpdate` cannot decode. + */ +function zeroFlood() { + return Buffer.alloc(600 * 1024) +} + +/** `messageSync` with nothing after it: `readSyncMessage` reads past the end. */ +function truncatedSyncMessage() { + return Uint8Array.from([messageSync]) +} + +/** `messageSync` + sync step 1 whose state vector is shorter than it claims. */ +function truncatedStateVector() { + return Uint8Array.from([messageSync, syncStep1, 8, 1, 2, 3]) +} + +/** `messageAwareness` + an awareness payload shorter than it claims. */ +function truncatedAwarenessMessage() { + return Uint8Array.from([messageAwareness, 8, 1, 2, 3]) +} + +/** + * `messageAwareness` + one well-formed entry whose state is not JSON, so + * `applyAwarenessUpdate` fails inside `JSON.parse`. V8 quotes the offending + * input back in its message, which is how peer bytes — ESC included — can reach + * a log line that interpolates `error.message`. + */ +function awarenessWithUnparseableState() { + const update = encoding.createEncoder() + encoding.writeVarUint(update, 1) // one client + encoding.writeVarUint(update, 42) // client id + encoding.writeVarUint(update, 1) // clock + // Not starting with '{': V8 only quotes the input back for a value that is + // invalid from position 0, which is the case this needs to reproduce. + encoding.writeVarString(update, `x${ESC}[2J OWNED THE LOG`) + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageAwareness) + encoding.writeVarUint8Array(encoder, encoding.toUint8Array(update)) + return encoding.toUint8Array(encoder) +} + +const MALFORMED = [ + ['600 KiB of zeros', zeroFlood], + ['a truncated sync message', truncatedSyncMessage], + ['a truncated sync step 1 state vector', truncatedStateVector], + ['a truncated awareness update', truncatedAwarenessMessage], + ['an awareness state that is not JSON', awarenessWithUnparseableState] +] + +/** Connect, wait for the server's sync step 1, i.e. for the peer to be authenticated. */ +async function connectAuthenticated(server, token, { onOpen } = {}) { + const client = openClient(`${server.url}/${DOC_PATH}?token=${token}`, { onOpen }) + await waitFor(() => hasSyncType(client, syncStep1) || client.closeCode !== undefined, { + message: 'the server to send sync step 1' + }) + assert.equal(client.closeCode, undefined, 'connection was closed before it was set up') + return client +} + +for (const [label, payload] of MALFORMED) { + test(`an authenticated client sending ${label} is closed, and the server survives`, { timeout: 60000 }, async (t) => { + const jwks = await startJwksServer() + const server = await startMultiplayerServer({ WINDMILL_BASE_URL: jwks.baseUrl }) + t.after(async () => { + await server.close() + await jwks.close() + }) + + const token = mintToken(jwks.privateKey, { workspaceId: WORKSPACE }) + + // A bystander on the same document, synced before anything goes wrong. + const bystander = await connectAuthenticated(server, token, { + onOpen: (ws) => ws.send(syncStep1Message()) + }) + await waitFor(() => hasSyncType(bystander, syncStep2), { + message: 'the server to answer the bystander with sync step 2' + }) + t.after(() => bystander.ws.close()) + + // The offender is fully authenticated — the server has already answered it + // with sync step 1 — so this frame goes through the live message handler, + // not the pre-auth buffer. + const offender = await connectAuthenticated(server, token) + offender.ws.send(payload()) + + await waitFor(() => offender.closeCode !== undefined, { + message: 'the offending connection to be closed' + }) + assert.equal(offender.closeCode, INVALID_PAYLOAD) + assert.equal(server.exitStatus, null, `server died: ${server.output}`) + // The frame itself is never logged: 600 KiB of zeros must not reach the log. + assert.ok(server.output.length < 8192, `server logged ${server.output.length} bytes, payload leaked?`) + + // Exactly one refusal line, and every assertion about it is made against the + // line itself: CONNECT and DISCONNECT also name the document, so checking the + // whole log would pass even if the refusal stopped naming anything. + const refusals = server.output.split('\n').filter((line) => line.includes(REFUSED)) + assert.equal(refusals.length, 1, `expected one refusal line, got:\n${server.output}`) + assert.ok(refusals[0].includes(`doc="${DOC_PATH}"`), `the refusal must name the document: ${refusals[0]}`) + assert.ok(refusals[0].includes('from='), `the refusal must name the peer: ${refusals[0]}`) + // Nothing a peer put in the frame can escape that line: an error message can + // quote the payload (V8 does so for JSON.parse), so it must carry no control + // characters at all. + assert.doesNotMatch(refusals[0], /[\x00-\x1f\x7f]/, 'the refusal line must have no control characters') + + // The bystander was not disturbed, and a new client still syncs. + assert.equal(bystander.closeCode, undefined) + const latecomer = await connectAuthenticated(server, token, { + onOpen: (ws) => ws.send(syncStep1Message()) + }) + t.after(() => latecomer.ws.close()) + await waitFor(() => hasSyncType(latecomer, syncStep2), { + message: 'the server to answer a new client with sync step 2' + }) + + // And a real edit still propagates from one client to the other. + const bystanderDoc = syncedDoc(bystander) + latecomer.ws.send(updateMessage('hello')) + await waitFor(() => bystanderDoc().getText('content').toString() === 'hello', { + message: 'the edit to propagate to the bystander' + }) + }) +} + +test('a malformed frame replayed from the pre-auth buffer is refused, not fatal', { timeout: 60000 }, async (t) => { + const jwks = await startJwksServer({ hold: true }) + const server = await startMultiplayerServer({ WINDMILL_BASE_URL: jwks.baseUrl }) + t.after(async () => { + await server.close() + await jwks.close() + }) + + const token = mintToken(jwks.privateKey, { workspaceId: WORKSPACE }) + + // The key is parked, so this frame is buffered by the pre-auth path and only + // decoded later, when `setupWSConnection` replays it. One 600 KiB frame stays + // under MAX_PREAUTH_BYTES (1 MiB), so the flood cap cannot be what closes it. + await waitFor(() => jwks.requests >= 1, { message: 'the server to request the JWKS' }) + let framesWritten = 0 + const offender = openClient(`${server.url}/${DOC_PATH}?token=${token}`, { + onOpen: (ws) => ws.send(zeroFlood(), () => framesWritten++) + }) + await waitFor(() => framesWritten === 1, { message: 'the offending client to write its frame' }) + // The frame is on the wire while the server demonstrably has no key, so it + // cannot have been authenticated yet. + assert.ok(!server.output.includes(KEY_LOADED)) + await delay(FLIGHT_MARGIN_MS) + jwks.release() + + await waitFor(() => offender.closeCode !== undefined, { + message: 'the offending connection to be closed' + }) + assert.equal(offender.closeCode, INVALID_PAYLOAD) + // Without this the test would pass just as well if the frame had arrived after + // setup and gone through the live handler — the same close code, the same log + // line — so it would silently stop covering the replay path it is named for. + // server.mjs logs REPLAY only when it replays a pre-auth buffer. + assert.ok( + server.output.includes(`${REPLAYED} doc="${DOC_PATH}" from=127.0.0.1 messages=1`), + `the frame did not go through the pre-auth buffer:\n${server.output}` + ) + assert.equal(server.exitStatus, null, `server died: ${server.output}`) + assert.ok(server.output.includes(REFUSED), `server did not log the refused frame:\n${server.output}`) + + // The server is unharmed and still syncs a well-behaved client. + const healthy = await connectAuthenticated(server, token, { + onOpen: (ws) => ws.send(syncStep1Message()) + }) + t.after(() => healthy.ws.close()) + await waitFor(() => hasSyncType(healthy, syncStep2), { + message: 'the server to still answer a well-behaved client with sync step 2' + }) +}) diff --git a/multiplayer/test/protocol.mjs b/multiplayer/test/protocol.mjs new file mode 100644 index 0000000000..77aeea3db9 --- /dev/null +++ b/multiplayer/test/protocol.mjs @@ -0,0 +1,94 @@ +/** + * y-websocket protocol helpers shared by the multiplayer server tests: + * building the frames a real client sends, reading back what the server + * answers, and a socket wrapper that records both. + */ + +import { WebSocket } from 'ws' +import * as Y from 'yjs' +import * as syncProtocol from 'y-protocols/sync' +import * as encoding from 'lib0/encoding' +import * as decoding from 'lib0/decoding' + +// Message types, as server.mjs numbers them. +export const messageSync = 0 +export const messageAwareness = 1 + +// y-protocols/sync sub-types. These overlap numerically with the message types +// above (`syncStep1 === messageSync === 0`), so no runtime check can catch a +// caller that passes the wrong family: `hasSyncType` takes only these three. +export const syncStep1 = 0 +export const syncStep2 = 1 +export const syncUpdate = 2 + +/** `messageSync` + sync step 1 for an empty document, as y-websocket sends on open. */ +export function syncStep1Message() { + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + syncProtocol.writeSyncStep1(encoder, new Y.Doc()) + return encoding.toUint8Array(encoder) +} + +/** `messageSync` + a Yjs update inserting `text` into the 'content' text type. */ +export function updateMessage(text) { + const doc = new Y.Doc() + doc.getText('content').insert(0, text) + const encoder = encoding.createEncoder() + encoding.writeVarUint(encoder, messageSync) + syncProtocol.writeUpdate(encoder, Y.encodeStateAsUpdate(doc)) + return encoding.toUint8Array(encoder) +} + +/** Read the (messageType, syncType) pair that prefixes a sync message. */ +export function messageKind(data) { + const decoder = decoding.createDecoder(data) + const messageType = decoding.readVarUint(decoder) + if (messageType !== messageSync) return { messageType } + return { messageType, syncType: decoding.readVarUint(decoder) } +} + +/** + * A live mirror of what `client` has been sent: one document that each call + * brings up to date with the frames received since the last one. Cheap enough + * to call from a polling predicate, and never re-applies a frame. + */ +export function syncedDoc(client) { + const doc = new Y.Doc() + let applied = 0 + return () => { + while (applied < client.received.length) { + const data = client.received[applied++] + if (messageKind(data).messageType !== messageSync) continue + const decoder = decoding.createDecoder(data) + decoding.readVarUint(decoder) // messageSync + syncProtocol.readSyncMessage(decoder, encoding.createEncoder(), doc, null) + } + return doc + } +} + +/** Open a socket and collect every frame it receives, plus its close code. */ +export function openClient(url, { onOpen } = {}) { + const ws = new WebSocket(url) + const received = [] + const client = { ws, received, closeCode: undefined, lastError: undefined } + // Under the default binaryType 'nodebuffer' `ws` hands every frame over as a + // Buffer, text frames included, so this is lossless for both. + ws.on('message', (data) => received.push(new Uint8Array(data))) + ws.on('close', (code) => { + client.closeCode = code + }) + // Keep the error rather than only silencing it: without a listener `ws` would + // throw, and without the record a failed connection shows up as a `waitFor` + // timeout with no cause attached. + ws.on('error', (error) => { + client.lastError = error + }) + if (onOpen) ws.on('open', () => onOpen(ws)) + return client +} + +/** Has `client` received a sync message of this sub-type (never a message type)? */ +export function hasSyncType(client, syncType) { + return client.received.some((data) => messageKind(data).syncType === syncType) +} diff --git a/multiplayer/test/socket_error.test.mjs b/multiplayer/test/socket_error.test.mjs new file mode 100644 index 0000000000..4978906a50 --- /dev/null +++ b/multiplayer/test/socket_error.test.mjs @@ -0,0 +1,128 @@ +/** + * Regression tests for WebSocket-level protocol errors. + * + * A frame the `ws` library itself cannot parse — an unmasked frame from a + * client, a reserved opcode, a bad RSV bit — never reaches the application's + * 'message' handler. `ws` fails it in its Receiver and emits an 'error' on the + * WebSocket (`receiverOnError` in ws/lib/websocket.js), and an unhandled 'error' + * on an EventEmitter throws, so the process used to exit here too — the same + * blast radius as a malformed application payload, reached one layer lower. + * + * Raw bytes are written straight to the TCP socket, since the `ws` client would + * never produce an illegal frame on its own. + */ + +import assert from 'node:assert/strict' +import net from 'node:net' +import test from 'node:test' + +import { + mintToken, + runMultiplayerServerUntilExit, + startJwksServer, + startMultiplayerServer, + waitFor +} from './helpers.mjs' +import { hasSyncType, openClient, syncStep1, syncStep1Message, syncStep2 } from './protocol.mjs' + +const WORKSPACE = 'test_workspace' +const DOC_PATH = `${WORKSPACE}/f/foo/bar` +// Logged by server.mjs for every connection `ws` failed at the protocol level. +const SOCKET_ERROR = 'SOCKET ERROR' +// RFC 6455 "protocol error": what `ws` closes with when its Receiver rejects a +// frame, carried on the Receiver's error as `Symbol(status-code)`. +const PROTOCOL_ERROR = 1002 + +/** + * A FIN + text frame of 3 bytes with the MASK bit clear. RFC 6455 requires every + * client-to-server frame to be masked, so `ws` rejects it with + * WS_ERR_EXPECTED_MASK (close status 1002). + */ +const UNMASKED_FRAME = Buffer.from([0x81, 0x03, 0x61, 0x62, 0x63]) + +/** Assert the server is alive by making it serve a fresh client. */ +async function assertStillServing(t, server, token) { + const healthy = openClient(`${server.url}/${DOC_PATH}?token=${token}`, { + onOpen: (ws) => ws.send(syncStep1Message()) + }) + t.after(() => healthy.ws.close()) + // Resolve as soon as either outcome is settled, so a dead server fails fast + // and with its own output rather than by timing out. + await waitFor(() => hasSyncType(healthy, syncStep2) || server.exitStatus !== null, { + message: 'the server to answer a new client with sync step 2' + }) + assert.equal(server.exitStatus, null, `server died: ${server.output}`) + assert.ok(hasSyncType(healthy, syncStep2)) +} + +test('an illegal WebSocket frame from an authenticated client does not exit the server', { timeout: 60000 }, async (t) => { + const jwks = await startJwksServer() + const server = await startMultiplayerServer({ WINDMILL_BASE_URL: jwks.baseUrl }) + t.after(async () => { + await server.close() + await jwks.close() + }) + + const token = mintToken(jwks.privateKey, { workspaceId: WORKSPACE }) + + const offender = openClient(`${server.url}/${DOC_PATH}?token=${token}`) + // The server's sync step 1 means this peer is past authentication. + await waitFor(() => hasSyncType(offender, syncStep1), { message: 'the server to send sync step 1' }) + offender.ws._socket.write(UNMASKED_FRAME) + + await waitFor(() => offender.closeCode !== undefined, { + message: 'the offending connection to be closed' + }) + assert.equal(offender.closeCode, PROTOCOL_ERROR) + assert.equal(server.exitStatus, null, `server died: ${server.output}`) + assert.ok(server.output.includes(SOCKET_ERROR), `server did not log the socket error:\n${server.output}`) + + await assertStillServing(t, server, token) +}) + +test('an illegal WebSocket frame before authentication does not exit the server', { timeout: 60000 }, async (t) => { + const jwks = await startJwksServer({ hold: true }) + const server = await startMultiplayerServer({ WINDMILL_BASE_URL: jwks.baseUrl }) + t.after(async () => { + await server.close() + await jwks.close() + }) + + const token = mintToken(jwks.privateKey, { workspaceId: WORKSPACE }) + + // The JWKS response is parked and never released before the assertions below, + // so the server cannot authenticate anyone for the whole of this test. + const offender = openClient(`${server.url}/${DOC_PATH}?token=${token}`, { + onOpen: (ws) => ws._socket.write(UNMASKED_FRAME) + }) + + await waitFor(() => offender.closeCode !== undefined, { + message: 'the offending connection to be closed' + }) + assert.equal(offender.closeCode, PROTOCOL_ERROR) + // server.mjs logs CONNECT only once a peer is past verification, so its absence + // is what makes this the pre-auth case rather than a repeat of the test above. + assert.ok(!server.output.includes('CONNECT:'), `a connection was accepted:\n${server.output}`) + assert.equal(server.exitStatus, null, `server died: ${server.output}`) + assert.ok(server.output.includes(SOCKET_ERROR), `server did not log the socket error:\n${server.output}`) + + jwks.release() + await assertStillServing(t, server, token) +}) + +test('a server-level error is fatal, not swallowed', { timeout: 60000 }, async (t) => { + // Hold the port so server.mjs's listen fails with EADDRINUSE. `ws` forwards + // the HTTP server's errors to the WebSocketServer, so this is what reaches the + // `wss.on('error')` handler — and a process with no listening socket must not + // report a clean exit, or nothing upstream knows to replace it. + const blocker = net.createServer() + await new Promise((resolve) => blocker.listen(0, '127.0.0.1', resolve)) + const { port } = blocker.address() + t.after(() => new Promise((resolve) => blocker.close(resolve))) + + const { code, output, killedByTimeout } = await runMultiplayerServerUntilExit({ PORT: String(port) }) + + assert.equal(killedByTimeout, false, `the server had to be killed rather than exiting:\n${output}`) + assert.ok(output.includes('EADDRINUSE'), `expected a listen failure, got:\n${output}`) + assert.notEqual(code, 0, `expected a non-zero exit, got ${code}:\n${output}`) +})