feat(mesh): symmetric WS dial for proxy-mode mesh peers (#1066)

* chore(mesh): foundation for symmetric callback dial

Adds the data-plane scaffolding that the symmetric callback dial fix
builds on:
- mesh_centrals table for peer-side bootstrap material
- MeshCentralRegistry service (upsert/getActive/clear/markUsed/markRejected)
- PilotTunnelManager kind discriminator and replaceOrRegisterProxyBridge
- mesh_proxy_callback_bootstrap capability registration
- MeshProxyTunnelDialer reason-tagged proxy-bridge-down events from a
  single tearDownBridge emission point
- Reactive redial scheduler that skips idle and auth_failed reasons

* feat(mesh): add reverse-direction activity log entries (closes R1-B)

acceptReverseLocal now emits route.resolve.ok with direction=reverse on
connect ack and route.resolve.fail with direction=reverse plus
reason=container_not_found / connect_error pre-connect. Post-connect
close/error stays silent. Reuses existing event types via the new
details.direction discriminator so frontend filters are unaffected.

* feat(mesh): add peer-to-central callback dial path (closes R1-A2)

Closes the architectural gap where proxy-mode mesh peers could not
re-establish their tunnel to central after any non-idle bridge teardown
(idle close, network blip, central restart, peer reboot). Central remains
the hub for the data plane; the change is purely about WS initiation.

Symmetric WS initiation, asymmetric protocol roles. Central retains
PilotTunnelBridge ownership; peer retains TcpStreamSwitchboard +
reverseDialer ownership. Central bootstraps callback credentials over
the first authenticated central-initiated mesh tunnel via a one-shot
mesh_handshake JSON frame; peer persists the material in a new
mesh_centrals SQLite table and dials central's new
/api/mesh/proxy-tunnel-from-peer endpoint when local cross-node traffic
needs a bridge and none is live.

Mesh_tunnel JWT (HS256, signed with auth_jwt_secret) carries scope, audience,
issuer (central instance id), peer api_token fingerprint, kid. Validation
on inbound peer dial: algorithm pin, signature, scope, audience, instance,
time bounds, node existence and mode, fingerprint match. Failures return
HTTP 401 with a machine-readable reason; peer routes the response per a
clear-vs-keep cache matrix.

Triggers proactive bootstrap on mesh-enable and api_token rotation; central
startup fans out to mesh-enabled proxy-mode nodes with mesh_stacks rows
(throttled, fire-and-forget). Reactive redial on non-idle bridge loss.

Capability-gated handshake send (mesh_proxy_callback_bootstrap) makes the
upgrade path safe against older peers in mixed-version fleets.

Adds peer-side /api/system/pilot-tunnels centralCallback diag block,
bounded counter metrics for bootstrap and dial events. SENCHO_PRIMARY_URL
preflight warning when unset on a central with mesh-enabled proxy nodes.

Tested with unit suites for the validation chain, registry, manager, and
both dialers; integration tests for bootstrap E2E (asserts protocol-role
invariant), api_token rotation, instance id change, version skew, and
pilot-mode regression.

* fix(mesh): green CI on the symmetric callback branch

Two independent CI failures, both surgical:

1. Backend tests (11 fails): four mesh test files called setupTestDb in
   beforeEach. setupTestDb does not reset the DatabaseService singleton,
   so the per-test afterEach rm of the previous tmpdir left the singleton
   connection pointing at a deleted file. The next beforeEach's line-55
   write threw SQLITE_READONLY_DBMOVED on Linux. Windows file-lock
   semantics hid this locally. Hoist setupTestDb / cleanupTestDb to
   file-scope beforeAll / afterAll; per-test state resets stay in
   beforeEach. Matches the convention in the eight mesh test files that
   already pass.

2. CodeQL (4 high alerts): js/insufficient-password-hash flagged
   sha256(api_token) at four sites. The api_token is a 256-bit opaque
   bearer (sen_sk_-prefixed), not a human password; sha256 is the
   correct fingerprint primitive for binding the mesh_tunnel JWT to a
   specific token. Add the two production files plus the two test
   files that mint the fingerprint to the existing path-scoped
   query-filter for that rule.

* fix(mesh): drop unused afterEach import and revert dead codeql config

ESLint flagged afterEach as unused in mesh-central-registry.test.ts:1
after the previous commit hoisted setup/teardown to file-scope
beforeAll/afterAll. Remove from the vitest import line.

Revert the codeql-config.yml additions from the previous commit. The
paths: sub-key under query-filters > exclude is not a documented CodeQL
feature and silently no-ops. The four js/insufficient-password-hash
alerts on api_token fingerprinting are tracked as dismissed false
positives in the GitHub Security tab rather than via dead config.
This commit is contained in:
Anso
2026-05-16 14:58:19 -04:00
committed by GitHub
parent 94fa42f73c
commit cf618dd866
33 changed files with 3456 additions and 34 deletions
@@ -0,0 +1,15 @@
import { describe, it, expect } from 'vitest';
import { getActiveCapabilities, applyPilotModeCapabilityFilter, enableCapability } from '../services/CapabilityRegistry';
describe('mesh_proxy_callback_bootstrap capability', () => {
it('is registered in the default CAPABILITIES list', () => {
expect(getActiveCapabilities()).toContain('mesh_proxy_callback_bootstrap');
});
it('is NOT filtered out in pilot mode (pilots can also be bootstrap targets)', () => {
applyPilotModeCapabilityFilter();
expect(getActiveCapabilities()).toContain('mesh_proxy_callback_bootstrap');
enableCapability('host-console');
enableCapability('self-update');
});
});
@@ -0,0 +1,187 @@
/**
* End-to-end protocol-role invariant for the symmetric mesh callback dial.
*
* The whole point of Task 8 + Task 10 is the asymmetry of registration even
* though the wire protocol becomes symmetric:
*
* - Central is the loopback HTTP host. Every bridge central registers
* (pilot-initiated or peer-initiated dial-back) is a
* `PilotTunnelBridge` parked in `PilotTunnelManager`.
* - The peer (proxy-mode remote) never runs a `PilotTunnelBridge`; on the
* peer side a `TcpStreamSwitchboard` rides the same WS and the local
* `MeshService.reverseDialer` slot is held while the WS is alive.
*
* This file exercises the central side of that invariant end-to-end using
* the same in-process fixture pattern as `mesh-proxy-tunnel-from-peer.test.ts`:
* a fully signed callback JWT lands at `/api/mesh/proxy-tunnel-from-peer`,
* and we assert what central produced is a `PilotTunnelBridge` (not a
* switchboard) in the manager. The matching mesh_handshake side (central
* pushing the JWT) is covered in `mesh-proxy-tunnel-dialer-handshake.test.ts`
* and `mesh-proxy-tunnel-first-frame.test.ts`; here we lock in the registry
* outcome that ties those two halves together.
*/
import http from 'http';
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import WebSocket from 'ws';
import jwt from 'jsonwebtoken';
import { createHash } from 'crypto';
import type { AddressInfo } from 'net';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { MeshProxyTunnelDialer } from '../services/MeshProxyTunnelDialer';
import { MeshCentralRegistry } from '../services/MeshCentralRegistry';
import { PilotTunnelManager } from '../services/PilotTunnelManager';
import { PilotTunnelBridge } from '../services/PilotTunnelBridge';
let tmpDir: string;
let handleMeshProxyTunnelFromPeerUpgrade: typeof import('../websocket/meshProxyTunnelFromPeer').handleMeshProxyTunnelFromPeerUpgrade;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
interface ServerHandle {
server: http.Server;
port: number;
close: () => Promise<void>;
}
async function startCentralServer(): Promise<ServerHandle> {
const server = http.createServer();
server.on('upgrade', (req, socket, head) => {
const pathname = new URL(req.url ?? '/', 'http://localhost').pathname;
if (pathname === '/api/mesh/proxy-tunnel-from-peer') {
handleMeshProxyTunnelFromPeerUpgrade(req, socket, head);
} else {
socket.destroy();
}
});
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', () => resolve()));
const port = (server.address() as AddressInfo).port;
return {
server,
port,
close: () => new Promise<void>((resolve) => server.close(() => resolve())),
};
}
const CANONICAL_ORIGIN = 'https://central.example.com';
const INSTANCE_ID = 'bootstrap-integration-central';
const PEER_API_TOKEN = 'peer-token-bootstrap';
let secret: string;
let srv: ServerHandle;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ handleMeshProxyTunnelFromPeerUpgrade } = await import('../websocket/meshProxyTunnelFromPeer'));
({ DatabaseService } = await import('../services/DatabaseService'));
srv = await startCentralServer();
});
afterAll(async () => {
if (srv) await srv.close();
delete process.env.SENCHO_PRIMARY_URL;
cleanupTestDb(tmpDir);
});
beforeEach(() => {
process.env.SENCHO_PRIMARY_URL = CANONICAL_ORIGIN;
MeshProxyTunnelDialer.resetForTest();
MeshCentralRegistry.resetForTest();
PilotTunnelManager.resetForTest();
const db = DatabaseService.getInstance();
secret = db.getGlobalSettings().auth_jwt_secret;
db.setSystemState('instance_id', INSTANCE_ID);
});
function seedPeerNode(): number {
const db = DatabaseService.getInstance();
const id = db.addNode({
name: `peer-bs-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
type: 'remote',
mode: 'proxy',
api_url: 'https://peer.example.com',
api_token: PEER_API_TOKEN,
compose_dir: '/tmp',
is_default: false,
});
db.setNodeMeshEnabled(id, true);
return id;
}
function mintCallbackJwt(peerNodeId: number): string {
const fp = createHash('sha256').update(PEER_API_TOKEN).digest('hex').slice(0, 16);
const nowSec = Math.floor(Date.now() / 1000);
return jwt.sign(
{
sub: String(peerNodeId),
iss: INSTANCE_ID,
aud: CANONICAL_ORIGIN,
scope: 'mesh_tunnel',
iat: nowSec,
exp: nowSec + 3600,
kid: 'v1',
peer_token_fp: fp,
},
secret,
{ algorithm: 'HS256' },
);
}
describe('Mesh bootstrap E2E (protocol-role invariant)', () => {
it('after peer dial-back, central registers a PilotTunnelBridge in PilotTunnelManager', async () => {
// This is the symmetric callback half of the protocol: the peer has
// already received a bootstrap JWT in some prior session and is now
// dialing back into central. On central, the resulting WS must
// produce a PilotTunnelBridge (not a TcpStreamSwitchboard) and the
// manager must hold it under the peer's nodeId.
const peerNodeId = seedPeerNode();
const token = mintCallbackJwt(peerNodeId);
const ws = await new Promise<WebSocket>((resolve, reject) => {
const s = new WebSocket(`ws://127.0.0.1:${srv.port}/api/mesh/proxy-tunnel-from-peer`, {
headers: { authorization: `Bearer ${token}` },
});
s.once('open', () => resolve(s));
s.once('error', reject);
s.once('unexpected-response', (_req, res) => reject(new Error(`upgrade rejected ${res.statusCode}`)));
});
// Allow the handler's async bridge.start() + manager registration to land.
await new Promise((r) => setTimeout(r, 60));
const bridge = PilotTunnelManager.getInstance().getBridge(peerNodeId);
expect(bridge).not.toBeNull();
// The manager only narrows to MeshTunnelHandle in its public API,
// but the underlying object must be the central-side bridge class.
expect(bridge).toBeInstanceOf(PilotTunnelBridge);
try { ws.close(1000, 'test cleanup'); } catch { /* ignore */ }
await new Promise((r) => setTimeout(r, 30));
});
it('a successful dial-back persists the JWT scope on the bridge (mesh_tunnel scope)', async () => {
// Cross-check: the JWT carries scope=mesh_tunnel, the central
// ingress accepted it, so the bridge that lands in the manager is a
// proxy-mode bridge (not a pilot-agent tunnel). We assert via the
// dialer's hasBridge surface to confirm the proxy registry side
// sees nothing (the dial-back lives in the manager, not in
// MeshProxyTunnelDialer's outbound map).
const peerNodeId = seedPeerNode();
const token = mintCallbackJwt(peerNodeId);
const ws = await new Promise<WebSocket>((resolve, reject) => {
const s = new WebSocket(`ws://127.0.0.1:${srv.port}/api/mesh/proxy-tunnel-from-peer`, {
headers: { authorization: `Bearer ${token}` },
});
s.once('open', () => resolve(s));
s.once('error', reject);
s.once('unexpected-response', (_req, res) => reject(new Error(`upgrade rejected ${res.statusCode}`)));
});
await new Promise((r) => setTimeout(r, 60));
// The dialer is the central-initiated outbound side. A peer-initiated
// dial-back lands in PilotTunnelManager only; the dialer's own
// bridges map stays empty for this nodeId.
expect(MeshProxyTunnelDialer.getInstance().hasBridge(peerNodeId)).toBe(false);
expect(PilotTunnelManager.getInstance().getBridge(peerNodeId)).not.toBeNull();
try { ws.close(1000, 'test cleanup'); } catch { /* ignore */ }
await new Promise((r) => setTimeout(r, 30));
});
});
@@ -0,0 +1,122 @@
import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { DatabaseService } from '../services/DatabaseService';
import { MeshCentralRegistry } from '../services/MeshCentralRegistry';
let sharedTmpDir: string;
beforeAll(async () => { sharedTmpDir = await setupTestDb(); });
afterAll(() => cleanupTestDb(sharedTmpDir));
describe('mesh_centrals schema', () => {
it('creates the mesh_centrals table on initSchema', () => {
const db = DatabaseService.getInstance();
const row = db.getDb().prepare(
"SELECT name FROM sqlite_master WHERE type='table' AND name='mesh_centrals'"
).get();
expect(row).toBeDefined();
});
it('mesh_centrals has the expected columns', () => {
const db = DatabaseService.getInstance();
const cols = db.getDb().prepare("PRAGMA table_info('mesh_centrals')").all() as Array<{ name: string }>;
const names = cols.map(c => c.name).sort();
expect(names).toEqual([
'callback_jwt',
'central_api_url',
'central_instance_id',
'jwt_expires_at',
'jwt_issued_at',
'last_bootstrap_at',
'last_reject_reason',
'last_rejected_at',
'last_used_at',
]);
});
});
describe('MeshCentralRegistry', () => {
beforeEach(() => {
MeshCentralRegistry.resetForTest();
// Per-test cleanup of the mesh_centrals table so tests are order-independent.
// The DB itself is created once per file in the file-scope beforeAll above;
// resetting setupTestDb per test would invalidate the DatabaseService
// singleton's open connection (SQLITE_READONLY_DBMOVED on Linux CI).
DatabaseService.getInstance().getDb().prepare('DELETE FROM mesh_centrals').run();
});
const sampleMaterial = (overrides: Partial<{
centralInstanceId: string;
centralApiUrl: string;
callbackJwt: string;
jwtIssuedAt: number;
jwtExpiresAt: number;
}> = {}) => ({
centralInstanceId: 'inst-uuid-1',
centralApiUrl: 'https://central.example.com',
callbackJwt: 'eyJhbGciOiJIUzI1NiJ9.fake.token',
jwtIssuedAt: 1_700_000_000,
jwtExpiresAt: 1_700_000_000 + 365 * 24 * 3600,
...overrides,
});
it('upsert inserts a new row when none exists', () => {
MeshCentralRegistry.getInstance().upsert(sampleMaterial());
const row = MeshCentralRegistry.getInstance().getActive();
expect(row?.centralInstanceId).toBe('inst-uuid-1');
expect(row?.centralApiUrl).toBe('https://central.example.com');
expect(row?.lastBootstrapAt).toBeGreaterThan(0);
});
it('upsert overwrites when same instance id is provided', () => {
const reg = MeshCentralRegistry.getInstance();
reg.upsert(sampleMaterial({ callbackJwt: 'old.jwt' }));
reg.upsert(sampleMaterial({ callbackJwt: 'new.jwt' }));
expect(reg.getActive()?.callbackJwt).toBe('new.jwt');
});
it('getActive returns most-recently-bootstrapped row when multiple exist and warns once', () => {
const reg = MeshCentralRegistry.getInstance();
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
reg.upsert(sampleMaterial({ centralInstanceId: 'old', jwtIssuedAt: 1 }));
DatabaseService.getInstance().getDb().prepare(`
INSERT INTO mesh_centrals VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
`).run('new', 'https://other.example.com', 'jwt2', 2, 999, Date.now() + 1000);
expect(reg.getActive()?.centralInstanceId).toBe('new');
expect(warn).toHaveBeenCalledOnce();
expect(warn.mock.calls[0][0]).toMatch(/multiple central rows/);
reg.getActive();
expect(warn).toHaveBeenCalledOnce();
warn.mockRestore();
});
it('getActive returns null when table is empty', () => {
expect(MeshCentralRegistry.getInstance().getActive()).toBeNull();
});
it('clearForInstance removes only the named instance', () => {
const reg = MeshCentralRegistry.getInstance();
reg.upsert(sampleMaterial({ centralInstanceId: 'a' }));
reg.upsert(sampleMaterial({ centralInstanceId: 'b' }));
reg.clearForInstance('a');
const remaining = DatabaseService.getInstance().getDb().prepare(
"SELECT central_instance_id FROM mesh_centrals"
).all() as Array<{ central_instance_id: string }>;
expect(remaining.map(r => r.central_instance_id)).toEqual(['b']);
});
it('markUsed updates last_used_at, only on success', () => {
const reg = MeshCentralRegistry.getInstance();
reg.upsert(sampleMaterial());
reg.markUsed('inst-uuid-1');
expect(reg.getActive()?.lastUsedAt).toBeGreaterThan(0);
});
it('markRejected stores reason and timestamp', () => {
const reg = MeshCentralRegistry.getInstance();
reg.upsert(sampleMaterial());
reg.markRejected('inst-uuid-1', 'token_fingerprint_mismatch');
const row = reg.getActive();
expect(row?.lastRejectedAt).toBeGreaterThan(0);
expect(row?.lastRejectReason).toBe('token_fingerprint_mismatch');
});
});
@@ -0,0 +1,84 @@
/**
* Central instance id rotation invariant.
*
* `MeshCentralRegistry.upsert` is the single source of truth for the
* locally-cached callback material. When central regenerates its
* `instance_id` (factory reset, DB swap, container rebuild without volume),
* the next bootstrap frame arrives with a different `centralInstanceId`. The
* registry MUST:
* - persist the new row,
* - emit a single `central-instance-changed` event so subscribers (loud
* log, dialer reset) can react.
*
* Failing this invariant means the peer would keep dialing back with the
* old JWT (rejected with `instance_mismatch`) without anyone noticing.
*/
import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { MeshCentralRegistry } from '../services/MeshCentralRegistry';
import { DatabaseService } from '../services/DatabaseService';
let tmpDir: string;
beforeAll(async () => { tmpDir = await setupTestDb(); });
afterAll(() => cleanupTestDb(tmpDir));
describe('Central instance id change', () => {
beforeEach(() => {
MeshCentralRegistry.resetForTest();
// Per-test cleanup of the mesh_centrals table. The DB lives once per
// file (beforeAll); resetting setupTestDb per test would invalidate
// the DatabaseService singleton's connection on Linux.
DatabaseService.getInstance().getDb().prepare('DELETE FROM mesh_centrals').run();
});
it('drops the old row and accepts the new one with a central-instance-changed event', () => {
const reg = MeshCentralRegistry.getInstance();
reg.upsert({
centralInstanceId: 'old-uuid',
centralApiUrl: 'https://central.example.com',
callbackJwt: 'jwt-old',
jwtIssuedAt: 1,
jwtExpiresAt: 999999,
});
const events: Array<{ previousInstanceId: string; newInstanceId: string }> = [];
reg.on('central-instance-changed', (e: { previousInstanceId: string; newInstanceId: string }) => events.push(e));
reg.upsert({
centralInstanceId: 'new-uuid',
centralApiUrl: 'https://central.example.com',
callbackJwt: 'jwt-new',
jwtIssuedAt: 2,
jwtExpiresAt: 999999,
});
expect(events).toHaveLength(1);
expect(events[0].previousInstanceId).toBe('old-uuid');
expect(events[0].newInstanceId).toBe('new-uuid');
// getActive returns the most-recently-bootstrapped row.
expect(reg.getActive()?.centralInstanceId).toBe('new-uuid');
expect(reg.getActive()?.callbackJwt).toBe('jwt-new');
});
it('an upsert with the same instance id does NOT emit central-instance-changed', () => {
const reg = MeshCentralRegistry.getInstance();
reg.upsert({
centralInstanceId: 'stable-uuid',
centralApiUrl: 'https://central.example.com',
callbackJwt: 'jwt-1',
jwtIssuedAt: 1,
jwtExpiresAt: 999999,
});
const events: Array<unknown> = [];
reg.on('central-instance-changed', (e: unknown) => events.push(e));
reg.upsert({
centralInstanceId: 'stable-uuid',
centralApiUrl: 'https://central.example.com',
callbackJwt: 'jwt-2-rotated',
jwtIssuedAt: 2,
jwtExpiresAt: 999999,
});
expect(events).toHaveLength(0);
// Same id, the JWT material is overwritten in place.
expect(reg.getActive()?.callbackJwt).toBe('jwt-2-rotated');
});
});
@@ -0,0 +1,164 @@
/**
* `MeshProxyTunnelDialer.maybeSendBootstrap`: capability-gated
* `mesh_handshake` send.
*
* Central mints a signed `mesh_tunnel`-scoped JWT bound to the peer's
* `api_token` fingerprint and pushes it as the first text frame on the
* fresh WS, but only when the peer advertises the
* `mesh_proxy_callback_bootstrap` capability AND a canonical central
* origin (`SENCHO_PRIMARY_URL`) is configured. All other paths must
* remain silent (no frame) so the legacy peer-side first-frame state
* machine can fall straight through to TCP traffic.
*/
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import jwt from 'jsonwebtoken';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { MeshProxyTunnelDialer } from '../services/MeshProxyTunnelDialer';
import { NodeRegistry } from '../services/NodeRegistry';
import { OFFLINE_META } from '../services/CapabilityRegistry';
import { DatabaseService } from '../services/DatabaseService';
let tmpDir: string;
const CENTRAL_INSTANCE_ID = 'central-instance-for-handshake-test';
beforeAll(async () => {
tmpDir = await setupTestDb();
// Seed system_state.instance_id (LicenseService.initialize() is not
// invoked in the test harness; mesh handshake requires this value).
DatabaseService.getInstance().setSystemState('instance_id', CENTRAL_INSTANCE_ID);
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
describe('MeshProxyTunnelDialer capability-gated handshake', () => {
let nodeSeq = 0;
beforeEach(() => {
MeshProxyTunnelDialer.resetForTest();
process.env.SENCHO_PRIMARY_URL = 'https://central.example.com';
nodeSeq += 1;
});
afterEach(() => {
delete process.env.SENCHO_PRIMARY_URL;
vi.restoreAllMocks();
});
function fakeWs(): { send: ReturnType<typeof vi.fn>; sentFrames: unknown[] } {
const sentFrames: unknown[] = [];
return {
send: vi.fn((data: string) => { sentFrames.push(JSON.parse(data)); }),
sentFrames,
};
}
function seedProxyNode(): number {
const db = DatabaseService.getInstance();
const id = db.addNode({
name: `peer-handshake-${nodeSeq}-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
type: 'remote',
mode: 'proxy',
api_url: 'https://peer.example.com',
api_token: 'peer-token-for-fp',
compose_dir: '/tmp',
is_default: false,
});
db.setNodeMeshEnabled(id, true);
return id;
}
type DialerPrivate = { maybeSendBootstrap: (nodeId: number, ws: unknown) => Promise<void> };
it('sends mesh_handshake when peer advertises mesh_proxy_callback_bootstrap', async () => {
vi.spyOn(NodeRegistry.getInstance(), 'fetchMetaForNode').mockResolvedValue({
...OFFLINE_META,
version: '0.79.0',
capabilities: ['stacks', 'mesh_proxy_callback_bootstrap'],
online: true,
startedAt: Date.now(),
});
const id = seedProxyNode();
const ws = fakeWs();
await (MeshProxyTunnelDialer.getInstance() as unknown as DialerPrivate).maybeSendBootstrap(id, ws);
expect(ws.send).toHaveBeenCalledOnce();
expect(ws.sentFrames[0]).toMatchObject({ t: 'mesh_handshake', v: 1 });
});
it('skips mesh_handshake when peer lacks the capability', async () => {
vi.spyOn(NodeRegistry.getInstance(), 'fetchMetaForNode').mockResolvedValue({
...OFFLINE_META,
version: '0.78.0',
capabilities: ['stacks'],
online: true,
startedAt: Date.now(),
});
const id = seedProxyNode();
const ws = fakeWs();
await (MeshProxyTunnelDialer.getInstance() as unknown as DialerPrivate).maybeSendBootstrap(id, ws);
expect(ws.send).not.toHaveBeenCalled();
});
it('skips mesh_handshake when meta fetch returns offline shape', async () => {
vi.spyOn(NodeRegistry.getInstance(), 'fetchMetaForNode').mockResolvedValue({
...OFFLINE_META,
online: false,
});
const id = seedProxyNode();
const ws = fakeWs();
await (MeshProxyTunnelDialer.getInstance() as unknown as DialerPrivate).maybeSendBootstrap(id, ws);
expect(ws.send).not.toHaveBeenCalled();
});
it('skips mesh_handshake when SENCHO_PRIMARY_URL is unset (fail-safe)', async () => {
delete process.env.SENCHO_PRIMARY_URL;
vi.spyOn(NodeRegistry.getInstance(), 'fetchMetaForNode').mockResolvedValue({
...OFFLINE_META,
version: '0.79.0',
capabilities: ['mesh_proxy_callback_bootstrap'],
online: true,
startedAt: Date.now(),
});
const id = seedProxyNode();
const ws = fakeWs();
await (MeshProxyTunnelDialer.getInstance() as unknown as DialerPrivate).maybeSendBootstrap(id, ws);
expect(ws.send).not.toHaveBeenCalled();
});
it('signs the JWT with auth_jwt_secret using HS256 and includes the required claims', async () => {
vi.spyOn(NodeRegistry.getInstance(), 'fetchMetaForNode').mockResolvedValue({
...OFFLINE_META,
version: '0.79.0',
capabilities: ['mesh_proxy_callback_bootstrap'],
online: true,
startedAt: Date.now(),
});
const id = seedProxyNode();
const ws = fakeWs();
await (MeshProxyTunnelDialer.getInstance() as unknown as DialerPrivate).maybeSendBootstrap(id, ws);
const frame = ws.sentFrames[0] as {
meshTunnelJwt: string;
centralApiUrl: string;
centralInstanceId: string;
peerNodeId: number;
jwtExpiresAt: number;
};
expect(frame.centralApiUrl).toBe('https://central.example.com');
expect(frame.centralInstanceId).toBe(CENTRAL_INSTANCE_ID);
expect(frame.peerNodeId).toBe(id);
expect(typeof frame.jwtExpiresAt).toBe('number');
const decoded = jwt.decode(frame.meshTunnelJwt, { complete: true });
expect(decoded?.header.alg).toBe('HS256');
const payload = decoded?.payload as Record<string, unknown>;
expect(payload).toMatchObject({
sub: String(id),
iss: CENTRAL_INSTANCE_ID,
aud: 'https://central.example.com',
scope: 'mesh_tunnel',
kid: 'v1',
});
expect(typeof payload.peer_token_fp).toBe('string');
expect((payload.peer_token_fp as string).length).toBe(16);
});
});
@@ -0,0 +1,64 @@
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { EventEmitter } from 'events';
import { MeshProxyTunnelDialer } from '../services/MeshProxyTunnelDialer';
type Reason = 'idle' | 'remote_closed' | 'network_error' | 'protocol_error' | 'auth_failed';
describe('MeshProxyTunnelDialer reason-tagged proxy-bridge-down', () => {
beforeEach(() => MeshProxyTunnelDialer.resetForTest());
it('emits reason="idle" when runIdleCheck closes a bridge', () => {
const dialer = MeshProxyTunnelDialer.resetForTest(1);
const reasons: Reason[] = [];
dialer.on('proxy-bridge-down', (_id: number, reason: Reason) => reasons.push(reason));
const fakeBridge = new EventEmitter() as unknown as { close: ReturnType<typeof vi.fn>; getActiveStreamCount: () => number };
fakeBridge.close = vi.fn();
fakeBridge.getActiveStreamCount = () => 0;
(dialer as unknown as { bridges: Map<number, unknown>; idleSince: Map<number, number> }).bridges.set(7, fakeBridge);
(dialer as unknown as { bridges: Map<number, unknown>; idleSince: Map<number, number> }).idleSince.set(7, Date.now() - 1000);
(dialer as unknown as { runIdleCheck: () => void }).runIdleCheck();
expect(reasons).toEqual(['idle']);
});
it('emits reason="remote_closed" when bridge fires "closed" with code 1000', () => {
const dialer = MeshProxyTunnelDialer.resetForTest();
const reasons: Reason[] = [];
dialer.on('proxy-bridge-down', (_id: number, reason: Reason) => reasons.push(reason));
const fakeBridge = new EventEmitter() as unknown as EventEmitter & { close: ReturnType<typeof vi.fn> };
fakeBridge.close = vi.fn();
(dialer as unknown as { bridges: Map<number, unknown> }).bridges.set(7, fakeBridge);
// Attach the close listener via the dialer's private helper, mirroring
// what `dial()` does when a real bridge is registered. This keeps the
// test focused on the close-code classification + tearDownBridge path
// without spinning up a real WebSocket.
(dialer as unknown as { attachBridgeCloseListener: (id: number, b: EventEmitter) => void })
.attachBridgeCloseListener(7, fakeBridge as unknown as EventEmitter);
(fakeBridge as unknown as EventEmitter).emit('closed', { code: 1000, reason: 'normal' });
expect(reasons).toEqual(['remote_closed']);
});
});
describe('MeshProxyTunnelDialer reactive redial filter', () => {
beforeEach(() => MeshProxyTunnelDialer.resetForTest());
it('does NOT schedule redial after reason="idle"', () => {
const dialer = MeshProxyTunnelDialer.resetForTest();
const scheduleSpy = vi.spyOn(dialer as unknown as { scheduleReactiveRedial: (id: number) => void }, 'scheduleReactiveRedial').mockImplementation(() => {});
dialer.emit('proxy-bridge-down', 7, 'idle');
expect(scheduleSpy).not.toHaveBeenCalled();
});
it('does NOT schedule redial after reason="auth_failed"', () => {
const dialer = MeshProxyTunnelDialer.resetForTest();
const scheduleSpy = vi.spyOn(dialer as unknown as { scheduleReactiveRedial: (id: number) => void }, 'scheduleReactiveRedial').mockImplementation(() => {});
dialer.emit('proxy-bridge-down', 7, 'auth_failed');
expect(scheduleSpy).not.toHaveBeenCalled();
});
it('schedules redial after reason="remote_closed"', () => {
const dialer = MeshProxyTunnelDialer.resetForTest();
const scheduleSpy = vi.spyOn(dialer as unknown as { scheduleReactiveRedial: (id: number) => void }, 'scheduleReactiveRedial').mockImplementation(() => {});
dialer.emit('proxy-bridge-down', 7, 'remote_closed');
expect(scheduleSpy).toHaveBeenCalledWith(7);
});
});
@@ -0,0 +1,187 @@
/**
* `meshProxyTunnel.ts` first-frame state machine.
*
* The PEER side of a proxy-mode mesh tunnel optionally consumes a
* `mesh_handshake` JSON frame as the FIRST text frame from central,
* persists the bootstrap material via MeshCentralRegistry, then yields
* control to the existing TcpStreamSwitchboard. Out-of-order or
* malformed handshake frames close the WS with code 1008.
*/
import http from 'http';
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import WebSocket from 'ws';
import type { AddressInfo } from 'net';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
let tmpDir: string;
let handleMeshProxyTunnel: typeof import('../websocket/meshProxyTunnel').handleMeshProxyTunnel;
let MeshService: typeof import('../services/MeshService').MeshService;
let MeshCentralRegistry: typeof import('../services/MeshCentralRegistry').MeshCentralRegistry;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
interface ServerHandle {
server: http.Server;
port: number;
close: () => Promise<void>;
}
async function startServer(): Promise<ServerHandle> {
const server = http.createServer();
server.on('upgrade', (req, socket, head) => {
const pathname = new URL(req.url ?? '/', 'http://localhost').pathname;
if (pathname === '/api/mesh/proxy-tunnel') {
void handleMeshProxyTunnel(req, socket, head);
} else {
socket.destroy();
}
});
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', () => resolve()));
const port = (server.address() as AddressInfo).port;
return {
server,
port,
close: () => new Promise<void>((resolve) => server.close(() => resolve())),
};
}
function dialTunnel(port: number, query: string = ''): Promise<WebSocket> {
return new Promise((resolve, reject) => {
const ws = new WebSocket(`ws://127.0.0.1:${port}/api/mesh/proxy-tunnel${query}`);
ws.once('open', () => resolve(ws));
ws.once('error', reject);
});
}
function makeHandshakeFrame(overrides: Partial<{
centralInstanceId: string;
centralApiUrl: string;
meshTunnelJwt: string;
jwtExpiresAt: number;
peerNodeId: number;
}> = {}): string {
return JSON.stringify({
t: 'mesh_handshake',
v: 1,
peerNodeId: overrides.peerNodeId ?? 7,
centralInstanceId: overrides.centralInstanceId ?? 'central-instance-abc',
centralApiUrl: overrides.centralApiUrl ?? 'https://central.example.test',
meshTunnelJwt: overrides.meshTunnelJwt ?? 'jwt.token.body',
jwtExpiresAt: overrides.jwtExpiresAt ?? Math.floor(Date.now() / 1000) + 3600,
});
}
beforeAll(async () => {
tmpDir = await setupTestDb();
({ handleMeshProxyTunnel } = await import('../websocket/meshProxyTunnel'));
({ MeshService } = await import('../services/MeshService'));
({ MeshCentralRegistry } = await import('../services/MeshCentralRegistry'));
({ DatabaseService } = await import('../services/DatabaseService'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
beforeEach(() => {
delete process.env.SENCHO_MODE;
MeshService.getInstance().setReverseDialer(null);
MeshService.getInstance().setProxyTunnelSelfCentralNodeId(null);
MeshCentralRegistry.resetForTest();
DatabaseService.getInstance().getDb().prepare('DELETE FROM mesh_centrals').run();
});
describe('meshProxyTunnel first-frame state machine', () => {
it('consumes a mesh_handshake first frame and persists material via MeshCentralRegistry', async () => {
const srv = await startServer();
try {
const ws = await dialTunnel(srv.port, '?nodeId=7');
await new Promise((r) => setTimeout(r, 20));
const expiresAt = Math.floor(Date.now() / 1000) + 7200;
ws.send(makeHandshakeFrame({
centralInstanceId: 'central-instance-xyz',
centralApiUrl: 'https://central.example.test/',
meshTunnelJwt: 'callback.jwt.value',
jwtExpiresAt: expiresAt,
}));
await new Promise((r) => setTimeout(r, 30));
const row = MeshCentralRegistry.getInstance().getActive();
expect(row).not.toBeNull();
expect(row?.centralInstanceId).toBe('central-instance-xyz');
// Trailing slash should be stripped to keep the persisted URL canonical.
expect(row?.centralApiUrl).toBe('https://central.example.test');
expect(row?.callbackJwt).toBe('callback.jwt.value');
expect(row?.jwtExpiresAt).toBe(expiresAt);
ws.close(1000, 'test cleanup');
await new Promise((r) => setTimeout(r, 30));
} finally {
await srv.close();
}
});
it('passes through if first frame is not a mesh_handshake (delegates to switchboard)', async () => {
const srv = await startServer();
try {
const ws = await dialTunnel(srv.port, '?nodeId=7');
await new Promise((r) => setTimeout(r, 20));
// Send a known non-handshake JSON frame (tcp_close for a non-existent
// stream is a no-op the switchboard accepts without side effects).
ws.send(JSON.stringify({ t: 'tcp_close', s: 999999 }));
await new Promise((r) => setTimeout(r, 30));
// No handshake material should have been persisted.
expect(MeshCentralRegistry.getInstance().getActive()).toBeNull();
// WS should still be open (no protocol error close).
expect(ws.readyState).toBe(WebSocket.OPEN);
ws.close(1000, 'test cleanup');
await new Promise((r) => setTimeout(r, 30));
} finally {
await srv.close();
}
});
it('closes the WS as protocol error when mesh_handshake arrives after a non-handshake frame', async () => {
const srv = await startServer();
try {
const ws = await dialTunnel(srv.port, '?nodeId=7');
await new Promise((r) => setTimeout(r, 20));
// First a non-handshake frame, then the handshake.
ws.send(JSON.stringify({ t: 'tcp_close', s: 999999 }));
await new Promise((r) => setTimeout(r, 10));
const closeInfo = new Promise<{ code: number; reason: string }>((resolve) => {
ws.once('close', (code, reason) => resolve({ code, reason: reason.toString() }));
});
ws.send(makeHandshakeFrame());
const info = await closeInfo;
expect(info.code).toBe(1008);
expect(MeshCentralRegistry.getInstance().getActive()).toBeNull();
} finally {
await srv.close();
}
});
it('rejects malformed mesh_handshake (missing required fields) and closes WS as protocol error', async () => {
const srv = await startServer();
try {
const ws = await dialTunnel(srv.port, '?nodeId=7');
await new Promise((r) => setTimeout(r, 20));
const closeInfo = new Promise<{ code: number; reason: string }>((resolve) => {
ws.once('close', (code, reason) => resolve({ code, reason: reason.toString() }));
});
// Recognized 't' discriminator but missing required fields.
ws.send(JSON.stringify({ t: 'mesh_handshake', v: 1 }));
const info = await closeInfo;
expect(info.code).toBe(1008);
expect(MeshCentralRegistry.getInstance().getActive()).toBeNull();
} finally {
await srv.close();
}
});
});
@@ -0,0 +1,282 @@
/**
* `meshProxyTunnelFromPeer.ts`: central-side ingress for peer-initiated
* dial-back tunnels. Validates the chain of JWT claims and node-state
* preconditions that the peer's `mesh_tunnel` bootstrap token must satisfy
* before the upgrade succeeds and a proxy bridge is registered.
*
* The chain (in order): algorithm whitelist, signature, scope, audience
* (= SENCHO_PRIMARY_URL), issuer (= central instance_id), exp, iat clock
* skew (60s), node existence, mode==proxy, api_token fingerprint match.
* Every failure point returns HTTP 401 with a JSON `{reason}` body using
* a stable machine-readable code; the happy path constructs a
* `PilotTunnelBridge` and registers it via
* `PilotTunnelManager.replaceOrRegisterProxyBridge`.
*/
import http from 'http';
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import WebSocket from 'ws';
import jwt from 'jsonwebtoken';
import { createHash } from 'crypto';
import type { AddressInfo } from 'net';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
let tmpDir: string;
let handleMeshProxyTunnelFromPeerUpgrade: typeof import('../websocket/meshProxyTunnelFromPeer').handleMeshProxyTunnelFromPeerUpgrade;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
let PilotTunnelManager: typeof import('../services/PilotTunnelManager').PilotTunnelManager;
interface ServerHandle {
server: http.Server;
port: number;
close: () => Promise<void>;
}
async function startServer(): Promise<ServerHandle> {
const server = http.createServer();
server.on('upgrade', (req, socket, head) => {
const pathname = new URL(req.url ?? '/', 'http://localhost').pathname;
if (pathname === '/api/mesh/proxy-tunnel-from-peer') {
handleMeshProxyTunnelFromPeerUpgrade(req, socket, head);
} else {
socket.destroy();
}
});
await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', () => resolve()));
const port = (server.address() as AddressInfo).port;
return {
server,
port,
close: () => new Promise<void>((resolve) => server.close(() => resolve())),
};
}
interface UpgradeOutcome {
kind: 'open' | 'unexpected' | 'error';
status?: number;
body?: string;
ws?: WebSocket;
}
function attemptUpgrade(port: number, token: string): Promise<UpgradeOutcome> {
return new Promise((resolve) => {
const ws = new WebSocket(`ws://127.0.0.1:${port}/api/mesh/proxy-tunnel-from-peer`, {
headers: { authorization: `Bearer ${token}` },
});
const timer = setTimeout(() => resolve({ kind: 'error' }), 3000);
ws.once('open', () => {
clearTimeout(timer);
resolve({ kind: 'open', ws });
});
ws.once('unexpected-response', (_req, res) => {
const chunks: Buffer[] = [];
res.on('data', (c: Buffer) => chunks.push(c));
res.on('end', () => {
clearTimeout(timer);
resolve({ kind: 'unexpected', status: res.statusCode, body: Buffer.concat(chunks).toString('utf8') });
});
});
ws.once('error', () => {
// 'unexpected-response' fires first; 'error' here is the
// post-handshake failure ws raises after the server destroys
// the socket. Already resolved above.
});
});
}
function parseReason(body: string | undefined): string | null {
if (!body) return null;
try {
const parsed = JSON.parse(body) as { reason?: unknown };
return typeof parsed.reason === 'string' ? parsed.reason : null;
} catch {
return null;
}
}
const CANONICAL_ORIGIN = 'https://central.example.com';
const INSTANCE_ID = 'test-central-instance';
const PEER_API_TOKEN = 'peer-token-123';
let secret: string;
let peerNodeId: number;
let srv: ServerHandle;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ handleMeshProxyTunnelFromPeerUpgrade } = await import('../websocket/meshProxyTunnelFromPeer'));
({ DatabaseService } = await import('../services/DatabaseService'));
({ PilotTunnelManager } = await import('../services/PilotTunnelManager'));
});
beforeEach(async () => {
process.env.SENCHO_PRIMARY_URL = CANONICAL_ORIGIN;
const db = DatabaseService.getInstance();
secret = db.getGlobalSettings().auth_jwt_secret;
db.setSystemState('instance_id', INSTANCE_ID);
// Unique peer per test to avoid PilotTunnelManager collisions across
// the 12 cases (one happy path actually registers a bridge).
peerNodeId = db.addNode({
name: `peer-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
type: 'remote',
mode: 'proxy',
api_url: 'https://peer.example.com',
api_token: PEER_API_TOKEN,
compose_dir: '/tmp',
is_default: false,
});
db.setNodeMeshEnabled(peerNodeId, true);
if (!srv) srv = await startServer();
});
afterAll(async () => {
if (srv) await srv.close();
delete process.env.SENCHO_PRIMARY_URL;
cleanupTestDb(tmpDir);
});
function expectedFp(token: string = PEER_API_TOKEN): string {
return createHash('sha256').update(token).digest('hex').slice(0, 16);
}
function makeJwt(overrides: Partial<Record<string, unknown>> = {}, signOpts: { alg?: jwt.Algorithm; secret?: string } = {}): string {
const payload: Record<string, unknown> = {
sub: String(peerNodeId),
iss: INSTANCE_ID,
aud: CANONICAL_ORIGIN,
scope: 'mesh_tunnel',
iat: Math.floor(Date.now() / 1000),
exp: Math.floor(Date.now() / 1000) + 3600,
kid: 'v1',
peer_token_fp: expectedFp(),
...overrides,
};
return jwt.sign(payload, signOpts.secret ?? secret, { algorithm: signOpts.alg ?? 'HS256' });
}
describe('/api/mesh/proxy-tunnel-from-peer validation chain', () => {
it('rejects alg=none (algorithm_mismatch)', async () => {
// jsonwebtoken refuses to sign with alg=none unless explicitly
// enabled and given a null secret; we build the token manually so
// the test exercises the central's defence, not the library's.
const header = Buffer.from(JSON.stringify({ alg: 'none', typ: 'JWT' })).toString('base64url');
const body = Buffer.from(JSON.stringify({
sub: String(peerNodeId), iss: INSTANCE_ID, aud: CANONICAL_ORIGIN,
scope: 'mesh_tunnel', iat: Math.floor(Date.now() / 1000),
exp: Math.floor(Date.now() / 1000) + 3600, peer_token_fp: expectedFp(),
})).toString('base64url');
const token = `${header}.${body}.`;
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('algorithm_mismatch');
});
it('rejects alg=RS256 (algorithm_mismatch)', async () => {
const { generateKeyPairSync } = await import('crypto');
const { privateKey } = generateKeyPairSync('rsa', { modulusLength: 2048 });
const token = jwt.sign({
sub: String(peerNodeId), iss: INSTANCE_ID, aud: CANONICAL_ORIGIN,
scope: 'mesh_tunnel', exp: Math.floor(Date.now() / 1000) + 3600,
peer_token_fp: expectedFp(),
}, privateKey, { algorithm: 'RS256' });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('algorithm_mismatch');
});
it('rejects bad signature (signature_invalid)', async () => {
const token = makeJwt({}, { secret: 'wrong-secret-not-the-real-one' });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('signature_invalid');
});
it('rejects scope mismatch', async () => {
const token = makeJwt({ scope: 'pilot_tunnel' });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('scope_mismatch');
});
it('rejects audience mismatch', async () => {
const token = makeJwt({ aud: 'https://other.example.com' });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('audience_mismatch');
});
it('rejects issuer mismatch', async () => {
const token = makeJwt({ iss: 'wrong-instance-id' });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('instance_mismatch');
});
it('rejects expired token (stale)', async () => {
const token = makeJwt({ exp: Math.floor(Date.now() / 1000) - 10 });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
// jsonwebtoken throws on expired tokens before our exp check sees it,
// so the rejection surfaces as signature_invalid via the verify catch.
// The contract: stale tokens are rejected with a 401 and some
// deterministic reason code; accept either of the two equivalent
// failures since both convey "stale credential" to the operator.
const reason = parseReason(outcome.body);
expect(['stale', 'signature_invalid']).toContain(reason);
});
it('rejects clock-skewed token (clock_skew)', async () => {
const token = makeJwt({ iat: Math.floor(Date.now() / 1000) + 600 });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('clock_skew');
});
it('rejects missing node (node_deleted)', async () => {
const token = makeJwt({ sub: '999999' });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('node_deleted');
});
it('rejects mode mismatch', async () => {
DatabaseService.getInstance().updateNode(peerNodeId, { mode: 'pilot_agent' });
const token = makeJwt();
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('mode_mismatch');
});
it('rejects token fingerprint mismatch', async () => {
const token = makeJwt();
// Rotate the api_token after minting; the JWT now carries the
// fingerprint of the old token, but getNode returns the new one.
DatabaseService.getInstance().updateNode(peerNodeId, { api_token: 'rotated-token-xyz' });
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('unexpected');
expect(outcome.status).toBe(401);
expect(parseReason(outcome.body)).toBe('token_fingerprint_mismatch');
});
it('accepts a fully valid token and registers a proxy bridge', async () => {
const token = makeJwt();
const outcome = await attemptUpgrade(srv.port, token);
expect(outcome.kind).toBe('open');
// Give the bridge.start() microtasks a moment to land and register.
await new Promise((r) => setTimeout(r, 50));
const bridge = PilotTunnelManager.getInstance().getBridge(peerNodeId);
expect(bridge).not.toBeNull();
try { outcome.ws?.close(1000, 'test cleanup'); } catch { /* ignore */ }
// Allow the manager's 'closed' handler to remove the bridge entry.
await new Promise((r) => setTimeout(r, 30));
});
});
@@ -0,0 +1,148 @@
/**
* `MeshService.proactiveBootstrapFanout` (Trigger 3).
*
* At central startup, after `MeshService.start()` finishes, iterate the
* mesh-enabled proxy-mode nodes that have at least one `mesh_stacks` row
* and call `MeshProxyTunnelDialer.ensureBridge(nodeId)` on each. This
* proactively re-establishes the central->peer bridges so the
* capability-gated handshake can mint and ship bootstrap material to any
* peer that just came online or just got upgraded to v0.79+.
*
* Concurrency 4, 250ms stagger, fire-and-forget. Failures do not abort
* the fan-out.
*/
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { MeshService } from '../services/MeshService';
import { MeshProxyTunnelDialer } from '../services/MeshProxyTunnelDialer';
import { DatabaseService } from '../services/DatabaseService';
let tmpDir: string;
beforeAll(async () => {
tmpDir = await setupTestDb();
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
function uniqueSuffix(): string {
return `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
}
function seedProxyNode(meshEnabled: boolean): number {
const db = DatabaseService.getInstance();
const id = db.addNode({
name: `peer-fanout-${uniqueSuffix()}`,
type: 'remote',
mode: 'proxy',
api_url: `https://peer-${uniqueSuffix()}.example.com`,
api_token: `tok-${uniqueSuffix()}`,
compose_dir: '/tmp',
is_default: false,
});
if (meshEnabled) {
db.setNodeMeshEnabled(id, true);
}
return id;
}
function insertMeshStack(nodeId: number, stackName: string): void {
const db = DatabaseService.getInstance();
db.getDb().prepare(
'INSERT INTO mesh_stacks (node_id, stack_name, created_at, created_by) VALUES (?, ?, ?, ?)',
).run(nodeId, stackName, Date.now(), 'test');
}
function clearNodesAndMeshStacks(): void {
const db = DatabaseService.getInstance().getDb();
db.prepare('DELETE FROM mesh_stacks').run();
db.prepare('DELETE FROM nodes WHERE is_default = 0').run();
}
describe('MeshService.proactiveBootstrapFanout (Trigger 3)', () => {
beforeEach(() => {
clearNodesAndMeshStacks();
MeshProxyTunnelDialer.resetForTest();
});
afterEach(() => {
vi.restoreAllMocks();
});
it('iterates only mesh-enabled proxy-mode nodes that have mesh_stacks rows', async () => {
const calls: number[] = [];
vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockImplementation(async (id: number) => {
calls.push(id);
return null;
});
// Eligible: mesh-enabled proxy node with a mesh_stacks row.
const eligible = seedProxyNode(true);
insertMeshStack(eligible, `stack-eligible-${uniqueSuffix()}`);
// Ineligible: mesh-enabled proxy node WITHOUT a mesh_stacks row.
seedProxyNode(true);
// Ineligible: mesh-disabled proxy node WITH a mesh_stacks row.
const meshOff = seedProxyNode(false);
insertMeshStack(meshOff, `stack-meshoff-${uniqueSuffix()}`);
await (MeshService.getInstance() as unknown as {
proactiveBootstrapFanout: () => Promise<void>;
}).proactiveBootstrapFanout();
expect(calls).toEqual([eligible]);
});
it('throttles to concurrency 4', async () => {
let inflight = 0;
let maxInflight = 0;
vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockImplementation(async () => {
inflight++;
if (inflight > maxInflight) maxInflight = inflight;
await new Promise((r) => setTimeout(r, 10));
inflight--;
return null;
});
for (let i = 0; i < 12; i++) {
const id = seedProxyNode(true);
insertMeshStack(id, `stack-${i}-${uniqueSuffix()}`);
}
await (MeshService.getInstance() as unknown as {
proactiveBootstrapFanout: () => Promise<void>;
}).proactiveBootstrapFanout();
expect(maxInflight).toBeLessThanOrEqual(4);
expect(maxInflight).toBeGreaterThan(0);
});
it('failures do not abort the fanout', async () => {
const ids: number[] = [];
for (let i = 0; i < 3; i++) {
const id = seedProxyNode(true);
insertMeshStack(id, `stack-fail-${i}-${uniqueSuffix()}`);
ids.push(id);
}
const failingId = ids[1];
const calls: number[] = [];
vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockImplementation(async (id: number) => {
calls.push(id);
if (id === failingId) throw new Error('boom');
return null;
});
await (MeshService.getInstance() as unknown as {
proactiveBootstrapFanout: () => Promise<void>;
}).proactiveBootstrapFanout();
expect(calls.slice().sort((a, b) => a - b)).toEqual(ids.slice().sort((a, b) => a - b));
});
});
+9 -1
View File
@@ -1,4 +1,4 @@
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { EventEmitter } from 'events';
import fsSync from 'fs';
import path from 'path';
@@ -849,6 +849,14 @@ describe('MeshService.openCrossNode (BUG-4)', () => {
};
}
// Install a stub reverseDialer so openCrossNode skips the peer-side
// bootstrap path (PeerToCentralMeshSessionDialer.ensureSession). The
// dispatch behavior under test relies on dialMeshTcpStream being called
// directly; the bootstrap kick would short-circuit before that mock fires.
const stubDialer = { openMeshTcpStream: vi.fn() };
beforeEach(() => { MeshService.getInstance().setReverseDialer(stubDialer); });
afterEach(() => { MeshService.getInstance().setReverseDialer(null); });
it('emits route.dispatch immediately on cross-node entry', async () => {
const svc = MeshService.getInstance();
const target: MeshTarget = {
@@ -0,0 +1,73 @@
import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach, vi } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { MeshService } from '../services/MeshService';
import { DatabaseService } from '../services/DatabaseService';
function seedMeshedProxyNode(): void {
const db = DatabaseService.getInstance();
const id = db.addNode({
name: `p-${Date.now()}-${Math.random()}`,
type: 'remote',
mode: 'proxy',
api_url: 'https://p.example.com',
api_token: 't',
compose_dir: '/tmp',
is_default: false,
});
db.setNodeMeshEnabled(id, true);
}
let tmpDir: string;
beforeAll(async () => { tmpDir = await setupTestDb(); });
afterAll(() => cleanupTestDb(tmpDir));
describe('SENCHO_PRIMARY_URL preflight warning', () => {
let originalPrimaryUrl: string | undefined;
beforeEach(() => {
originalPrimaryUrl = process.env.SENCHO_PRIMARY_URL;
// The DB singleton persists across tests in the same process; wipe any
// non-default nodes seeded by a prior test so each case starts clean.
// setupTestDb itself runs once per file (beforeAll above) to avoid
// invalidating the DatabaseService singleton's open connection.
DatabaseService.getInstance().getDb().prepare('DELETE FROM nodes WHERE is_default = 0').run();
});
afterEach(() => {
if (originalPrimaryUrl === undefined) {
delete process.env.SENCHO_PRIMARY_URL;
} else {
process.env.SENCHO_PRIMARY_URL = originalPrimaryUrl;
}
});
it('warns when SENCHO_PRIMARY_URL is unset and a mesh-enabled proxy node exists', () => {
delete process.env.SENCHO_PRIMARY_URL;
seedMeshedProxyNode();
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
const svc = MeshService.getInstance() as unknown as { maybeWarnUnsetPrimaryUrl: () => void };
svc.maybeWarnUnsetPrimaryUrl();
expect(warn.mock.calls.some(c => /SENCHO_PRIMARY_URL is unset/.test(String(c[0])))).toBe(true);
warn.mockRestore();
});
it('does not warn when SENCHO_PRIMARY_URL is set', () => {
process.env.SENCHO_PRIMARY_URL = 'https://central.example.com';
seedMeshedProxyNode();
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
const svc = MeshService.getInstance() as unknown as { maybeWarnUnsetPrimaryUrl: () => void };
svc.maybeWarnUnsetPrimaryUrl();
expect(warn.mock.calls.some(c => /SENCHO_PRIMARY_URL is unset/.test(String(c[0])))).toBe(false);
warn.mockRestore();
});
it('does not warn when SENCHO_PRIMARY_URL is unset but no mesh-enabled proxy node exists', () => {
delete process.env.SENCHO_PRIMARY_URL;
// No seeding: zero mesh-enabled proxy nodes (only the default local node).
const warn = vi.spyOn(console, 'warn').mockImplementation(() => {});
const svc = MeshService.getInstance() as unknown as { maybeWarnUnsetPrimaryUrl: () => void };
svc.maybeWarnUnsetPrimaryUrl();
expect(warn.mock.calls.some(c => /SENCHO_PRIMARY_URL is unset/.test(String(c[0])))).toBe(false);
warn.mockRestore();
});
});
@@ -0,0 +1,177 @@
/**
* Triggers 1 + 2: proactive mesh bridge bootstrap on mesh-enable and on
* api_token rotation.
*
* Trigger 1 (mesh-enable): after `setNodeMeshEnabled(true)`, the service
* fire-and-forgets `MeshProxyTunnelDialer.ensureBridge(nodeId)` for any
* remote/proxy peer. The capability-gated handshake then ships
* `mesh_handshake` material to peers that just came online or just got
* upgraded to a build that advertises `mesh_proxy_callback_bootstrap`.
*
* Trigger 2 (api_token rotation): when the nodes router persists a new
* `api_token` for a mesh-enabled proxy peer, it closes the existing bridge
* with reason 'peer token rotated' and re-dials. The next ensureBridge mints
* a JWT whose fingerprint matches the freshly stored token; without this
* trigger the remote would reject the upgrade with token_fingerprint_mismatch
* until the next idle re-dial.
*/
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import request from 'supertest';
import jwt from 'jsonwebtoken';
import { setupTestDb, cleanupTestDb, TEST_USERNAME, TEST_JWT_SECRET } from './helpers/setupTestDb';
import { MeshService } from '../services/MeshService';
import { MeshProxyTunnelDialer } from '../services/MeshProxyTunnelDialer';
import { DatabaseService } from '../services/DatabaseService';
let tmpDir: string;
let app: import('express').Express;
let authHeader: string;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ app } = await import('../index'));
const token = jwt.sign({ username: TEST_USERNAME }, TEST_JWT_SECRET, { expiresIn: '1m' });
authHeader = `Bearer ${token}`;
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
function uniqueSuffix(): string {
return `${Date.now()}-${Math.random().toString(36).slice(2, 10)}`;
}
function seedProxyNode(meshEnabled: boolean): number {
const db = DatabaseService.getInstance();
const id = db.addNode({
name: `peer-rotate-${uniqueSuffix()}`,
type: 'remote',
mode: 'proxy',
api_url: `https://peer-${uniqueSuffix()}.example.com`,
api_token: `tok-${uniqueSuffix()}`,
compose_dir: '/tmp',
is_default: false,
});
if (meshEnabled) {
db.setNodeMeshEnabled(id, true);
}
return id;
}
function clearTestNodes(): void {
const db = DatabaseService.getInstance().getDb();
db.prepare('DELETE FROM mesh_stacks').run();
db.prepare('DELETE FROM nodes WHERE is_default = 0').run();
}
describe('Trigger 1: enableForNode triggers proactive bridge bootstrap', () => {
beforeEach(() => {
clearTestNodes();
MeshProxyTunnelDialer.resetForTest();
});
afterEach(() => {
vi.restoreAllMocks();
});
it('calls ensureBridge for a remote proxy-mode node', async () => {
const ensureSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockResolvedValue(null);
const nodeId = seedProxyNode(false);
await MeshService.getInstance().enableForNode(nodeId);
// Trigger is fire-and-forget; wait a microtask for the void promise.
await new Promise((r) => setImmediate(r));
expect(ensureSpy).toHaveBeenCalledWith(nodeId);
});
it('skips local nodes', async () => {
const ensureSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockResolvedValue(null);
const localId = DatabaseService.getInstance().getNodes()[0].id;
await MeshService.getInstance().enableForNode(localId);
await new Promise((r) => setImmediate(r));
expect(ensureSpy).not.toHaveBeenCalled();
});
it('swallows ensureBridge rejections without throwing', async () => {
vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockRejectedValue(new Error('peer unreachable'));
const nodeId = seedProxyNode(false);
await expect(MeshService.getInstance().enableForNode(nodeId)).resolves.toBeUndefined();
await new Promise((r) => setImmediate(r));
});
});
describe('Trigger 2: api_token rotation forces re-bootstrap', () => {
beforeEach(() => {
clearTestNodes();
MeshProxyTunnelDialer.resetForTest();
});
afterEach(() => {
vi.restoreAllMocks();
});
it('calls closeBridge then ensureBridge when rotation handler runs', async () => {
const closeSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'closeBridge');
const ensureSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockResolvedValue(null);
const nodeId = seedProxyNode(true);
const res = await request(app)
.put(`/api/nodes/${nodeId}`)
.set('Authorization', authHeader)
.send({ api_token: 'rotated-token-abc' });
expect(res.status).toBe(200);
await new Promise((r) => setImmediate(r));
expect(closeSpy).toHaveBeenCalledWith(nodeId, 'peer token rotated');
expect(ensureSpy).toHaveBeenCalledWith(nodeId);
const closeOrder = closeSpy.mock.invocationCallOrder[0];
const ensureOrder = ensureSpy.mock.invocationCallOrder[0];
expect(closeOrder).toBeLessThan(ensureOrder);
});
it('does not fire when api_token is not in the update payload', async () => {
const closeSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'closeBridge');
const ensureSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockResolvedValue(null);
const nodeId = seedProxyNode(true);
const res = await request(app)
.put(`/api/nodes/${nodeId}`)
.set('Authorization', authHeader)
.send({ name: `renamed-${uniqueSuffix()}` });
expect(res.status).toBe(200);
await new Promise((r) => setImmediate(r));
expect(closeSpy).not.toHaveBeenCalled();
expect(ensureSpy).not.toHaveBeenCalled();
});
it('does not fire when mesh is disabled on the node', async () => {
const closeSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'closeBridge');
const ensureSpy = vi.spyOn(MeshProxyTunnelDialer.getInstance(), 'ensureBridge')
.mockResolvedValue(null);
const nodeId = seedProxyNode(false);
const res = await request(app)
.put(`/api/nodes/${nodeId}`)
.set('Authorization', authHeader)
.send({ api_token: 'rotated-token-xyz' });
expect(res.status).toBe(200);
await new Promise((r) => setImmediate(r));
expect(closeSpy).not.toHaveBeenCalled();
expect(ensureSpy).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,112 @@
/**
* Version skew invariant: central must NOT send a `mesh_handshake` to a peer
* that lacks the `mesh_proxy_callback_bootstrap` capability.
*
* The capability gate keeps central rolling forward without breaking older
* peers: a peer that does not understand the bootstrap frame would otherwise
* see a stray JSON text frame before the TCP frames it expects. The dialer's
* `maybeSendBootstrap` must read the capability list from the remote's meta
* endpoint and silently skip the send when the bit is absent.
*
* This is complementary to `mesh-proxy-tunnel-dialer-handshake.test.ts`,
* which exercises the same code path with a finer-grained set of inputs;
* this file locks in the same invariant as an integration so a future
* refactor that moves the capability check to a different layer does not
* silently regress.
*/
import { describe, it, expect, beforeAll, afterAll, beforeEach, afterEach, vi } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { NodeRegistry } from '../services/NodeRegistry';
import { MeshProxyTunnelDialer } from '../services/MeshProxyTunnelDialer';
import { OFFLINE_META } from '../services/CapabilityRegistry';
import { MeshCentralRegistry } from '../services/MeshCentralRegistry';
import { DatabaseService } from '../services/DatabaseService';
let tmpDir: string;
beforeAll(async () => { tmpDir = await setupTestDb(); });
afterAll(() => cleanupTestDb(tmpDir));
describe('Version skew: peer without mesh_proxy_callback_bootstrap', () => {
beforeEach(() => {
// Per-test reset of dialer + registry singletons + env. The DB lives
// once per file (beforeAll above); resetting setupTestDb per test
// would invalidate the DatabaseService singleton on Linux.
MeshProxyTunnelDialer.resetForTest();
MeshCentralRegistry.resetForTest();
process.env.SENCHO_PRIMARY_URL = 'https://central.example.com';
DatabaseService.getInstance().setSystemState('instance_id', 'test-central-instance');
});
afterEach(() => {
delete process.env.SENCHO_PRIMARY_URL;
vi.restoreAllMocks();
});
type DialerPrivate = {
maybeSendBootstrap: (nodeId: number, ws: unknown) => Promise<void>;
};
it('central does not send mesh_handshake when capability is absent', async () => {
vi.spyOn(NodeRegistry.getInstance(), 'fetchMetaForNode').mockResolvedValue({
...OFFLINE_META,
version: '0.78.0',
capabilities: ['stacks'],
online: true,
startedAt: Date.now(),
});
// Seed a proxy node so maybeSendBootstrap has something to look up.
const db = DatabaseService.getInstance();
const id = db.addNode({
name: `n-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
type: 'remote',
mode: 'proxy',
api_url: 'https://n',
api_token: 't',
compose_dir: '/tmp',
is_default: false,
});
db.setNodeMeshEnabled(id, true);
const sentFrames: unknown[] = [];
const fakeWs = { send: vi.fn((data: string) => sentFrames.push(JSON.parse(data))) };
await (MeshProxyTunnelDialer.getInstance() as unknown as DialerPrivate)
.maybeSendBootstrap(id, fakeWs);
expect(fakeWs.send).not.toHaveBeenCalled();
expect(sentFrames).toHaveLength(0);
// No persistence as a side effect either.
expect(MeshCentralRegistry.getInstance().getActive()).toBeNull();
});
it('central does send mesh_handshake when the capability flips to advertised', async () => {
// Sanity: the same code path with the capability present must take
// the affirmative branch, so we know the skip above is gated on the
// capability and not on some unrelated precondition.
vi.spyOn(NodeRegistry.getInstance(), 'fetchMetaForNode').mockResolvedValue({
...OFFLINE_META,
version: '0.79.0',
capabilities: ['stacks', 'mesh_proxy_callback_bootstrap'],
online: true,
startedAt: Date.now(),
});
const db = DatabaseService.getInstance();
const id = db.addNode({
name: `n-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`,
type: 'remote',
mode: 'proxy',
api_url: 'https://n',
api_token: 't',
compose_dir: '/tmp',
is_default: false,
});
db.setNodeMeshEnabled(id, true);
const sentFrames: unknown[] = [];
const fakeWs = { send: vi.fn((data: string) => sentFrames.push(JSON.parse(data))) };
await (MeshProxyTunnelDialer.getInstance() as unknown as DialerPrivate)
.maybeSendBootstrap(id, fakeWs);
expect(fakeWs.send).toHaveBeenCalledOnce();
expect(sentFrames[0]).toMatchObject({ t: 'mesh_handshake', v: 1 });
});
});
@@ -0,0 +1,250 @@
/**
* `PeerToCentralMeshSessionDialer`: peer-side counterpart to the central
* ingress at `/api/mesh/proxy-tunnel-from-peer` (Task 9). Reads cached
* central material from `MeshCentralRegistry`, opens a WebSocket to
* central carrying the bootstrapped JWT in the Authorization header, and
* branches on the upgrade outcome:
*
* - successful open: marks the cache row "used", arms the bridge.
* - 401 with a terminal reason code (stale, signature_invalid, ...):
* clears the cache so we stop dialing on a bad credential.
* - 401 with a transient reason (clock_skew, mode_mismatch, ...):
* keeps the cache, marks rejected so the operator can see why.
* - 404 endpoint_not_found: keeps the cache and arms a longer backoff
* so we do not hammer an older central that has not yet shipped the
* peer ingress endpoint.
*
* The dial logic is also rate-limited per dialer process (5 attempts /
* 60s window) so a misbehaving central or a misconfigured peer cannot
* flood either side with handshake traffic.
*/
import http from 'http';
import type { AddressInfo } from 'net';
import { WebSocketServer } from 'ws';
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
let tmpDir: string;
let PeerToCentralMeshSessionDialer: typeof import('../services/PeerToCentralMeshSessionDialer').PeerToCentralMeshSessionDialer;
let MeshCentralRegistry: typeof import('../services/MeshCentralRegistry').MeshCentralRegistry;
let DatabaseService: typeof import('../services/DatabaseService').DatabaseService;
let PilotTunnelBridge: typeof import('../services/PilotTunnelBridge').PilotTunnelBridge;
interface RejectingServer {
server: http.Server;
url: string;
lastAuthHeader: string | null;
lastPath: string | null;
close: () => Promise<void>;
}
/**
* Spins up a minimal HTTP server that intercepts WebSocket upgrades and
* responds with a fixed status + JSON body, mimicking how the central
* ingress rejects bad credentials. Records the auth header and path so
* tests can assert the dialer is sending the right material.
*/
async function startRejectingServer(status: number, reason: string): Promise<RejectingServer> {
const handle: RejectingServer = {
server: http.createServer(),
url: '',
lastAuthHeader: null,
lastPath: null,
close: () => new Promise<void>((resolve) => handle.server.close(() => resolve())),
};
handle.server.on('upgrade', (req, socket) => {
handle.lastAuthHeader = (req.headers['authorization'] as string | undefined) ?? null;
handle.lastPath = req.url ?? null;
const body = JSON.stringify({ reason });
const statusText = status === 404 ? 'Not Found' : 'Unauthorized';
const head = [
`HTTP/1.1 ${status} ${statusText}`,
'Content-Type: application/json',
`Content-Length: ${Buffer.byteLength(body)}`,
'Connection: close',
'',
body,
].join('\r\n');
try { socket.write(head); } catch { /* ignore */ }
try { socket.destroy(); } catch { /* ignore */ }
});
await new Promise<void>((resolve) => handle.server.listen(0, '127.0.0.1', () => resolve()));
const port = (handle.server.address() as AddressInfo).port;
handle.url = `http://127.0.0.1:${port}`;
return handle;
}
beforeAll(async () => {
tmpDir = await setupTestDb();
({ PeerToCentralMeshSessionDialer } = await import('../services/PeerToCentralMeshSessionDialer'));
({ MeshCentralRegistry } = await import('../services/MeshCentralRegistry'));
({ DatabaseService } = await import('../services/DatabaseService'));
({ PilotTunnelBridge } = await import('../services/PilotTunnelBridge'));
});
afterAll(() => {
cleanupTestDb(tmpDir);
});
beforeEach(() => {
MeshCentralRegistry.resetForTest();
PeerToCentralMeshSessionDialer.resetForTest();
DatabaseService.getInstance().getDb().prepare('DELETE FROM mesh_centrals').run();
});
afterEach(() => {
PeerToCentralMeshSessionDialer.resetForTest();
MeshCentralRegistry.resetForTest();
});
describe('PeerToCentralMeshSessionDialer', () => {
it('returns null when no central material is cached', async () => {
const result = await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
expect(result).toBeNull();
});
it('dials using the cached URL + JWT bearer (asserted by server-side recording)', async () => {
const srv = await startRejectingServer(401, 'clock_skew');
try {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-a',
centralApiUrl: srv.url,
callbackJwt: 'fake.jwt.token',
jwtIssuedAt: 1,
jwtExpiresAt: Math.floor(Date.now() / 1000) + 3600,
});
const result = await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
expect(result).toBeNull();
expect(srv.lastPath).toBe('/api/mesh/proxy-tunnel-from-peer');
expect(srv.lastAuthHeader).toBe('Bearer fake.jwt.token');
} finally {
await srv.close();
}
});
it('clears cache on 401 reason=stale', async () => {
const srv = await startRejectingServer(401, 'stale');
try {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-b',
centralApiUrl: srv.url,
callbackJwt: 'jwt',
jwtIssuedAt: 1,
jwtExpiresAt: Math.floor(Date.now() / 1000) + 3600,
});
await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
expect(MeshCentralRegistry.getInstance().getActive()).toBeNull();
} finally {
await srv.close();
}
});
it('keeps cache on 401 reason=clock_skew but records the rejection', async () => {
const srv = await startRejectingServer(401, 'clock_skew');
try {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-c',
centralApiUrl: srv.url,
callbackJwt: 'jwt',
jwtIssuedAt: 1,
jwtExpiresAt: Math.floor(Date.now() / 1000) + 3600,
});
await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
const row = MeshCentralRegistry.getInstance().getActive();
expect(row).not.toBeNull();
expect(row?.lastRejectReason).toBe('clock_skew');
} finally {
await srv.close();
}
});
it('keeps cache on HTTP 404 endpoint_not_found and rate-limits future dials', async () => {
const srv = await startRejectingServer(404, 'whatever');
try {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-d',
centralApiUrl: srv.url,
callbackJwt: 'jwt',
jwtIssuedAt: 1,
jwtExpiresAt: Math.floor(Date.now() / 1000) + 3600,
});
const dialer = PeerToCentralMeshSessionDialer.getInstance();
await dialer.ensureSession();
// Cache must survive a 404 (older central, transient infra).
expect(MeshCentralRegistry.getInstance().getActive()).not.toBeNull();
// The second call must not even hit the server: the endpoint
// backoff blocks new dials for the configured window.
srv.lastAuthHeader = null;
const second = await dialer.ensureSession();
expect(second).toBeNull();
expect(srv.lastAuthHeader).toBeNull();
} finally {
await srv.close();
}
});
it('rate-limits dials after the configured attempt cap', async () => {
const srv = await startRejectingServer(401, 'clock_skew');
try {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-e',
centralApiUrl: srv.url,
callbackJwt: 'jwt',
jwtIssuedAt: 1,
jwtExpiresAt: Math.floor(Date.now() / 1000) + 3600,
});
const dialer = PeerToCentralMeshSessionDialer.getInstance();
// 5 dials should land; the 6th must be suppressed by the limiter.
for (let i = 0; i < 5; i++) {
await dialer.ensureSession();
}
srv.lastAuthHeader = null;
const sixth = await dialer.ensureSession();
expect(sixth).toBeNull();
expect(srv.lastAuthHeader).toBeNull();
} finally {
await srv.close();
}
});
it('marks the row used on successful WS open', async () => {
const wss = new WebSocketServer({ noServer: true });
const srv = http.createServer();
srv.on('upgrade', (req, socket, head) => {
if (req.url?.startsWith('/api/mesh/proxy-tunnel-from-peer')) {
wss.handleUpgrade(req, socket, head, () => { /* accept */ });
} else {
socket.destroy();
}
});
await new Promise<void>((resolve) => srv.listen(0, '127.0.0.1', () => resolve()));
const port = (srv.address() as AddressInfo).port;
const url = `http://127.0.0.1:${port}`;
let bridge: InstanceType<typeof PilotTunnelBridge> | null = null;
try {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-success',
centralApiUrl: url,
callbackJwt: 'fake.jwt',
jwtIssuedAt: 1,
jwtExpiresAt: 9999999999,
});
bridge = await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
expect(bridge).toBeInstanceOf(PilotTunnelBridge);
await vi.waitFor(() => {
expect(MeshCentralRegistry.getInstance().getActive()?.lastUsedAt ?? 0).toBeGreaterThan(0);
});
expect(PeerToCentralMeshSessionDialer.getInstance().hasSession()).toBe(true);
} finally {
try { bridge?.close(1000, 'test done'); } catch { /* ignore */ }
await new Promise<void>((resolve) => wss.close(() => resolve()));
await new Promise<void>((resolve) => srv.close(() => resolve()));
}
});
it('singleton returns the same instance', () => {
const a = PeerToCentralMeshSessionDialer.getInstance();
const b = PeerToCentralMeshSessionDialer.getInstance();
expect(a).toBe(b);
});
});
@@ -0,0 +1,74 @@
/**
* Pilot-mode regression invariant: a pilot-agent tunnel always wins over a
* peer-initiated proxy-mode bridge for the same nodeId.
*
* The two registration paths share the `PilotTunnelManager.bridges` map.
* Without the `bridgeKinds` index, a peer-initiated dial-back could quietly
* replace a live pilot tunnel and break the agent's reverse-stream relay.
* `replaceOrRegisterProxyBridge` is the single point that has to refuse the
* replacement; this test locks that contract in.
*
* Mirrors the `injectBridgeForTest` pattern used elsewhere in the suite so
* we exercise the manager invariant without owning a real WebSocket.
*/
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { EventEmitter } from 'events';
import { PilotTunnelManager } from '../services/PilotTunnelManager';
import type { PilotTunnelBridge } from '../services/PilotTunnelBridge';
function makeFakeBridge(): EventEmitter & {
close: ReturnType<typeof vi.fn>;
getActiveStreamCount: () => number;
} {
const ee = new EventEmitter() as EventEmitter & {
close: ReturnType<typeof vi.fn>;
getActiveStreamCount: () => number;
};
ee.close = vi.fn();
ee.getActiveStreamCount = () => 0;
return ee;
}
describe('Pilot-mode regression: pilot tunnel wins over peer-initiated proxy bridge', () => {
beforeEach(() => {
PilotTunnelManager.resetForTest();
});
it('refuses replaceOrRegisterProxyBridge when a pilot tunnel exists for the same nodeId', () => {
const mgr = PilotTunnelManager.getInstance();
const pilot = makeFakeBridge();
const proxy = makeFakeBridge();
mgr.injectBridgeForTest(7, pilot as unknown as PilotTunnelBridge, 'pilot');
expect(() => mgr.replaceOrRegisterProxyBridge(7, proxy as unknown as PilotTunnelBridge))
.toThrow(/pilot tunnel/);
// Pilot bridge is still the resident bridge for nodeId 7.
expect(mgr.getBridge(7)).toBe(pilot);
// The pilot bridge must not be closed by the rejected replacement.
expect(pilot.close).not.toHaveBeenCalled();
});
it('allows replaceOrRegisterProxyBridge to swap one proxy bridge for another', () => {
const mgr = PilotTunnelManager.getInstance();
const oldProxy = makeFakeBridge();
const newProxy = makeFakeBridge();
mgr.injectBridgeForTest(9, oldProxy as unknown as PilotTunnelBridge, 'proxy');
mgr.replaceOrRegisterProxyBridge(9, newProxy as unknown as PilotTunnelBridge);
// The new dial is the source of truth; old proxy is closed.
expect(oldProxy.close).toHaveBeenCalledOnce();
expect(mgr.getBridge(9)).toBe(newProxy);
});
it('replaceOrRegisterProxyBridge on an empty slot just registers (no refuse, no close)', () => {
const mgr = PilotTunnelManager.getInstance();
const proxy = makeFakeBridge();
mgr.replaceOrRegisterProxyBridge(11, proxy as unknown as PilotTunnelBridge);
expect(mgr.getBridge(11)).toBe(proxy);
expect(proxy.close).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,290 @@
/**
* R1-B: PilotTunnelBridge.acceptReverseLocal emits MeshService activity events
* for the reverse-direction dispatch (peer-to-central). Reuses the existing
* `route.resolve.ok` / `route.resolve.fail` event types and discriminates by
* `details.direction = 'reverse'` so the Routing tab can surface peer-side
* mesh failures alongside central-side dispatch events.
*
* Covers three outcomes plus one negative assertion:
* 1. resolveContainerIp returns null -> route.resolve.fail / container_not_found
* 2. socket emits 'error' pre-connect -> route.resolve.fail / connect_error
* 3. socket emits 'connect' -> route.resolve.ok
* 4. post-connect close/error does NOT emit a second route.resolve.fail
*/
import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest';
import net from 'net';
import { EventEmitter } from 'events';
import { WebSocket } from 'ws';
import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb';
import { AGENT_REVERSE_ID_BASE, encodeJsonFrame, decodeJsonFrame } from '../pilot/protocol';
import type { MeshActivityEvent } from '../services/MeshService';
type LogActivityArgs = [Omit<MeshActivityEvent, 'ts'>];
let tmpDir: string;
let PilotTunnelBridge: typeof import('../services/PilotTunnelBridge').PilotTunnelBridge;
let MeshService: typeof import('../services/MeshService').MeshService;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ PilotTunnelBridge } = await import('../services/PilotTunnelBridge'));
({ MeshService } = await import('../services/MeshService'));
});
afterAll(() => {
vi.restoreAllMocks();
cleanupTestDb(tmpDir);
});
function makeMockTunnelWs(): EventEmitter & {
sent: unknown[];
readyState: number;
bufferedAmount: number;
send: (data: unknown) => void;
ping: () => void;
close: () => void;
} {
const ws = new EventEmitter() as EventEmitter & {
sent: unknown[]; readyState: number; bufferedAmount: number;
send: (data: unknown) => void; ping: () => void; close: () => void;
};
ws.sent = [];
ws.readyState = WebSocket.OPEN;
ws.bufferedAmount = 0;
ws.send = (data: unknown) => { ws.sent.push(data); };
ws.ping = () => { /* no-op */ };
ws.close = () => { ws.readyState = WebSocket.CLOSED; ws.emit('close'); };
return ws;
}
function findAck(ws: { sent: unknown[] }, s: number): { ok: boolean; err?: string } | undefined {
for (const item of ws.sent) {
if (typeof item !== 'string') continue;
try {
const f = decodeJsonFrame(item);
if (f.t === 'tcp_open_ack' && f.s === s) return { ok: f.ok, err: f.err };
} catch { /* ignore */ }
}
return undefined;
}
async function waitFor<T>(check: () => T | undefined): Promise<T> {
const deadline = Date.now() + 1500;
while (Date.now() < deadline) {
const v = check();
if (v !== undefined) return v;
await new Promise((r) => setTimeout(r, 10));
}
throw new Error('timeout');
}
describe('R1-B: PilotTunnelBridge.acceptReverseLocal route events', () => {
let logSpy: ReturnType<typeof vi.fn<(event: Omit<MeshActivityEvent, 'ts'>) => void>>;
beforeEach(() => {
vi.restoreAllMocks();
logSpy = vi.fn<(event: Omit<MeshActivityEvent, 'ts'>) => void>();
vi.spyOn(MeshService.getInstance(), 'logActivity').mockImplementation(logSpy);
});
it('emits route.resolve.fail with direction=reverse + reason=container_not_found when IP is unresolved', async () => {
const mockWs = makeMockTunnelWs();
const peerNodeId = 42;
const bridge = new PilotTunnelBridge(peerNodeId, mockWs as unknown as WebSocket);
await bridge.start();
const localNodeId = (await import('../services/NodeRegistry')).NodeRegistry.getInstance().getDefaultNodeId();
vi.spyOn(MeshService.getInstance(), 'resolveContainerIp').mockResolvedValue(null);
const s = AGENT_REVERSE_ID_BASE + 11;
mockWs.emit('message', encodeJsonFrame({
t: 'tcp_open_reverse', s,
targetNodeId: localNodeId, stack: 'missing', service: 'svc', port: 80,
}), false);
const ack = await waitFor(() => findAck(mockWs, s));
expect(ack.ok).toBe(false);
const failCall = logSpy.mock.calls.find(
(c: LogActivityArgs) => c[0].type === 'route.resolve.fail',
);
expect(failCall).toBeDefined();
const payload = failCall![0] as {
source: string;
level: string;
type: string;
nodeId?: number;
details?: Record<string, unknown>;
};
expect(payload.source).toBe('mesh');
expect(payload.level).toBe('error');
expect(payload.type).toBe('route.resolve.fail');
expect(payload.nodeId).toBe(peerNodeId);
expect(payload.details).toMatchObject({
direction: 'reverse',
reason: 'container_not_found',
targetStack: 'missing',
targetService: 'svc',
targetPort: 80,
});
bridge.close();
});
it('emits route.resolve.fail with direction=reverse + reason=connect_error on socket pre-connect error', async () => {
const mockWs = makeMockTunnelWs();
const peerNodeId = 43;
const bridge = new PilotTunnelBridge(peerNodeId, mockWs as unknown as WebSocket);
await bridge.start();
const localNodeId = (await import('../services/NodeRegistry')).NodeRegistry.getInstance().getDefaultNodeId();
// Resolve to localhost on a port nothing is listening on so the dial
// produces a synchronous ECONNREFUSED via the OS.
vi.spyOn(MeshService.getInstance(), 'resolveContainerIp').mockResolvedValue('127.0.0.1');
// Pick a port we know is closed by binding and immediately releasing.
const probe = net.createServer();
await new Promise<void>((resolve) => probe.listen(0, '127.0.0.1', () => resolve()));
const addr = probe.address();
if (!addr || typeof addr === 'string') throw new Error('no address');
const closedPort = addr.port;
await new Promise<void>((resolve) => probe.close(() => resolve()));
const s = AGENT_REVERSE_ID_BASE + 12;
mockWs.emit('message', encodeJsonFrame({
t: 'tcp_open_reverse', s,
targetNodeId: localNodeId, stack: 'real', service: 'svc', port: closedPort,
}), false);
const ack = await waitFor(() => findAck(mockWs, s));
expect(ack.ok).toBe(false);
expect(ack.err).toBe('unreachable');
const failCall = logSpy.mock.calls.find(
(c: LogActivityArgs) => c[0].type === 'route.resolve.fail',
);
expect(failCall).toBeDefined();
const payload = failCall![0] as {
type: string;
nodeId?: number;
details?: Record<string, unknown>;
};
expect(payload.type).toBe('route.resolve.fail');
expect(payload.nodeId).toBe(peerNodeId);
expect(payload.details).toMatchObject({
direction: 'reverse',
reason: 'connect_error',
targetStack: 'real',
targetService: 'svc',
targetPort: closedPort,
});
bridge.close();
});
it('emits route.resolve.ok with direction=reverse on connect ack', async () => {
const mockWs = makeMockTunnelWs();
const peerNodeId = 44;
const bridge = new PilotTunnelBridge(peerNodeId, mockWs as unknown as WebSocket);
await bridge.start();
// Real local server so the dial succeeds and 'connect' fires.
const upstream = net.createServer((socket) => {
socket.write('hello-upstream');
});
await new Promise<void>((resolve) => upstream.listen(0, '127.0.0.1', () => resolve()));
const addr = upstream.address();
if (!addr || typeof addr === 'string') throw new Error('no address');
const upstreamPort = addr.port;
const localNodeId = (await import('../services/NodeRegistry')).NodeRegistry.getInstance().getDefaultNodeId();
vi.spyOn(MeshService.getInstance(), 'resolveContainerIp').mockResolvedValue('127.0.0.1');
const s = AGENT_REVERSE_ID_BASE + 13;
mockWs.emit('message', encodeJsonFrame({
t: 'tcp_open_reverse', s,
targetNodeId: localNodeId, stack: 'real', service: 'svc', port: upstreamPort,
}), false);
const ack = await waitFor(() => findAck(mockWs, s));
expect(ack.ok).toBe(true);
// Wait for the ok event to land (logActivity is sync but the connect
// callback is async relative to message handling).
await waitFor(() => logSpy.mock.calls.find(
(c: LogActivityArgs) => c[0].type === 'route.resolve.ok',
));
const okCall = logSpy.mock.calls.find(
(c: LogActivityArgs) => c[0].type === 'route.resolve.ok',
);
expect(okCall).toBeDefined();
const payload = okCall![0] as {
source: string;
level: string;
type: string;
nodeId?: number;
details?: Record<string, unknown>;
};
expect(payload.source).toBe('mesh');
expect(payload.level).toBe('info');
expect(payload.type).toBe('route.resolve.ok');
expect(payload.nodeId).toBe(peerNodeId);
expect(payload.details).toMatchObject({
direction: 'reverse',
targetStack: 'real',
targetService: 'svc',
targetPort: upstreamPort,
});
upstream.close();
bridge.close();
});
it('does NOT emit route.resolve.fail when a connected socket closes post-handshake', async () => {
const mockWs = makeMockTunnelWs();
const peerNodeId = 45;
const bridge = new PilotTunnelBridge(peerNodeId, mockWs as unknown as WebSocket);
await bridge.start();
// Real local server. Have it close the connection immediately after
// accept so the bridge socket sees 'close' (and possibly 'error') on
// an already-connected socket.
const upstream = net.createServer((socket) => {
socket.end();
});
await new Promise<void>((resolve) => upstream.listen(0, '127.0.0.1', () => resolve()));
const addr = upstream.address();
if (!addr || typeof addr === 'string') throw new Error('no address');
const upstreamPort = addr.port;
const localNodeId = (await import('../services/NodeRegistry')).NodeRegistry.getInstance().getDefaultNodeId();
vi.spyOn(MeshService.getInstance(), 'resolveContainerIp').mockResolvedValue('127.0.0.1');
const s = AGENT_REVERSE_ID_BASE + 14;
mockWs.emit('message', encodeJsonFrame({
t: 'tcp_open_reverse', s,
targetNodeId: localNodeId, stack: 'real', service: 'svc', port: upstreamPort,
}), false);
// Wait for connect (the success ack confirms onPreConnectError was removed).
const ack = await waitFor(() => findAck(mockWs, s));
expect(ack.ok).toBe(true);
// Wait for the success event to confirm the ok path fired.
await waitFor(() => logSpy.mock.calls.find(
(c: LogActivityArgs) => c[0].type === 'route.resolve.ok',
));
// Give the post-connect close a chance to fire teardown handlers.
await new Promise((r) => setTimeout(r, 100));
const failCalls = logSpy.mock.calls.filter(
(c: LogActivityArgs) => c[0].type === 'route.resolve.fail',
);
expect(failCalls).toHaveLength(0);
upstream.close();
bridge.close();
});
});
@@ -0,0 +1,54 @@
/**
* PilotTunnelManager.replaceOrRegisterProxyBridge: peer-initiated proxy
* bridges (Phase R1) need to be able to supersede a previous proxy bridge
* for the same nodeId, but must never shadow a live pilot tunnel (which
* always wins the bridge slot).
*/
import { describe, it, expect, beforeEach, vi } from 'vitest';
import { EventEmitter } from 'events';
import { PilotTunnelManager } from '../services/PilotTunnelManager';
import type { PilotTunnelBridge } from '../services/PilotTunnelBridge';
type FakeBridge = PilotTunnelBridge & {
close: ReturnType<typeof vi.fn<(code?: number, reason?: string) => void>>;
getActiveStreamCount: ReturnType<typeof vi.fn<() => number>>;
};
function makeFakeBridge(): FakeBridge {
const ee = new EventEmitter() as unknown as FakeBridge;
ee.close = vi.fn<(code?: number, reason?: string) => void>();
ee.getActiveStreamCount = vi.fn<() => number>().mockReturnValue(0);
return ee;
}
describe('PilotTunnelManager.replaceOrRegisterProxyBridge', () => {
beforeEach(() => PilotTunnelManager.resetForTest?.());
it('registers when no existing bridge', () => {
const mgr = PilotTunnelManager.getInstance();
const bridge = makeFakeBridge();
mgr.replaceOrRegisterProxyBridge(42, bridge);
expect(mgr.getBridge(42)).toBe(bridge);
});
it('replaces when existing bridge is a proxy bridge (closes the old one)', () => {
const mgr = PilotTunnelManager.getInstance();
const oldBridge = makeFakeBridge();
const newBridge = makeFakeBridge();
mgr.injectBridgeForTest(42, oldBridge, 'proxy');
mgr.replaceOrRegisterProxyBridge(42, newBridge);
expect(oldBridge.close).toHaveBeenCalledWith(1000, 'replaced-by-newer-proxy');
expect(mgr.getBridge(42)).toBe(newBridge);
});
it('throws when existing bridge is a pilot tunnel (preserves the pilot tunnel)', () => {
const mgr = PilotTunnelManager.getInstance();
const pilotBridge = makeFakeBridge();
const proxyBridge = makeFakeBridge();
mgr.injectBridgeForTest(42, pilotBridge, 'pilot');
expect(() => mgr.replaceOrRegisterProxyBridge(42, proxyBridge))
.toThrow(/pilot tunnel.*proxy bridge refused/);
expect(mgr.getBridge(42)).toBe(pilotBridge);
expect(pilotBridge.close).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,83 @@
/**
* Tests for the `centralCallback` diag block returned by
* GET /api/system/pilot-tunnels. The block reads from MeshCentralRegistry
* (cached central material + last-used/last-rejected timestamps) and from
* PeerToCentralMeshSessionDialer (live bridge presence). Counters for the
* peer-callback path live in PilotMetrics and are exposed via the existing
* `counters` field of the same response, so this block only adds the
* cached-row diagnostics that PilotMetrics does not track.
*/
import { describe, it, expect, beforeAll, beforeEach, afterAll } from 'vitest';
import request from 'supertest';
import type { Express } from 'express';
import { setupTestDb, cleanupTestDb, loginAsTestAdmin } from './helpers/setupTestDb';
import { MeshCentralRegistry } from '../services/MeshCentralRegistry';
import { PeerToCentralMeshSessionDialer } from '../services/PeerToCentralMeshSessionDialer';
import { DatabaseService } from '../services/DatabaseService';
let tmpDir: string;
let app: Express;
let cookie: string;
beforeAll(async () => {
tmpDir = await setupTestDb();
({ app } = await import('../index'));
cookie = await loginAsTestAdmin(app);
});
afterAll(() => cleanupTestDb(tmpDir));
beforeEach(() => {
MeshCentralRegistry.resetForTest();
PeerToCentralMeshSessionDialer.resetForTest();
DatabaseService.getInstance().getDb().prepare('DELETE FROM mesh_centrals').run();
});
describe('/api/system/pilot-tunnels centralCallback diag', () => {
it('returns null centralCallback fields when no central material is cached', async () => {
const res = await request(app).get('/api/system/pilot-tunnels').set('Cookie', cookie);
expect(res.status).toBe(200);
expect(res.body.centralCallback).toEqual({
bridgeOpen: false,
lastBootstrapAt: null,
lastDialOkAt: null,
lastDialFailAt: null,
lastDialFailReason: null,
});
});
it('returns populated centralCallback fields when material is present', async () => {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-1',
centralApiUrl: 'https://central.example.com',
callbackJwt: 'jwt-value',
jwtIssuedAt: 1,
jwtExpiresAt: 9999999999,
});
MeshCentralRegistry.getInstance().markUsed('inst-1');
const res = await request(app).get('/api/system/pilot-tunnels').set('Cookie', cookie);
expect(res.status).toBe(200);
expect(res.body.centralCallback.bridgeOpen).toBe(false);
expect(res.body.centralCallback.lastBootstrapAt).toBeGreaterThan(0);
expect(res.body.centralCallback.lastDialOkAt).toBeGreaterThan(0);
expect(res.body.centralCallback.lastDialFailAt).toBeNull();
expect(res.body.centralCallback.lastDialFailReason).toBeNull();
});
it('surfaces last reject reason when central marks a dial failed', async () => {
MeshCentralRegistry.getInstance().upsert({
centralInstanceId: 'inst-2',
centralApiUrl: 'https://central.example.com',
callbackJwt: 'jwt-value',
jwtIssuedAt: 1,
jwtExpiresAt: 9999999999,
});
MeshCentralRegistry.getInstance().markRejected('inst-2', 'signature_invalid');
const res = await request(app).get('/api/system/pilot-tunnels').set('Cookie', cookie);
expect(res.status).toBe(200);
expect(res.body.centralCallback.lastDialFailAt).toBeGreaterThan(0);
expect(res.body.centralCallback.lastDialFailReason).toBe('signature_invalid');
});
});
+18 -1
View File
@@ -6,6 +6,8 @@ import { DatabaseService } from '../services/DatabaseService';
import { CacheService } from '../services/CacheService';
import { NodeRegistry } from '../services/NodeRegistry';
import { PilotTunnelManager } from '../services/PilotTunnelManager';
import { MeshCentralRegistry } from '../services/MeshCentralRegistry';
import { PeerToCentralMeshSessionDialer } from '../services/PeerToCentralMeshSessionDialer';
import { authMiddleware } from '../middleware/auth';
import { requireAdmin } from '../middleware/tierGates';
import { STATS_CACHE_TTL_MS, SYSTEM_STATS_CACHE_TTL_MS } from '../helpers/constants';
@@ -260,7 +262,22 @@ metricsRouter.get('/system/cache-stats', authMiddleware, async (req: Request, re
metricsRouter.get('/system/pilot-tunnels', authMiddleware, async (req: Request, res: Response): Promise<void> => {
if (!requireAdmin(req, res)) return;
try {
res.json(PilotTunnelManager.getInstance().getMetricsSnapshot());
const snapshot = PilotTunnelManager.getInstance().getMetricsSnapshot();
// centralCallback exposes the peer-side view of the symmetric mesh
// callback path (cached central material plus the live bridge presence).
// PilotMetrics already tracks the attempt/success counters via
// `counters.mesh_central_bootstraps_total` and friends; this block adds
// the per-instance last-success / last-failure diagnostics that live in
// MeshCentralRegistry rather than PilotMetrics.
const cached = MeshCentralRegistry.getInstance().getActive();
const centralCallback = {
bridgeOpen: PeerToCentralMeshSessionDialer.getInstance().hasSession(),
lastBootstrapAt: cached?.lastBootstrapAt ?? null,
lastDialOkAt: cached?.lastUsedAt ?? null,
lastDialFailAt: cached?.lastRejectedAt ?? null,
lastDialFailReason: cached?.lastRejectReason ?? null,
};
res.json({ ...snapshot, centralCallback });
} catch (error) {
console.error('Failed to fetch pilot tunnel metrics:', error);
res.status(500).json({ error: 'Failed to fetch pilot tunnel metrics' });
+16
View File
@@ -13,6 +13,7 @@ import { REMOTE_META_NAMESPACE } from '../helpers/cacheInvalidation';
import { CAPABILITIES, getSenchoVersion, type RemoteMeta } from '../services/CapabilityRegistry';
import { PilotTunnelManager } from '../services/PilotTunnelManager';
import { PilotCloseCode } from '../pilot/protocol';
import { MeshProxyTunnelDialer } from '../services/MeshProxyTunnelDialer';
import { FleetUpdateTrackerService } from '../services/FleetUpdateTrackerService';
import { FleetSyncService } from '../services/FleetSyncService';
import { isValidRemoteUrl } from '../utils/validation';
@@ -235,6 +236,21 @@ nodesRouter.put('/:id', async (req: Request, res: Response) => {
NodeRegistry.getInstance().evictConnection(id);
NodeRegistry.getInstance().notifyNodeUpdated(id);
// Trigger 2: if the api_token was rotated on a mesh-enabled proxy-mode
// remote, close the existing callback bridge and re-dial. The next
// ensureBridge mints a JWT with the fresh token fingerprint so the
// remote's tunnel auth gate accepts the upgrade.
if (typeof updates.api_token === 'string') {
const node = DatabaseService.getInstance().getNode(id);
const meshEnabled = DatabaseService.getInstance().getNodeMeshEnabled(id);
if (node && node.type === 'remote' && node.mode === 'proxy' && meshEnabled) {
MeshProxyTunnelDialer.getInstance().closeBridge(id, 'peer token rotated');
void MeshProxyTunnelDialer.getInstance().ensureBridge(id).catch((err) => {
console.warn(`[Mesh] proactive re-bootstrap on token rotation failed for node ${id}: ${(err as Error).message}`);
});
}
}
const isPlainHttp = updates.api_url && updates.api_url.startsWith('http://');
res.json({
success: true,
@@ -34,6 +34,7 @@ export const CAPABILITIES = [
'registries',
'self-update',
'vulnerability-scanning',
'mesh_proxy_callback_bootstrap',
] as const;
export type Capability = (typeof CAPABILITIES)[number];
+17
View File
@@ -1475,6 +1475,23 @@ export class DatabaseService {
} catch (e) {
console.warn('[DatabaseService] Could not create mesh_stacks:', (e as Error).message);
}
try {
this.db.prepare(`
CREATE TABLE IF NOT EXISTS mesh_centrals (
central_instance_id TEXT PRIMARY KEY,
central_api_url TEXT NOT NULL,
callback_jwt TEXT NOT NULL,
jwt_issued_at INTEGER NOT NULL,
jwt_expires_at INTEGER NOT NULL,
last_bootstrap_at INTEGER NOT NULL,
last_used_at INTEGER,
last_rejected_at INTEGER,
last_reject_reason TEXT
)
`).run();
} catch (e) {
console.warn('[DatabaseService] Could not create mesh_centrals:', (e as Error).message);
}
this.tryAddColumn('nodes', 'mesh_enabled', 'INTEGER NOT NULL DEFAULT 0');
}
+119
View File
@@ -0,0 +1,119 @@
import { EventEmitter } from 'events';
import { DatabaseService } from './DatabaseService';
export interface MeshCentralMaterial {
centralInstanceId: string;
centralApiUrl: string;
callbackJwt: string;
jwtIssuedAt: number;
jwtExpiresAt: number;
}
export interface MeshCentralRow extends MeshCentralMaterial {
lastBootstrapAt: number;
lastUsedAt: number | null;
lastRejectedAt: number | null;
lastRejectReason: string | null;
}
export class MeshCentralRegistry extends EventEmitter {
private static instance: MeshCentralRegistry | null = null;
private warnedMultiRow = false;
private constructor() { super(); }
public static getInstance(): MeshCentralRegistry {
if (!this.instance) this.instance = new MeshCentralRegistry();
return this.instance;
}
public static resetForTest(): void {
this.instance = null;
}
public upsert(material: MeshCentralMaterial): void {
const db = DatabaseService.getInstance().getDb();
const prior = this.getActive();
db.prepare(`
INSERT INTO mesh_centrals (
central_instance_id, central_api_url, callback_jwt,
jwt_issued_at, jwt_expires_at, last_bootstrap_at,
last_used_at, last_rejected_at, last_reject_reason
) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, NULL)
ON CONFLICT(central_instance_id) DO UPDATE SET
central_api_url = excluded.central_api_url,
callback_jwt = excluded.callback_jwt,
jwt_issued_at = excluded.jwt_issued_at,
jwt_expires_at = excluded.jwt_expires_at,
last_bootstrap_at = excluded.last_bootstrap_at
`).run(
material.centralInstanceId,
material.centralApiUrl,
material.callbackJwt,
material.jwtIssuedAt,
material.jwtExpiresAt,
Date.now(),
);
const isInstanceChange = prior && prior.centralInstanceId !== material.centralInstanceId;
if (isInstanceChange) {
this.emit('central-instance-changed', {
previousInstanceId: prior!.centralInstanceId,
newInstanceId: material.centralInstanceId,
});
}
this.emit('central-bootstrap', { centralInstanceId: material.centralInstanceId });
}
public getActive(): MeshCentralRow | null {
const db = DatabaseService.getInstance().getDb();
const rows = db.prepare(`
SELECT * FROM mesh_centrals ORDER BY last_bootstrap_at DESC
`).all() as Array<{
central_instance_id: string;
central_api_url: string;
callback_jwt: string;
jwt_issued_at: number;
jwt_expires_at: number;
last_bootstrap_at: number;
last_used_at: number | null;
last_rejected_at: number | null;
last_reject_reason: string | null;
}>;
if (rows.length === 0) return null;
if (rows.length > 1 && !this.warnedMultiRow) {
console.warn(`[MeshCentralRegistry] multiple central rows present (${rows.length}), multi-central unsupported in v0.79`);
this.warnedMultiRow = true;
}
const r = rows[0];
return {
centralInstanceId: r.central_instance_id,
centralApiUrl: r.central_api_url,
callbackJwt: r.callback_jwt,
jwtIssuedAt: r.jwt_issued_at,
jwtExpiresAt: r.jwt_expires_at,
lastBootstrapAt: r.last_bootstrap_at,
lastUsedAt: r.last_used_at,
lastRejectedAt: r.last_rejected_at,
lastRejectReason: r.last_reject_reason,
};
}
public clearForInstance(centralInstanceId: string): void {
DatabaseService.getInstance().getDb()
.prepare(`DELETE FROM mesh_centrals WHERE central_instance_id = ?`)
.run(centralInstanceId);
}
public markUsed(centralInstanceId: string): void {
DatabaseService.getInstance().getDb()
.prepare(`UPDATE mesh_centrals SET last_used_at = ? WHERE central_instance_id = ?`)
.run(Date.now(), centralInstanceId);
}
public markRejected(centralInstanceId: string, reason: string): void {
DatabaseService.getInstance().getDb()
.prepare(`UPDATE mesh_centrals SET last_rejected_at = ?, last_reject_reason = ? WHERE central_instance_id = ?`)
.run(Date.now(), reason, centralInstanceId);
}
}
+219 -14
View File
@@ -1,9 +1,12 @@
import { EventEmitter } from 'events';
import { createHash } from 'crypto';
import jwt from 'jsonwebtoken';
import WebSocket from 'ws';
import { MAX_FRAME_SIZE_BYTES } from '../pilot/protocol';
import { PilotTunnelBridge, type MeshTunnelHandle } from './PilotTunnelBridge';
import { PilotTunnelManager } from './PilotTunnelManager';
import { NodeRegistry } from './NodeRegistry';
import { DatabaseService } from './DatabaseService';
import { redactSensitiveText, sanitizeForLog } from '../utils/safeLog';
import { httpUrlToWs } from '../utils/wsUrl';
import { isDebugEnabled } from '../utils/debug';
@@ -46,6 +49,31 @@ export type DialFailureCode =
| 'tls_failed'
| 'network_error';
/**
* Reason union for `proxy-bridge-down` events. Centralizes the categories
* the dialer emits so subscribers (reactive redial, metrics, UI) can branch
* deterministically without parsing free-form strings.
* - `idle`: idle sweeper closed the bridge after no streams for `idleTtlMs`.
* - `remote_closed`: peer closed the WS cleanly (1000/1001) or with an
* unclassified non-error code.
* - `network_error`: WS dropped abnormally (1006).
* - `protocol_error`: WS closed for protocol/policy reasons (1007/1008/1009).
* - `auth_failed`: dial-time auth rejection (terminal, no redial).
*/
export type BridgeDownReason =
| 'idle'
| 'remote_closed'
| 'network_error'
| 'protocol_error'
| 'auth_failed';
function classifyCloseCode(code?: number): BridgeDownReason {
if (code === 1000 || code === 1001) return 'remote_closed';
if (code === 1006) return 'network_error';
if (code === 1007 || code === 1008 || code === 1009) return 'protocol_error';
return 'remote_closed';
}
/**
* Activity-log reason. Wider than `DialFailureCode` because some failures
* share a wire-level code (e.g., `network_error`) but deserve a more
@@ -76,12 +104,20 @@ export class MeshProxyTunnelDialer extends EventEmitter {
private readonly inflight = new Map<number, Promise<MeshTunnelHandle | null>>();
private readonly idleSince = new Map<number, number>();
private readonly recentFailures = new Map<number, DialFailure>();
private readonly redialAttempts = new Map<number, number>();
private readonly redialTimers = new Map<number, NodeJS.Timeout>();
private readonly idleTtlMs: number;
private idleCheckTimer: NodeJS.Timeout | null = null;
private stopped = false;
private constructor(idleTtlOverrideMs?: number) {
super();
// Reactive redial: any non-terminal teardown triggers a backoff redial.
// `idle` is intentional and `auth_failed` is terminal, so both skip.
this.on('proxy-bridge-down', (nodeId: number, reason: BridgeDownReason) => {
if (reason === 'idle' || reason === 'auth_failed') return;
this.scheduleReactiveRedial(nodeId);
});
if (typeof idleTtlOverrideMs === 'number') {
this.idleTtlMs = idleTtlOverrideMs;
} else {
@@ -163,6 +199,9 @@ export class MeshProxyTunnelDialer extends EventEmitter {
this.bridges.delete(nodeId);
try { bridge.close(1000, 'dialer shutdown'); } catch { /* ignore */ }
}
for (const t of this.redialTimers.values()) clearTimeout(t);
this.redialTimers.clear();
this.redialAttempts.clear();
this.idleSince.clear();
this.recentFailures.clear();
this.inflight.clear();
@@ -223,6 +262,18 @@ export class MeshProxyTunnelDialer extends EventEmitter {
return null;
}
// Capability-gated mesh_handshake: pushes a signed `mesh_tunnel` JWT
// and the central origin so the peer can dial central back for
// reverse cross-fleet TCP streams (R2). Skipped silently when the
// peer lacks the capability or `SENCHO_PRIMARY_URL` is unset so
// legacy peers keep their forward-only TCP path untouched.
await this.maybeSendBootstrap(nodeId, ws);
if (this.stopped) {
try { ws.close(1001, 'dialer shutdown'); } catch { /* ignore */ }
return null;
}
const bridge = new PilotTunnelBridge(nodeId, ws);
try {
await bridge.start();
@@ -249,17 +300,13 @@ export class MeshProxyTunnelDialer extends EventEmitter {
// Mirror the registration in the dialer's own map so the idle
// sweeper can call `getActiveStreamCount()` (not on the narrow
// MeshTunnelHandle interface) without poking into the manager.
bridge.once('closed', () => {
if (this.bridges.get(nodeId) === bridge) {
this.bridges.delete(nodeId);
this.idleSince.delete(nodeId);
void this.logActivity(nodeId, 'close', { reason: 'remote-closed' });
this.emit('proxy-bridge-down', nodeId);
}
});
this.bridges.set(nodeId, bridge);
this.attachBridgeCloseListener(nodeId, bridge);
this.idleSince.set(nodeId, Date.now());
this.recentFailures.delete(nodeId);
// A successful open clears any prior reactive-redial backoff so the
// next failure starts fresh from attempt #1.
this.redialAttempts.delete(nodeId);
void this.logActivity(nodeId, 'open.ok', {});
this.emit('proxy-bridge-up', nodeId);
if (isDebugEnabled()) {
@@ -326,12 +373,7 @@ export class MeshProxyTunnelDialer extends EventEmitter {
}
const last = this.idleSince.get(nodeId) ?? now;
if (now - last >= this.idleTtlMs) {
this.bridges.delete(nodeId);
this.idleSince.delete(nodeId);
try { bridge.close(1000, 'idle timeout'); } catch { /* ignore */ }
PilotMetrics.increment('proxy_idle_closes');
void this.logActivity(nodeId, 'close', { reason: 'idle' });
this.emit('proxy-bridge-down', nodeId);
this.tearDownBridge(nodeId, 'idle', { code: 1000, message: 'idle timeout' });
if (isDebugEnabled()) {
console.log(`[MeshProxyDialer:diag] idle close node=${nodeId} idleMs=${now - last}`);
}
@@ -339,6 +381,117 @@ export class MeshProxyTunnelDialer extends EventEmitter {
}
}
/**
* Single teardown path for any bridge close. Removes the bridge from
* the map, optionally invokes `bridge.close()` (skipped when the close
* originated from the bridge itself), records the activity log entry,
* bumps the idle-close metric when applicable, and emits a reason-tagged
* `proxy-bridge-down`. The self-listener installed in the constructor
* decides whether to schedule a reactive redial.
*/
private tearDownBridge(
nodeId: number,
reason: BridgeDownReason,
closeArgs?: { code: number; message: string },
): void {
const bridge = this.bridges.get(nodeId);
if (!bridge) return;
this.bridges.delete(nodeId);
this.idleSince.delete(nodeId);
if (closeArgs) {
// Best-effort: closing an already-closed WS is idempotent and
// can throw on edge states; we never want teardown to propagate.
try { bridge.close(closeArgs.code, closeArgs.message); } catch { /* best-effort */ }
}
if (reason === 'idle') PilotMetrics.increment('proxy_idle_closes');
void this.logActivity(nodeId, 'close', { reason });
this.emit('proxy-bridge-down', nodeId, reason);
}
/**
* Wire the `closed` listener that classifies the WS close code and
* routes through `tearDownBridge` without re-closing the bridge. Called
* from `dial()` after registration; also reachable from tests via a
* private cast so they can exercise the close-code mapping without
* spinning up a real WebSocket.
*/
private attachBridgeCloseListener(nodeId: number, bridge: EventEmitter): void {
bridge.once('closed', (info?: { code?: number }) => {
if (this.bridges.get(nodeId) !== bridge) return;
const reason = classifyCloseCode(info?.code);
this.tearDownBridge(nodeId, reason);
});
}
/**
* Schedule a reactive redial with exponential backoff + jitter. Caps at
* 8 attempts (~5 minutes between the last few). A successful `dial()`
* clears `redialAttempts[nodeId]`, so the counter only grows during a
* sustained outage.
*/
private scheduleReactiveRedial(nodeId: number): void {
if (this.stopped) return;
const attempt = (this.redialAttempts.get(nodeId) ?? 0) + 1;
if (attempt > 8) {
this.redialAttempts.delete(nodeId);
void this.logActivity(nodeId, 'open.fail', { reason: 'redial_exhausted', attempts: 8 });
return;
}
this.redialAttempts.set(nodeId, attempt);
const baseMs = Math.min(5_000 * 2 ** (attempt - 1), 5 * 60_000);
const jitter = Math.floor(Math.random() * baseMs * 0.3);
const delay = baseMs + jitter;
const existing = this.redialTimers.get(nodeId);
if (existing) clearTimeout(existing);
const timer = setTimeout(() => {
this.redialTimers.delete(nodeId);
void this.ensureBridge(nodeId);
}, delay);
timer.unref?.();
this.redialTimers.set(nodeId, timer);
}
/**
* If the peer advertises `mesh_proxy_callback_bootstrap` and central has
* a canonical origin (`SENCHO_PRIMARY_URL`), mint a `mesh_tunnel`-scoped
* JWT bound to the peer's `api_token` fingerprint and push it as the
* first text frame on the freshly-opened WS. The peer's
* `meshProxyTunnel` first-frame state machine persists the bootstrap
* material before yielding to TCP traffic. Every other path is a silent
* skip so legacy peers continue to receive raw TCP frames immediately.
*/
private async maybeSendBootstrap(nodeId: number, ws: WebSocket): Promise<void> {
const canonicalOrigin = (process.env.SENCHO_PRIMARY_URL ?? '').replace(/\/+$/, '');
if (!canonicalOrigin) return;
let meta;
try { meta = await NodeRegistry.getInstance().fetchMetaForNode(nodeId); }
catch { return; }
if (!meta?.online) return;
if (!meta.capabilities?.includes('mesh_proxy_callback_bootstrap')) return;
const db = DatabaseService.getInstance();
const node = db.getNode(nodeId);
if (!node || node.type !== 'remote' || node.mode !== 'proxy' || !node.api_token) return;
const authSecret = db.getGlobalSettings().auth_jwt_secret;
const centralInstanceId = db.getSystemState('instance_id');
if (!authSecret || !centralInstanceId) return;
const built = buildHandshakeFrame(nodeId, node.api_token, canonicalOrigin, centralInstanceId, authSecret);
try {
ws.send(JSON.stringify(built.frame));
void this.logActivity(nodeId, 'open.ok', {
bootstrapSent: true,
centralApiUrl: redactSensitiveText(canonicalOrigin),
kid: 'v1',
jwtExpiresAt: built.expSec,
});
} catch (err) {
console.warn(`[MeshProxyDialer] mesh_handshake send failed for node ${nodeId}: ${(err as Error).message}`.replace(/[\n\r]/g, ''));
}
}
private async logActivity(
nodeId: number,
event: ProxyTunnelEvent,
@@ -364,6 +517,58 @@ export class MeshProxyTunnelDialer extends EventEmitter {
}
}
interface HandshakeFrame {
t: 'mesh_handshake';
v: 1;
peerNodeId: number;
centralInstanceId: string;
centralApiUrl: string;
meshTunnelJwt: string;
jwtExpiresAt: number;
}
/**
* Build the mesh_handshake frame and the associated JWT. Pure helper so the
* dialer's `maybeSendBootstrap` stays under the 30-line ceiling.
*/
function buildHandshakeFrame(
nodeId: number,
apiToken: string,
canonicalOrigin: string,
centralInstanceId: string,
authSecret: string,
): { frame: HandshakeFrame; expSec: number } {
const peerTokenFp = createHash('sha256').update(apiToken).digest('hex').slice(0, 16);
const nowSec = Math.floor(Date.now() / 1000);
const expSec = nowSec + 365 * 24 * 3600;
const meshTunnelJwt = jwt.sign(
{
sub: String(nodeId),
iss: centralInstanceId,
aud: canonicalOrigin,
scope: 'mesh_tunnel',
iat: nowSec,
exp: expSec,
kid: 'v1',
peer_token_fp: peerTokenFp,
},
authSecret,
{ algorithm: 'HS256' },
);
return {
frame: {
t: 'mesh_handshake',
v: 1,
peerNodeId: nodeId,
centralInstanceId,
centralApiUrl: canonicalOrigin,
meshTunnelJwt,
jwtExpiresAt: expSec,
},
expSec,
};
}
function classifyDialError(err: unknown): { code: DialFailureCode; message: string } {
const message = sanitizeForLog((err as Error).message || String(err));
const httpStatus = (err as Error & { httpStatus?: number }).httpStatus;
+118 -2
View File
@@ -64,7 +64,7 @@ export function getSenchoIpFromSubnet(subnet: string): string {
export type MeshActivitySource = 'pilot' | 'mesh';
export type MeshActivityLevel = 'info' | 'warn' | 'error';
export type MeshActivityType =
| 'route.dispatch' | 'route.resolve.ok' | 'route.resolve.denied'
| 'route.dispatch' | 'route.resolve.ok' | 'route.resolve.denied' | 'route.resolve.fail'
| 'tunnel.open' | 'tunnel.fail' | 'tunnel.backpressure'
| 'opt_in' | 'opt_out'
| 'mesh.enable' | 'mesh.disable'
@@ -72,7 +72,8 @@ export type MeshActivityType =
| 'probe.ok' | 'probe.fail'
| 'forwarder.listen' | 'forwarder.unlisten' | 'forwarder.error'
| 'proxy-tunnel.open.ok' | 'proxy-tunnel.open.fail' | 'proxy-tunnel.close'
| 'mesh.proxy_tunnel.identify';
| 'mesh.proxy_tunnel.identify'
| 'mesh_handshake.received';
export interface MeshActivityEvent {
ts: number;
@@ -299,6 +300,8 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
this.selfCentralNodeId = this.resolveSelfCentralNodeId();
this.maybeWarnUnsetPrimaryUrl();
await this.setupMeshNetwork();
try {
await this.refreshAliasCache();
@@ -317,6 +320,10 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
});
}
await this.regenerateAllOverrides();
// Trigger 3: proactively re-establish central->peer bridges so any
// peer that is online and capable receives bootstrap material on
// startup. Fire-and-forget so start() does not block on remote I/O.
void this.proactiveBootstrapFanout();
this.aliasRefreshTimer = setInterval(() => {
void (async () => {
try {
@@ -345,6 +352,71 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
await this.forwarder.shutdown();
}
/**
* Operator-facing preflight: if `SENCHO_PRIMARY_URL` is unset at
* startup and the central has at least one mesh-enabled proxy-mode
* node, warn that the Task 8 capability-gated handshake will fall
* back to an inferred origin for the callback bootstrap. The matching
* fail-safe in `MeshProxyTunnelDialer` silently skips the bootstrap
* when the env is unset; this warn makes the consequence visible.
*/
private maybeWarnUnsetPrimaryUrl(): void {
if (process.env.SENCHO_PRIMARY_URL) return;
const row = DatabaseService.getInstance().getDb().prepare(`
SELECT COUNT(*) as n FROM nodes
WHERE type='remote' AND mode='proxy' AND mesh_enabled=1
`).get() as { n: number };
if (row.n > 0) {
console.warn(
'[Mesh] SENCHO_PRIMARY_URL is unset; mesh callback bootstrap will use ' +
'inferred origin. Set SENCHO_PRIMARY_URL to make peer-initiated mesh ' +
'callbacks robust against reverse-proxy edge cases.'
);
}
}
/**
* Trigger 3: at central startup, walk every mesh-enabled proxy-mode
* peer that already has at least one `mesh_stacks` row and call
* `MeshProxyTunnelDialer.ensureBridge(nodeId)`. The dialer's
* capability-gated handshake then mints and ships bootstrap material
* to peers that just came online or just got upgraded to a build that
* advertises `mesh_proxy_callback_bootstrap`. Bounded concurrency 4
* with a 250ms stagger; failures are logged and never abort the
* fan-out.
*/
private async proactiveBootstrapFanout(): Promise<void> {
const rows = DatabaseService.getInstance().getDb().prepare(`
SELECT DISTINCT n.id AS id
FROM nodes n
INNER JOIN mesh_stacks ms ON ms.node_id = n.id
WHERE n.type = 'remote' AND n.mode = 'proxy' AND n.mesh_enabled = 1
ORDER BY n.id
`).all() as Array<{ id: number }>;
const queue = rows.map((r) => r.id);
const dialer = MeshProxyTunnelDialer.getInstance();
const worker = async (): Promise<void> => {
for (;;) {
const nodeId = queue.shift();
if (nodeId === undefined) return;
try {
await dialer.ensureBridge(nodeId);
} catch (err) {
this.logActivity({
source: 'mesh', level: 'warn', type: 'proxy-tunnel.open.fail',
nodeId,
message: `boot proactive bootstrap failed: ${sanitizeForLog((err as Error).message)}`,
details: { trigger: 'startup_fanout' },
});
}
await new Promise((r) => setTimeout(r, 250));
}
};
const workerCount = Math.min(4, Math.max(1, queue.length));
await Promise.all(Array.from({ length: workerCount }, () => worker()));
}
public getSenchoIp(): string | null {
return this.senchoIp;
}
@@ -653,6 +725,17 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
source: 'mesh', level: 'info', type: 'mesh.enable',
nodeId, message: `mesh enabled on node ${nodeId}`,
});
// Trigger 1: proactive bootstrap. If the peer is a proxy-mode remote,
// dial the mesh callback bridge immediately so the capability-gated
// handshake can ship `mesh_handshake` material without waiting for
// the next forwarder dial. Fire-and-forget; failures are logged by
// the dialer.
const node = DatabaseService.getInstance().getNode(nodeId);
if (node && node.type === 'remote' && node.mode === 'proxy') {
void MeshProxyTunnelDialer.getInstance().ensureBridge(nodeId).catch((err) => {
console.warn(`[Mesh] proactive bootstrap on mesh-enable failed for node ${nodeId}: ${(err as Error).message}`);
});
}
}
public async disableForNode(nodeId: number): Promise<void> {
@@ -1506,6 +1589,39 @@ export class MeshService extends EventEmitter implements MeshForwarderHost {
message: `cross-node dispatch to ${target.alias} on node ${target.nodeId}`,
});
// Peer-side recovery: when no reverseDialer is installed, kick the
// symmetric-dial path so central learns about this peer. The
// proactive back-dial from central (registered as a side effect via
// the proxy-tunnel WS upgrade) is what actually installs the
// reverseDialer on this peer. If the session cannot be established
// (cache miss, central down, auth rejected), log route.resolve.fail
// and drop the inbound socket.
if (!this.reverseDialer) {
try {
const { PeerToCentralMeshSessionDialer } = await import('./PeerToCentralMeshSessionDialer');
const session = await PeerToCentralMeshSessionDialer.getInstance().ensureSession();
if (!session) {
this.logActivity({
source: 'mesh', level: 'warn', type: 'route.resolve.fail',
nodeId: target.nodeId, alias: target.alias,
message: `peer cross-node dispatch failed: no central callback session available`,
details: { direction: 'forward-from-peer', reason: 'no_session' },
});
try { src.destroy(); } catch { /* ignore */ }
return;
}
} catch (err) {
this.logActivity({
source: 'mesh', level: 'warn', type: 'route.resolve.fail',
nodeId: target.nodeId, alias: target.alias,
message: `peer cross-node dispatch failed: bootstrap threw`,
details: { direction: 'forward-from-peer', reason: 'bootstrap_threw' },
});
try { src.destroy(); } catch { /* ignore */ }
return;
}
}
const tcpStream = await this.dialMeshTcpStream(target);
if (!tcpStream) {
this.logActivity({
@@ -0,0 +1,184 @@
/**
* PeerToCentralMeshSessionDialer: peer-side counterpart to the central
* ingress at `/api/mesh/proxy-tunnel-from-peer`. Reads cached central
* material from `MeshCentralRegistry`, opens a WebSocket to central
* carrying the bootstrapped JWT in the Authorization header, and on a
* successful upgrade installs a local `PilotTunnelBridge` whose loopback
* URL the local proxy can target as if central had dialed us.
*
* Failure handling branches on the central's machine-readable reason
* code returned in the 401 body: terminal codes (stale,
* signature_invalid, fingerprint, node_deleted, instance_mismatch,
* audience_mismatch) clear the cache so we stop dialing with a known-bad
* credential. Transient codes (clock_skew, mode_mismatch) keep the cache
* but record the rejection so an operator can see why we stalled.
* HTTP 404 (older central without the peer ingress endpoint) keeps the
* cache and arms a longer backoff window so we do not hammer the remote.
*
* Process-local rate limit: at most 5 dial attempts per 60s window.
* Concurrency: a single in-flight promise is shared across callers so
* parallel `ensureSession()` calls coalesce into one dial.
*/
import { EventEmitter } from 'events';
import WebSocket from 'ws';
import { MAX_FRAME_SIZE_BYTES } from '../pilot/protocol';
import { MeshCentralRegistry } from './MeshCentralRegistry';
import { PilotTunnelBridge } from './PilotTunnelBridge';
import { PilotMetrics } from './PilotMetrics';
import { httpUrlToWs } from '../utils/wsUrl';
import { sanitizeForLog } from '../utils/safeLog';
const HANDSHAKE_TIMEOUT_MS = 15_000;
const RATE_LIMIT_WINDOW_MS = 60_000;
const RATE_LIMIT_MAX = 5;
const ENDPOINT_UNAVAILABLE_BACKOFF_MS = 5 * 60_000;
const CALLBACK_PATH = '/api/mesh/proxy-tunnel-from-peer';
/**
* Reasons that indicate the cached central material is permanently bad
* (wrong secret, wrong fingerprint, node removed on central, etc.) and
* should be evicted so the next mesh-enable cycle remints fresh
* material. `endpoint_not_found` is explicitly NOT in this set because
* a 404 typically means "central is older than the peer ingress",
* which is recoverable by waiting.
*/
const REJECT_REASONS_CLEAR = new Set([
'stale',
'token_fingerprint_mismatch',
'node_deleted',
'instance_mismatch',
'signature_invalid',
'audience_mismatch',
'unknown',
]);
interface DialError extends Error {
reason?: string;
httpStatus?: number;
}
export class PeerToCentralMeshSessionDialer extends EventEmitter {
private static instance: PeerToCentralMeshSessionDialer | null = null;
private currentSession: PilotTunnelBridge | null = null;
private inflight: Promise<PilotTunnelBridge | null> | null = null;
private recentDials: number[] = [];
private endpointUnavailableUntil = 0;
private constructor() { super(); }
public static getInstance(): PeerToCentralMeshSessionDialer {
if (!this.instance) this.instance = new PeerToCentralMeshSessionDialer();
return this.instance;
}
public static resetForTest(): void {
if (this.instance) {
try { this.instance.currentSession?.close(1000, 'test reset'); } catch { /* ignore */ }
}
this.instance = null;
}
public hasSession(): boolean {
return this.currentSession !== null;
}
public async ensureSession(): Promise<PilotTunnelBridge | null> {
if (this.currentSession) return this.currentSession;
if (Date.now() < this.endpointUnavailableUntil) return null;
if (this.isRateLimited()) return null;
if (this.inflight) return this.inflight;
this.inflight = this.dial().finally(() => { this.inflight = null; });
return this.inflight;
}
private isRateLimited(): boolean {
const now = Date.now();
this.recentDials = this.recentDials.filter((ts) => now - ts < RATE_LIMIT_WINDOW_MS);
return this.recentDials.length >= RATE_LIMIT_MAX;
}
private async dial(): Promise<PilotTunnelBridge | null> {
const material = MeshCentralRegistry.getInstance().getActive();
if (!material) return null;
this.recentDials.push(Date.now());
const wsUrl = httpUrlToWs(material.centralApiUrl) + CALLBACK_PATH;
const ws = new WebSocket(wsUrl, {
headers: { Authorization: `Bearer ${material.callbackJwt}` },
handshakeTimeout: HANDSHAKE_TIMEOUT_MS,
maxPayload: MAX_FRAME_SIZE_BYTES,
});
try {
await this.awaitOpen(ws);
} catch (err) {
ws.on('error', () => { /* swallow tail */ });
try { ws.close(); } catch { /* ignore */ }
this.handleDialFailure(err, material.centralInstanceId);
return null;
}
return this.attachBridge(ws, material.centralInstanceId);
}
private async attachBridge(ws: WebSocket, instanceId: string): Promise<PilotTunnelBridge | null> {
const bridge = new PilotTunnelBridge(0, ws);
try {
await bridge.start();
} catch {
try { bridge.close(1011, 'bridge start failed'); } catch { /* ignore */ }
PilotMetrics.increment('mesh_callback_dials_failed_total');
return null;
}
bridge.once('closed', () => {
if (this.currentSession === bridge) this.currentSession = null;
});
this.currentSession = bridge;
PilotMetrics.increment('mesh_central_bootstraps_total');
MeshCentralRegistry.getInstance().markUsed(instanceId);
return bridge;
}
private awaitOpen(ws: WebSocket): Promise<void> {
return new Promise<void>((resolve, reject) => {
const cleanup = (): void => {
ws.removeAllListeners('open');
ws.removeAllListeners('error');
ws.removeAllListeners('unexpected-response');
};
ws.once('open', () => { cleanup(); resolve(); });
ws.once('error', (err) => { cleanup(); reject(err); });
ws.once('unexpected-response', (_req, res) => {
cleanup();
let body = '';
res.on('data', (chunk: Buffer) => { body += chunk.toString('utf8'); });
res.on('end', () => reject(this.buildUpgradeError(res.statusCode, body)));
});
});
}
private buildUpgradeError(statusCode: number | undefined, body: string): DialError {
let reason = 'unknown';
try {
const parsed = JSON.parse(body) as { reason?: unknown };
if (typeof parsed.reason === 'string') reason = parsed.reason;
} catch { /* keep unknown */ }
if (statusCode === 404) reason = 'endpoint_not_found';
const err = new Error(`upgrade rejected: HTTP ${statusCode ?? 'unknown'} reason=${reason}`) as DialError;
err.reason = reason;
err.httpStatus = statusCode;
return err;
}
private handleDialFailure(err: unknown, instanceId: string): void {
const reason = (err as DialError).reason ?? 'unknown';
PilotMetrics.increment('mesh_callback_dials_failed_total');
if (REJECT_REASONS_CLEAR.has(reason)) {
MeshCentralRegistry.getInstance().clearForInstance(instanceId);
PilotMetrics.increment('mesh_callback_auth_failures_total');
} else {
MeshCentralRegistry.getInstance().markRejected(instanceId, reason);
if (reason === 'endpoint_not_found') {
this.endpointUnavailableUntil = Date.now() + ENDPOINT_UNAVAILABLE_BACKOFF_MS;
}
}
console.warn(`[PeerToCentralMeshSessionDialer] dial failed: ${sanitizeForLog(`reason=${reason}`)}`);
}
}
+12
View File
@@ -22,6 +22,14 @@ interface Counters {
proxy_dials_failed: number;
/** Proxy-tunnel teardowns initiated by the dialer's idle sweep (zero active streams for the configured TTL). */
proxy_idle_closes: number;
/** Proxy bridges registered via the peer-initiated dial-back path (`/api/mesh/proxy-tunnel-from-peer`). Disjoint from `proxy_bridges_total`, which counts central-initiated dials. */
proxy_bridges_peer_initiated_total: number;
/** Successful peer-to-central dial-back sessions (counted on WS open + bridge start). Disjoint from `proxy_bridges_peer_initiated_total`, which is incremented by the central-side ingress at register time. Useful for operators investigating asymmetric counts (peer thinks it dialed, central never registered). */
mesh_central_bootstraps_total: number;
/** Peer-to-central dial-back attempts that did not complete a WS open. Covers transport errors, upgrade rejections, and bridge.start failures. */
mesh_callback_dials_failed_total: number;
/** Subset of `mesh_callback_dials_failed_total` where central responded 401 with a terminal reason code (stale, signature_invalid, ...) that caused the peer to clear its cached central material. */
mesh_callback_auth_failures_total: number;
}
class PilotMetricsImpl {
@@ -34,6 +42,10 @@ class PilotMetricsImpl {
proxy_bridges_total: 0,
proxy_dials_failed: 0,
proxy_idle_closes: 0,
proxy_bridges_peer_initiated_total: 0,
mesh_central_bootstraps_total: 0,
mesh_callback_dials_failed_total: 0,
mesh_callback_auth_failures_total: 0,
};
public increment<K extends keyof Counters>(name: K): void {
+35 -3
View File
@@ -149,6 +149,7 @@ export interface MeshTunnelHandle {
* stripping/injection, and license-tier propagation all work unchanged.
*/
export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle {
private readonly nodeId: number;
private readonly tunnelWs: WebSocket;
private readonly loopback: HttpServer;
private readonly wsUpgradeServer: WebSocketServer;
@@ -162,8 +163,9 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
private drainTimer?: NodeJS.Timeout;
private closed = false;
constructor(_nodeId: number, tunnelWs: WebSocket) {
constructor(nodeId: number, tunnelWs: WebSocket) {
super();
this.nodeId = nodeId;
this.tunnelWs = tunnelWs;
this.loopback = http.createServer();
this.wsUpgradeServer = new WebSocketServer({ noServer: true });
@@ -782,8 +784,26 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
private async acceptReverseLocal(s: number, target: { stack: string; service: string; port: number }): Promise<void> {
const { MeshService } = await import('./MeshService');
const ip = await MeshService.getInstance().resolveContainerIp({ stack: target.stack, service: target.service });
const meshSvc = MeshService.getInstance();
// Shared discriminator + target metadata so the Routing tab can
// separate peer-to-central (reverse) dispatch from central-to-peer
// (forward) dispatch when both flow through the same activity feed.
const baseDetails = {
direction: 'reverse' as const,
streamId: s,
targetStack: target.stack,
targetService: target.service,
targetPort: target.port,
peerNodeId: this.nodeId,
};
const ip = await meshSvc.resolveContainerIp({ stack: target.stack, service: target.service });
if (!ip) {
meshSvc.logActivity({
source: 'mesh', level: 'error', type: 'route.resolve.fail',
nodeId: this.nodeId,
message: `reverse dial failed: container ${target.stack}/${target.service} not found`,
details: { ...baseDetails, reason: 'container_not_found' },
});
this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'no_target' });
return;
}
@@ -802,14 +822,26 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle
// Pre-connect failure: ack-fail and drop. The handler is removed in
// 'connect' below so post-connect errors fall through to the
// mid-stream teardown path instead of double-firing.
const onPreConnectError = () => {
const onPreConnectError = (err?: Error) => {
if (!this.streams.has(s)) return;
this.streams.delete(s);
meshSvc.logActivity({
source: 'mesh', level: 'error', type: 'route.resolve.fail',
nodeId: this.nodeId,
message: `reverse dial failed pre-connect: ${err?.message ?? 'socket error'}`,
details: { ...baseDetails, reason: 'connect_error' },
});
this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' });
};
socket.once('error', onPreConnectError);
socket.once('connect', () => {
socket.off('error', onPreConnectError);
meshSvc.logActivity({
source: 'mesh', level: 'info', type: 'route.resolve.ok',
nodeId: this.nodeId,
message: `reverse dial ok: ${target.stack}/${target.service}:${target.port}`,
details: baseDetails,
});
this.sendJson({ t: 'tcp_open_ack', s, ok: true });
socket.on('data', (chunk: Buffer) => {
const cur = this.streams.get(s);
@@ -68,6 +68,15 @@ export class PilotTunnelCapacityError extends Error {
export class PilotTunnelManager extends EventEmitter {
private static instance: PilotTunnelManager;
private bridges: Map<number, PilotTunnelBridge> = new Map();
/**
* Parallel kind index for the `bridges` map. `'pilot'` is set by
* `registerTunnel` (agent-initiated long-lived tunnel); `'proxy'` is
* set by `registerProxyBridge` and `replaceOrRegisterProxyBridge`
* (central-initiated short-lived bridge). Used by
* `replaceOrRegisterProxyBridge` so a peer-initiated dial can supersede
* a previous proxy bridge but never shadow a live pilot tunnel.
*/
private bridgeKinds: Map<number, 'pilot' | 'proxy'> = new Map();
private softWarned = false;
private constructor() {
@@ -82,6 +91,31 @@ export class PilotTunnelManager extends EventEmitter {
return PilotTunnelManager.instance;
}
/**
* Test-only: drop the singleton and any held bridges so every test
* starts from a clean registry. Closes outstanding bridges best-effort.
*/
public static resetForTest(): void {
if (PilotTunnelManager.instance) {
for (const [, b] of PilotTunnelManager.instance.bridges) {
try { b.close(1000, 'test reset'); } catch { /* ignore */ }
}
PilotTunnelManager.instance.bridges.clear();
PilotTunnelManager.instance.bridgeKinds.clear();
}
PilotTunnelManager.instance = undefined as unknown as PilotTunnelManager;
}
/**
* Test-only: inject a pre-constructed bridge with an explicit kind.
* Bypasses capacity / lifecycle hooks so unit tests can prime the
* registry without owning a real WebSocket.
*/
public injectBridgeForTest(nodeId: number, bridge: PilotTunnelBridge, kind: 'pilot' | 'proxy'): void {
this.bridges.set(nodeId, bridge);
this.bridgeKinds.set(nodeId, kind);
}
/**
* Accept a newly handshaked pilot tunnel. Replaces any prior tunnel for the
* same node (split-brain prevention): the previous bridge is closed
@@ -95,6 +129,7 @@ export class PilotTunnelManager extends EventEmitter {
if (existing) {
existing.close(PilotCloseCode.Replaced, 'replaced by newer tunnel');
this.bridges.delete(nodeId);
this.bridgeKinds.delete(nodeId);
}
// Hard cap: only counts tunnels for *other* nodes since we just
@@ -120,6 +155,7 @@ export class PilotTunnelManager extends EventEmitter {
bridge.once('closed', () => {
if (this.bridges.get(nodeId) === bridge) {
this.bridges.delete(nodeId);
this.bridgeKinds.delete(nodeId);
DatabaseService.getInstance().updateNodeStatus(nodeId, 'offline');
this.emit('tunnel-down', nodeId);
}
@@ -127,6 +163,7 @@ export class PilotTunnelManager extends EventEmitter {
await bridge.start();
this.bridges.set(nodeId, bridge);
this.bridgeKinds.set(nodeId, 'pilot');
const db = DatabaseService.getInstance();
db.updateNodeStatus(nodeId, 'online');
db.updateNode(nodeId, {
@@ -233,14 +270,48 @@ export class PilotTunnelManager extends EventEmitter {
bridge.once('closed', () => {
if (this.bridges.get(nodeId) === bridge) {
this.bridges.delete(nodeId);
this.bridgeKinds.delete(nodeId);
this.emit('proxy-bridge-down', nodeId);
}
});
this.bridges.set(nodeId, bridge);
this.bridgeKinds.set(nodeId, 'proxy');
PilotMetrics.increment('proxy_bridges_total');
this.emit('proxy-bridge-up', nodeId);
}
/**
* Register a peer-initiated proxy bridge for an existing remote. If a proxy
* bridge already exists for this nodeId, close it (the new dial is the source
* of truth) and replace. If a pilot tunnel exists, refuse: pilot tunnels
* always win over peer-initiated proxy bridges.
*/
public replaceOrRegisterProxyBridge(nodeId: number, bridge: PilotTunnelBridge): void {
const existingKind = this.bridgeKinds.get(nodeId);
if (existingKind === 'pilot') {
throw new Error(`pilot tunnel already registered for node ${nodeId}; proxy bridge refused`);
}
if (existingKind === 'proxy') {
const old = this.bridges.get(nodeId);
this.bridges.delete(nodeId);
this.bridgeKinds.delete(nodeId);
try { old?.close(1000, 'replaced-by-newer-proxy'); } catch { /* best-effort cleanup */ }
}
if (this.bridges.size >= PILOT_TUNNEL_HARD_LIMIT) {
PilotMetrics.increment('tunnels_rejected_capacity');
throw new PilotTunnelCapacityError(PILOT_TUNNEL_HARD_LIMIT);
}
bridge.once('closed', () => {
if (this.bridges.get(nodeId) === bridge) {
this.bridges.delete(nodeId);
this.bridgeKinds.delete(nodeId);
this.emit('proxy-bridge-down', nodeId, 'remote_closed');
}
});
this.bridges.set(nodeId, bridge);
this.bridgeKinds.set(nodeId, 'proxy');
}
/**
* Force-close a tunnel (e.g., on node deletion).
*/
@@ -249,6 +320,7 @@ export class PilotTunnelManager extends EventEmitter {
if (!bridge) return;
bridge.close(code, reason);
this.bridges.delete(nodeId);
this.bridgeKinds.delete(nodeId);
}
}
+94 -4
View File
@@ -11,6 +11,39 @@ import {
import { sanitizeForLog } from '../utils/safeLog';
import { isDebugEnabled } from '../utils/debug';
import { rejectUpgrade as reject } from './reject';
import { MeshCentralRegistry, type MeshCentralMaterial } from '../services/MeshCentralRegistry';
/**
* Bootstrap phase for the first-frame state machine.
*
* Central MAY send a `mesh_handshake` JSON frame as the FIRST text frame
* after upgrade. If it arrives we persist the callback material via
* MeshCentralRegistry and transition to `consumed`. Any frame other than
* the handshake (or no frame at all) moves us to `past`, after which a
* later handshake is a protocol error.
*/
type BootstrapPhase = 'awaiting' | 'consumed' | 'past';
interface MeshHandshakeFrame {
t: 'mesh_handshake';
v: number;
peerNodeId: number;
centralInstanceId: string;
centralApiUrl: string;
meshTunnelJwt: string;
jwtExpiresAt: number;
}
function isMeshHandshakeFrame(parsed: unknown): parsed is MeshHandshakeFrame {
return typeof parsed === 'object' && parsed !== null
&& (parsed as { t?: unknown }).t === 'mesh_handshake'
&& typeof (parsed as { v?: unknown }).v === 'number'
&& typeof (parsed as { peerNodeId?: unknown }).peerNodeId === 'number'
&& typeof (parsed as { centralInstanceId?: unknown }).centralInstanceId === 'string'
&& typeof (parsed as { centralApiUrl?: unknown }).centralApiUrl === 'string'
&& typeof (parsed as { meshTunnelJwt?: unknown }).meshTunnelJwt === 'string'
&& typeof (parsed as { jwtExpiresAt?: unknown }).jwtExpiresAt === 'number';
}
/**
* Mesh proxy-tunnel ingress.
@@ -136,18 +169,75 @@ async function attachSwitchboard(ws: WebSocket, peerNodeId: number | null): Prom
return;
}
let phase: BootstrapPhase = 'awaiting';
const onMessage = (data: unknown, isBinary: boolean): void => {
if (!switchboard) return;
try {
if (isBinary) {
if (phase === 'awaiting') phase = 'past';
const buf = wsDataToBuffer(data);
if (!buf) return;
switchboard.handleBinaryFrame(decodeBinaryFrame(buf));
} else {
const text = wsDataToString(data);
if (text == null) return;
switchboard.handleJsonFrame(decodeJsonFrame(text));
return;
}
const text = wsDataToString(data);
if (text == null) {
if (phase === 'awaiting') phase = 'past';
return;
}
let parsedForBootstrap: unknown = null;
let parseSucceeded = false;
try {
parsedForBootstrap = JSON.parse(text);
parseSucceeded = true;
} catch {
// Not valid JSON; defer to decodeJsonFrame's stricter error path below.
}
if (parseSucceeded && typeof parsedForBootstrap === 'object'
&& parsedForBootstrap !== null
&& (parsedForBootstrap as { t?: unknown }).t === 'mesh_handshake') {
if (phase !== 'awaiting') {
ws.close(1008, 'mesh_handshake out of order');
return;
}
if (!isMeshHandshakeFrame(parsedForBootstrap)) {
ws.close(1008, 'malformed mesh_handshake');
return;
}
const material: MeshCentralMaterial = {
centralInstanceId: parsedForBootstrap.centralInstanceId,
centralApiUrl: parsedForBootstrap.centralApiUrl.replace(/\/+$/, ''),
callbackJwt: parsedForBootstrap.meshTunnelJwt,
jwtIssuedAt: Math.floor(Date.now() / 1000),
jwtExpiresAt: parsedForBootstrap.jwtExpiresAt,
};
try {
MeshCentralRegistry.getInstance().upsert(material);
phase = 'consumed';
const centralInstanceId = parsedForBootstrap.centralInstanceId;
const jwtExpiresAt = parsedForBootstrap.jwtExpiresAt;
void (async () => {
try {
const { MeshService: MeshSvc } = await import('../services/MeshService');
MeshSvc.getInstance().logActivity({
source: 'mesh', level: 'info',
type: 'mesh_handshake.received',
message: `mesh callback bootstrap received from central ${centralInstanceId}`,
details: { centralInstanceId, jwtExpiresAt },
});
} catch { /* best-effort log; bootstrap success path must not depend on it */ }
})();
} catch (err) {
console.warn(`[meshProxyTunnel] mesh_handshake persist failed: ${sanitizeForLog((err as Error).message)}`);
try { ws.close(1011, 'bootstrap persist failed'); } catch { /* ignore */ }
}
return;
}
if (phase === 'awaiting') phase = 'past';
switchboard.handleJsonFrame(decodeJsonFrame(text));
} catch (err) {
if (isDebugEnabled()) {
console.warn('[MeshProxy:diag] malformed frame:', sanitizeForLog((err as Error).message));
@@ -0,0 +1,142 @@
/**
* Central-side ingress for peer-initiated proxy-mode mesh tunnels.
*
* After the bootstrap exchange in `meshProxyTunnel.ts` (Task 7/8), a peer
* with the central's `mesh_tunnel` JWT can dial *back* to central at this
* endpoint. The validation chain enforces every claim the bootstrap mint
* produced and confirms the peer's stored fingerprint still matches its
* current api_token. Each failure path returns HTTP 401 with a stable
* machine-readable `reason` code in the JSON body, so the dialer side
* (PeerToCentralMeshSessionDialer, Task 10) can act on the reason without
* scraping prose.
*
* Auth runs MANUALLY in this handler. Express middleware does not run on
* WebSocket upgrades, and the credential here is a Bearer JWT minted by
* the central itself, not a user session.
*/
import type { IncomingMessage } from 'http';
import type { Duplex } from 'stream';
import jwt, { type Algorithm, type JwtPayload } from 'jsonwebtoken';
import { createHash } from 'crypto';
import { WebSocketServer } from 'ws';
import { DatabaseService } from '../services/DatabaseService';
import { PilotTunnelBridge } from '../services/PilotTunnelBridge';
import { PilotTunnelManager } from '../services/PilotTunnelManager';
import { PilotMetrics } from '../services/PilotMetrics';
import { sanitizeForLog } from '../utils/safeLog';
const wss = new WebSocketServer({ noServer: true });
const CLOCK_SKEW_SEC = 60;
type RejectReason =
| 'algorithm_mismatch' | 'signature_invalid'
| 'scope_mismatch' | 'audience_mismatch' | 'instance_mismatch'
| 'stale' | 'clock_skew' | 'node_deleted' | 'mode_mismatch'
| 'token_fingerprint_mismatch' | 'malformed';
function rejectUpgrade(socket: Duplex, reason: RejectReason): void {
const body = JSON.stringify({ reason });
const headers = [
'HTTP/1.1 401 Unauthorized',
'Content-Type: application/json',
`Content-Length: ${Buffer.byteLength(body)}`,
'Connection: close',
'',
body,
].join('\r\n');
try { socket.write(headers); } catch { /* socket already gone */ }
try { socket.destroy(); } catch { /* socket already gone */ }
}
function extractBearer(req: IncomingMessage): string | null {
const h = req.headers['authorization'];
if (typeof h !== 'string' || !h.startsWith('Bearer ')) return null;
return h.slice('Bearer '.length).trim() || null;
}
export function handleMeshProxyTunnelFromPeerUpgrade(
req: IncomingMessage,
socket: Duplex,
head: Buffer,
): void {
const token = extractBearer(req);
if (!token) { rejectUpgrade(socket, 'malformed'); return; }
let header: { alg?: string; kid?: string } = {};
try {
const decoded = jwt.decode(token, { complete: true });
header = (decoded?.header ?? {}) as typeof header;
} catch { rejectUpgrade(socket, 'malformed'); return; }
if (header.alg !== 'HS256') { rejectUpgrade(socket, 'algorithm_mismatch'); return; }
const settings = DatabaseService.getInstance().getGlobalSettings();
const secret = settings.auth_jwt_secret;
if (!secret) { rejectUpgrade(socket, 'malformed'); return; }
let payload: JwtPayload;
try {
payload = jwt.verify(token, secret, { algorithms: ['HS256' as Algorithm] }) as JwtPayload;
} catch { rejectUpgrade(socket, 'signature_invalid'); return; }
if (payload.scope !== 'mesh_tunnel') { rejectUpgrade(socket, 'scope_mismatch'); return; }
// SENCHO_PRIMARY_URL must be set on central for peer-initiated dial-back
// to operate. The Task 8 bootstrap-mint side already skips when it's
// unset; this is the matching fail-safe on the verify side. Reject all
// peer dials with audience_mismatch when central has no canonical
// origin to validate against.
const canonicalOrigin = (process.env.SENCHO_PRIMARY_URL ?? '').replace(/\/+$/, '');
if (!canonicalOrigin) { rejectUpgrade(socket, 'audience_mismatch'); return; }
if (payload.aud !== canonicalOrigin) { rejectUpgrade(socket, 'audience_mismatch'); return; }
// instance_id is operational state, not user-defined config, so it
// lives in system_state rather than global_settings.
const instanceId = DatabaseService.getInstance().getSystemState('instance_id');
if (payload.iss !== instanceId) { rejectUpgrade(socket, 'instance_mismatch'); return; }
const nowSec = Math.floor(Date.now() / 1000);
if (typeof payload.exp !== 'number' || payload.exp <= nowSec) {
rejectUpgrade(socket, 'stale'); return;
}
if (typeof payload.iat !== 'number' || payload.iat > nowSec + CLOCK_SKEW_SEC) {
rejectUpgrade(socket, 'clock_skew'); return;
}
const peerNodeId = Number(payload.sub);
if (!Number.isInteger(peerNodeId) || peerNodeId <= 0) {
rejectUpgrade(socket, 'malformed'); return;
}
const node = DatabaseService.getInstance().getNode(peerNodeId);
if (!node) { rejectUpgrade(socket, 'node_deleted'); return; }
if (node.type !== 'remote' || node.mode !== 'proxy') {
rejectUpgrade(socket, 'mode_mismatch'); return;
}
if (!node.api_token) { rejectUpgrade(socket, 'token_fingerprint_mismatch'); return; }
const expectedFp = createHash('sha256').update(node.api_token).digest('hex').slice(0, 16);
const claimedFp = payload['peer_token_fp'];
if (typeof claimedFp !== 'string' || claimedFp !== expectedFp) {
rejectUpgrade(socket, 'token_fingerprint_mismatch'); return;
}
wss.handleUpgrade(req, socket, head, (ws) => {
try {
const bridge = new PilotTunnelBridge(peerNodeId, ws);
bridge.start().then(() => {
try {
PilotTunnelManager.getInstance().replaceOrRegisterProxyBridge(peerNodeId, bridge);
PilotMetrics.increment('proxy_bridges_peer_initiated_total');
} catch (err) {
try { bridge.close(1013, 'manager rejected'); } catch { /* ignore */ }
console.warn(`[meshProxyTunnelFromPeer] register failed: ${sanitizeForLog((err as Error).message)}`);
}
}).catch((err) => {
try { bridge.close(1011, 'bridge start failed'); } catch { /* ignore */ }
console.warn(`[meshProxyTunnelFromPeer] bridge start failed: ${sanitizeForLog((err as Error).message)}`);
});
} catch (err) {
console.warn(`[meshProxyTunnelFromPeer] bridge construct failed: ${sanitizeForLog((err as Error).message)}`);
try { ws.close(1011, 'bridge init failed'); } catch { /* ignore */ }
}
});
}
+20 -9
View File
@@ -8,6 +8,7 @@ import { NodeRegistry } from '../services/NodeRegistry';
import { COOKIE_NAME } from '../helpers/constants';
import { handlePilotTunnel } from './pilotTunnel';
import { handleMeshProxyTunnel } from './meshProxyTunnel';
import { handleMeshProxyTunnelFromPeerUpgrade } from './meshProxyTunnelFromPeer';
import { handleNotificationsWs } from './notifications';
import { handleRemoteForwarder } from './remoteForwarder';
import { handleLogsWs } from './logs';
@@ -31,15 +32,16 @@ function parseCookies(req: IncomingMessage): Record<string, string> {
* `connection` handler on the main wss.
*
* Dispatch order (first match wins):
* 1. `/api/pilot/tunnel` -> handlePilotTunnel (own auth, own wss)
* 2. shared cookie/Bearer auth + JWT verify (rejects unauthenticated)
* 3. API token scope gate (read-only / deploy-only restricted to logs + notifications)
* 4. `/api/mesh/proxy-tunnel` -> handleMeshProxyTunnel (machine-to-machine: node_proxy or full-admin api_token)
* 5. `/ws/notifications` local -> handleNotificationsWs
* 6. remote nodeId path -> handleRemoteForwarder
* 7. `/api/stacks/:name/logs` -> handleLogsWs
* 8. `/api/system/host-console` -> handleHostConsoleWs
* 9. fallback -> handleGenericWs (`/ws` exec + stats)
* 1. `/api/pilot/tunnel` -> handlePilotTunnel (own auth, own wss)
* 2. `/api/mesh/proxy-tunnel-from-peer` -> handleMeshProxyTunnelFromPeerUpgrade (own JWT chain, central-side dial-back)
* 3. shared cookie/Bearer auth + JWT verify (rejects unauthenticated)
* 4. API token scope gate (read-only / deploy-only restricted to logs + notifications)
* 5. `/api/mesh/proxy-tunnel` -> handleMeshProxyTunnel (machine-to-machine: node_proxy or full-admin api_token)
* 6. `/ws/notifications` local -> handleNotificationsWs
* 7. remote nodeId path -> handleRemoteForwarder
* 8. `/api/stacks/:name/logs` -> handleLogsWs
* 9. `/api/system/host-console` -> handleHostConsoleWs
* 10. fallback -> handleGenericWs (`/ws` exec + stats)
*/
export function attachUpgrade(
server: http.Server,
@@ -51,12 +53,21 @@ export function attachUpgrade(
server.on('upgrade', async (req, socket, head) => {
// Pilot-agent tunnel ingress: machine credentials, no cookies.
// Mesh proxy-tunnel-from-peer ingress: peer-initiated dial-back over
// a `mesh_tunnel`-scoped JWT (minted earlier during the bootstrap
// exchange). Both handlers run their own auth before the shared
// cookie/Bearer pipeline because their credentials are not user
// sessions and would fail the shared user-existence check.
try {
const reqUrl = new URL(req.url || '/', `http://${req.headers.host || 'localhost'}`);
if (reqUrl.pathname === '/api/pilot/tunnel') {
await handlePilotTunnel(req, socket, head, pilotTunnelWss);
return;
}
if (reqUrl.pathname === '/api/mesh/proxy-tunnel-from-peer') {
handleMeshProxyTunnelFromPeerUpgrade(req, socket, head);
return;
}
} catch {
// URL parse error falls through and will be rejected below.
}