From 79b36d38c80603a18848617d5adea114da300be9 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 30 Aug 2026 11:47:37 +0000 Subject: [PATCH 1/5] feat(fleet): run one command, or a package upgrade, across many servers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DiskPush moves bytes to a server. This adds the other thing you do with a list of servers: run work on all of them. diskpush fleet check --on tag:production # what does each one need diskpush fleet upgrade --on tag:production --sudo diskpush fleet run "systemctl reload nginx" --on 'web-*' --sudo diskpush fleet script ./deploy.sh --on all '!db-01' In the desktop app it is the Fleet button: tick servers, pick a recipe or type a command, watch each host report on its own. New package `packages/fleet-core`, with no Electron in it, the same way `rsync-core` has none. It takes connections and a script and produces a stream of per-host events; it opens sessions only through a `connect` function the caller supplies, so the CLI opens one connection per host and closes it while the desktop hands it a pooled session it keeps — and so the runner is testable without a network. Upgrades detect the package manager on each host, inside the script, so one command covers a fleet mixing Debian, Rocky and Alpine (apt, dnf, yum, zypper, pacman, apk, brew, pkg). Every invocation is non-interactive and keeps the installed config file where the question comes up. Nothing is removed: upgrade, never dist-upgrade or autoremove. Rebooting is off by default and the run reports which servers need one. The decisions worth knowing about: - Script text is never interpolated into a command line. It goes to the remote interpreter on stdin; the command line only names the interpreter. A quote, a backtick or a newline in a script cannot become a different command. - A selector term matching nothing is an error, not a smaller fleet. `--on web-O3` with a letter O fails rather than quietly patching eleven of twelve servers. Same for an exclusion, which was meant to protect a host. - `unreachable` is not `failed`. A server that was switched off did not run the command, and collapsing those two is how a fleet tool reports a powered-down box as a failed deploy. - A script matching a known way to lose a machine needs confirming first, the same bargain Mirror makes on the transfer side. Re-checked in the Electron main process, so a renderer that skipped the dialog cannot skip the check with it. It is a tripwire against the accident, not a sandbox. - `--sudo` uses `sudo -n`, which fails rather than hanging on a prompt nobody can see. `--sudo-password` asks once without echo and feeds `sudo -S` on stdin: never stored, never logged, never on a command line where `ps` on the server would show it. - Unknown host keys fail during a fan-out instead of prompting one at a time; `--accept-new` is opt-in per run. A changed key is still never accepted. - A run records the script it ran and the server names it ran on, not pointers to them, so editing a saved command cannot rewrite the history of what was executed on production last Tuesday. - Exit 71 when any server did not succeed. One code rather than a failing host's own status: across twelve servers there may be several, and picking one would mean inventing a winner. `--json` carries them all. Also adds `SshSession.execStream`, since `exec` buffers to completion and a fifteen-minute upgrade is only watchable if the output arrives while it happens. Verified against real SSH: streaming interleaves across hosts, per-host exit codes and timeouts are reported correctly, script bytes round-trip unmangled, and an unreachable host stays distinct from a failing one. 36 new tests; suite is 452 across 38 files. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01UeSWg1Czsb2Lwxj8vHUnA4 --- README.md | 14 + apps/cli/package.json | 1 + apps/cli/src/bin.ts | 3 + apps/cli/src/commands/fleet.ts | 831 ++++++++++++++++++ apps/cli/src/exit-codes.ts | 9 + apps/cli/src/help.ts | 38 + apps/cli/src/parse-argv.ts | 15 + apps/desktop/electron/main/index.ts | 7 +- apps/desktop/electron/main/ipc.ts | 32 + apps/desktop/electron/main/services/fleet.ts | 212 +++++ apps/desktop/electron/preload/index.ts | 16 + apps/desktop/electron/shared/contract.ts | 59 ++ .../electron/shared/fleet-contract.test.ts | 91 ++ apps/desktop/package.json | 1 + apps/desktop/src/app/page.tsx | 18 + apps/desktop/src/components/fleet-dialog.tsx | 618 +++++++++++++ apps/desktop/src/lib/api.ts | 106 ++- apps/desktop/src/lib/fleet-events.test.ts | 104 +++ apps/desktop/src/lib/fleet-events.ts | 69 ++ apps/web/lib/docs.ts | 2 + docs/architecture.md | 8 + docs/cli.md | 46 + docs/desktop.md | 37 + docs/fleet.md | 259 ++++++ docs/security.md | 22 + packages/database/src/fleet-store.test.ts | 185 ++++ packages/database/src/migrations.ts | 65 ++ packages/database/src/store.ts | 240 +++++ packages/fleet-core/package.json | 26 + packages/fleet-core/src/check.ts | 93 ++ packages/fleet-core/src/command.test.ts | 113 +++ packages/fleet-core/src/command.ts | 159 ++++ packages/fleet-core/src/guard.test.ts | 118 +++ packages/fleet-core/src/guard.ts | 172 ++++ packages/fleet-core/src/index.ts | 7 + packages/fleet-core/src/recipes.ts | 112 +++ packages/fleet-core/src/runner.test.ts | 319 +++++++ packages/fleet-core/src/runner.ts | 298 +++++++ packages/fleet-core/src/select.test.ts | 122 +++ packages/fleet-core/src/select.ts | 127 +++ packages/fleet-core/src/upgrade.test.ts | 159 ++++ packages/fleet-core/src/upgrade.ts | 269 ++++++ packages/fleet-core/tsconfig.json | 6 + packages/schemas/src/fleet.ts | 205 +++++ packages/schemas/src/index.ts | 1 + packages/ssh-core/src/session.ts | 117 ++- pnpm-lock.yaml | 21 + 47 files changed, 5549 insertions(+), 3 deletions(-) create mode 100644 apps/cli/src/commands/fleet.ts create mode 100644 apps/desktop/electron/main/services/fleet.ts create mode 100644 apps/desktop/electron/shared/fleet-contract.test.ts create mode 100644 apps/desktop/src/components/fleet-dialog.tsx create mode 100644 apps/desktop/src/lib/fleet-events.test.ts create mode 100644 apps/desktop/src/lib/fleet-events.ts create mode 100644 docs/fleet.md create mode 100644 packages/database/src/fleet-store.test.ts create mode 100644 packages/fleet-core/package.json create mode 100644 packages/fleet-core/src/check.ts create mode 100644 packages/fleet-core/src/command.test.ts create mode 100644 packages/fleet-core/src/command.ts create mode 100644 packages/fleet-core/src/guard.test.ts create mode 100644 packages/fleet-core/src/guard.ts create mode 100644 packages/fleet-core/src/index.ts create mode 100644 packages/fleet-core/src/recipes.ts create mode 100644 packages/fleet-core/src/runner.test.ts create mode 100644 packages/fleet-core/src/runner.ts create mode 100644 packages/fleet-core/src/select.test.ts create mode 100644 packages/fleet-core/src/select.ts create mode 100644 packages/fleet-core/src/upgrade.test.ts create mode 100644 packages/fleet-core/src/upgrade.ts create mode 100644 packages/fleet-core/tsconfig.json create mode 100644 packages/schemas/src/fleet.ts diff --git a/README.md b/README.md index d6cc7d7..7ebcc44 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,9 @@ diskpush ./data/ prod:/data/ -- --checksum # your own rsync flags - **Skips unchanged files.** Re-running a job moves almost nothing. - **Never deletes** destination-only files unless you explicitly enable Mirror, and Mirror always shows you the delete list first. +- **One command, many servers.** Package upgrades, a health sweep, or a script + you already have — run across a whole tagged fleet, each server reported + separately. - **No cloud account, no relay.** For a server-to-server job the payload moves directly between the two servers; DiskPush only orchestrates. @@ -89,6 +92,15 @@ diskpush ./dist/ production:/srv/app/ --dry-run # Do it diskpush ./dist/ production:/srv/app/ + +# Ask every production server what it needs +diskpush fleet check --on tag:production + +# Install it +diskpush fleet upgrade --on tag:production --sudo + +# Or run anything, anywhere +diskpush fleet run "systemctl reload nginx" --on 'web-*' --sudo ``` ## Safety @@ -123,6 +135,7 @@ to make that flag hard to trigger by accident. | [docs/direct-server-to-server.md](docs/direct-server-to-server.md) | How the no-relay guarantee is implemented | | [docs/file-browser.md](docs/file-browser.md) | Why browsing is SFTP and transfers are rsync | | [docs/profiles.md](docs/profiles.md) | Saved, repeatable directory pairs | +| [docs/fleet.md](docs/fleet.md) | Running one command, or an upgrade, across many servers | | [docs/security.md](docs/security.md) | Threat model and the decisions that follow from it | | [docs/architecture.md](docs/architecture.md) | Packages, processes and boundaries | | [docs/troubleshooting.md](docs/troubleshooting.md) | What the errors mean | @@ -137,6 +150,7 @@ apps/web diskpush.com packages/schemas typed option model shared by every surface packages/rsync-core the transfer engine, with no Electron in it packages/ssh-core SSH sessions, SFTP browsing, host keys, preflight +packages/fleet-core one command across many servers, with no Electron in it packages/database the local store shared by desktop and CLI ``` diff --git a/apps/cli/package.json b/apps/cli/package.json index 9b56331..deb325a 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -17,6 +17,7 @@ }, "dependencies": { "@diskpush/database": "workspace:*", + "@diskpush/fleet-core": "workspace:*", "@diskpush/rsync-core": "workspace:*", "@diskpush/schemas": "workspace:*", "@diskpush/ssh-core": "workspace:*", diff --git a/apps/cli/src/bin.ts b/apps/cli/src/bin.ts index 15fa4b4..c23a600 100644 --- a/apps/cli/src/bin.ts +++ b/apps/cli/src/bin.ts @@ -3,6 +3,7 @@ import { DiskPushStore } from '@diskpush/database' import { EndpointParseError } from '@diskpush/rsync-core' import { ZodError } from 'zod' import { runConnections } from './commands/connections.js' +import { runFleetCommand } from './commands/fleet.js' import { runJob, runJobs, runRetry } from './commands/jobs.js' import { runDoctor, runUninstall, runUpdate } from './commands/self.js' import { runDesktop } from './commands/desktop.js' @@ -94,6 +95,8 @@ async function main(argv: readonly string[]): Promise { return await runDesktop(parsed, output) case 'tui': return await runTui(parsed, store, output) + case 'fleet': + return await runFleetCommand(parsed, store, output) case 'ls': return await runLs(parsed, store, output) default: diff --git a/apps/cli/src/commands/fleet.ts b/apps/cli/src/commands/fleet.ts new file mode 100644 index 0000000..969cb98 --- /dev/null +++ b/apps/cli/src/commands/fleet.ts @@ -0,0 +1,831 @@ +import { randomUUID } from 'node:crypto' +import { readFileSync } from 'node:fs' +import { createInterface } from 'node:readline/promises' +import { knownHostsPath, type DiskPushStore } from '@diskpush/database' +import { + BUILTIN_RECIPES, + buildUpgradeScript, + checkFleet, + copyRecipe, + describeHazards, + inspectScript, + needsAttention, + runFleet, + selectConnections, + SelectionError, + type SudoMode, +} from '@diskpush/fleet-core' +import { + FLEET_DEFAULT_CONCURRENCY, + FLEET_DEFAULT_TIMEOUT_SECONDS, + FleetInterpreterSchema, + type Connection, + type FleetHostResult, + type FleetInterpreter, + type HostUpdateReport, +} from '@diskpush/schemas' +import { SshSession } from '@diskpush/ssh-core' +import { EXIT } from '../exit-codes.js' +import { formatDuration, table } from '../format.js' +import { failure, type Output } from '../output.js' +import { ArgvError, flagValue, flagValues, hasFlag, numberFlag, type ParsedArgv } from '../parse-argv.js' +import { sshConfigHosts } from '../resolve.js' + +/** + * `diskpush fleet` — one command, many servers. + * + * The transfer side of DiskPush moves bytes to a server. This moves work to a + * set of them, and it holds itself to the same bargain the transfer side + * does: show exactly what will run, on exactly which hosts, before running + * it, and never report success on behalf of a host that did not report it. + */ + +export async function runFleetCommand(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + try { + return await dispatch(parsed, store, output) + } catch (error) { + // A bad selector is a configuration mistake, not a crash, and under + // --json it has to come back as JSON like every other failure rather than + // as a bare line on stderr from the top-level handler. + if (error instanceof SelectionError) return failure(output, error.message, EXIT.configuration) + if (error instanceof ArgvError) return failure(output, error.message, EXIT.usage) + throw error + } +} + +async function dispatch(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const subcommand = parsed.positionals[0] ?? 'help' + + switch (subcommand) { + case 'run': + case 'exec': + return fleetRun(parsed, store, output) + case 'script': + return fleetScript(parsed, store, output) + case 'upgrade': + return fleetUpgrade(parsed, store, output) + case 'check': + case 'status': + return fleetCheck(parsed, store, output) + case 'servers': + case 'targets': + return fleetServers(parsed, store, output) + case 'commands': + return fleetCommands(parsed, store, output) + case 'runs': + return fleetRuns(parsed, store, output) + case 'show': + return fleetShow(parsed, store, output) + default: + return failure( + output, + `Unknown subcommand ${JSON.stringify(subcommand)}. ` + + 'Try: run, script, upgrade, check, servers, commands, runs, show.', + EXIT.usage, + ) + } +} + +// --- selecting the fleet --------------------------------------------------- + +/** + * Everything a fleet command may run on. + * + * Saved connections and `~/.ssh/config` hosts, with saved winning on a name + * clash. Including ssh_config matters: most people already have their servers + * in that file, and making them re-enter twenty hosts before the first fleet + * command is how a feature goes unused. + */ +async function availableConnections(store: DiskPushStore): Promise { + const saved = await store.listConnections() + const savedNames = new Set(saved.map((connection) => connection.name)) + return [...saved, ...sshConfigHosts().filter((host) => !savedNames.has(host.name))] +} + +type Targets = { connections: Connection[]; selector: string[] } + +async function resolveTargets(parsed: ParsedArgv, store: DiskPushStore, fallback: readonly string[] = []): Promise { + const selector = flagValues(parsed, '--on') + const terms = selector.length > 0 ? selector : [...fallback] + + if (terms.length === 0) { + throw new ArgvError( + 'No servers selected. Add --on with a name, a glob, tag:NAME, or all.\n' + + 'Run `diskpush fleet servers` to see what is available.', + ) + } + + const available = await availableConnections(store) + const selection = selectConnections(available, terms) + + if (selection.unmatched.length > 0) { + // A typo'd host is not a smaller fleet. Refusing here is the difference + // between "upgraded 11 of 12" and "upgraded 11, silently skipped one". + throw new SelectionError( + `No server matches ${selection.unmatched.map((term) => JSON.stringify(term)).join(', ')}. ` + + 'Run `diskpush fleet servers` to see the names and tags DiskPush knows about.', + ) + } + if (selection.matched.length === 0) { + throw new SelectionError('That selector matched no servers after exclusions.') + } + + return { connections: selection.matched, selector: terms } +} + +// --- prompts --------------------------------------------------------------- + +/** Reads a line without echoing it. Used for a sudo password and nothing else. */ +async function readSecret(promptText: string): Promise { + if (!process.stdin.isTTY) throw new ArgvError('A sudo password can only be asked for on a terminal.') + + const rl = createInterface({ input: process.stdin, output: process.stderr, terminal: true }) + const asAny = rl as unknown as { output: NodeJS.WriteStream; _writeToOutput?: (text: string) => void } + asAny._writeToOutput = (text: string) => { + // Echo the prompt itself, then nothing: the password must not be visible + // and must not survive in the terminal's scrollback. + if (text.includes(promptText)) asAny.output.write(promptText) + } + try { + const value = await rl.question(promptText) + process.stderr.write('\n') + return value + } finally { + rl.close() + } +} + +async function confirm(output: Output, question: string): Promise { + if (!process.stdin.isTTY) { + output.error('Refusing to continue without confirmation, and there is no terminal to ask on. Re-run with --yes.') + return false + } + const rl = createInterface({ input: process.stdin, output: process.stderr }) + try { + const answer = await rl.question(`${question} [y/N] `) + return /^y(es)?$/i.test(answer.trim()) + } finally { + rl.close() + } +} + +// --- the shared run path --------------------------------------------------- + +type FleetInvocation = { + label: string + script: string + interpreter: FleetInterpreter + sudo: boolean + workingDirectory: string | null + timeoutSeconds: number + commandId: string | null + targetFallback: readonly string[] + /** + * The caller already asked about this script's hazards in terms specific to + * what it does. `fleet upgrade --reboot` names the servers it will restart, + * which is a better question than "this contains `shutdown`, continue?" — + * and asking both makes the second one furniture. + */ + hazardsAcknowledged?: boolean +} + +function invocationFromFlags(parsed: ParsedArgv): Pick { + const interpreter = flagValue(parsed, '--interpreter') + return { + interpreter: interpreter ? FleetInterpreterSchema.parse(interpreter) : 'sh', + workingDirectory: flagValue(parsed, '--cwd') ?? null, + timeoutSeconds: numberFlag(parsed, '--timeout') ?? FLEET_DEFAULT_TIMEOUT_SECONDS, + } +} + +/** `--env KEY=VALUE`, repeatable. */ +function envFromFlags(parsed: ParsedArgv): Record { + const env: Record = {} + for (const entry of flagValues(parsed, '--env')) { + const equals = entry.indexOf('=') + if (equals <= 0) throw new ArgvError(`--env takes KEY=VALUE, got ${JSON.stringify(entry)}.`) + env[entry.slice(0, equals)] = entry.slice(equals + 1) + } + return env +} + +/** + * The one place a fleet command actually runs. + * + * Shared by `run`, `script` and `upgrade` so that the confirmation, the + * hazard check, the live output, the summary and the recorded history are + * identical whichever door you came in through. + */ +async function execute( + invocation: FleetInvocation, + parsed: ParsedArgv, + store: DiskPushStore, + output: Output, +): Promise { + const targets = await resolveTargets(parsed, store, invocation.targetFallback) + const concurrency = numberFlag(parsed, '--concurrency') ?? FLEET_DEFAULT_CONCURRENCY + const onFailure = hasFlag(parsed, '--stop-on-error') ? 'stop' : 'continue' + const env = envFromFlags(parsed) + const assumeYes = hasFlag(parsed, '--yes') + + const sudo: SudoMode = !invocation.sudo ? 'off' : hasFlag(parsed, '--sudo-password') ? 'password' : 'non-interactive' + + // `--print-command` is a pipeline: `diskpush fleet upgrade --print-command + // > upgrade.sh` has to produce a script and nothing else, so it returns + // before any of the framing below. + if (hasFlag(parsed, '--print-command')) { + process.stdout.write(`${invocation.script}\n`) + return EXIT.ok + } + + // --- what is about to happen, before anything happens -------------------- + // + // On stderr, not stdout: the command's own output is the result of this + // command, and `diskpush fleet run "cat /etc/hostname" --on all | sort` + // must not have a three-line preamble in the middle of it. + output.warn(`Command: ${invocation.label}`) + output.warn( + `Servers: ${targets.connections.length} (${targets.connections.map((connection) => connection.name).join(', ')})`, + ) + output.warn( + `Running: ${concurrency} at a time, ${invocation.timeoutSeconds}s timeout each${sudo === 'off' ? '' : ', via sudo'}\n`, + ) + + if (hasFlag(parsed, '--dry-run')) { + if (output.isJson) { + output.json({ + status: 'ok', + dryRun: true, + script: invocation.script, + servers: targets.connections.map((connection) => ({ + id: connection.id, + name: connection.name, + host: connection.host, + })), + }) + } else { + output.line('--dry-run: nothing was run. The script above would go to each of those servers.') + } + return EXIT.ok + } + + // A script that matches a known way to lose a machine does not fan out + // until someone says so out loud. + const hazards = invocation.hazardsAcknowledged ? [] : inspectScript(invocation.script) + if (hazards.length > 0) { + output.warn(`This command matches ${hazards.length === 1 ? 'a pattern' : 'patterns'} that can destroy a server:\n`) + for (const line of describeHazards(hazards)) output.warn(` ${line}\n`) + output.warn(`It would run on ${targets.connections.length} server(s).\n`) + if (!assumeYes && !(await confirm(output, 'Run it anyway?'))) { + return failure(output, 'Cancelled.', EXIT.refused) + } + } + + const sudoPassword = sudo === 'password' ? await readSecret('sudo password: ') : undefined + + // --- record the run before it starts ------------------------------------- + + const runId = randomUUID() + await store.createFleetRun({ + id: runId, + commandId: invocation.commandId, + label: invocation.label, + script: invocation.script, + interpreter: invocation.interpreter, + sudo: invocation.sudo, + workingDirectory: invocation.workingDirectory, + timeoutSeconds: invocation.timeoutSeconds, + concurrency, + onFailure, + targetSelector: targets.selector, + state: 'running', + hostsTotal: targets.connections.length, + hostsSucceeded: 0, + hostsFailed: 0, + completedAt: null, + }) + + // Ctrl-C stops the run rather than killing the process mid-write, so the + // per-host results already collected are still recorded. + const controller = new AbortController() + const onInterrupt = () => { + output.warn('\nStopping. Servers already running are being signalled; the rest are cancelled.') + controller.abort() + } + process.once('SIGINT', onInterrupt) + + const width = Math.max(...targets.connections.map((connection) => connection.name.length)) + const live = !output.isJson && !hasFlag(parsed, '--quiet') + + try { + const run = await runFleet({ + connections: targets.connections, + script: invocation.script, + interpreter: invocation.interpreter, + sudo, + sudoPassword, + workingDirectory: invocation.workingDirectory, + ...(Object.keys(env).length > 0 ? { env } : {}), + timeoutSeconds: invocation.timeoutSeconds, + concurrency, + onFailure, + // `sh -e` is the default. `--no-fail-fast` is for a probe script whose + // commands are expected to fail as part of doing their job. + failFast: !hasFlag(parsed, '--no-fail-fast'), + runId, + signal: controller.signal, + connect: (connection) => SshSession.connect(connection, sessionOptions(parsed)), + // The CLI opens one connection per host and owns it. The desktop pools + // sessions and deliberately does not. + release: (session) => session.close(), + onEvent: (event) => { + if (!live) return + const name = (id: string) => + targets.connections.find((connection) => connection.id === id)?.name ?? id + if (event.type === 'host-stdout') output.line(`${name(event.connectionId).padEnd(width)} | ${event.line}`) + if (event.type === 'host-stderr') output.line(`${name(event.connectionId).padEnd(width)} ! ${event.line}`) + }, + }) + + for (const result of run.results) await store.saveFleetHostResult(result) + await store.completeFleetRun(runId, { + state: run.state, + hostsSucceeded: run.succeeded, + hostsFailed: run.failed, + }) + + if (output.isJson) { + output.json({ + status: run.state === 'completed' ? 'ok' : 'failed', + runId, + state: run.state, + succeeded: run.succeeded, + failed: run.failed, + skipped: run.skipped, + hosts: run.results, + }) + } else { + output.line() + output.line(resultTable(run.results)) + output.line() + output.line( + `${run.succeeded} succeeded, ${run.failed} failed` + + `${run.skipped > 0 ? `, ${run.skipped} not run` : ''}. Run ${runId.slice(0, 8)}.`, + ) + if (run.failed > 0) output.line(`Full output: diskpush fleet show ${runId.slice(0, 8)}`) + } + + return run.failed > 0 || run.skipped > 0 ? EXIT.fleetIncomplete : EXIT.ok + } finally { + process.off('SIGINT', onInterrupt) + } +} + +function sessionOptions(parsed: ParsedArgv) { + return { + knownHostsPath: knownHostsPath(), + onUnknownHostKey: async (details: { host: string; keyType: string; fingerprint: string }) => { + // A fan-out is exactly the wrong moment to be answering host key + // prompts one at a time, so an unknown host fails unless --accept-new + // was passed deliberately for this run. + if (hasFlag(parsed, '--accept-new')) return true + throw new Error( + `${details.host} is not in known_hosts (${details.keyType} ${details.fingerprint}). ` + + 'Run `diskpush connections test NAME` to check it once, or pass --accept-new.', + ) + }, + } +} + +function resultTable(results: readonly FleetHostResult[]): string { + return table( + results.map((result) => [ + result.connectionName, + result.host, + result.state, + result.exitCode === null ? '-' : String(result.exitCode), + result.durationMs === null ? '-' : formatDuration(result.durationMs / 1000), + (result.errorSummary ?? '').slice(0, 60), + ]), + ['SERVER', 'HOST', 'STATE', 'EXIT', 'TIME', 'NOTE'], + ) +} + +// --- subcommands ----------------------------------------------------------- + +async function fleetRun(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const savedName = flagValue(parsed, '--command') + const inline = parsed.positionals.slice(1).join(' ') + + if (savedName) { + const command = await store.findFleetCommand(savedName, BUILTIN_RECIPES) + if (!command) { + return failure( + output, + `No saved command or recipe named ${JSON.stringify(savedName)}. Run \`diskpush fleet commands\`.`, + EXIT.configuration, + ) + } + return execute( + { + label: command.name, + script: command.script, + interpreter: command.interpreter, + sudo: command.sudo || hasFlag(parsed, '--sudo'), + workingDirectory: flagValue(parsed, '--cwd') ?? command.workingDirectory, + timeoutSeconds: numberFlag(parsed, '--timeout') ?? command.timeoutSeconds, + commandId: command.builtin ? null : command.id, + targetFallback: command.targets, + }, + parsed, + store, + output, + ) + } + + if (!inline) { + return failure( + output, + 'Usage: diskpush fleet run "COMMAND" --on SELECTOR\n' + + ' or: diskpush fleet run --command NAME --on SELECTOR\n' + + ' or: diskpush fleet script FILE --on SELECTOR', + EXIT.usage, + ) + } + + const flags = invocationFromFlags(parsed) + return execute( + { + label: inline.length > 60 ? `${inline.slice(0, 57)}...` : inline, + script: inline, + // A one-liner typed at a shell prompt should behave like one; `sh -es` + // around `uptime` buys nothing and surprises anyone who pipes. + interpreter: flagValue(parsed, '--interpreter') ? flags.interpreter : 'raw', + sudo: hasFlag(parsed, '--sudo') || hasFlag(parsed, '--sudo-password'), + workingDirectory: flags.workingDirectory, + timeoutSeconds: flags.timeoutSeconds, + commandId: null, + targetFallback: [], + }, + parsed, + store, + output, + ) +} + +async function fleetScript(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const path = parsed.positionals[1] ?? flagValue(parsed, '--script') + if (!path) return failure(output, 'Usage: diskpush fleet script FILE --on SELECTOR', EXIT.usage) + + let script: string + try { + script = readFileSync(path, 'utf8') + } catch (error) { + return failure(output, `Could not read ${path}: ${(error as Error).message}`, EXIT.configuration) + } + if (script.trim().length === 0) return failure(output, `${path} is empty.`, EXIT.usage) + + const flags = invocationFromFlags(parsed) + return execute( + { + label: path, + script, + // A `#!/bin/bash` line means the author has already said which shell + // this needs, and honouring it is cheaper than making them repeat it. + interpreter: flagValue(parsed, '--interpreter') + ? flags.interpreter + : /^#!.*\bbash\b/.test(script) + ? 'bash' + : 'sh', + sudo: hasFlag(parsed, '--sudo') || hasFlag(parsed, '--sudo-password'), + workingDirectory: flags.workingDirectory, + timeoutSeconds: flags.timeoutSeconds, + commandId: null, + targetFallback: [], + }, + parsed, + store, + output, + ) +} + +async function fleetUpgrade(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const rebootFlag = flagValue(parsed, '--reboot') + const reboot = + rebootFlag === 'always' + ? 'always' + : hasFlag(parsed, '--reboot') + ? 'if-needed' + : 'never' + + // Neither of those inspects anything, so neither may prompt: `--print-command` + // piped into a file must not stop on a question nobody is there to answer. + const inspecting = hasFlag(parsed, '--print-command') || hasFlag(parsed, '--dry-run') + + if (reboot !== 'never' && !inspecting && !hasFlag(parsed, '--yes')) { + const targets = await resolveTargets(parsed, store) + output.warn( + `--reboot will restart ${targets.connections.length} server(s) ` + + `${reboot === 'always' ? 'whether or not they need it' : 'that report needing one'}.`, + ) + if (!(await confirm(output, 'Continue?'))) return failure(output, 'Cancelled.', EXIT.refused) + } + + return execute( + { + label: `upgrade packages${reboot === 'never' ? '' : ` (reboot: ${reboot})`}`, + script: buildUpgradeScript({ reboot }), + interpreter: 'sh', + // Installing packages needs root everywhere. --no-sudo is there for the + // fleet that already connects as root. + sudo: !hasFlag(parsed, '--no-sudo'), + workingDirectory: null, + timeoutSeconds: numberFlag(parsed, '--timeout') ?? 3600, + commandId: null, + targetFallback: [], + // Reaching here with a reboot policy means the question above was + // already answered, by name, for these servers. + hazardsAcknowledged: reboot !== 'never', + }, + parsed, + store, + output, + ) +} + +async function fleetCheck(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const targets = await resolveTargets(parsed, store) + const concurrency = numberFlag(parsed, '--concurrency') ?? FLEET_DEFAULT_CONCURRENCY + + if (!output.isJson) output.line(`Checking ${targets.connections.length} server(s)...\n`) + + const { reports, unreachable } = await checkFleet({ + connections: targets.connections, + concurrency, + ...(numberFlag(parsed, '--timeout') !== undefined ? { timeoutSeconds: numberFlag(parsed, '--timeout')! } : {}), + connect: (connection) => SshSession.connect(connection, sessionOptions(parsed)), + release: (session) => session.close(), + }) + + const shown = hasFlag(parsed, '--pending') ? needsAttention(reports) : reports + + if (output.isJson) { + output.json({ status: unreachable > 0 ? 'partial' : 'ok', reports: shown }) + return unreachable > 0 ? EXIT.fleetIncomplete : EXIT.ok + } + + if (shown.length === 0) { + output.line('Every server is up to date and none is waiting on a reboot.') + return EXIT.ok + } + + output.line(checkTable(shown)) + + // Why a host could not be reached goes below the table, not in a column: + // "Timed out while waiting for handshake" does not fit next to an uptime, + // and squeezing it in there is how it ends up looking like one. + const unreachableReports = shown.filter((report) => !report.reachable) + if (unreachableReports.length > 0) { + output.line() + for (const report of unreachableReports) { + output.line(`${report.connectionName}: ${report.error ?? 'did not answer'}`) + } + } + + const pending = reports.filter((report) => (report.updates ?? 0) > 0).length + const rebooting = reports.filter((report) => report.rebootRequired === true).length + output.line() + output.line( + [ + `${pending} server(s) with updates`, + `${rebooting} waiting on a reboot`, + unreachable > 0 ? `${unreachable} unreachable` : null, + ] + .filter(Boolean) + .join(', ') + '.', + ) + if (pending > 0) output.line('Install them with: diskpush fleet upgrade --on ' + (flagValues(parsed, '--on')[0] ?? 'all')) + + return unreachable > 0 ? EXIT.fleetIncomplete : EXIT.ok +} + +function checkTable(reports: readonly HostUpdateReport[]): string { + return table( + reports.map((report) => [ + report.connectionName, + report.reachable ? (report.os ?? 'unknown') : 'unreachable', + report.packageManager, + report.updates === null ? '?' : String(report.updates), + report.securityUpdates === null ? '-' : String(report.securityUpdates), + report.rebootRequired === null ? '?' : report.rebootRequired ? 'YES' : 'no', + report.diskUsedPercent === null ? '-' : `${report.diskUsedPercent}%`, + report.uptimeSeconds === null ? '-' : formatUptime(report.uptimeSeconds), + ]), + ['SERVER', 'OS', 'PM', 'UPD', 'SEC', 'REBOOT', 'DISK', 'UPTIME'], + ) +} + +function formatUptime(seconds: number): string { + const days = Math.floor(seconds / 86400) + if (days > 0) return `${days}d` + return `${Math.floor(seconds / 3600)}h` +} + +async function fleetServers(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const available = await availableConnections(store) + const selector = flagValues(parsed, '--on') + const shown = selector.length > 0 ? selectConnections(available, selector).matched : available + + if (output.isJson) { + output.json({ status: 'ok', servers: shown }) + return EXIT.ok + } + if (shown.length === 0) { + output.line('No servers. Add one with `diskpush connections add NAME user@host`, or import ~/.ssh/config.') + return EXIT.ok + } + + const savedIds = new Set((await store.listConnections()).map((connection) => connection.id)) + output.line( + table( + shown.map((connection) => [ + connection.name, + `${connection.username}@${connection.host}:${connection.port}`, + connection.tags.join(',') || '-', + savedIds.has(connection.id) ? 'saved' : 'ssh_config', + ]), + ['SERVER', 'TARGET', 'TAGS', 'FROM'], + ), + ) + return EXIT.ok +} + +async function fleetCommands(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const action = parsed.positionals[1] ?? 'list' + + if (action === 'list') { + const commands = await store.listFleetCommands(BUILTIN_RECIPES) + if (output.isJson) { + output.json({ status: 'ok', commands }) + return EXIT.ok + } + output.line( + table( + commands.map((command) => [ + command.name, + command.builtin ? 'built-in' : 'saved', + command.sudo ? 'sudo' : '-', + command.targets.join(',') || '-', + command.description.slice(0, 56), + ]), + ['NAME', 'SOURCE', 'ROOT', 'DEFAULT TARGETS', 'DESCRIPTION'], + ), + ) + return EXIT.ok + } + + if (action === 'show') { + const name = parsed.positionals[2] + if (!name) return failure(output, 'Usage: diskpush fleet commands show NAME', EXIT.usage) + const command = await store.findFleetCommand(name, BUILTIN_RECIPES) + if (!command) return failure(output, `No command named ${JSON.stringify(name)}.`, EXIT.configuration) + if (output.isJson) output.json({ status: 'ok', command }) + else { + output.line(`${command.name}${command.builtin ? ' (built-in)' : ''}`) + if (command.description) output.line(command.description) + output.line() + output.line(command.script) + } + return EXIT.ok + } + + if (action === 'save') { + const name = parsed.positionals[2] + const scriptPath = flagValue(parsed, '--script') + const inline = parsed.positionals.slice(3).join(' ') + if (!name || (!scriptPath && !inline)) { + return failure(output, 'Usage: diskpush fleet commands save NAME --script FILE | "COMMAND"', EXIT.usage) + } + + let script = inline + if (scriptPath) { + try { + script = readFileSync(scriptPath, 'utf8') + } catch (error) { + return failure(output, `Could not read ${scriptPath}: ${(error as Error).message}`, EXIT.configuration) + } + } + + const flags = invocationFromFlags(parsed) + const saved = await store.saveFleetCommand({ + name, + description: flagValue(parsed, '--description') ?? '', + script, + interpreter: flags.interpreter, + sudo: hasFlag(parsed, '--sudo'), + workingDirectory: flags.workingDirectory, + timeoutSeconds: flags.timeoutSeconds, + targets: flagValues(parsed, '--on'), + tags: flagValues(parsed, '--tag'), + }) + if (output.isJson) output.json({ status: 'ok', command: saved }) + else output.line(`Saved command ${saved.name}. Run it with: diskpush fleet run --command ${saved.name} --on SELECTOR`) + return EXIT.ok + } + + if (action === 'copy') { + const source = parsed.positionals[2] + const target = parsed.positionals[3] ?? flagValue(parsed, '--as') + if (!source || !target) return failure(output, 'Usage: diskpush fleet commands copy NAME NEW-NAME', EXIT.usage) + const command = await store.findFleetCommand(source, BUILTIN_RECIPES) + if (!command) return failure(output, `No command named ${JSON.stringify(source)}.`, EXIT.configuration) + + const saved = await store.saveFleetCommand(copyRecipe(command, target)) + if (output.isJson) output.json({ status: 'ok', command: saved }) + else output.line(`Copied ${command.name} to ${saved.name}. It is yours to edit now.`) + return EXIT.ok + } + + if (action === 'remove' || action === 'rm') { + const name = parsed.positionals[2] + if (!name) return failure(output, 'Usage: diskpush fleet commands remove NAME', EXIT.usage) + const removed = await store.deleteFleetCommand(name) + if (!removed) { + const builtin = BUILTIN_RECIPES.find((recipe) => recipe.name === name) + return failure( + output, + builtin + ? `${name} is a built-in recipe and cannot be removed. Copy it instead: diskpush fleet commands copy ${name} my-${name}` + : `No saved command named ${JSON.stringify(name)}.`, + EXIT.configuration, + ) + } + if (output.isJson) output.json({ status: 'ok', removed: name }) + else output.line(`Removed command ${name}.`) + return EXIT.ok + } + + return failure(output, `Unknown action ${JSON.stringify(action)}. Try: list, show, save, copy, remove.`, EXIT.usage) +} + +async function fleetRuns(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const runs = await store.listFleetRuns(numberFlag(parsed, '--limit') ?? 25) + if (output.isJson) { + output.json({ status: 'ok', runs }) + return EXIT.ok + } + if (runs.length === 0) { + output.line('No fleet runs yet.') + return EXIT.ok + } + output.line( + table( + runs.map((run) => [ + run.id.slice(0, 8), + run.createdAt.slice(0, 19).replace('T', ' '), + run.state, + `${run.hostsSucceeded}/${run.hostsTotal}`, + run.label.slice(0, 52), + ]), + ['ID', 'WHEN', 'STATE', 'OK', 'COMMAND'], + ), + ) + return EXIT.ok +} + +async function fleetShow(parsed: ParsedArgv, store: DiskPushStore, output: Output): Promise { + const id = parsed.positionals[1] + if (!id) return failure(output, 'Usage: diskpush fleet show RUN-ID', EXIT.usage) + + const run = await store.findFleetRun(id) + if (!run) return failure(output, `No fleet run matching ${JSON.stringify(id)}.`, EXIT.configuration) + const hosts = await store.listFleetRunHosts(run.id) + + if (output.isJson) { + output.json({ status: 'ok', run, hosts }) + return EXIT.ok + } + + output.line(`Run ${run.id}`) + output.line(`${run.label} · ${run.state} · ${run.createdAt.slice(0, 19).replace('T', ' ')}`) + output.line(`Selector: ${run.targetSelector.join(' ') || '-'}`) + output.line() + output.line(resultTable(hosts)) + + // Full output only for the hosts that need explaining. Printing every + // successful `apt upgrade` transcript is how this becomes unreadable. + const failed = hosts.filter((host) => host.state !== 'succeeded') + const wanted = hasFlag(parsed, '--all') ? hosts : failed + for (const host of wanted) { + output.line() + output.line(`--- ${host.connectionName} (${host.host}) ${host.state} ---`) + if (host.stdout.trim()) output.line(host.stdout.trimEnd()) + if (host.stderr.trim()) output.line(host.stderr.trimEnd()) + } + if (failed.length === 0 && !hasFlag(parsed, '--all')) { + output.line() + output.line('Every server succeeded. Pass --all to print their output too.') + } + + return EXIT.ok +} diff --git a/apps/cli/src/exit-codes.ts b/apps/cli/src/exit-codes.ts index 1e02e40..7037a96 100644 --- a/apps/cli/src/exit-codes.ts +++ b/apps/cli/src/exit-codes.ts @@ -17,6 +17,15 @@ export const EXIT = { unavailable: 69, /** DiskPush itself broke. */ internal: 70, + /** + * A fleet run did not succeed everywhere. + * + * Deliberately one code rather than the failing host's own exit status: + * across twelve servers there may be several, and picking one to pass + * through would mean inventing a winner. `--json` carries every host's real + * code for anything that needs them. + */ + fleetIncomplete: 71, } as const export type ExitCode = (typeof EXIT)[keyof typeof EXIT] | number diff --git a/apps/cli/src/help.ts b/apps/cli/src/help.ts index b7f38c4..6b28cdc 100644 --- a/apps/cli/src/help.ts +++ b/apps/cli/src/help.ts @@ -31,6 +31,14 @@ COMMANDS desktop launch the desktop app tui [SRC] [DST] the two-pane browser, in this terminal + fleet run "CMD" run one command on every selected server + fleet script FILE run a local script file on every selected server + fleet upgrade install pending package updates, per host + fleet check what each server needs: updates, reboot, disk + fleet servers what --on can select, with tags + fleet commands saved commands and built-in recipes + fleet runs / show ID fleet history, and one run's full output + ls [ENDPOINT] list a local or remote directory over SFTP connections list saved connections connections add NAME [user@]host @@ -78,6 +86,30 @@ OPTIONS --progress rsync's own per-file progress instead of the aggregate line DiskPush draws --stats include rsync's transfer statistics +FLEET OPTIONS + --on SELECTOR which servers (repeatable, or comma-separated): + all every server DiskPush knows + web-01 one, by name + web-* a glob over names + tag:production every server with that tag + host:10.0.0.* a glob over hostnames + !web-03 remove one from the set + --concurrency N how many servers at once (default 4) + --sudo run through \`sudo -n\`, which fails rather than + hangs when a password is wanted + --sudo-password ask for a sudo password once, use it everywhere; + never written to disk + --timeout SECONDS per-server deadline (default 900, upgrade 3600) + --stop-on-error do not start further servers after one fails + --accept-new trust an unknown host key instead of failing + --interpreter NAME sh | bash | raw + --cwd PATH cd here on the server first + --env KEY=VALUE set an environment variable (repeatable) + --print-command print the script and exit + --reboot[=always] upgrade only: reboot hosts that need one, or all + --no-sudo upgrade only: you already connect as root + --pending check only: hide servers with nothing to do + -y, --yes confirm a mirror's deletions without prompting --non-interactive never prompt; fail instead --json machine-readable result on stdout @@ -94,6 +126,7 @@ EXIT CODES 66 DiskPush declined (unconfirmed mirror, blocked argument) 69 precondition failed (rsync missing, host unreachable) 70 internal error + 71 a fleet run did not succeed on every server EXAMPLES diskpush ./dist/ production:/srv/app/ @@ -101,4 +134,9 @@ EXAMPLES diskpush media-01:/srv/media/ backup-02:/data/media/ diskpush mirror ./site/ prod:/var/www/site/ --exclude-preset node diskpush sync ./data/ prod:/data/ -- -aHAX --checksum + + diskpush fleet check --on tag:production + diskpush fleet upgrade --on tag:production --sudo --concurrency 2 + diskpush fleet run "systemctl reload nginx" --on web-* --sudo + diskpush fleet script ./rotate-keys.sh --on all '!db-01' ` diff --git a/apps/cli/src/parse-argv.ts b/apps/cli/src/parse-argv.ts index 6405455..0e1bd29 100644 --- a/apps/cli/src/parse-argv.ts +++ b/apps/cli/src/parse-argv.ts @@ -48,6 +48,20 @@ export const VALUE_FLAGS = new Set([ '--key', '--path', '--jump', + // fleet + '--on', + '--concurrency', + '--command', + '--script', + '--interpreter', + '--cwd', + '--env', + '--description', + '--tag', + '--as', + // `--reboot` is deliberately absent, for the same reason `--compress` is: + // its value is optional, so `--reboot --on web-01` must not read `--on` as + // the reboot policy. `--reboot=always` uses the attached form. ]) /** Single-letter aliases. */ @@ -84,6 +98,7 @@ const KNOWN_COMMANDS = new Set([ 'doctor', 'desktop', 'tui', + 'fleet', 'help', 'version', ]) diff --git a/apps/desktop/electron/main/index.ts b/apps/desktop/electron/main/index.ts index 79f7300..f4d61f3 100644 --- a/apps/desktop/electron/main/index.ts +++ b/apps/desktop/electron/main/index.ts @@ -9,6 +9,7 @@ import { contentSecurityPolicy, inlineScriptHashes } from './csp.js' import { registerIpc } from './ipc.js' import { checkForUpdates } from './services/updater.js' import { closeAllSessions } from './services/sessions.js' +import { cancelAllFleetRuns, hasActiveFleetRun } from './services/fleet.js' import { cancelAll, hasActiveTransfer } from './services/transfers.js' // ESM has no __dirname. import.meta.dirname exists in Electron 33's Node 20.18, @@ -164,7 +165,10 @@ app.whenReady().then(() => { createWindow() // Not awaited: a slow or unreachable GitHub must not delay the window. - void checkForUpdates(hasActiveTransfer) + // A fleet upgrade counts as busy too: restarting the app under an + // `apt upgrade` running on eight servers would drop every live connection + // mid-transaction. + void checkForUpdates(() => hasActiveTransfer() || hasActiveFleetRun()) app.on('activate', () => { if (BrowserWindow.getAllWindows().length === 0) createWindow() @@ -179,5 +183,6 @@ app.on('before-quit', () => { // Stopping with SIGINT leaves rsync's partial files intact, so anything in // flight is resumable rather than lost. cancelAll() + cancelAllFleetRuns() closeAllSessions() }) diff --git a/apps/desktop/electron/main/ipc.ts b/apps/desktop/electron/main/ipc.ts index 2db69b8..12d440a 100644 --- a/apps/desktop/electron/main/ipc.ts +++ b/apps/desktop/electron/main/ipc.ts @@ -9,6 +9,9 @@ import { CreateEntryRequestSchema, DeleteEntryRequestSchema, ExternalUrlSchema, + FleetCheckRequestSchema, + FleetRequestSchema, + FleetRunIdSchema, IPC, JobIdSchema, PathSchema, @@ -18,6 +21,15 @@ import { type IpcResult, } from '../shared/contract.js' import { requireConnection } from './services/connections.js' +import { + cancelFleet, + checkFleetServers, + fleetCommands, + fleetRunDetail, + fleetServers, + previewFleet, + startFleet, +} from './services/fleet.js' import { browserFor, dropSession, sessionFor } from './services/sessions.js' import { store } from './services/store.js' import { cancelTransfer, previewTransfer, startTransfer } from './services/transfers.js' @@ -293,6 +305,26 @@ export function registerIpc(): void { handle(IPC.profilesRemove, z.object({ id: z.string().min(1) }), async ({ id }) => (await store()).deleteProfile(id)) + // --- fleet --------------------------------------------------------------- + + handle(IPC.fleetServers, z.undefined(), async () => fleetServers()) + + handle(IPC.fleetCommands, z.undefined(), async () => fleetCommands()) + + handle(IPC.fleetPreview, FleetRequestSchema, async (request) => previewFleet(request)) + + handle(IPC.fleetStart, FleetRequestSchema, async (request, event) => startFleet(request, event.sender)) + + handle(IPC.fleetCancel, z.object({ runId: FleetRunIdSchema }), async ({ runId }) => cancelFleet(runId)) + + handle(IPC.fleetCheck, FleetCheckRequestSchema, async (input) => checkFleetServers(input)) + + handle(IPC.fleetRuns, z.object({ limit: z.number().int().min(1).max(200).default(25) }), async ({ limit }) => + (await store()).listFleetRuns(limit), + ) + + handle(IPC.fleetRunDetail, z.object({ runId: FleetRunIdSchema }), async ({ runId }) => fleetRunDetail(runId)) + // --- shell --------------------------------------------------------------- handle(IPC.shellOpenExternal, z.object({ url: ExternalUrlSchema }), async ({ url }) => { diff --git a/apps/desktop/electron/main/services/fleet.ts b/apps/desktop/electron/main/services/fleet.ts new file mode 100644 index 0000000..68a6627 --- /dev/null +++ b/apps/desktop/electron/main/services/fleet.ts @@ -0,0 +1,212 @@ +import { randomUUID } from 'node:crypto' +import type { WebContents } from 'electron' +import { + BUILTIN_RECIPES, + checkFleet, + inspectScript, + runFleet, + type Hazard, + type SudoMode, +} from '@diskpush/fleet-core' +import type { Connection, FleetCommand, FleetHostResult, HostUpdateReport } from '@diskpush/schemas' +import { sshConfigConnections } from '@diskpush/ssh-core' +import { IPC, type FleetRequest } from '../../shared/contract.js' +import { dropSession, sessionFor } from './sessions.js' +import { store } from './store.js' + +/** + * Fleet operations for the desktop. + * + * The renderer sends connection ids and script text. Everything that turns + * those into something a server executes — the interpreter, the sudo mode, + * the host lookup, the hazard check — happens here, so the dialog is a view + * of the operation rather than the thing that defines it. + */ + +type RunningFleet = { runId: string; controller: AbortController } +const running = new Map() + +/** Saved connections plus ~/.ssh/config hosts, saved winning a name clash. */ +export async function fleetServers(): Promise { + const saved = await (await store()).listConnections() + const savedNames = new Set(saved.map((connection) => connection.name)) + return [...saved, ...sshConfigConnections().filter((host) => !savedNames.has(host.name))] +} + +export async function fleetCommands(): Promise { + return (await store()).listFleetCommands(BUILTIN_RECIPES) +} + +/** + * Resolves ids to real connections. + * + * An id the renderer no longer has a server for is an error rather than a + * silently smaller fleet — the same rule the CLI applies to a typo'd + * selector, and for the same reason. + */ +async function connectionsFor(ids: readonly string[]): Promise { + const available = await fleetServers() + const byId = new Map(available.map((connection) => [connection.id, connection])) + const resolved: Connection[] = [] + for (const id of ids) { + const connection = byId.get(id) + if (!connection) throw new Error('One of the selected servers no longer exists. Close and reopen Fleet.') + resolved.push(connection) + } + return resolved +} + +export type FleetPreview = { + servers: { id: string; name: string; host: string }[] + hazards: Hazard[] + /** The readable form of what will run. Not the runnable one. */ + command: string +} + +/** What the dialog shows before anything runs. */ +export async function previewFleet(request: FleetRequest): Promise { + const connections = await connectionsFor(request.connectionIds) + return { + servers: connections.map((connection) => ({ + id: connection.id, + name: connection.name, + host: connection.host, + })), + hazards: inspectScript(request.script), + command: request.script, + } +} + +export type StartedFleetRun = { runId: string; hosts: { connectionId: string; connectionName: string }[] } + +export async function startFleet(request: FleetRequest, sender: WebContents): Promise { + const connections = await connectionsFor(request.connectionIds) + + // Re-checked here, not taken on the renderer's word. The dialog showing a + // confirmation is what makes this true; a renderer that skipped the dialog + // must not be able to skip the check with it. + const hazards = inspectScript(request.script) + if (hazards.length > 0 && !request.hazardsConfirmed) { + throw new Error( + `This command matches ${hazards.length} destructive pattern(s) and was not confirmed: ` + + hazards.map((hazard) => `line ${hazard.lineNumber}, ${hazard.explanation}`).join('; '), + ) + } + + const sudo: SudoMode = !request.sudo ? 'off' : request.sudoPassword ? 'password' : 'non-interactive' + const runId = randomUUID() + const db = await store() + + await db.createFleetRun({ + id: runId, + commandId: request.commandId, + label: request.label, + script: request.script, + interpreter: request.interpreter, + sudo: request.sudo, + workingDirectory: request.workingDirectory, + timeoutSeconds: request.timeoutSeconds, + concurrency: request.concurrency, + onFailure: request.onFailure, + targetSelector: connections.map((connection) => connection.name), + state: 'running', + hostsTotal: connections.length, + hostsSucceeded: 0, + hostsFailed: 0, + completedAt: null, + }) + + const controller = new AbortController() + running.set(runId, { runId, controller }) + + void (async () => { + try { + const run = await runFleet({ + connections, + script: request.script, + interpreter: request.interpreter, + sudo, + ...(request.sudoPassword ? { sudoPassword: request.sudoPassword } : {}), + workingDirectory: request.workingDirectory, + timeoutSeconds: request.timeoutSeconds, + concurrency: request.concurrency, + onFailure: request.onFailure, + runId, + signal: controller.signal, + connect: (connection) => sessionFor(connection), + // No release: the desktop pools sessions across browsing and + // transfers, so closing one here would shut a file pane's connection + // out from under it. + onEvent: (event) => { + // The window can go away mid-run. The run carries on and its + // outcome is still recorded, the same as a transfer. + if (!sender.isDestroyed()) sender.send(IPC.eventFleet, { runId, event }) + }, + }) + + for (const result of run.results) await db.saveFleetHostResult(result) + await db.completeFleetRun(runId, { + state: run.state, + hostsSucceeded: run.succeeded, + hostsFailed: run.failed, + }) + } catch (error) { + await db.completeFleetRun(runId, { state: 'failed', hostsSucceeded: 0, hostsFailed: connections.length }) + if (!sender.isDestroyed()) { + sender.send(IPC.eventFleet, { + runId, + event: { type: 'run-error', message: error instanceof Error ? error.message : String(error) }, + }) + } + } finally { + running.delete(runId) + // A command that changed sshd, the login shell or a key would leave a + // pooled session pointing at a server that no longer works the way the + // session assumes. Cheaper to reconnect than to debug that later. + if (request.sudo) for (const connection of connections) dropSession(connection.id) + } + })() + + return { + runId, + hosts: connections.map((connection) => ({ connectionId: connection.id, connectionName: connection.name })), + } +} + +export function cancelFleet(runId: string): boolean { + const run = running.get(runId) + if (!run) return false + run.controller.abort() + return true +} + +export async function checkFleetServers(input: { + connectionIds: readonly string[] + concurrency: number + timeoutSeconds: number +}): Promise { + const connections = await connectionsFor(input.connectionIds) + const { reports } = await checkFleet({ + connections, + concurrency: input.concurrency, + timeoutSeconds: input.timeoutSeconds, + connect: (connection) => sessionFor(connection), + }) + return reports +} + +export async function fleetRunDetail(runId: string): Promise<{ run: unknown; hosts: FleetHostResult[] } | null> { + const db = await store() + const run = await db.findFleetRun(runId) + if (!run) return null + return { run, hosts: await db.listFleetRunHosts(run.id) } +} + +/** True while any fleet command is in flight; the updater defers a restart on it. */ +export function hasActiveFleetRun(): boolean { + return running.size > 0 +} + +export function cancelAllFleetRuns(): void { + for (const run of running.values()) run.controller.abort() +} diff --git a/apps/desktop/electron/preload/index.ts b/apps/desktop/electron/preload/index.ts index d6ac72c..d9fe464 100644 --- a/apps/desktop/electron/preload/index.ts +++ b/apps/desktop/electron/preload/index.ts @@ -49,6 +49,17 @@ const api = { list: () => call(IPC.profilesList), remove: (id: string) => call(IPC.profilesRemove, { id }), }, + fleet: { + servers: () => call(IPC.fleetServers), + commands: () => call(IPC.fleetCommands), + preview: (request: unknown) => call(IPC.fleetPreview, request), + start: (request: unknown) => call(IPC.fleetStart, request), + cancel: (runId: string) => call(IPC.fleetCancel, { runId }), + check: (connectionIds: string[], concurrency = 4, timeoutSeconds = 180) => + call(IPC.fleetCheck, { connectionIds, concurrency, timeoutSeconds }), + runs: (limit = 25) => call(IPC.fleetRuns, { limit }), + runDetail: (runId: string) => call(IPC.fleetRunDetail, { runId }), + }, shell: { openExternal: (url: string) => call(IPC.shellOpenExternal, { url }), }, @@ -63,6 +74,11 @@ const api = { ipcRenderer.on(IPC.eventTransfer, wrapped) return () => ipcRenderer.off(IPC.eventTransfer, wrapped) }, + onFleet(listener: (payload: { runId: string; event: unknown }) => void): () => void { + const wrapped = (_event: unknown, payload: { runId: string; event: unknown }) => listener(payload) + ipcRenderer.on(IPC.eventFleet, wrapped) + return () => ipcRenderer.off(IPC.eventFleet, wrapped) + }, }, } diff --git a/apps/desktop/electron/shared/contract.ts b/apps/desktop/electron/shared/contract.ts index ee1546a..a4b4279 100644 --- a/apps/desktop/electron/shared/contract.ts +++ b/apps/desktop/electron/shared/contract.ts @@ -38,10 +38,21 @@ export const IPC = { profilesSave: 'profiles:save', profilesRemove: 'profiles:remove', + fleetServers: 'fleet:servers', + fleetCommands: 'fleet:commands', + fleetPreview: 'fleet:preview', + fleetStart: 'fleet:start', + fleetCancel: 'fleet:cancel', + fleetCheck: 'fleet:check', + fleetRuns: 'fleet:runs', + fleetRunDetail: 'fleet:run-detail', + shellOpenExternal: 'shell:open-external', /** Main -> renderer, one channel carrying every job event. */ eventTransfer: 'event:transfer', + /** Main -> renderer, one channel carrying every fleet event. */ + eventFleet: 'event:fleet', } as const /** A path the renderer asked for. Length-capped, and never joined by the renderer. */ @@ -170,6 +181,54 @@ export const DeleteEntryRequestSchema = z.object({ isDirectory: z.boolean(), }) +/** + * A fleet request. + * + * The renderer names connections by id and supplies script text; it does not + * assemble a command line, choose an interpreter binary, or supply a remote + * shell. The main process turns those into an invocation — the same one the + * CLI builds — so a compromised renderer's worst case is a script running + * where the user already has a shell, not a command line of its own design. + */ +export const FleetRequestSchema = z.object({ + connectionIds: z.array(ConnectionIdSchema).min(1).max(500), + script: z.string().min(1).max(256 * 1024), + interpreter: z.enum(['sh', 'bash', 'raw']).default('sh'), + sudo: z.boolean().default(false), + /** + * Held in memory for one run and written to `sudo -S` on stdin. + * + * Never stored, never logged, and never echoed back to the renderer. The + * cap is here so a renderer cannot use this field to push megabytes through + * the boundary. + */ + sudoPassword: z.string().max(1024).optional(), + workingDirectory: PathSchema.nullable().default(null), + timeoutSeconds: z.number().int().min(1).max(86400).default(900), + concurrency: z.number().int().min(1).max(64).default(4), + onFailure: z.enum(['continue', 'stop']).default('continue'), + /** + * The user saw the hazard list and chose to continue. + * + * Checked again in the main process rather than trusted from the dialog: + * `false` here means a script matching a destructive pattern is refused, + * whatever the renderer believes it showed. + */ + hazardsConfirmed: z.boolean().default(false), + /** A saved command this came from, recorded with the run. */ + commandId: z.string().min(1).max(128).nullable().default(null), + label: z.string().min(1).max(200), +}) +export type FleetRequest = z.infer + +export const FleetCheckRequestSchema = z.object({ + connectionIds: z.array(ConnectionIdSchema).min(1).max(500), + concurrency: z.number().int().min(1).max(64).default(4), + timeoutSeconds: z.number().int().min(1).max(3600).default(180), +}) + +export const FleetRunIdSchema = z.string().min(1).max(128) + /** Only http(s) may be handed to the system browser. */ export const ExternalUrlSchema = z.string().url().refine((value) => /^https?:\/\//i.test(value), { message: 'Only http and https URLs can be opened externally.', diff --git a/apps/desktop/electron/shared/fleet-contract.test.ts b/apps/desktop/electron/shared/fleet-contract.test.ts new file mode 100644 index 0000000..8fbb1b2 --- /dev/null +++ b/apps/desktop/electron/shared/fleet-contract.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, it } from 'vitest' +import { FleetCheckRequestSchema, FleetRequestSchema } from './contract.js' + +/** + * The fleet boundary carries script text to servers, which makes it the most + * consequential thing the renderer can reach. These assert what it cannot + * express. + */ + +const base = { + connectionIds: ['c1'], + script: 'uptime', + label: 'uptime', +} + +describe('FleetRequestSchema', () => { + it('names servers by id, never by host', () => { + const parsed = FleetRequestSchema.parse(base) + expect(parsed.connectionIds).toEqual(['c1']) + // No host, user, port or key field exists here: those come from the + // stored connection, so a renderer cannot invent a server to reach. + expect(Object.keys(parsed)).not.toContain('host') + expect(Object.keys(parsed)).not.toContain('username') + }) + + it('refuses a run with no servers', () => { + expect(FleetRequestSchema.safeParse({ ...base, connectionIds: [] }).success).toBe(false) + }) + + it('refuses an empty script', () => { + expect(FleetRequestSchema.safeParse({ ...base, script: '' }).success).toBe(false) + }) + + it('allows only the two shells and raw, not an arbitrary binary', () => { + expect(FleetRequestSchema.safeParse({ ...base, interpreter: 'sh' }).success).toBe(true) + expect(FleetRequestSchema.safeParse({ ...base, interpreter: 'bash' }).success).toBe(true) + expect(FleetRequestSchema.safeParse({ ...base, interpreter: 'raw' }).success).toBe(true) + expect(FleetRequestSchema.safeParse({ ...base, interpreter: '/usr/bin/python3' }).success).toBe(false) + expect(FleetRequestSchema.safeParse({ ...base, interpreter: 'sh -c evil' }).success).toBe(false) + }) + + it('defaults hazardsConfirmed to false, so silence is never consent', () => { + expect(FleetRequestSchema.parse(base).hazardsConfirmed).toBe(false) + }) + + it('defaults to a modest concurrency and caps it', () => { + expect(FleetRequestSchema.parse(base).concurrency).toBe(4) + expect(FleetRequestSchema.safeParse({ ...base, concurrency: 5000 }).success).toBe(false) + expect(FleetRequestSchema.safeParse({ ...base, concurrency: 0 }).success).toBe(false) + }) + + it('bounds the timeout rather than letting a run hold a connection forever', () => { + expect(FleetRequestSchema.parse(base).timeoutSeconds).toBe(900) + expect(FleetRequestSchema.safeParse({ ...base, timeoutSeconds: 999999 }).success).toBe(false) + }) + + it('caps the fleet size and the script length', () => { + const many = Array.from({ length: 501 }, (_, index) => `c${index}`) + expect(FleetRequestSchema.safeParse({ ...base, connectionIds: many }).success).toBe(false) + expect(FleetRequestSchema.safeParse({ ...base, script: 'x'.repeat(300 * 1024) }).success).toBe(false) + }) + + it('caps the sudo password so the field cannot be used as a data channel', () => { + expect(FleetRequestSchema.safeParse({ ...base, sudo: true, sudoPassword: 'x'.repeat(2000) }).success).toBe(false) + expect(FleetRequestSchema.safeParse({ ...base, sudo: true, sudoPassword: 'hunter2' }).success).toBe(true) + }) + + it('leaves the sudo password absent when none was given, rather than empty', () => { + expect(FleetRequestSchema.parse(base).sudoPassword).toBeUndefined() + }) + + it('is not root unless it says so', () => { + expect(FleetRequestSchema.parse(base).sudo).toBe(false) + }) + + it('continues through failures unless asked to stop', () => { + expect(FleetRequestSchema.parse(base).onFailure).toBe('continue') + expect(FleetRequestSchema.safeParse({ ...base, onFailure: 'panic' }).success).toBe(false) + }) +}) + +describe('FleetCheckRequestSchema', () => { + it('has a short default timeout, because it only reads state', () => { + expect(FleetCheckRequestSchema.parse({ connectionIds: ['c1'] }).timeoutSeconds).toBe(180) + }) + + it('takes no script at all: a status sweep cannot be turned into a command', () => { + const parsed = FleetCheckRequestSchema.parse({ connectionIds: ['c1'], script: 'rm -rf /' } as never) + expect(Object.keys(parsed)).toEqual(['connectionIds', 'concurrency', 'timeoutSeconds']) + }) +}) diff --git a/apps/desktop/package.json b/apps/desktop/package.json index 3f489ab..e7cfe5b 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -16,6 +16,7 @@ "dependencies": { "@base-ui/react": "^1.7.0", "@diskpush/database": "workspace:*", + "@diskpush/fleet-core": "workspace:*", "@diskpush/rsync-core": "workspace:*", "@diskpush/schemas": "workspace:*", "@diskpush/ssh-core": "workspace:*", diff --git a/apps/desktop/src/app/page.tsx b/apps/desktop/src/app/page.tsx index dda8de6..2366b3f 100644 --- a/apps/desktop/src/app/page.tsx +++ b/apps/desktop/src/app/page.tsx @@ -9,11 +9,13 @@ import { FileDown, MonitorOff, Plus, + Server, Settings, Users, X, } from 'lucide-react' import { ConnectionDialog } from '@/components/connection-dialog' +import { FleetDialog } from '@/components/fleet-dialog' import { endpointLabel, loadPane, Pane, type PaneEndpoint, type PaneState } from '@/components/pane' import { TransferRail } from '@/components/transfer-rail' import { MirrorPreviewDialog, TransferBand, type ActiveJob } from '@/components/transfer-panel' @@ -60,6 +62,7 @@ export default function Workspace() { const [job, setJob] = useState(null) const [error, setError] = useState(null) const [showConnection, setShowConnection] = useState(false) + const [showFleet, setShowFleet] = useState(false) const [outsideShell, setOutsideShell] = useState(false) const refreshConnections = useCallback(async () => { @@ -276,6 +279,19 @@ export default function Workspace() { )}
+ {/* + Fleet sits beside "New server" rather than inside the menu: it is + the other half of what this app does with a list of servers, and + a feature nobody can find is a feature nobody has. + */} + +
+ + + {error ? ( +
+ + {error} +
+ ) : null} + +
+ {/* --- servers ------------------------------------------------- */} + + + {/* --- command and results ------------------------------------- */} +
+
+
+ Recipes: + {commands.map((command) => ( + + ))} +
+ +