feat(multi-site): live catalog replication + identical-directory signing key
The shipped join flow (v2.2.0-v2.3.0) was a one-time snapshot: a spoke's catalog never updated after joining. This adds the two pieces that were explicitly designed but missing: - Live replication: a spoke registers its own endpoint with the master right after joining (POST /api/site/spokes, Bearer join-key), receiving a pushToken. Every successful catalog write on the master now fires a fire-and-forget resync ping (utils/site_replicate.js) at every known spoke, concurrently -- one unreachable spoke never blocks or delays another (wired into the existing write-gate middleware in api_directory_admin.js). The spoke's POST /api/site/resync handler reuses the already-tested export+import path rather than applying a partial diff. - Identical directories: POST /api/site/export now best-effort includes the master's agent-signing key; a spoke adopts it via agent_keys.adopt() on both join and every resync, so every site's sso-manager can validly sign a command for any agent enrolled anywhere -- the accepted tradeoff discussed for this deployment's scale (blast radius for simplicity). New SiteSpoke model tracks registered spokes (endpoint + pushToken); registered it in models/index.js (a real bug the e2e test below caught -- SiteSpoke.list() 500'd with "Cannot read properties of null (reading 'adapter')" until the model was added to initORM's model list). Verified end-to-end against docker-compose.multisite-e2e.yml: mint join key -> join with selfUrl -> write a NEW resource on master post-join -> poll the spoke -> it shows up within a few seconds via the resync push, no manual re-join needed. MULTISITE E2E PASS. Unit tests: nodejs/tests/site_replicate.test.js (concurrent fan-out, one failing spoke doesn't block another, empty-registry and list()-throws edge cases).
This commit is contained in:
+195
-50
@@ -14,6 +14,7 @@
|
||||
// calls it with a join key), so it is defined BEFORE the auth middleware.
|
||||
|
||||
const express = require('express');
|
||||
const crypto = require('crypto');
|
||||
const { execFile } = require('child_process');
|
||||
const { promisify } = require('util');
|
||||
const os = require('os');
|
||||
@@ -25,10 +26,13 @@ const permission = require('../utils/permission');
|
||||
const conf = require('@simpleworkjs/conf');
|
||||
const { Resource, ResourceEdge } = require('../models/resource');
|
||||
const { SiteJoinKey } = require('../models/site_join_key');
|
||||
const { SiteSpoke } = require('../models/site_spoke');
|
||||
const { replicateToSpokes } = require('../utils/site_replicate');
|
||||
const User = require('../models/user');
|
||||
const { Agent } = require('../models/agent');
|
||||
const siteConfig = require('../utils/site_config');
|
||||
const { importDirectory, ldapAddArgs, baseDnFrom, siteIsFresh } = require('../utils/site_join');
|
||||
const agentKeys = require('../utils/agent_keys');
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
const router = express.Router();
|
||||
@@ -63,10 +67,16 @@ router.post('/export', async (req, res, next) => {
|
||||
const key = await SiteJoinKey.authenticate(rawKey);
|
||||
if (!key) return res.status(401).json({ status: 'error', message: 'invalid or revoked site join key' });
|
||||
|
||||
const [ldif, resources, edges] = await Promise.all([
|
||||
const [ldif, resources, edges, signingKey] = await Promise.all([
|
||||
slurpLdif(),
|
||||
Resource.list(),
|
||||
ResourceEdge.list()
|
||||
ResourceEdge.list(),
|
||||
// Best-effort: a master with no OpenBao reachable (or no key generated
|
||||
// yet) still exports successfully -- signingKey is just omitted, and
|
||||
// the spoke keeps whatever key (if any) it already has. Identical
|
||||
// signing keys across sites is a nice-to-have on top of the join
|
||||
// working at all, never a reason to fail the join.
|
||||
agentKeys.load().then((k) => k && { privateKeyPem: k.privateKeyPem, publicKeyPem: k.publicKeyPem }).catch(() => null)
|
||||
]);
|
||||
|
||||
await key.update({ use_count: (key.use_count || 0) + 1, last_used_on: Math.floor(Date.now() / 1000) }).catch(() => {});
|
||||
@@ -77,7 +87,8 @@ router.post('/export', async (req, res, next) => {
|
||||
baseDn: baseDnFrom(conf),
|
||||
ldif,
|
||||
resources: (resources || []).map(r => (r.toJSON ? r.toJSON() : r)),
|
||||
edges: (edges || []).map(e => (e.toJSON ? e.toJSON() : e))
|
||||
edges: (edges || []).map(e => (e.toJSON ? e.toJSON() : e)),
|
||||
...(signingKey ? { signingKey } : {})
|
||||
});
|
||||
} catch (e) { next(e); }
|
||||
});
|
||||
@@ -95,6 +106,70 @@ router.post('/ping', async (req, res, next) => {
|
||||
} catch (e) { next(e); }
|
||||
});
|
||||
|
||||
// ── Spoke registration (MASTER side, Bearer site-join-key; no admin session)
|
||||
// A spoke calls this right after adopting a join, handing over its own
|
||||
// reachable endpoint so the master can push live-replication resync pings to
|
||||
// it later (see utils/site_replicate.js). Idempotent on endpoint: calling it
|
||||
// again (e.g. a spoke re-registering after its own restart) returns the same
|
||||
// pushToken rather than minting a new one, so the spoke doesn't need to
|
||||
// re-learn a credential it already has.
|
||||
router.post('/spokes', async (req, res, next) => {
|
||||
try {
|
||||
const auth = req.headers.authorization || '';
|
||||
const rawKey = auth.startsWith('Bearer ') ? auth.slice(7).trim() : '';
|
||||
const key = await SiteJoinKey.authenticate(rawKey);
|
||||
if (!key) return res.status(401).json({ status: 'error', message: 'invalid or revoked site join key' });
|
||||
|
||||
const { endpoint, siteSlug } = req.body || {};
|
||||
if (!endpoint || !/^https?:\/\//.test(endpoint)) {
|
||||
return res.status(400).json({ status: 'error', message: 'a valid http(s) endpoint is required' });
|
||||
}
|
||||
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
let spoke = (await SiteSpoke.list({ where: { endpoint } }))[0];
|
||||
if (spoke) {
|
||||
await spoke.update({ siteSlug: siteSlug || spoke.siteSlug, last_seen_on: now });
|
||||
} else {
|
||||
spoke = await SiteSpoke.create({
|
||||
id: crypto.randomUUID(),
|
||||
endpoint,
|
||||
siteSlug: siteSlug || null,
|
||||
pushToken: SiteSpoke.generatePushToken(),
|
||||
created_on: now,
|
||||
last_seen_on: now
|
||||
});
|
||||
}
|
||||
logAudit('spoke_registered', { endpoint, siteSlug: spoke.siteSlug });
|
||||
res.json({ status: 'ok', pushToken: spoke.pushToken });
|
||||
} catch (e) { next(e); }
|
||||
});
|
||||
|
||||
// ── Resync (SPOKE side, Bearer pushToken; no admin session) ─────────────────
|
||||
// The receiving end of utils/site_replicate.js's fire-and-forget push: the
|
||||
// master pings this when its catalog changes. Deliberately just
|
||||
// re-runs the same export-pull + import this node already did at join time
|
||||
// (adoptFromMaster below) rather than applying a partial diff -- one tested
|
||||
// code path for "make my catalog match the master's," not two.
|
||||
router.post('/resync', async (req, res, next) => {
|
||||
try {
|
||||
const cfg = siteConfig.get();
|
||||
if (cfg.isMaster) return res.status(400).json({ status: 'error', message: 'this node is master; resync is a spoke-only operation' });
|
||||
|
||||
const auth = req.headers.authorization || '';
|
||||
const presented = auth.startsWith('Bearer ') ? auth.slice(7).trim() : '';
|
||||
if (!cfg.replicationPushToken || presented !== cfg.replicationPushToken) {
|
||||
return res.status(401).json({ status: 'error', message: 'invalid resync push token' });
|
||||
}
|
||||
if (!cfg.masterUrl || !cfg.masterJoinKey) {
|
||||
return res.status(409).json({ status: 'error', message: 'no master join credentials on file' });
|
||||
}
|
||||
|
||||
const imp = await adoptFromMaster({ masterUrl: cfg.masterUrl, joinKey: cfg.masterJoinKey });
|
||||
logAudit('resynced', { reason: (req.body && req.body.reason) || 'unspecified', resourcesCreated: imp.created, resourcesUpdated: imp.updated });
|
||||
res.json({ status: 'ok', resources: { created: imp.created, updated: imp.updated, edges: imp.edgeCount } });
|
||||
} catch (e) { next(e); }
|
||||
});
|
||||
|
||||
// ── Everything below requires an admin session ──────────────────────────────
|
||||
router.use(middleware.auth);
|
||||
router.use(async (req, res, next) => {
|
||||
@@ -161,9 +236,77 @@ router.delete('/join-keys/:id', async (req, res, next) => {
|
||||
// role. Only valid on a node that is currently the master (i.e. a fresh
|
||||
// bring-up that has not joined anything yet) — see setup.sh wiring for the
|
||||
// pre-seed timing (this pass is server endpoints only).
|
||||
// Shared by /join (first adoption) and /resync (live-replication re-pull):
|
||||
// fetch the master's export and apply it locally (catalog + LDAP). Throws on
|
||||
// any failure that should surface as a 502 to the caller.
|
||||
async function adoptFromMaster({ masterUrl, joinKey }) {
|
||||
const base = String(masterUrl).replace(/\/+$/, '');
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), 30000);
|
||||
let resp;
|
||||
try {
|
||||
resp = await fetch(base + '/api/site/export', {
|
||||
method: 'POST',
|
||||
headers: { Authorization: 'Bearer ' + joinKey, 'Content-Type': 'application/json' },
|
||||
body: '{}',
|
||||
signal: controller.signal
|
||||
});
|
||||
} finally { clearTimeout(timer); }
|
||||
|
||||
if (!resp.ok) {
|
||||
const text = (await resp.text().catch(() => '')).slice(0, 200);
|
||||
const err = new Error('master export failed: HTTP ' + resp.status + ' ' + text);
|
||||
err.httpStatus = 502;
|
||||
throw err;
|
||||
}
|
||||
const exportData = await resp.json();
|
||||
if (!exportData || exportData.status !== 'ok' || !exportData.ldif) {
|
||||
const err = new Error('master export returned no directory');
|
||||
err.httpStatus = 502;
|
||||
throw err;
|
||||
}
|
||||
|
||||
// 1. Adopt the resource catalog.
|
||||
const imp = await importDirectory({ Resource, ResourceEdge, exportData });
|
||||
|
||||
// 2. Adopt the LDAP tree. The spoke keeps its own cn=admin / base DN;
|
||||
// ldapadd -c skips existing entries, so users/groups come from master.
|
||||
let ldapNote = 'imported';
|
||||
try {
|
||||
const adminDn = conf.ldap && conf.ldap.bindDN;
|
||||
// The admin credential for the local slapd (read from config at runtime —
|
||||
// never hardcoded; named without the literal "password" keyword so secret
|
||||
// scanners don't false-positive on a variable assignment).
|
||||
const ldapCred = conf.ldap && conf.ldap.bindPassword;
|
||||
const ldifFile = path.join(os.tmpdir(), 'theta-site-join.ldif');
|
||||
fs.writeFileSync(ldifFile, exportData.ldif, 'utf8');
|
||||
const argv = ldapAddArgs({ bindDN: adminDn, ldapCred, ldifFile, ldapUrl: conf.ldap && conf.ldap.url });
|
||||
await execFileAsync(argv[0], argv.slice(1), { maxBuffer: 4 * 1024 * 1024, timeout: 120000 });
|
||||
fs.unlink(ldifFile).catch(() => {});
|
||||
} catch (e) {
|
||||
ldapNote = 'skipped/failed: ' + e.message;
|
||||
}
|
||||
|
||||
// 3. Adopt the master's agent-signing key, if it sent one (MULTI_SITE_SPEC.md
|
||||
// §2 -- identical directories). Best-effort: OpenBao being unreachable
|
||||
// here shouldn't fail a join/resync any more than it would on a
|
||||
// standalone install.
|
||||
let signingKeyNote = 'not provided by master';
|
||||
if (exportData.signingKey) {
|
||||
try {
|
||||
await agentKeys.adopt(exportData.signingKey);
|
||||
signingKeyNote = 'adopted';
|
||||
} catch (e) {
|
||||
signingKeyNote = 'failed: ' + e.message;
|
||||
}
|
||||
}
|
||||
|
||||
return { imp, ldapNote, signingKeyNote, exportData, base };
|
||||
}
|
||||
|
||||
router.post('/join', async (req, res, next) => {
|
||||
try {
|
||||
const { masterUrl, joinKey } = req.body || {};
|
||||
const { masterUrl, joinKey, selfUrl } = req.body || {};
|
||||
if (!masterUrl || !joinKey) {
|
||||
return res.status(400).json({ status: 'error', message: 'masterUrl and joinKey are required' });
|
||||
}
|
||||
@@ -181,54 +324,52 @@ router.post('/join', async (req, res, next) => {
|
||||
});
|
||||
}
|
||||
|
||||
const base = String(masterUrl).replace(/\/+$/, '');
|
||||
const controller = new AbortController();
|
||||
const timer = setTimeout(() => controller.abort(), 30000);
|
||||
let resp;
|
||||
let adopted;
|
||||
try {
|
||||
resp = await fetch(base + '/api/site/export', {
|
||||
method: 'POST',
|
||||
headers: { Authorization: 'Bearer ' + joinKey, 'Content-Type': 'application/json' },
|
||||
body: '{}',
|
||||
signal: controller.signal
|
||||
});
|
||||
} finally { clearTimeout(timer); }
|
||||
|
||||
if (!resp.ok) {
|
||||
const text = (await resp.text().catch(() => '')).slice(0, 200);
|
||||
return res.status(502).json({ status: 'error', message: 'master export failed: HTTP ' + resp.status + ' ' + text });
|
||||
}
|
||||
const exportData = await resp.json();
|
||||
if (!exportData || exportData.status !== 'ok' || !exportData.ldif) {
|
||||
return res.status(502).json({ status: 'error', message: 'master export returned no directory' });
|
||||
}
|
||||
|
||||
// 1. Adopt the resource catalog.
|
||||
const imp = await importDirectory({ Resource, ResourceEdge, exportData });
|
||||
|
||||
// 2. Adopt the LDAP tree. The spoke keeps its own cn=admin / base DN;
|
||||
// ldapadd -c skips existing entries, so users/groups come from master.
|
||||
let ldapNote = 'imported';
|
||||
try {
|
||||
const adminDn = conf.ldap && conf.ldap.bindDN;
|
||||
// The admin credential for the local slapd (read from config at runtime —
|
||||
// never hardcoded; named without the literal "password" keyword so secret
|
||||
// scanners don't false-positive on a variable assignment).
|
||||
const ldapCred = conf.ldap && conf.ldap.bindPassword;
|
||||
const ldifFile = path.join(os.tmpdir(), 'theta-site-join.ldif');
|
||||
fs.writeFileSync(ldifFile, exportData.ldif, 'utf8');
|
||||
const argv = ldapAddArgs({ bindDN: adminDn, ldapCred, ldifFile, ldapUrl: conf.ldap && conf.ldap.url });
|
||||
await execFileAsync(argv[0], argv.slice(1), { maxBuffer: 4 * 1024 * 1024, timeout: 120000 });
|
||||
fs.unlink(ldifFile).catch(() => {});
|
||||
adopted = await adoptFromMaster({ masterUrl, joinKey });
|
||||
} catch (e) {
|
||||
ldapNote = 'skipped/failed: ' + e.message;
|
||||
return res.status(e.httpStatus || 502).json({ status: 'error', message: e.message });
|
||||
}
|
||||
const { imp, ldapNote, signingKeyNote, exportData, base } = adopted;
|
||||
|
||||
// 3. Register with the master for live replication, if this node knows
|
||||
// its own reachable endpoint (selfUrl -- see setup.env's
|
||||
// CFG_SELF_DIRECTORY_URL). Best-effort: a spoke that can't/won't
|
||||
// register still joins successfully, it just won't receive live
|
||||
// resync pushes (falls back to being exactly today's one-time
|
||||
// snapshot for that spoke, not a hard failure).
|
||||
let replicationPushToken = null;
|
||||
let replicationNote = 'not registered (no selfUrl given)';
|
||||
if (selfUrl) {
|
||||
try {
|
||||
const regResp = await fetch(base + '/api/site/spokes', {
|
||||
method: 'POST',
|
||||
headers: { Authorization: 'Bearer ' + joinKey, 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ endpoint: selfUrl, siteSlug: exportData.siteSlug || cfg.siteSlug })
|
||||
});
|
||||
if (regResp.ok) {
|
||||
const regBody = await regResp.json();
|
||||
replicationPushToken = regBody.pushToken;
|
||||
replicationNote = 'registered for live replication';
|
||||
} else {
|
||||
replicationNote = 'registration failed: HTTP ' + regResp.status;
|
||||
}
|
||||
} catch (e) {
|
||||
replicationNote = 'registration failed: ' + e.message;
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Persist the spoke role (survives restarts). The join key is kept so
|
||||
// the spoke can run WAN-health checks (and, in a later layer, proxy
|
||||
// writes) against the master — it is a spoke-to-master credential, not
|
||||
// a shared secret.
|
||||
siteConfig.save({ isMaster: false, masterUrl: base, siteSlug: exportData.siteSlug || cfg.siteSlug, masterJoinKey: joinKey });
|
||||
// 4. Persist the spoke role (survives restarts). The join key is kept so
|
||||
// the spoke can run WAN-health checks against the master; the push
|
||||
// token (if registration succeeded) is what authenticates the
|
||||
// master's future resync pushes back to THIS node.
|
||||
siteConfig.save({
|
||||
isMaster: false,
|
||||
masterUrl: base,
|
||||
siteSlug: exportData.siteSlug || cfg.siteSlug,
|
||||
masterJoinKey: joinKey,
|
||||
...(replicationPushToken ? { replicationPushToken } : {})
|
||||
});
|
||||
|
||||
logAudit('joined', {
|
||||
actor: req.user.uid,
|
||||
@@ -237,7 +378,9 @@ router.post('/join', async (req, res, next) => {
|
||||
resourcesCreated: imp.created,
|
||||
resourcesUpdated: imp.updated,
|
||||
edges: imp.edgeCount,
|
||||
ldap: ldapNote
|
||||
ldap: ldapNote,
|
||||
signingKey: signingKeyNote,
|
||||
replication: replicationNote
|
||||
});
|
||||
|
||||
res.json({
|
||||
@@ -245,7 +388,9 @@ router.post('/join', async (req, res, next) => {
|
||||
message: 'Joined master site ' + base,
|
||||
siteSlug: exportData.siteSlug || cfg.siteSlug,
|
||||
resources: { created: imp.created, updated: imp.updated, edges: imp.edgeCount },
|
||||
ldap: { note: ldapNote }
|
||||
ldap: { note: ldapNote },
|
||||
signingKey: { note: signingKeyNote },
|
||||
replication: { note: replicationNote, live: !!replicationPushToken }
|
||||
});
|
||||
} catch (e) { next(e); }
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user