runtime: connect with a sealed credential, scoped, over pinned amqps (ADR 0048)
A module reads its broker credential from MESH_BROKER_FILE — the sealed
{url,fingerprint} the mesh delivered — and connects over amqps pinned to
exactly that certificate. The pin is two-phase (fetch cert, verify, then
trust only it), because Node's checkServerIdentity does not run under
rejectUnauthorized:false, so a naive connect-then-check would already have
sent the password to whoever answered.
A scoped module (assumeExchanges) never declares the exchanges (its account
may not) nor its own queue with a dead-letter (the broker refuses that to a
non-administrator) — the mesh pre-declared the queue, so it passively checks
it, binds and consumes. The RPC reply queue is lazy, and a module that
registered no tools serves none: a pure-events consumer touches only what its
account allows.
Verified end-to-end against a real broker as the scoped account: the audit
logger consumes # and records events, over an account that is not the
broker's own.
Claude-Session: https://claude.ai/code/session_01LrgweAeERJYBg88c5cKDzF
This commit is contained in:
+116
-28
@@ -5,7 +5,8 @@
|
||||
// prefetch, dead-letter — none of which a module ever sees.
|
||||
|
||||
import amqp from "amqplib";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import * as tls from "node:tls";
|
||||
import { randomUUID, createHash } from "node:crypto";
|
||||
import type { Broker, Envelope, EventHeaders } from "@novox/mesh-sdk/messaging";
|
||||
|
||||
// Two topic exchanges, kept apart on purpose (ADR 0047): tool invocations are request/reply and are
|
||||
@@ -23,40 +24,69 @@ interface Reply {
|
||||
error?: string;
|
||||
}
|
||||
|
||||
/** Connect to the mesh broker and return a Broker. `close()` tears both channel and connection down. */
|
||||
export async function connectAmqp(url: string): Promise<Broker> {
|
||||
const conn = await amqp.connect(url);
|
||||
/** A broker credential as the mesh delivers it (novox/hq ADR 0048): an amqps URL and the
|
||||
* fingerprint of the certificate the broker must present. A plain string is a bootstrap URL. */
|
||||
export interface Credential {
|
||||
url: string;
|
||||
fingerprint?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Connect to the mesh broker and return a Broker. `close()` tears both channel and connection down.
|
||||
*
|
||||
* A scoped module (novox/hq ADR 0048) passes `assumeExchanges: true`: its account may not declare
|
||||
* an exchange, and the substrate already owns them, so it declares only its own queue. A credential
|
||||
* carrying a fingerprint is dialled over amqps, pinned to exactly that certificate.
|
||||
*/
|
||||
export async function connectAmqp(
|
||||
target: string | Credential,
|
||||
opts: { assumeExchanges?: boolean } = {},
|
||||
): Promise<Broker> {
|
||||
const cred: Credential = typeof target === "string" ? { url: target } : target;
|
||||
const conn = cred.fingerprint
|
||||
? await amqp.connect(cred.url, await pinnedOptions(cred.url, cred.fingerprint))
|
||||
: await amqp.connect(cred.url);
|
||||
|
||||
// A confirm channel, so an event publish awaits the broker's ack: a publish the broker never
|
||||
// accepted (it was mid-restart, the connection dropped) fails the emit rather than vanishing —
|
||||
// at-least-once starts at the emitter, not only the consumer (ADR 0047).
|
||||
const ch = await conn.createConfirmChannel();
|
||||
await ch.assertExchange(RPC_EXCHANGE, "topic", { durable: true });
|
||||
await ch.assertExchange(EVENTS_EXCHANGE, "topic", { durable: true });
|
||||
|
||||
// The dead-letter home for poison events. A durable queue bound to `#` retains them for
|
||||
// inspection — a dead-letter exchange with no queue behind it would drop them silently, which is
|
||||
// exactly the loss the audit trail exists to prevent.
|
||||
await ch.assertExchange(DEAD_EXCHANGE, "topic", { durable: true });
|
||||
await ch.assertQueue(DEAD_EXCHANGE, { durable: true });
|
||||
await ch.bindQueue(DEAD_EXCHANGE, DEAD_EXCHANGE, "#");
|
||||
// The substrate owns the exchanges (ADR 0048). A bootstrap/admin connection declares them; a
|
||||
// scoped module assumes they exist and never tries — its account could not, and the dead-letter
|
||||
// queue behind the exchange is the substrate's to keep, not a module's.
|
||||
if (!opts.assumeExchanges) {
|
||||
await ch.assertExchange(RPC_EXCHANGE, "topic", { durable: true });
|
||||
await ch.assertExchange(EVENTS_EXCHANGE, "topic", { durable: true });
|
||||
await ch.assertExchange(DEAD_EXCHANGE, "topic", { durable: true });
|
||||
await ch.assertQueue(DEAD_EXCHANGE, { durable: true });
|
||||
await ch.bindQueue(DEAD_EXCHANGE, DEAD_EXCHANGE, "#");
|
||||
}
|
||||
|
||||
await ch.prefetch(EVENT_PREFETCH);
|
||||
|
||||
// Request/reply: one exclusive reply queue, correlationId → resolver.
|
||||
const { queue: replyQueue } = await ch.assertQueue("", { exclusive: true });
|
||||
// Request/reply is set up lazily: a consumer-only module (the audit logger) never calls a tool,
|
||||
// and its scoped account may not declare the exclusive reply queue this would otherwise need.
|
||||
const pending = new Map<string, (r: Reply) => void>();
|
||||
await ch.consume(
|
||||
replyQueue,
|
||||
(msg) => {
|
||||
if (!msg) return;
|
||||
const resolve = pending.get(msg.properties.correlationId);
|
||||
if (resolve) {
|
||||
pending.delete(msg.properties.correlationId);
|
||||
resolve(JSON.parse(msg.content.toString()) as Reply);
|
||||
}
|
||||
},
|
||||
{ noAck: true },
|
||||
);
|
||||
let replyQueue: string | undefined;
|
||||
async function ensureReply(): Promise<string> {
|
||||
if (replyQueue) return replyQueue;
|
||||
const { queue } = await ch.assertQueue("", { exclusive: true });
|
||||
replyQueue = queue;
|
||||
await ch.consume(
|
||||
queue,
|
||||
(msg) => {
|
||||
if (!msg) return;
|
||||
const resolve = pending.get(msg.properties.correlationId);
|
||||
if (resolve) {
|
||||
pending.delete(msg.properties.correlationId);
|
||||
resolve(JSON.parse(msg.content.toString()) as Reply);
|
||||
}
|
||||
},
|
||||
{ noAck: true },
|
||||
);
|
||||
return queue;
|
||||
}
|
||||
|
||||
// One durable event queue per consumer (ADR 0047: <node>.<module>.events), with many bindings and
|
||||
// a single consumer that fans out to the handlers whose pattern matches. AMQP delivers a message
|
||||
@@ -94,6 +124,7 @@ export async function connectAmqp(url: string): Promise<Broker> {
|
||||
|
||||
return {
|
||||
async request<Req, Res>(key: string, body: Req): Promise<Res> {
|
||||
const reply = await ensureReply();
|
||||
const id = randomUUID();
|
||||
const answered = new Promise<Res>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
@@ -107,7 +138,7 @@ export async function connectAmqp(url: string): Promise<Broker> {
|
||||
});
|
||||
ch.publish(RPC_EXCHANGE, key, Buffer.from(JSON.stringify(body)), {
|
||||
correlationId: id,
|
||||
replyTo: replyQueue,
|
||||
replyTo: reply,
|
||||
});
|
||||
return answered;
|
||||
},
|
||||
@@ -168,7 +199,14 @@ export async function connectAmqp(url: string): Promise<Broker> {
|
||||
if (node && mod) {
|
||||
if (!eventQueue) {
|
||||
const name = `${node}.${mod}.events`;
|
||||
await ch.assertQueue(name, { durable: true, deadLetterExchange: DEAD_EXCHANGE });
|
||||
if (opts.assumeExchanges) {
|
||||
// The mesh pre-declared this queue with its dead-letter when it issued the account: a
|
||||
// scoped account may not declare a dead-lettered queue itself (the broker refuses that
|
||||
// to a non-administrator). Passively check it is there, then bind and consume.
|
||||
await ch.checkQueue(name);
|
||||
} else {
|
||||
await ch.assertQueue(name, { durable: true, deadLetterExchange: DEAD_EXCHANGE });
|
||||
}
|
||||
eventQueue = name;
|
||||
}
|
||||
await ch.bindQueue(eventQueue, EVENTS_EXCHANGE, pattern);
|
||||
@@ -216,6 +254,56 @@ export async function connectAmqp(url: string): Promise<Broker> {
|
||||
};
|
||||
}
|
||||
|
||||
/** Normalise a certificate fingerprint to bare lower-case hex, dropping an `sha256:` prefix and
|
||||
* any colon grouping, so two spellings of the same fingerprint compare equal. */
|
||||
function normalizeFingerprint(fingerprint: string): string {
|
||||
return fingerprint.replace(/^sha256:/i, "").replace(/:/g, "").toLowerCase();
|
||||
}
|
||||
|
||||
/**
|
||||
* Socket options that pin the broker to exactly the certificate whose fingerprint the mesh
|
||||
* delivered (novox/hq ADR 0048, as the builder does). Done in two phases so a credential never
|
||||
* reaches an impostor: first a bare TLS connection that sends nothing fetches the certificate and
|
||||
* the fingerprint is checked; only then does the real connection trust *that* certificate as its
|
||||
* own authority, so the AMQP login flows solely to the broker that proved it holds the pinned key.
|
||||
* Node's `checkServerIdentity` does not run under `rejectUnauthorized: false`, so a one-phase
|
||||
* "connect then check" would have already sent the password to whoever answered.
|
||||
*/
|
||||
async function pinnedOptions(rawUrl: string, fingerprint: string): Promise<tls.ConnectionOptions> {
|
||||
const url = new URL(rawUrl);
|
||||
const host = url.hostname;
|
||||
const port = url.port ? Number(url.port) : 5671;
|
||||
|
||||
const certificate = await new Promise<tls.DetailedPeerCertificate>((resolve, reject) => {
|
||||
const socket = tls.connect({ host, port, servername: host, rejectUnauthorized: false }, () => {
|
||||
const peer = socket.getPeerCertificate(true);
|
||||
socket.destroy();
|
||||
if (!peer || !peer.raw) reject(new Error("the broker presented no certificate to pin"));
|
||||
else resolve(peer);
|
||||
});
|
||||
socket.setTimeout(15_000, () => {
|
||||
socket.destroy();
|
||||
reject(new Error("timed out fetching the broker's certificate"));
|
||||
});
|
||||
socket.on("error", reject);
|
||||
});
|
||||
|
||||
const seen = createHash("sha256").update(certificate.raw).digest("hex");
|
||||
if (seen !== normalizeFingerprint(fingerprint)) {
|
||||
throw new Error(
|
||||
`the broker's certificate (sha256:${seen}) does not match the pinned ${fingerprint} — refusing`,
|
||||
);
|
||||
}
|
||||
|
||||
const pem =
|
||||
"-----BEGIN CERTIFICATE-----\n" +
|
||||
(certificate.raw.toString("base64").match(/.{1,64}/g) ?? []).join("\n") +
|
||||
"\n-----END CERTIFICATE-----\n";
|
||||
// Trust that one certificate and nothing else; the mesh's own name is not in any public store,
|
||||
// so identity is the pin, not the hostname — checkServerIdentity is satisfied deliberately.
|
||||
return { ca: [pem], checkServerIdentity: () => undefined };
|
||||
}
|
||||
|
||||
/** Read a broker message back into an Envelope: string headers, contentType folded in, body parsed. */
|
||||
function toEnvelope<T>(msg: amqp.ConsumeMessage): Envelope<T> {
|
||||
const raw = msg.properties.headers ?? {};
|
||||
|
||||
Reference in New Issue
Block a user