Initial import of verae-fleet from zapier monorepo

src/ssh.js

@@ -0,0 +1,170 @@
+/**
+ * SSH control of remote fleet workers. Uses username + host + identity file path.
+ * Never logs or stores private-key material.
+ * @module ssh
+ */
+
+import { spawn } from 'node:child_process';
+import fs from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const HERE = path.dirname(fileURLToPath(import.meta.url));
+
+export function expandHome(p) {
+  if (!p) return p;
+  if (p === '~') return os.homedir();
+  if (p.startsWith('~/')) return path.join(os.homedir(), p.slice(2));
+  return p;
+}
+
+export function shellQuote(s) {
+  return `'${String(s).replace(/'/g, `'\"'\"'`)}'`;
+}
+
+export function sshTarget(machine) {
+  return `${machine.user || 'marchon'}@${machine.host}`;
+}
+
+export function sshBaseArgs(machine) {
+  const ident = expandHome(machine.identityFile || '');
+  const args = [
+    '-n',
+    '-o', 'BatchMode=yes',
+    '-o', 'StrictHostKeyChecking=accept-new',
+    '-o', 'IdentitiesOnly=yes',
+    '-o', 'ConnectTimeout=8',
+    '-p', String(machine.sshPort || 22),
+  ];
+  if (ident) {
+    if (!fs.existsSync(ident)) throw new Error(`identityFile not found: ${ident}`);
+    args.push('-i', ident);
+  }
+  return args;
+}
+
+export function scpBaseArgs(machine) {
+  const ident = expandHome(machine.identityFile || '');
+  const args = [
+    '-o', 'BatchMode=yes',
+    '-o', 'StrictHostKeyChecking=accept-new',
+    '-o', 'IdentitiesOnly=yes',
+    '-o', 'ConnectTimeout=8',
+    '-P', String(machine.sshPort || 22),
+  ];
+  if (ident) args.push('-i', ident);
+  return args;
+}
+
+export function sshExec(machine, remoteCommand, { timeoutMs = 20000 } = {}) {
+  return new Promise((resolve, reject) => {
+    const args = [...sshBaseArgs(machine), sshTarget(machine), remoteCommand];
+    const child = spawn('ssh', args, { stdio: ['ignore', 'pipe', 'pipe'] });
+    const out = [];
+    const err = [];
+    child.stdout.on('data', (c) => out.push(c));
+    child.stderr.on('data', (c) => err.push(c));
+    const t = setTimeout(() => {
+      child.kill('SIGKILL');
+      reject(new Error(`ssh timeout: ${machine.id || machine.host}`));
+    }, timeoutMs);
+    child.on('error', (e) => {
+      clearTimeout(t);
+      reject(e);
+    });
+    child.on('close', (code) => {
+      clearTimeout(t);
+      const stdout = Buffer.concat(out).toString('utf8').trim();
+      const stderr = Buffer.concat(err).toString('utf8').trim();
+      if (code !== 0) {
+        reject(new Error(stderr || stdout || `ssh exit ${code}`));
+        return;
+      }
+      resolve({ stdout, stderr, code });
+    });
+  });
+}
+
+export function scpTo(machine, localFiles, remoteDir) {
+  return new Promise((resolve, reject) => {
+    const dest = `${sshTarget(machine)}:${remoteDir}/`;
+    const args = [...scpBaseArgs(machine), ...localFiles, dest];
+    const child = spawn('scp', args, { stdio: ['ignore', 'pipe', 'pipe'] });
+    const err = [];
+    child.stderr.on('data', (c) => err.push(c));
+    const t = setTimeout(() => {
+      child.kill('SIGKILL');
+      reject(new Error('scp timeout'));
+    }, 30000);
+    child.on('close', (code) => {
+      clearTimeout(t);
+      if (code !== 0) reject(new Error(Buffer.concat(err).toString('utf8') || `scp exit ${code}`));
+      else resolve();
+    });
+    child.on('error', (e) => {
+      clearTimeout(t);
+      reject(e);
+    });
+  });
+}
+
+const synced = new Set();
+
+export async function ensureRemoteRuntime(machine) {
+  const remoteDir = machine.remoteDir || '~/verae-fleet-runtime';
+  const key = `${machine.id}:${remoteDir}`;
+  if (synced.has(key)) return remoteDir;
+  await sshExec(machine, `mkdir -p ${remoteDir}/src ${remoteDir}/data`);
+  await scpTo(machine, [path.join(HERE, 'worker.js'), path.join(HERE, 'rtt.js')], `${remoteDir}/src`);
+  synced.add(key);
+  return remoteDir;
+}
+
+export async function sshSpawnWorker(machine, { instance, env }) {
+  const remoteDir = await ensureRemoteRuntime(machine);
+  const fleetEnv = Object.entries(env)
+    .filter(([k]) => k.startsWith('FLEET_'))
+    .map(([k, v]) => `${k}=${shellQuote(v)}`)
+    .join(' ');
+  const inner = `node src/worker.js </dev/null >data/${instance}/worker.log 2>&1 & echo $!`;
+  const cmd = `mkdir -p ${remoteDir}/data/${instance} && cd ${remoteDir} && ${fleetEnv} bash -c ${shellQuote(inner)}`;
+  const { stdout } = await sshExec(machine, cmd);
+  const pid = Number(String(stdout).split('\n').pop());
+  if (!Number.isFinite(pid)) throw new Error(`remote spawn produced no pid: ${stdout}`);
+  return { pid, remoteDir };
+}
+
+export async function sshKillWorker(machine, pid, instance) {
+  const cmd = pid
+    ? `kill -TERM ${Number(pid)} 2>/dev/null || true`
+    : `pkill -f ${shellQuote(`FLEET_INSTANCE=${instance}`)} 2>/dev/null || true`;
+  await sshExec(machine, cmd);
+}
+
+export async function sshHttp(machine, method, urlPath, { port, body } = {}) {
+  const url = `http://127.0.0.1:${port}${urlPath}`;
+  let cmd = `curl -sS -m 3 -X ${method}`;
+  if (body) {
+    cmd += ` -H 'content-type: application/json' -d ${shellQuote(JSON.stringify(body))}`;
+  }
+  cmd += ` ${shellQuote(url)}`;
+  const { stdout } = await sshExec(machine, cmd, { timeoutMs: 8000 });
+  try {
+    return JSON.parse(stdout || '{}');
+  } catch {
+    return { raw: stdout };
+  }
+}
+
+export async function sshCheck(machine) {
+  const { stdout } = await sshExec(machine, 'echo OK; whoami; hostname; command -v node; node -v');
+  const lines = stdout.split('\n');
+  return {
+    ok: lines[0] === 'OK',
+    user: lines[1],
+    hostname: lines[2],
+    node: lines[3] ? `${lines[3]} ${lines[4] || ''}`.trim() : null,
+    identityFile: expandHome(machine.identityFile || ''),
+  };
+}