Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eba5d732f2 | ||
|
|
b1397465ee | ||
|
|
b5046ad680 | ||
|
|
9a211ee5d7 | ||
|
|
dbd159be70 | ||
|
|
d6e66b537f | ||
|
|
929fdb4a28 | ||
|
|
f934347041 |
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"version": 1,
|
||||
"sourceRepo": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos",
|
||||
"sourceCommit": "039f055a7dcff2a29d59a8c37d7433b269b50f7a",
|
||||
"sourceCommit": "932fe7714761fb8b5db4bd6401357cdff762a428",
|
||||
"sourcePath": "quixos-instance/quixos-nix-helpers",
|
||||
"exportName": "quixos-nix-helpers",
|
||||
"mirrorRemote": "https://gitea-external.egads.tutti.syntaxblitz.net/quixos/quixos-nix-helpers.git"
|
||||
|
||||
+12
-7
@@ -113,8 +113,9 @@ service = {
|
||||
};
|
||||
```
|
||||
|
||||
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
|
||||
Each invocation gets a fresh Worker in a reusable engine process. An engine runs
|
||||
one active frame; the adapter has bounded concurrency and queueing. 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.
|
||||
@@ -127,11 +128,15 @@ parent environment credentials, and unlisted modules are denied. Explicit dates,
|
||||
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.
|
||||
The packaged runtime uses the platform's Bubblewrap/seccomp launcher. Only the
|
||||
engine's read-only runtime closure and private broker directory are mounted;
|
||||
native filesystem/process/network escape probes qualify that boundary. Systemd
|
||||
limits the whole package in a workspace VM. Shared authority additionally charges
|
||||
each engine to its own cgroup before launch. No per-isolate memory protection or
|
||||
Cloudflare hosted side-channel defenses are claimed. See the
|
||||
[runtime decision](../../docs/RESTRICTED_RUNTIME_DECISION.md#os-process-confinement-implementation)
|
||||
for the exact boundary and residual risks. A guest user can control its VM, so
|
||||
authoritative graph grants and credentials must remain outside it.
|
||||
|
||||
Qualification commands:
|
||||
|
||||
|
||||
+1
-12
@@ -96,18 +96,7 @@ 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;
|
||||
restrictedSDK = "${(import ./restricted-execution { inherit pkgs core; }).sdk}/core.mjs";
|
||||
}
|
||||
);
|
||||
compiled =
|
||||
|
||||
+41
-4
@@ -128,8 +128,7 @@ 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 restricted = !!config.restricted && target === "server";
|
||||
const coreImport = target === "browser" ? coreURL : restricted ? config.restricted.sdk : coreServer;
|
||||
const coreImport = target === "browser" ? coreURL : coreServer;
|
||||
const metadata = path.join(work, "bundle-meta.json");
|
||||
measureBuild("portable-bundle", { package: pkg.revision, target, mode }, () =>
|
||||
execFileSync(
|
||||
@@ -137,7 +136,7 @@ for (const { target, mode, settings } of builds) {
|
||||
[
|
||||
entry,
|
||||
"--bundle",
|
||||
`--platform=${target === "browser" || restricted ? "browser" : "node"}`,
|
||||
`--platform=${target === "browser" ? "browser" : "node"}`,
|
||||
"--format=esm",
|
||||
"--minify-whitespace",
|
||||
"--target=es2024",
|
||||
@@ -159,7 +158,7 @@ for (const { target, mode, settings } of builds) {
|
||||
(i) =>
|
||||
i.external &&
|
||||
i.path !== coreImport &&
|
||||
!(target === "server" && !restricted && builtinModules.includes(i.path.replace(/^node:/, ""))),
|
||||
!(target === "server" && builtinModules.includes(i.path.replace(/^node:/, ""))),
|
||||
),
|
||||
)
|
||||
)
|
||||
@@ -182,6 +181,44 @@ for (const { target, mode, settings } of builds) {
|
||||
files: [await recordFile(bundlePath, "text/javascript"), ...sharedFiles],
|
||||
};
|
||||
implementations.push({ ...body, contentDigest: compiler.implementationDigest(body) });
|
||||
if (target === "server" && mode !== "remote" && exports.length) {
|
||||
// A closed candidate is not an eligibility claim. The host checks these
|
||||
// bytes and its exact qualified SDK before admitting it to the sandbox.
|
||||
// Keep a normal service implementation for unsupported dependencies.
|
||||
const restrictedPath = "share/quixos/restricted/module.mjs";
|
||||
const sdkPath = "share/quixos/restricted/core.mjs";
|
||||
const sdkName = "quixos:restricted-sdk/v1";
|
||||
await fs.mkdir(path.dirname(path.join(output, restrictedPath)), { recursive: true });
|
||||
await fs.copyFile(config.restrictedSDK, path.join(output, sdkPath));
|
||||
measureBuild("restricted-bundle", { package: pkg.revision }, () =>
|
||||
execFileSync(
|
||||
config.esbuild,
|
||||
[
|
||||
entry,
|
||||
"--bundle",
|
||||
"--platform=node",
|
||||
"--format=esm",
|
||||
"--minify-whitespace",
|
||||
"--target=es2024",
|
||||
`--outfile=${path.join(output, restrictedPath)}`,
|
||||
`--alias:@quixos/camino-replica-core=${sdkName}`,
|
||||
`--external:${sdkName}`,
|
||||
],
|
||||
{ cwd: work, stdio: "inherit" },
|
||||
),
|
||||
);
|
||||
const restricted = {
|
||||
id: "restricted-module",
|
||||
target: "server",
|
||||
compatibility: "quixos-restricted-module-v1",
|
||||
path: restrictedPath,
|
||||
ports: "quixos-transaction-ports-v1",
|
||||
registryExport: settings.registryExport,
|
||||
exports,
|
||||
files: [await recordFile(restrictedPath, "text/javascript"), await recordFile(sdkPath, "text/javascript")],
|
||||
};
|
||||
implementations.push({ ...restricted, contentDigest: compiler.implementationDigest(restricted) });
|
||||
}
|
||||
}
|
||||
const components =
|
||||
config.group && !config.components
|
||||
|
||||
+4
-57
@@ -13,62 +13,6 @@ 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 });
|
||||
@@ -78,7 +22,10 @@ export async function buildService({
|
||||
throw Error("Custom service entry must be JavaScript; use a separate typed/language build before packaging");
|
||||
serviceEntry = path.join(config.source, relative(config.service.entry));
|
||||
} else {
|
||||
const modules = implementations.filter((implementation) => implementation.target === "server");
|
||||
const modules = implementations.filter(
|
||||
(implementation) =>
|
||||
implementation.target === "server" && implementation.compatibility === "quixos-js-module-v1",
|
||||
);
|
||||
if (pkg.exports.some((entry) => !modules.some((module) => module.exports.includes(entry.id))))
|
||||
throw Error("Generated service does not cover every package export; provide a custom service");
|
||||
if (!modules.length) throw Error("Generated service requires a checked registry");
|
||||
|
||||
@@ -1,6 +1,36 @@
|
||||
{ pkgs, core }:
|
||||
let
|
||||
workerd = import ./workerd.nix { inherit pkgs; };
|
||||
sandbox = import ./sandbox.nix {
|
||||
inherit pkgs;
|
||||
runtime = workerd;
|
||||
executable = "${workerd}/bin/workerd";
|
||||
};
|
||||
nativeProbe =
|
||||
pkgs.runCommand "quixos-restricted-native-probe"
|
||||
{
|
||||
nativeBuildInputs = [ pkgs.stdenv.cc ];
|
||||
}
|
||||
''
|
||||
mkdir -p "$out/bin"
|
||||
$CC -O2 -Wall -Wextra -Werror ${./sandbox-probe.c} -o "$out/bin/probe"
|
||||
'';
|
||||
nativeProbeSandbox = import ./sandbox.nix {
|
||||
inherit pkgs;
|
||||
runtime = nativeProbe;
|
||||
executable = "${nativeProbe}/bin/probe";
|
||||
};
|
||||
identityLauncher =
|
||||
pkgs.runCommand "quixos-restricted-identity-launcher"
|
||||
{
|
||||
nativeBuildInputs = [ pkgs.stdenv.cc ];
|
||||
}
|
||||
''
|
||||
mkdir -p "$out/bin"
|
||||
$CC -O2 -Wall -Wextra -Werror ${./identity-launcher.c} \
|
||||
-DSANDBOX_EXECUTABLE='"${sandbox}/bin/quixos-restricted-sandbox"' \
|
||||
-o "$out/bin/quixos-restricted-engine"
|
||||
'';
|
||||
# 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 ]; } ''
|
||||
@@ -23,6 +53,7 @@ let
|
||||
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_SANDBOX ${sandbox}/bin/quixos-restricted-sandbox \
|
||||
--set QUIXOS_RESTRICTED_WORKER "$out/lib/worker.mjs"
|
||||
'';
|
||||
qualification =
|
||||
@@ -40,6 +71,9 @@ in
|
||||
{
|
||||
inherit
|
||||
workerd
|
||||
sandbox
|
||||
identityLauncher
|
||||
nativeProbeSandbox
|
||||
sdk
|
||||
runtime
|
||||
qualification
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
#define _GNU_SOURCE
|
||||
#include <dirent.h>
|
||||
#include <fcntl.h>
|
||||
#include <grp.h>
|
||||
#include <pwd.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/prctl.h>
|
||||
#include <sys/resource.h>
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
|
||||
/* Installed only as a narrowly executable NixOS wrapper. This trusted helper
|
||||
* accepts no executable, environment, filesystem root or arbitrary UID. The
|
||||
* caller is platform authority, never a workspace guest or authored Worker. */
|
||||
static void fail(const char *message) {
|
||||
fprintf(stderr, "RESTRICTED_IDENTITY: %s\n", message);
|
||||
exit(125);
|
||||
}
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
const char *root = "/run/quixos-restricted-engines/";
|
||||
uid_t caller = getuid();
|
||||
gid_t group = getgid();
|
||||
struct passwd *owner = getpwnam("quixos-graph");
|
||||
if (geteuid() != 0) fail("privileged wrapper unavailable");
|
||||
if (!owner) fail("authority identity unavailable");
|
||||
if (caller != owner->pw_uid) fail("caller identity mismatch");
|
||||
if (argc != 3) fail("invalid arguments");
|
||||
char *end;
|
||||
unsigned long number = strtoul(argv[2], &end, 10);
|
||||
if (!*argv[2] || *end || number < 610000 || number >= 610064)
|
||||
fail("invalid engine identity");
|
||||
uid_t identity = (uid_t)number;
|
||||
if (strlen(argv[1]) != strlen(root) + 16 || strncmp(argv[1], root, strlen(root)))
|
||||
fail("invalid engine directory");
|
||||
const char *name = argv[1] + strlen(root);
|
||||
if (strncmp(name, "qx-engine-", 10) ||
|
||||
strlen(name) != 16 || strspn(name + 10, "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") != 6)
|
||||
fail("invalid engine directory");
|
||||
int parent = open("/run/quixos-restricted-engines", O_RDONLY | O_DIRECTORY | O_NOFOLLOW);
|
||||
struct stat info;
|
||||
if (parent < 0 || fstat(parent, &info) || info.st_uid != caller || (info.st_mode & 066))
|
||||
fail("invalid engine root");
|
||||
int directory = openat(parent, name, O_RDONLY | O_DIRECTORY | O_NOFOLLOW);
|
||||
if (directory < 0 || fstat(directory, &info) || info.st_uid != caller || (info.st_mode & 077))
|
||||
fail("invalid engine owner");
|
||||
DIR *entries = fdopendir(dup(directory));
|
||||
if (!entries) fail("engine entries");
|
||||
struct dirent *entry;
|
||||
unsigned int files = 0;
|
||||
while ((entry = readdir(entries))) {
|
||||
const char *file = entry->d_name;
|
||||
if (!strcmp(file, ".") || !strcmp(file, "..")) continue;
|
||||
int socket = !strcmp(file, "broker.sock");
|
||||
if ((!socket && strcmp(file, "config.capnp") && strcmp(file, "parent.mjs")) ||
|
||||
fstatat(directory, file, &info, AT_SYMLINK_NOFOLLOW) || info.st_uid != caller ||
|
||||
info.st_nlink != 1 || (socket ? !S_ISSOCK(info.st_mode) : !S_ISREG(info.st_mode)) ||
|
||||
info.st_size > 4 * 1024 * 1024)
|
||||
fail("unexpected engine file");
|
||||
files++;
|
||||
}
|
||||
closedir(entries);
|
||||
if (files != 3) fail("incomplete engine configuration");
|
||||
const char *files_to_transfer[] = {"config.capnp", "parent.mjs", "broker.sock"};
|
||||
for (size_t i = 0; i < 3; i++) {
|
||||
if (fchownat(directory, files_to_transfer[i], identity, group, AT_SYMLINK_NOFOLLOW) ||
|
||||
fchmodat(directory, files_to_transfer[i], 0660, 0))
|
||||
fail("engine file ownership");
|
||||
}
|
||||
if (fchown(directory, identity, group) || fchmod(directory, 0770))
|
||||
fail("engine directory ownership");
|
||||
close(directory);
|
||||
close(parent);
|
||||
struct rlimit core = {0, 0};
|
||||
if (setrlimit(RLIMIT_CORE, &core) || setgroups(0, NULL) ||
|
||||
setresgid(identity, identity, identity) || setresuid(identity, identity, identity) ||
|
||||
prctl(PR_SET_DUMPABLE, 0) || prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0))
|
||||
fail("identity confinement");
|
||||
/* The containing directory grants only its owner and authority access. A
|
||||
* socket created by workerd must also admit authority's different UID/GID. */
|
||||
umask(0000);
|
||||
char *args[] = {SANDBOX_EXECUTABLE, argv[1], NULL};
|
||||
clearenv();
|
||||
execv(SANDBOX_EXECUTABLE, args);
|
||||
fail("sandbox launch");
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
{
|
||||
executor,
|
||||
enable,
|
||||
controlGroup,
|
||||
}:
|
||||
{ config, lib, ... }:
|
||||
{
|
||||
config = lib.mkIf enable {
|
||||
assertions = [
|
||||
{
|
||||
assertion = lib.all (user: user.uid == null || user.uid < 610000 || user.uid >= 610064) (
|
||||
builtins.attrValues config.users.users
|
||||
);
|
||||
message = "UIDs 610000–610063 are reserved for restricted engine identities.";
|
||||
}
|
||||
];
|
||||
systemd.tmpfiles.rules = [
|
||||
"d /run/quixos-restricted-engines 0711 quixos-graph ${controlGroup} -"
|
||||
];
|
||||
security.wrappers.quixos-restricted-engine = {
|
||||
source = "${executor.identityLauncher}/bin/quixos-restricted-engine";
|
||||
owner = "root";
|
||||
group = controlGroup;
|
||||
permissions = "u+rx,g+rx,o-rwx";
|
||||
setuid = true;
|
||||
};
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
#define _GNU_SOURCE
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <sched.h>
|
||||
#include <stddef.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/ptrace.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/stat.h>
|
||||
#include <sys/un.h>
|
||||
#include <sys/wait.h>
|
||||
#include <unistd.h>
|
||||
|
||||
static void check(int condition, const char *label) {
|
||||
if (!condition) {
|
||||
fprintf(stderr, "ESCAPE: %s (%s)\n", label, strerror(errno));
|
||||
exit(1);
|
||||
}
|
||||
}
|
||||
int main(void) {
|
||||
for (int fd = 3; fd < 256; fd++)
|
||||
check(fcntl(fd, F_GETFD) < 0 && errno == EBADF, "inherited descriptor");
|
||||
check(!getenv("QUIXOS_VM_SECRET"), "parent environment");
|
||||
check(open("/etc/test-secret", O_RDONLY) < 0, "authority filesystem");
|
||||
check(open("/sys/fs/cgroup/cgroup.procs", O_WRONLY) < 0, "resource controls");
|
||||
check(open("/proc/1/root/etc/test-secret", O_RDONLY) < 0, "proc root escape");
|
||||
check(socket(AF_INET, SOCK_STREAM, 0) < 0 && errno == EPERM, "IPv4 socket");
|
||||
check(socket(AF_INET6, SOCK_STREAM, 0) < 0 && errno == EPERM, "IPv6 socket");
|
||||
check(socket(AF_NETLINK, SOCK_RAW, 0) < 0 && errno == EPERM, "netlink socket");
|
||||
check(ptrace(PTRACE_TRACEME, 0, NULL, NULL) < 0 && errno == EPERM, "ptrace");
|
||||
check(unshare(CLONE_NEWUSER) < 0, "nested user namespace");
|
||||
pid_t child = fork();
|
||||
if (child == 0) _exit(2);
|
||||
check(child < 0 && errno == EPERM, "subprocess");
|
||||
check(access("/bin/sh", X_OK) < 0, "ambient executable");
|
||||
check(getppid() <= 1, "host process visibility");
|
||||
struct sockaddr_un address = {.sun_family = AF_UNIX};
|
||||
int outside = socket(AF_UNIX, SOCK_STREAM, 0);
|
||||
check(outside >= 0, "Unix socket support");
|
||||
strcpy(address.sun_path, "/run/quixos-native-authority/test.sock");
|
||||
check(connect(outside, (struct sockaddr *)&address, sizeof(address)) < 0,
|
||||
"authority Unix socket");
|
||||
close(outside);
|
||||
outside = socket(AF_UNIX, SOCK_STREAM, 0);
|
||||
memset(address.sun_path, 0, sizeof(address.sun_path));
|
||||
strcpy(address.sun_path + 1, "quixos-native-authority");
|
||||
check(connect(outside, (struct sockaddr *)&address,
|
||||
offsetof(struct sockaddr_un, sun_path) + 1 + strlen(address.sun_path + 1)) < 0,
|
||||
"authority abstract socket");
|
||||
close(outside);
|
||||
char directory[4096];
|
||||
check(getcwd(directory, sizeof(directory)) != NULL, "engine directory");
|
||||
check(snprintf(address.sun_path, sizeof(address.sun_path), "%s/broker.sock", directory) <
|
||||
(int)sizeof(address.sun_path), "broker path length");
|
||||
int broker = socket(AF_UNIX, SOCK_STREAM, 0);
|
||||
check(broker >= 0 && connect(broker, (struct sockaddr *)&address, sizeof(address)) == 0,
|
||||
"authorized broker");
|
||||
check(write(broker, "sandbox-probe", 13) == 13, "broker write");
|
||||
close(broker);
|
||||
puts("RESTRICTED_NATIVE_SANDBOX_PASSED");
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
#define _GNU_SOURCE
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
#include <unistd.h>
|
||||
|
||||
/* Trusted, single-process admission gate. The supervisor charges this PID to
|
||||
* its engine cgroup before releasing stdin. All subsequently created namespace
|
||||
* helpers and threads inherit that charge. No authored code runs here. */
|
||||
static void fail(const char *message) {
|
||||
fprintf(stderr, "RESTRICTED_SANDBOX: %s\n", message);
|
||||
exit(125);
|
||||
}
|
||||
|
||||
int main(int argc, char **argv) {
|
||||
struct stat info;
|
||||
if (argc != 2 || argv[1][0] != '/' || strstr(argv[1], "/../") ||
|
||||
lstat(argv[1], &info) || !S_ISDIR(info.st_mode) ||
|
||||
info.st_uid != getuid() || (info.st_mode & 007))
|
||||
fail("invalid engine directory");
|
||||
char gate;
|
||||
if (read(STDIN_FILENO, &gate, 1) != 1 || gate != 1)
|
||||
fail("engine admission was not released");
|
||||
int input = open("/dev/null", O_RDONLY);
|
||||
if (input < 0 || dup2(input, STDIN_FILENO) < 0)
|
||||
fail("stdin");
|
||||
close(input);
|
||||
int filter = open(SANDBOX_FILTER, O_RDONLY);
|
||||
if (filter < 0)
|
||||
fail("seccomp filter");
|
||||
char filter_fd[32];
|
||||
snprintf(filter_fd, sizeof(filter_fd), "%d", filter);
|
||||
char *args[4096];
|
||||
size_t count = 0;
|
||||
#define ARG(value) args[count++] = (value)
|
||||
ARG(SANDBOX_BWRAP);
|
||||
ARG("--unshare-all");
|
||||
ARG("--unshare-user");
|
||||
ARG("--die-with-parent");
|
||||
ARG("--disable-userns");
|
||||
ARG("--assert-userns-disabled");
|
||||
ARG("--cap-drop"); ARG("ALL");
|
||||
ARG("--hostname"); ARG("quixos-restricted");
|
||||
ARG("--clearenv");
|
||||
ARG("--setenv"); ARG("TZ"); ARG("UTC");
|
||||
ARG("--setenv"); ARG("LANG"); ARG("C.UTF-8");
|
||||
FILE *closure = fopen(SANDBOX_CLOSURE, "r");
|
||||
if (!closure)
|
||||
fail("runtime closure");
|
||||
char *line = NULL;
|
||||
size_t capacity = 0;
|
||||
while (getline(&line, &capacity, closure) >= 0) {
|
||||
line[strcspn(line, "\n")] = 0;
|
||||
if (strncmp(line, "/nix/store/", 11) || strchr(line, ' ') || count > 4000)
|
||||
fail("runtime closure entry");
|
||||
char *path = strdup(line);
|
||||
if (!path)
|
||||
fail("runtime closure allocation");
|
||||
ARG("--ro-bind"); ARG(path); ARG(path);
|
||||
}
|
||||
fclose(closure);
|
||||
free(line);
|
||||
ARG("--proc"); ARG("/proc");
|
||||
ARG("--dev"); ARG("/dev");
|
||||
ARG("--tmpfs"); ARG("/tmp");
|
||||
ARG("--bind"); ARG(argv[1]); ARG(argv[1]);
|
||||
ARG("--chdir"); ARG(argv[1]);
|
||||
ARG("--seccomp"); ARG(filter_fd);
|
||||
ARG("--"); ARG(SANDBOX_EXECUTABLE);
|
||||
ARG("serve"); ARG("--experimental"); ARG("config.capnp");
|
||||
ARG(NULL);
|
||||
execv(SANDBOX_BWRAP, args);
|
||||
fail("bubblewrap failed");
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
{
|
||||
pkgs,
|
||||
runtime,
|
||||
executable,
|
||||
}:
|
||||
let
|
||||
closure = pkgs.closureInfo { rootPaths = [ runtime ]; };
|
||||
filter =
|
||||
pkgs.runCommand "quixos-restricted-seccomp"
|
||||
{
|
||||
nativeBuildInputs = [
|
||||
pkgs.stdenv.cc
|
||||
pkgs.pkg-config
|
||||
];
|
||||
buildInputs = [ pkgs.libseccomp ];
|
||||
}
|
||||
''
|
||||
$CC -O2 -Wall -Wextra -Werror ${./seccomp.c} $(pkg-config --cflags --libs libseccomp) -o filter
|
||||
./filter > "$out"
|
||||
'';
|
||||
in
|
||||
pkgs.runCommand "quixos-restricted-sandbox" { nativeBuildInputs = [ pkgs.stdenv.cc ]; } ''
|
||||
mkdir -p "$out/bin"
|
||||
$CC -O2 -Wall -Wextra -Werror ${./sandbox.c} \
|
||||
-DSANDBOX_BWRAP='"${pkgs.bubblewrap}/bin/bwrap"' \
|
||||
-DSANDBOX_EXECUTABLE='"${executable}"' \
|
||||
-DSANDBOX_FILTER='"${filter}"' \
|
||||
-DSANDBOX_CLOSURE='"${closure}/store-paths"' \
|
||||
-o "$out/bin/quixos-restricted-sandbox"
|
||||
''
|
||||
@@ -0,0 +1,54 @@
|
||||
#define _GNU_SOURCE
|
||||
#include <errno.h>
|
||||
#include <linux/sched.h>
|
||||
#include <seccomp.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <sys/socket.h>
|
||||
#include <unistd.h>
|
||||
|
||||
/* Layered on an empty mount tree, fresh PID/user/network namespaces, no caps,
|
||||
* no inherited descriptors and an immutable runtime closure. Deny kernel and
|
||||
* process escape surfaces; retain V8's ordinary memory/thread syscalls. */
|
||||
static void checked(int result) {
|
||||
if (result < 0) {
|
||||
fprintf(stderr, "seccomp rule: %d\n", result);
|
||||
exit(1);
|
||||
}
|
||||
}
|
||||
int main(void) {
|
||||
scmp_filter_ctx filter = seccomp_init(SCMP_ACT_ALLOW);
|
||||
if (!filter) return 1;
|
||||
const char *denied[] = {
|
||||
"fork", "vfork", "execveat", "ptrace", "process_vm_readv", "process_vm_writev",
|
||||
"pidfd_getfd", "kcmp", "bpf", "perf_event_open", "userfaultfd", "io_uring_setup",
|
||||
"mount", "umount2", "pivot_root", "chroot", "move_mount", "open_tree", "fsopen",
|
||||
"fsconfig", "fsmount", "fspick", "mount_setattr", "open_by_handle_at",
|
||||
"name_to_handle_at", "setns", "unshare", "keyctl", "add_key", "request_key",
|
||||
"reboot", "kexec_load", "kexec_file_load", "init_module", "finit_module",
|
||||
"delete_module", "swapon", "swapoff", "syslog", "acct", "sethostname",
|
||||
"setdomainname", "iopl", "ioperm", "lookup_dcookie", NULL
|
||||
};
|
||||
for (size_t i = 0; denied[i]; i++) {
|
||||
int syscall = seccomp_syscall_resolve_name(denied[i]);
|
||||
if (syscall != __NR_SCMP_ERROR)
|
||||
checked(seccomp_rule_add(filter, SCMP_ACT_ERRNO(EPERM), syscall, 0));
|
||||
}
|
||||
/* glibc falls back to clone when clone3 is unavailable. Only thread creation
|
||||
* is admitted; subprocesses and new namespaces cannot be manufactured. */
|
||||
checked(seccomp_rule_add(filter, SCMP_ACT_ERRNO(ENOSYS), SCMP_SYS(clone3), 0));
|
||||
checked(seccomp_rule_add(filter, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(clone), 1,
|
||||
SCMP_A0(SCMP_CMP_MASKED_EQ, CLONE_THREAD, 0)));
|
||||
const unsigned long namespaces[] = {CLONE_NEWUSER, CLONE_NEWNS, CLONE_NEWNET,
|
||||
CLONE_NEWPID, CLONE_NEWIPC, CLONE_NEWUTS, CLONE_NEWCGROUP};
|
||||
for (size_t i = 0; i < sizeof(namespaces) / sizeof(namespaces[0]); i++)
|
||||
checked(seccomp_rule_add(filter, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(clone), 1,
|
||||
SCMP_A0(SCMP_CMP_MASKED_EQ, namespaces[i], namespaces[i])));
|
||||
checked(seccomp_rule_add(filter, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(socket), 1,
|
||||
SCMP_A0(SCMP_CMP_NE, AF_UNIX)));
|
||||
checked(seccomp_rule_add(filter, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(ioctl), 1,
|
||||
SCMP_A1(SCMP_CMP_EQ, 0x5412))); /* TIOCSTI */
|
||||
checked(seccomp_export_bpf(filter, STDOUT_FILENO));
|
||||
seccomp_release(filter);
|
||||
return 0;
|
||||
}
|
||||
+445
-146
@@ -1,6 +1,8 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { randomUUID } from "node:crypto";
|
||||
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";
|
||||
|
||||
@@ -13,15 +15,76 @@ const absolute = (value) =>
|
||||
!/[\s\u0000-\u001f]/.test(value) &&
|
||||
!value.split("/").includes("..");
|
||||
|
||||
/** Host-wide engine reservations. Only trusted adapters hold these callbacks;
|
||||
* workers cannot alter admission or evict another invocation's engine. */
|
||||
export class RestrictedEngineBudget {
|
||||
#leases = new Set();
|
||||
constructor(limit = 8, reserve = async () => ({})) {
|
||||
if (!Number.isSafeInteger(limit) || limit < 1 || limit > 64) throw fixedError("INVALID_ENGINE_BUDGET");
|
||||
this.limit = limit;
|
||||
this.reserve = reserve;
|
||||
}
|
||||
async acquire(idle, evict) {
|
||||
while (this.#leases.size >= this.limit) {
|
||||
const victim = [...this.#leases].find((entry) => entry.idle());
|
||||
if (!victim) throw fixedError("RESTRICTED_ENGINE_LIMIT");
|
||||
await victim.evict();
|
||||
}
|
||||
const lease = { idle, evict };
|
||||
this.#leases.add(lease);
|
||||
try {
|
||||
const resource = await this.reserve();
|
||||
let releasing;
|
||||
const release = () =>
|
||||
(releasing ??= (async () => {
|
||||
await resource.release?.();
|
||||
this.#leases.delete(lease);
|
||||
})());
|
||||
release.attach = resource.attach;
|
||||
release.usage = resource.usage;
|
||||
release.kill = resource.kill;
|
||||
release.uid = resource.uid;
|
||||
return release;
|
||||
} catch (error) {
|
||||
this.#leases.delete(lease);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
stats() {
|
||||
return { engines: this.#leases.size, limit: this.limit };
|
||||
}
|
||||
}
|
||||
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.
|
||||
* Per-frame process replacement prevents retained globals/handles crossing calls. */
|
||||
* Each engine runs one fresh Worker per invocation. Host tokens expire before
|
||||
* reuse; cancellation and watchdogs destroy the affected engine, not neighbors. */
|
||||
export async function serveRestrictedRegistry(config, io = process) {
|
||||
if (
|
||||
!object(config) ||
|
||||
!absolute(config.workerd) ||
|
||||
!absolute(config.worker) ||
|
||||
!absolute(config.module) ||
|
||||
(config.sandbox !== undefined && !absolute(config.sandbox)) ||
|
||||
(config.sandboxIdentity !== undefined && typeof config.sandboxIdentity !== "boolean") ||
|
||||
!/^[$A-Z_a-z][$\w]*$/.test(config.registryExport) ||
|
||||
!Array.isArray(config.exports) ||
|
||||
!config.exports.length ||
|
||||
@@ -47,6 +110,9 @@ 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,
|
||||
engineIdleMs: config.engineIdleMs ?? 30000,
|
||||
};
|
||||
if (
|
||||
!Number.isSafeInteger(limits.maxBytes) ||
|
||||
@@ -60,18 +126,31 @@ 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 ||
|
||||
!Number.isSafeInteger(limits.engineIdleMs) ||
|
||||
limits.engineIdleMs < 1 ||
|
||||
limits.engineIdleMs > 60000
|
||||
)
|
||||
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 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 engineBudget = config.engineBudget ?? new RestrictedEngineBudget(limits.maxActive + limits.maxNested);
|
||||
const active = new Map(),
|
||||
engines = new Set(),
|
||||
idleEngines = [];
|
||||
const metrics = { starts: 0, reused: 0, completed: 0, peakRSSBytes: 0 };
|
||||
let buffered = Buffer.alloc(0),
|
||||
stopped = false,
|
||||
failure;
|
||||
const queued = [];
|
||||
const send = (value) => {
|
||||
if (stopped) return;
|
||||
@@ -83,11 +162,61 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
throw fixedError("RESTRICTED_BACKPRESSURE");
|
||||
io.stdout.write(line);
|
||||
};
|
||||
// A sandbox has namespace helpers as well as workerd. Charge the complete
|
||||
// delegated cgroup when available; the qualification fallback follows the
|
||||
// bounded process tree, never trusting a worker's self-reported counters.
|
||||
const engineUsage = async (engine) => {
|
||||
if (engine.release?.usage) return engine.release.usage();
|
||||
engine.processTicks ??= new Map();
|
||||
const pending = [engine.child.pid];
|
||||
let memoryBytes = 0,
|
||||
visited = 0;
|
||||
while (pending.length) {
|
||||
if (++visited > 64) throw fixedError("RESTRICTED_PROCESS_LIMIT");
|
||||
const pid = pending.pop();
|
||||
try {
|
||||
const stat = await readFile(`/proc/${pid}/stat`, "utf8");
|
||||
const fields = stat.slice(stat.lastIndexOf(")") + 2).split(" ");
|
||||
engine.processTicks.set(`${pid}:${fields[19]}`, Number(fields[11]) + Number(fields[12]));
|
||||
const status = await readFile(`/proc/${pid}/status`, "utf8");
|
||||
memoryBytes += Number(status.match(/^VmRSS:\s+(\d+)/m)?.[1] ?? 0) * 1024;
|
||||
const children = await readFile(`/proc/${pid}/task/${pid}/children`, "utf8");
|
||||
pending.push(...children.trim().split(/\s+/).filter(Boolean).map(Number));
|
||||
} catch (error) {
|
||||
if (!["ENOENT", "ESRCH"].includes(error.code)) throw error;
|
||||
}
|
||||
}
|
||||
return {
|
||||
cpuMicroseconds: [...engine.processTicks.values()].reduce((sum, ticks) => sum + ticks * 10000, 0),
|
||||
memoryBytes,
|
||||
};
|
||||
};
|
||||
const sampleCPU = async (call) => {
|
||||
if (!call.child?.pid) return;
|
||||
const { cpuMicroseconds: total } = await engineUsage(call.engine);
|
||||
const cpuMicroseconds = total - (call.cpuBaseline ?? 0);
|
||||
if (call.engine) call.engine.cpuTotal = total;
|
||||
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,29 +224,46 @@ 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(() => {});
|
||||
if (call.child?.pid) {
|
||||
try {
|
||||
process.kill(-call.child.pid, "SIGKILL");
|
||||
} catch (error) {
|
||||
if (error.code !== "ESRCH") throw error;
|
||||
}
|
||||
await call.exited;
|
||||
clearTimeout(call.cpuTimer);
|
||||
await call.cpuRead?.catch(() => {});
|
||||
let usageFailure;
|
||||
try {
|
||||
await sampleCPU(call);
|
||||
} catch (error) {
|
||||
usageFailure = error;
|
||||
}
|
||||
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;
|
||||
const engine = call.engine;
|
||||
if (engine) {
|
||||
// Revoke the frame token before returning an engine to the idle pool.
|
||||
// A delayed child request can never acquire the next call's authority.
|
||||
if (engine.current === call) engine.current = null;
|
||||
if (!stopped && reason.code === "INVOCATION_ENDED" && !usageFailure && !engine.destroying) {
|
||||
const rss = (await engineUsage(engine)).memoryBytes;
|
||||
metrics.peakRSSBytes = Math.max(metrics.peakRSSBytes, rss);
|
||||
// Bound retained engine heaps as well as lifetime. Hard aggregate
|
||||
// physical memory remains the supervising process/cgroup's contract.
|
||||
if (rss <= 128 * 1024 * 1024) {
|
||||
idleEngines.push(engine);
|
||||
engine.idleTimer = setTimeout(() => void destroyEngine(engine).catch(fail), limits.engineIdleMs);
|
||||
} else await destroyEngine(engine);
|
||||
} else await destroyEngine(engine);
|
||||
}
|
||||
active.delete(call.message.invocation);
|
||||
if (usageFailure) throw usageFailure;
|
||||
})();
|
||||
return call.retiring;
|
||||
};
|
||||
const fail = async (error) => {
|
||||
if (stopped) return;
|
||||
const fail = (error) => {
|
||||
if (failure) return failure;
|
||||
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;
|
||||
return (failure = (async () => {
|
||||
io.stdin.destroy();
|
||||
for (const call of [...queued]) await retire(call, error);
|
||||
await Promise.all([...active.values()].map((call) => retire(call, error)));
|
||||
await Promise.all([...engines].map(destroyEngine));
|
||||
io.stdout.end();
|
||||
io.exitCode = 1;
|
||||
})());
|
||||
};
|
||||
const body = async (stream) => {
|
||||
const chunks = [];
|
||||
@@ -129,17 +275,201 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
}
|
||||
return JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(Buffer.concat(chunks)));
|
||||
};
|
||||
const destroyEngine = async (engine) => {
|
||||
if (engine.destroying) return engine.destroying;
|
||||
engine.destroying = (async () => {
|
||||
clearTimeout(engine.idleTimer);
|
||||
const index = idleEngines.indexOf(engine);
|
||||
if (index >= 0) idleEngines.splice(index, 1);
|
||||
engine.current = null;
|
||||
if (engine.child?.pid) {
|
||||
if (engine.release.kill) await engine.release.kill();
|
||||
else {
|
||||
try {
|
||||
process.kill(-engine.child.pid, "SIGKILL");
|
||||
} catch (error) {
|
||||
if (error.code !== "ESRCH") throw error;
|
||||
}
|
||||
}
|
||||
await engine.exited;
|
||||
}
|
||||
engine.server?.closeAllConnections();
|
||||
if (engine.server) await new Promise((resolve) => engine.server.close(resolve));
|
||||
try {
|
||||
if (engine.directory) await rm(engine.directory, { recursive: true, force: true });
|
||||
} finally {
|
||||
engines.delete(engine);
|
||||
await engine.release?.();
|
||||
}
|
||||
})();
|
||||
return engine.destroying;
|
||||
};
|
||||
const prepareEngine = async (call) => {
|
||||
const cached = idleEngines.pop();
|
||||
if (cached) {
|
||||
clearTimeout(cached.idleTimer);
|
||||
cached.current = call;
|
||||
call.engine = cached;
|
||||
call.child = cached.child;
|
||||
call.cpuBaseline = cached.cpuTotal ?? 0;
|
||||
metrics.reused++;
|
||||
watchCPU(call);
|
||||
return;
|
||||
}
|
||||
const engine = { current: call, cpuTotal: 0 };
|
||||
engines.add(engine);
|
||||
call.engine = engine;
|
||||
engine.release = await engineBudget.acquire(
|
||||
() => !engine.current && !engine.destroying,
|
||||
() => destroyEngine(engine),
|
||||
);
|
||||
if (!call.open) return;
|
||||
engine.directory = await mkdtemp(join(tmpdir(), "qx-engine-"));
|
||||
if (!call.open) return;
|
||||
const broker = join(engine.directory, "broker.sock");
|
||||
engine.socket = join(engine.directory, "engine.sock");
|
||||
engine.server = createServer(async (request, response) => {
|
||||
const call = engine.current;
|
||||
const message = call?.message;
|
||||
try {
|
||||
if (!call?.open || request.method !== "POST" || request.url !== "/port/" + call.token)
|
||||
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 > 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 }));
|
||||
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) => {
|
||||
engine.server.once("error", reject);
|
||||
engine.server.listen(broker, resolve);
|
||||
});
|
||||
const modules = { "entry.mjs": { js: workerSource }, "package.mjs": { js: packageSource } };
|
||||
for (const module of sdk) modules[module.name] = { js: module.source };
|
||||
const setting = { registryExport: config.registryExport, exports: config.exports, maxBytes: limits.maxBytes };
|
||||
// The parent contains immutable checked source only. It never imports the
|
||||
// authored module or keeps a child stub, request, result or frame handle.
|
||||
const parent = `const modules=${JSON.stringify(modules)}, setting=${JSON.stringify(setting)};
|
||||
export default {async fetch(request,env) {
|
||||
const input=await request.json();
|
||||
const child=env.LOADER.load({compatibilityDate:"2026-10-05",mainModule:"entry.mjs",modules,
|
||||
globalOutbound:null,env:{BROKER:env.BROKER,CONFIG:{...setting,frameToken:input.frameToken}}});
|
||||
const response=await child.getEntrypoint().fetch(new Request("http://invocation/",{
|
||||
method:"POST",body:JSON.stringify(input.message)}));
|
||||
const body=await response.arrayBuffer();
|
||||
return new Response(body,{status:response.status,headers:response.headers});
|
||||
}};`;
|
||||
await writeFile(join(engine.directory, "parent.mjs"), parent);
|
||||
const configuration = `using Workerd = import "/workerd/workerd.capnp"; const config :Workerd.Config = (
|
||||
services=[(name="main",worker=(modules=[(name="parent.mjs",esModule=embed "parent.mjs")],
|
||||
compatibilityDate="2026-10-05",compatibilityFlags=["experimental"],globalOutbound="deny",
|
||||
bindings=[(name="LOADER",workerLoader=()),(name="BROKER",service="broker")])),
|
||||
(name="broker",external=(address=${JSON.stringify("unix:" + broker)},http=())),
|
||||
(name="deny",network=(allow=[]))],
|
||||
sockets=[(name="frame",address=${JSON.stringify("unix:" + engine.socket)},http=(),service="main")]);`;
|
||||
const file = join(engine.directory, "config.capnp");
|
||||
await writeFile(file, configuration);
|
||||
if (!call.open) return;
|
||||
if (config.sandboxIdentity && !Number.isSafeInteger(engine.release.uid))
|
||||
throw fixedError("RESTRICTED_IDENTITY_UNAVAILABLE");
|
||||
engine.child = spawn(
|
||||
config.sandbox ?? config.workerd,
|
||||
config.sandbox
|
||||
? [engine.directory, ...(config.sandboxIdentity ? [String(engine.release.uid)] : [])]
|
||||
: ["serve", "--experimental", file],
|
||||
{
|
||||
detached: true,
|
||||
env: { TZ: "UTC", LANG: "C.UTF-8" },
|
||||
cwd: engine.directory,
|
||||
stdio: [config.sandbox ? "pipe" : "ignore", "ignore", "pipe"],
|
||||
},
|
||||
);
|
||||
metrics.starts++;
|
||||
engine.exited = new Promise((resolve) => {
|
||||
engine.child.once("exit", resolve);
|
||||
engine.child.once("error", resolve);
|
||||
});
|
||||
engine.child.stdin?.on("error", () => engine.current?.terminate("RESTRICTED_START_FAILED"));
|
||||
call.child = engine.child;
|
||||
let diagnostics = 0;
|
||||
engine.child.stderr.on("data", (chunk) => {
|
||||
if ((diagnostics += chunk.length) > 65536) {
|
||||
if (engine.current?.open) engine.current.terminate("RESTRICTED_DIAGNOSTIC_LIMIT");
|
||||
else void destroyEngine(engine).catch(fail);
|
||||
} else config.diagnostics?.(chunk);
|
||||
});
|
||||
await engine.release.attach?.(engine.child.pid);
|
||||
if (config.sandbox && call.open) engine.child.stdin.end(Buffer.from([1]));
|
||||
watchCPU(call);
|
||||
engine.child.once("error", () => engine.current?.terminate("RESTRICTED_START_FAILED"));
|
||||
void engine.exited.then(() => {
|
||||
if (engine.current?.open) engine.current.terminate("RESTRICTED_ENGINE_EXITED");
|
||||
else void destroyEngine(engine).catch(fail);
|
||||
});
|
||||
};
|
||||
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);
|
||||
metrics.lastInvocation = {
|
||||
queueMs: (call.startedAt ?? performance.now()) - call.receivedAt,
|
||||
preparationMs: call.preparationMs ?? 0,
|
||||
roundtripMs: call.roundtripMs ?? 0,
|
||||
initializationMs: call.initializationMs ?? null,
|
||||
executionMs: call.executionMs ?? null,
|
||||
totalMs: performance.now() - call.receivedAt,
|
||||
};
|
||||
send(call.cancellationReply ?? reply);
|
||||
pump();
|
||||
})());
|
||||
};
|
||||
const invoke = async (call) => {
|
||||
const { message } = call;
|
||||
active = call;
|
||||
call.startedAt = performance.now();
|
||||
active.set(message.invocation, call);
|
||||
const terminate = (code) => {
|
||||
if (!call.open) return;
|
||||
void finish(call, fixedError(code), {
|
||||
@@ -150,119 +480,46 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
};
|
||||
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");
|
||||
});
|
||||
})();
|
||||
call.token = randomUUID();
|
||||
call.preparing = prepareEngine(call);
|
||||
await call.preparing;
|
||||
call.preparationMs = performance.now() - call.startedAt;
|
||||
if (!call.open) return;
|
||||
const roundtripAt = performance.now();
|
||||
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);
|
||||
}
|
||||
});
|
||||
const request = httpRequest(
|
||||
{ socketPath: call.engine.socket, path: "/", method: "POST", agent: false },
|
||||
async (response) => {
|
||||
try {
|
||||
const result = await body(response);
|
||||
for (const [header, key] of [
|
||||
["x-quixos-initialization-ms", "initializationMs"],
|
||||
["x-quixos-execution-ms", "executionMs"],
|
||||
]) {
|
||||
const value = Number(response.headers[header]);
|
||||
if (Number.isFinite(value) && value >= 0 && value <= limits.deadlineMs) call[key] = value;
|
||||
}
|
||||
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));
|
||||
request.end(JSON.stringify({ message, frameToken: call.token }));
|
||||
};
|
||||
attempt();
|
||||
});
|
||||
call.roundtripMs = performance.now() - roundtripAt;
|
||||
if (!call.open) return;
|
||||
if (
|
||||
!object(result) ||
|
||||
@@ -286,7 +543,9 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
error: { code: result.error.code, message: result.error.code },
|
||||
};
|
||||
await finish(call, fixedError("INVOCATION_ENDED"), reply);
|
||||
metrics.completed++;
|
||||
} catch (error) {
|
||||
config.diagnostics?.(Buffer.from(`RESTRICTED_HOST_ERROR:${error.code ?? error.name}\n`));
|
||||
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 } });
|
||||
@@ -311,9 +570,37 @@ 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,
|
||||
receivedAt: performance.now(),
|
||||
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 +608,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 +635,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");
|
||||
@@ -390,7 +679,16 @@ export async function serveRestrictedRegistry(config, io = process) {
|
||||
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")) };
|
||||
return {
|
||||
close: () => fail(fixedError("SERVICE_SHUTDOWN")),
|
||||
stats: () => ({
|
||||
...metrics,
|
||||
active: active.size,
|
||||
queued: queued.length,
|
||||
engines: engines.size,
|
||||
idle: idleEngines.length,
|
||||
}),
|
||||
};
|
||||
}
|
||||
|
||||
if (process.argv[1] === new URL(import.meta.url).pathname) {
|
||||
@@ -399,6 +697,7 @@ if (process.argv[1] === new URL(import.meta.url).pathname) {
|
||||
...config,
|
||||
workerd: process.env.QUIXOS_RESTRICTED_WORKERD,
|
||||
worker: process.env.QUIXOS_RESTRICTED_WORKER,
|
||||
sandbox: process.env.QUIXOS_RESTRICTED_SANDBOX,
|
||||
module: join(dirname(process.argv[2]), config.module),
|
||||
});
|
||||
}
|
||||
|
||||
@@ -5,7 +5,7 @@ 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";
|
||||
import { serveRestrictedRegistry, RestrictedEngineBudget } from "./service.mjs";
|
||||
|
||||
const workerd = process.env.QUIXOS_TEST_WORKERD;
|
||||
if (!workerd) throw Error("Set QUIXOS_TEST_WORKERD to the pinned engine");
|
||||
@@ -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);
|
||||
}
|
||||
@@ -45,12 +50,21 @@ async function fixture(source, options = {}) {
|
||||
}
|
||||
};
|
||||
const service = await serveRestrictedRegistry(
|
||||
{ workerd, worker, module, registryExport: "registry", exports: ["test"], sdk: [], ...options },
|
||||
{
|
||||
workerd,
|
||||
worker,
|
||||
module,
|
||||
sandbox: process.env.QUIXOS_TEST_SANDBOX,
|
||||
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 start = (input = null, budget) => {
|
||||
const invocation = "invoke-" + ++next,
|
||||
frame = "frame-" + next;
|
||||
send({
|
||||
@@ -62,6 +76,7 @@ async function fixture(source, options = {}) {
|
||||
receiver: null,
|
||||
input,
|
||||
effect: "snapshot",
|
||||
...(budget ? { budget } : {}),
|
||||
});
|
||||
return { invocation, frame };
|
||||
};
|
||||
@@ -70,6 +85,8 @@ async function fixture(source, options = {}) {
|
||||
receive,
|
||||
send,
|
||||
start,
|
||||
usage,
|
||||
stats: service.stats,
|
||||
async close() {
|
||||
await service.close();
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
@@ -78,6 +95,115 @@ async function fixture(source, options = {}) {
|
||||
}
|
||||
const registry = (code) => `export const registry={exports:['test'],async invoke(request,channel){${code}}};`;
|
||||
|
||||
test("uncertain OS cleanup cannot release engine capacity", async () => {
|
||||
const budget = new RestrictedEngineBudget(1, async () => ({
|
||||
async release() {
|
||||
throw Error("kernel cleanup not confirmed");
|
||||
},
|
||||
}));
|
||||
const release = await budget.acquire(
|
||||
() => false,
|
||||
async () => {},
|
||||
);
|
||||
await assert.rejects(release(), /cleanup not confirmed/);
|
||||
assert.equal(budget.stats().engines, 1);
|
||||
await assert.rejects(
|
||||
budget.acquire(
|
||||
() => false,
|
||||
async () => {},
|
||||
),
|
||||
{ code: "RESTRICTED_ENGINE_LIMIT" },
|
||||
);
|
||||
});
|
||||
|
||||
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 +348,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 +398,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,15 +437,21 @@ 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: 4000, maxActive: 1 },
|
||||
);
|
||||
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"));
|
||||
assert.equal((await f.receive()).kind, "port");
|
||||
// Expire while the first call is deliberately blocked. Two equal relative
|
||||
// deadlines let fast startup legitimately begin the queued call first.
|
||||
const queued = f.start(null, { root: "queued-deadline", deadlineMs: 50, depth: 1, calls: 10 });
|
||||
const expired = await f.receive();
|
||||
assert.equal(expired.invocation, queued.invocation);
|
||||
assert.equal(expired.error.code, "RESTRICTED_DEADLINE");
|
||||
f.send({ kind: "cancel", ...first });
|
||||
const cancelled = await f.receive();
|
||||
assert.equal(cancelled.invocation, first.invocation);
|
||||
assert.equal(cancelled.kind, "cancelled");
|
||||
const fresh = f.start();
|
||||
const port = await f.receive();
|
||||
assert.equal(port.invocation, fresh.invocation);
|
||||
@@ -325,3 +461,174 @@ test("queue wait consumes the deadline and cancelled work never reaches its port
|
||||
await f.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("warm engines create fresh transitive module state, closures and built-ins", async () => {
|
||||
const directory = await mkdtemp(join(tmpdir(), "qx-transitive-"));
|
||||
const helper = join(directory, "helper.mjs");
|
||||
await writeFile(
|
||||
helper,
|
||||
"let value=0, retained; export function touch(channel){const old=retained; retained=channel; return {count:++value, captured:!!old};}",
|
||||
);
|
||||
const f = await fixture(
|
||||
`import {touch} from 'helper.mjs'; let count=0;
|
||||
export const registry={exports:['test'],async invoke(request,channel){
|
||||
const before={object:Object.prototype.retained??null,array:Array.prototype.retained??null,
|
||||
map:Map.prototype.retained??null, global:globalThis.retained??null};
|
||||
Object.prototype.retained=request.input; Array.prototype.retained=request.input;
|
||||
Map.prototype.retained=request.input; globalThis.retained=request.input;
|
||||
return {count:++count, helper:touch(channel), before};
|
||||
}};`,
|
||||
{ sdk: [{ name: "helper.mjs", path: helper }] },
|
||||
);
|
||||
try {
|
||||
for (const value of ["private graph A", "private graph B", "private graph C"]) {
|
||||
f.start(value);
|
||||
assert.deepEqual((await f.receive()).output, {
|
||||
count: 1,
|
||||
helper: { count: 1, captured: false },
|
||||
before: { object: null, array: null, map: null, global: null },
|
||||
});
|
||||
}
|
||||
assert.equal(f.stats().starts, 1, "Qualification must exercise actual engine reuse");
|
||||
assert.equal(f.stats().reused, 2);
|
||||
assert.equal(f.stats().idle, 1);
|
||||
} finally {
|
||||
await f.close();
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("an expired frame token cannot borrow a later invocation's broker", async () => {
|
||||
const directory = await mkdtemp(join(tmpdir(), "qx-token-"));
|
||||
const worker = join(directory, "token-test.mjs");
|
||||
// Trusted test entrypoint deliberately replays the old token. Authored
|
||||
// registries never receive CONFIG or BROKER directly.
|
||||
await writeFile(
|
||||
worker,
|
||||
`export default {async fetch(request,env) {
|
||||
const message=await request.json();
|
||||
const output=message.input ? (await env.BROKER.fetch('http://frame/port/'+message.input,{
|
||||
method:'POST',body:JSON.stringify({portId:'read',action:{kind:'state',primitive:'read'},input:null})})).status
|
||||
: env.CONFIG.frameToken;
|
||||
return Response.json({status:'returned',output});
|
||||
}};`,
|
||||
);
|
||||
const f = await fixture(registry("return null;"), { worker });
|
||||
try {
|
||||
f.start();
|
||||
const token = (await f.receive()).output;
|
||||
assert.match(token, /^[a-f0-9-]{36}$/);
|
||||
f.start(token);
|
||||
assert.equal((await f.receive()).output, 409);
|
||||
assert.equal(f.stats().starts, 1);
|
||||
assert.equal(f.stats().reused, 1);
|
||||
} finally {
|
||||
await f.close();
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("idle engines evict without a graph poll and restart with fresh state", async () => {
|
||||
const f = await fixture(registry("return request.input;"), { engineIdleMs: 30 });
|
||||
try {
|
||||
f.start("first");
|
||||
assert.equal((await f.receive()).output, "first");
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
assert.equal(f.stats().engines, 0);
|
||||
f.start("second");
|
||||
assert.equal((await f.receive()).output, "second");
|
||||
assert.equal(f.stats().starts, 2);
|
||||
} finally {
|
||||
await f.close();
|
||||
}
|
||||
});
|
||||
|
||||
for (const loop of [
|
||||
"while(true) await Promise.resolve();",
|
||||
"const memory=[]; for(let n=0;n<128;n++) memory.push(new Uint8Array(256*1024)); while(true) await Promise.resolve(memory);",
|
||||
])
|
||||
test("await/allocation pressure is fenced without delaying a neighboring engine: " + loop.slice(0, 28), async () => {
|
||||
const f = await fixture(registry(`if(request.input==='spin'){${loop}} return request.input;`), {
|
||||
deadlineMs: 1500,
|
||||
});
|
||||
try {
|
||||
const stuck = f.start("spin", { root: "stuck", depth: 1, deadlineMs: 350, calls: 10 });
|
||||
const before = performance.now();
|
||||
const healthy = f.start("healthy");
|
||||
const replies = [await f.receive(), await f.receive()];
|
||||
assert.equal(replies.find((r) => r.invocation === healthy.invocation).output, "healthy");
|
||||
assert.equal(replies.find((r) => r.invocation === stuck.invocation).error.code, "RESTRICTED_DEADLINE");
|
||||
assert.equal(replies[0].invocation, healthy.invocation, "Healthy work must finish before the stuck deadline");
|
||||
assert.ok(performance.now() - before < 1200);
|
||||
f.start("after fault");
|
||||
assert.equal((await f.receive()).output, "after fault");
|
||||
assert.equal(f.stats().starts, 2, "The healthy engine survives its neighbor's termination");
|
||||
} finally {
|
||||
await f.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("detached microtask loops cannot escape the invocation wall deadline", async () => {
|
||||
const f = await fixture(
|
||||
registry("void(async()=>{while(true)await Promise.resolve();})(); return 'nominal return';"),
|
||||
{ deadlineMs: 200 },
|
||||
);
|
||||
try {
|
||||
const before = performance.now();
|
||||
f.start();
|
||||
assert.equal((await f.receive()).error.code, "RESTRICTED_DEADLINE");
|
||||
assert.ok(performance.now() - before < 1000);
|
||||
assert.equal(f.stats().engines, 0);
|
||||
} finally {
|
||||
await f.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("detached digest callbacks cannot use a completed frame or contaminate reuse", async () => {
|
||||
const f = await fixture(
|
||||
registry(`
|
||||
void(async()=>{
|
||||
for(let n=0;n<100;n++)await crypto.subtle.digest('SHA-256',new Uint8Array(32768));
|
||||
await channel({portId:'late',action:{kind:'state',primitive:'read'},input:request.input});
|
||||
})().catch(()=>{});
|
||||
return request.input;
|
||||
`),
|
||||
);
|
||||
try {
|
||||
f.start("private A");
|
||||
assert.equal((await f.receive()).output, "private A");
|
||||
await new Promise((resolve) => setTimeout(resolve, 75));
|
||||
f.start("private B");
|
||||
const reply = await f.receive();
|
||||
assert.equal(reply.status, "returned", "No late graph port may appear after completion");
|
||||
assert.equal(reply.output, "private B");
|
||||
assert.equal(f.stats().starts, 1);
|
||||
} finally {
|
||||
await f.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("the shared engine budget evicts idle code families and rejects busy oversubscription", async () => {
|
||||
const engineBudget = new RestrictedEngineBudget(1);
|
||||
const a = await fixture(registry("return request.input;"), { engineBudget });
|
||||
const b = await fixture(
|
||||
registry("return await channel({portId:'read',action:{kind:'state',primitive:'read'},input:null});"),
|
||||
{ engineBudget },
|
||||
);
|
||||
try {
|
||||
a.start("first family");
|
||||
assert.equal((await a.receive()).output, "first family");
|
||||
const busy = b.start();
|
||||
const port = await b.receive();
|
||||
assert.equal(a.stats().engines, 0, "Idle foreign code is evicted before allocating another engine");
|
||||
assert.equal(engineBudget.stats().engines, 1);
|
||||
a.start("overload");
|
||||
assert.equal((await a.receive()).error.code, "RESTRICTED_ENGINE_LIMIT");
|
||||
assert.equal(engineBudget.stats().engines, 1);
|
||||
b.send({ kind: "port-result", ...busy, sequence: port.sequence, status: "returned", output: "survived" });
|
||||
assert.equal((await b.receive()).output, "survived");
|
||||
} finally {
|
||||
await Promise.all([a.close(), b.close()]);
|
||||
}
|
||||
assert.equal(engineBudget.stats().engines, 0);
|
||||
});
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
// Platform entrypoint, evaluated before loading the authored registry. A fresh
|
||||
// engine process owns exactly one frame. BROKER grants only that frame's ports.
|
||||
// Worker owns exactly one frame. The parent engine is reusable; BROKER's host
|
||||
// token grants only this frame and expires before the next invocation.
|
||||
const monotonic = performance.now.bind(performance);
|
||||
const deny = () => {
|
||||
throw Error("RESTRICTED_AMBIENT_ACCESS");
|
||||
};
|
||||
@@ -48,7 +50,9 @@ export default {
|
||||
if (used) return new Response("FRAME_EXPIRED", { status: 409 });
|
||||
used = true;
|
||||
const input = await request.json();
|
||||
const loading = monotonic();
|
||||
const source = await import("./package.mjs");
|
||||
const initializationMs = monotonic() - loading;
|
||||
const registry = source[env.CONFIG.registryExport];
|
||||
if (
|
||||
!registry ||
|
||||
@@ -59,7 +63,7 @@ export default {
|
||||
let open = true;
|
||||
const channel = async (value) => {
|
||||
if (!open) throw Error("EXPIRED_CONTEXT");
|
||||
const response = await env.BROKER.fetch("http://frame/port", {
|
||||
const response = await env.BROKER.fetch("http://frame/port/" + env.CONFIG.frameToken, {
|
||||
method: "POST",
|
||||
body: JSON.stringify(value),
|
||||
});
|
||||
@@ -71,6 +75,7 @@ export default {
|
||||
});
|
||||
return result.output;
|
||||
};
|
||||
const running = monotonic();
|
||||
try {
|
||||
const output = await registry.invoke(
|
||||
{ exportId: input.exportId, workspace: input.workspace, receiver: input.receiver, input: input.input },
|
||||
@@ -78,7 +83,13 @@ export default {
|
||||
);
|
||||
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" } });
|
||||
return new Response(body, {
|
||||
headers: {
|
||||
"content-type": "application/json",
|
||||
"x-quixos-initialization-ms": String(initializationMs),
|
||||
"x-quixos-execution-ms": String(monotonic() - running),
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
// Only authored/codec rejection codes cross the engine boundary. Engine
|
||||
// diagnostics stay private to the supervisor, never become protocol data.
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
# Pinned upstream engine. This binary supplies an import/capability boundary;
|
||||
# production execution additionally requires a workspace VM and OS limits.
|
||||
# production execution additionally requires the qualified OS identity,
|
||||
# namespace, syscall, cgroup and invocation-broker boundaries.
|
||||
{ pkgs }:
|
||||
let
|
||||
version = "1.20261005.1";
|
||||
@@ -37,7 +38,7 @@ pkgs.stdenvNoCC.mkDerivation {
|
||||
"$out/bin/workerd" --version | grep -Fx 'workerd 2026-10-05'
|
||||
'';
|
||||
meta = {
|
||||
description = "Pinned workerd engine for VM-scoped Quixos restricted execution";
|
||||
description = "Pinned workerd engine for Quixos restricted execution";
|
||||
homepage = "https://github.com/cloudflare/workerd";
|
||||
license = pkgs.lib.licenses.asl20;
|
||||
platforms = builtins.attrNames platforms;
|
||||
|
||||
Reference in New Issue
Block a user