diff --git a/restricted-execution/service.mjs b/restricted-execution/service.mjs index 4e5967d..b1ea480 100644 --- a/restricted-execution/service.mjs +++ b/restricted-execution/service.mjs @@ -1,6 +1,7 @@ import { spawn } from "node:child_process"; import { createServer, request as httpRequest } from "node:http"; -import { mkdtemp, readFile, writeFile, rm } from "node:fs/promises"; +import { mkdtemp, readFile, writeFile, rm, open } from "node:fs/promises"; +import { constants } from "node:fs"; import { tmpdir } from "node:os"; import { join, dirname } from "node:path"; @@ -12,6 +13,24 @@ const absolute = (value) => value.startsWith("/") && !/[\s\u0000-\u001f]/.test(value) && !value.split("/").includes(".."); +async function boundedSource(path, maximum, code) { + const handle = await open(path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.size > maximum) throw fixedError(code); + const bytes = Buffer.alloc(stat.size + 1); + let offset = 0; + while (offset < bytes.length) { + const read = await handle.read(bytes, offset, bytes.length - offset, offset); + if (!read.bytesRead) break; + offset += read.bytesRead; + } + if (offset !== stat.size) throw fixedError(code); + return new TextDecoder("utf-8", { fatal: true }).decode(bytes.subarray(0, offset)); + } finally { + await handle.close(); + } +} /** Trusted stdio adapter inside one workspace VM. Engine processes never own * coordinator credentials. The supervising service must impose VM/OS limits. @@ -47,6 +66,8 @@ export async function serveRestrictedRegistry(config, io = process) { maxPortCalls: config.maxPortCalls ?? 1000, deadlineMs: config.deadlineMs ?? 4000, maxQueued: config.maxQueued ?? 8, + maxActive: config.maxActive ?? 2, + maxNested: config.maxNested ?? 4, }; if ( !Number.isSafeInteger(limits.maxBytes) || @@ -60,17 +81,22 @@ export async function serveRestrictedRegistry(config, io = process) { limits.deadlineMs > 60000 || !Number.isSafeInteger(limits.maxQueued) || limits.maxQueued < 0 || - limits.maxQueued > 64 + limits.maxQueued > 64 || + !Number.isSafeInteger(limits.maxActive) || + limits.maxActive < 1 || + limits.maxActive > 16 || + !Number.isSafeInteger(limits.maxNested) || + limits.maxNested < 1 || + limits.maxNested > 16 ) throw fixedError("INVALID_RESTRICTED_LIMITS"); - const packageSource = await readFile(config.module, "utf8"); - if (Buffer.byteLength(packageSource) > 1024 * 1024) throw fixedError("RESTRICTED_SOURCE_LIMIT"); - const workerSource = await readFile(config.worker, "utf8"); - if (Buffer.byteLength(workerSource) > 65536) throw fixedError("RESTRICTED_ENTRY_LIMIT"); - const sdk = await Promise.all(config.sdk.map(async (m) => ({ ...m, source: await readFile(m.path, "utf8") }))); - if (sdk.some((m) => Buffer.byteLength(m.source) > 2 * 1024 * 1024)) throw fixedError("RESTRICTED_SDK_LIMIT"); - let active, - buffered = Buffer.alloc(0), + const packageSource = await boundedSource(config.module, 1024 * 1024, "RESTRICTED_SOURCE_LIMIT"); + const workerSource = await boundedSource(config.worker, 65536, "RESTRICTED_ENTRY_LIMIT"); + const sdk = []; + for (const m of config.sdk) + sdk.push({ ...m, source: await boundedSource(m.path, 2 * 1024 * 1024, "RESTRICTED_SDK_LIMIT") }); + const active = new Map(); + let buffered = Buffer.alloc(0), stopped = false; const queued = []; const send = (value) => { @@ -83,11 +109,40 @@ export async function serveRestrictedRegistry(config, io = process) { throw fixedError("RESTRICTED_BACKPRESSURE"); io.stdout.write(line); }; + // Linux USER_HZ is 100 on the supported NixOS architectures. Read process + // CPU outside the engine; authored code cannot forge these usage messages. + const sampleCPU = async (call) => { + if (!call.child?.pid) return; + let text; + try { + text = await readFile(`/proc/${call.child.pid}/stat`, "utf8"); + } catch (error) { + if (error.code === "ENOENT" || error.code === "ESRCH") return; + throw error; + } + const fields = text.slice(text.lastIndexOf(")") + 2).split(" "); + const cpuMicroseconds = (Number(fields[11]) + Number(fields[12])) * 10000; + if (!Number.isSafeInteger(cpuMicroseconds) || cpuMicroseconds < (call.cpuMicroseconds ?? 0)) + throw fixedError("RESTRICTED_CPU_ACCOUNTING"); + if (cpuMicroseconds !== (call.cpuMicroseconds ?? 0)) { + call.cpuMicroseconds = cpuMicroseconds; + send({ kind: "usage", invocation: call.message.invocation, frame: call.message.frame, cpuMicroseconds }); + } + }; + const watchCPU = (call) => { + call.cpuTimer = setTimeout(() => { + call.cpuRead = sampleCPU(call).then(() => { + if (call.open) watchCPU(call); + }); + void call.cpuRead.catch(fail); + }, 20); + }; const retire = async (call, reason) => { if (call.retiring) return call.retiring; call.open = false; call.retiring = (async () => { clearTimeout(call.timer); + clearTimeout(call.cpuTimer); const index = queued.indexOf(call); if (index >= 0) queued.splice(index, 1); for (const pending of call.pending.values()) pending.reject(reason); @@ -95,6 +150,14 @@ export async function serveRestrictedRegistry(config, io = process) { // Cancellation can arrive during async directory/socket preparation. // Wait for it to settle before claiming every process/handle is gone. await call.preparing?.catch(() => {}); + clearTimeout(call.cpuTimer); + await call.cpuRead?.catch(() => {}); + let usageFailure; + try { + await sampleCPU(call); + } catch (error) { + usageFailure = error; + } if (call.child?.pid) { try { process.kill(-call.child.pid, "SIGKILL"); @@ -106,7 +169,8 @@ export async function serveRestrictedRegistry(config, io = process) { call.server?.closeAllConnections(); if (call.server) await new Promise((resolve) => call.server.close(resolve)); if (call.directory) await rm(call.directory, { recursive: true, force: true }); - if (active === call) active = undefined; + active.delete(call.message.invocation); + if (usageFailure) throw usageFailure; })(); return call.retiring; }; @@ -115,7 +179,7 @@ export async function serveRestrictedRegistry(config, io = process) { stopped = true; io.stdin.destroy(); for (const call of [...queued]) await retire(call, error); - if (active) await retire(active, error); + await Promise.all([...active.values()].map((call) => retire(call, error))); io.stdout.end(); io.exitCode = 1; }; @@ -129,17 +193,27 @@ export async function serveRestrictedRegistry(config, io = process) { } return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(Buffer.concat(chunks))); }; + const available = (call) => + [...active.values()].filter((entry) => entry.nested === call.nested).length < + (call.nested ? limits.maxNested : limits.maxActive); const pump = () => { - if (!stopped && !active && queued.length) void invoke(queued.shift()).catch(fail); + while (!stopped) { + const index = queued.findIndex(available); + if (index < 0) return; + void invoke(queued.splice(index, 1)[0]).catch(fail); + } }; const finish = async (call, reason, reply) => { - await retire(call, reason); - send(reply); - pump(); + if (reply.kind === "cancelled") call.cancellationReply = reply; + return (call.finished ??= (async () => { + await retire(call, reason); + send(call.cancellationReply ?? reply); + pump(); + })()); }; const invoke = async (call) => { const { message } = call; - active = call; + active.set(message.invocation, call); const terminate = (code) => { if (!call.open) return; void finish(call, fixedError(code), { @@ -169,7 +243,10 @@ export async function serveRestrictedRegistry(config, io = process) { !Object.hasOwn(port, "input") ) throw fixedError("INVALID_RESTRICTED_PORT"); - if (++call.sequence > limits.maxPortCalls || call.pending.size >= limits.maxPortCalls) + if ( + ++call.sequence > Math.min(limits.maxPortCalls, message.budget?.calls ?? limits.maxPortCalls) || + call.pending.size >= limits.maxPortCalls + ) throw fixedError("RESTRICTED_PORT_LIMIT"); const sequence = call.sequence; const result = new Promise((resolve, reject) => call.pending.set(sequence, { resolve, reject })); @@ -229,6 +306,7 @@ export async function serveRestrictedRegistry(config, io = process) { call.child.once("exit", resolve); call.child.once("error", resolve); }); + watchCPU(call); let diagnostics = 0; call.child.stderr.on("data", (chunk) => { if ((diagnostics += chunk.length) > 65536) terminate("RESTRICTED_DIAGNOSTIC_LIMIT"); @@ -246,15 +324,18 @@ export async function serveRestrictedRegistry(config, io = process) { reject(fixedError("EXPIRED_CONTEXT")); return; } - const request = httpRequest({ socketPath: engine, path: "/", method: "POST" }, async (response) => { - try { - const result = await body(response); - if (response.statusCode !== 200) throw fixedError("RESTRICTED_PACKAGE_FAILED"); - resolve(result); - } catch (error) { - reject(error); - } - }); + const request = httpRequest( + { socketPath: engine, path: "/", method: "POST", agent: false }, + async (response) => { + try { + const result = await body(response); + if (response.statusCode !== 200) throw fixedError("RESTRICTED_PACKAGE_FAILED"); + resolve(result); + } catch (error) { + reject(error); + } + }, + ); request.once("error", (error) => { if (call.open && ["ENOENT", "ECONNREFUSED"].includes(error.code)) setTimeout(attempt, 5); else reject(error); @@ -311,9 +392,30 @@ export async function serveRestrictedRegistry(config, io = process) { message.frame.length > 4096 ) throw fixedError("INVALID_RESTRICTED_INVOCATION"); - if ([active, ...queued].some((call) => call?.message.invocation === message.invocation)) + if ( + message.budget !== undefined && + (!object(message.budget) || + Object.keys(message.budget).sort().join(",") !== "calls,deadlineMs,depth,root" || + typeof message.budget.root !== "string" || + !message.budget.root || + message.budget.root.length > 4096 || + !Number.isSafeInteger(message.budget.depth) || + message.budget.depth < 0 || + message.budget.depth > 32 || + !Number.isSafeInteger(message.budget.calls) || + message.budget.calls < 1 || + message.budget.calls > 1000000 || + !Number.isSafeInteger(message.budget.deadlineMs) || + message.budget.deadlineMs < 1 || + message.budget.deadlineMs > 60000) + ) + throw fixedError("INVALID_RESTRICTED_BUDGET"); + if (active.has(message.invocation) || queued.some((call) => call.message.invocation === message.invocation)) throw fixedError("RESTRICTED_DUPLICATE_INVOCATION"); - if (active && queued.length >= limits.maxQueued) { + const call = { message, open: true, pending: new Map(), sequence: 0, nested: (message.budget?.depth ?? 0) > 1 }; + // Descendants cannot wait behind ancestors that hold all nested capacity. + // Reject excess fan-out/depth rather than deadlock an admitted call tree. + if (!available(call) && (call.nested || queued.length >= limits.maxQueued)) { send({ invocation: message.invocation, status: "failed", @@ -321,16 +423,18 @@ export async function serveRestrictedRegistry(config, io = process) { }); return; } - const call = { message, open: true, pending: new Map(), sequence: 0 }; // Deadline includes queue wait: overload cannot extend a transaction frame. - call.timer = setTimeout(() => { - if (!call.open) return; - void finish(call, fixedError("RESTRICTED_DEADLINE"), { - invocation: message.invocation, - status: "failed", - error: { code: "RESTRICTED_DEADLINE", message: "RESTRICTED_DEADLINE" }, - }).catch(fail); - }, limits.deadlineMs); + call.timer = setTimeout( + () => { + if (!call.open) return; + void finish(call, fixedError("RESTRICTED_DEADLINE"), { + invocation: message.invocation, + status: "failed", + error: { code: "RESTRICTED_DEADLINE", message: "RESTRICTED_DEADLINE" }, + }).catch(fail); + }, + Math.min(limits.deadlineMs, message.budget?.deadlineMs ?? limits.deadlineMs), + ); queued.push(call); pump(); return; @@ -346,7 +450,7 @@ export async function serveRestrictedRegistry(config, io = process) { }).catch(fail); return; } - const call = active; + const call = active.get(message.invocation); if (!call || message.invocation !== call.message.invocation || message.frame !== call.message.frame) { if (message.kind === "cancel") send({ kind: "cancelled", invocation: message.invocation, frame: message.frame }); else if (message.kind !== "port-result") throw fixedError("RESTRICTED_UNKNOWN_INVOCATION"); diff --git a/restricted-execution/service.test.mjs b/restricted-execution/service.test.mjs index b3634b2..e127ae8 100644 --- a/restricted-execution/service.test.mjs +++ b/restricted-execution/service.test.mjs @@ -19,6 +19,7 @@ async function fixture(source, options = {}) { const io = { stdin: new PassThrough(), stdout: new PassThrough(), exitCode: 0 }; const waiting = [], queued = []; + const usage = []; let buffer = ""; io.stdout.on("data", (chunk) => { buffer += chunk; @@ -26,6 +27,10 @@ async function fixture(source, options = {}) { while ((index = buffer.indexOf("\n")) >= 0) { const value = JSON.parse(buffer.slice(0, index)); buffer = buffer.slice(index + 1); + if (value.kind === "usage") { + usage.push(value); + continue; + } if (waiting.length) waiting.shift()(value); else queued.push(value); } @@ -50,7 +55,7 @@ async function fixture(source, options = {}) { ); assert.deepEqual(await receive(), { abi, kind: "ready", exports: ["test"] }); const send = (value) => io.stdin.write(JSON.stringify({ abi, ...value }) + "\n"); - const start = (input = null) => { + const start = (input = null, budget) => { const invocation = "invoke-" + ++next, frame = "frame-" + next; send({ @@ -62,6 +67,7 @@ async function fixture(source, options = {}) { receiver: null, input, effect: "snapshot", + ...(budget ? { budget } : {}), }); return { invocation, frame }; }; @@ -70,6 +76,7 @@ async function fixture(source, options = {}) { receive, send, start, + usage, async close() { await service.close(); await rm(directory, { recursive: true, force: true }); @@ -78,6 +85,94 @@ async function fixture(source, options = {}) { } const registry = (code) => `export const registry={exports:['test'],async invoke(request,channel){${code}}};`; +test("parallel frames route independently and cancellation cannot retire a neighbor", async () => { + const f = await fixture( + registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:request.input});"), + ); + try { + const first = f.start("a"), + second = f.start("b"); + const ports = new Map([await f.receive(), await f.receive()].map((port) => [port.invocation, port])); + assert.equal(ports.get(first.invocation).input, "a"); + assert.equal(ports.get(second.invocation).input, "b"); + f.send({ kind: "cancel", ...first }); + assert.equal((await f.receive()).kind, "cancelled"); + f.send({ + kind: "port-result", + ...first, + sequence: ports.get(first.invocation).sequence, + status: "returned", + output: "late", + }); + f.send({ + kind: "port-result", + ...second, + sequence: ports.get(second.invocation).sequence, + status: "returned", + output: "survived", + }); + const result = await f.receive(); + assert.equal(result.invocation, second.invocation); + assert.equal(result.output, "survived"); + assert.equal(f.io.exitCode, 0); + } finally { + await f.close(); + } +}); + +test("A to B to A uses reserved nested capacity while both root lanes are saturated", async () => { + const source = registry( + "if(request.input==='leaf')return 'done'; return await channel({portId:'next',action:{kind:'state',primitive:'read'},input:request.input});", + ); + const a = await fixture(source, { maxActive: 1, maxNested: 1, maxQueued: 1 }); + const b = await fixture(source, { maxActive: 1, maxNested: 1, maxQueued: 1 }); + const budget = (depth) => ({ root: "root-chain", depth, deadlineMs: 2000, calls: 20 }); + try { + const parent = a.start("B", budget(1)), + aPort = await a.receive(); + const neighbor = b.start("held", { ...budget(1), root: "other-root" }); + await b.receive(); + const child = b.start("A", budget(2)), + bPort = await b.receive(); + const leaf = a.start("leaf", budget(3)); + const leafResult = await a.receive(); + assert.equal(leafResult.invocation, leaf.invocation); + assert.equal(leafResult.output, "done"); + b.send({ kind: "port-result", ...child, sequence: bPort.sequence, status: "returned", output: leafResult.output }); + const childResult = await b.receive(); + a.send({ + kind: "port-result", + ...parent, + sequence: aPort.sequence, + status: "returned", + output: childResult.output, + }); + assert.equal((await a.receive()).output, "done"); + b.send({ kind: "cancel", ...neighbor }); + assert.equal((await b.receive()).kind, "cancelled"); + } finally { + await a.close(); + await b.close(); + } +}); + +test("the remaining root wall budget overrides a longer adapter deadline", async () => { + const f = await fixture( + registry("return await channel({portId:'held',action:{kind:'state',primitive:'read'},input:null});"), + { deadlineMs: 2000 }, + ); + try { + const started = performance.now(); + f.start(null, { root: "short-root", depth: 2, deadlineMs: 150, calls: 10 }); + await f.receive(); + const result = await f.receive(); + assert.equal(result.error.code, "RESTRICTED_DEADLINE"); + assert.ok(performance.now() - started < 1000); + } finally { + await f.close(); + } +}); + test("real engine returns results and resets all globals between frames", async () => { const f = await fixture( registry( @@ -222,6 +317,10 @@ test("initialization and regex loops are killed, then a fresh frame can start", f.start(); const result = await f.receive(); assert.equal(result.error.code, "RESTRICTED_DEADLINE"); + assert.ok( + f.usage.some((value) => value.cpuMicroseconds >= 50000), + "External watchdog did not account spinning engine CPU", + ); } } finally { await f.close(); @@ -268,7 +367,7 @@ test("cancellation during asynchronous preparation leaves no running frame", asy test("bounded FIFO rejects overload without losing active or queued frames", async () => { const f = await fixture( registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:request.input});"), - { maxQueued: 2 }, + { maxQueued: 2, maxActive: 1 }, ); try { const first = f.start(1), @@ -307,7 +406,7 @@ test("bounded FIFO rejects overload without losing active or queued frames", asy test("queue wait consumes the deadline and cancelled work never reaches its ports", async () => { const f = await fixture( registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"), - { deadlineMs: 350 }, + { deadlineMs: 350, maxActive: 1 }, ); try { const first = f.start();