stack/packages/bus/src/host-run.mjs

// Row 210 (#1707, design D5 to D7): run one sudo command on a registered host over ssh, with the
// agent's host key handed to a private ssh-agent and never written to disk, argv or env.
//
// Each call gets a 0700 dir under <configDir>/secrets/run holding known_hosts (the registry's pinned
// keys), id.pub, the agent socket and a pids file. ssh-agent and ssh each lead their own process
// group and run under `setpriv --pdeathsig KILL`, so they die with the bus child. The `finally`
// kills both groups and removes the dir. If the bus child dies first, sweepRunDir clears the dir
// at the next start.
import { spawn } from 'node:child_process';
import { appendFileSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { join } from 'node:path';
import { BusError } from './broker.mjs';

const deny = (code) => {
  throw new BusError(code);
};
const KILL_WITH_PARENT = ['/usr/bin/setpriv', '--pdeathsig', 'KILL', '--'];
const PATH = '/usr/bin:/bin';
// Linux sun_path holds 108 bytes with the NUL.
const SOCKET_MAX = 107;
const RUN_DIR = /^h-[A-Za-z0-9]{6}$/;
export const HOST_LIMITS = Object.freeze({ outputMax: 1024 * 1024, graceMs: 5000, agentWaitMs: 5000 });

// D6: each element POSIX single-quoted, so the remote shell hands sudo exactly these strings.
export const quote = (element) => "'" + element.replaceAll("'", "'\\''") + "'";
export const remoteCommand = (argv) => ['sudo', '-n', '--', ...argv.map(quote)].join(' ');

// D7 step 4: every option set here, nothing read from a config file.
export function sshArgs({ hostId, address, port, user, dir }) {
  const o = (x) => ['-o', x];
  return [
    // ssh's own messages go to the run dir's log, apart from the remote command's stderr (N3).
    '-F', '/dev/null', '-T', '-E', join(dir, 'ssh.log'), '-p', String(port),
    ...o('BatchMode=yes'), ...o('IdentitiesOnly=yes'), ...o(`IdentityFile=${join(dir, 'id.pub')}`),
    ...o(`IdentityAgent=${join(dir, 'a.sock')}`),
    ...o('StrictHostKeyChecking=yes'), ...o(`UserKnownHostsFile=${join(dir, 'known_hosts')}`),
    ...o('GlobalKnownHostsFile=/dev/null'),
    ...o(`HostKeyAlias=mosaic-host-${hostId}`), ...o('UpdateHostKeys=no'), ...o('CheckHostIP=no'), ...o('VerifyHostKeyDNS=no'),
    ...o('ForwardAgent=no'), ...o('ForwardX11=no'), ...o('ClearAllForwardings=yes'), ...o('PermitLocalCommand=no'),
    ...o('ProxyCommand=none'), ...o('ProxyJump=none'), ...o('ControlMaster=no'), ...o('ControlPath=none'),
    ...o('ConnectTimeout=10'), ...o('ServerAliveInterval=15'), ...o('ServerAliveCountMax=3'),
    '-l', user, '--', address,
  ];
}

// Field 22 of /proc/<pid>/stat, the start time in clock ticks, or null. The comm field may hold
// spaces and parentheses, so the count starts after the last ")" (field 2's end).
export function startTime(pid) {
  try {
    const stat = readFileSync(`/proc/${pid}/stat`, 'utf8');
    return stat.slice(stat.lastIndexOf(')') + 2).split(' ')[19] ?? null;
  } catch {
    return null;
  }
}
const alive = (pid, start) => start !== null && startTime(pid) === start;
const record = (dir, pid) => appendFileSync(join(dir, 'pids'), `${pid} ${startTime(pid)}\n`, { mode: 0o600 });
const group = (child, signal) => {
  if (!child?.pid) return;
  try {
    process.kill(-child.pid, signal);
  } catch {}
};
const start = (bin, args, env, stdio) =>
  spawn(KILL_WITH_PARENT[0], [...KILL_WITH_PARENT.slice(1), bin, ...args], { env, detached: true, stdio });

// The run root is the stack instance's user's own, like the credential store.
function runRoot(root) {
  mkdirSync(root, { recursive: true, mode: 0o700 });
  const st = lstatSync(root);
  if (!st.isDirectory() || st.uid !== process.getuid() || (st.mode & 0o077) !== 0) deny('credential-file');
  return root;
}

// Waits for ssh-agent's socket, or for the agent to exit first.
async function waitForSocket(agent, sock, limitMs) {
  const until = Date.now() + limitMs;
  while (!existsSync(sock)) {
    if (agent.exitCode !== null || agent.signalCode !== null || Date.now() > until) deny('host-failed');
    await new Promise((r) => setTimeout(r, 20));
  }
}

// Runs ssh-add with the key on its stdin; resolves on exit 0. A hung ssh-add is killed at
// `limitMs` (host-failed) or on abort (credential-unavailable), so a revoke or a stuck agent
// never holds the host busy (N2).
export function addKey(sock, text, lifetime, { limitMs = HOST_LIMITS.agentWaitMs, signal = null } = {}) {
  return new Promise((resolve, reject) => {
    if (signal?.aborted) return reject(new BusError('credential-unavailable'));
    const add = start('ssh-add', ['-q', '-t', String(lifetime), '-'], { PATH, SSH_AUTH_SOCK: sock }, ['pipe', 'ignore', 'ignore']);
    const stop = (code) => {
      group(add, 'SIGKILL');
      reject(new BusError(code));
    };
    const timer = setTimeout(() => stop('host-failed'), limitMs);
    const abort = () => stop('credential-unavailable');
    signal?.addEventListener('abort', abort, { once: true });
    const settle = () => {
      clearTimeout(timer);
      signal?.removeEventListener('abort', abort);
    };
    add.stdin.on('error', () => {});
    add.stdin.end(text);
    add.on('error', () => (settle(), reject(new BusError('host-failed'))));
    add.on('close', (code) => (settle(), code === 0 ? resolve() : reject(new BusError('host-failed'))));
  });
}

// ssh itself: stdin written once, stdout and stderr each kept to outputMax and drained past it.
function runSsh(args, stdin, timeoutMs, signal, limits, onSpawn) {
  return new Promise((resolve, reject) => {
    const began = Date.now();
    const ssh = start('ssh', args, { PATH }, ['pipe', 'pipe', 'pipe']);
    onSpawn(ssh);
    const streams = { stdout: [], stderr: [] };
    const sizes = { stdout: 0, stderr: 0 };
    const truncated = { stdout: false, stderr: false };
    for (const name of ['stdout', 'stderr'])
      ssh[name].on('data', (chunk) => {
        const room = limits.outputMax - sizes[name];
        if (chunk.length > room) truncated[name] = true;
        if (room > 0) {
          streams[name].push(chunk.subarray(0, room));
          sizes[name] += Math.min(room, chunk.length);
        }
      });
    ssh.stdin.on('error', () => {});
    ssh.stdin.end(stdin);
    let timedOut = false;
    let kill = null;
    const term = setTimeout(() => {
      timedOut = true;
      group(ssh, 'SIGTERM');
      kill = setTimeout(() => group(ssh, 'SIGKILL'), limits.graceMs);
    }, timeoutMs);
    const abort = () => group(ssh, 'SIGKILL');
    signal?.addEventListener('abort', abort, { once: true });
    const settle = (fn) => {
      clearTimeout(term);
      clearTimeout(kill);
      signal?.removeEventListener('abort', abort);
      fn();
    };
    ssh.on('error', () => settle(() => reject(new BusError('host-failed'))));
    ssh.on('close', (code, sig) =>
      settle(() =>
        resolve({
          exit: code,
          signal: sig ?? (timedOut ? 'SIGTERM' : null),
          stdout: Buffer.concat(streams.stdout).toString('utf8'),
          stderr: Buffer.concat(streams.stderr).toString('utf8'),
          truncated,
          ms: Date.now() - began,
        }),
      ),
    );
  });
}

// D7: one call. `key` is the parsed entry from credentials.mjs (text and publicLine); `entry` is
// the registry's { address, port, user, hostKeys }. `root` is <configDir>/secrets/run.
// Row 234 (#1740): `command`, when set, is sent as is in place of the sudo argv. The installer key's
// forced command reads it (tools/host/mosaic-keys).
export async function runHost({ root, hostId, entry, key, argv, stdin, timeout, signal = null, limits = HOST_LIMITS, command = null }) {
  if (signal?.aborted) deny('credential-unavailable');
  const dir = mkdtempSync(join(runRoot(root), 'h-'));
  const sock = join(dir, 'a.sock');
  let agent = null,
    ssh = null;
  try {
    if (Buffer.byteLength(sock) > SOCKET_MAX) deny('host-failed');
    writeFileSync(join(dir, 'known_hosts'), entry.hostKeys.map((k) => `mosaic-host-${hostId} ${k}\n`).join(''), { mode: 0o600, flag: 'wx' });
    writeFileSync(join(dir, 'id.pub'), key.publicLine + '\n', { mode: 0o600, flag: 'wx' });
    writeFileSync(join(dir, 'pids'), `${process.pid} ${startTime(process.pid)}\n`, { mode: 0o600, flag: 'wx' });
    agent = start('ssh-agent', ['-D', '-a', sock], { PATH }, ['ignore', 'ignore', 'ignore']);
    agent.on('error', () => {});
    if (agent.pid) record(dir, agent.pid);
    await waitForSocket(agent, sock, limits.agentWaitMs);
    // D7 step 3: the agent drops the key at the call's limit plus 30 s even if nothing else runs.
    await addKey(sock, key.text, timeout + 30, { limitMs: limits.agentWaitMs, signal });
    if (signal?.aborted) deny('credential-unavailable');
    const out = await runSsh(
      [...sshArgs({ hostId, ...entry, dir }), command ?? remoteCommand(argv)],
      stdin,
      timeout * 1000,
      signal,
      limits,
      (child) => {
        ssh = child;
        if (child.pid) record(dir, child.pid);
      },
    );
    if (signal?.aborted) deny('credential-unavailable');
    // Only ssh writes its log, so a remote command that prints the same text can't fake a pin
    // refusal. On ssh's own exit 255 the agent still gets its message, after the remote stderr.
    const log = out.exit === 255 && existsSync(join(dir, 'ssh.log')) ? readFileSync(join(dir, 'ssh.log'), 'utf8').slice(0, 4096) : '';
    if (log.includes('Host key verification failed')) deny('host-key-mismatch');
    if (log && !out.truncated.stderr) out.stderr += log;
    return out;
  } finally {
    group(ssh, 'SIGKILL');
    group(agent, 'SIGKILL');
    rmSync(dir, { recursive: true, force: true });
  }
}

// D7 step 6, at bus start before broker.sock listens. A dir whose bus pid is gone gets each of its
// live recorded processes that names the dir killed, then is removed. A dir whose bus is alive is
// left alone. Any entry that isn't a run dir refuses start.
export function sweepRunDir(root) {
  let names;
  try {
    names = readdirSync(root);
  } catch (e) {
    if (e.code === 'ENOENT') return { removed: [] };
    deny('credential-file');
  }
  runRoot(root);
  const removed = [];
  for (const name of names) {
    const dir = join(root, name);
    const st = lstatSync(dir);
    if (!RUN_DIR.test(name) || !st.isDirectory() || st.uid !== process.getuid()) deny('credential-file');
    let lines = [];
    try {
      lines = readFileSync(join(dir, 'pids'), 'utf8').split('\n').filter(Boolean).map((l) => l.split(' '));
    } catch {}
    const [bus, ...children] = lines;
    if (bus && alive(Number(bus[0]), bus[1])) continue;
    for (const [pid, at] of children) {
      if (!alive(Number(pid), at)) continue;
      let cmdline = '';
      try {
        cmdline = readFileSync(`/proc/${pid}/cmdline`, 'utf8');
      } catch {}
      if (cmdline.includes(dir)) {
        try {
          process.kill(Number(pid), 'SIGKILL');
        } catch {}
      }
    }
    rmSync(dir, { recursive: true, force: true });
    removed.push(name);
  }
  return { removed };
}

// The broker's executor (design D5): withHost beside withGitea, the key used only through useFor.
export function hostExecutor({ credentials, root, limits = HOST_LIMITS }) {
  return {
    busy: (session, hostId) => credentials.busy(session.business + '/' + session.role, 'host.' + hostId),
    run: (session, hostId, entry, { argv, stdin, timeout }) =>
      credentials.useFor(
        session,
        'host.' + hostId,
        (_token, { key, signal }) => runHost({ root, hostId, entry, key, argv, stdin, timeout, signal, limits }),
        { input: { argv, stdin } },
      ),
  };
}