diff --git a/backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts b/backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts index 460a4680..076004d1 100644 --- a/backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts +++ b/backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts @@ -38,7 +38,8 @@ interface ServerHandle { async function startServer(): Promise { const server = http.createServer(); server.on('upgrade', (req, socket, head) => { - if (req.url === '/api/mesh/proxy-tunnel') { + const pathname = new URL(req.url ?? '/', 'http://localhost').pathname; + if (pathname === '/api/mesh/proxy-tunnel') { void handleMeshProxyTunnel(req, socket, head); } else { socket.destroy(); @@ -53,9 +54,9 @@ async function startServer(): Promise { }; } -function dialTunnel(port: number): Promise { +function dialTunnel(port: number, query: string = ''): Promise { return new Promise((resolve, reject) => { - const ws = new WebSocket(`ws://127.0.0.1:${port}/api/mesh/proxy-tunnel`); + const ws = new WebSocket(`ws://127.0.0.1:${port}/api/mesh/proxy-tunnel${query}`); ws.once('open', () => resolve(ws)); ws.once('error', reject); }); @@ -75,6 +76,7 @@ beforeEach(() => { delete process.env.SENCHO_MODE; // Defensive: clear any reverse dialer left over from a prior test. MeshService.getInstance().setReverseDialer(null); + MeshService.getInstance().setProxyTunnelSelfCentralNodeId(null); }); describe('handleMeshProxyTunnel', () => { @@ -149,6 +151,108 @@ describe('handleMeshProxyTunnel', () => { } }); + it('installs the central-namespace nodeId from the ?nodeId= query param', async () => { + const srv = await startServer(); + try { + const ws = await dialTunnel(srv.port, '?nodeId=14'); + await new Promise((r) => setTimeout(r, 20)); + expect((MeshService.getInstance() as unknown as { proxyTunnelSelfCentralNodeId: number | null }).proxyTunnelSelfCentralNodeId).toBe(14); + + ws.close(1000, 'test cleanup'); + await new Promise((r) => setTimeout(r, 30)); + // Tunnel close clears the value. + expect((MeshService.getInstance() as unknown as { proxyTunnelSelfCentralNodeId: number | null }).proxyTunnelSelfCentralNodeId).toBeNull(); + } finally { + await srv.close(); + } + }); + + it('does not install or reject when the nodeId query param is missing (warns and proceeds)', async () => { + const srv = await startServer(); + try { + const ws = await dialTunnel(srv.port); + await new Promise((r) => setTimeout(r, 20)); + // Upgrade succeeded (reverse dialer installed), but no nodeId + // is recorded because central did not pass it. + const dialer = (MeshService.getInstance() as unknown as { reverseDialer: unknown }).reverseDialer; + expect(dialer).not.toBeNull(); + expect((MeshService.getInstance() as unknown as { proxyTunnelSelfCentralNodeId: number | null }).proxyTunnelSelfCentralNodeId).toBeNull(); + + ws.close(1000, 'test cleanup'); + await new Promise((r) => setTimeout(r, 30)); + } finally { + await srv.close(); + } + }); + + it('ignores malformed nodeId query params (non-numeric, zero, negative, decimal, leading zeros, exponent, whitespace)', async () => { + const srv = await startServer(); + try { + // Strict regex rejects everything that is not a positive + // decimal integer with no leading zero. parseInt would have + // silently truncated `14.5` to `14`, accepted `00014`, etc. + const cases = [ + '?nodeId=abc', '?nodeId=0', '?nodeId=-3', + '?nodeId=14.5', '?nodeId=00014', '?nodeId=1e2', + '?nodeId=%2014', '?nodeId=14abc', '?nodeId=', + ]; + for (const bogus of cases) { + const ws = await dialTunnel(srv.port, bogus); + await new Promise((r) => setTimeout(r, 20)); + expect((MeshService.getInstance() as unknown as { proxyTunnelSelfCentralNodeId: number | null }).proxyTunnelSelfCentralNodeId).toBeNull(); + ws.close(1000, 'test cleanup'); + await new Promise((r) => setTimeout(r, 30)); + } + } finally { + await srv.close(); + } + }); + + it('setProxyTunnelSelfCentralNodeId warns on a non-null overwrite to a different value', () => { + const svc = MeshService.getInstance(); + const warns: string[] = []; + const origWarn = console.warn; + console.warn = (...args: unknown[]) => { warns.push(args.map(String).join(' ')); }; + try { + svc.setProxyTunnelSelfCentralNodeId(14); + svc.setProxyTunnelSelfCentralNodeId(14); // same value: no warn + svc.setProxyTunnelSelfCentralNodeId(15); // different value: warn + svc.setProxyTunnelSelfCentralNodeId(null); // clear: no warn + svc.setProxyTunnelSelfCentralNodeId(20); // install after clear: no warn + } finally { + console.warn = origWarn; + } + const overwriteWarns = warns.filter((w) => w.includes('proxyTunnelSelfCentralNodeId overwritten')); + expect(overwriteWarns).toHaveLength(1); + expect(overwriteWarns[0]).toContain('14 -> 15'); + }); + + it('a CAS-rejected reverse-dialer install does not leak the nodeId into MeshService', async () => { + const svc = MeshService.getInstance(); + // Pre-seed the reverse-dialer slot so the handler's CAS install + // fails. The contract: the nodeId installer runs ONLY after the + // CAS install succeeds; a rejected upgrade must leave the + // identity slot unchanged. + const blockingDialer = { openMeshTcpStream: () => null }; + svc.setReverseDialer(blockingDialer as unknown as Parameters[0]); + try { + const srv = await startServer(); + try { + const ws = new WebSocket(`ws://127.0.0.1:${srv.port}/api/mesh/proxy-tunnel?nodeId=14`); + const closeInfo = await new Promise<{ code: number }>((resolve, reject) => { + ws.once('close', (code) => resolve({ code })); + ws.once('error', reject); + }); + expect(closeInfo.code).toBe(1013); + expect((svc as unknown as { proxyTunnelSelfCentralNodeId: number | null }).proxyTunnelSelfCentralNodeId).toBeNull(); + } finally { + await srv.close(); + } + } finally { + svc.setReverseDialer(null); + } + }); + it('on WS error the reverse dialer slot is cleared (defensive: matches close-path teardown)', async () => { const srv = await startServer(); try { diff --git a/backend/src/__tests__/mesh-service.test.ts b/backend/src/__tests__/mesh-service.test.ts index 09b7531c..c887c017 100644 --- a/backend/src/__tests__/mesh-service.test.ts +++ b/backend/src/__tests__/mesh-service.test.ts @@ -34,6 +34,7 @@ beforeEach(() => { meshSubnet: string; networkSetupError: string | null; selfCentralNodeId: number | null; + proxyTunnelSelfCentralNodeId: number | null; }; svc.aliasCache = new Map(); svc.aliasByPort = new Map(); @@ -45,6 +46,7 @@ beforeEach(() => { svc.meshSubnet = '172.30.0.0/24'; svc.networkSetupError = null; svc.selfCentralNodeId = null; + svc.proxyTunnelSelfCentralNodeId = null; delete process.env.SENCHO_ENROLL_TOKEN; vi.restoreAllMocks(); }); @@ -903,4 +905,74 @@ describe('MeshService pilot handleAccept dispatch', () => { expect(openCross).toHaveBeenCalledOnce(); expect(openSame).not.toHaveBeenCalled(); }); + + it('handleAccept on a proxy peer uses proxyTunnelSelfCentralNodeId to route cross-node aliases correctly (R1)', async () => { + // Repro for the R1 bug: a proxy peer receives an overlay carrying + // central-namespace nodeIds (e.g., Local = 1, this peer = 14). Pre-R1 + // the peer had no selfCentralNodeId source, fell back to its local DB + // default (always 1), and falsely matched alias.nodeId=1 to its own + // selfNodeId=1 — dispatching cross-node aliases as same-node. + const svc = MeshService.getInstance(); + const internals = svc as unknown as { + proxyTunnelSelfCentralNodeId: number | null; + selfCentralNodeId: number | null; + aliasByPort: Map; + openSameNode: (t: MeshTarget, s: unknown) => Promise; + openCrossNode: (t: MeshTarget, s: unknown) => void; + }; + // Proxy peer: selfCentralNodeId is null (no SENCHO_ENROLL_TOKEN), + // the proxy-tunnel handler installed central's view of this peer. + internals.selfCentralNodeId = null; + svc.setProxyTunnelSelfCentralNodeId(14); + // Overlay alias for central's own stack (Local = nodeId 1 in + // central's namespace). + internals.aliasByPort.set(9000, { + host: 'echo.audit-mesh-central.Local.sencho', + nodeId: 1, + nodeName: 'Local', + stackName: 'audit-mesh-central', + serviceName: 'echo', + port: 9000, + }); + + const openSame = vi.spyOn(internals, 'openSameNode').mockResolvedValue(undefined); + const openCross = vi.spyOn(internals, 'openCrossNode').mockImplementation(() => undefined); + const fakeSrc = { remoteAddress: '127.0.0.1', destroy: vi.fn() } as unknown as import('net').Socket; + + await svc.handleAccept(9000, fakeSrc); + + expect(openCross).toHaveBeenCalledOnce(); + expect(openSame).not.toHaveBeenCalled(); + }); + + it('handleAccept on a proxy peer routes same-node aliases (matching the proxy-tunnel nodeId) to openSameNode', async () => { + const svc = MeshService.getInstance(); + const internals = svc as unknown as { + proxyTunnelSelfCentralNodeId: number | null; + selfCentralNodeId: number | null; + aliasByPort: Map; + openSameNode: (t: MeshTarget, s: unknown) => Promise; + openCrossNode: (t: MeshTarget, s: unknown) => void; + }; + internals.selfCentralNodeId = null; + svc.setProxyTunnelSelfCentralNodeId(14); + // Alias for a stack on this peer (nodeId 14 in central's namespace). + internals.aliasByPort.set(9002, { + host: 'echo.audit-mesh-proxy.sencho-test-03.sencho', + nodeId: 14, + nodeName: 'sencho-test-03', + stackName: 'audit-mesh-proxy', + serviceName: 'echo', + port: 9002, + }); + + const openSame = vi.spyOn(internals, 'openSameNode').mockResolvedValue(undefined); + const openCross = vi.spyOn(internals, 'openCrossNode').mockImplementation(() => undefined); + const fakeSrc = { remoteAddress: '127.0.0.1', destroy: vi.fn() } as unknown as import('net').Socket; + + await svc.handleAccept(9002, fakeSrc); + + expect(openSame).toHaveBeenCalledOnce(); + expect(openCross).not.toHaveBeenCalled(); + }); }); diff --git a/backend/src/services/MeshProxyTunnelDialer.ts b/backend/src/services/MeshProxyTunnelDialer.ts index 50c1472c..3a0c8561 100644 --- a/backend/src/services/MeshProxyTunnelDialer.ts +++ b/backend/src/services/MeshProxyTunnelDialer.ts @@ -186,7 +186,12 @@ export class MeshProxyTunnelDialer extends EventEmitter { console.log(`[MeshProxyDialer:diag] dialing node=${nodeId} url=${sanitizeForLog(target.apiUrl)}`.replace(/[\n\r]/g, '')); } - const wsUrl = httpUrlToWs(target.apiUrl) + '/api/mesh/proxy-tunnel'; + // Pass the peer's nodeId in central's namespace as a query param so + // the remote MeshService can dispatch its own `handleAccept` correctly: + // overlay aliases carry central-namespace nodeIds, and without this + // the peer falls back to its local DB default (always 1) and treats + // cross-node aliases as same-node. + const wsUrl = httpUrlToWs(target.apiUrl) + `/api/mesh/proxy-tunnel?nodeId=${nodeId}`; let ws: WebSocket; try { ws = new WebSocket(wsUrl, { diff --git a/backend/src/services/MeshService.ts b/backend/src/services/MeshService.ts index b9dcc313..4fb77ba8 100644 --- a/backend/src/services/MeshService.ts +++ b/backend/src/services/MeshService.ts @@ -71,7 +71,8 @@ export type MeshActivityType = | 'mesh.override.preserved' | 'probe.ok' | 'probe.fail' | 'forwarder.listen' | 'forwarder.unlisten' | 'forwarder.error' - | 'proxy-tunnel.open.ok' | 'proxy-tunnel.open.fail' | 'proxy-tunnel.close'; + | 'proxy-tunnel.open.ok' | 'proxy-tunnel.open.fail' | 'proxy-tunnel.close' + | 'mesh.proxy_tunnel.identify'; export interface MeshActivityEvent { ts: number; @@ -232,6 +233,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { // pilot's own local DB id (always 1) inverts dispatch. Null on central // (fallback to getDefaultNodeId()). Populated from SENCHO_ENROLL_TOKEN. private selfCentralNodeId: number | null = null; + // On a proxy-mode peer, central's DB id for this node, communicated + // through the `?nodeId=` query param on the `/api/mesh/proxy-tunnel` WS + // upgrade. Same purpose as `selfCentralNodeId` but for proxy peers, + // where there is no SENCHO_ENROLL_TOKEN to read at boot. Takes + // precedence in `handleAccept`'s self-id resolution because the active + // upstream tunnel is the most authoritative source. Cleared on tunnel + // close. + private proxyTunnelSelfCentralNodeId: number | null = null; private constructor() { super(); @@ -1316,7 +1325,15 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { try { src.destroy(); } catch { /* ignore */ } return; } - const selfNodeId = this.selfCentralNodeId ?? NodeRegistry.getInstance().getDefaultNodeId(); + // Resolution order: active proxy-tunnel install > boot-time enroll + // token > local DB default. The proxy-tunnel value is the most + // authoritative when present because the upstream central just told + // this peer how it sees it; the enroll token covers pilot mode; the + // default-node fallback covers central itself. + const selfNodeId = + this.proxyTunnelSelfCentralNodeId + ?? this.selfCentralNodeId + ?? NodeRegistry.getInstance().getDefaultNodeId(); if (target.nodeId === selfNodeId) { await this.openSameNode(target, src); } else { @@ -1418,6 +1435,40 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { return true; } + /** + * Install (or clear) the central-namespace nodeId communicated by the + * upstream central at proxy-tunnel upgrade. Called by the proxy-tunnel + * WS handler; cleared on disconnect. Consumed by `handleAccept` to + * dispatch cross-node aliases correctly on proxy peers. + * + * The caller is implicitly single-tenant: it runs only after + * `setReverseDialer`'s CAS install succeeds (slot already guarded). A + * second non-null install while a different non-null value is present + * indicates a misconfigured deployment, so we warn loudly instead of + * silently overwriting. + */ + public setProxyTunnelSelfCentralNodeId(nodeId: number | null): void { + if ( + nodeId !== null + && this.proxyTunnelSelfCentralNodeId !== null + && this.proxyTunnelSelfCentralNodeId !== nodeId + ) { + console.warn( + `[MeshService] proxyTunnelSelfCentralNodeId overwritten ${this.proxyTunnelSelfCentralNodeId} -> ${nodeId}; concurrent proxy-tunnel install or misconfigured deployment`, + ); + } + if (nodeId !== null && this.proxyTunnelSelfCentralNodeId !== nodeId) { + this.logActivity({ + source: 'mesh', level: 'info', type: 'mesh.proxy_tunnel.identify', + message: `proxy-tunnel: this node is nodeId=${nodeId} in central's namespace`, + }); + } + // Null-clear (tunnel teardown) is intentionally not logged: it + // would double the entries on every reconnect cycle without adding + // signal beyond the existing `proxy-tunnel.close` event. + this.proxyTunnelSelfCentralNodeId = nodeId; + } + private async dialMeshTcpStream(target: MeshTarget): Promise { if (this.reverseDialer) { return this.reverseDialer.openMeshTcpStream({ diff --git a/backend/src/websocket/meshProxyTunnel.ts b/backend/src/websocket/meshProxyTunnel.ts index 0abbf8e2..78784785 100644 --- a/backend/src/websocket/meshProxyTunnel.ts +++ b/backend/src/websocket/meshProxyTunnel.ts @@ -48,14 +48,38 @@ export async function handleMeshProxyTunnel(req: IncomingMessage, socket: Duplex return reject(socket, 404, 'Not Found'); } + // Central appends `?nodeId=` so this peer can tag + // its own MeshService with the right nodeId for `handleAccept` dispatch. + // The value is unsigned, but the Bearer credential at the upgrade has + // already proven the caller is the upstream central (auth + scope gate + // in `upgradeHandler.ts`); the query param's trust ceiling is the + // token's trust ceiling. + // Strict regex (no parseInt) so `?nodeId=14.5`, `?nodeId=00014`, + // `?nodeId=1e2`, leading whitespace, etc. are rejected rather than + // silently truncated. + let peerNodeId: number | null = null; + try { + const parsed = new URL(req.url ?? '/', `http://${req.headers.host || 'localhost'}`); + const raw = parsed.searchParams.get('nodeId'); + if (raw != null && /^[1-9][0-9]*$/.test(raw)) { + const n = Number(raw); + if (Number.isSafeInteger(n)) peerNodeId = n; + } + } catch { + // Malformed URL; treat as missing param. + } + if (peerNodeId === null) { + console.warn('[MeshProxy] proxy-tunnel upgrade missing or malformed nodeId query param; reverse-direction mesh dispatch will be undefined until central upgrades'); + } + await new Promise((resolve) => { wss.handleUpgrade(req, socket as Parameters[1], head, (ws) => { - void attachSwitchboard(ws).finally(resolve); + void attachSwitchboard(ws, peerNodeId).finally(resolve); }); }); } -async function attachSwitchboard(ws: WebSocket): Promise { +async function attachSwitchboard(ws: WebSocket, peerNodeId: number | null): Promise { let switchboard: TcpStreamSwitchboard | null = null; let meshServiceCleanup: (() => void) | null = null; @@ -87,8 +111,17 @@ async function attachSwitchboard(ws: WebSocket): Promise { switchboard = null; return; } + // Install central's view of this peer's nodeId so handleAccept + // dispatches cross-node aliases correctly. Done after setReverseDialer + // succeeds so a CAS-rejected tunnel does not leak nodeId state. + if (peerNodeId !== null) { + meshService.setProxyTunnelSelfCentralNodeId(peerNodeId); + } meshServiceCleanup = () => { meshService.setReverseDialer(null, localDialer); + if (peerNodeId !== null) { + meshService.setProxyTunnelSelfCentralNodeId(null); + } }; } catch (err) { if (isDebugEnabled()) {