199 lines
12 KiB
JavaScript
199 lines
12 KiB
JavaScript
// Slice 1 S5 (#1522): the Console's read of the bus. It runs the S4 human
|
|
// transport (packages/bus/src/human-cli.mjs, lead decision 70) as a child
|
|
// with spawn, not spawnSync, so one slow read never stalls the HTTP server.
|
|
// Only the four verbs of the Q1 module (lead decision 56), row 57's
|
|
// `decisions`, row 60's `credentials` and row 81's `replies` reads are
|
|
// reachable as reads: no read writes, not even seen or delivered (Q4). The
|
|
// one write is row 76's human `message.send` (lead decision 95), built by
|
|
// the CLI only with --send. The bus does the proof; a server started inside
|
|
// an agent run gets `human-required` from the broker.
|
|
|
|
import { spawn } from 'node:child_process';
|
|
import { randomBytes } from 'node:crypto';
|
|
import { fileURLToPath } from 'node:url';
|
|
import { HUMAN_CLI } from '../../cli/src/transport.mjs';
|
|
import { readHostState } from '../../cli/src/host.mjs';
|
|
import { views } from '../../bus/src/views.mjs';
|
|
import { CONSOLE_VERBS } from '../../bus/src/broker.mjs';
|
|
|
|
export class BusReadError extends Error {
|
|
constructor(code, message = code) { super(message); this.code = code; }
|
|
}
|
|
|
|
// The broker's identifier rule (packages/bus/src/broker.mjs `id`), checked
|
|
// here so a bad subject is a 400 and never reaches the bus.
|
|
export const subjectOk = s => typeof s === 'string' && s.length > 0 && s.length <= 160 && /^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$/.test(s);
|
|
|
|
// Same protocol as humanTransport in packages/cli/src/transport.mjs: the
|
|
// request on stdin, JSON on stdout, or a bus error code as the last
|
|
// `^[a-z-]{1,64}$` line of stderr.
|
|
export function humanCall({ socket, business, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024 }) {
|
|
return (verb, args = {}) => transport({ cli, socket, env: cli === HUMAN_CLI ? { ...env, MOSAIC_BUS_CLI_NONCE: randomBytes(32).toString('hex') } : env, input: { business, verb, args }, timeoutMs, maxBytes });
|
|
}
|
|
function transport({ cli, socket, env, input, timeoutMs, maxBytes }) {
|
|
return new Promise((resolve, reject) => {
|
|
let child;
|
|
try {
|
|
child = spawn(process.execPath, [cli, socket], { env, stdio: ['pipe', 'pipe', 'pipe'] });
|
|
} catch { return reject(new BusReadError('outcome-unknown', 'the bus transport did not start')); }
|
|
const out = [], err = [];
|
|
let size = 0, settled = false;
|
|
const done = (fn, value) => { if (!settled) { settled = true; clearTimeout(timer); fn(value); } };
|
|
const timer = setTimeout(() => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish in time')); }, timeoutMs);
|
|
child.stdout.on('data', chunk => {
|
|
size += chunk.length;
|
|
if (size > maxBytes) { child.kill('SIGKILL'); return done(reject, new BusReadError('response-too-large')); }
|
|
out.push(chunk);
|
|
});
|
|
child.stderr.on('data', chunk => { if (err.length < 64) err.push(chunk); });
|
|
child.on('error', () => done(reject, new BusReadError('outcome-unknown', 'the bus transport did not start')));
|
|
child.on('close', status => {
|
|
if (status === null) return done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish'));
|
|
if (status !== 0) {
|
|
const code = Buffer.concat(err).toString('utf8').trim().split('\n').reverse().find(l => /^[a-z-]{1,64}$/.test(l)) ?? 'invalid-response';
|
|
return done(reject, new BusReadError(code));
|
|
}
|
|
try { done(resolve, JSON.parse(Buffer.concat(out).toString('utf8'))); }
|
|
catch { done(reject, new BusReadError('invalid-response')); }
|
|
});
|
|
child.stdin.on('error', () => {}); // A child that exits early reports through 'close'.
|
|
child.stdin.end(`${JSON.stringify(input)}\n`);
|
|
});
|
|
}
|
|
|
|
// Row 252 (#1753, design section 2): the Console owner's answer to a grant request. Each call spawns
|
|
// a fresh console-cli with a fresh nonce, and the business and the signed-in (provider, iss, sub) go
|
|
// in its spawn environment, where the bus reads them back. Inherited console keys are dropped first,
|
|
// so only this call's values reach the wrapper. Only the two console verbs pass (N4). A resolve pins
|
|
// the business its confirmation was issued for.
|
|
export const CONSOLE_CLI = fileURLToPath(new URL('../../bus/src/console-cli.mjs', import.meta.url));
|
|
export function consoleCall({ dataRoot, socket, business = null, env = process.env, cli = CONSOLE_CLI, timeoutMs = 30000 }) {
|
|
return async (verb, args, { subject, business: pinned = null }) => {
|
|
if (!CONSOLE_VERBS.includes(verb)) throw new Error(`the Console does not call ${verb}`);
|
|
const id = pinned ?? (business || hostBusiness(dataRoot));
|
|
const own = Object.fromEntries(Object.entries(env).filter(([k]) => !k.startsWith('MOSAIC_BUS_CONSOLE_')));
|
|
return transport({
|
|
cli, socket, timeoutMs, maxBytes: (verb === 'request.artifact.console' ? 2 : 1) * 1024 * 1024, input: { verb, args },
|
|
env: { ...own, MOSAIC_BUS_CONSOLE_NONCE: randomBytes(32).toString('hex'), MOSAIC_BUS_CONSOLE_BUSINESS: id,
|
|
MOSAIC_BUS_CONSOLE_PROVIDER: subject.provider, MOSAIC_BUS_CONSOLE_ISS: subject.iss, MOSAIC_BUS_CONSOLE_SUB: subject.sub },
|
|
});
|
|
};
|
|
}
|
|
|
|
// The business is the --business flag, or else the live bus host's, read
|
|
// on every request so a host started after the Console is picked up.
|
|
export function busCaller({ dataRoot, socket, business = null, env = process.env, cli = HUMAN_CLI, timeoutMs = 30000 }) {
|
|
return async (verb, args) => {
|
|
return humanCall({ socket, business: business || hostBusiness(dataRoot), env, cli, timeoutMs })(verb, args);
|
|
};
|
|
}
|
|
function hostBusiness(dataRoot) {
|
|
let state;
|
|
try { state = readHostState(dataRoot); } catch { throw new BusReadError('no-bus-host', 'the bus host state file is unreadable'); }
|
|
if (!state?.live) throw new BusReadError('no-bus-host', 'no bus host is running and no --business was given');
|
|
return state.business;
|
|
}
|
|
export const busReader = options => views({ call: busCaller(options) });
|
|
|
|
// FED R6 (#1800): a granted user's bus transport, R3's user-cli.mjs (#414, design 3.6). Same
|
|
// protocol as humanCall: the request on stdin, JSON on stdout, or a bus error code as the last
|
|
// `^[a-z-]{1,64}$` line of stderr. Row 301 (#1808) moved it here from user-bus.mjs.
|
|
export const USER_CLI = fileURLToPath(new URL('../../bus/src/user-cli.mjs', import.meta.url));
|
|
|
|
// The Console sets the user nonce itself, so user-cli.mjs doesn't re-exec and the broker sees the
|
|
// Console (`cli.mjs serve`) as its parent (principal.mjs). The human nonce is dropped, so this child
|
|
// can never pass as human-cli, and it only ever runs USER_CLI.
|
|
export function userCall({ socket, business, user, env = process.env, timeoutMs = 30000, maxBytes = 8 * 1024 * 1024 }) {
|
|
const { MOSAIC_BUS_CLI_NONCE, ...rest } = env;
|
|
return (verb, args = {}) => new Promise((resolve, reject) => {
|
|
let child;
|
|
try {
|
|
child = spawn(process.execPath, [USER_CLI, socket], { env: { ...rest, MOSAIC_BUS_USER_NONCE: randomBytes(32).toString('hex') }, stdio: ['pipe', 'pipe', 'pipe'] });
|
|
} catch { return reject(new BusReadError('outcome-unknown', 'the bus transport did not start')); }
|
|
const out = [], err = [];
|
|
let size = 0, settled = false;
|
|
const done = (fn, value) => { if (!settled) { settled = true; clearTimeout(timer); fn(value); } };
|
|
const timer = setTimeout(() => { child.kill('SIGKILL'); done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish in time')); }, timeoutMs);
|
|
child.stdout.on('data', chunk => {
|
|
size += chunk.length;
|
|
if (size > maxBytes) { child.kill('SIGKILL'); return done(reject, new BusReadError('response-too-large')); }
|
|
out.push(chunk);
|
|
});
|
|
child.stderr.on('data', chunk => { if (err.length < 64) err.push(chunk); });
|
|
child.on('error', () => done(reject, new BusReadError('outcome-unknown', 'the bus transport did not start')));
|
|
child.on('close', status => {
|
|
if (status === null) return done(reject, new BusReadError('outcome-unknown', 'the bus transport did not finish'));
|
|
if (status !== 0) {
|
|
const code = Buffer.concat(err).toString('utf8').trim().split('\n').reverse().find(l => /^[a-z-]{1,64}$/.test(l)) ?? 'invalid-response';
|
|
return done(reject, new BusReadError(code));
|
|
}
|
|
try { done(resolve, JSON.parse(Buffer.concat(out).toString('utf8'))); }
|
|
catch { done(reject, new BusReadError('invalid-response')); }
|
|
});
|
|
child.stdin.on('error', () => {}); // A child that exits early reports through 'close'.
|
|
child.stdin.end(`${JSON.stringify({ business, user, verb, args })}\n`);
|
|
});
|
|
}
|
|
|
|
// The business is the --business flag, or else the live bus host's, read on every call, as busCaller does.
|
|
export function busBusiness({ dataRoot, business = null }) {
|
|
return () => {
|
|
if (business) return business;
|
|
let state;
|
|
try { state = readHostState(dataRoot); } catch { throw new BusReadError('no-bus-host', 'the bus host state file is unreadable'); }
|
|
if (!state?.live) throw new BusReadError('no-bus-host', 'no bus host is running and no --business was given');
|
|
return state.business;
|
|
};
|
|
}
|
|
|
|
// One granted user's caller, for that user's session only.
|
|
export function userCaller({ dataRoot, socket, business = null, user, env = process.env, timeoutMs = 30000 }) {
|
|
const which = busBusiness({ dataRoot, business });
|
|
return async (verb, args) => userCall({ socket, business: which(), user, env, timeoutMs })(verb, args);
|
|
}
|
|
|
|
// A send's error, sorted. The broker writes in one transaction, so any code it
|
|
// returns means nothing was written. A transport that lost the answer, or an
|
|
// answer that can't be read, leaves the outcome open: the message may exist.
|
|
// Row 239 (duckworth-0165 C1): so does a write that failed after its revoke.
|
|
const OPEN = ['outcome-unknown', 'invalid-response', 'response-too-large', 'outcome-unknown-revoked'];
|
|
export function sendOutcome(err) {
|
|
const code = err instanceof BusReadError ? err.code : 'outcome-unknown';
|
|
return OPEN.includes(code) ? { code: 'outcome-unknown', sent: 'unknown' } : { code, sent: false };
|
|
}
|
|
|
|
// HTTP status by bus error code. 403: the bus refused this caller; 503: no
|
|
// bus to ask, or no answer; 502: an answer that can't be used.
|
|
const STATUS = {
|
|
'human-required': 403, 'peercred-required': 403, unauthenticated: 403, 'unknown-business': 403, 'read-only': 403,
|
|
'outcome-unknown': 503, 'no-bus-host': 503, 'not-configured': 503,
|
|
'revision-conflict':409,'organization-busy':409,'organization-uncertain':503,
|
|
// Row 239 (#1745): a fire's or a delete's write refused after its revoke ran.
|
|
'revision-conflict-revoked': 409, 'outcome-unknown-revoked': 503,
|
|
'invalid-request': 400, 'unknown-role': 400, 'message-not-found': 400, 'request-too-large': 400,
|
|
// Row 148: the spec and docs editor. A stale base, a busy or dirty repository and an unchanged
|
|
// file are the user's to resolve by reloading; a path or content the project refuses is a 400.
|
|
'stale-base': 409, 'project-busy': 409, 'project-dirty': 409,
|
|
'project-name-is-repo': 409,
|
|
'project-record-unchanged': 400, 'project-record-path': 400, 'project-record-invalid': 400,
|
|
'project-read-path': 400, 'project-not-declared': 400, 'project-read-missing': 404, 'project-not-configured': 503,
|
|
// Row 155 (#1649): the assistant's settings, seat names and stop. A name in use or no run to stop
|
|
// is the user's to resolve by reading again.
|
|
'organization-not-assistant': 400, 'organization-definition-refused': 400, 'unknown-model': 400,
|
|
'name-taken': 409, 'name-is-instance': 400, 'name-is-role': 400, 'not-assistant': 400, 'not-held': 409,
|
|
// Row 165 (#1662): role instances. An id already taken is the user's to resolve by reading again.
|
|
'organization-instance-exists': 409, 'organization-unknown-instance': 400,
|
|
'organization-credentials-out-of-scope': 400, 'organization-launch-not-configured': 400,
|
|
// Row 194 (brief item 7, sage-1610 a): only the CLI names a Vikunja bot on add.
|
|
'organization-tracker-required': 400, 'organization-tracker-bot-taken': 409,
|
|
'organization-no-change': 400, 'organization-project-mismatch': 400, 'organization-unresolvable-instance': 400,
|
|
// Row 254 (#1759): library sync. A sync already running, a held divergence and a stale accept are
|
|
// the user's to resolve by reading library.status again; a missing credential or a failed fetch
|
|
// leaves the remote unreachable, like no bus to ask.
|
|
'library-sync-busy': 409, 'library-diverged': 409, 'library-accept-stale': 409, 'library-accept-invalid': 400,
|
|
'library-credential-missing': 503, 'library-fetch-failed': 503,
|
|
// Row 225 (C2-10b): a Revert of a commit that isn't an agent's is hers to read, like stale-base.
|
|
'library-revert-not-agent': 409,
|
|
};
|
|
export const busStatus = code => STATUS[code] ?? 502;
|