stack/packages/model-proxy/src/admin.mjs

// #1873 proxy setup state and host actions shared by `mosaic proxy` and the Console page.
// Never imports the management client and never reads the management key file: state is
// file presence, the pinned install hashes, the fence file and the proxy unit's systemd state.
// Results and receipts name files by label, never by absolute path, and carry no key.
import fs from 'node:fs';
import path from 'node:path';
import net from 'node:net';
import { execFile } from 'node:child_process';
import { fileURLToPath } from 'node:url';
import { PIN, refuse } from './common.mjs';
import { installPaths, loadProxy, verifyInstall } from './settings.mjs';

const REPO = fileURLToPath(new URL('../../../', import.meta.url)).replace(/\/$/, '');
export const INSTANCE_NAME = /^[a-z][a-z0-9]{1,15}$/;


export const FILE_LABELS = Object.freeze({ settings: 'proxy.json', key: 'the management key file', proxyDir: 'the proxy state directory', lock: 'the init lock' });
export function initFiles(configDir, dataRoot) {
  return {
    settings: path.join(configDir, 'proxy.json'),
    key: path.join(configDir, 'secrets', 'proxy-management.key'),
    proxyDir: path.join(dataRoot, 'proxy'),
    lock: path.join(configDir, '.proxy-init.lock'),
  };
}
// Presence only. A path that cannot be examined counts as present, so nothing clears on doubt.
function present(file) { try { fs.lstatSync(file); return true; } catch (error) { return error.code !== 'ENOENT'; } }

const verified = new Map();
// The pinned release: hashes are checked once per file identity (dev, inode, size, mtime, ctime, mode),
// in nanoseconds. mtime can be set back by the file's owner; ctime cannot, so an in-place rewrite re-hashes.
export function releaseState(dataRoot, pin = PIN) {
  const paths = installPaths(dataRoot, pin);
  const out = { version: pin.version, panelVersion: pin.panel.version, fetched: false, verified: false, problem: null };
  let stats;
  try { stats = [paths.binary, paths.panel].map(file => fs.lstatSync(file, { bigint: true })); }
  catch (error) { out.problem = error.code === 'ENOENT' ? 'proxy-install-missing' : 'proxy-install-invalid'; return out; }
  out.fetched = true;
  const identity = stats.map(s => `${s.dev}:${s.ino}:${s.size}:${s.mtimeNs}:${s.ctimeNs}:${s.mode}`).join('|');
  if (verified.get(paths.dir) === identity) { out.verified = true; return out; }
  try { verifyInstall(dataRoot, pin); verified.set(paths.dir, identity); out.verified = true; }
  catch (error) { out.problem = /^proxy-install-(missing|invalid)$/.test(error.code ?? '') ? error.code : 'proxy-install-invalid'; }
  return out;
}

export function initState(configDir, dataRoot) {
  const files = initFiles(configDir, dataRoot);
  const found = Object.fromEntries(Object.entries(files).map(([name, file]) => [name, present(file)]));
  let settings = null, problem = null;
  if (found.settings) {
    try {
      const proxy = loadProxy(configDir, dataRoot);
      settings = { port: proxy.port, rotateGraceMinutes: proxy.rotateGraceMinutes ?? null, firstOutputSeconds: proxy.firstOutputSeconds ?? null };
    } catch (error) { problem = /^proxy-[a-z-]+$/.test(error.code ?? '') ? error.code : 'proxy-config-invalid'; }
  }
  const initialized = found.settings && found.key && found.proxyDir && !problem;
  const partial = !initialized && (found.settings || found.key || found.proxyDir);
  return { initialized, partial, problem, settings, files: found, lock: found.lock };
}

// Read-only view of the uncertain-write fence. Never constructs a proxy lock, which writes.
export function fenceState(dataRoot) {
  const file = path.join(dataRoot, 'proxy', 'mosaic-keys.uncertain');
  let stat;
  try { stat = fs.lstatSync(file); } catch (error) { if (error.code === 'ENOENT') return null; return { invalid: true }; }
  if (!stat.isFile() || stat.uid !== process.getuid() || (stat.mode & 0o077) || stat.size > 4096) return { invalid: true };
  let entry;
  try { entry = JSON.parse(fs.readFileSync(file, 'utf8')); } catch { return { invalid: true }; }
  if (!entry || typeof entry !== 'object' || Array.isArray(entry)) return { invalid: true };
  return {
    org: typeof entry.org === 'string' ? entry.org : null,
    op: typeof entry.op === 'string' ? entry.op : null,
    time: typeof entry.time === 'string' ? entry.time : null,
    pid: Number.isSafeInteger(entry.pid) ? entry.pid : null,
  };
}

// README conditions for a stale init lock: the bus is stopped (so no init is running), and
// none of proxy.json, the key file or the proxy dir exists. `buses` is the caller's proof:
// { stopped: boolean, running: [label] }.
export function initLockStatus({ configDir, dataRoot, buses }) {
  const state = initState(configDir, dataRoot);
  const blockers = [];
  if (!buses || buses.stopped !== true) blockers.push(`bus running: ${(buses?.running ?? ['unknown']).join(', ')}`);
  for (const name of ['settings', 'key', 'proxyDir']) if (state.files[name]) blockers.push(`${FILE_LABELS[name]} exists`);
  return { present: state.lock, clearable: state.lock && blockers.length === 0, blockers: state.lock ? blockers : [] };
}

export function clearInitLock({ configDir, dataRoot, buses, now = () => new Date() }) {
  const status = initLockStatus({ configDir, dataRoot, buses });
  if (!status.present) refuse('proxy-init-lock-absent');
  if (!status.clearable) refuse('proxy-init-lock-held');
  const lock = initFiles(configDir, dataRoot).lock;
  const stat = fs.lstatSync(lock);
  if (!stat.isFile() || stat.uid !== process.getuid()) refuse('proxy-init-lock-held');
  fs.unlinkSync(lock);
  return appendReceipt(dataRoot, { op: 'init-lock.clear', time: now().toISOString(), conditions: ['bus stopped', 'no proxy.json', 'no management key file', 'no proxy state directory'] });
}

// Append-only admin receipts beside the other state logs. Labels and key ids only.
export function receiptFile(dataRoot) { return path.join(dataRoot, 'state', 'proxy-admin.jsonl'); }
export function appendReceipt(dataRoot, record) {
  const dir = path.join(dataRoot, 'state');
  fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
  const fd = fs.openSync(receiptFile(dataRoot), 'a', 0o600);
  try { fs.writeFileSync(fd, JSON.stringify(record) + '\n'); fs.fsyncSync(fd); } finally { fs.closeSync(fd); }
  return record;
}

// The last few receipts, oldest first, for the page after a reload. Reads only the file's tail and
// skips a line that does not parse (a torn last write). Receipts never hold a key or a body.
export function recentReceipts(dataRoot, limit = 10) {
  let fd;
  try { fd = fs.openSync(receiptFile(dataRoot), 'r'); } catch { return []; }
  try {
    const size = fs.fstatSync(fd).size, length = Math.min(size, 65536), buffer = Buffer.alloc(length);
    fs.readSync(fd, buffer, 0, length, size - length);
    const lines = buffer.toString('utf8').split('\n').slice(size > length ? 1 : 0).filter(Boolean);
    const out = [];
    for (const line of lines) { try { const value = JSON.parse(line); if (value && typeof value === 'object' && typeof value.op === 'string') out.push(value); } catch { /* torn line */ } }
    return out.slice(-limit);
  } finally { fs.closeSync(fd); }
}

export function unitName(instance) {
  if (instance !== null && instance !== undefined && instance !== 'dev' && !INSTANCE_NAME.test(instance)) refuse('proxy-unit-usage');
  return instance && instance !== 'dev' ? `mosaic-proxy-${instance}.service` : 'mosaic-proxy.service';
}

export function defaultRun(command, args, { env, timeout = 60000 } = {}) {
  return new Promise(resolve => {
    execFile(command, args, { env, timeout, maxBuffer: 1024 * 1024, killSignal: 'SIGKILL' }, (error, stdout, stderr) => {
      resolve({ code: error ? (typeof error.code === 'number' ? error.code : 1) : 0, stdout: String(stdout ?? ''), stderr: String(stderr ?? '') });
    });
  });
}

export function probePort(port, { host = '127.0.0.1', timeout = 1000 } = {}) {
  return new Promise(resolve => {
    if (!Number.isInteger(port)) return resolve(false);
    const socket = net.connect({ host, port });
    const done = value => { socket.destroy(); resolve(value); };
    socket.setTimeout(timeout, () => done(false));
    socket.once('connect', () => done(true));
    socket.once('error', () => done(false));
  });
}

const UNIT_CODES = ['proxy-unit-usage', 'proxy-unit-exists', 'proxy-unit-dropin', 'proxy-unit-copy', 'proxy-unit-path', 'proxy-not-configured'];
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));

// The proxy user unit for one instance. `port` is a function so a fresh init is seen.
// `unitDir` is passed to proxy-service.sh --dir; null keeps that script's default.
export function createUnitControl({ instance = null, port = () => null, unitDir = null, env = process.env, run = defaultRun, repo = REPO, probe = probePort, waitMs = 30000, pollMs = 250, now = () => new Date() } = {}) {
  const unit = unitName(instance);
  if (unitDir !== null && (typeof unitDir !== 'string' || !path.isAbsolute(unitDir))) refuse('proxy-unit-path');
  const script = path.join(repo, 'scripts', 'proxy-service.sh');
  const instanceArgs = instance && instance !== 'dev' ? ['--instance', instance] : [];
  async function service(op) {
    const result = await run('bash', [script, ...instanceArgs, op, ...(unitDir && op === 'install' ? ['--dir', unitDir] : [])], { env });
    if (result.code === 0) return result;
    const code = /^proxy-service: (proxy-[a-z-]+)$/m.exec(result.stderr)?.[1];
    refuse(UNIT_CODES.includes(code) ? code : 'proxy-unit-failed');
  }
  async function status() {
    const result = await run('systemctl', ['--user', 'show', unit, '--property=LoadState,ActiveState,SubState,MainPID,ExecMainStartTimestampMonotonic,NRestarts'], { env, timeout: 15000 });
    if (result.code !== 0) refuse('proxy-unit-failed');
    const fields = Object.fromEntries(result.stdout.split('\n').filter(Boolean).map(line => { const i = line.indexOf('='); return [line.slice(0, i), line.slice(i + 1)]; }));
    const pid = /^[1-9][0-9]*$/.test(fields.MainPID ?? '') ? Number(fields.MainPID) : null;
    const since = /^[1-9][0-9]*$/.test(fields.ExecMainStartTimestampMonotonic ?? '') ? Number(fields.ExecMainStartTimestampMonotonic) : null;
    const p = port();
    const listening = fields.ActiveState === 'active' && Number.isInteger(p) ? await probe(p) : false;
    return {
      unit,
      installed: fields.LoadState === 'loaded',
      load: fields.LoadState ?? 'unknown', active: fields.ActiveState ?? 'unknown', sub: fields.SubState ?? 'unknown',
      pid, since, restarts: /^[0-9]+$/.test(fields.NRestarts ?? '') ? Number(fields.NRestarts) : null,
      port: Number.isInteger(p) ? p : null, listening,
      healthy: fields.ActiveState === 'active' && pid !== null && listening,
    };
  }
  async function waitFor(test) {
    const deadline = Date.now() + waitMs;
    for (;;) {
      const state = await status();
      if (test(state)) return state;
      if (Date.now() >= deadline) return state;
      await sleep(pollMs);
    }
  }
  async function systemctl(verb) {
    const result = await run('systemctl', ['--user', verb, unit], { env, timeout: 90000 });
    if (result.code !== 0) refuse('proxy-unit-failed');
  }
  return {
    unit,
    status,
    async render() { return (await service('render')).stdout; },
    async install() { await service('install'); return status(); },
    async start() {
      const before = await status();
      if (!before.installed) refuse('proxy-unit-missing');
      await systemctl('start');
      const after = await waitFor(s => s.healthy);
      if (!after.healthy) refuse('proxy-unit-unhealthy');
      return after;
    },
    async stop() {
      const before = await status();
      if (!before.installed) refuse('proxy-unit-missing');
      await systemctl('stop');
      return waitFor(s => s.active === 'inactive' || s.active === 'failed');
    },
    // Restart evidence is what a fence clear needs: a new main process, started after the old
    // one, active and answering on the proxy port. Anything less is not a confirmed restart.
    async restart() {
      const before = await status();
      if (!before.installed) refuse('proxy-unit-missing');
      await systemctl('restart');
      const changed = s => s.healthy && s.pid !== null && s.pid !== before.pid && s.since !== null && (before.since === null || s.since > before.since);
      const after = await waitFor(changed);
      const confirmed = changed(after);
      return {
        state: after,
        evidence: {
          unit, confirmed,
          before: { pid: before.pid, since: before.since, active: before.active },
          after: { pid: after.pid, since: after.since, active: after.active, listening: after.listening },
          observedAt: now().toISOString(),
        },
      };
    },
  };
}

// Process start time as wall-clock milliseconds, from /proc (start ticks since boot + btime).
export function processStartMs(pid, { proc = '/proc' } = {}) {
  try {
    const stat = fs.readFileSync(path.join(proc, String(pid), 'stat'), 'utf8');
    const fields = stat.slice(stat.lastIndexOf(')') + 2).split(' ');
    const ticks = Number(fields[19]);
    const btime = Number(/^btime ([0-9]+)$/m.exec(fs.readFileSync(path.join(proc, 'stat'), 'utf8'))?.[1]);
    if (!Number.isFinite(ticks) || !Number.isFinite(btime)) return null;
    return btime * 1000 + Math.floor(ticks * 10);
  } catch { return null; }
}
// `pid` and every process descended from it, from each process's parent field in /proc.
export function processTree(pid, { proc = '/proc' } = {}) {
  const parents = new Map();
  let names; try { names = fs.readdirSync(proc); } catch { return [pid]; }
  for (const name of names) {
    if (!/^[1-9][0-9]*$/.test(name)) continue;
    try {
      const stat = fs.readFileSync(path.join(proc, name, 'stat'), 'utf8');
      const ppid = Number(stat.slice(stat.lastIndexOf(')') + 2).split(' ')[1]);
      if (!parents.has(ppid)) parents.set(ppid, []);
      parents.get(ppid).push(Number(name));
    } catch { /* gone */ }
  }
  const out = [pid];
  for (let i = 0; i < out.length && out.length < 256; i++) out.push(...(parents.get(out[i]) ?? []));
  return out;
}
// True when `pid` or one of its descendants holds a listening TCP socket on `port` (IPv4 or
// IPv6), read from /proc. The unit's main process is the node runner; the pinned binary it
// spawns is the listener.
export function pidListensOn(pid, port, { proc = '/proc' } = {}) {
  const hex = port.toString(16).toUpperCase().padStart(4, '0');
  const inodes = new Set();
  for (const table of ['tcp', 'tcp6']) {
    let text; try { text = fs.readFileSync(path.join(proc, 'net', table), 'utf8'); } catch { continue; }
    for (const line of text.split('\n').slice(1)) {
      const cols = line.trim().split(/\s+/);
      if (cols.length > 9 && cols[1].endsWith(`:${hex}`) && cols[3] === '0A') inodes.add(cols[9]);
    }
  }
  if (inodes.size === 0) return false;
  for (const member of processTree(pid, { proc })) {
    let fds; try { fds = fs.readdirSync(path.join(proc, String(member), 'fd')); } catch { continue; }
    for (const fd of fds) {
      let link; try { link = fs.readlinkSync(path.join(proc, String(member), 'fd', fd)); } catch { continue; }
      const m = /^socket:\[([0-9]+)\]$/.exec(link);
      if (m && inodes.has(m[1])) return true;
    }
  }
  return false;
}

// A fence clear accepts restart evidence only after re-proving it here, on the host: the
// evidence names a confirmed restart to a new process, and that process is alive now, started
// after the fence was set, and is the one listening on the proxy port. Caller-supplied fields
// are never trusted on their own.
export function verifyRestart(evidence, { fence, port, startMs = processStartMs, listens = pidListensOn, now = () => Date.now() } = {}) {
  const int = v => Number.isSafeInteger(v) && v > 0;
  const plain = v => v !== null && typeof v === 'object' && !Array.isArray(v);
  if (!plain(evidence) || Object.keys(evidence).sort().join(',') !== 'after,before,confirmed,observedAt,unit'
      || typeof evidence.unit !== 'string' || !/^mosaic-proxy(-[a-z][a-z0-9]{1,15})?\.service$/.test(evidence.unit)
      || typeof evidence.observedAt !== 'string' || Number.isNaN(Date.parse(evidence.observedAt))
      || !plain(evidence.before) || !plain(evidence.after)) refuse('proxy-restart-unconfirmed');
  const { before, after } = evidence;
  if (evidence.confirmed !== true || !int(after.pid) || !int(after.since) || after.pid === before.pid) refuse('proxy-restart-unconfirmed');
  const observed = Date.parse(evidence.observedAt);
  if (observed > now() + 60000 || now() - observed > 10 * 60 * 1000) refuse('proxy-restart-unconfirmed');
  const started = startMs(after.pid);
  if (started === null) refuse('proxy-restart-unconfirmed');
  const fenceTime = fence && typeof fence.time === 'string' ? Date.parse(fence.time) : NaN;
  if (fence && !(Number.isFinite(fenceTime) && started > fenceTime)) refuse('proxy-restart-unconfirmed');
  if (!Number.isInteger(port) || !listens(after.pid, port)) refuse('proxy-restart-unconfirmed');
  return { unit: evidence.unit, before: { pid: int(before.pid) ? before.pid : null }, after: { pid: after.pid, startedAt: new Date(started).toISOString() }, observedAt: evidence.observedAt, listeningPort: port };
}