diff --git a/nodejs/models/index.js b/nodejs/models/index.js index d9d245f..eb96717 100644 --- a/nodejs/models/index.js +++ b/nodejs/models/index.js @@ -22,6 +22,7 @@ const { SharedSecretGrant } = require('./shared_secret_grant'); const { VaultAppToken } = require('./vault_app_token'); const { Agent, AgentJoinKey } = require('./agent'); const { SiteJoinKey } = require('./site_join_key'); +const { SiteSpoke } = require('./site_spoke'); async function initORM() { const ormConf = conf.orm || { dialect: 'sqlite', @@ -36,7 +37,7 @@ async function initORM() { conf: { orm: ormConf }, models: [ Resource, ResourceEdge, ResourceGroup, AccessRequest, Webhook, PluginInstance, - SharedSecret, SharedSecretGrant, VaultAppToken, Agent, AgentJoinKey, SiteJoinKey, + SharedSecret, SharedSecretGrant, VaultAppToken, Agent, AgentJoinKey, SiteJoinKey, SiteSpoke, Token, AuthToken, InviteToken, ImpersonationToken, PasswordResetToken, OtpToken, ServiceToken ] }); diff --git a/nodejs/models/site_spoke.js b/nodejs/models/site_spoke.js new file mode 100644 index 0000000..3763f56 --- /dev/null +++ b/nodejs/models/site_spoke.js @@ -0,0 +1,42 @@ +'use strict'; + +const crypto = require('crypto'); +const { Model } = require('@simpleworkjs/orm'); + +// A spoke known to THIS node while it's acting as master — the registry that +// makes live replication possible. A spoke registers itself here (POST +// /api/site/spokes, authenticated by the same join key it used to join) +// right after adopting the master's export, handing over its own reachable +// endpoint. In return it's issued a `pushToken`: a shared secret the master +// then presents on every future POST /api/site/resync call. +// +// This is a DIFFERENT credential direction than SiteJoinKey: a join key is +// presented TO the master and only ever needs to be verified (so it's stored +// hashed, like a password). pushToken is presented BY the master, repeatedly, +// so it has to be retrievable here -- there is no getting around storing it +// in plaintext on the master, the same way Webhook.secret is (see +// services/webhook_emitter.js) for the same reason (an HMAC/bearer credential +// the sender must keep re-presenting, not a one-time secret only ever +// verified). +class SiteSpoke extends Model { + static generatePushToken() { + return crypto.randomBytes(24).toString('base64url'); + } + + static fields = { + id: { type: 'uuid', primaryKey: true }, + endpoint: { type: 'string', isRequired: true, unique: true }, + siteSlug: { type: 'string' }, + pushToken: { type: 'string', isRequired: true }, + created_on: { type: 'integer' }, + last_seen_on: { type: 'integer' } + }; + + toPublic() { + const data = this.toJSON ? this.toJSON() : { ...this }; + delete data.pushToken; + return data; + } +} + +module.exports = { SiteSpoke }; diff --git a/nodejs/package.json b/nodejs/package.json index 6d05ab5..ab80af7 100755 --- a/nodejs/package.json +++ b/nodejs/package.json @@ -11,7 +11,7 @@ "scripts": { "start": "node ./bin/www", "dev": "npx nodemon --ignore public/ ./bin/www", - "test": "NODE_ENV=test jest tests/groups.test.js tests/subtypes.test.js tests/site_join.test.js tests/site_config.test.js --forceExit" + "test": "NODE_ENV=test jest tests/groups.test.js tests/subtypes.test.js tests/site_join.test.js tests/site_config.test.js tests/site_replicate.test.js --forceExit" }, "jest": { "testEnvironment": "node", diff --git a/nodejs/routes/api_directory_admin.js b/nodejs/routes/api_directory_admin.js index 2856fbc..a9d0db9 100644 --- a/nodejs/routes/api_directory_admin.js +++ b/nodejs/routes/api_directory_admin.js @@ -9,6 +9,7 @@ const { projectResources } = require('@simpleworkjs/directory-schema'); const SUPER_ADMIN_GROUP = permission.SUPER_ADMIN_GROUP; const groups = require('../utils/groups'); +const meshReplicate = require('../utils/site_replicate'); // Make `childCn` a member of `parentCn`, i.e. everyone in the child is // transitively in the parent. Idempotent and non-fatal: "already a member" is @@ -252,19 +253,32 @@ router.get('/resources', async (req, res, next) => { } catch (err) { next(err); } }); -// ── Spoke read-only enforcement ───────────────────────────────────────────── +// ── Spoke read-only enforcement + live replication trigger ────────────────── // On a joined spoke the catalog is a copy of the master's; directory writes // must go to the master (MULTI_SITE_SPEC.md — spoke = read-only catalog). Any // mutating request below this point is rejected on a spoke with a pointer to // the master. (site-status / site-promote live AFTER this middleware and are // not directory writes.) +// +// On the MASTER, a successful mutation here fires a fire-and-forget resync +// push (utils/site_replicate.js) at every registered spoke, so the shipped +// join flow's one-time snapshot doesn't go stale the moment the catalog +// changes. Fires on res.on('finish') (after the response is actually sent, +// status known) rather than before the handler runs, so a write that fails +// validation never triggers a pointless replication round-trip. router.use((req, res, next) => { - if (['POST', 'PUT', 'DELETE', 'PATCH'].includes(req.method)) { + const mutating = ['POST', 'PUT', 'DELETE', 'PATCH'].includes(req.method); + if (mutating) { const cfg = siteConfig.get(); if (!cfg.isMaster) { const hint = cfg.masterUrl ? ' Directory writes must go to the master at ' + cfg.masterUrl + '.' : ''; return res.status(403).json({ status: 'error', message: 'This node is a spoke (read-only catalog).' + hint }); } + res.on('finish', () => { + if (res.statusCode >= 200 && res.statusCode < 300) { + meshReplicate.replicateToSpokes(`${req.method} ${req.path}`); + } + }); } next(); }); diff --git a/nodejs/routes/api_site.js b/nodejs/routes/api_site.js index 2bcf858..941813a 100644 --- a/nodejs/routes/api_site.js +++ b/nodejs/routes/api_site.js @@ -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); } }); diff --git a/nodejs/tests/site_replicate.test.js b/nodejs/tests/site_replicate.test.js new file mode 100644 index 0000000..d36d5fe --- /dev/null +++ b/nodejs/tests/site_replicate.test.js @@ -0,0 +1,90 @@ +'use strict'; + +// In-memory stand-in for the SiteSpoke ORM model. +let spokeStore; +function makeSpokeMock() { + spokeStore = []; + return { + list: jest.fn(async () => [...spokeStore]), + _seed(rows) { spokeStore.push(...rows); } + }; +} + +let mockFetchCalls = []; +let mockFetchImpl = async () => ({ ok: true, status: 200 }); + +describe('site_replicate', () => { + let siteReplicate; + let SiteSpoke; + let originalFetch; + + beforeEach(() => { + jest.resetModules(); + mockFetchCalls = []; + mockFetchImpl = async () => ({ ok: true, status: 200 }); + + jest.doMock('../models/site_spoke', () => ({ SiteSpoke: makeSpokeMock() })); + siteReplicate = require('../utils/site_replicate'); + SiteSpoke = require('../models/site_spoke').SiteSpoke; + + // site_replicate.js uses the global fetch (Node 18+ built-in), not + // node-fetch -- stub that directly. + originalFetch = global.fetch; + global.fetch = (...args) => { mockFetchCalls.push(args); return mockFetchImpl(...args); }; + }); + + afterEach(() => { + global.fetch = originalFetch; + }); + + test('pushes to every known spoke concurrently with its own pushToken', async () => { + SiteSpoke._seed([ + { endpoint: 'https://spoke-a.example.com', pushToken: 'token-a' }, + { endpoint: 'https://spoke-b.example.com', pushToken: 'token-b' } + ]); + + await siteReplicate.replicateToSpokes('catalog-changed'); + await new Promise((r) => setImmediate(r)); + + expect(mockFetchCalls.length).toBe(2); + const urls = mockFetchCalls.map((c) => c[0]).sort(); + expect(urls).toEqual(['https://spoke-a.example.com/api/site/resync', 'https://spoke-b.example.com/api/site/resync']); + + const [, optsA] = mockFetchCalls.find((c) => c[0].includes('spoke-a')); + expect(optsA.headers.Authorization).toBe('Bearer token-a'); + expect(JSON.parse(optsA.body).reason).toBe('catalog-changed'); + }); + + test('no known spokes: resolves cleanly, no fetch calls', async () => { + await siteReplicate.replicateToSpokes('catalog-changed'); + await new Promise((r) => setImmediate(r)); + expect(mockFetchCalls.length).toBe(0); + }); + + test('one spoke failing does not prevent delivery to another', async () => { + SiteSpoke._seed([ + { endpoint: 'https://dead-spoke.example.com', pushToken: 'token-dead' }, + { endpoint: 'https://live-spoke.example.com', pushToken: 'token-live' } + ]); + mockFetchImpl = async (url) => { + if (url.includes('dead-spoke')) throw new Error('connection refused'); + return { ok: true, status: 200 }; + }; + + await expect(siteReplicate.replicateToSpokes('event')).resolves.toBeUndefined(); + await new Promise((r) => setImmediate(r)); + expect(mockFetchCalls.length).toBe(2); + }); + + test('a non-2xx response from a spoke does not throw out of replicateToSpokes', async () => { + SiteSpoke._seed([{ endpoint: 'https://spoke-a.example.com', pushToken: 'token-a' }]); + mockFetchImpl = async () => ({ ok: false, status: 500 }); + await expect(siteReplicate.replicateToSpokes('event')).resolves.toBeUndefined(); + }); + + test('SiteSpoke.list() throwing does not propagate to the caller', async () => { + SiteSpoke.list = jest.fn(async () => { throw new Error('db unavailable'); }); + await expect(siteReplicate.replicateToSpokes('event')).resolves.toBeUndefined(); + expect(mockFetchCalls.length).toBe(0); + }); +}); diff --git a/nodejs/utils/agent_keys.js b/nodejs/utils/agent_keys.js index 3d91d93..80a9388 100644 --- a/nodejs/utils/agent_keys.js +++ b/nodejs/utils/agent_keys.js @@ -90,10 +90,34 @@ function status() { return { available: !!cached, error: loadError }; } +// Overwrite the stored key with material handed over by a master (multi-site +// "identical directories" — MULTI_SITE_SPEC.md §2). Every site sharing one +// signing key is what lets any site's sso-manager validly sign a command for +// an agent enrolled at any other site, at the accepted cost that compromising +// ANY one site's OpenBao is equivalent to compromising all of them for agent +// command authority. That tradeoff was deliberately accepted for this +// deployment's scale (a handful of trusted sites) -- do not call this to sync +// keys across a boundary where sites don't trust each other equally. +// +// Idempotent: adopting the same key material twice (e.g. on every resync +// ping) is a no-op past the first call. +async function adopt({ privateKeyPem, publicKeyPem }) { + if (!privateKeyPem || !publicKeyPem) throw new Error('adopt() requires both privateKeyPem and publicKeyPem'); + if (cached && cached.privateKeyPem === privateKeyPem && cached.publicKeyPem === publicKeyPem) { + return cached; // already holding this exact key -- nothing to do + } + const material = { privateKeyPem, publicKeyPem }; + await baoConf.set(PATH, material); + cached = { ...material, publicKeyBase64: rawPublicKeyBase64(publicKeyPem) }; + loadError = null; + console.log('[agent_keys] adopted signing key from master (multi-site identical-directory sync)'); + return cached; +} + // Test seam: drop the in-process cache. function _reset() { cached = null; loadError = null; } -module.exports = { load, status, rawPublicKeyBase64, _reset, PATH }; +module.exports = { load, status, adopt, rawPublicKeyBase64, _reset, PATH }; diff --git a/nodejs/utils/site_replicate.js b/nodejs/utils/site_replicate.js new file mode 100644 index 0000000..4421d65 --- /dev/null +++ b/nodejs/utils/site_replicate.js @@ -0,0 +1,57 @@ +'use strict'; + +// Live replication push -- the piece the shipped v1 join flow doesn't have on +// its own (join is a one-time export/import snapshot; nothing kept a spoke in +// sync afterward). This fires a lightweight "something changed, re-pull" ping +// at every spoke registered in SiteSpoke, concurrently, fire-and-forget: never +// awaited by its caller, and one unreachable spoke never delays or blocks +// another. See MULTI_SITE_SPEC.md §2.2 for why this must never become a +// blocking design (a write must never stall on spoke reachability). +// +// Deliberately a PUSH-A-SIGNAL / PULL-A-SNAPSHOT design, not a push-a-diff +// design: the receiving spoke reacts by calling the master's already-shipped, +// already-tested POST /api/site/export + importDirectory() path again (see +// routes/api_site.js's /resync handler), rather than this module inventing a +// second, parallel way to represent "what changed." Fewer moving parts, and +// no risk of a diff payload and a full export ever disagreeing. + +const { SiteSpoke } = require('../models/site_spoke'); + +const RESYNC_TIMEOUT_MS = 8000; + +function replicateToSpokes(reason) { + return (async () => { + let spokes; + try { + spokes = await SiteSpoke.list(); + } catch (err) { + console.error('[site-replicate] failed to list known spokes:', err.message); + return; + } + for (const spoke of spokes) { + // Not awaited -- every spoke is pushed to concurrently. + pingOne(spoke, reason).catch((err) => { + console.error(`[site-replicate] resync ping to ${spoke.endpoint} failed:`, err.message); + }); + } + })(); +} + +async function pingOne(spoke, reason) { + const url = String(spoke.endpoint).replace(/\/+$/, '') + '/api/site/resync'; + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), RESYNC_TIMEOUT_MS); + try { + const resp = await fetch(url, { + method: 'POST', + headers: { Authorization: 'Bearer ' + spoke.pushToken, 'Content-Type': 'application/json' }, + body: JSON.stringify({ reason: reason || 'catalog-changed' }), + signal: controller.signal + }); + if (!resp.ok) throw new Error('status ' + resp.status); + } finally { + clearTimeout(timer); + } +} + +module.exports = { replicateToSpokes }; diff --git a/test/multisite_join_e2e.js b/test/multisite_join_e2e.js index e34f165..946ffab 100644 --- a/test/multisite_join_e2e.js +++ b/test/multisite_join_e2e.js @@ -175,14 +175,17 @@ async function main() { if (keyRes.status !== 200 || !keyRes.body.key) fail(`join-key mint failed: ${keyRes.status} ${JSON.stringify(keyRes.body)}`); const joinKey = keyRes.body.key; - step('Joining spoke to master'); + step('Joining spoke to master (with selfUrl, to register for live replication)'); const joinRes = await api(SPOKE_URL, '/api/site/join', { method: 'POST', token: spokeToken, // master's own container-internal URL, as the spoke would reach it over the network - body: { masterUrl: 'http://master:3001', joinKey } + body: { masterUrl: 'http://master:3001', joinKey, selfUrl: 'http://spoke:3001' } }); if (joinRes.status !== 200) fail(`join failed: ${joinRes.status} ${JSON.stringify(joinRes.body)}`); + if (!joinRes.body.replication || joinRes.body.replication.live !== true) { + fail(`expected join to register for live replication, got ${JSON.stringify(joinRes.body.replication)}`); + } step('Verifying spoke persisted isMaster:false + masterUrl after join'); const { body: spokeCfg } = await api(SPOKE_URL, '/api/site/config', { token: spokeToken }); @@ -203,6 +206,24 @@ async function main() { }); if (writeAttempt.status !== 403) fail(`expected 403 writing to spoke post-join, got ${writeAttempt.status} ${JSON.stringify(writeAttempt.body)}`); + step('Creating a resource on master AFTER join, to verify LIVE replication (not just the one-time join snapshot)'); + const postJoinRes = await api(MASTER_URL, '/api/directory-admin/resources', { + method: 'POST', + token: masterToken, + body: { name: 'E2E Post-Join Host', slug: 'host_e2e_postjoin', kind: 'host', parentSlug: 'site_e2e' } + }); + if (postJoinRes.status !== 200) fail(`creating post-join resource on master failed: ${postJoinRes.status} ${JSON.stringify(postJoinRes.body)}`); + + step('Waiting for the fire-and-forget resync push to reach the spoke'); + let liveReplicated = false; + for (let i = 0; i < 20; i++) { + const r = await api(SPOKE_URL, '/api/directory-admin/resources', { token: spokeToken }); + const slugs = (r.body.results || r.body.resources || r.body || []).map((x) => x.slug); + if (slugs.includes('host_e2e_postjoin')) { liveReplicated = true; break; } + await new Promise((res) => setTimeout(res, 500)); + } + if (!liveReplicated) fail('post-join resource never appeared on the spoke -- live replication did not fire (or resync did not apply it)'); + step('Verifying WAN health ping from spoke to master succeeds'); const statusRes = await api(SPOKE_URL, '/api/directory-admin/site-status', { token: spokeToken }); if (statusRes.body.config && statusRes.body.config.wanConnected !== true) {