diff --git a/PORTABLE_ARTIFACTS.md b/PORTABLE_ARTIFACTS.md index b285675..bde2e74 100644 --- a/PORTABLE_ARTIFACTS.md +++ b/PORTABLE_ARTIFACTS.md @@ -91,3 +91,59 @@ proves that a component-only edit preserves a mixed package's service digest. Run `nix run .#camino-component-source-qualification` from the monorepo to materialize separate retained Git resources and compile an ordinary Nix candidate. This is a source/build qualification harness; it does not deploy a workspace. + +## Restricted execution in a workspace VM + +Add `service.restricted = true` to a checked portable package to select the pinned +`workerd` executor. Generated bindings and the ordinary transaction-port protocol +stay the same. All server exports must belong to one checked portable registry; +remote exports and custom process services retain their ordinary VM execution +path. The restricted build rejects Node imports and unresolved dependencies, +and supplies an explicit SDK for value codecs, identities and declared graph ports. + +```nix +service = { + restricted = true; + limits = { + deadlineMs = 4000; + maxBytes = 1048576; + maxPortCalls = 1000; + maxQueued = 8; + }; +}; +``` + +Each invocation gets a fresh engine process. The adapter admits one active frame +and a bounded FIFO; queue wait consumes the deadline. Overflow is an explicit +`RESTRICTED_QUEUE_FULL` result. Cancellation kills the engine process group and +removes its sockets before acknowledgement. All graph calls inherit the trusted +invocation identity; authored code cannot choose another frame or workspace. +Native graph grants and transaction commit remain enforced by Camino outside +the guest. Initialization and synchronous loops are subject to the same deadline. + +Direct network access, native filesystem imports, ambient clock and entropy, +parent environment credentials, and unlisted modules are denied. Explicit dates, +`Intl.Segmenter`, deterministic hashes and normal JavaScript computation remain +available. Host-managed HTTP bridges expose only the current frame's declared +graph ports. Authored serialization is untrusted protocol input. + +This is an inner execution restriction, **not a hardened shared-process sandbox**. +One workspace owns the surrounding VM. Systemd limits the entire package cgroup, +including adapter and child engines; memory exhaustion terminates that group. +No per-isolate memory protection is claimed. A guest user can control its VM, +so authoritative graph grants and credentials must remain outside it. + +Qualification commands: + +```sh +nix build .#checks.x86_64-linux.camino-restricted-execution +nix build .#checks.x86_64-linux.camino-restricted-package-runners +nix build .#checks.x86_64-linux.camino-restricted-execution-vm +``` + +The first uses the actual pinned engine for frame isolation, graph ports, +SDK codecs, denied ambient access, cancellation and overload. The second runs +ordinary package dispatch, rollback, PostgreSQL and browser replica tests with +the restricted reader fixture. The VM check adds actual cgroup allocation +exhaustion, neighboring-service survival, native socket egress rejection and +fresh-frame timing. VM qualification is independent of hosted rollout. diff --git a/artifacts.nix b/artifacts.nix index f5ca613..735df8e 100644 --- a/artifacts.nix +++ b/artifacts.nix @@ -96,6 +96,18 @@ let esbuild = "${pkgs.esbuild}/bin/esbuild"; shell = "${pkgs.runtimeShell}"; serviceAdapter = ./portable-service.mjs; + restricted = + if service.restricted or false then + let + executor = import ./restricted-execution { inherit pkgs core; }; + in + { + runtime = "${executor.runtime}/bin/quixos-restricted-registry"; + sdk = "${executor.sdk}/core.mjs"; + limits = service.limits or { }; + } + else + null; } ); compiled = diff --git a/build-artifacts.mjs b/build-artifacts.mjs index 040e3b3..4715d82 100644 --- a/build-artifacts.mjs +++ b/build-artifacts.mjs @@ -128,7 +128,8 @@ for (const { target, mode, settings } of builds) { const bundlePath = mode === "remote" ? "share/quixos/remote/server/module.mjs" : `share/quixos/portable/${target}/module.mjs`; await fs.mkdir(path.dirname(path.join(output, bundlePath)), { recursive: true }); - const coreImport = target === "browser" ? coreURL : coreServer; + const restricted = !!config.restricted && target === "server"; + const coreImport = target === "browser" ? coreURL : restricted ? config.restricted.sdk : coreServer; const metadata = path.join(work, "bundle-meta.json"); measureBuild("portable-bundle", { package: pkg.revision, target, mode }, () => execFileSync( @@ -136,7 +137,7 @@ for (const { target, mode, settings } of builds) { [ entry, "--bundle", - `--platform=${target === "browser" ? "browser" : "node"}`, + `--platform=${target === "browser" || restricted ? "browser" : "node"}`, "--format=esm", "--minify-whitespace", "--target=es2024", @@ -158,7 +159,7 @@ for (const { target, mode, settings } of builds) { (i) => i.external && i.path !== coreImport && - !(target === "server" && builtinModules.includes(i.path.replace(/^node:/, ""))), + !(target === "server" && !restricted && builtinModules.includes(i.path.replace(/^node:/, ""))), ), ) ) diff --git a/build-service.mjs b/build-service.mjs index 5e3ca09..57e2335 100644 --- a/build-service.mjs +++ b/build-service.mjs @@ -13,6 +13,62 @@ export async function buildService({ recordFile, }) { const relative = compiler.artifactPath; + if (config.restricted) { + if (config.service.entry || pkg.exports.some((e) => e.execution.kind === "remote")) + throw Error( + "Restricted services require portable snapshot/transaction exports; remote/custom services keep their VM path", + ); + const modules = implementations.filter((i) => i.target === "server"); + if (modules.length !== 1 || pkg.exports.some((e) => !modules[0].exports.includes(e.id))) + throw Error("Restricted service requires exactly one checked portable registry covering all exports"); + const limits = config.restricted.limits; + if (!limits || typeof limits !== "object" || Array.isArray(limits)) + throw Error("Restricted limits must be an object"); + if (Object.keys(limits).some((key) => !["deadlineMs", "maxBytes", "maxPortCalls", "maxQueued"].includes(key))) + throw Error("Unknown restricted service limit"); + for (const [key, min, max] of [ + ["deadlineMs", 1, 60000], + ["maxBytes", 1024, 4194304], + ["maxPortCalls", 1, 10000], + ["maxQueued", 0, 64], + ]) + if (limits[key] !== undefined && (!Number.isSafeInteger(limits[key]) || limits[key] < min || limits[key] > max)) + throw Error("Invalid restricted service limit " + key); + await fs.mkdir(path.join(output, "bin"), { recursive: true }); + await fs.mkdir(path.join(output, "share/quixos/service"), { recursive: true }); + const configFile = "share/quixos/service/restricted.json"; + const module = modules[0]; + await fs.writeFile( + path.join(output, configFile), + JSON.stringify({ + module: path.relative("share/quixos/service", module.path), + registryExport: module.registryExport, + exports: module.exports, + sdk: [{ name: config.restricted.sdk, path: config.restricted.sdk }], + ...limits, + }), + ); + await fs.writeFile( + path.join(output, "bin/service"), + `#!${config.shell}\nservice_root=\${0%/*}/..\nexec ${config.restricted.runtime} "$service_root/${configFile}"\n`, + { mode: 0o755 }, + ); + const body = { + id: "service", + target: "server", + compatibility: "quixos-service-v1", + path: "bin/service", + ports: "quixos-transaction-ports-v1", + exports: module.exports, + files: [ + await recordFile("bin/service", "text/x-shellscript"), + await recordFile(configFile, "application/json"), + ...module.files, + ], + }; + implementations.unshift({ ...body, contentDigest: compiler.implementationDigest(body) }); + return; + } if (pkg.exports.length || config.service.entry) { await fs.mkdir(path.join(output, "bin"), { recursive: true }); await fs.mkdir(path.join(output, "share/quixos/service"), { recursive: true }); diff --git a/build-tools.nix b/build-tools.nix index c80795d..ad04960 100644 --- a/build-tools.nix +++ b/build-tools.nix @@ -8,6 +8,10 @@ builtins.path { let name = builtins.baseNameOf filename; in - type == "directory" - || (builtins.match ".*\\.mjs" name != null && builtins.match ".*\\.test\\.mjs" name == null); + (type == "directory" && filename == toString ./.) + || ( + builtins.dirOf filename == toString ./. + && builtins.match ".*\\.mjs" name != null + && builtins.match ".*\\.test\\.mjs" name == null + ); } diff --git a/restricted-execution/default.nix b/restricted-execution/default.nix new file mode 100644 index 0000000..383bb93 --- /dev/null +++ b/restricted-execution/default.nix @@ -0,0 +1,47 @@ +{ pkgs, core }: +let + workerd = import ./workerd.nix { inherit pkgs; }; + # Transaction-port/value adapters do not need the browser's CRDT/Wasm engine. + # Keep this surface explicit; it is bundled into checked restricted handlers. + sdk = pkgs.runCommand "quixos-restricted-sdk" { nativeBuildInputs = [ pkgs.esbuild ]; } '' + mkdir -p "$out" + cat > entry.mjs <<'JS' + export {definePortableRegistry,QxReject} from '${core}/dist/portable.js'; + export {encodeArgument,decodeArgument,canonicalJson,parseJson} from '${core}/dist/value.js'; + export {workspaceId,operationId,objectId,entryId,hashIdentity} from '${core}/dist/identity.js'; + export {reference} from '${core}/dist/references.js'; + export {CodecError,ExecutionError} from '${core}/dist/limits.js'; + JS + esbuild entry.mjs --bundle --platform=neutral --format=esm --target=es2024 --outfile="$out/core.mjs" + ''; + runtime = + pkgs.runCommand "quixos-restricted-execution" { nativeBuildInputs = [ pkgs.makeWrapper ]; } + '' + mkdir -p "$out/lib" "$out/bin" + cp ${./service.mjs} "$out/lib/service.mjs" + cp ${./worker.mjs} "$out/lib/worker.mjs" + makeWrapper ${pkgs.nodejs_24}/bin/node "$out/bin/quixos-restricted-registry" \ + --add-flags "$out/lib/service.mjs" \ + --set QUIXOS_RESTRICTED_WORKERD ${workerd}/bin/workerd \ + --set QUIXOS_RESTRICTED_WORKER "$out/lib/worker.mjs" + ''; + qualification = + pkgs.runCommand "quixos-restricted-execution-qualification" + { + nativeBuildInputs = [ pkgs.nodejs_24 ]; + QUIXOS_TEST_WORKERD = "${workerd}/bin/workerd"; + QUIXOS_TEST_SDK = "${sdk}/core.mjs"; + } + '' + node --test ${./.}/service.test.mjs + touch "$out" + ''; +in +{ + inherit + workerd + sdk + runtime + qualification + ; +} diff --git a/restricted-execution/service.mjs b/restricted-execution/service.mjs new file mode 100644 index 0000000..4e5967d --- /dev/null +++ b/restricted-execution/service.mjs @@ -0,0 +1,404 @@ +import { spawn } from "node:child_process"; +import { createServer, request as httpRequest } from "node:http"; +import { mkdtemp, readFile, writeFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, dirname } from "node:path"; + +const abi = "quixos-transaction-ports-v1"; +const fixedError = (code) => Object.assign(Error(code), { code }); +const object = (v) => !!v && typeof v === "object" && !Array.isArray(v); +const absolute = (value) => + typeof value === "string" && + value.startsWith("/") && + !/[\s\u0000-\u001f]/.test(value) && + !value.split("/").includes(".."); + +/** Trusted stdio adapter inside one workspace VM. Engine processes never own + * coordinator credentials. The supervising service must impose VM/OS limits. + * Per-frame process replacement prevents retained globals/handles crossing calls. */ +export async function serveRestrictedRegistry(config, io = process) { + if ( + !object(config) || + !absolute(config.workerd) || + !absolute(config.worker) || + !absolute(config.module) || + !/^[$A-Z_a-z][$\w]*$/.test(config.registryExport) || + !Array.isArray(config.exports) || + !config.exports.length || + config.exports.length > 10000 || + config.exports.some((x) => typeof x !== "string" || !x || x.length > 4096) || + new Set(config.exports).size !== config.exports.length || + !Array.isArray(config.sdk) || + config.sdk.length > 8 || + config.sdk.some( + (m) => + !object(m) || + !absolute(m.path) || + typeof m.name !== "string" || + !m.name || + /[\u0000-\u001f]/.test(m.name) || + ["entry.mjs", "package.mjs"].includes(m.name), + ) || + new Set(config.sdk.map((m) => m.name)).size !== config.sdk.length + ) + throw fixedError("INVALID_RESTRICTED_CONFIGURATION"); + const limits = { + maxBytes: config.maxBytes ?? 1024 * 1024, + maxPortCalls: config.maxPortCalls ?? 1000, + deadlineMs: config.deadlineMs ?? 4000, + maxQueued: config.maxQueued ?? 8, + }; + if ( + !Number.isSafeInteger(limits.maxBytes) || + limits.maxBytes < 1024 || + limits.maxBytes > 4 * 1024 * 1024 || + !Number.isSafeInteger(limits.maxPortCalls) || + limits.maxPortCalls < 1 || + limits.maxPortCalls > 10000 || + !Number.isSafeInteger(limits.deadlineMs) || + limits.deadlineMs < 1 || + limits.deadlineMs > 60000 || + !Number.isSafeInteger(limits.maxQueued) || + limits.maxQueued < 0 || + limits.maxQueued > 64 + ) + 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), + stopped = false; + const queued = []; + const send = (value) => { + if (stopped) return; + const line = JSON.stringify({ abi, ...value }) + "\n"; + if ( + Buffer.byteLength(line) > limits.maxBytes || + io.stdout.writableLength + Buffer.byteLength(line) > limits.maxBytes * 2 + ) + throw fixedError("RESTRICTED_BACKPRESSURE"); + io.stdout.write(line); + }; + const retire = async (call, reason) => { + if (call.retiring) return call.retiring; + call.open = false; + call.retiring = (async () => { + clearTimeout(call.timer); + const index = queued.indexOf(call); + if (index >= 0) queued.splice(index, 1); + for (const pending of call.pending.values()) pending.reject(reason); + call.pending.clear(); + // Cancellation can arrive during async directory/socket preparation. + // Wait for it to settle before claiming every process/handle is gone. + await call.preparing?.catch(() => {}); + if (call.child?.pid) { + try { + process.kill(-call.child.pid, "SIGKILL"); + } catch (error) { + if (error.code !== "ESRCH") throw error; + } + await call.exited; + } + 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; + })(); + return call.retiring; + }; + const fail = async (error) => { + if (stopped) return; + stopped = true; + io.stdin.destroy(); + for (const call of [...queued]) await retire(call, error); + if (active) await retire(active, error); + io.stdout.end(); + io.exitCode = 1; + }; + const body = async (stream) => { + const chunks = []; + let bytes = 0; + for await (const chunk of stream) { + bytes += chunk.length; + if (bytes > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT"); + chunks.push(chunk); + } + return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(Buffer.concat(chunks))); + }; + const pump = () => { + if (!stopped && !active && queued.length) void invoke(queued.shift()).catch(fail); + }; + const finish = async (call, reason, reply) => { + await retire(call, reason); + send(reply); + pump(); + }; + const invoke = async (call) => { + const { message } = call; + active = call; + const terminate = (code) => { + if (!call.open) return; + void finish(call, fixedError(code), { + invocation: message.invocation, + status: "failed", + error: { code, message: code }, + }).catch(fail); + }; + call.terminate = terminate; + try { + let engine; + call.preparing = (async () => { + call.directory = await mkdtemp(join(tmpdir(), "qx-frame-")); + if (!call.open) return; + const broker = join(call.directory, "broker.sock"); + engine = join(call.directory, "engine.sock"); + call.server = createServer(async (request, response) => { + try { + if (!call.open || request.method !== "POST" || request.url !== "/port") throw fixedError("EXPIRED_CONTEXT"); + const port = await body(request); + if ( + !call.open || + !object(port) || + Object.keys(port).some((key) => !["portId", "action", "input", "key"].includes(key)) || + typeof port.portId !== "string" || + !object(port.action) || + !Object.hasOwn(port, "input") + ) + throw fixedError("INVALID_RESTRICTED_PORT"); + if (++call.sequence > 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 })); + void result.catch(() => {}); + send({ + kind: "port", + invocation: message.invocation, + frame: message.frame, + sequence, + portId: port.portId, + action: { ...port.action, ...(port.key === undefined ? {} : { key: port.key }) }, + input: port.input, + }); + const value = await result; + if (!call.open) throw fixedError("EXPIRED_CONTEXT"); + const encoded = JSON.stringify(value); + if (Buffer.byteLength(encoded) > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT"); + response.writeHead(200, { "content-type": "application/json" }); + response.end(encoded); + } catch { + response.writeHead(409); + response.end('{"status":"failed","error":{"code":"RESTRICTED_PORT_FAILED"}}'); + } + }); + await new Promise((resolve, reject) => { + call.server.once("error", reject); + call.server.listen(broker, resolve); + }); + await writeFile(join(call.directory, "package.mjs"), packageSource); + await writeFile(join(call.directory, "entry.mjs"), workerSource); + const modules = [ + { name: "entry.mjs", esModule: 'embed "entry.mjs"' }, + { name: "package.mjs", esModule: 'embed "package.mjs"' }, + ]; + for (const [index, module] of sdk.entries()) { + const file = `sdk-${index}.mjs`; + await writeFile(join(call.directory, file), module.source); + modules.push({ name: module.name, esModule: "embed " + JSON.stringify(file) }); + } + const setting = { registryExport: config.registryExport, exports: config.exports, maxBytes: limits.maxBytes }; + const configuration = `using Workerd = import "/workerd/workerd.capnp"; const config :Workerd.Config = ( + services = [(name="main", worker=(modules=[${modules.map((m) => `(name=${JSON.stringify(m.name)}, esModule=${m.esModule})`).join(",")}], + compatibilityDate="2026-10-05", globalOutbound="deny", + bindings=[(name="BROKER", service="broker"),(name="CONFIG",json=${JSON.stringify(JSON.stringify(setting))})])), + (name="broker",external=(address=${JSON.stringify("unix:" + broker)},http=())), (name="deny",network=(allow=[]))], + sockets=[(name="frame",address=${JSON.stringify("unix:" + engine)},http=(),service="main")]);`; + const file = join(call.directory, "config.capnp"); + await writeFile(file, configuration); + if (!call.open) return; + call.child = spawn(config.workerd, ["serve", file], { + detached: true, + env: { TZ: "UTC", LANG: "C.UTF-8" }, + cwd: call.directory, + stdio: ["ignore", "ignore", "pipe"], + }); + call.exited = new Promise((resolve) => { + call.child.once("exit", resolve); + call.child.once("error", resolve); + }); + let diagnostics = 0; + call.child.stderr.on("data", (chunk) => { + if ((diagnostics += chunk.length) > 65536) terminate("RESTRICTED_DIAGNOSTIC_LIMIT"); + }); + call.child.once("error", () => terminate("RESTRICTED_START_FAILED")); + void call.exited.then(() => { + if (call.open) terminate("RESTRICTED_ENGINE_EXITED"); + }); + })(); + await call.preparing; + if (!call.open) return; + const result = await new Promise((resolve, reject) => { + const attempt = () => { + if (!call.open) { + 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); + } + }); + request.once("error", (error) => { + if (call.open && ["ENOENT", "ECONNREFUSED"].includes(error.code)) setTimeout(attempt, 5); + else reject(error); + }); + request.end(JSON.stringify(message)); + }; + attempt(); + }); + if (!call.open) return; + if ( + !object(result) || + !["returned", "failed"].includes(result.status) || + Object.keys(result).some( + (key) => !(result.status === "returned" ? ["status", "output"] : ["status", "error"]).includes(key), + ) || + (result.status === "returned" + ? !Object.hasOwn(result, "output") + : !object(result.error) || !/^[A-Z0-9_]{1,128}$/.test(result.error.code)) + ) + throw fixedError("RESTRICTED_PACKAGE_FAILED"); + // Authored code can alter serialization prototypes. Treat even the + // platform worker's JSON as untrusted; never accept routing/ABI fields. + const reply = + result.status === "returned" + ? { invocation: message.invocation, status: "returned", output: result.output } + : { + invocation: message.invocation, + status: "failed", + error: { code: result.error.code, message: result.error.code }, + }; + await finish(call, fixedError("INVOCATION_ENDED"), reply); + } catch (error) { + if (call.open) { + const code = /^RESTRICTED_[A-Z_]+$/.test(error.code ?? "") ? error.code : "RESTRICTED_PACKAGE_FAILED"; + await finish(call, error, { invocation: message.invocation, status: "failed", error: { code, message: code } }); + } + } + }; + const receive = (message) => { + if (!object(message) || message.abi !== abi) throw fixedError("RESTRICTED_ABI_MISMATCH"); + if (message.kind === "shutdown") { + void fail(fixedError("SERVICE_SHUTDOWN")); + return; + } + if (message.kind === "invoke") { + if ( + !config.exports.includes(message.exportId) || + !["snapshot", "transaction"].includes(message.effect) || + typeof message.invocation !== "string" || + !message.invocation || + message.invocation.length > 4096 || + typeof message.frame !== "string" || + !message.frame || + message.frame.length > 4096 + ) + throw fixedError("INVALID_RESTRICTED_INVOCATION"); + if ([active, ...queued].some((call) => call?.message.invocation === message.invocation)) + throw fixedError("RESTRICTED_DUPLICATE_INVOCATION"); + if (active && queued.length >= limits.maxQueued) { + send({ + invocation: message.invocation, + status: "failed", + error: { code: "RESTRICTED_QUEUE_FULL", message: "RESTRICTED_QUEUE_FULL" }, + }); + 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); + queued.push(call); + pump(); + return; + } + const waiting = queued.find( + (call) => call.message.invocation === message.invocation && call.message.frame === message.frame, + ); + if (waiting && message.kind === "cancel") { + void finish(waiting, fixedError("INVOCATION_CANCELLED"), { + kind: "cancelled", + invocation: message.invocation, + frame: message.frame, + }).catch(fail); + return; + } + const call = active; + 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"); + return; + } + if (message.kind === "cancel") { + void finish(call, fixedError("INVOCATION_CANCELLED"), { + kind: "cancelled", + invocation: message.invocation, + frame: message.frame, + }).catch(fail); + return; + } + if (message.kind !== "port-result" || !["returned", "failed"].includes(message.status)) + throw fixedError("INVALID_RESTRICTED_RESULT"); + const pending = call.pending.get(message.sequence); + if (!pending) throw fixedError("RESTRICTED_UNKNOWN_PORT"); + call.pending.delete(message.sequence); + pending.resolve(message); + }; + io.stdin.on("data", (chunk) => { + if (stopped) return; + try { + let offset = 0; + while (offset < chunk.length) { + const newline = chunk.indexOf(10, offset), + end = newline < 0 ? chunk.length : newline; + if (buffered.length + end - offset > limits.maxBytes) throw fixedError("RESTRICTED_MESSAGE_LIMIT"); + buffered = Buffer.concat([buffered, chunk.subarray(offset, end)]); + if (newline < 0) break; + receive(JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(buffered))); + buffered = Buffer.alloc(0); + offset = newline + 1; + if (stopped) break; + } + } catch (error) { + void fail(error); + } + }); + io.stdin.on("end", () => void fail(fixedError("COORDINATOR_DISCONNECTED"))); + io.stdin.on("error", (error) => void fail(error)); + io.stdout.on("error", (error) => void fail(error)); + send({ kind: "ready", exports: config.exports }); + return { close: () => fail(fixedError("SERVICE_SHUTDOWN")) }; +} + +if (process.argv[1] === new URL(import.meta.url).pathname) { + const config = JSON.parse(await readFile(process.argv[2], "utf8")); + await serveRestrictedRegistry({ + ...config, + workerd: process.env.QUIXOS_RESTRICTED_WORKERD, + worker: process.env.QUIXOS_RESTRICTED_WORKER, + module: join(dirname(process.argv[2]), config.module), + }); +} diff --git a/restricted-execution/service.test.mjs b/restricted-execution/service.test.mjs new file mode 100644 index 0000000..b3634b2 --- /dev/null +++ b/restricted-execution/service.test.mjs @@ -0,0 +1,327 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { PassThrough } from "node:stream"; +import { mkdtemp, writeFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { serveRestrictedRegistry } from "./service.mjs"; + +const workerd = process.env.QUIXOS_TEST_WORKERD; +if (!workerd) throw Error("Set QUIXOS_TEST_WORKERD to the pinned engine"); +const worker = fileURLToPath(new URL("./worker.mjs", import.meta.url)); +const abi = "quixos-transaction-ports-v1"; +let next = 0; +async function fixture(source, options = {}) { + const directory = await mkdtemp(join(tmpdir(), "qx-restricted-test-")); + const module = join(directory, "module.mjs"); + await writeFile(module, source); + const io = { stdin: new PassThrough(), stdout: new PassThrough(), exitCode: 0 }; + const waiting = [], + queued = []; + let buffer = ""; + io.stdout.on("data", (chunk) => { + buffer += chunk; + let index; + while ((index = buffer.indexOf("\n")) >= 0) { + const value = JSON.parse(buffer.slice(0, index)); + buffer = buffer.slice(index + 1); + if (waiting.length) waiting.shift()(value); + else queued.push(value); + } + }); + const receive = async () => { + if (queued.length) return queued.shift(); + let timer; + try { + return await Promise.race([ + new Promise((resolve) => waiting.push(resolve)), + new Promise((_, reject) => { + timer = setTimeout(() => reject(Error("TEST_MESSAGE_TIMEOUT")), 5000); + }), + ]); + } finally { + clearTimeout(timer); + } + }; + const service = await serveRestrictedRegistry( + { workerd, worker, module, registryExport: "registry", exports: ["test"], sdk: [], ...options }, + io, + ); + 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 invocation = "invoke-" + ++next, + frame = "frame-" + next; + send({ + kind: "invoke", + invocation, + frame, + exportId: "test", + workspace: "workspace:restricted", + receiver: null, + input, + effect: "snapshot", + }); + return { invocation, frame }; + }; + return { + io, + receive, + send, + start, + async close() { + await service.close(); + await rm(directory, { recursive: true, force: true }); + }, + }; +} +const registry = (code) => `export const registry={exports:['test'],async invoke(request,channel){${code}}};`; + +test("real engine returns results and resets all globals between frames", async () => { + const f = await fixture( + registry( + "globalThis.retained=(globalThis.retained??0)+1; return {count:globalThis.retained,text:new Intl.Segmenter('zh',{granularity:'word'}).segment('雨后的书店').containing(3).segment};", + ), + ); + try { + for (let n = 0; n < 3; n++) { + const call = f.start(); + const result = await f.receive(); + assert.equal(result.invocation, call.invocation); + assert.equal(result.status, "returned"); + assert.equal(result.output.count, 1); + assert.equal(typeof result.output.text, "string"); + } + } finally { + await f.close(); + } +}); +test("captured frame is the sole source of nested port identities", async () => { + const f = await fixture( + registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"), + ); + try { + const call = f.start(); + const port = await f.receive(); + assert.equal(port.kind, "port"); + assert.equal(port.frame, call.frame); + assert.equal(port.invocation, call.invocation); + f.send({ kind: "port-result", ...call, sequence: port.sequence, status: "returned", output: 42 }); + const result = await f.receive(); + assert.equal(result.status, "returned"); + assert.equal(result.output, 42); + } finally { + await f.close(); + } +}); +test("authored serialization cannot forge result routing or ABI fields", async () => { + const f = await fixture( + registry( + "Object.prototype.toJSON=function(){return this.status==='returned'?{status:'returned',output:42,invocation:'forged',abi:'forged'}:this;};return 42;", + ), + ); + try { + const call = f.start(); + const result = await f.receive(); + assert.equal(result.invocation, call.invocation); + assert.equal(result.abi, abi); + assert.equal(result.status, "failed"); + assert.equal(result.error.code, "RESTRICTED_PACKAGE_FAILED"); + } finally { + await f.close(); + } +}); +test("the checked portable SDK keeps codecs and nested graph ports inside the frame", async () => { + const sdk = process.env.QUIXOS_TEST_SDK; + assert.ok(sdk, "Set QUIXOS_TEST_SDK to the checked portable SDK"); + const C = await import(sdk), + workspace = C.workspaceId("workspace:restricted"); + const integer = { kind: "scalar", name: "int64" }, + unit = { kind: "builtin", name: "unit" }; + const declarations = [ + { + id: "test", + name: "test", + effect: "snapshot", + receiver: null, + input: integer, + output: integer, + methods: [ + { + name: "get", + portName: "count", + portId: "count-read", + action: { kind: "state", primitive: "read" }, + input: unit, + output: integer, + keyed: false, + }, + ], + }, + ]; + const source = `import {definePortableRegistry} from 'qx:core'; export const registry=definePortableRegistry(${JSON.stringify(declarations)},{test:async ctx=>ctx.input+await ctx.ports.count.get()});`; + const f = await fixture(source, { sdk: [{ name: "qx:core", path: sdk }] }); + try { + const call = f.start(C.encodeArgument(integer, 2n, { workspace })); + const port = await f.receive(); + assert.equal(port.kind, "port"); + assert.equal(C.decodeArgument(unit, port.input, { workspace }), null); + f.send({ + kind: "port-result", + ...call, + sequence: port.sequence, + status: "returned", + output: C.encodeArgument(integer, 4n, { workspace }), + }); + const result = await f.receive(); + assert.equal(result.status, "returned"); + assert.equal(C.decodeArgument(integer, result.output, { workspace }), 6n); + } finally { + await f.close(); + } +}); +test("ambient clock, entropy, network and native imports are denied", async () => { + for (const code of [ + "Date.now()", + "new Date()", + "new Intl.DateTimeFormat().format()", + "new Intl.DateTimeFormat().formatToParts()", + "Math.random()", + "crypto.randomUUID()", + "fetch('https://example.com')", + "(await import('node:fs')).readFileSync('/etc/passwd')", + "(await import('./unlisted.mjs')).secret", + ]) { + const f = await fixture(registry(`return ${code};`)); + try { + f.start(); + assert.equal((await f.receive()).status, "failed", code); + } finally { + await f.close(); + } + } +}); +test("explicit dates remain useful without an implicit clock", async () => { + const f = await fixture(registry("return new Date('2026-01-01T00:00:00Z').toISOString();")); + try { + f.start(); + assert.equal((await f.receive()).output, "2026-01-01T00:00:00.000Z"); + } finally { + await f.close(); + } +}); +test("initialization and regex loops are killed, then a fresh frame can start", async () => { + for (const source of [ + "while(true){};" + registry("return 1;"), + registry("return /(a+)+$/.test('a'.repeat(100000)+'!');"), + ]) { + const f = await fixture(source, { deadlineMs: 350 }); + try { + for (let n = 0; n < 2; n++) { + f.start(); + const result = await f.receive(); + assert.equal(result.error.code, "RESTRICTED_DEADLINE"); + } + } finally { + await f.close(); + } + } +}); +test("cancellation kills the entire frame before acknowledgment and rejects late port replies", async () => { + const f = await fixture( + registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"), + ); + try { + const call = f.start(); + const port = await f.receive(); + f.send({ kind: "cancel", ...call }); + const cancelled = await f.receive(); + assert.equal(cancelled.kind, "cancelled"); + f.send({ kind: "port-result", ...call, sequence: port.sequence, status: "returned", output: "late" }); + const next = f.start(); + const nextPort = await f.receive(); + assert.equal(nextPort.invocation, next.invocation); + f.send({ kind: "port-result", ...next, sequence: nextPort.sequence, status: "returned", output: "fresh" }); + const result = await f.receive(); + assert.equal(result.output, "fresh"); + assert.equal(result.invocation, next.invocation); + } finally { + await f.close(); + } +}); +test("cancellation during asynchronous preparation leaves no running frame", async () => { + const f = await fixture(registry("return 1;")); + try { + for (let n = 0; n < 5; n++) { + const call = f.start(); + f.send({ kind: "cancel", ...call }); + assert.equal((await f.receive()).kind, "cancelled"); + } + f.start(); + assert.equal((await f.receive()).output, 1); + } finally { + await f.close(); + } +}); + +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 }, + ); + try { + const first = f.start(1), + port = await f.receive(); + const second = f.start(2), + third = f.start(3), + overflow = f.start(4); + const refused = await f.receive(); + assert.equal(refused.invocation, overflow.invocation); + assert.equal(refused.error.code, "RESTRICTED_QUEUE_FULL"); + f.send({ kind: "cancel", ...second }); + assert.equal((await f.receive()).invocation, second.invocation); + f.send({ kind: "port-result", ...first, sequence: port.sequence, status: "returned", output: 1 }); + assert.equal((await f.receive()).output, 1); + const nextPort = await f.receive(); + assert.equal(nextPort.invocation, third.invocation); + assert.equal(nextPort.input, 3); + f.send({ kind: "port-result", ...third, sequence: nextPort.sequence, status: "returned", output: 3 }); + assert.equal((await f.receive()).output, 3); + f.start(5); + const fresh = await f.receive(); + f.send({ + kind: "port-result", + invocation: fresh.invocation, + frame: fresh.frame, + sequence: fresh.sequence, + status: "returned", + output: 5, + }); + assert.equal((await f.receive()).output, 5); + } finally { + await f.close(); + } +}); + +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 }, + ); + try { + const first = f.start(); + await f.receive(); + const queued = f.start(); + const replies = [await f.receive(), await f.receive()]; + assert.deepEqual(new Set(replies.map((x) => x.invocation)), new Set([first.invocation, queued.invocation])); + assert.ok(replies.every((x) => x.error.code === "RESTRICTED_DEADLINE")); + const fresh = f.start(); + const port = await f.receive(); + assert.equal(port.invocation, fresh.invocation); + f.send({ kind: "port-result", ...fresh, sequence: port.sequence, status: "returned", output: "fresh" }); + assert.equal((await f.receive()).output, "fresh"); + } finally { + await f.close(); + } +}); diff --git a/restricted-execution/worker.mjs b/restricted-execution/worker.mjs new file mode 100644 index 0000000..a2e21b6 --- /dev/null +++ b/restricted-execution/worker.mjs @@ -0,0 +1,94 @@ +// Platform entrypoint, evaluated before loading the authored registry. A fresh +// engine process owns exactly one frame. BROKER grants only that frame's ports. +const deny = () => { + throw Error("RESTRICTED_AMBIENT_ACCESS"); +}; +const replace = (object, key, value) => + Object.defineProperty(object, key, { value, configurable: false, writable: false }); +const NativeDate = Date; +const FixedDate = new Proxy(NativeDate, { + construct(target, args) { + if (!args.length) deny(); + return Reflect.construct(target, args, target); + }, + apply() { + deny(); + }, +}); +replace(NativeDate, "now", deny); +replace(NativeDate.prototype, "constructor", FixedDate); +replace(globalThis, "Date", FixedDate); +replace(Math, "random", deny); +// Identity/value codecs need deterministic SHA-256, never entropy or key creation. +const digest = crypto.subtle.digest.bind(crypto.subtle); +replace(globalThis, "crypto", Object.freeze({ subtle: Object.freeze({ digest }) })); +for (const name of ["performance", "Temporal", "scheduler", "WebSocket", "EventSource", "SharedArrayBuffer", "Atomics"]) + replace(globalThis, name, undefined); +for (const name of ["setTimeout", "setInterval", "fetch"]) replace(globalThis, name, deny); +// Intl's no-argument format defaults to the clock without consulting Date.now. +const format = Object.getOwnPropertyDescriptor(Intl.DateTimeFormat.prototype, "format").get; +Object.defineProperty(Intl.DateTimeFormat.prototype, "format", { + configurable: false, + get() { + const bound = format.call(this); + return (value) => { + if (value === undefined) deny(); + return bound(value); + }; + }, +}); +const formatToParts = Intl.DateTimeFormat.prototype.formatToParts; +replace(Intl.DateTimeFormat.prototype, "formatToParts", function (value) { + if (value === undefined) deny(); + return formatToParts.call(this, value); +}); +let used = false; +export default { + async fetch(request, env) { + if (used) return new Response("FRAME_EXPIRED", { status: 409 }); + used = true; + const input = await request.json(); + const source = await import("./package.mjs"); + const registry = source[env.CONFIG.registryExport]; + if ( + !registry || + typeof registry.invoke !== "function" || + JSON.stringify(registry.exports) !== JSON.stringify(env.CONFIG.exports) + ) + throw Error("RESTRICTED_REGISTRY_MISMATCH"); + let open = true; + const channel = async (value) => { + if (!open) throw Error("EXPIRED_CONTEXT"); + const response = await env.BROKER.fetch("http://frame/port", { + method: "POST", + body: JSON.stringify(value), + }); + const result = await response.json(); + if (!open) throw Error("EXPIRED_CONTEXT"); + if (result.status !== "returned") + throw Object.assign(Error(result.error?.message ?? "Port failed"), { + code: result.error?.code ?? "PORT_FAILED", + }); + return result.output; + }; + try { + const output = await registry.invoke( + { exportId: input.exportId, workspace: input.workspace, receiver: input.receiver, input: input.input }, + channel, + ); + const body = JSON.stringify({ status: "returned", output }); + if (new TextEncoder().encode(body).length > env.CONFIG.maxBytes) throw Error("RESTRICTED_OUTPUT_LIMIT"); + return new Response(body, { headers: { "content-type": "application/json" } }); + } catch (error) { + // Only authored/codec rejection codes cross the engine boundary. Engine + // diagnostics stay private to the supervisor, never become protocol data. + const code = + typeof error?.code === "string" && /^[A-Z0-9_]{1,128}$/.test(error.code) + ? error.code + : "RESTRICTED_PACKAGE_FAILED"; + return Response.json({ status: "failed", error: { code, message: code } }); + } finally { + open = false; + } + }, +}; diff --git a/restricted-execution/workerd.nix b/restricted-execution/workerd.nix new file mode 100644 index 0000000..087b9ef --- /dev/null +++ b/restricted-execution/workerd.nix @@ -0,0 +1,45 @@ +# Pinned upstream engine. This binary supplies an import/capability boundary; +# production execution additionally requires a workspace VM and OS limits. +{ pkgs }: +let + version = "1.20261005.1"; + platforms = { + x86_64-linux = { + name = "linux-64"; + hash = "sha256-zm0T3YLo49ORPwR4iwP+WkQBMkiZQYftc+ArBQEoNGY="; + }; + aarch64-linux = { + name = "linux-arm64"; + hash = "sha256-jkvHyexr8Sn2wyqmLJJEOOGfAkfaVPLI7llHOX+FbxU="; + }; + }; + platform = + platforms.${pkgs.stdenv.hostPlatform.system} + or (throw "Restricted execution requires Linux x86_64 or aarch64"); +in +pkgs.stdenvNoCC.mkDerivation { + pname = "quixos-workerd"; + inherit version; + src = pkgs.fetchurl { + url = "https://registry.npmjs.org/@cloudflare/workerd-${platform.name}/-/workerd-${platform.name}-${version}.tgz"; + inherit (platform) hash; + }; + nativeBuildInputs = [ pkgs.autoPatchelfHook ]; + buildInputs = [ pkgs.stdenv.cc.cc.lib ]; + sourceRoot = "package"; + installPhase = '' + runHook preInstall + install -Dm755 bin/workerd "$out/bin/workerd" + runHook postInstall + ''; + doInstallCheck = true; + installCheckPhase = '' + "$out/bin/workerd" --version | grep -Fx 'workerd 2026-10-05' + ''; + meta = { + description = "Pinned workerd engine for VM-scoped Quixos restricted execution"; + homepage = "https://github.com/cloudflare/workerd"; + license = pkgs.lib.licenses.asl20; + platforms = builtins.attrNames platforms; + }; +}