217 lines
10 KiB
JavaScript
217 lines
10 KiB
JavaScript
import { createServer } from 'node:net';
|
|
import { lstatSync, chmodSync, unlinkSync } from 'node:fs';
|
|
import { BusError, TASK_VERBS, WEB_VERBS, HOST_VERBS, HOSTS_VERBS, GRANTS_VERBS, PROJECT_FILE_VERBS, LIBRARY_VERBS, PROJECT_SAVE_VERB, PROJECT_SAVE_REQUEST_MAX, CONSOLE_VERBS } from './broker.mjs';
|
|
import { inspectInterpreter } from './interpreter.mjs';
|
|
import { createPeerReader } from './peercred.mjs';
|
|
const LIMIT = 65536;
|
|
// Row 148: a document save may carry up to 1 MiB of text. Only a request that names that verb near
|
|
// its start (the proof object comes first) may pass LIMIT; the parsed verb is checked again below.
|
|
const PROJECT_SAVE_AT_START = new RegExp(`"verb":"${PROJECT_SAVE_VERB.replaceAll('.', '\\.')}"`);
|
|
const LIBRARY_SAVE_AT_START = /"verb":"library\.file\.save"/;
|
|
const SAVE_VERB_AT_START = { test: text => PROJECT_SAVE_AT_START.test(text) || LIBRARY_SAVE_AT_START.test(text) };
|
|
// Preserve every code and scoped reason, adding only static rail/next-step guidance.
|
|
const refusal = (e, other) => {
|
|
const error = e instanceof BusError ? e : new BusError(other);
|
|
return { ok: false, error: error.code, rail: error.rail, nextStep: error.nextStep,
|
|
...(typeof error.reason === 'string' ? { reason: error.reason } : {}) };
|
|
};
|
|
// One request per connection. There is deliberately no reconnect/retry or SQL verb.
|
|
// `tasks` ({handle, timeout}) is the trusted S3 adapter; only the eight task verbs reach it.
|
|
// `web` is the trusted web service (web.mjs), or a function from a business id to one (or null); only
|
|
// the web verbs reach it, and without one they refuse web-not-configured (or web-not-granted).
|
|
// `host` (row 210) is the trusted host.run executor (host-run.mjs), or null: host.run then refuses
|
|
// host-not-configured.
|
|
// A request carries exactly one of cap, human (row S4), tester (row 110), user (FED R3) and console
|
|
// (row 252) proof.
|
|
|
|
export async function serve({ broker, path, authenticateHuman = null, authenticateTester = null, authenticateUser = null, authenticateConsole = null, peerReader = null, timeout = 5000, tasks = null, web = null, host = null }) {
|
|
try {
|
|
lstatSync(path);
|
|
throw new BusError('socket-exists');
|
|
} catch (e) {
|
|
if (e.code !== 'ENOENT') throw e;
|
|
}
|
|
const peers = peerReader ?? createPeerReader({ interpreter: inspectInterpreter() });
|
|
const sockets = new Set();
|
|
const server = createServer({ allowHalfOpen: true }, (socket) => {
|
|
sockets.add(socket);
|
|
let boundCap;
|
|
const release = () => { if (boundCap) { broker.disconnect(boundCap); boundCap = null; } };
|
|
socket.on('close', () => { sockets.delete(socket); release(); });
|
|
socket.on('error', () => {});
|
|
let input = Buffer.alloc(0),
|
|
done = false, processing = false;
|
|
const finish = (result) => {
|
|
if (done) return;
|
|
done = true;
|
|
release();
|
|
socket.resume();
|
|
socket.end(JSON.stringify(result) + '\n');
|
|
};
|
|
socket.setTimeout(timeout, () => {
|
|
finish(refusal(new BusError('request-timeout')));
|
|
socket.destroySoon();
|
|
});
|
|
socket.on('data', async (chunk) => {
|
|
if (done) return;
|
|
if (processing) return finish(refusal(new BusError('invalid-envelope')));
|
|
const total = input.length + chunk.length;
|
|
if (total > LIMIT && !(total <= PROJECT_SAVE_REQUEST_MAX &&
|
|
SAVE_VERB_AT_START.test(Buffer.concat([input, chunk]).subarray(0, 1024).toString('latin1')))) {
|
|
finish(refusal(new BusError('request-too-large')));
|
|
return;
|
|
}
|
|
input = Buffer.concat([input, chunk]);
|
|
const end = input.indexOf(10);
|
|
if (end < 0) return;
|
|
processing = true;
|
|
let transient;
|
|
try {
|
|
if (end !== input.length - 1) throw new BusError('invalid-envelope');
|
|
const r = JSON.parse(input.subarray(0, end).toString('utf8'));
|
|
if (input.length > LIMIT && ![PROJECT_SAVE_VERB, 'library.file.save'].includes(r?.verb)) throw new BusError('request-too-large');
|
|
if (
|
|
!r ||
|
|
Array.isArray(r) ||
|
|
typeof r !== 'object' ||
|
|
Object.keys(r).some((k) => !['cap', 'human', 'tester', 'user', 'console', 'verb', 'args'].includes(k)) ||
|
|
typeof r.verb !== 'string' ||
|
|
[r.cap, r.human, r.tester, r.user, r.console].filter(Boolean).length !== 1
|
|
)
|
|
throw new BusError('invalid-envelope');
|
|
let cap = r.cap;
|
|
const kind = r.human ? 'human' : r.tester ? 'tester' : r.user ? 'user' : r.console ? 'console' : null;
|
|
// Row 252 (#1753): a Console owner proof reaches its two verbs only, refused before any proof check.
|
|
if (kind === 'console' && !CONSOLE_VERBS.includes(r.verb)) throw new BusError('human-required');
|
|
if (kind) {
|
|
const authenticate = kind === 'human' ? authenticateHuman : kind === 'tester' ? authenticateTester : kind === 'user' ? authenticateUser : authenticateConsole;
|
|
if (!authenticate) throw new BusError(kind === 'tester' ? 'tester-required' : 'human-required');
|
|
socket.pause();
|
|
let peer;
|
|
try { peer = await peers.read(socket); }
|
|
catch { throw new BusError('peercred-required'); }
|
|
if (done || socket.destroyed) return;
|
|
transient = cap = authenticate(r[kind], peer);
|
|
boundCap = transient;
|
|
socket.resume();
|
|
}
|
|
if (typeof cap !== 'string') throw new BusError('unauthenticated');
|
|
if (tasks && TASK_VERBS.includes(r.verb)) {
|
|
const own = transient;
|
|
transient = null;
|
|
// A Vikunja write can outlast the idle timeout; a late reply is still outcome-unknown to the client.
|
|
socket.setTimeout(tasks.timeout);
|
|
broker
|
|
.requestTask(cap, { verb: r.verb, args: r.args ?? {} }, tasks.handle)
|
|
.then(
|
|
(result) => finish({ ok: true, result }),
|
|
(e) => finish(refusal(e, 'adapter-failed')),
|
|
)
|
|
.finally(() => {
|
|
if (own) release();
|
|
});
|
|
return;
|
|
}
|
|
if (WEB_VERBS.includes(r.verb)) {
|
|
const own = transient;
|
|
transient = null;
|
|
const service = typeof web === 'function' ? web(broker.identity(cap).business) : web;
|
|
// A fetch outlasts the idle timeout; the service's own deadline bounds it. The longer wait
|
|
// is given only to a call the broker admitted (grant and arguments checked), so a refused
|
|
// call can't hold the connection for it (vikunja:19/307).
|
|
broker
|
|
.requestWeb(cap, { verb: r.verb, args: r.args ?? {} }, service, { onAdmit: () => socket.setTimeout(service.timeout) })
|
|
.then(
|
|
(result) => finish({ ok: true, result }),
|
|
(e) => finish(refusal(e, 'web-failed')),
|
|
)
|
|
.finally(() => {
|
|
if (own) release();
|
|
});
|
|
return;
|
|
}
|
|
// Row 210 (#1707, design D5): a sudo command can run for minutes. As for web, the longer wait
|
|
// goes only to a call the broker admitted: the call's own limit plus 10 s.
|
|
if (HOST_VERBS.includes(r.verb)) {
|
|
const own = transient;
|
|
transient = null;
|
|
broker
|
|
.requestHost(cap, { verb: r.verb, args: r.args ?? {} }, host, { onAdmit: (ms) => socket.setTimeout(ms + 10000) })
|
|
.then(
|
|
(result) => finish({ ok: true, result }),
|
|
(e) => finish(refusal(e, 'host-failed')),
|
|
)
|
|
.finally(() => {
|
|
if (own) release();
|
|
});
|
|
return;
|
|
}
|
|
// Async effects retain the synchronous path for every other verb.
|
|
if (r.verb === 'processing.halt' || r.verb === 'processing.resume' ||
|
|
r.verb === 'usage.begin' || r.verb === 'usage.end' ||
|
|
r.verb === 'spec.read' || r.verb === 'record.write' || PROJECT_FILE_VERBS.includes(r.verb) || LIBRARY_VERBS.includes(r.verb) ||
|
|
// Row 234 (#1740): a grant raise awaits its keyscan, and hosts.check its ssh.
|
|
HOSTS_VERBS.includes(r.verb) ||
|
|
// Row 236 (#1742): the grants verbs may call Gitea several times, 15 s each at most.
|
|
GRANTS_VERBS.includes(r.verb) ||
|
|
r.verb === 'proxy.init' ||
|
|
(r.verb === 'decision.raise' && (Object.hasOwn(r.args ?? {}, 'spec_path') || Object.hasOwn(r.args ?? {}, 'spec_commit') ||
|
|
Object.hasOwn(r.args ?? {}, 'grant')))) {
|
|
const own = transient;
|
|
transient = null;
|
|
// Local Git may take 100 seconds; keep its result beyond the ordinary idle deadline.
|
|
socket.setTimeout(120_000);
|
|
Promise.resolve().then(() => broker.request(cap, { verb: r.verb, args: r.args ?? {} })).then(
|
|
result => { if (!socket.destroyed) finish({ ok: true, result }); },
|
|
e => { if (!socket.destroyed) finish(refusal(e, 'invalid-envelope')); },
|
|
).finally(() => { if (own) release(); });
|
|
return;
|
|
}
|
|
const result = broker.request(cap, { verb: r.verb, args: r.args ?? {} });
|
|
// C2-06: a guarded message.send and a library business's thread.messages read Git first, so
|
|
// the broker answers them with a promise. Every other verb stays synchronous.
|
|
if (result instanceof Promise) {
|
|
const own = transient;
|
|
transient = null;
|
|
socket.setTimeout(120_000);
|
|
result.then(
|
|
value => { if (!socket.destroyed) finish({ ok: true, result: value }); },
|
|
e => { if (!socket.destroyed) finish(refusal(e, 'invalid-envelope')); },
|
|
).finally(() => { if (own) release(); });
|
|
return;
|
|
}
|
|
finish({ ok: true, result });
|
|
} catch (e) {
|
|
finish(refusal(e, 'invalid-envelope'));
|
|
} finally {
|
|
if (transient) release();
|
|
}
|
|
});
|
|
socket.on('end', () => {
|
|
if (!done && !processing) finish(refusal(new BusError('incomplete-request')));
|
|
});
|
|
});
|
|
await new Promise((resolve, reject) => {
|
|
server.once('error', reject);
|
|
server.listen(path, () => {
|
|
server.removeListener('error', reject);
|
|
resolve();
|
|
});
|
|
});
|
|
chmodSync(path, 0o600);
|
|
const owned = lstatSync(path);
|
|
return {
|
|
async close() {
|
|
await peers.close();
|
|
for (const s of sockets) s.destroy();
|
|
await new Promise((resolve) => server.close(resolve));
|
|
try {
|
|
const cur = lstatSync(path);
|
|
if (cur.ino === owned.ino && cur.dev === owned.dev) unlinkSync(path);
|
|
} catch (e) {
|
|
if (e.code !== 'ENOENT') throw e;
|
|
}
|
|
},
|
|
};
|
|
}
|