217 lines
13 KiB
TypeScript
217 lines
13 KiB
TypeScript
import {createHash} from 'node:crypto';
|
|
import {types} from 'node:util';
|
|
import type {Socket} from 'socket.io-client';
|
|
import type {bot_instance_context} from '../bot/instance/bot_instance_init.ts';
|
|
import {get_bot_context} from '../bot/instance/index.ts';
|
|
import type {bot_config_plugin} from '../types/bot/plugins/plugin.ts';
|
|
import type {workflow_config} from '../types/bot/workflow/workflow.ts';
|
|
import {compile_workflow} from '../bot/workflow/runtime/compile.ts';
|
|
import {plain_plugin_object, valid_plugin_config, valid_plugin_name, valid_plugin_path} from './plugin_validation.ts';
|
|
import {console_print} from './console_print.ts';
|
|
|
|
type Context = bot_instance_context<unknown> & {readonly is_disposed?: boolean};
|
|
interface Request {
|
|
command_id: string;
|
|
payload_version: 1;
|
|
node_session_id: string;
|
|
bot_id: number;
|
|
config_revision: number;
|
|
plugins: bot_config_plugin<unknown>[];
|
|
workflows: workflow_config[];
|
|
}
|
|
type Reply = {status: 'success'; data: Pick<Request, 'command_id' | 'bot_id' | 'config_revision'> & {
|
|
state: 'completed' | 'processing' | 'no_context';
|
|
}} | {status: 'error'; data: {code: string; message: string}};
|
|
type State = {revision: number; fingerprint?: string; applied_revision: number; applied_fingerprint?: string; tail: Promise<void>};
|
|
type Job = {fingerprint: string; context: Context | undefined; reply: Reply; pending: boolean; failure_reported?: boolean};
|
|
const max_bytes = 256 * 1024;
|
|
const fields = ['command_id', 'payload_version', 'node_session_id', 'bot_id', 'config_revision', 'plugins', 'workflows'];
|
|
const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
|
|
const positive = (value: unknown): value is number => typeof value === 'number' && Number.isSafeInteger(value) && value > 0;
|
|
const failure = (code: string, message: string): Reply => ({status: 'error', data: {code, message}});
|
|
class ApplyGuardError extends Error {
|
|
constructor(readonly code: string, message: string) { super(message); }
|
|
}
|
|
|
|
// Reject accessors, proxies, sparse arrays and unsafe keys before hashing or cloning input.
|
|
function bounded_json(value: unknown): void {
|
|
let bytes = 0, entries = 0;
|
|
const ancestors = new Set<object>();
|
|
const visit = (item: unknown, depth: number): void => {
|
|
if (depth > 64 || ++entries > 200000) throw new Error('Invalid JSON');
|
|
if (item === null || typeof item === 'boolean' || typeof item === 'number' && Number.isFinite(item)) {
|
|
bytes += JSON.stringify(item).length;
|
|
} else if (typeof item === 'string') {
|
|
if (item.length > max_bytes) throw new Error('Invalid JSON');
|
|
bytes += Buffer.byteLength(JSON.stringify(item));
|
|
} else if (item && typeof item === 'object') {
|
|
if (types.isProxy(item) || ancestors.has(item)) throw new Error('Invalid JSON');
|
|
const array = Array.isArray(item);
|
|
if (!array && !plain_plugin_object(item)) throw new Error('Invalid JSON');
|
|
const keys = Reflect.ownKeys(item);
|
|
if (keys.length > 200000 || array && keys.length !== item.length + 1) throw new Error('Invalid JSON');
|
|
ancestors.add(item);
|
|
bytes += 2;
|
|
let count = 0;
|
|
for (const key of keys) {
|
|
if (array && key === 'length') continue;
|
|
if (typeof key !== 'string' || ['__proto__', 'prototype', 'constructor'].includes(key)) throw new Error('Invalid JSON');
|
|
const descriptor = Object.getOwnPropertyDescriptor(item, key)!;
|
|
if (!descriptor.enumerable || !Object.hasOwn(descriptor, 'value')) throw new Error('Invalid JSON');
|
|
if (array && (!/^(0|[1-9]\d*)$/.test(key) || Number(key) >= item.length)) throw new Error('Invalid JSON');
|
|
if (!array) bytes += Buffer.byteLength(JSON.stringify(key)) + 1;
|
|
if (count++) bytes++;
|
|
visit(descriptor.value, depth + 1);
|
|
}
|
|
ancestors.delete(item);
|
|
} else throw new Error('Invalid JSON');
|
|
if (bytes > max_bytes) throw new Error('Invalid JSON');
|
|
};
|
|
visit(value, 0);
|
|
}
|
|
|
|
function fingerprint(value: unknown): string {
|
|
const canonical = (item: any): any => Array.isArray(item) ? item.map(canonical)
|
|
: item && typeof item === 'object' ? Object.fromEntries(Object.keys(item).sort().map(key => [key, canonical(item[key])])) : item;
|
|
return createHash('sha256').update(JSON.stringify(canonical(value))).digest('hex');
|
|
}
|
|
|
|
function parse(value: unknown): Request {
|
|
bounded_json(value);
|
|
if (!plain_plugin_object(value) || Object.keys(value).length !== fields.length
|
|
|| !fields.every(field => Object.hasOwn(value, field)) || typeof value.command_id !== 'string' || !uuid.test(value.command_id)
|
|
|| value.payload_version !== 1 || typeof value.node_session_id !== 'string' || !value.node_session_id.trim()
|
|
|| Buffer.byteLength(value.node_session_id) > 256 || !positive(value.bot_id) || !positive(value.config_revision)
|
|
|| !Array.isArray(value.plugins) || !Array.isArray(value.workflows)) throw new Error('Invalid payload');
|
|
const names = new Set<string>();
|
|
for (const item of value.plugins) {
|
|
if (!plain_plugin_object(item) || Object.keys(item).length !== 3 || !valid_plugin_name(item.name)
|
|
|| !valid_plugin_path(item.path) || !valid_plugin_config(item.config) || names.has(item.name)) throw new Error('Invalid plugins');
|
|
names.add(item.name);
|
|
}
|
|
// Compile the entire graph set before any plugin mutation, including duplicate IDs.
|
|
const ids = new Set<string>();
|
|
for (const item of value.workflows) {
|
|
const {definition} = compile_workflow(item);
|
|
if (ids.has(definition.id)) throw new Error('Duplicate workflow');
|
|
ids.add(definition.id);
|
|
}
|
|
return structuredClone(value) as unknown as Request;
|
|
}
|
|
|
|
export function create_bot_config_apply_handler(lookup: (bot_id: number) => Context | undefined = get_bot_context) {
|
|
const jobs = new Map<string, Job>();
|
|
const states = new WeakMap<Context, State>();
|
|
let pending = 0;
|
|
const trim = () => {
|
|
let completed = jobs.size - pending;
|
|
for (const [id, job] of jobs) {
|
|
if (completed <= 100) break;
|
|
if (!job.pending) { jobs.delete(id); completed--; }
|
|
}
|
|
};
|
|
const success = (request: Request, state: 'completed' | 'processing' | 'no_context'): Reply => ({status: 'success', data: {
|
|
command_id: request.command_id, bot_id: request.bot_id, config_revision: request.config_revision, state,
|
|
}});
|
|
const current = (request: Request, context: Context) => lookup(request.bot_id) === context && !context.is_disposed;
|
|
const session = (socket: Socket, request: Request) => socket.connected && socket.id === request.node_session_id;
|
|
|
|
return async (socket: Socket, data: unknown, callback: unknown): Promise<void> => {
|
|
if (typeof callback !== 'function') return;
|
|
const ack = (reply: Reply) => { try { callback(structuredClone(reply)); } catch {} };
|
|
let request: Request;
|
|
try { request = parse(data); }
|
|
catch { ack(failure('INVALID_PAYLOAD', '配置应用请求格式无效')); return; }
|
|
if (!session(socket, request)) { ack(failure('SESSION_MISMATCH', '节点会话已失效')); return; }
|
|
const content = fingerprint({plugins: request.plugins, workflows: request.workflows});
|
|
const identity = fingerprint({bot_id: request.bot_id, config_revision: request.config_revision, content});
|
|
const previous = jobs.get(request.command_id);
|
|
if (previous) {
|
|
if (previous.fingerprint !== identity) { ack(failure('COMMAND_CONFLICT', '命令标识已用于其他配置')); return; }
|
|
if (previous.context && !current(request, previous.context)) { ack(failure('CONTEXT_CHANGED', '机器人上下文已失效')); return; }
|
|
const retryable = !previous.pending && previous.reply.status === 'error'
|
|
&& ['APPLY_FAILED', 'PARTIAL_APPLY_FAILED', 'SESSION_MISMATCH'].includes(previous.reply.data.code);
|
|
if (!retryable || !previous.failure_reported) {
|
|
ack(previous.reply);
|
|
// Report the settled failure once, then allow the same command to retry safely.
|
|
if (retryable) previous.failure_reported = true;
|
|
return;
|
|
}
|
|
}
|
|
const context = lookup(request.bot_id);
|
|
if (!context) {
|
|
const reply = success(request, 'no_context');
|
|
jobs.set(request.command_id, {fingerprint: identity, context, reply, pending: false});
|
|
trim(); ack(reply); return;
|
|
}
|
|
if (!current(request, context)) { ack(failure('CONTEXT_CHANGED', '机器人上下文已失效')); return; }
|
|
if (!context.plugin_manager || !context.workflow_manager) {
|
|
ack(failure('MANAGER_UNAVAILABLE', '机器人配置管理器尚未就绪')); return;
|
|
}
|
|
let state = states.get(context);
|
|
if (!state) {
|
|
const config = context.bot_config;
|
|
const revision = positive(config.config_revision) ? config.config_revision : 0;
|
|
const initial = Array.isArray(config.plugins) && Array.isArray(config.workflow)
|
|
? fingerprint({plugins: config.plugins, workflows: config.workflow}) : undefined;
|
|
state = {revision, fingerprint: initial, applied_revision: revision, applied_fingerprint: initial, tail: Promise.resolve()};
|
|
states.set(context, state);
|
|
}
|
|
if (request.config_revision < state.revision) { ack(failure('STALE_REVISION', '配置版本已过期')); return; }
|
|
if (request.config_revision === state.revision && state.fingerprint && state.fingerprint !== content) {
|
|
ack(failure('REVISION_CONFLICT', '相同配置版本的内容不一致')); return;
|
|
}
|
|
if (pending >= 32) { ack(failure('APPLY_BUSY', '配置应用队列已满')); return; }
|
|
state.revision = request.config_revision;
|
|
state.fingerprint = content;
|
|
const job: Job = {fingerprint: identity, context, reply: success(request, 'processing'), pending: true};
|
|
jobs.set(request.command_id, job);
|
|
pending++;
|
|
const installed_state = state;
|
|
const manager = context.workflow_manager;
|
|
const guard = () => {
|
|
if (!current(request, context)) throw new ApplyGuardError('CONTEXT_CHANGED', '机器人上下文已失效');
|
|
if (!session(socket, request)) throw new ApplyGuardError('SESSION_MISMATCH', '节点会话已失效');
|
|
if (context.workflow_manager !== manager) throw new ApplyGuardError('CONTEXT_CHANGED', '机器人配置管理器已失效');
|
|
};
|
|
state.tail = state.tail.then(async () => {
|
|
let plugin_applied = false;
|
|
try {
|
|
guard();
|
|
if (request.config_revision < installed_state.revision) throw new ApplyGuardError('STALE_REVISION', '配置版本已过期');
|
|
if (installed_state.applied_revision !== request.config_revision || installed_state.applied_fingerprint !== content) {
|
|
await context.plugin_manager.apply_desired(request.plugins);
|
|
plugin_applied = true;
|
|
guard();
|
|
await manager.apply(request.workflows, request.config_revision);
|
|
guard();
|
|
context.bot_config.plugins = structuredClone(request.plugins);
|
|
context.bot_config.workflow = structuredClone(request.workflows);
|
|
context.bot_config.config_revision = request.config_revision;
|
|
installed_state.applied_revision = request.config_revision;
|
|
installed_state.applied_fingerprint = content;
|
|
}
|
|
job.reply = success(request, 'completed');
|
|
const message = `机器人 ${request.bot_id} 配置版本 ${request.config_revision} 应用完成`;
|
|
try { console_print.log(console_print.success(message)); } catch {}
|
|
try { if (context.logger) context.log([context.logger.info(message)]); } catch {}
|
|
} catch (error) {
|
|
// Transport replies and logs must never contain thrown plugin parameters or code.
|
|
job.reply = error instanceof ApplyGuardError ? failure(error.code, error.message)
|
|
: failure(plugin_applied ? 'PARTIAL_APPLY_FAILED' : 'APPLY_FAILED',
|
|
plugin_applied ? '插件已应用，工作流应用失败，请重试完整配置' : '配置应用失败，请重试完整配置');
|
|
const message = `机器人 ${request.bot_id} 配置版本 ${request.config_revision} 应用失败`;
|
|
try { console_print.log(console_print.error(message)); } catch {}
|
|
try { if (context.logger) context.log([context.logger.error(message)]); } catch {}
|
|
} finally {
|
|
job.pending = false;
|
|
pending--;
|
|
trim();
|
|
}
|
|
});
|
|
ack(job.reply);
|
|
};
|
|
}
|
|
|
|
export const bot_config_apply_handler = create_bot_config_apply_handler();
|