From 38c34881cc857cb0dbfda7f6129b69854ca63e79 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Wed, 9 Sep 2026 01:31:16 +0000 Subject: [PATCH] Split UI updates 5/6: supervision --- scripts/package-release.mjs | 15 ++ scripts/test-updates-systemd.ts | 137 +++++++++++++++++ src/cli/main.ts | 172 +++++++++++++++++++-- src/server.ts | 3 + src/server/updates.server.ts | 259 ++++++++++++++++++++++++++++++++ src/updater/client.ts | 54 +++++++ src/updater/daemon.ts | 242 +++++++++++++++++++++++++++++ src/updater/peer-broker.py | 61 ++++++++ src/updater/systemd.ts | 231 ++++++++++++++++++++++++++++ tests/ui-updates.test.ts | 31 ++++ tests/update-http.test.ts | 87 +++++++++++ tests/update-socket.test.ts | 82 ++++++++++ tests/update-work.test.ts | 71 +++++++++ 13 files changed, 1430 insertions(+), 15 deletions(-) create mode 100644 scripts/test-updates-systemd.ts create mode 100644 src/server/updates.server.ts create mode 100644 src/updater/client.ts create mode 100644 src/updater/daemon.ts create mode 100644 src/updater/peer-broker.py create mode 100644 src/updater/systemd.ts create mode 100644 tests/update-http.test.ts create mode 100644 tests/update-socket.test.ts create mode 100644 tests/update-work.test.ts diff --git a/scripts/package-release.mjs b/scripts/package-release.mjs index 7498cc5..7b45a3a 100644 --- a/scripts/package-release.mjs +++ b/scripts/package-release.mjs @@ -182,6 +182,21 @@ writeFileSync( )}\n`, ); +writeFileSync( + join(bundle, "compatibility.json"), + JSON.stringify({ + protocol: 1, + startupGate: 1, + app: { min: 0, max: 10, output: 10 }, + auth: { min: 0, max: 0, output: 0 }, + data: "complete-snapshot-v1", + externalState: "unchanged", + codex: pins.codex.version, + }), +); +for (const file of ["extract.py", "peer-broker.py"]) + copyReleaseTree(join(root, "src/updater", file), join(bundle, "cli", file)); + const archive = join(output, "roost-linux-x64.tar.gz"); execFileSync( diff --git a/scripts/test-updates-systemd.ts b/scripts/test-updates-systemd.ts new file mode 100644 index 0000000..5d995bd --- /dev/null +++ b/scripts/test-updates-systemd.ts @@ -0,0 +1,137 @@ +/** Disposable-guest-only lifecycle driver. Never used by application or helper. + * Bundle with Vite SSR and run as the guest installation user in its own systemd + * unit. The guest must have the explicit marker below and this exact test root. */ + +import { randomUUID } from "node:crypto"; +import { existsSync } from "node:fs"; +import { readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { UpdateEngine } from "../src/updater/engine"; +import { readGate } from "../src/updater/gate"; +import { durableJson } from "../src/updater/journal"; +import { parseOffer } from "../src/updater/releases"; +import { systemdAdapter } from "../src/updater/systemd"; + +const root = "/home/ubuntu/roost-update-test"; +if (!existsSync("/etc/roost-update-disposable") || process.getuid?.() !== 1000) + throw new Error( + "This driver requires the explicitly marked disposable Ubuntu guest.", + ); +const config = JSON.parse(await readFile(join(root, "config.json"), "utf8")); +if (config.root !== root || config.user !== "ubuntu") + throw new Error("Fixture installation mismatch."); +const [ + version = "0.1.41", + boundary = "none", + mode = "run", + run = randomUUID(), + scenario = "normal", +] = process.argv.slice(2); +const adapter = systemdAdapter(config, join(root, "current", "cli")); +const readyDeadline = Date.now() + 120000; +while (true) { + const open = readGate(root).mode === "open"; + let healthy = false; + if (open) { + if (!(await adapter.running())) healthy = true; + else + try { + healthy = ( + await fetch(`http://127.0.0.1:${config.port}/api/health`, { + signal: AbortSignal.timeout(2000), + }) + ).ok; + } catch { + /* Wait for the test app. */ + } + } + if (healthy) break; + if (Date.now() > readyDeadline) + throw new Error("Disposable fixture did not become ready."); + await delay(500); +} +// Artifacts are preinstalled copies of the built package in this lifecycle test. +// Network pinning and hostile archive rejection have separate isolated tests. +adapter.stage = async () => { + if (scenario === "staging-failure") + throw new Error("Disposable staging failure"); + if (scenario === "cancel") await delay(200); +}; +if (scenario === "busy") + adapter.blockers = async () => ["Disposable uncertain worker"]; +if (scenario === "rollback-probe") + adapter.probe = async () => { + throw new Error("Disposable readiness failure"); + }; +// Rollback-boundary cases inject the readiness failure; the independent native +// browser and lifecycle suites exercise real bad-health detection and deadlines. +if ( + scenario === "candidate-probe-failure" || + (scenario === "normal" && + /restor|failed-data|rolled-back|rollback-/.test(boundary)) +) { + const probe = adapter.probe; + adapter.probe = async (target, id, token) => { + if (target === version) + throw new Error("Disposable candidate readiness failure"); + return probe(target, id, token); + }; +} +const engine = new UpdateEngine( + root, + adapter, + scenario === "busy" ? 50 : 30000, + async (phase) => { + if (phase !== boundary) return; + await durableJson(join(root, "boundary.json"), { + phase, + mode, + run, + version, + pid: process.pid, + boot: (await readFile("/proc/sys/kernel/random/boot_id", "utf8")).trim(), + operation: (await engine.status())?.id, + reachedAt: Date.now(), + }); + if (mode === "pause") { + const deadline = Date.now() + 300000; + while (!existsSync(join(root, `continue-${run}`))) { + if (Date.now() > deadline) + throw new Error("Disposable boundary pause expired"); + await delay(100); + } + } + if (mode === "kill") process.kill(process.pid, "SIGKILL"); + if (mode === "reboot") await new Promise(() => setInterval(() => {}, 1000)); + }, +); +const offer = parseOffer( + { + id: 1, + tag_name: `v${version}`, + draft: false, + prerelease: false, + assets: [ + { + id: 2, + name: "roost-linux-x64.tar.gz", + size: 1, + digest: `sha256:${"a".repeat(64)}`, + url: "https://api.github.com/repos/srctl/roost/releases/assets/2", + }, + ], + }, + "srctl/roost", + Date.now(), +); +const operation = await engine.accept({ + actor: "b".repeat(64), + key: randomUUID(), + offer, + confirmedVersion: version, +}); +await writeFile(join(root, "test-operation"), operation.id, { mode: 0o600 }); +if (scenario === "cancel") await engine.cancel(operation.id); +await engine.settled(); +console.log(JSON.stringify(await engine.status())); diff --git a/src/cli/main.ts b/src/cli/main.ts index 5dcd765..728d208 100644 --- a/src/cli/main.ts +++ b/src/cli/main.ts @@ -15,6 +15,11 @@ import { dirname, join, resolve } from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; import { Effect } from "effect"; import { AuthStore } from "../server/auth/store.server"; +import { updaterRequest } from "../updater/client"; +import { serveUpdater } from "../updater/daemon"; +import { enroll, readEnrollment } from "../updater/enrollment"; +import { readGate, startupGuard } from "../updater/gate"; +import { withKernelLock } from "../updater/lock"; import { activate, downloadRelease, @@ -45,6 +50,9 @@ const help = `Roost roost setup [--repository owner/repo] [--port 3000] [--skip-login] roost setup --login Sign in with the bundled Codex roost update [--version 0.1.0] + roost updates enroll Explicit operator enrollment (sudo) + roost updates status [--id UUID] Read-only helper diagnostics + roost updates repair --id UUID --decision restore|resume --confirm-version X.Y.Z roost auth setup --origin https://roost.example.com roost auth recover [--origin https://roost.example.com] roost server start @@ -256,6 +264,88 @@ async function main() { return; } + if (action === "updates") { + const [operation, ...rest] = args; + if (operation === "serve") { + flags(rest, {}); + await serveUpdater(root, join(bundle, "cli")); + return; + } + if (operation === "enroll") { + flags(rest, {}); + await enroll(await config(), (await readRelease(bundle)).version); + return; + } + if (operation === "repair") { + const options = flags(rest, { + "--id": "value", + "--decision": "value", + "--confirm-version": "value", + }); + console.log( + await updaterRequest(root, { + action: "repair", + id: options["--id"], + decision: options["--decision"], + version: options["--confirm-version"], + }), + ); + return; + } + if (operation === "status") { + const options = flags(rest, { "--id": "value" }); + // Diagnostics are independently available even if the web app/socket is down. + const { readJournal } = await import("../updater/journal"); + const { readdir } = await import("node:fs/promises"); + const enrollment = await readEnrollment(root); + console.log( + JSON.stringify( + { + unit: serviceName(enrollment.installation), + helper: `roost-${enrollment.installation.uid}-updater.service`, + root, + gate: { + mode: readGate(root).mode, + operation: readGate(root).operation, + }, + lockOwner: existsSync(join(root, "updater-owner.json")) + ? JSON.parse( + await readFile(join(root, "updater-owner.json"), "utf8"), + ) + : null, + }, + null, + 2, + ), + ); + for (const id of await readdir(join(root, "updates"))) { + if ( + !/^[a-f0-9-]{36}$/.test(id) || + (options["--id"] && id !== options["--id"]) + ) + continue; + const journal = await readJournal(root, id); + console.log( + JSON.stringify( + { + id, + phase: journal.phase, + previous: journal.previous, + candidate: journal.candidate, + snapshotDigest: journal.snapshotDigest, + location: join(root, "updates", id), + }, + null, + 2, + ), + ); + } + return; + } + throw new Error( + "Use roost updates enroll or status. Recovery runs automatically in the supervised helper; inspect its journal before operator repair.", + ); + } if (action === "auth") { const [operation, ...rest] = args; if (operation !== "setup" && operation !== "recover") @@ -263,20 +353,29 @@ async function main() { const options = flags(rest, { "--origin": "value" }); if (operation === "setup" && !options["--origin"]) throw new Error("Specify --origin https://your-host."); - const store = new AuthStore( - resolve(process.env.ROOST_DATA_DIR ?? join(root, "data")), - ); - try { - const link = store.setup(options["--origin"], operation === "recover"); - console.log( - "Open this private, single-use link within 15 minutes to register your passkey:", + await withKernelLock(root, async () => { + if ( + existsSync(join(root, "updater.json")) && + readGate(root).mode !== "open" + ) + throw new Error( + "Authentication recovery must wait for updater recovery.", + ); + const store = new AuthStore( + resolve(process.env.ROOST_DATA_DIR ?? join(root, "data")), ); - console.log(link); - if (operation === "recover") - console.log("Previous passkeys and sessions were revoked."); - } finally { - store.close(); - } + try { + const link = store.setup(options["--origin"], operation === "recover"); + console.log( + "Open this private, single-use link within 15 minutes to register your passkey:", + ); + console.log(link); + if (operation === "recover") + console.log("Previous passkeys and sessions were revoked."); + } finally { + store.close(); + } + }); return; } if (process.platform !== "linux" || process.arch !== "x64") @@ -298,7 +397,37 @@ async function main() { } if (action === "update") { const options = flags(args, { "--version": "value" }); - await withLock(root, () => update(options)); + if (existsSync(join(root, "updater.json"))) { + const status = await updaterRequest<{ + latest?: { id: string; version: string; expiresAt: number }; + }>(root, { action: "status" }); + const offer = + status.latest && status.latest.expiresAt > Date.now() + ? status.latest + : await updaterRequest<{ id: string; version: string }>(root, { + action: "check", + }); + if (!options["--version"]) + throw new Error( + `Confirm the exact offer with roost update --version ${offer.version}. This requires an outage.`, + ); + if (options["--version"] !== offer.version) + throw new Error( + "Enrolled CLI updates accept only the pinned latest stable offer.", + ); + const { createHash, randomUUID } = await import("node:crypto"); + console.log( + await updaterRequest(root, { + action: "accept", + offerId: offer.id, + version: offer.version, + key: randomUUID(), + actor: createHash("sha256") + .update(`operator:${userInfo().uid}`) + .digest("hex"), + }), + ); + } else await withLock(root, () => update(options)); return; } @@ -312,6 +441,13 @@ async function main() { const c = await config(); if (operation === "run") { const release = await readRelease(bundle); + // Read verification credentials after systemd has switched to the app UID. + // Never ask privileged PID 1 to read a file in this user-writable directory. + const gate = readGate(root); + if (gate.mode === "verify") process.env.ROOST_UPDATE_TOKEN = gate.token; + else delete process.env.ROOST_UPDATE_TOKEN; + startupGuard(root, release.version, process.env.ROOST_UPDATE_TOKEN); + process.env.ROOST_PUBLIC_DIR = join(bundle, "app/public"); process.env.HOST = "127.0.0.1"; process.env.NITRO_HOST = "127.0.0.1"; process.env.PORT = String(c.port); @@ -340,11 +476,17 @@ async function main() { return; } if (operation === "start") { - await withLock(root, () => start(c)); + if (existsSync(join(root, "updater.json"))) + await updaterRequest(root, { action: "start" }); + else await withLock(root, () => start(c)); return; } if (operation === "stop") { + if (existsSync(join(root, "updater.json"))) { + await updaterRequest(root, { action: "stop" }); + return; + } await withLock(root, async () => { await service(c, "stop"); console.log( diff --git a/src/server.ts b/src/server.ts index 774e70a..0cefb5b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -12,6 +12,7 @@ import { trackUpdateRequest, updateGateRequest, } from "./server/update-gate.server"; +import { updatesRequest } from "./server/updates.server"; const handler = createStartHandler(defaultStreamHandler); @@ -31,6 +32,8 @@ export default createServerEntry({ return trackUpdateRequest(async () => { const auth = await authGate(request, authPage); if (auth) return auth; + if (new URL(request.url).pathname.startsWith("/api/updates")) + return updatesRequest(request); const response = await followStartupRedirect( request, await handler(request, options), diff --git a/src/server/updates.server.ts b/src/server/updates.server.ts new file mode 100644 index 0000000..d800fc4 --- /dev/null +++ b/src/server/updates.server.ts @@ -0,0 +1,259 @@ +import { randomBytes } from "node:crypto"; +import { existsSync } from "node:fs"; +import { readFile, realpath } from "node:fs/promises"; +import { join } from "node:path"; +import { updaterRequest } from "../updater/client"; +import { capability } from "../updater/contract"; +import type { UpdateSummary } from "../updater/daemon"; +import { type Offer, ReleaseChecker } from "../updater/releases"; +import { authorizeMutation, csrfToken, updateBody } from "../updater/security"; +import { sessionId } from "./auth/session.server"; +import { digest, openAuth } from "./auth/store.server"; + +const secret = randomBytes(32).toString("hex"); +let checker: ReleaseChecker | undefined; +const json = (value: unknown, status = 200) => + Response.json(value, { + status, + headers: { + "Cache-Control": "private, no-store", + "X-Content-Type-Options": "nosniff", + }, + }); +async function installation() { + const root = process.env.ROOST_HOME; + if ( + !root || + !process.env.ROOST_RELEASE_VERSION || + process.env.ROOST_DATA_DIR !== join(root, "data") + ) + return { packaged: false, root: undefined, repository: undefined }; + try { + const c = JSON.parse(await readFile(join(root, "config.json"), "utf8")); + if (c.root !== (await realpath(root)) || c.uid !== process.getuid?.()) + throw new Error("Installation mismatch"); + const os = await readFile("/etc/os-release", "utf8").catch(() => ""); + const field = (name: string) => + new RegExp(`^${name}="?([^"\\n]+)`, "m").exec(os)?.[1] ?? "unknown"; + const mounts = ( + await readFile("/proc/self/mountinfo", "utf8").catch(() => "") + ) + .split("\n") + .map((line) => { + const [left, right] = line.split(" - "); + return { + path: (left?.split(" ")[4] ?? "").replace( + /\\([0-7]{3})/g, + (_, n: string) => String.fromCharCode(Number.parseInt(n, 8)), + ), + fs: right?.split(" ")[0], + }; + }) + .filter( + (mount) => + mount.path && + (root === mount.path || + root.startsWith(mount.path === "/" ? "/" : `${mount.path}/`)), + ) + .sort((a, b) => b.path.length - a.path.length); + return { + packaged: true, + distribution: `${field("ID")}:${field("VERSION_ID")}`, + filesystem: mounts[0]?.fs ?? "unknown", + root, + repository: c.repository as string | undefined, + }; + } catch { + return { packaged: false, root: undefined, repository: undefined }; + } +} +export async function updatesRequest(request: Request) { + const facts = await installation(); + const detected = capability({ + ...facts, + platform: process.platform, + arch: process.arch, + systemd: existsSync("/run/systemd/system"), + }); + const auth = openAuth(); + try { + const session = auth?.session(sessionId(request)); + if ( + auth && + (!session || + new URL(request.url).host !== new URL(auth.config().origin).host) + ) + return json({ error: "Native sign-in required." }, 401); + const enrolled = !!( + facts.root && existsSync(join(facts.root, "updater.json")) + ); + const url = new URL(request.url); + const match = /^\/api\/updates(?:\/([a-f0-9-]{36})(\/cancel)?)?$/.exec( + url.pathname, + ); + if (!match && url.pathname !== "/api/updates/check") + return json({ error: "Not found." }, 404); + if (facts.repository && checker?.repository !== facts.repository) + checker = new ReleaseChecker(facts.repository); + if (request.method === "GET" && match) { + const key = url.searchParams.get("key"); + if ( + [...url.searchParams.keys()].some((name) => name !== "key") || + (key !== null && + (match[1] || + url.searchParams.getAll("key").length !== 1 || + !/^[a-zA-Z0-9_-]{16,100}$/.test(key))) + ) + return json({ error: "Invalid status lookup." }, 400); + let helper: + | { + qualified: boolean; + latest: Offer | null; + operation: UpdateSummary | null; + } + | undefined; + let error: string | undefined; + if (enrolled && session) + try { + helper = await updaterRequest(facts.root!, { + action: "status", + ...(match[1] ? { id: match[1] } : key ? { key } : {}), + }); + } catch { + error = + "Updater is unavailable. Existing operations may still be recovering; do not submit again."; + } + return json({ + capability: + enrolled && detected.code === "setup-required" + ? { + code: !session + ? "native-auth-required" + : !helper + ? "updater-unavailable" + : helper.qualified + ? "supported" + : "qualification-required", + canActivate: !!helper?.qualified, + reason: !session + ? "Native passkey sign-in is required for UI updates." + : !helper + ? "The enrolled updater is unavailable. Inspect its service and durable status before retrying." + : helper.qualified + ? "Enrolled supervised updater is ready." + : "Updater is enrolled. Activation remains disabled until this helper build passes systemd and reboot qualification.", + } + : detected, + version: facts.packaged ? process.env.ROOST_RELEASE_VERSION : "dev", + latest: session ? (helper?.latest ?? checker?.cached ?? null) : null, + operation: helper?.operation ?? null, + enrolled, + error, + canCheck: !!(session && facts.repository), + csrf: session ? csrfToken(secret, session.id) : null, + recent: !!session && session.created >= Date.now() - 300000, + }); + } + if (request.method !== "POST") + return json({ error: "Method not allowed." }, 405); + if (!auth || !session) + return json({ error: "Native passkey sign-in required." }, 401); + let body: Record; + try { + authorizeMutation(request, { + origin: auth.config().origin, + session, + secret, + }); + body = await updateBody( + request, + url.pathname === "/api/updates" ? ["offerId", "version", "key"] : [], + ); + // Reading a bounded streaming body may outlive revocation or recent auth. + authorizeMutation(request, { + origin: auth.config().origin, + session: auth.session(session.id), + secret, + }); + } catch { + return json({ error: "Sign in again and retry from Settings." }, 403); + } + if (url.pathname === "/api/updates/check") { + if (!facts.repository || !checker) + return json( + { error: "No packaged release repository configured." }, + 409, + ); + if (!auth.limit("updates-check", 2)) + return json({ error: "Wait a minute before checking again." }, 429); + try { + return json({ + latest: enrolled + ? await updaterRequest(facts.root!, { action: "check" }) + : await checker.check(), + }); + } catch { + return json( + { + error: + "Release check failed. Check repository access and retry in a minute.", + }, + 503, + ); + } + } + if (!enrolled) + return json({ error: "Explicit terminal enrollment is required." }, 409); + if (match?.[2]) { + try { + return json({ + operation: await updaterRequest(facts.root!, { + action: "cancel", + id: match[1], + }), + }); + } catch { + return json( + { + error: + "Cancellation could not be confirmed. Read the operation status.", + }, + 409, + ); + } + } + if (!auth.limit("updates-start", 4)) + return json({ error: "Too many update requests." }, 429); + if ( + typeof body.offerId !== "string" || + !/^([a-f0-9]{64})$/.test(body.offerId) || + typeof body.version !== "string" || + body.version.length > 32 || + typeof body.key !== "string" || + !/^[a-zA-Z0-9_-]{16,100}$/.test(body.key) + ) + return json({ error: "Confirm the exact offered version." }, 400); + try { + return json( + { + operation: await updaterRequest(facts.root!, { + action: "accept", + ...body, + actor: digest(session.id), + }), + }, + 202, + ); + } catch { + return json( + { + error: + "Acceptance was not confirmed. Check durable status before any new confirmation.", + }, + 409, + ); + } + } finally { + auth?.close(); + } +} diff --git a/src/updater/client.ts b/src/updater/client.ts new file mode 100644 index 0000000..f3450ce --- /dev/null +++ b/src/updater/client.ts @@ -0,0 +1,54 @@ +import { connect } from "node:net"; +import { join } from "node:path"; +export async function updaterRequest( + root: string, + message: unknown, +): Promise { + const input = `${JSON.stringify({ protocol: 1, ...(message as object) })}\n`; + if (Buffer.byteLength(input) > 8192) + throw new Error("Update request too large."); + // Service shutdown may take 90 seconds; repair also copies data and probes + // startup. Keep ordinary HTTP-facing requests short, with fixed CLI budgets. + const action = (message as { action?: string })?.action; + const timeout = + action === "repair" + ? 610000 + : action === "start" || action === "stop" + ? 130000 + : 20000; + return new Promise((resolve, reject) => { + const socket = connect(join(root, "updates", "helper.sock")); + let output = ""; + socket.setTimeout(timeout, () => + socket.destroy( + new Error( + "Updater did not respond. Check durable status; do not repeat activation.", + ), + ), + ); + socket.once("connect", () => socket.write(input)); + socket.on("data", (chunk) => { + output += chunk; + if (output.length > 65536) + socket.destroy(new Error("Updater response too large.")); + }); + socket.once("error", reject); + socket.once("end", () => { + if (!output.trim()) { + reject( + new Error( + "Updater connection ended without a result. Check durable status before repeating an operation.", + ), + ); + return; + } + try { + const result = JSON.parse(output); + if (result.error) reject(new Error(result.error)); + else resolve(result.value); + } catch { + reject(new Error("Invalid updater response.")); + } + }); + }); +} diff --git a/src/updater/daemon.ts b/src/updater/daemon.ts new file mode 100644 index 0000000..b8f23cc --- /dev/null +++ b/src/updater/daemon.ts @@ -0,0 +1,242 @@ +import { spawn } from "node:child_process"; +import { lstat, readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { type Operation, terminal, UpdateEngine } from "./engine"; +import { readEnrollment, validateInstallation } from "./enrollment"; +import { openGate, readGate, writeGate } from "./gate"; +import { durableJson } from "./journal"; +import { withKernelLock } from "./lock"; +import { execute } from "./process"; +import { type Offer, ReleaseChecker } from "./releases"; +import { systemdAdapter } from "./systemd"; + +export function summary(record: Operation | null) { + return record + ? { + id: record.id, + requestKey: record.request.key, + phase: record.phase, + previous: record.previous, + version: record.candidate, + updatedAt: record.updatedAt, + cancellable: ["accepted", "staged", "draining"].includes(record.phase), + error: record.error, + blockers: record.blockers, + bytes: record.bytes, + committed: record.committed, + } + : null; +} +export type UpdateSummary = NonNullable>; +export async function serveUpdater(root: string, support: string) { + const mainPid = await execute( + "/usr/bin/systemctl", + [ + "show", + `roost-${process.getuid?.()}-updater.service`, + "--property=MainPID", + "--value", + ], + 10000, + ); + if (mainPid !== String(process.pid)) + throw new Error("Run the updater through its enrolled systemd supervisor."); + // A second manual/service invocation must not close another helper's gate or + // replace its live socket. This lifetime lock is separate from transaction ownership. + return withKernelLock(root, () => supervise(root, support), "supervisor"); +} +async function supervise(root: string, support: string) { + const enrollment = await readEnrollment(root); + await validateInstallation(enrollment.installation, true); + const tokenPath = join(root, "updates", "github-token"); + let token: string | undefined; + try { + const info = await lstat(tokenPath); + if ( + !info.isFile() || + info.isSymbolicLink() || + info.uid !== process.getuid?.() || + info.mode & 0o077 || + info.size > 1024 + ) + throw new Error("Updater credential must be an owner-only regular file."); + token = (await readFile(tokenPath, "utf8")).trim(); + } catch (e) { + if ((e as NodeJS.ErrnoException).code !== "ENOENT") throw e; + } + const adapter = systemdAdapter(enrollment.installation, support, token); + const engine = new UpdateEngine(root, adapter); + const priorGate = readGate(root); + await writeGate(root, { + ...openGate, + mode: "hold", + operation: priorGate.operation ?? "boot", + }); + await durableJson(join(root, "updates", "ready.json"), { + boot: (await readFile("/proc/sys/kernel/random/boot_id", "utf8")).trim(), + }); + try { + if (priorGate.mode === "manual" && !priorGate.operation) + throw new Error("Missing or corrupt gate evidence."); + await engine.recover(); + } catch { + await writeGate(root, { ...openGate, mode: "manual", operation: "boot" }); + } + const checker = new ReleaseChecker( + enrollment.installation.repository!, + fetch, + token, + ); + let offer: Offer | undefined; + try { + offer = JSON.parse( + await readFile(join(root, "updates", "offer.json"), "utf8"), + ); + } catch { + /* A missing offer is never permission to activate. */ + } + const handle = async (message: Record) => { + if (message.protocol !== 1 || typeof message.action !== "string") + throw new Error("Unsupported updater protocol."); + const fields: Record = { + status: ["id", "key", "actor"], + check: [], + accept: ["offerId", "version", "actor", "key"], + cancel: ["id"], + start: [], + stop: [], + repair: ["id", "decision", "version"], + }; + if ( + !fields[message.action] || + Object.keys(message).some( + (key) => + ![ + "protocol", + "action", + ...fields[message.action as string]!, + ].includes(key), + ) + ) + throw new Error("Unexpected updater request fields."); + if (message.action === "status") { + // Authenticated owner reads may reconcile a key after native reauth. + // This grants no mutation authority; accept still checks the original actor. + const record = message.key + ? ((await engine.records()).find( + (r) => + r.request.key === message.key && + (message.actor === undefined || + r.request.actor === message.actor), + ) ?? null) + : await engine.status( + typeof message.id === "string" ? message.id : undefined, + ); + return { + protocol: 1, + qualified: true, + latest: offer ?? null, + operation: summary(record), + }; + } + if (message.action === "check") { + offer = await checker.check(); + await durableJson(join(root, "updates", "offer.json"), offer); + return offer; + } + if (message.action === "accept") { + const existing = (await engine.records()).find( + (r) => r.request.key === message.key, + ); + if (existing) { + if ( + existing.request.actor !== message.actor || + existing.request.offer.id !== message.offerId || + existing.candidate !== message.version + ) + throw new Error("Idempotency conflict."); + return summary(existing); + } + if ( + !offer || + offer.id !== message.offerId || + typeof message.actor !== "string" || + typeof message.key !== "string" || + typeof message.version !== "string" + ) + throw new Error("Check releases and confirm the offered version."); + return summary( + await engine.accept({ + actor: message.actor, + key: message.key, + offer, + confirmedVersion: message.version, + }), + ); + } + if (message.action === "repair") { + if ( + typeof message.id !== "string" || + typeof message.version !== "string" || + !["restore", "resume"].includes(String(message.decision)) + ) + throw new Error("Invalid repair confirmation."); + return summary( + await engine.repair( + message.id, + message.decision as "restore" | "resume", + message.version, + ), + ); + } + if (message.action === "cancel") { + if (typeof message.id !== "string") + throw new Error("Missing operation ID."); + return summary(await engine.cancel(message.id)); + } + return withKernelLock(root, async () => { + const latest = await engine.status(); + if ( + latest && + (!terminal(latest.phase) || latest.phase === "manual-recovery") + ) + throw new Error("Resolve the active update first."); + if (message.action === "start") await adapter.start(); + else await adapter.stop(); + return { ok: true }; + }); + }; + const child = spawn( + "/usr/bin/python3", + [join(support, "peer-broker.py"), join(root, "updates", "helper.sock")], + { stdio: ["pipe", "pipe", "inherit"] }, + ); + const lines = createInterface({ input: child.stdout }); + for await (const line of lines) { + if (line.length > 16384) { + child.kill(); + throw new Error("Invalid socket bridge frame."); + } + const envelope = JSON.parse(line) as { + ready?: boolean; + id: string; + message: Record; + }; + if (envelope.ready) continue; + // Requests may observe/cancel an operation while its long transaction runs. + void handle(envelope.message) + .then( + (value) => ({ value }), + () => ({ + error: + "Updater request refused. Check enrollment, confirmation and durable status.", + }), + ) + .then((result) => { + if (child.stdin.writable) + child.stdin.write(`${JSON.stringify({ id: envelope.id, result })}\n`); + }); + } + throw new Error("Updater socket bridge stopped."); +} diff --git a/src/updater/peer-broker.py b/src/updater/peer-broker.py new file mode 100644 index 0000000..b0b6169 --- /dev/null +++ b/src/updater/peer-broker.py @@ -0,0 +1,61 @@ +"""Private Unix socket transport with Linux SO_PEERCRED; no command execution. +Node owns policy/state. This bridge only multiplexes bounded JSON envelopes. +""" +import json, os, queue, socket, struct, sys, threading, uuid +path=sys.argv[1] +pending={} +lock=threading.Lock() +slots=threading.BoundedSemaphore(32) +server=socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) +# systemd owns this process group; on a daemon restart no old broker remains. +try: os.unlink(path) +except FileNotFoundError: pass +server.bind(path); os.chmod(path,0o600);server.listen(16) + +def reply_reader(): + for line in sys.stdin: + try: + message=json.loads(line) + with lock: target=pending.get(message['id']) + if target: target.put_nowait(message['result']) + except (ValueError,KeyError,queue.Full): pass + os._exit(0) +threading.Thread(target=reply_reader,daemon=True).start() + +def client(connection): + request_id=None + try: + connection.settimeout(20) + _,uid,_=struct.unpack('3i',connection.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12)) + if uid != os.getuid(): return + data=b'' + while b'\n' not in data: + chunk=connection.recv(4096) + if not chunk: return + data+=chunk + if len(data)>8192: return + if data.count(b'\n')!=1 or data.split(b'\n',1)[1]: return + message=json.loads(data) + if not isinstance(message,dict): return + # Fixed CLI-only service/repair budgets; clients cannot supply a timeout. + action=message.get('action') + wait=600 if action=='repair' else 120 if action in ('start','stop') else 18 + request_id=str(uuid.uuid4()); result=queue.Queue(1) + with lock: + pending[request_id]=result + print(json.dumps({'id':request_id,'message':message}),flush=True) + try: response=result.get(timeout=wait) + except queue.Empty: + response={'error':'Updater response deadline exceeded. Check durable status before repeating an operation.'} + answer=json.dumps(response).encode()+b'\n' + if len(answer)>65536: return + connection.sendall(answer) + except (ValueError,OSError,queue.Empty): pass + finally: + with lock: pending.pop(request_id,None) + connection.close(); slots.release() +print(json.dumps({'ready':True}),flush=True) +while True: + connection,_=server.accept() + if not slots.acquire(False): connection.close();continue + threading.Thread(target=client,args=(connection,),daemon=True).start() diff --git a/src/updater/systemd.ts b/src/updater/systemd.ts new file mode 100644 index 0000000..a9f5538 --- /dev/null +++ b/src/updater/systemd.ts @@ -0,0 +1,231 @@ +import { readFile } from "node:fs/promises"; +import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { setTimeout as delay } from "node:timers/promises"; +import { readRelease } from "../cli/releases"; +import { type Installation, serviceName } from "../cli/service"; +import { maintenance } from "../cli/state"; +import { stageArtifact } from "./artifact"; +import type { EngineAdapter } from "./engine"; +import { validateInstallation, validateService } from "./enrollment"; +import { readGate } from "./gate"; +import { execute } from "./process"; + +/** Labels alone do not establish quiescence. Missing/stale/changed identity + * observations never authorize stopping an installation. */ +export function codingWorkUncertain( + j: Record, + now = Date.now(), +) { + if ( + j.lastWorkerState === "not_started" && + j.status === "blocked" && + !j.observedWorking && + !j.sessionIdentity && + !j.nativeSessionId + ) + return false; + return ( + j.status !== "review" || + !j.observedWorking || + !["idle", "done"].includes(String(j.lastWorkerState)) || + !!j.cancelRequested || + !j.sessionIdentity || + !j.nativeSessionId || + Number(j.lastCheckedAt) < now - 15000 || + !Number.isFinite(Number(j.lastCheckedAt)) || + !!j.error + ); +} +export function systemdAdapter( + c: Installation, + support: string, + token?: string, +): EngineAdapter { + const unit = serviceName(c); + const control = (action: "start" | "stop") => + execute("/usr/bin/sudo", ["-n", "/usr/bin/systemctl", action, unit]); + async function appProbe(path: string) { + const gate = readGate(c.root); + const response = await fetch(`http://127.0.0.1:${c.port}${path}`, { + headers: { "X-Roost-Updater": gate.token ?? gate.operation ?? "" }, + signal: AbortSignal.timeout(2000), + }); + if (!response.ok) throw new Error("Application update probe unavailable."); + return response.json(); + } + return { + async running() { + const status = await execute("/usr/bin/systemctl", [ + "show", + unit, + "--property=ActiveState", + "--value", + ]); + return status === "active" || status === "activating"; + }, + stage: (offer, id, signal, progress) => + stageArtifact(c.root, offer, id, support, signal, progress, token), + async preflight(candidate) { + await validateInstallation(c); + await validateService(c); + await readRelease(join(c.root, "releases", candidate)); + // sudo -n -l validates authorization without actually starting/stopping. + for (const action of ["start", "stop"]) + await execute("/usr/bin/sudo", [ + "-n", + "-l", + "/usr/bin/systemctl", + action, + unit, + ]); + const fragment = await execute("/usr/bin/systemctl", [ + "show", + unit, + "--property=ExecStart", + "--value", + ]); + if (!fragment.includes(join(c.root, "current", "runtime/node"))) + throw new Error("Service identity differs from enrollment."); + }, + async admission(blocked) { + maintenance(c.root, blocked); + }, + async blockers(frozen) { + let db: DatabaseSync | undefined; + try { + db = new DatabaseSync(join(c.root, "data", "roost.sqlite"), { + readOnly: true, + }); + db.exec("PRAGMA busy_timeout=5000"); + } catch { + db?.close(); + return ["Work cannot be verified because its store is unavailable."]; + } + const blockers: string[] = []; + try { + if ( + Number( + db + .prepare( + "SELECT count(*) AS n FROM runs WHERE status IN ('running','steering')", + ) + .get()?.n, + ) + ) + blockers.push("Conversations or steering are still active."); + const jobs = db + .prepare( + "SELECT * FROM coding_jobs WHERE status NOT IN ('completed','cancelled','failed','queued')", + ) + .all(); + for (const j of jobs) { + if (codingWorkUncertain(j)) + blockers.push( + "A coding worker is active, missing, or uncertain; inspect it before retrying.", + ); + } + if ( + Number( + db + .prepare( + "SELECT count(*) AS n FROM coding_job_inputs WHERE status IN ('launching','dispatching')", + ) + .get()?.n, + ) + ) + blockers.push("Coding submission is in flight."); + } catch { + blockers.push( + "Work cannot be verified because its store is busy or unreadable.", + ); + } finally { + db.close(); + } + if (await this.running()) { + try { + const result = await appProbe("/api/updates/quiescence"); + if (result.tasks || result.requests || result.login) + blockers.push("Application tasks or requests are still finishing."); + if (frozen && !result.frozen) + blockers.push("Writer gate has not closed."); + } catch { + blockers.push("Application quiescence cannot be verified."); + } + } + return [...new Set(blockers)]; + }, + async stop() { + const before = await execute("/usr/bin/systemctl", [ + "show", + unit, + "--property=ControlGroup", + "--value", + ]); + await control("stop"); + const values = await execute("/usr/bin/systemctl", [ + "show", + unit, + "--property=ActiveState,SubState,MainPID,ControlGroup", + ]); + if ( + !values.includes("ActiveState=inactive") || + !values.includes("MainPID=0") + ) + throw new Error("Service did not stop cleanly."); + const reported = values + .split("\n") + .find((s) => s.startsWith("ControlGroup=")) + ?.slice(13); + const group = before || reported; + if (group) { + try { + if ( + ( + await readFile( + join("/sys/fs/cgroup", group, "cgroup.events"), + "utf8", + ) + ).includes("populated 1") + ) + throw new Error("Service group still has writers."); + } catch (e) { + if ((e as NodeJS.ErrnoException).code !== "ENOENT") throw e; + } + } + }, + async start() { + await control("start"); + }, + async probe(version, id, tokenValue) { + const contract = JSON.parse( + await readFile( + join(c.root, "releases", version, "compatibility.json"), + "utf8", + ), + ); + const deadline = Date.now() + 120000; + while (Date.now() < deadline) { + try { + const result = await appProbe("/api/updates/probe"); + if ( + result.version === version && + result.operation === id && + result.token === tokenValue && + result.integrity === true && + result.appSchema === contract.app.output && + result.authSchema === contract.auth.output && + result.assets === true && + result.shell === true && + result.workerReady === true + ) + return; + } catch { + /* Startup may still be in progress. */ + } + await delay(500); + } + throw new Error("Candidate readiness deadline exceeded."); + }, + }; +} diff --git a/tests/ui-updates.test.ts b/tests/ui-updates.test.ts index b2adf6b..c3e70d8 100644 --- a/tests/ui-updates.test.ts +++ b/tests/ui-updates.test.ts @@ -15,6 +15,7 @@ import { join } from "node:path"; import { DatabaseSync } from "node:sqlite"; import { test } from "node:test"; import { activeRuns, withLock } from "../src/cli/state"; +import { updatesRequest } from "../src/server/updates.server"; import { assertCompatible, capability, @@ -448,3 +449,33 @@ test("journal updates reject sequence reuse and rollback after commit", async () } }); +test("source status is read-only and unauthenticated activation cannot reach any updater", async () => { + const directory = await mkdtemp("/tmp/ui-update-http-"); + const old = process.env.ROOST_DATA_DIR; + process.env.ROOST_DATA_DIR = directory; + try { + const response = await updatesRequest( + new Request("https://roost.example/api/updates"), + ); + assert.equal(response.status, 200); + assert.equal(response.headers.get("cache-control"), "private, no-store"); + const status = await response.json(); + assert.equal(status.capability.canActivate, false); + assert.equal(status.version, "dev"); + assert.equal(status.canCheck, false); + assert.equal(status.csrf, null); + for (const path of ["/api/updates", "/api/updates/check"]) { + const denied = await updatesRequest( + new Request(`https://roost.example${path}`, { + method: "POST", + body: JSON.stringify({ unit: "arbitrary.service" }), + }), + ); + assert.equal(denied.status, 401); + } + } finally { + if (old === undefined) delete process.env.ROOST_DATA_DIR; + else process.env.ROOST_DATA_DIR = old; + await rm(directory, { recursive: true, force: true }); + } +}); diff --git a/tests/update-http.test.ts b/tests/update-http.test.ts new file mode 100644 index 0000000..f7262dc --- /dev/null +++ b/tests/update-http.test.ts @@ -0,0 +1,87 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { test } from "node:test"; +import { AuthStore, digest } from "../src/server/auth/store.server"; +import { updatesRequest } from "../src/server/updates.server"; + +test("HTTP update authorization rechecks a native session after reading a slow request body", async () => { + const directory = await mkdtemp("/tmp/roost-update-http-"); + const previous = process.env.ROOST_DATA_DIR; + process.env.ROOST_DATA_DIR = directory; + const store = new AuthStore(directory); + try { + const origin = "https://roost.example"; + store.setup(origin); + const secret = store.createSession("test-credential"); + const cookie = `__Host-roost-session=${secret}`; + const status = await ( + await updatesRequest( + new Request(`${origin}/api/updates`, { headers: { cookie } }), + ) + ).json(); + assert.ok(status.csrf); + for (const query of [ + "?key=short", + `?key=${"x".repeat(101)}`, + "?key=abcdefghijklmnop&key=abcdefghijklmnop", + "?path=/etc/passwd", + ]) { + assert.equal( + ( + await updatesRequest( + new Request(`${origin}/api/updates${query}`, { + headers: { cookie }, + }), + ) + ).status, + 400, + ); + } + assert.equal( + ( + await updatesRequest( + new Request(`${origin}/api/updates?key=abcdefghijklmnop`, { + headers: { cookie }, + }), + ) + ).status, + 200, + ); + let controller!: ReadableStreamDefaultController; + const body = new ReadableStream({ + start(c) { + controller = c; + }, + }); + const request = new Request(`${origin}/api/updates`, { + method: "POST", + headers: { + cookie, + origin, + "Content-Type": "application/json", + "X-Roost-CSRF": status.csrf, + }, + body, + duplex: "half", + } as RequestInit); + const pending = updatesRequest(request); + await new Promise((resolve) => setTimeout(resolve, 20)); + store.revokeSession(digest(secret)); + controller.enqueue(new TextEncoder().encode("{}")); + controller.close(); + assert.equal((await pending).status, 403); + assert.equal( + ( + await updatesRequest( + new Request(`${origin}/api/updates`, { headers: { cookie } }), + ) + ).status, + 401, + ); + } finally { + store.close(); + if (previous === undefined) delete process.env.ROOST_DATA_DIR; + else process.env.ROOST_DATA_DIR = previous; + await rm(directory, { recursive: true, force: true }); + } +}); diff --git a/tests/update-socket.test.ts b/tests/update-socket.test.ts new file mode 100644 index 0000000..bebfe5c --- /dev/null +++ b/tests/update-socket.test.ts @@ -0,0 +1,82 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { mkdtemp, rm, stat } from "node:fs/promises"; +import { join } from "node:path"; +import { createInterface } from "node:readline"; +import { test } from "node:test"; +import { setTimeout as delay } from "node:timers/promises"; +import { updaterRequest } from "../src/updater/client"; + +test("real private Unix transport bounds requests and routes responses through the credential-checking bridge", async () => { + const root = await mkdtemp("/tmp/roost-socket-"); + const directory = join(root, "updates"); + const { mkdir } = await import("node:fs/promises"); + await mkdir(directory, { mode: 0o700 }); + const child = spawn( + "/usr/bin/python3", + ["src/updater/peer-broker.py", join(directory, "helper.sock")], + { stdio: ["pipe", "pipe", "pipe"] }, + ); + const reader = createInterface({ input: child.stdout }); + let ready!: () => void; + const started = new Promise((resolve) => { + ready = resolve; + }); + reader.on("line", (line) => { + const frame = JSON.parse(line); + if (frame.ready) { + ready(); + return; + } + const reply = () => + child.stdin.write( + `${JSON.stringify({ + id: frame.id, + result: { + value: { + seen: frame.message.action, + protocol: frame.message.protocol, + }, + }, + })}\n`, + ); + if (frame.message.action === "repair") setTimeout(reply, 21000); + else reply(); + }); + try { + await started; + assert.equal( + (await stat(join(directory, "helper.sock"))).mode & 0o777, + 0o600, + ); + assert.deepEqual(await updaterRequest(root, { action: "status" }), { + seen: "status", + protocol: 1, + }); + await assert.rejects( + updaterRequest(root, { action: "status", payload: "x".repeat(9000) }), + /too large/, + ); + // Exceeds both former 18-second broker and 20-second client limits. Status + // must still be observable while the same connection waits for repair. + const repair = updaterRequest(root, { action: "repair" }); + await delay(100); + const observed = await Promise.race([ + updaterRequest(root, { action: "status" }), + delay(5000).then(() => { + throw new Error("Status was blocked by the pending repair."); + }), + ]); + assert.deepEqual(observed, { + seen: "status", + protocol: 1, + }); + assert.deepEqual(await repair, { seen: "repair", protocol: 1 }); + } finally { + child.kill("SIGTERM"); + await once(child, "exit"); + reader.close(); + await rm(root, { recursive: true, force: true }); + } +}); diff --git a/tests/update-work.test.ts b/tests/update-work.test.ts new file mode 100644 index 0000000..60f0586 --- /dev/null +++ b/tests/update-work.test.ts @@ -0,0 +1,71 @@ +import assert from "node:assert/strict"; +import { mkdtemp, rm } from "node:fs/promises"; +import { test } from "node:test"; +import { codingWorkUncertain, systemdAdapter } from "../src/updater/systemd"; + +test("coding quiescence requires fresh verified idle identity; labels never substitute", () => { + const now = 100000; + const idle = { + status: "review", + observedWorking: 1, + lastWorkerState: "idle", + sessionIdentity: "owner", + nativeSessionId: "session", + lastCheckedAt: now, + cancelRequested: 0, + }; + assert.equal(codingWorkUncertain(idle, now), false); + assert.equal( + codingWorkUncertain({ ...idle, lastWorkerState: "done" }, now), + false, + ); + for (const patch of [ + { status: "blocked" }, + { status: "running" }, + { observedWorking: 0 }, + { lastWorkerState: "working" }, + { lastWorkerState: "missing" }, + { lastWorkerState: "unknown" }, + { lastWorkerState: null }, + { lastCheckedAt: now - 15001 }, + { lastCheckedAt: undefined }, + { sessionIdentity: null }, + { nativeSessionId: null }, + { cancelRequested: 1 }, + { error: "identity changed" }, + ]) + assert.equal( + codingWorkUncertain({ ...idle, ...patch }, now), + true, + JSON.stringify(patch), + ); + assert.equal( + codingWorkUncertain( + { status: "blocked", lastWorkerState: "not_started" }, + now, + ), + false, + ); + assert.equal( + codingWorkUncertain( + { ...idle, status: "blocked", lastWorkerState: "not_started" }, + now, + ), + true, + ); +}); + +test("unavailable work storage defers instead of granting quiescence", async () => { + const root = await mkdtemp("/tmp/roost-work-unavailable-"); + try { + const adapter = systemdAdapter( + { root, user: "test", uid: 1000, home: root, port: 12345 }, + root, + ); + assert.deepEqual(await adapter.blockers(false), [ + "Work cannot be verified because its store is unavailable.", + ]); + } finally { + await rm(root, { recursive: true, force: true }); + } +});