// Phase 1 PULL peering gate (the reverse of mixedTopologyGate's push/promote). // mixedTopologyGate proves dev -> t1 PROMOTE (push to /api/ingest). This gate // proves the PULL path (/admin/dev-deploy/pull -> the peer's signed // /api/journal) across the same multi-tenant boundary, which nothing else // covers: e2e.js exercises pull only single-instance (SQLite MAIN <-> MAIN), // never across the Postgres tenant boundary or tenant<->tenant. // // Topology: STANDALONE dev (SQLite MAIN :3000) + tenants t1, t2 on the // multi-tenant Postgres "prod" (:3002). We verify: // 1. REVERSE PULL (mixed topology): t1 PULLS dev's journal -> dev's ops land // in t1 (applied >= 1), proving the host-bound HMAC works on the GET pull // path (signedFetch targetHost = dev's host), not just the push path. // 2. ISOLATION ON PULL: t2's journal is UNCHANGED by t1 pulling from dev -- // a pull into one tenant never leaks into a sibling tenant. // 3. IDEMPOTENT RE-PULL: pulling again with the advanced inbound anchor // returns "nothing to pull" (no duplicate apply). // 4. TENANT<->TENANT PULL: t2 PULLS t1's journal (same Postgres server, two // tenants) -> t1's OWN ops land in t2 while dev is untouched. apiJournal // serves only first-party ops (source_env_id == self), so t2 receives // t1's ops, NOT the dev ops t1 itself pulled in step 1 (no transitive // relay / no loops). // // Run: node test/pullGate.js (MAIN :3000 + PG :3002 up, dev-deploy installed // on MAIN and per-tenant on t1/t2). Self-skips if either is down. const http = require("http"); const net = require("net"); const DEV = { port: 3000, host: "localhost:3000", base: "http://localhost:3000" }; const PROD_PORT = 3002; const ADMIN_PW = "AdminP@ss1"; const RUN = Date.now(); const DEV_PROBE = "pull_dev_" + RUN; const T1_PROBE = "pull_t1_" + RUN; const thost = (t) => t + ".localhost.localdomain:" + PROD_PORT; const jars = { dev: {}, t1: {}, t2: {} }; let pass = 0; let fail = 0; const ok = (cond, msg) => { if (cond) { pass++; console.log(" PASS " + msg); } else { fail++; console.log(" FAIL " + msg); } }; const portOpen = (port) => { return new Promise((resolve) => { const s = net.connect(port, "127.0.0.1"); const done = (up) => { s.destroy(); resolve(up); }; s.setTimeout(1000); s.on("connect", () => done(true)); s.on("timeout", () => done(false)); s.on("error", () => done(false)); }); }; const HOSTOF = (inst) => (inst === "dev" ? DEV.host : thost(inst)); const PORTOF = (inst) => (inst === "dev" ? DEV.port : PROD_PORT); const ADMIN = (inst) => (inst === "dev" ? "admin@local" : "admin@" + inst + ".local"); const BASEOF = (inst) => (inst === "dev" ? DEV.base : "http://" + thost(inst)); const storeCookies = (inst, headers) => { const sc = headers["set-cookie"]; if (!sc) { return; } for (const line of sc) { const pair = line.split(";")[0]; const eq = pair.indexOf("="); if (eq > 0) { jars[inst][pair.slice(0, eq).trim()] = pair.slice(eq + 1).trim(); } } }; const request = (inst, method, path, opts) => { const options = opts || {}; return new Promise((resolve, reject) => { const headers = Object.assign({ Host: options.host || HOSTOF(inst) }, options.headers || {}); const jar = jars[inst]; if (jar && Object.keys(jar).length > 0) { headers["Cookie"] = Object.keys(jar).map((k) => k + "=" + jar[k]).join("; "); } let data = null; if (options.body) { data = new URLSearchParams(options.body).toString(); headers["Content-Type"] = "application/x-www-form-urlencoded"; headers["Content-Length"] = Buffer.byteLength(data); } const r = http.request({ host: "127.0.0.1", port: PORTOF(inst), method: method, path: path, headers: headers }, (resp) => { storeCookies(inst, resp.headers); let body = ""; resp.on("data", (c) => { body += c; }); resp.on("end", () => resolve({ status: resp.statusCode, headers: resp.headers, body: body })); }); r.on("error", reject); if (data !== null) { r.write(data); } r.end(); }); }; const csrfOf = (h) => { const m = h.match(/name="_csrf" value="([^"]+)"/); return m ? m[1] : ""; }; const envIdOf = (h) => { const m = h.match(/env_id<\/strong> is ([0-9a-f-]{36})<\/code>/); return m ? m[1] : ""; }; const opCountOf = (h) => { const m = h.match(/Ops recorded<\/th>(\d+)<\/td>/); return m ? parseInt(m[1], 10) : null; }; const secretOf = (h) => { const m = h.match(/

([0-9a-f]+)<\/p>/); return m ? m[1] : ""; }; // pull redirects 302 with ?msg=... (success) or ?err=... (failure); the message // reads "pulled N ops (A applied, E errors, C conflicts)" or "nothing to pull". const redirectMsg = (res) => { const loc = res.headers.location || ""; const m = loc.match(/[?&](?:msg|err)=([^&]+)/); return m ? decodeURIComponent(m[1]) : ""; }; const pulledCounts = (msg) => { const m = msg.match(/pulled (\d+) ops \((\d+) applied, (\d+) errors, (\d+) conflicts\)/); return m ? { n: +m[1], applied: +m[2], errors: +m[3], conflicts: +m[4] } : null; }; const peerIdFor = (html, envId) => { for (const row of html.split("")) { if (row.indexOf("" + envId + "") >= 0) { const m = row.match(/name="peer_id" value="(\d+)"/); if (m) { return m[1]; } } } return null; }; // Tracked-tables admin page renders one row per tracked table: // NAMEUUID8LOCAL_ID... // Map a probe table's NAME to its LOCAL_ID so cleanup can drop it (the source // copy AND any copy materialized by a pull) without tracking create locations. const localIdByName = async (inst, name) => { const html = (await request(inst, "GET", "/admin/dev-deploy/tables")).body; const re = new RegExp("" + name + "\\s*[0-9a-f]{8}\\s*(\\d+)"); const m = html.match(re); return m ? m[1] : null; }; const authed = async (inst) => /true/.test((await request(inst, "GET", "/auth/authenticated")).body); const bootstrap = async (inst) => { const lp = await request(inst, "GET", "/auth/login"); await request(inst, "POST", "/auth/login", { body: { email: ADMIN(inst), password: ADMIN_PW, _csrf: csrfOf(lp.body) } }); if (await authed(inst)) { return true; } const cp = await request(inst, "GET", "/auth/create_first_user"); const cc = csrfOf(cp.body); if (cc) { await request(inst, "POST", "/auth/create_first_user", { body: { email: ADMIN(inst), password: ADMIN_PW, default_language: "en", _csrf: cc } }); } return await authed(inst); }; const ddCsrf = async (inst) => csrfOf((await request(inst, "GET", "/admin/dev-deploy/peers")).body); const clearPeer = async (inst, envId) => { const pid = peerIdFor((await request(inst, "GET", "/admin/dev-deploy/peers")).body, envId); if (pid) { await request(inst, "POST", "/admin/dev-deploy/peers/delete", { body: { peer_id: pid, _csrf: await ddCsrf(inst) } }); } }; // Returns the issued shared secret when the peer is added WITHOUT one (the // adding side generates it); returns "" when an existing secret is supplied. const addPeer = async (inst, envId, label, baseUrl, secret) => { const body = { env_id: envId, label: label, base_url: baseUrl, _csrf: await ddCsrf(inst) }; if (secret) { body.existing_secret = secret; } const res = await request(inst, "POST", "/admin/dev-deploy/peers/add", { body: body }); return secretOf(res.body); }; // Create a table via Saltcorn's own POST /table (journaled by dev-deploy as a // create_table op authored by THIS instance). The _csrf comes from dev-deploy's // self-rendered page: a freshly created themeless tenant 500s on /table/new // (sendWrap with no layout), but POST /table only mutates + redirects, so it // succeeds. Returns true on the 302 -> /table/ success. const createTable = async (inst, name) => { const csrf = await ddCsrf(inst); const res = await request(inst, "POST", "/table", { body: { name: name, _csrf: csrf } }); return res.status >= 300 && res.status < 400 && /\/table\/\d+/.test(res.headers.location || ""); }; const dropTableByName = async (inst, name) => { const id = await localIdByName(inst, name); if (id) { await request(inst, "POST", "/table/delete/" + id, { body: { _csrf: await ddCsrf(inst) } }); } }; const opCount = async (inst) => opCountOf((await request(inst, "GET", "/admin/dev-deploy/")).body); const doPull = async (inst, peerId) => { return await request(inst, "POST", "/admin/dev-deploy/pull", { body: { peer_id: peerId, _csrf: await ddCsrf(inst) } }); }; const run = async () => { ok(await bootstrap("dev"), "dev (standalone MAIN :3000) admin session"); ok(await bootstrap("t1"), "t1 admin session"); ok(await bootstrap("t2"), "t2 admin session"); const devEnv = envIdOf((await request("dev", "GET", "/admin/dev-deploy/peers")).body); const t1Env = envIdOf((await request("t1", "GET", "/admin/dev-deploy/peers")).body); const t2Env = envIdOf((await request("t2", "GET", "/admin/dev-deploy/peers")).body); ok(/^[0-9a-f-]{36}$/.test(devEnv), "dev env_id (" + devEnv + ")"); ok(t1Env && t2Env && t1Env !== t2Env && t1Env !== devEnv, "dev/t1/t2 have distinct env_ids"); // Fresh pairings (idempotent reruns): a new peer_id => a fresh inbound anchor // => the pull starts from the beginning of the source journal. await clearPeer("dev", t1Env); await clearPeer("t1", devEnv); await clearPeer("t1", t2Env); await clearPeer("t2", t1Env); // Pair dev <-> t1 so t1 can pull from dev: t1 needs dev as a peer (base_url + // secret to initiate); dev needs t1 as a peer (to AUTH t1's signed pull, which // peerAuth looks up by t1's env_id). dev issues the secret. const secretDevT1 = await addPeer("dev", t1Env, "prod-t1", BASEOF("t1")); ok(/^[0-9a-f]{64}$/.test(secretDevT1), "dev<->t1 paired (shared secret issued)"); await addPeer("t1", devEnv, "dev", BASEOF("dev"), secretDevT1); // Give dev a brand-new op to serve: a uniquely named table guarantees an op // t1 has never seen (status "applied", not "already_applied"), so the assert // holds on every rerun regardless of accumulated history. ok(await createTable("dev", DEV_PROBE), "dev created a probe table to seed its journal"); const t1Before = await opCount("t1"); const t2Before = await opCount("t2"); // (1) REVERSE PULL: t1 pulls dev's journal. const t1DevPeer = peerIdFor((await request("t1", "GET", "/admin/dev-deploy/peers")).body, devEnv); ok(!!t1DevPeer, "t1 has a peer_id for dev (" + t1DevPeer + ")"); const pull1 = await doPull("t1", t1DevPeer); const pull1Msg = redirectMsg(pull1); const c1 = pulledCounts(pull1Msg); ok(pull1.status === 302, "t1 pull from dev returned 302 (" + pull1.status + ")"); ok(c1 !== null, "t1 pull reported a pulled-ops summary (\"" + pull1Msg + "\")"); ok(c1 && c1.applied >= 1, "t1 APPLIED >= 1 op pulled from dev (applied=" + (c1 ? c1.applied : "?") + ")"); ok(c1 && c1.errors === 0 && c1.conflicts === 0, "t1 pull had no errors/conflicts"); const t1After1 = await opCount("t1"); const t2After1 = await opCount("t2"); // (1) lands in t1. ok(t1After1 > t1Before, "t1 journal grew from the pull (" + t1Before + " -> " + t1After1 + ")"); // (2) ISOLATION ON PULL: t2 untouched. ok(t2After1 === t2Before, "t2 journal UNCHANGED by t1 pulling from dev (" + t2Before + " -> " + t2After1 + ")"); // (3) IDEMPOTENT RE-PULL: anchor advanced, nothing new to fetch. const pull2 = await doPull("t1", t1DevPeer); const pull2Msg = redirectMsg(pull2); ok(/nothing to pull/.test(pull2Msg), "re-pull from dev is idempotent (\"" + pull2Msg + "\")"); const t1After2 = await opCount("t1"); ok(t1After2 === t1After1, "t1 journal UNCHANGED by the idempotent re-pull (" + t1After1 + " -> " + t1After2 + ")"); // (4) TENANT<->TENANT PULL: pair t1<->t2 (t1 issues), seed t1 with its own op, // then t2 pulls t1. t1's apiJournal serves only t1-authored ops, so t2 gets // t1's probe -- NOT the dev ops t1 pulled in step 1 (no transitive relay). const secretT1T2 = await addPeer("t1", t2Env, "prod-t2", BASEOF("t2")); ok(/^[0-9a-f]{64}$/.test(secretT1T2), "t1<->t2 paired (shared secret issued)"); await addPeer("t2", t1Env, "prod-t1", BASEOF("t1"), secretT1T2); ok(await createTable("t1", T1_PROBE), "t1 created a probe table to seed its own journal"); const devBefore4 = await opCount("dev"); const t2Before4 = await opCount("t2"); const t2T1Peer = peerIdFor((await request("t2", "GET", "/admin/dev-deploy/peers")).body, t1Env); ok(!!t2T1Peer, "t2 has a peer_id for t1 (" + t2T1Peer + ")"); const pull3 = await doPull("t2", t2T1Peer); const pull3Msg = redirectMsg(pull3); const c3 = pulledCounts(pull3Msg); ok(c3 && c3.applied >= 1, "t2 APPLIED >= 1 op pulled from t1 (tenant<->tenant; applied=" + (c3 ? c3.applied : "?") + ")"); ok(c3 && c3.errors === 0 && c3.conflicts === 0, "t2 pull from t1 had no errors/conflicts"); const t2After4 = await opCount("t2"); const devAfter4 = await opCount("dev"); ok(t2After4 > t2Before4, "t2 journal grew from pulling t1 (" + t2Before4 + " -> " + t2After4 + ")"); ok(devAfter4 === devBefore4, "dev journal UNCHANGED by the t2<->t1 pull (" + devBefore4 + " -> " + devAfter4 + ")"); // Cleanup (best-effort): drop each probe ONLY on its SOURCE instance (dev // authored DEV_PROBE, t1 authored T1_PROBE), never on the puller. A drop is a // first-party op too: dropping on the source makes the journal create->drop // LINEAR, so the next run's pull replays it in order and cleans up the pulled // copy with no conflict. Dropping the PULLED copy instead would author a // competing local op, and the next run (replaying the source's create) would // legitimately conflict with it -- a self-inflicted divergence, not a bug. await dropTableByName("dev", DEV_PROBE); await dropTableByName("t1", T1_PROBE); await clearPeer("dev", t1Env); await clearPeer("t1", devEnv); await clearPeer("t1", t2Env); await clearPeer("t2", t1Env); console.log("\nRESULT: " + pass + " passed, " + fail + " failed"); process.exit(fail ? 1 : 0); }; const main = async () => { if (!(await portOpen(DEV.port))) { console.log("SKIP: standalone dev not reachable on 127.0.0.1:" + DEV.port); process.exit(0); } if (!(await portOpen(PROD_PORT))) { console.log("SKIP: Postgres multi-tenant prod not reachable on 127.0.0.1:" + PROD_PORT); process.exit(0); } await run(); }; main().catch((e) => { console.error("PULL GATE ERROR:", e); process.exit(2); });