import crypto from "node:crypto"; import fs from "node:fs/promises"; import { createServer } from "node:net"; import path from "node:path"; import { resolveStateDir } from "../config/paths.js"; import { backupFleetCell, restoreFleetCell } from "./backup.runtime.js"; import { allocateHostPort, buildCellEnvironment, cellAuthSecretDir, cellContainerName, cellDataDir, cellNetworkName, cellOwnerId, DEFAULT_FLEET_IMAGE, FLEET_ATTEMPT_LABEL, FLEET_OWNER_LABEL, FLEET_TENANT_LABEL, parseEnvAssignments, validateCellContainerProfile, validateFleetImage, validateDiskSize, validateTenantId, type CellContainerProfile, type FleetContainerRuntimeName, } from "./cell-profile.js"; import { createFleetContainerRuntime, type FleetContainerInspectResult, type FleetContainerRuntime, } from "./containers.runtime.js"; import { runFleetDoctor } from "./doctor.runtime.js"; import { deleteFleetCell, getFleetCell, listFleetCells, reserveFleetCell, updateFleetCellImage, } from "./registry.js"; import { assertCurrentReservation, assertManagedInspection, assertManagedNetwork, buildProfileBaseFromInspection, cleanupFailedCreateContainer, cleanupFailedCreateNetwork, detectHostSelinux, inspectionState, prepareCellConfig, prepareCellDirectories, probeCellHealth, readHostIdentity, requireInspectedAttemptId, requireInspectedGatewayToken, requireCell, resolveContainerUser, resolvePurgeTarget, restorePreviousCell, withFleetCellOperation, verifyReplacementHealthy, } from "./service-support.runtime.js"; const OFFICIAL_IMAGE_UID = 1_000; const OFFICIAL_IMAGE_GID = 1_000; // Mirrors the compose healthcheck contract: an upgrade commits only after /healthz // answers. The deadline bounds how long a broken image can hold the cell before // restore without rolling back slow-booting cells prematurely. const CELL_VERIFY_TIMEOUT_MS = 60_000; const CELL_VERIFY_POLL_MS = 1_000; async function probeLoopbackPort(port: number): Promise { return await new Promise((resolve) => { const server = createServer(); server.once("error", (error: NodeJS.ErrnoException) => { // The probe exists only to catch the one legible failure early (address in // use). Anything else - e.g. EACCES on a privileged port an unprivileged CLI // cannot bind but a rootful daemon can - defers to the authoritative runtime bind. resolve(error.code !== "EADDRINUSE"); }); server.listen(port, "127.0.0.1", () => { server.close(() => resolve(true)); }); }); } export type FleetCreateOptions = { tenant: string; image?: string; runtime?: FleetContainerRuntimeName; port?: number; memory?: string; cpus?: string; pidsLimit?: number; disk?: string; network?: "bridge" | "internal"; env?: string[]; gatewayToken?: string; start?: boolean; }; type FleetCreateResult = { tenant: string; containerName: string; port: number; image: string; runtime: FleetContainerRuntimeName; started: boolean; token: string; tokenNote: string; url: string; nextStep: string; }; type FleetListEntry = { tenant: string; state: string; port: number; image: string; created: string; }; export type FleetHealthResult = | { status: "ok"; url: string; httpStatus: number } | { status: "failed"; url: string; error: string; httpStatus?: number } | { status: "skipped"; url: string; reason: string }; type FleetStatusResult = { tenant: string; containerName: string; runtime: FleetContainerRuntimeName; port: number; image: string; created: string; dataDir: string; container: { imageId?: string } & ( | { state: string; running: boolean; managed: boolean } | { state: "missing"; running: false; managed: false } | { state: "unknown"; running: false; managed: false; error: string } ); health: FleetHealthResult; }; export type FleetLifecycleAction = "start" | "stop" | "restart"; export type FleetLogsOptions = { tenant: string; follow?: boolean; timestamps?: boolean; tail?: number; since?: string; }; type FleetActionResult = { tenant: string; action: FleetLifecycleAction | "upgrade" | "rm"; image?: string; dataPurged?: boolean; }; type FleetServiceOptions = { env?: NodeJS.ProcessEnv; containers?: FleetContainerRuntime; fetch?: typeof fetch; now?: () => number; generateToken?: () => string; generateAttemptId?: () => string; getuid?: () => number | undefined; getgid?: () => number | undefined; sleep?: (ms: number) => Promise; probePort?: (port: number) => Promise; selinuxEnabled?: () => Promise; updateImage?: typeof updateFleetCellImage; }; export function createFleetService(options: FleetServiceOptions = {}) { const env = options.env ?? process.env; const containers = options.containers ?? createFleetContainerRuntime(); const fetchImpl = options.fetch ?? fetch; const now = options.now ?? Date.now; const { generateToken = () => crypto.randomBytes(16).toString("hex") } = options; const generateAttemptId = options.generateAttemptId ?? (() => crypto.randomBytes(16).toString("hex")); const getuid = options.getuid ?? (() => process.getuid?.()); const getgid = options.getgid ?? (() => process.getgid?.()); const sleep = options.sleep ?? ((ms: number) => new Promise((resolve) => { setTimeout(resolve, ms); })); const selinuxEnabled = options.selinuxEnabled ?? detectHostSelinux; const updateImage = options.updateImage ?? updateFleetCellImage; const probePort = options.probePort ?? probeLoopbackPort; return { async create(createOptions: FleetCreateOptions): Promise { const tenantId = validateTenantId(createOptions.tenant); const image = validateFleetImage(createOptions.image ?? DEFAULT_FLEET_IMAGE); const runtime = createOptions.runtime ?? "docker"; const network = createOptions.network ?? "bridge"; const diskSize = createOptions.disk === undefined ? undefined : validateDiskSize(createOptions.disk); const { gatewayToken } = createOptions; if (gatewayToken !== undefined && !gatewayToken.trim()) { throw new Error("Gateway token must not be empty."); } const token = gatewayToken ?? generateToken(); const userEnvironment = parseEnvAssignments(createOptions.env ?? []); const environment = buildCellEnvironment(token, userEnvironment); const userEnvironmentKeys = Object.keys(userEnvironment); const attemptId = generateAttemptId(); await containers.assertLocal(runtime); if (network === "internal" && runtime === "docker") { throw new Error( "Docker cannot publish loopback ports for containers on --internal networks, so the cell Gateway's 127.0.0.1 port would be unreachable and health-gated operations would always fail. Use --runtime podman for internal cells, or keep the default bridge network and enforce Docker egress policy with host firewall rules (DOCKER-USER chain).", ); } return await withFleetCellOperation({ env, tenantId, operationName: "create", operation: async (checkpoint) => { checkpoint(); const stateDir = resolveStateDir(env); const usedPorts = new Set(listFleetCells(env).map((cell) => cell.hostPort)); const reservation = { tenantId, createdAtMs: now(), image, runtime, containerName: cellContainerName(tenantId), dataDir: cellDataDir(stateDir, tenantId), }; let record: ReturnType | undefined; if (createOptions.port !== undefined) { const candidatePort = allocateHostPort(usedPorts, createOptions.port); if (!(await probePort(candidatePort))) { throw new Error( `Host port ${candidatePort} is already in use on 127.0.0.1 by another process.`, ); } // The probe is best-effort UX; the runtime bind remains authoritative across this TOCTOU gap. record = reserveFleetCell(env, { ...reservation, requestedPort: candidatePort }); } else { const unavailablePorts = new Set(usedPorts); // The exclusion set only grows, so this terminates: allocateHostPort throws // its range-exhaustion error once every port through 65535 is excluded. while (!record) { for (const cell of listFleetCells(env)) { unavailablePorts.add(cell.hostPort); } const candidate = allocateHostPort(unavailablePorts); if (!(await probePort(candidate))) { unavailablePorts.add(candidate); continue; } try { // The probe is best-effort UX; the runtime bind remains authoritative across this TOCTOU gap. record = reserveFleetCell(env, { ...reservation, requestedPort: candidate }); } catch (error) { if (getFleetCell(env, tenantId)) { throw error; } const candidateWasReserved = listFleetCells(env).some( (cell) => cell.hostPort === candidate, ); if (!candidateWasReserved) { throw error; } unavailablePorts.add(candidate); } } } let result: FleetCreateResult; let networkAttempted = false; let containerAttempted = false; try { const authSecretDir = cellAuthSecretDir(stateDir, tenantId); const hostIdentity = readHostIdentity(getuid, getgid); const containerUser = await resolveContainerUser({ runtime, containers, hostIdentity, }); const imageOwner = hostIdentity?.uid === 0 && !containerUser ? { uid: OFFICIAL_IMAGE_UID, gid: OFFICIAL_IMAGE_GID } : undefined; const profile: CellContainerProfile = { tenantId, containerName: record.containerName, networkName: cellNetworkName(tenantId), image, runtime, hostPort: record.hostPort, dataDir: record.dataDir, authSecretDir, ownerId: cellOwnerId(record.dataDir), attemptId, memory: createOptions.memory ?? "2g", cpus: createOptions.cpus ?? "2", ...(diskSize ? { diskSize } : {}), pidsLimit: createOptions.pidsLimit ?? 512, environment, userEnvironmentKeys, ...(containerUser ? { containerUser } : {}), selinuxRelabel: await selinuxEnabled(), }; validateCellContainerProfile(profile); checkpoint(); await prepareCellDirectories(record, authSecretDir, imageOwner); assertCurrentReservation(env, record); const started = createOptions.start !== false; networkAttempted = true; checkpoint(); await containers.createNetwork( runtime, profile.networkName, { [FLEET_TENANT_LABEL]: tenantId, [FLEET_OWNER_LABEL]: profile.ownerId, [FLEET_ATTEMPT_LABEL]: attemptId, }, { internal: network === "internal" }, ); assertCurrentReservation(env, record); containerAttempted = true; checkpoint(); await containers.run(profile, false); assertCurrentReservation(env, record); checkpoint(); await prepareCellConfig(record, imageOwner); assertCurrentReservation(env, record); if (started) { checkpoint(); await containers.start(runtime, record.containerName); assertCurrentReservation(env, record); } const url = `http://127.0.0.1:${record.hostPort}`; result = { tenant: tenantId, containerName: record.containerName, port: record.hostPort, image, runtime, started, token, tokenNote: "Shown once. Store this Gateway token securely.", url, nextStep: `Open ${url}, then configure per-tenant channel accounts inside the cell.`, }; } catch (error) { let releaseReservation = true; try { if (containerAttempted) { releaseReservation = await cleanupFailedCreateContainer( record, containers, attemptId, checkpoint, ); } if (releaseReservation && networkAttempted) { releaseReservation = await cleanupFailedCreateNetwork( record, containers, attemptId, checkpoint, ); } } catch { releaseReservation = false; } if (releaseReservation) { try { checkpoint(); deleteFleetCell(env, tenantId); } catch { // Preserve the provisioning error; a stale reservation remains recoverable via fleet list/rm. } } if ( diskSize && error instanceof Error && /storage[ -]?opt|pquota|backingfs/iu.test(error.message) ) { throw new Error( `Fleet cannot enforce --disk on this container storage backend: ${error.message}. --disk requires Docker overlay2 on XFS with the pquota mount option (or btrfs/zfs storage drivers), or Podman overlay storage on XFS. Retry without --disk or move container storage to a supported filesystem.`, { cause: error }, ); } throw error; } if (result.started) { try { await verifyReplacementHealthy({ containers, record, attemptId, fetchImpl, now, sleep, checkpoint, timeoutMs: CELL_VERIFY_TIMEOUT_MS, pollMs: CELL_VERIFY_POLL_MS, context: "create", }); } catch (error) { // Unlike upgrade/restore, create has no previous container to bring back; // keep the sick cell (container + registry row) as diagnosable evidence. throw new Error( `Fleet cell ${tenantId} was created but did not become healthy within 60s; inspect it with \`openclaw fleet status ${tenantId}\` or \`openclaw fleet logs ${tenantId}\`, or remove it with \`openclaw fleet rm ${tenantId} --force\`.`, { cause: error }, ); } } return result; }, }); }, async list(): Promise { const records = listFleetCells(env); const localityChecks = new Map>(); const inspections = await Promise.all( records.map(async (record) => { try { let locality = localityChecks.get(record.runtime); if (!locality) { locality = containers.assertLocal(record.runtime); localityChecks.set(record.runtime, locality); } await locality; return await containers.inspect(record.runtime, record.containerName); } catch (error) { return { kind: "unavailable" as const, state: "unknown" as const, error: error instanceof Error ? error.message : String(error), }; } }), ); return records.map((record, index) => ({ tenant: record.tenantId, state: inspectionState( record, inspections[index] ?? { kind: "unavailable", state: "unknown", error: "inspect result missing", }, ), port: record.hostPort, image: record.image, created: new Date(record.createdAtMs).toISOString(), })); }, async status(tenant: string): Promise { const record = requireCell(env, tenant); let inspection: FleetContainerInspectResult; try { await containers.assertLocal(record.runtime); inspection = await containers.inspect(record.runtime, record.containerName); } catch (error) { inspection = { kind: "unavailable" as const, state: "unknown" as const, error: error instanceof Error ? error.message : String(error), }; } const url = `http://127.0.0.1:${record.hostPort}/healthz`; let container: FleetStatusResult["container"]; let health: FleetHealthResult; if (inspection.kind === "ok") { const managed = inspection.labels[FLEET_TENANT_LABEL] === record.tenantId && inspection.labels[FLEET_OWNER_LABEL] === cellOwnerId(record.dataDir); container = { state: managed ? inspection.state : "unknown", running: inspection.running, managed, ...(managed ? { imageId: inspection.imageId } : {}), }; health = managed && inspection.running ? await probeCellHealth({ port: record.hostPort, fetchImpl }) : { status: "skipped", url, reason: managed ? "container is not running" : "fleet ownership mismatch", }; } else if (inspection.kind === "missing") { container = { state: "missing", running: false, managed: false }; health = { status: "skipped", url, reason: "container is missing" }; } else { container = { state: "unknown", running: false, managed: false, error: inspection.error }; health = { status: "skipped", url, reason: "container runtime unavailable" }; } return { tenant: record.tenantId, containerName: record.containerName, runtime: record.runtime, port: record.hostPort, image: record.image, created: new Date(record.createdAtMs).toISOString(), dataDir: record.dataDir, container, health, }; }, async lifecycle(tenant: string, action: FleetLifecycleAction): Promise { const tenantId = validateTenantId(tenant); await containers.assertLocal(requireCell(env, tenantId).runtime); return await withFleetCellOperation({ env, tenantId, operationName: action, operation: async (checkpoint) => { const record = requireCell(env, tenantId); await containers.assertLocal(record.runtime); const inspection = assertManagedInspection( record, await containers.inspect(record.runtime, record.containerName), ); checkpoint(); // Pin the inspected generation: the ownership guard above proved this // container, and a name can point at a different one by the time we act. await containers[action](record.runtime, inspection.containerId); return { tenant: record.tenantId, action }; }, }); }, async logs(logOptions: FleetLogsOptions): Promise { const record = requireCell(env, validateTenantId(logOptions.tenant)); await containers.assertLocal(record.runtime); // Ownership must be proven before streaming; never stream a foreign name-squatting container. const inspection = assertManagedInspection( record, await containers.inspect(record.runtime, record.containerName), ); const gatewayCredential = inspection.environment.OPENCLAW_GATEWAY_TOKEN; // Pin the inspected generation so a concurrent restore cannot redirect the stream. await containers.logs(record.runtime, inspection.containerId, { follow: logOptions.follow, timestamps: logOptions.timestamps, tail: logOptions.tail, since: logOptions.since, redactValues: gatewayCredential ? [gatewayCredential] : [], }); }, async upgrade(tenant: string, requestedImage?: string): Promise { const tenantId = validateTenantId(tenant); const explicitImage = requestedImage === undefined ? undefined : validateFleetImage(requestedImage); await containers.assertLocal(requireCell(env, tenantId).runtime); return await withFleetCellOperation({ env, tenantId, operationName: "upgrade", operation: async (checkpoint) => { const record = requireCell(env, tenantId); await containers.assertLocal(record.runtime); const inspection = assertManagedInspection( record, await containers.inspect(record.runtime, record.containerName), ); const image = explicitImage ?? validateFleetImage(record.image); // The token is intentionally absent from SQLite; capture container env before removal and replay it. const token = requireInspectedGatewayToken(inspection, "upgrade"); const containerUser = await resolveContainerUser({ runtime: record.runtime, containers, hostIdentity: readHostIdentity(getuid, getgid), user: inspection.user, }); const previousAttemptId = requireInspectedAttemptId(inspection, "upgrade"); const nextAttemptId = generateAttemptId(); const profileBase = buildProfileBaseFromInspection({ record, stateDir: resolveStateDir(env), inspection, containerUser, selinuxRelabel: await selinuxEnabled(), token, context: "upgrade", }); const oldProfile: CellContainerProfile = { ...profileBase, image: inspection.imageId, attemptId: previousAttemptId, }; const nextProfile: CellContainerProfile = { ...profileBase, image, attemptId: nextAttemptId, }; validateCellContainerProfile(oldProfile); validateCellContainerProfile(nextProfile); checkpoint(); await containers.pull(record.runtime, image); checkpoint(); assertManagedNetwork( record, await containers.inspectNetwork(record.runtime, cellNetworkName(record.tenantId)), ); try { if (inspection.running) { checkpoint(); await containers.stop(record.runtime, inspection.containerId); } checkpoint(); await containers.remove(record.runtime, inspection.containerId, false); checkpoint(); await containers.run(nextProfile, true); // `run -d` succeeds once the container launches, and a broken image can stay // "running" briefly before crashing. Commit only after the replacement answers // /healthz (the image's compose health contract): exit/restart-loop fails fast, // and the deadline restores the old cell instead of leaving a dead replacement. await verifyReplacementHealthy({ containers, record, attemptId: nextAttemptId, fetchImpl, now, sleep, checkpoint, timeoutMs: CELL_VERIFY_TIMEOUT_MS, pollMs: CELL_VERIFY_POLL_MS, context: "upgrade", }); checkpoint(); updateImage(env, record.tenantId, image); } catch (error) { try { await restorePreviousCell({ record, containers, oldProfile, previousAttemptId, nextAttemptId, wasRunning: inspection.running, checkpoint, }); } catch { throw new Error( `Fleet upgrade failed for ${record.tenantId}; the previous container could not be restored.`, { cause: error }, ); } throw new Error( `Fleet upgrade failed for ${record.tenantId}; the previous container was restored.`, { cause: error }, ); } return { tenant: record.tenantId, action: "upgrade", image }; }, }); }, async backup(params: { tenant: string; out?: string; maxBytes?: number }) { const tenantId = validateTenantId(params.tenant); return await withFleetCellOperation({ env, tenantId, operationName: "backup", operation: async (checkpoint) => { checkpoint(); const record = requireCell(env, tenantId); return await backupFleetCell({ record, stateDir: resolveStateDir(env), containers, now, checkpoint, out: params.out, maxBytes: params.maxBytes, }); }, }); }, async restore(params: { tenant: string; from: string; force?: boolean; maxBytes?: number }) { const tenantId = validateTenantId(params.tenant); return await withFleetCellOperation({ env, tenantId, operationName: "restore", operation: async (checkpoint) => { const record = requireCell(env, tenantId); return await restoreFleetCell({ record, stateDir: resolveStateDir(env), containers, fetchImpl, now, sleep, checkpoint, generateToken, generateAttemptId, hostIdentity: readHostIdentity(getuid, getgid), selinuxRelabel: await selinuxEnabled(), from: params.from, force: params.force, maxBytes: params.maxBytes, }); }, }); }, async doctor(tenant?: string) { return await runFleetDoctor({ env, containers, fetchImpl, tenant, getuid, getgid }); }, async remove(params: { tenant: string; force?: boolean; purgeData?: boolean; }): Promise { if (params.purgeData && !params.force) { throw new Error("--purge-data requires --force."); } const tenantId = validateTenantId(params.tenant); await containers.assertLocal(requireCell(env, tenantId).runtime); return await withFleetCellOperation({ env, tenantId, operationName: "rm", operation: async (checkpoint) => { const record = requireCell(env, tenantId); await containers.assertLocal(record.runtime); const stateDir = resolveStateDir(env); const authSecretDir = cellAuthSecretDir(stateDir, record.tenantId); const purgeTargets: string[] = []; if (params.purgeData) { const dataTarget = await resolvePurgeTarget( path.join(stateDir, "fleet", "cells"), record.dataDir, record.tenantId, ); if (dataTarget) { purgeTargets.push(dataTarget); } const authTarget = await resolvePurgeTarget( path.join(stateDir, "fleet", "auth-profile-secrets"), authSecretDir, record.tenantId, ); if (authTarget) { purgeTargets.push(authTarget); } } const inspection = await containers.inspect(record.runtime, record.containerName); if (inspection.kind === "unavailable") { throw new Error( `Cannot inspect ${record.runtime} container for tenant ${record.tenantId}: ${inspection.error}`, ); } const networkName = cellNetworkName(record.tenantId); const networkInspection = await containers.inspectNetwork(record.runtime, networkName); if (networkInspection.kind === "unavailable") { throw new Error( `Cannot inspect ${record.runtime} network for tenant ${record.tenantId}: ${networkInspection.error}`, ); } if (networkInspection.kind === "ok") { assertManagedNetwork(record, networkInspection); } if (inspection.kind === "ok") { const managed = assertManagedInspection(record, inspection); if (inspection.running && !params.force) { throw new Error( `Fleet cell ${record.tenantId} is running; use --force to remove it.`, ); } checkpoint(); // Pin the inspected generation: `--force --purge-data` must never // destroy a container that did not pass the ownership guard. await containers.remove(record.runtime, managed.containerId, params.force === true); } if (networkInspection.kind === "ok") { checkpoint(); await containers.removeNetwork(record.runtime, networkName); } if (purgeTargets.length > 0) { checkpoint(); await Promise.all( purgeTargets.map((target) => fs.rm(target, { recursive: true, force: true })), ); } checkpoint(); deleteFleetCell(env, record.tenantId); return { tenant: record.tenantId, action: "rm", dataPurged: params.purgeData === true, }; }, }); }, }; } /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */