import assert from 'node:assert/strict'; import {randomUUID} from 'node:crypto'; import {EventEmitter} from 'node:events'; import {readFile, mkdtemp, mkdir, writeFile, rm, unlink} from 'node:fs/promises'; import {tmpdir} from 'node:os'; import path from 'node:path'; import {createRequire} from 'node:module'; import {setTimeout as delay} from 'node:timers/promises'; import {fileURLToPath} from 'node:url'; import test from 'node:test'; import {build} from 'esbuild'; import {io} from 'socket.io-client'; import {tsImport} from 'tsx/esm/api'; const root = fileURLToPath(new URL('../', import.meta.url)); const vectors = JSON.parse(await readFile(new URL('./fixtures/forward_vectors.json', import.meta.url), 'utf8')); const compiled = await build({ stdin: {contents: `export * from './bot/forward/manager.ts'; export * from './bot/forward/stream.ts'; export * from './bot/forward/files.ts'; export * from './utils/forward_protocol.ts';`, resolveDir: root, loader: 'ts'}, bundle: true, platform: 'node', format: 'cjs', packages: 'external', write: false, metafile: true, plugins: [{name: 'isolated-default-bot-registry', setup(build) { build.onResolve({filter: /^\.\.\/instance\/index\.ts$/}, () => ({path: 'registry', namespace: 'test'})); build.onLoad({filter: /.*/, namespace: 'test'}, () => ({loader: 'js', contents: 'export const get_bot_context = () => undefined; export const observe_bot_context = () => () => {};'})); }}], }); const module = {exports: {}}; new Function('require', 'module', 'exports', compiled.outputFiles[0].text)(createRequire(import.meta.url), module, module.exports); const {NodeForwardManager, BotStream, BotFileService, validate_command, validate_join, validate_rpc, success_data, error_ack} = module.exports; const success = data => ({status: 'success', data}); const scope = () => ({forward_id: 1, forward_session_id: randomUUID(), node_session_id: 'main-socket-session', node_id: 2, bot_id: 3}); const command = (s = vectors.scope, overrides = {}) => ({...s, url: 'http://127.0.0.1:1', path: '/relay/socket.io', namespace: '/forward', token: vectors.token, expires_at: Date.now() + 10_000, deadline_at: Date.now() + 2000, ...overrides}); const request = (method = 'bot.snapshot', payload = {}, request_id = randomUUID()) => ({request_id, method, payload}); const node_request = (s, overrides = {}) => ({...request(), correlation_id: randomUUID(), forward_session_id: s.forward_session_id, deadline_at: Date.now() + 1000, ...overrides}); function code(ack, expected) { assert.equal(ack.status, 'error'); assert.deepEqual(Object.keys(ack.data).sort(), ['code', 'message', 'retryable']); assert.equal(ack.data.code, expected); assert.equal(typeof ack.data.message, 'string'); assert.equal(typeof ack.data.retryable, 'boolean'); } const rejects = (action, expected = 'INVALID_REQUEST') => assert.throws(action, error => error.code === expected); function received(socket, event, predicate = () => true) { return new Promise((resolve, reject) => { const timer = setTimeout(() => { socket.off(event, listener); reject(new Error(`Missing ${event}`)); }, 2500); const listener = value => { if (!predicate(value)) return; clearTimeout(timer); socket.off(event, listener); resolve(value); }; socket.on(event, listener); }); } const ack = (socket, name, payload) => socket.timeout(2500).emitWithAck(name, payload); class FakeSocket extends EventEmitter { connected = false; packets = []; io = {engine: {writeBuffer: []}}; volatile = {emit: (event, value) => this.packets.push([event, value])}; constructor(cmd, responder) { super(); this.cmd = cmd; this.responder = responder; } connect() { queueMicrotask(() => {this.connected = true; super.emit('connect');}); } disconnect() { if (this.connected) {this.connected = false; super.emit('disconnect', 'test');} } timeout() { return {emit: (event, payload, callback) => { this.packets.push([event, payload]); const data = event === 'forward_node_ready' ? {node_ready: true} : { ...Object.fromEntries(Object.keys(vectors.scope).map(key => [key, this.cmd[key]])), role: 'node', node_ready: event === 'reauth', expires_at: this.cmd.expires_at, }; if (this.responder) this.responder(event, success(data), callback); else queueMicrotask(() => callback(null, success(data))); }}; } rpc(raw) { return new Promise(resolve => super.emit('relay_request', raw, resolve)); } } function fake_manager(t, cmd, responder, stream_factory) { const sockets = []; const manager = new NodeForwardManager(cmd.node_id, cmd.node_session_id, { socket_factory: (_url, options) => { const socket = new FakeSocket(cmd, responder); socket.options = options; sockets.push(socket); return socket; }, stream_factory, }); t.after(() => manager.stop()); return {manager, sockets}; } test('independent Node bundle has no relay or token issuer imports; fixed token stays opaque and byte-identical', async t => { assert.ok(Object.keys(compiled.metafile.inputs).every(path => !path.includes('forward_server'))); assert.doesNotMatch(compiled.outputFiles[0].text, /createCipheriv|createDecipheriv|encode_token|decode_token/); const cmd = command(); const f = fake_manager(t, cmd); assert.equal(validate_command(cmd).token, vectors.token); const ready = await f.manager.ensure(cmd); assert.deepEqual(ready, success({...vectors.scope, node_ready: true, lease_expires_at: cmd.expires_at})); assert.equal(f.sockets[0].options.auth.token, vectors.token); assert.equal(f.sockets[0].options.reconnection, false); assert.deepEqual(f.sockets[0].packets, [ ['join_room', {token: vectors.token}], ['forward_node_ready', {forward_session_id: cmd.forward_session_id}], ]); // The copied token has Web claims and fixed historical times. Node never interprets them. const opaque = command({...vectors.scope, forward_session_id: randomUUID()}, {token: 'opaque-not-a-token'}); const second = fake_manager(t, opaque); assert.equal((await second.manager.ensure(opaque)).status, 'success'); assert.equal(second.sockets[0].packets[0][1].token, opaque.token); }); test('local command validator enforces exact wire DTO, scope IDs, URL and numeric bounds', () => { const valid = command(); for (const key of Object.keys(valid)) { const missing = {...valid}; delete missing[key]; rejects(() => validate_command(missing)); } for (const patch of [{extra: 'SECRET'}, {node_id: 0}, {bot_id: '2'}, {forward_id: -1}, {forward_session_id: 'session'}, {node_session_id: ' '}, {node_session_id: '\u754c'.repeat(86)}, {token: ''}, {token: 'x'.repeat(8193)}, {deadline_at: Infinity}, {expires_at: 1.5}, {url: 'file:///tmp/relay'}, {url: 'http://user:password@localhost'}, {url: 'http://localhost/path'}, {url: 'http://localhost?token=SECRET'}, {url: 'http://localhost#SECRET'}, {path: 'socket.io'}, {namespace: '/bad?x'}]) { rejects(() => validate_command({...valid, ...patch})); } assert.equal(validate_command({...valid, forward_id: 0, node_session_id: 'x'.repeat(256)}).forward_id, 0); assert.equal(validate_command({...valid, forward_session_id: '11111111-1111-1111-1111-111111111111'}).forward_session_id, '11111111-1111-1111-1111-111111111111'); }); test('local Node RPC validator allows only exact public method payloads and relay correlation metadata', () => { const valid = node_request(vectors.scope, {request_id: vectors.request_id}); for (const [method, payload] of [['bot.snapshot', {}], ['bot.logs', {limit: 200, stream_id: randomUUID(), after_seq: 0}], ['get_log_history', {}], ['get_context_info', {}], ['get_data_file_list', {}], ['get_file_data', {id: randomUUID()}], ['get_file_data', {id: randomUUID(), offset: 0, length: 262144}], ['tab_complete', {text: '/he'}], ['tab_complete', {text: ''}], ['bot.chat', {message: 'hello'}], ['execute_function', {function: 'inventory', name: 'left_click_slot', args: [9]}]]) assert.equal(validate_rpc({...valid, method, payload}).method, method); for (const payload of [{}, {function: '../inventory', name: 'left_click_slot', args: []}, {function: 'inventory', name: '', args: []}, {function: 'inventory', name: 'x', args: {}}, {function: 'inventory', name: 'x', args: [], extra: true}]) rejects(() => validate_rpc({...valid, method: 'execute_function', payload})); for (const payload of [{}, {text: 1}, {text: 'x'.repeat(257)}, {text: '/he\n'}, {text: '/he', extra: true}]) rejects(() => validate_rpc({...valid, method: 'tab_complete', payload})); rejects(() => validate_rpc({...valid, method: 'get_log_history', payload: {limit: 21}})); rejects(() => validate_rpc({...valid, method: 'get_context_info', payload: {name: 'health'}})); rejects(() => validate_rpc({...valid, method: 'get_data_file_list', payload: {path: '../'}})); for (const payload of [{id: '../secret'}, {id: randomUUID(), path: '/tmp'}, {id: randomUUID(), offset: -1}, {id: randomUUID(), offset: 0.5}, {id: randomUUID(), length: 0}, {id: randomUUID(), length: 262145}]) rejects(() => validate_rpc({...valid, method: 'get_file_data', payload})); for (const key of Object.keys(valid)) { const missing = {...valid}; delete missing[key]; rejects(() => validate_rpc(missing)); } for (const patch of [{token: 'SECRET'}, {request_id: 'bad'}, {correlation_id: 'bad'}, {deadline_at: NaN}, {method: 'bot.delete'}, {method: 'bot.subscribe', payload: {events: ['log']}}, {payload: {config: 'SECRET'}}, {method: 'bot.logs', payload: {limit: 0}}, {method: 'bot.logs', payload: {limit: 201}}, {method: 'bot.logs', payload: {limit: 1, after_seq: 0}}, {method: 'bot.chat', payload: {message: 'hello', password: 'SECRET'}}, ...['', ' ', 'x'.repeat(257), 'hello\nworld', 'hello\rworld', 'hello\0world'].map(message => ({method: 'bot.chat', payload: {message}}))]) { rejects(() => validate_rpc({...valid, ...patch})); } }); test('local join/ACK validation binds every scope field, lease, role and readiness without exposing errors', () => { const cmd = command(); const data = {...vectors.scope, role: 'node', node_ready: false, expires_at: cmd.expires_at}; assert.deepEqual(validate_join(success(data), cmd, true), data); for (const patch of [{forward_id: 9}, {forward_session_id: randomUUID()}, {node_session_id: 'old'}, {node_id: 9}, {bot_id: 9}, {role: 'web'}, {node_ready: true}, {expires_at: cmd.expires_at + 1}, {secret: 'SECRET'}]) { rejects(() => validate_join(success({...data, ...patch}), cmd, true), 'SESSION_STALE'); } for (const value of [null, {}, {status: 'success'}, {status: 'success', data, extra: true}, {status: 'error', data: {code: 'SECRET', message: 'SECRET', retryable: false}}]) { rejects(() => success_data(value), 'RELAY_UNAVAILABLE'); } rejects(() => success_data(error_ack('TIMEOUT')), 'TIMEOUT'); }); test('manager rejects stale scopes and expired commands before creating a transport', async t => { const cmd = command(); const f = fake_manager(t, cmd); for (const [patch, expected] of [[{node_session_id: 'old'}, 'SESSION_STALE'], [{node_id: 99}, 'FORBIDDEN'], [{deadline_at: Date.now() - 1}, 'TIMEOUT'], [{expires_at: Date.now() - 1}, 'TOKEN_EXPIRED']]) { code(await f.manager.ensure({...cmd, ...patch}), expected); } assert.equal(f.sockets.length, 0); f.manager.stop(); code(await f.manager.ensure(cmd), 'SESSION_STALE'); }); test('manager deadline rejects late join ACK and prevents readiness or later RPC side effects', async t => { const cmd = command(vectors.scope, {deadline_at: Date.now() + 40}); let late; const f = fake_manager(t, cmd, (_event, response, callback) => {late = () => callback(null, response);}); code(await f.manager.ensure(cmd), 'TIMEOUT'); late(); await delay(10); assert.equal(f.sockets[0].connected, false); assert.equal(f.sockets[0].packets.some(([name]) => name === 'forward_node_ready'), false); assert.equal(f.sockets[0].listenerCount('relay_request'), 0); }); test('manager independently enforces RPC deadlines, session binding, correlation replay and safe error ACKs', async t => { const cmd = command(); let calls = 0; const stream = {subscribe: () => () => {}, resync() {}, dispose() {}, snapshot: () => ({seq: ++calls}), chat() {calls++; throw new Error('SECRET password/stack');}}; const f = fake_manager(t, cmd, undefined, () => stream); assert.equal((await f.manager.ensure(cmd)).status, 'success'); const socket = f.sockets[0]; const rpc = node_request(cmd); code(await socket.rpc({...rpc, deadline_at: Date.now() - 1}), 'TIMEOUT'); code(await socket.rpc({...rpc, forward_session_id: randomUUID()}), 'SESSION_STALE'); code(await socket.rpc({...rpc, method: 'bot.chat', payload: {message: 'hello', token: 'SECRET'}}), 'INVALID_REQUEST'); assert.equal(calls, 0); assert.deepEqual(await socket.rpc(rpc), success({seq: 1})); code(await socket.rpc(rpc), 'CANCELLED'); const failed = await socket.rpc(node_request(cmd, {method: 'bot.chat', payload: {message: 'hello'}})); code(failed, 'INTERNAL_ERROR'); assert.equal(JSON.stringify(failed).includes('SECRET'), false); assert.equal(calls, 2); }); // The integration layer alone loads relay code. Issuing credentials is test-only. test('disk history RPC discards responses after deadline or transport replacement', async t => { const cmd = command(); let release; const stream = {subscribe: () => () => {}, resync() {}, dispose() {}, get_log_history: () => new Promise(resolve => {release = resolve;})}; const f = fake_manager(t, cmd, undefined, () => stream); assert.equal((await f.manager.ensure(cmd)).status, 'success'); const socket = f.sockets[0]; const pending = socket.rpc(node_request(cmd, {method: 'get_log_history', payload: {}, deadline_at: Date.now() + 15})); await delay(25); release({records: []}); code(await pending, 'TIMEOUT'); const stale = socket.rpc(node_request(cmd, {method: 'get_log_history', payload: {}})); socket.disconnect(); release({records: []}); code(await stale, 'SESSION_STALE'); }); async function real_fixture(t, limits = {}) { const [{start_forward_server}, {encode_token}] = await Promise.all([ tsImport('../forward_server/index.ts', import.meta.url), tsImport('../forward_server/utils/token_utils.ts', import.meta.url), ]); const key = 'node-forward-integration-test-secret-only'; const control = new EventEmitter(); const statuses = []; control.on('forward_session_status', value => statuses.push(value)); const handle = await start_forward_server({forward_id: 1, key, host: '127.0.0.1', port: 0, path: '/node-test/socket_io', namespace: '/node-test', allowed_origins: [], allow_no_origin: true, control, limits: {rpc_timeout_ms: 250, unused_timeout_ms: 10_000, idle_timeout_ms: 10_000, clock_skew_ms: 0, ...limits}}); const url = `http://127.0.0.1:${handle.http_server.address().port}`; const clients = []; const node_sockets = []; const managers = []; const sources = new Map(); const streams = new Map(); const packets = []; handle.io.of(handle.namespace).on('connection', socket => socket.onAny((event, value) => packets.push({id: socket.id, event, value}))); t.after(async () => {for (const manager of managers) manager.stop(); for (const socket of clients) socket.disconnect(); await handle.stop();}); const control_ack = (event, value) => new Promise(resolve => control.emit(event, value, resolve)); const permissions = ['bot.snapshot', 'bot.logs', 'bot.subscribe', 'bot.unsubscribe', 'bot.chat', 'execute_function', 'bot.files']; function token(s, role, expires_at, grant_id, granted = permissions) { const now = Date.now(); return encode_token(key, {...s, v: 1, role, permissions: granted, issued_at: now, join_before: Math.min(now + 1000, expires_at), expires_at, jti: randomUUID(), ...(role === 'web' ? {grant_id, user_id: 4} : {})}); } function source(bot_id) { if (!sources.has(bot_id)) { let current; const listeners = new Set(); sources.set(bot_id, {get: () => current, observe(_id, listener) {listeners.add(listener); listener(current); return () => listeners.delete(listener);}, set(value) {current = value; for (const listener of listeners) listener(value);}, get count() {return listeners.size;}}); } return sources.get(bot_id); } function manager(s, file_factory) { const result = new NodeForwardManager(s.node_id, s.node_session_id, { socket_factory: (...args) => {const socket = io(...args); node_sockets.push(socket); return socket;}, stream_factory: bot_id => {const stream = new BotStream(bot_id, {source: source(bot_id), capacity: 30}); streams.set(bot_id, stream); return stream;}, file_factory, }); managers.push(result); return result; } async function open(s, expires_at = Date.now() + 20_000) { assert.equal((await control_ack('forward_open_session', {...s, lease_expires_at: expires_at})).status, 'success'); return expires_at; } const cmd = (s, expires_at = Date.now() + 10_000) => command(s, {url, path: handle.path, namespace: handle.namespace, expires_at, token: token(s, 'node', expires_at), deadline_at: Date.now() + 2000}); async function web(s, granted = permissions, grant_id = randomUUID(), expires_at = Date.now() + 10_000) { assert.equal((await control_ack('forward_authorize_grant', {forward_session_id: s.forward_session_id, grant_id, user_id: 4, permissions: granted, expires_at})).status, 'success'); const socket = io(`${url}${handle.namespace}`, {path: handle.path, autoConnect: false, reconnection: false, forceNew: true, transports: ['websocket']}); clients.push(socket); const connected = received(socket, 'connect'); socket.connect(); await connected; const joined = await ack(socket, 'join_room', {token: token(s, 'web', expires_at, grant_id, granted)}); assert.equal(joined.status, 'success'); return {socket, grant_id, joined}; } return {handle, control_ack, statuses, permissions, token, source, streams, manager, open, cmd, web, packets, node_sockets}; } function online_context() { const sent = []; return Object.assign(new EventEmitter(), {bot_status: 'online', bot_instance: {chat: message => sent.push(message)}, sent, config: {password: 'SECRET'}, logger: {get_recent_log() {throw new Error('Disk access forbidden');}}}); } test('file tree and binary downloads use scoped IDs without requiring an online Bot', async t => { const directory = await mkdtemp(path.join(tmpdir(), 'forward-files-')); t.after(() => rm(directory, {recursive: true, force: true})); await mkdir(path.join(directory, 'log')); await mkdir(path.join(directory, 'empty')); const bytes = Buffer.alloc(600_123); for (let index = 0; index < bytes.length; index++) bytes[index] = index % 256; await writeFile(path.join(directory, 'binary.dat'), bytes); await writeFile(path.join(directory, 'log', 'sample.log'), 'a log\n'); const f = await real_fixture(t, {rpc_timeout_ms: 1500}); const s = scope(); await f.open(s); const manager = f.manager(s, bot_id => new BotFileService(bot_id, {root: directory})); assert.equal((await manager.ensure(f.cmd(s))).status, 'success'); const web = await f.web(s, ['bot.files']); const denied = await f.web(s, ['bot.snapshot']); const listing = success_data(await ack(web.socket, 'relay_request', request('get_data_file_list'))); assert.equal(listing.partial, false); assert.deepEqual(listing.children.map(entry => entry.name), ['binary.dat', 'empty', 'log']); assert.equal(listing.children[1].type, 'directory'); assert.deepEqual(listing.children[1].children, []); assert.equal(JSON.stringify(listing).includes(directory), false); const file = listing.children[0]; assert.equal(file.size, bytes.length); assert.match(file.id, /^[0-9a-f-]{36}$/); const chunks = []; let offset = 0; while (true) { const part = success_data(await ack(web.socket, 'relay_request', request('get_file_data', {id: file.id, offset}))); assert.equal(part.offset, offset); assert.equal(part.total_bytes, bytes.length); assert.equal(part.encoding, 'base64'); const chunk = Buffer.from(part.data, 'base64'); assert.equal(chunk.length, part.next_offset - offset); assert.ok(chunk.length <= 262144); chunks.push(chunk); offset = part.next_offset; if (part.eof) break; } assert.deepEqual(Buffer.concat(chunks), bytes); code(await ack(denied.socket, 'relay_request', request('get_data_file_list')), 'FORBIDDEN'); code(await ack(denied.socket, 'relay_request', request('get_file_data', {id: file.id})), 'FORBIDDEN'); code(await ack(web.socket, 'relay_request', request('get_file_data', {id: file.id, length: 262145})), 'INVALID_REQUEST'); code(await ack(web.socket, 'relay_request', request('get_file_data', {id: '../binary.dat'})), 'INVALID_REQUEST'); const another_scope = {...s, bot_id: s.bot_id + 1, forward_session_id: randomUUID()}; await f.open(another_scope); assert.equal((await manager.ensure(f.cmd(another_scope))).status, 'success'); const another = await f.web(another_scope, ['bot.files']); code(await ack(another.socket, 'relay_request', request('get_file_data', {id: file.id})), 'INVALID_REQUEST'); // Reusing a request ID must re-read disk instead of replaying a cached file body. const repeated = request('get_file_data', {id: file.id, length: 10}); assert.equal(success_data(await ack(web.socket, 'relay_request', repeated)).data, bytes.subarray(0, 10).toString('base64')); await unlink(path.join(directory, 'binary.dat')); const removed = await ack(web.socket, 'relay_request', repeated); assert.equal(removed.status, 'error'); assert.ok(['RESYNC_REQUIRED', 'INVALID_REQUEST'].includes(removed.data.code)); const expired = received(web.socket, 'relay_status', value => !value.node_ready); manager.stop(); await expired; code(await ack(web.socket, 'relay_request', request('get_data_file_list')), 'NODE_OFFLINE'); }); test('context info RPC and updates travel through actual relay with online and permission checks', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const bot = online_context(); let records = [{name: 'health', i18n: {lang: 'en', name: 'Health', description: ''}, value: 20}]; bot.info_manager = {get_all_info: () => records}; f.source(s.bot_id).set(bot); const manager = f.manager(s); assert.equal((await manager.ensure(f.cmd(s))).status, 'success'); const web = await f.web(s, ['bot.snapshot', 'bot.logs']); const denied = await f.web(s, ['bot.logs']); const snapshot_only = await f.web(s, ['bot.snapshot']); const no_info = []; denied.socket.on('info_update', value => no_info.push(value)); const rpc = request('get_context_info'); assert.deepEqual(success_data(await ack(web.socket, 'relay_request', rpc)), records); assert.deepEqual(success_data(await ack(snapshot_only.socket, 'relay_request', request('get_context_info'))), records); code(await ack(denied.socket, 'relay_request', request('get_context_info')), 'FORBIDDEN'); code(await ack(web.socket, 'relay_request', request('get_context_info', {name: 'health'})), 'INVALID_REQUEST'); const before = received(web.socket, 'bot_log'); bot.emit('log', [{message: 'before info', font_color: '#ffffff', font_style: ['normal']}]); const first_log = await before; const pushed = received(web.socket, 'info_update'); const snapshot_pushed = received(snapshot_only.socket, 'info_update'); records[0].value = 18; bot.emit('info_update', {name: 'health', value: 18}); const update = await pushed; assert.deepEqual((await snapshot_pushed).data, {name: 'health', value: 18}); assert.deepEqual(Object.keys(update).sort(), ['bot_id', 'context_generation', 'data', 'emitted_at', 'forward_session_id', 'seq', 'stream_id']); assert.deepEqual(update.data, {name: 'health', value: 18}); assert.equal(update.bot_id, s.bot_id); assert.equal(update.forward_session_id, s.forward_session_id); assert.equal(update.stream_id, first_log.stream_id); assert.ok(update.seq > first_log.seq); const after = received(web.socket, 'bot_log'); bot.emit('log', [{message: 'after info', font_color: '#ffffff', font_style: ['normal']}]); assert.ok((await after).seq > update.seq); // A completed request ID must re-read the Node rather than replay a stale success. bot.bot_status = 'offline'; code(await ack(web.socket, 'relay_request', rpc), 'BOT_NOT_ONLINE'); const updates_before = f.packets.filter(packet => packet.event === 'info_update').length; bot.emit('info_update', {name: 'health', value: 0}); await ack(web.socket, 'relay_request', request()); assert.equal(f.packets.filter(packet => packet.event === 'info_update').length, updates_before); assert.equal(no_info.length, 0); const stale = bot.listeners('info_update')[0]; f.source(s.bot_id).set(undefined); code(await ack(web.socket, 'relay_request', rpc), 'NO_CONTEXT'); stale({name: 'health', value: 1}); const replacement = online_context(); replacement.info_manager = {get_all_info: () => [{...records[0], value: 15}]}; f.source(s.bot_id).set(replacement); assert.equal(success_data(await ack(web.socket, 'relay_request', rpc))[0].value, 15); assert.equal(bot.listenerCount('info_update'), 0); records = [{...records[0], value: NaN}]; replacement.info_manager = {get_all_info: () => records}; code(await ack(web.socket, 'relay_request', request('get_context_info')), 'INTERNAL_ERROR'); manager.stop(); assert.equal(replacement.listenerCount('info_update'), 0); }); test('complex inventory from the real module survives Node query and push through the relay', async t => { const {init: InventoryInfo} = await tsImport('../bot/info/bot/inventory.ts', import.meta.url); const mf = createRequire(createRequire(import.meta.url).resolve('@wp2508/mineflayer')); const registry = mf('prismarine-registry')('1.21.4'); const windows = mf('prismarine-windows')(registry); const instance = new EventEmitter(); instance._client = new EventEmitter(); instance.inventory = windows.createWindow(0, 'minecraft:inventory', 'Inventory'); instance.currentWindow = windows.createWindow(1, 'minecraft:generic_9x6', 'Chest'); instance.quickBarSlot = 0; let nbt = {type: 'string', value: 'deep content'}; for (let i = 0; i < 15; i++) nbt = {type: 'compound', value: {nested: nbt}}; instance.currentWindow.slots = Array.from({length: 90}, () => ({ type: 1, name: 'stone', displayName: 'Stone', count: 64, metadata: 0, nbt, components: [ {type: 'custom_data', data: {description: 'x'.repeat(4096)}}, {type: 'custom_name', data: {type: 'compound', value: {text: {type: 'string', value: '菜单'}, color: {type: 'string', value: 'gold'}}}}, {type: 'lore', data: [{type: 'string', value: '说明'}]}, ], enchants: [], })); const inventory = new InventoryInfo(instance); t.after(() => inventory.dispose()); const f = await real_fixture(t); const s = scope(); await f.open(s); const bot = online_context(); bot.info_manager = {get_all_info: () => [{name: inventory.name, i18n: inventory.i18n, value: inventory.get_info()}]}; const unsubscribe = inventory.on_update(value => bot.emit('info_update', {name: 'inventory', value})); t.after(unsubscribe); f.source(s.bot_id).set(bot); const manager = f.manager(s); assert.equal((await manager.ensure(f.cmd(s))).status, 'success'); const web = await f.web(s, ['bot.snapshot']); const records = success_data(await ack(web.socket, 'relay_request', request('get_context_info'))); assert.ok(Buffer.byteLength(JSON.stringify(records)) > 196_608); assert.deepEqual(records[0], bot.info_manager.get_all_info()[0]); const item = records[0].value.window.slots[0]; assert.equal(item.displayName[0].message, '菜单'); assert.equal(item.displayName[0].font_color, '#FFAA00'); assert.equal(item.lore[0][0].message, '说明'); assert.equal('components' in item, false); assert.equal('nbt' in item, false); const pushed = received(web.socket, 'info_update'); instance.currentWindow.slots[0].count = 32; instance.currentWindow.emit('updateSlot'); const update = await pushed; assert.equal(update.data.name, 'inventory'); assert.equal(update.data.value.window.slots[0].count, 32); assert.equal(update.data.value.window.slots.length, 90); assert.deepEqual(update.data.value, inventory.get_info()); }); test('execute_function traverses the relay with boolean results, permission checks and no replay', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const manager = f.manager(s); await manager.ensure(f.cmd(s)); const web = await f.web(s); const reader = await f.web(s, ['bot.snapshot']); const bot = online_context(); const calls = []; bot.function_manager = {async run(path, name, args, before_run) { before_run(); calls.push({path, name, args}); if (name === 'missing') throw new Error('Unknown function'); return name === 'left_click_slot'; }}; f.source(s.bot_id).set(bot); const payload = {function: 'inventory', name: 'left_click_slot', args: [9]}; const rpc = request('execute_function', payload); assert.equal(success_data(await ack(web.socket, 'relay_request', rpc)), true); code(await ack(web.socket, 'relay_request', rpc), 'CANCELLED'); assert.equal(calls.length, 1); assert.deepEqual(calls[0], {path: 'inventory', name: 'left_click_slot', args: [9]}); code(await ack(reader.socket, 'relay_request', request('execute_function', payload)), 'FORBIDDEN'); assert.equal(success_data(await ack(web.socket, 'relay_request', request('execute_function', {...payload, name: 'missing'}))), false); assert.equal(success_data(await ack(web.socket, 'relay_request', request('execute_function', {...payload, name: 'right_click_slot'}))), false); bot.bot_status = 'offline'; assert.equal(success_data(await ack(web.socket, 'relay_request', request('execute_function', payload))), false); }); test('actual manager pushes lifecycle changes to snapshot readers without exposing other bot events', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const manager = f.manager(s); const cmd = f.cmd(s); assert.deepEqual(await manager.ensure(cmd), success({...s, node_ready: true, lease_expires_at: cmd.expires_at})); const node_id = f.node_sockets[0].id; assert.deepEqual(f.packets.filter(p => p.id === node_id && ['join_room', 'forward_node_ready'].includes(p.event)).map(p => p.event), ['join_room', 'forward_node_ready']); assert.ok(f.statuses.some(value => value.state === 'ready' && value.forward_session_id === s.forward_session_id)); const web = await f.web(s); const other = await f.web(s, ['bot.snapshot']); const absent = success_data(await ack(web.socket, 'relay_request', request())); assert.deepEqual(Object.keys(absent).sort(), ['bot_id', 'bot_status', 'captured_at', 'context_exists', 'context_generation', 'seq', 'stream_id']); assert.equal(absent.context_exists, false); assert.equal(absent.context_generation, null); assert.equal(absent.bot_status, 'offline'); code(await ack(web.socket, 'relay_request', request('bot.chat', {message: 'hello'})), 'NO_CONTEXT'); let leaks = 0; other.socket.on('relay_request', () => leaks++); const reader_events = []; other.socket.on('bot_event', event => reader_events.push(event)); assert.equal((await ack(web.socket, 'relay_request', request('bot.subscribe', {events: ['status_update', 'log', 'message', 'bot_error']}))).status, 'success'); const events = []; web.socket.on('bot_event', value => events.push(value)); const bot = online_context(); const initial_status = received(other.socket, 'bot_event', event => event.payload.status === 'online'); f.source(s.bot_id).set(bot); await initial_status; const snapshot = success_data(await ack(web.socket, 'relay_request', request())); assert.equal(snapshot.context_exists, true); assert.equal(snapshot.bot_status, 'online'); assert.equal(snapshot.stream_id, absent.stream_id); assert.ok(snapshot.context_generation); for (const state of ['offline', 'ready', 'connecting', 'online', 'disconnecting', 'reconnecting']) { const pushed = received(web.socket, 'bot_event', event => event.payload.status === state); const reader_pushed = received(other.socket, 'bot_event', event => event.payload.status === state); bot.bot_status = state; bot.emit('status_update', state); await Promise.all([pushed, reader_pushed]); const current = success_data(await ack(web.socket, 'relay_request', request())); assert.equal(current.context_exists, true); assert.equal(current.bot_status, state); assert.equal(current.context_generation, snapshot.context_generation); } const back_online = received(web.socket, 'bot_event', event => event.payload.status === 'online'); bot.bot_status = 'online'; bot.emit('status_update', 'online'); await back_online; const log_packet = received(web.socket, 'bot_log'); const metadata = {hoverEvent: {action: 'show_text', value: '悬停说明'}, clickEvent: {action: 'suggest_command', value: '/help'}, insertion: 'Alex'}; bot.emit('log', [{message: 'safe log', font_color: '#abcdef', font_style: ['normal'], ...metadata, password: 'SECRET'}]); const live_log = await log_packet; assert.equal(live_log.forward_session_id, s.forward_session_id); assert.equal(live_log.payload.records[0].segments[0].message, 'safe log'); for (const [key, value] of Object.entries(metadata)) assert.deepEqual(live_log.payload.records[0].segments[0][key], value); assert.equal(JSON.stringify(live_log).includes('SECRET'), false); const logs = success_data(await ack(web.socket, 'relay_request', request('bot.logs', {limit: 3}))); assert.equal(logs.records[0].segments[0].message, 'safe log'); assert.deepEqual(logs.records[0].segments[0], live_log.payload.records[0].segments[0]); for (let i = 0; i < 25; i++) bot.emit('log', [{message: `history ${i}`, font_color: '#ffffff', font_style: ['normal']}]); bot.logger.read_recent_log = async count => Array.from({length: 25}, (_, i) => ({ id: randomUUID(), timestamp: Date.now(), log: [{message: `history ${i}`, font_color: '#ffffff', font_style: ['normal']}], })).slice(-count); const history = success_data(await ack(web.socket, 'relay_request', request('get_log_history', {}))); assert.equal(history.records.length, 20); assert.deepEqual(history.records.map(record => record.segments[0].message), Array.from({length: 20}, (_, i) => `history ${i + 5}`)); code(await ack(other.socket, 'relay_request', request('get_log_history', {})), 'FORBIDDEN'); code(await ack(web.socket, 'relay_request', request('get_log_history', {limit: 21})), 'INVALID_REQUEST'); assert.equal((await ack(web.socket, 'relay_request', request('bot.chat', {message: 'hello'}))).data.accepted, true); assert.deepEqual(bot.sent, ['hello']); bot.bot_instance.tabComplete = async text => { assert.equal(text, '/he'); return ['help', {match: 'help'}, {match: 'hello'}]; }; assert.deepEqual(success_data(await ack(web.socket, 'relay_request', request('tab_complete', {text: '/he'}))), {matches: ['help', 'hello'], start: 1, length: 2}); bot.bot_instance.tabComplete = async () => ['Alex', 'Alice', 'Alex']; const input = '/tell Al'; const completion = success_data(await ack(web.socket, 'relay_request', request('tab_complete', {text: input}))); assert.deepEqual(completion, {matches: ['Alex', 'Alice'], start: 6, length: 2}); assert.equal(input.slice(0, completion.start) + completion.matches[0] + input.slice(completion.start + completion.length), '/tell Alex'); code(await ack(other.socket, 'relay_request', request('tab_complete', {text: '/he'})), 'FORBIDDEN'); bot.bot_status = 'offline'; code(await ack(web.socket, 'relay_request', request('bot.chat', {message: 'blocked'})), 'BOT_NOT_ONLINE'); assert.deepEqual(bot.sent, ['hello']); bot.emit('message', {json: {text: 'public text', password: 'SECRET'}}, 'chat'); bot.emit('error', new Error('SECRET stack')); assert.equal(JSON.stringify({snapshot, logs}).includes('SECRET'), false); await ack(other.socket, 'relay_request', request()); assert.equal(leaks, 0); assert.ok(events.length >= 7 && reader_events.length >= 7); assert.ok([...events, ...reader_events].every(event => event.name === 'status_update')); assert.ok(f.packets.some(packet => packet.event === 'bot_event')); manager.stop(); assert.equal(f.source(s.bot_id).count, 0); assert.equal(bot.listenerCount('log'), 0); }); test('actual manager reauth renews same transport; disconnect requires fresh token and preserves stream identity', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const manager = f.manager(s); const first = f.cmd(s, Date.now() + 3000); assert.equal((await manager.ensure(first)).status, 'success'); const web = await f.web(s); const before = success_data(await ack(web.socket, 'relay_request', request())); const renewed = f.cmd(s, Date.now() + 8000); assert.equal((await manager.ensure(renewed)).data.lease_expires_at, renewed.expires_at); assert.equal(f.node_sockets.length, 1); assert.ok(f.packets.some(p => p.event === 'reauth' && p.value.token === renewed.token)); const offline = received(web.socket, 'relay_status', value => !value.node_ready); f.node_sockets[0].disconnect(); await offline; code(await ack(web.socket, 'relay_request', request()), 'NODE_OFFLINE'); code(await manager.ensure({...renewed, deadline_at: Date.now() + 2000}), 'TOKEN_EXPIRED'); assert.equal(f.node_sockets.length, 1); const fresh = f.cmd(s, Date.now() + 9000); assert.equal((await manager.ensure(fresh)).status, 'success'); assert.equal(f.node_sockets.length, 2); const after = success_data(await ack(web.socket, 'relay_request', request())); assert.equal(after.stream_id, before.stream_id); assert.ok(after.seq > before.seq); }); test('actual multi-grant request_id reuse is private, read-deduped and chat executes once per grant', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const bot = online_context(); f.source(s.bot_id).set(bot); const manager = f.manager(s); assert.equal((await manager.ensure(f.cmd(s))).status, 'success'); const a = await f.web(s); const b = await f.web(s); const id = randomUUID(); const reads = await Promise.all([a, b].map(web => ack(web.socket, 'relay_request', request('bot.snapshot', {}, id)))); assert.ok(reads.every(value => value.status === 'success')); assert.deepEqual(await ack(a.socket, 'relay_request', request('bot.snapshot', {}, id)), reads[0]); code(await ack(a.socket, 'relay_request', request('bot.logs', {limit: 1}, id)), 'INVALID_REQUEST'); const chat_id = randomUUID(); const chats = await Promise.all([a, b].map(web => ack(web.socket, 'relay_request', request('bot.chat', {message: web.grant_id}, chat_id)))); assert.ok(chats.every(value => value.status === 'success')); assert.deepEqual(new Set(bot.sent), new Set([a.grant_id, b.grant_id])); code(await ack(a.socket, 'relay_request', request('bot.chat', {message: a.grant_id}, chat_id)), 'CANCELLED'); assert.equal(bot.sent.length, 2); const closed = received(a.socket, 'forward_session_closed'); assert.equal((await f.control_ack('forward_revoke_grant', {forward_session_id: s.forward_session_id, grant_id: a.grant_id, reason: 'permission_removed'})).status, 'success'); await closed; assert.equal((await ack(b.socket, 'relay_request', request())).status, 'success'); code(await f.control_ack('forward_authorize_grant', {forward_session_id: s.forward_session_id, grant_id: a.grant_id, user_id: 4, permissions: f.permissions, expires_at: Date.now() + 5000}), 'FORBIDDEN'); }); test('one Node manager isolates multiple Bot sessions and observes stop/start without closing Web transport', async t => { const f = await real_fixture(t); const a = scope(); const b = {...a, bot_id: a.bot_id + 1, forward_session_id: randomUUID()}; await f.open(a); await f.open(b); const manager = f.manager(a); const bot_a = online_context(); const bot_b = online_context(); f.source(a.bot_id).set(bot_a); f.source(b.bot_id).set(bot_b); assert.equal((await manager.ensure(f.cmd(a))).status, 'success'); assert.equal((await manager.ensure(f.cmd(b))).status, 'success'); const web_a = await f.web(a); const web_b = await f.web(b); const before = success_data(await ack(web_a.socket, 'relay_request', request())); const other = success_data(await ack(web_b.socket, 'relay_request', request())); assert.notEqual(before.stream_id, other.stream_id); assert.equal(before.bot_id, a.bot_id); assert.equal(other.bot_id, b.bot_id); await ack(web_a.socket, 'relay_request', request('bot.subscribe', {events: ['status_update', 'log']})); await ack(web_b.socket, 'relay_request', request('bot.subscribe', {events: ['status_update', 'log']})); const events_b = []; web_b.socket.on('bot_event', event => events_b.push(event)); const events_a = []; web_a.socket.on('bot_event', event => events_a.push(event)); bot_a.emit('log', [{message: 'Bot A only', font_color: '#ffffff', font_style: ['normal']}]); assert.equal(success_data(await ack(web_a.socket, 'relay_request', request('bot.logs', {limit: 10}))).records[0].segments[0].message, 'Bot A only'); assert.equal(success_data(await ack(web_b.socket, 'relay_request', request('bot.logs', {limit: 10}))).records.length, 0); assert.equal(events_b.length, 0); await ack(web_a.socket, 'relay_request', request('bot.chat', {message: 'a'})); await ack(web_b.socket, 'relay_request', request('bot.chat', {message: 'b'})); assert.deepEqual(bot_a.sent, ['a']); assert.deepEqual(bot_b.sent, ['b']); const stopped = received(web_a.socket, 'bot_event', event => event.payload.status === 'offline'); f.source(a.bot_id).set(undefined); await stopped; const absent = success_data(await ack(web_a.socket, 'relay_request', request())); assert.equal(absent.bot_status, 'offline'); assert.equal(absent.context_generation, null); const replacement = online_context(); const started = received(web_a.socket, 'bot_event', event => event.payload.status === 'online'); f.source(a.bot_id).set(replacement); await started; const after = success_data(await ack(web_a.socket, 'relay_request', request())); assert.equal(after.stream_id, before.stream_id); assert.notEqual(after.context_generation, before.context_generation); assert.equal(web_a.socket.connected, true); assert.equal(f.node_sockets.length, 2); assert.equal(events_b.length, 0); assert.deepEqual(events_a.map(event => event.payload.status), ['offline', 'online']); assert.equal(bot_a.listenerCount('log'), 0); }); test('actual session revocation retires manager, closes Web and cannot revive stale scope after a new session opens', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const manager = f.manager(s); assert.equal((await manager.ensure(f.cmd(s))).status, 'success'); const web = await f.web(s); const closed = received(web.socket, 'forward_session_closed'); const node_closed = received(f.node_sockets[0], 'forward_session_closed'); assert.equal((await f.control_ack('forward_revoke_session', {forward_session_id: s.forward_session_id, reason: 'ownership_changed'})).status, 'success'); assert.equal((await closed).code, 'SESSION_STALE'); await node_closed; code(await manager.ensure(f.cmd(s)), 'SESSION_STALE'); code(await f.control_ack('forward_open_session', {...s, lease_expires_at: Date.now() + 5000}), 'SESSION_STALE'); const next = {...s, forward_session_id: randomUUID(), node_session_id: 'new-main-socket-session'}; await f.open(next); code(await manager.ensure(f.cmd(next)), 'SESSION_STALE'); const replacement = f.manager(next); assert.equal((await replacement.ensure(f.cmd(next))).status, 'success'); const new_web = await f.web(next); assert.equal((await ack(new_web.socket, 'relay_request', request())).status, 'success'); code(await manager.ensure(f.cmd(s)), 'SESSION_STALE'); }); test('actual relay rejects wrong-scope token and manager lease expiry prevents session revival', async t => { const f = await real_fixture(t); const s = scope(); await f.open(s); const manager = f.manager(s); const wrong = f.cmd(s); wrong.token = f.token({...s, bot_id: s.bot_id + 1}, 'node', wrong.expires_at); code(await manager.ensure(wrong), 'SESSION_STALE'); const short = f.cmd(s, Date.now() + 200); assert.equal((await manager.ensure(short)).status, 'success'); const web = await f.web(s); await delay(Math.max(0, short.expires_at - Date.now()) + 25); code(await manager.ensure(f.cmd(s)), 'SESSION_STALE'); code(await ack(web.socket, 'relay_request', request()), 'NODE_OFFLINE'); assert.equal(f.source(s.bot_id).count, 0); }); test('actual relay deadline times out held chat; late manager delivery cannot send chat or replace the ACK', async t => { const f = await real_fixture(t, {rpc_timeout_ms: 60}); const s = scope(); await f.open(s); const bot = online_context(); f.source(s.bot_id).set(bot); const manager = f.manager(s); assert.equal((await manager.ensure(f.cmd(s))).status, 'success'); const web = await f.web(s); const node = f.node_sockets[0]; const [handler] = node.listeners('relay_request'); let late; let arrived; const held = new Promise(resolve => {arrived = resolve;}); node.off('relay_request', handler); node.on('relay_request', (raw, callback) => { late = () => new Promise(resolve => handler(raw, response => {callback(response); resolve(response);})); arrived(raw); }); const chat = request('bot.chat', {message: 'must never send'}); const response = ack(web.socket, 'relay_request', chat); const delivered = await held; assert.equal(delivered.forward_session_id, s.forward_session_id); assert.notEqual(delivered.correlation_id, chat.request_id); code(await response, 'TIMEOUT'); code(await late(), 'TIMEOUT'); assert.deepEqual(bot.sent, []); code(await ack(web.socket, 'relay_request', chat), 'CANCELLED'); assert.deepEqual(bot.sent, []); node.removeAllListeners('relay_request'); node.on('relay_request', handler); assert.equal((await ack(web.socket, 'relay_request', request())).status, 'success'); }); test('public relay restart closes old scope; a fresh session and replacement manager restore Web RPC', async t => { const first = await real_fixture(t); const s = scope(); await first.open(s); const manager = first.manager(s); assert.equal((await manager.ensure(first.cmd(s))).status, 'success'); const web = await first.web(s); const node_closed = received(first.node_sockets[0], 'forward_session_closed'); const web_closed = received(web.socket, 'forward_session_closed'); await first.handle.stop(); assert.equal((await node_closed).reason, 'relay_closed'); assert.equal((await web_closed).code, 'RELAY_UNAVAILABLE'); const restarted = await real_fixture(t); code(await manager.ensure(restarted.cmd(s)), 'SESSION_STALE'); const next = {...s, forward_session_id: randomUUID()}; await restarted.open(next); assert.equal((await manager.ensure(restarted.cmd(next))).status, 'success'); const current = await restarted.web(next); const before = success_data(await ack(current.socket, 'relay_request', request())); const offline = received(current.socket, 'relay_status', value => !value.node_ready); manager.stop(); await offline; const replacement = restarted.manager(next); assert.equal((await replacement.ensure(restarted.cmd(next))).status, 'success'); const after = success_data(await ack(current.socket, 'relay_request', request())); assert.notEqual(after.stream_id, before.stream_id); assert.equal(after.context_exists, false); });