133 lines
7.1 KiB
TypeScript
133 lines
7.1 KiB
TypeScript
import {Worker} from 'node:worker_threads';
|
|
import type {BranchState, NodeHandler} from '../../../types/bot/workflow/runtime.ts';
|
|
import type {WorkflowNodeTemplate} from '../../../types/bot/workflow/metadata.ts';
|
|
import type {Json} from '../../../types/bot/workflow/workflow.ts';
|
|
import {describe, failure_port, output, success_port} from '../metadata.ts';
|
|
import {assert_json} from '../runtime/value.ts';
|
|
|
|
export const config_template: WorkflowNodeTemplate = {
|
|
name: 'code', version: '1.0.0',
|
|
i18n: describe('执行代码', '执行 JavaScript 异步函数体，通过 state 读取状态副本，并返回 JSON 值；是否可用由宿主配置决定。'),
|
|
config: [
|
|
{name: 'source', type: 'code', language: 'javascript', required: true, max_length: 100_000,
|
|
i18n: describe('JavaScript 代码', '最多 100000 个字符，可使用 await；通过 return 返回有效 JSON 值，修改 state 副本不会写回工作流状态。')},
|
|
],
|
|
output: output('json', '代码返回值', '代码通过 return 返回并经校验的 JSON 值。'),
|
|
ports: [success_port, failure_port],
|
|
availability: {available: false, reason: 'CODE_EXECUTION_DISABLED'},
|
|
};
|
|
|
|
const worker_source = `
|
|
const {parentPort, workerData} = require('node:worker_threads');
|
|
function serialize_output(output) {
|
|
const ancestors = new Set();
|
|
let entries = 0;
|
|
let bytes = 0;
|
|
function visit(value, depth) {
|
|
if (depth > 64 || ++entries > 100000) throw new Error('Code output exceeds JSON limits');
|
|
if (value === null || typeof value === 'boolean') bytes += 5;
|
|
else if (typeof value === 'number' && Number.isFinite(value)) bytes += String(value).length;
|
|
else if (typeof value === 'string') {
|
|
if (value.length > 1048576) throw new Error('Code output exceeds JSON limits');
|
|
bytes += Buffer.byteLength(JSON.stringify(value));
|
|
} else if (typeof value === 'object') {
|
|
const array = Array.isArray(value);
|
|
if (!array && Object.getPrototypeOf(value) !== Object.prototype && Object.getPrototypeOf(value) !== null)
|
|
throw new Error('Code output must contain plain JSON objects');
|
|
if (ancestors.has(value)) throw new Error('Code output contains a cycle');
|
|
ancestors.add(value);
|
|
const keys = Reflect.ownKeys(value);
|
|
if (keys.length > 100000 || (array && value.length > 100000)) throw new Error('Code output exceeds JSON limits');
|
|
let count = 0;
|
|
bytes += 2;
|
|
for (const key of keys) {
|
|
if (array && key === 'length') continue;
|
|
if (typeof key !== 'string' || ['__proto__', 'prototype', 'constructor'].includes(key))
|
|
throw new Error('Code output contains an unsafe key');
|
|
const descriptor = Object.getOwnPropertyDescriptor(value, key);
|
|
if (!descriptor.enumerable || !('value' in descriptor)) throw new Error('Code output must contain data properties');
|
|
if (array) {
|
|
const index = Number(key);
|
|
if (!Number.isSafeInteger(index) || index < 0 || String(index) !== key || index >= value.length)
|
|
throw new Error('Code output contains an invalid array');
|
|
count++;
|
|
} else {
|
|
if (key.length > 1048576) throw new Error('Code output exceeds JSON limits');
|
|
bytes += Buffer.byteLength(JSON.stringify(key)) + 1;
|
|
}
|
|
bytes++;
|
|
visit(descriptor.value, depth + 1);
|
|
}
|
|
if (array && count !== value.length) throw new Error('Code output contains an array hole');
|
|
ancestors.delete(value);
|
|
} else throw new Error('Code output is not JSON');
|
|
if (bytes > 1048576) throw new Error('Code output exceeds JSON limits');
|
|
}
|
|
try { visit(output, 0); }
|
|
catch (error) { error.code = 'INVALID_JSON'; throw error; }
|
|
return JSON.stringify(output);
|
|
}
|
|
(async () => {
|
|
try {
|
|
const AsyncFunction = Object.getPrototypeOf(async function () {}).constructor;
|
|
const execute = new AsyncFunction('state', '"use strict";\\n' + workerData.source);
|
|
const output = await execute(JSON.parse(workerData.state));
|
|
parentPort.postMessage({status: 'success', output: serialize_output(output)});
|
|
} catch (error) {
|
|
parentPort.postMessage({status: 'failure',
|
|
code: typeof error?.code === 'string' ? error.code.slice(0, 256) : 'CODE_EXECUTION_FAILED',
|
|
message: (typeof error?.message === 'string' ? error.message : String(error)).slice(0, 4096)});
|
|
}
|
|
})();
|
|
`;
|
|
|
|
export const execute: NodeHandler = async (node, state, signal, deps) => {
|
|
if (node.type !== 'code') throw new Error('Expected a code node');
|
|
signal.throwIfAborted();
|
|
assert_json(state);
|
|
const serialized = JSON.stringify(state);
|
|
let output: Json;
|
|
if (deps.execute_code) {
|
|
output = await deps.execute_code(node.config.source, JSON.parse(serialized) as BranchState, signal);
|
|
} else {
|
|
const worker = new Worker(worker_source, {eval: true, execArgv: [],
|
|
workerData: {source: node.config.source, state: serialized},
|
|
stdout: true, stderr: true,
|
|
resourceLimits: {maxOldGenerationSizeMb: 32, maxYoungGenerationSizeMb: 8, stackSizeMb: 4}});
|
|
// Capture and discard worker output instead of forwarding it to the bot process.
|
|
worker.stdout.resume();
|
|
worker.stderr.resume();
|
|
let on_abort: () => void;
|
|
try {
|
|
output = await new Promise<Json>((resolve, reject) => {
|
|
on_abort = () => reject(signal.reason);
|
|
worker.once('message', message => {
|
|
if (message?.status === 'success') {
|
|
try {
|
|
if (typeof message.output !== 'string' || Buffer.byteLength(message.output) > 1_048_576) {
|
|
throw Object.assign(new Error('Invalid code worker output'), {code: 'INVALID_JSON'});
|
|
}
|
|
resolve(JSON.parse(message.output));
|
|
} catch (error) { reject(error); }
|
|
}
|
|
else reject(Object.assign(new Error(typeof message?.message === 'string'
|
|
? message.message : 'Code execution failed'),
|
|
{code: typeof message?.code === 'string' ? message.code : 'CODE_EXECUTION_FAILED'}));
|
|
});
|
|
worker.once('error', reject);
|
|
worker.once('exit', code => reject(Object.assign(
|
|
new Error(`Code worker exited before returning a result (${code})`), {code: 'CODE_WORKER_EXITED'})));
|
|
signal.addEventListener('abort', on_abort, {once: true});
|
|
if (signal.aborted) on_abort();
|
|
});
|
|
} finally {
|
|
signal.removeEventListener('abort', on_abort!);
|
|
await worker.terminate();
|
|
worker.removeAllListeners();
|
|
}
|
|
}
|
|
signal.throwIfAborted();
|
|
assert_json(output);
|
|
return {port: 'success', result: {status: 'success', output}};
|
|
};
|