From a38a3e022696019e6aa529f10fbffc27b1347c53 Mon Sep 17 00:00:00 2001 From: Anso Date: Thu, 14 May 2026 20:51:42 -0400 Subject: [PATCH] feat(mesh): route mesh traffic over Distributed API remotes (#1048) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(mesh): extract shared TCP stream switchboard Pull the `tcp_open` / `tcp_open_ack` / `tcp_open_reverse` / `tcp_close` + `TcpData` handling out of the pilot agent into a reusable module so a second caller (the upcoming proxy-mode WS handler) can run the same frame parser, stream allocator, idle timers, and Compose-label resolver. One parser, two callers, zero drift on a security-sensitive protocol surface. The pilot agent keeps its public `openMeshTcpStream` API and delegates to a per-connection switchboard constructed in `connect()` and torn down in `cleanupAfterDisconnect`. Existing reverse-stream and resolver tests are rewritten to exercise the shared module directly. * feat(mesh): route mesh traffic over Distributed API remotes Sencho Mesh now works against remotes in Distributed API mode in addition to Pilot Agent mode. Central opens a short-lived WebSocket tunnel to the remote's new `/api/mesh/proxy-tunnel` endpoint on demand and tears it down after 5 minutes of idle (configurable via `SENCHO_MESH_PROXY_TUNNEL_IDLE_MS`). Mesh dispatch is mode-agnostic at the routing layer; `PilotTunnelManager.ensureBridge` resolves an existing tunnel or asks the new `MeshProxyTunnelDialer` to dial. The Routing tab badges now use a `reachableMode` classifier: `★ Local` for local, `pilot offline` red badge for a pilot with a down tunnel, and a new `unreachable` red badge surfacing the specific reason (missing token, scope not full-admin, TLS failure, remote does not support proxy mesh). Distributed API remotes with valid credentials show no negative badge; the tunnel opens on first dial. Auth: the proxy-tunnel WS upgrade requires an `Authorization: Bearer` API token with the `full-admin` scope. Lower-scoped tokens are rejected at upgrade time so a leaked read-only token cannot reach the mesh data plane. Bidirectional: the proxy-mode WS handler registers itself as the local `MeshService` reverse dialer (compare-and-swap), so meshed containers on a Distributed API remote can dial cross-node aliases via `tcp_open_reverse` over the same tunnel. Cross-node relays (`PilotTunnelBridge.acceptReverseRelay`) await `ensureBridge` so proxy-to-proxy mesh works without standing tunnels. * docs(mesh): describe Distributed API mesh and unreachable troubleshooting Refresh the user-facing mesh documentation to reflect that mesh works over both Pilot Agent and Distributed API remotes. Adds a Troubleshooting accordion entry for the new `unreachable` Routing tab badge so operators can match a tooltip reason ("api token rejected (scope must be full-admin)", "remote does not support proxy mesh", "TLS handshake failed", "api_url not set", "api token missing") to a concrete fix. * refactor(mesh): apply /simplify review findings Five cleanups surfaced by the post-implementation code review pass. No behaviour change for fleets running on `main`; only internal structure improves. - **Deterministic container IP shared.** Extract the compose-default network preference (`pickContainerIp`) and the conventional-name fast path (`lookupContainerIp`) into `backend/src/mesh/containerLookup.ts`. Both `MeshService.resolveContainerIp` (existing same-node fast path) and the switchboard's `resolveByComposeLabels` (proxy/pilot inbound dial) now use the same logic. Earlier the switchboard helper grabbed the first `Object.values(Networks)` IP, which could flip across daemon versions on multi-network containers; fixed. - **WS URL upgrade extracted.** New `backend/src/utils/wsUrl.ts` exposes `httpUrlToWs(baseUrl)` that maps `http://` to `ws://` and `https://` to `wss://`. Used in both `pilot/agent.ts` and the proxy-tunnel dialer. The previous inline `replace(/^http/, 'ws')` silently downgraded `https://` to `ws://` (cleartext). - **`computeReachable` takes the pre-fetched node.** `getStatus` already iterated `db.getNodes()`; the helper previously re-queried by id per node (N+1 reads). Pass the row in. - **Reachable-reason ternary -> const map.** Replace the nested ternary in `MeshService.computeReachable` with a `Record` lookup keyed on the dialer's failure code. - **Activity-type ternary -> const map.** Same flattening inside `MeshProxyTunnelDialer.logActivity`. All mesh tests pass (85/85). Full backend suite green (2126/2126). * fix(mesh): cache proxy dial failures and contain WS handshake errors `ensureBridge` now consults the recent-failure cache before dialing so a continuous mesh workload against a misconfigured proxy-mode remote does not produce one upgrade attempt per cross-node TCP open. `recordFailure` emits at most one activity-log entry per cache window per (nodeId, code) so a connect-loop on a single bad remote cannot flush the ring buffer. Failure messages run through `redactSensitiveText` before reaching the log so any embedded Bearer / JWT / inline-URL credentials are scrubbed. `stop()` clears the inflight map and `dial()` checks `this.stopped` between awaits so a shutdown does not leak a half-opened bridge. When `awaitOpen` rejects (401, 4403, TLS), the dialer attaches a noop 'error' listener before calling `ws.close()`. Without it the ws library emits a tail 'error' on a still-CONNECTING socket that propagates as an unhandled exception. Surfaced by a live-network test against a real proxy-mode peer. Adds `mesh-proxy-tunnel-live.test.ts` (skipped unless MESH_AUDIT_URL and MESH_AUDIT_TOKEN_FILE env vars are set) covering both the happy path and the auth-rejected path against an actual remote Sencho. * feat(mesh): instrument proxy tunnel observability Adds three counters to `PilotMetrics`: - `proxy_bridges_total` (incremented on `registerProxyBridge` success) - `proxy_dials_failed` (every failed dial attempt, not deduped) - `proxy_idle_closes` (idle-sweep teardowns) All three surface automatically via `GET /api/system/pilot-tunnels` since the route returns the full `Counters` snapshot. Gates two pre-existing always-on `console.warn` calls in `tcpStreamSwitchboard.ts` (mid-stream socket errors and Docker resolve failures) behind `isDebugEnabled()`. Both fire on per-stream events and would otherwise violate the diagnostic-log safety rule under load. Adds `mesh-tcp-stream-switchboard.test.ts` covering the forward `tcp_open` path: real localhost dial, resolver errors (no_target, denied), per-tunnel cap saturation, frame-routing fall-through invariants, `tcp_close` socket teardown. Drops `MESH_CONNECT_TIMEOUT_MS` from the public surface; only the switchboard itself uses it. * fix(mesh): tighten Routing tab unreachable handling `RoutingNodeCard.tsx`: the **Add stack to mesh** button now disables when `reachableMode === 'unreachable'`, matching the existing TogglePill behavior. Without this, an operator on an unreachable node could open the opt-in sheet, confirm, and watch the redeploy proceed against a target whose mesh data plane will silently fail to route. Also drops the leftover `status.nodeId !== -1` guard on the pilot- offline badge. `MeshService.ts`: renames `isMeshReachable` to `isMeshConfigured` with the new predicate that returns true for proxy-mode remotes whose creds are valid (the tunnel is opened on demand). `getRouteDiagnostic` now distinguishes "routable" (configured) from "pilotLive" (live tunnel state, only meaningful for pilot mode); without the split, every idle proxy-mode route would report `tunnel down`. `MeshService.setReverseDialer` warns when the unconditional install path silently overwrites a non-null current dialer. By topology a Sencho is either pilot or central, so the branch flags a misconfigured deployment rather than an expected race. Drops the dead `export { ReverseTcpStreamHandle }` re-export from `pilot/agent.ts`; its only consumer imports straight from `mesh/tcpStreamSwitchboard.ts`. Fixes a stale doc comment in `MeshService.ts` that referenced the wrong source file. Adds `mesh-proxy-tunnel-handler.test.ts` covering the WS handler lifecycle: pilot-mode 404 rejection, post-upgrade reverse-dialer install, concurrent-upgrade 1013 rejection (single-tenant slot), and error-path teardown. Updates the troubleshooting accordion in the user docs to mention the disabled-state behavior. * fix(mesh): close ESLint and CodeQL findings on the proxy-tunnel diag logs Remove the unused `reverseDialer` local in `meshProxyTunnel.ts`; the closure target is `localDialer` above and this assignment was always dead. Strip line breaks inline on the three `MeshProxyTunnelDialer` diag log lines so CodeQL `js/log-injection` data flow recognises the sanitisation that `sanitizeForLog` already performs. No behaviour change; the diag logs render the same characters they do today. --- .env.example | 7 + .../__tests__/mesh-proxy-tunnel-dial.test.ts | 101 +++++ .../mesh-proxy-tunnel-handler.test.ts | 168 +++++++ .../__tests__/mesh-proxy-tunnel-live.test.ts | 105 +++++ .../mesh-tcp-stream-switchboard.test.ts | 210 +++++++++ .../pilot-agent-mesh-resolve.test.ts | 125 +++--- .../pilot-agent-reverse-stream.test.ts | 138 +++--- .../pilot-tunnel-manager-ensure.test.ts | 112 +++++ backend/src/mesh/containerLookup.ts | 84 ++++ backend/src/mesh/tcpStreamSwitchboard.ts | 415 ++++++++++++++++++ backend/src/pilot/agent.ts | 352 ++------------- backend/src/services/MeshProxyTunnelDialer.ts | 371 ++++++++++++++++ backend/src/services/MeshService.ts | 223 ++++++---- backend/src/services/PilotMetrics.ts | 9 + backend/src/services/PilotTunnelBridge.ts | 7 +- backend/src/services/PilotTunnelManager.ts | 89 +++- backend/src/utils/wsUrl.ts | 15 + backend/src/websocket/meshProxyTunnel.ts | 137 ++++++ backend/src/websocket/upgradeHandler.ts | 24 +- docs/features/sencho-mesh.mdx | 28 +- frontend/src/components/NodeManager.tsx | 2 +- .../src/components/fleet/RoutingNodeCard.tsx | 32 +- frontend/src/components/fleet/RoutingTab.tsx | 23 +- frontend/src/types/mesh.ts | 9 +- 24 files changed, 2225 insertions(+), 561 deletions(-) create mode 100644 backend/src/__tests__/mesh-proxy-tunnel-dial.test.ts create mode 100644 backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts create mode 100644 backend/src/__tests__/mesh-proxy-tunnel-live.test.ts create mode 100644 backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts create mode 100644 backend/src/__tests__/pilot-tunnel-manager-ensure.test.ts create mode 100644 backend/src/mesh/containerLookup.ts create mode 100644 backend/src/mesh/tcpStreamSwitchboard.ts create mode 100644 backend/src/services/MeshProxyTunnelDialer.ts create mode 100644 backend/src/utils/wsUrl.ts create mode 100644 backend/src/websocket/meshProxyTunnel.ts diff --git a/.env.example b/.env.example index 1af92bf6..e6770041 100644 --- a/.env.example +++ b/.env.example @@ -111,3 +111,10 @@ SSO_CALLBACK_URL= # sync tab, Fleet Actions tab). Backend routes for these surfaces stay # live regardless. This flag controls UI discovery only. SENCHO_EXPERIMENTAL=false + +# Idle teardown (in ms) for on-demand mesh tunnels that central opens +# to Distributed API (proxy-mode) remotes. After this many milliseconds +# of zero active mesh streams, central closes the WebSocket; the next +# mesh dial re-opens it. Defaults to 5 minutes. Set to 0 to keep +# tunnels open until the remote drops them or central shuts down. +SENCHO_MESH_PROXY_TUNNEL_IDLE_MS=300000 diff --git a/backend/src/__tests__/mesh-proxy-tunnel-dial.test.ts b/backend/src/__tests__/mesh-proxy-tunnel-dial.test.ts new file mode 100644 index 00000000..0a430324 --- /dev/null +++ b/backend/src/__tests__/mesh-proxy-tunnel-dial.test.ts @@ -0,0 +1,101 @@ +/** + * MeshProxyTunnelDialer: central-side dialer that opens an on-demand + * `/api/mesh/proxy-tunnel` WebSocket to a Distributed API remote. + * + * These tests cover the deterministic surface: failure caching when no + * proxy target is configured, idle-close behavior, and recent-failure + * lookup. End-to-end TLS / handshake paths are covered by the manual + * production verification recipe rather than by network-bound tests. + */ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; + +let tmpDir: string; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; +let MeshProxyTunnelDialer: typeof import('../services/MeshProxyTunnelDialer').MeshProxyTunnelDialer; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService } = await import('../services/DatabaseService')); + ({ MeshProxyTunnelDialer } = await import('../services/MeshProxyTunnelDialer')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +beforeEach(() => { + DatabaseService.getInstance().getDb().prepare("DELETE FROM nodes WHERE name LIKE 'proxy-test-%'").run(); +}); + +describe('MeshProxyTunnelDialer', () => { + it('returns null and caches a no_target failure when getProxyTarget yields no node', async () => { + const dialer = MeshProxyTunnelDialer.resetForTest(0); // idle-close disabled + const result = await dialer.ensureBridge(9999); // unknown nodeId + expect(result).toBeNull(); + const failure = dialer.getRecentFailure(9999); + expect(failure?.code).toBe('no_target'); + }); + + it('returns null and caches no_target when the node has no api_token', async () => { + const dialer = MeshProxyTunnelDialer.resetForTest(0); + const db = DatabaseService.getInstance(); + const nodeId = db.addNode({ + name: 'proxy-test-no-token', + type: 'remote', + compose_dir: '', + is_default: false, + mode: 'proxy', + api_url: 'https://proxy-test.invalid', + api_token: '', + }); + + const result = await dialer.ensureBridge(nodeId); + expect(result).toBeNull(); + const failure = dialer.getRecentFailure(nodeId); + expect(failure?.code).toBe('no_target'); + }); + + it('hasBridge reports false before the first successful dial', async () => { + const dialer = MeshProxyTunnelDialer.resetForTest(0); + const db = DatabaseService.getInstance(); + const nodeId = db.addNode({ + name: 'proxy-test-hasbridge', + type: 'remote', + compose_dir: '', + is_default: false, + mode: 'proxy', + api_url: '', + api_token: '', + }); + + expect(dialer.hasBridge(nodeId)).toBe(false); + await dialer.ensureBridge(nodeId); + expect(dialer.hasBridge(nodeId)).toBe(false); + }); + + it('expires the recent-failure cache entry after the cache TTL window', async () => { + const dialer = MeshProxyTunnelDialer.resetForTest(0); + const result = await dialer.ensureBridge(8888); + expect(result).toBeNull(); + const cached = dialer.getRecentFailure(8888); + expect(cached).not.toBeNull(); + type FailureMap = Map; + const cacheRef = (dialer as unknown as { recentFailures: FailureMap }).recentFailures; + const entry = cacheRef.get(8888); + if (!entry) throw new Error('cache entry missing'); + entry.ts = Date.now() - 90_000; // 90s old, well past the 60s TTL + expect(dialer.getRecentFailure(8888)).toBeNull(); + expect(cacheRef.has(8888)).toBe(false); + }); + + it('stop() tears down the singleton and the idle-check timer', () => { + const dialer = MeshProxyTunnelDialer.resetForTest(60_000); + const timerRef = (dialer as unknown as { idleCheckTimer: NodeJS.Timeout | null }).idleCheckTimer; + expect(timerRef).not.toBeNull(); + + dialer.stop(); + const timerAfter = (dialer as unknown as { idleCheckTimer: NodeJS.Timeout | null }).idleCheckTimer; + expect(timerAfter).toBeNull(); + }); +}); diff --git a/backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts b/backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts new file mode 100644 index 00000000..460a4680 --- /dev/null +++ b/backend/src/__tests__/mesh-proxy-tunnel-handler.test.ts @@ -0,0 +1,168 @@ +/** + * `meshProxyTunnel.ts` is the server-side ingress for Distributed API + * (proxy-mode) mesh tunnels. Auth + scope gating live one layer up in + * `upgradeHandler.ts`; this file covers the handler's own invariants: + * + * - It refuses to serve in pilot-mode deployments (returns 404). + * - On a successful upgrade it installs the shared switchboard AND + * CAS-installs itself as the local MeshService reverse dialer. + * - On WS close the reverse dialer slot is cleared so a subsequent + * dialer can install. + * - A concurrent second upgrade is rejected (1013) while the slot is + * held — this prevents two central instances from racing for the + * same remote's reverse-dial slot. + */ +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; + +interface ServerHandle { + server: http.Server; + port: number; + close: () => Promise; +} + +/** + * Stand up a bare http.Server that routes `/api/mesh/proxy-tunnel` + * upgrades through the handler. Skips the `upgradeHandler.ts` auth + * pipeline because that path is exercised by other integration tests; + * here the contract being verified is what the handler does AFTER + * auth has already accepted the upgrade. + */ +async function startServer(): Promise { + const server = http.createServer(); + server.on('upgrade', (req, socket, head) => { + if (req.url === '/api/mesh/proxy-tunnel') { + void handleMeshProxyTunnel(req, socket, head); + } else { + socket.destroy(); + } + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', () => resolve())); + const port = (server.address() as AddressInfo).port; + return { + server, + port, + close: () => new Promise((resolve) => server.close(() => resolve())), + }; +} + +function dialTunnel(port: number): Promise { + return new Promise((resolve, reject) => { + const ws = new WebSocket(`ws://127.0.0.1:${port}/api/mesh/proxy-tunnel`); + ws.once('open', () => resolve(ws)); + ws.once('error', reject); + }); +} + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ handleMeshProxyTunnel } = await import('../websocket/meshProxyTunnel')); + ({ MeshService } = await import('../services/MeshService')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +beforeEach(() => { + delete process.env.SENCHO_MODE; + // Defensive: clear any reverse dialer left over from a prior test. + MeshService.getInstance().setReverseDialer(null); +}); + +describe('handleMeshProxyTunnel', () => { + it('rejects with 404 when SENCHO_MODE=pilot (pilot-mode Sencho receives mesh via the pilot tunnel only)', async () => { + process.env.SENCHO_MODE = 'pilot'; + const srv = await startServer(); + try { + await new Promise((resolve, reject) => { + const ws = new WebSocket(`ws://127.0.0.1:${srv.port}/api/mesh/proxy-tunnel`); + ws.once('open', () => reject(new Error('upgrade should not have succeeded'))); + ws.once('unexpected-response', (_req, res) => { + expect(res.statusCode).toBe(404); + res.resume(); + resolve(); + }); + ws.once('error', () => { /* expected */ }); + }); + } finally { + await srv.close(); + } + }); + + it('on successful upgrade, installs a reverse dialer that exposes openMeshTcpStream', async () => { + const srv = await startServer(); + try { + const ws = await dialTunnel(srv.port); + // setReverseDialer is the load-bearing post-upgrade side effect; the + // mesh service's dispatch path consults it on every cross-node TCP + // open. The exact dialer instance is internal, but the presence of + // `openMeshTcpStream` is the contract MeshService.dialMeshTcpStream + // depends on. + // Allow the handler's microtasks to install the dialer. + await new Promise((r) => setTimeout(r, 20)); + const dialer = (MeshService.getInstance() as unknown as { reverseDialer: { openMeshTcpStream?: unknown } | null }).reverseDialer; + expect(dialer).not.toBeNull(); + expect(typeof dialer?.openMeshTcpStream).toBe('function'); + + ws.close(1000, 'test cleanup'); + // Wait for the close handler's CAS-uninstall. + await new Promise((r) => setTimeout(r, 30)); + const after = (MeshService.getInstance() as unknown as { reverseDialer: unknown }).reverseDialer; + expect(after).toBeNull(); + } finally { + await srv.close(); + } + }); + + it('a concurrent second upgrade is closed (the reverse dialer slot is single-tenant)', async () => { + const srv = await startServer(); + try { + const wsA = await dialTunnel(srv.port); + await new Promise((r) => setTimeout(r, 20)); + + // The second upgrade succeeds at the WS layer (the handler + // accepted the upgrade before discovering the slot was taken) + // but the handler then closes the socket with code 1013. + const wsB = new WebSocket(`ws://127.0.0.1:${srv.port}/api/mesh/proxy-tunnel`); + const closeInfo = await new Promise<{ code: number; reason: string }>((resolve, reject) => { + wsB.once('close', (code, reason) => resolve({ code, reason: reason.toString() })); + wsB.once('error', reject); + }); + expect(closeInfo.code).toBe(1013); + + // The first tunnel's dialer is still installed. + const dialer = (MeshService.getInstance() as unknown as { reverseDialer: unknown }).reverseDialer; + expect(dialer).not.toBeNull(); + + wsA.close(1000, 'test cleanup'); + await new Promise((r) => setTimeout(r, 30)); + } finally { + await srv.close(); + } + }); + + it('on WS error the reverse dialer slot is cleared (defensive: matches close-path teardown)', async () => { + const srv = await startServer(); + try { + const ws = await dialTunnel(srv.port); + await new Promise((r) => setTimeout(r, 20)); + + // Force-terminate without a clean close; the handler's `error` + // listener (line 133-138) should still run teardown. + ws.terminate(); + await new Promise((r) => setTimeout(r, 30)); + const after = (MeshService.getInstance() as unknown as { reverseDialer: unknown }).reverseDialer; + expect(after).toBeNull(); + } finally { + await srv.close(); + } + }); +}); diff --git a/backend/src/__tests__/mesh-proxy-tunnel-live.test.ts b/backend/src/__tests__/mesh-proxy-tunnel-live.test.ts new file mode 100644 index 00000000..a140195a --- /dev/null +++ b/backend/src/__tests__/mesh-proxy-tunnel-live.test.ts @@ -0,0 +1,105 @@ +/** + * Live-network test for the proxy-mode mesh tunnel against a real remote + * Sencho. SKIPPED unless `MESH_AUDIT_URL` and `MESH_AUDIT_TOKEN_FILE` + * env vars are set; run by hand from a developer machine, never in CI. + * + * The token file path is read indirectly so the credential never appears + * in the test source, environment dump, or vitest reporter output. + * + * Run manually: + * MESH_AUDIT_URL=http://:1852 \ + * MESH_AUDIT_TOKEN_FILE=/tmp/mesh-audit-token.txt \ + * npx vitest run --no-coverage src/__tests__/mesh-proxy-tunnel-live.test.ts + */ +import fs from 'fs'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; + +const REMOTE_URL = process.env.MESH_AUDIT_URL; +const TOKEN_FILE = process.env.MESH_AUDIT_TOKEN_FILE; +const enabled = !!(REMOTE_URL && TOKEN_FILE); +const describeLive = enabled ? describe : describe.skip; + +let tmpDir: string; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; +let MeshProxyTunnelDialer: typeof import('../services/MeshProxyTunnelDialer').MeshProxyTunnelDialer; +let PilotMetrics: typeof import('../services/PilotMetrics').PilotMetrics; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService } = await import('../services/DatabaseService')); + ({ MeshProxyTunnelDialer } = await import('../services/MeshProxyTunnelDialer')); + ({ PilotMetrics } = await import('../services/PilotMetrics')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +describeLive('MeshProxyTunnelDialer (live)', () => { + it('dials a real Sencho remote, registers a bridge, dedupes a second call, and tears down', async () => { + const token = fs.readFileSync(TOKEN_FILE!, 'utf8').trim(); + const dialer = MeshProxyTunnelDialer.resetForTest(0); // disable idle close + const db = DatabaseService.getInstance(); + const nodeId = db.addNode({ + name: 'audit-mesh-live', + type: 'remote', + compose_dir: '', + is_default: false, + mode: 'proxy', + api_url: REMOTE_URL!, + api_token: token, + }); + try { + const before = PilotMetrics.snapshot(); + const t0 = Date.now(); + const bridge = await dialer.ensureBridge(nodeId); + const elapsed = Date.now() - t0; + expect(bridge, `dial returned null after ${elapsed}ms`).not.toBeNull(); + expect(elapsed).toBeLessThan(15_000); + + // Dedupe: second call hits the cached bridge instantly. + const t1 = Date.now(); + const second = await dialer.ensureBridge(nodeId); + expect(second).toBe(bridge); + expect(Date.now() - t1).toBeLessThan(50); + + const after = PilotMetrics.snapshot(); + expect(after.proxy_bridges_total).toBeGreaterThan(before.proxy_bridges_total); + + dialer.closeBridge(nodeId, 'audit teardown'); + await new Promise((r) => setTimeout(r, 200)); + expect(dialer.hasBridge(nodeId)).toBe(false); + } finally { + db.getDb().prepare('DELETE FROM nodes WHERE id = ?').run(nodeId); + } + }, 30_000); + + it('records auth_failed and a single deduped activity entry when the token is rejected', async () => { + const dialer = MeshProxyTunnelDialer.resetForTest(0); + const db = DatabaseService.getInstance(); + const nodeId = db.addNode({ + name: 'audit-mesh-live-bad', + type: 'remote', + compose_dir: '', + is_default: false, + mode: 'proxy', + api_url: REMOTE_URL!, + api_token: 'this-is-not-a-real-jwt', + }); + try { + const result = await dialer.ensureBridge(nodeId); + expect(result).toBeNull(); + const failure = dialer.getRecentFailure(nodeId); + expect(failure?.code).toBe('auth_failed'); + + // Second call should short-circuit via the recent-failure cache. + const t0 = Date.now(); + const again = await dialer.ensureBridge(nodeId); + expect(again).toBeNull(); + expect(Date.now() - t0).toBeLessThan(20); + } finally { + db.getDb().prepare('DELETE FROM nodes WHERE id = ?').run(nodeId); + } + }, 20_000); +}); diff --git a/backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts b/backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts new file mode 100644 index 00000000..e860bf90 --- /dev/null +++ b/backend/src/__tests__/mesh-tcp-stream-switchboard.test.ts @@ -0,0 +1,210 @@ +/** + * TcpStreamSwitchboard forward path: `tcp_open` → resolveTarget → dial + * a local TCP target → splice bytes through `TcpData` frames. The + * reverse path is covered by `pilot-agent-reverse-stream.test.ts`; this + * file exercises the inbound side both the pilot agent and the proxy- + * mode WS handler share. + * + * Critical because forward `tcp_open` is the trust boundary on every + * mesh-receiving Sencho: the WS upgrade authenticates the credential, + * but every per-stream dial must resolve via Compose labels and never + * touch a non-mesh-opted target. Resolution policy lives in the + * caller's `resolveTarget` callback; the switchboard's job is to honor + * it without leaking on errors. + */ +import net from 'net'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; +import { + AGENT_REVERSE_ID_BASE, + BinaryFrameType, + decodeBinaryFrame, + decodeJsonFrame, + encodeBinaryFrame, +} from '../pilot/protocol'; + +let tmpDir: string; +let attachTcpStreamSwitchboard: typeof import('../mesh/tcpStreamSwitchboard').attachTcpStreamSwitchboard; + +interface CapturedSend { + raw: string | Buffer; + binary: boolean; +} + +function makeSwitchboard(opts: { + resolve?: typeof import('../mesh/tcpStreamSwitchboard').resolveByComposeLabels; + extraStreamCount?: () => number; +} = {}): { + switchboard: ReturnType; + sent: CapturedSend[]; + mockWs: { readyState: number; send: (data: unknown, opts?: { binary?: boolean }) => void }; +} { + const sent: CapturedSend[] = []; + const mockWs = { + readyState: 1, + send(data: unknown, sendOpts?: { binary?: boolean }) { + const isBinary = sendOpts?.binary === true; + sent.push({ + raw: isBinary ? (Buffer.isBuffer(data) ? data : Buffer.from(data as Uint8Array)) : String(data), + binary: isBinary, + }); + }, + }; + const switchboard = attachTcpStreamSwitchboard({ + ws: mockWs as unknown as import('ws').WebSocket, + resolveTarget: opts.resolve ?? (async () => ({ ok: false, err: 'no_target' })), + extraStreamCount: opts.extraStreamCount, + logLabel: 'Test', + }); + return { switchboard, sent, mockWs }; +} + +/** Spin up a one-shot localhost TCP server to serve as a dial target. */ +function startEchoServer(): Promise<{ port: number; server: net.Server; firstConn: Promise }> { + return new Promise((resolve) => { + let resolveFirst!: (s: net.Socket) => void; + const firstConn = new Promise((r) => { resolveFirst = r; }); + const server = net.createServer((sock) => { + resolveFirst(sock); + }); + server.listen(0, '127.0.0.1', () => { + const addr = server.address(); + if (!addr || typeof addr === 'string') throw new Error('no addr'); + resolve({ port: addr.port, server, firstConn }); + }); + }); +} + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ attachTcpStreamSwitchboard } = await import('../mesh/tcpStreamSwitchboard')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +describe('TcpStreamSwitchboard.onTcpOpen (forward path)', () => { + it('rejects with no_target when resolveTarget returns no_target, and never opens a socket', async () => { + const { switchboard, sent } = makeSwitchboard({ + resolve: async () => ({ ok: false, err: 'no_target' }), + }); + switchboard.handleJsonFrame({ t: 'tcp_open', s: 1, stack: 'api', service: 'db', port: 5432 }); + await new Promise((r) => setImmediate(r)); + const text = sent.find((s) => !s.binary); + expect(text).toBeDefined(); + const decoded = decodeJsonFrame(text!.raw as string); + if (decoded.t !== 'tcp_open_ack') throw new Error('expected ack'); + expect(decoded.s).toBe(1); + expect(decoded.ok).toBe(false); + expect(decoded.err).toBe('no_target'); + expect(switchboard.tcpStreamCount()).toBe(0); + }); + + it('rejects with the resolver-supplied err code (denied) without dialing', async () => { + const { switchboard, sent } = makeSwitchboard({ + resolve: async () => ({ ok: false, err: 'denied' }), + }); + switchboard.handleJsonFrame({ t: 'tcp_open', s: 7, stack: 'api', service: 'db', port: 5432 }); + await new Promise((r) => setImmediate(r)); + const ack = decodeJsonFrame(sent.find((s) => !s.binary)!.raw as string); + if (ack.t !== 'tcp_open_ack') throw new Error('expected ack'); + expect(ack.err).toBe('denied'); + }); + + it('happy path: resolves, dials a local TCP target, acks ok=true, splices echoed bytes back as TcpData frames', async () => { + const { port, server, firstConn } = await startEchoServer(); + try { + const { switchboard, sent } = makeSwitchboard({ + resolve: async () => ({ ok: true, host: '127.0.0.1', port }), + }); + switchboard.handleJsonFrame({ t: 'tcp_open', s: 42, stack: 'api', service: 'db', port: 5432 }); + + const targetSocket = await firstConn; + // Wait for the ack frame to land in `sent`. + await new Promise((r) => setTimeout(r, 25)); + const ack = sent.map((s) => s.binary ? null : decodeJsonFrame(s.raw as string)).find((d) => d?.t === 'tcp_open_ack'); + if (!ack || ack.t !== 'tcp_open_ack') throw new Error('no ack'); + expect(ack.ok).toBe(true); + expect(ack.s).toBe(42); + expect(switchboard.tcpStreamCount()).toBe(1); + + // Server writes some bytes; switchboard should encode them as a + // TcpData binary frame on the WS keyed on the same streamId. + targetSocket.write('pong'); + await new Promise((r) => setTimeout(r, 25)); + const dataFrame = sent + .filter((s) => s.binary) + .map((s) => decodeBinaryFrame(s.raw as Buffer)) + .find((d) => d.type === BinaryFrameType.TcpData && d.streamId === 42); + expect(dataFrame).toBeDefined(); + expect(dataFrame!.payload.toString()).toBe('pong'); + + // Inbound TcpData should land on the target socket. + const echoed = new Promise((r) => targetSocket.once('data', r)); + const buf = encodeBinaryFrame(BinaryFrameType.TcpData, 42, Buffer.from('ping')); + switchboard.handleBinaryFrame(decodeBinaryFrame(buf)); + const received = await echoed; + expect(received.toString()).toBe('ping'); + + switchboard.cleanup('test'); + } finally { + server.close(); + } + }); + + it('refuses to dial when the per-tunnel cap (with extraStreamCount) is already saturated', async () => { + const { switchboard, sent } = makeSwitchboard({ + resolve: async () => ({ ok: true, host: '127.0.0.1', port: 1 }), + // Pretend we already have MAX_STREAMS_PER_TUNNEL non-mesh streams. + extraStreamCount: () => 1024, + }); + switchboard.handleJsonFrame({ t: 'tcp_open', s: 99, stack: 's', service: 'svc', port: 80 }); + await new Promise((r) => setImmediate(r)); + const ack = decodeJsonFrame(sent.find((s) => !s.binary)!.raw as string); + if (ack.t !== 'tcp_open_ack') throw new Error('expected ack'); + expect(ack.ok).toBe(false); + expect(ack.err).toBe('agent_error'); + expect(switchboard.tcpStreamCount()).toBe(0); + }); + + it('returns false from handleJsonFrame for non-TCP frame types so the outer dispatcher can route them', () => { + const { switchboard } = makeSwitchboard(); + const consumed = switchboard.handleJsonFrame({ t: 'http_req', s: 1, method: 'GET', path: '/', headers: {} }); + expect(consumed).toBe(false); + }); + + it('returns false for tcp_open_ack with a forward-range id (low half) so the caller routes it elsewhere', () => { + const { switchboard } = makeSwitchboard(); + const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: 5, ok: true }); + expect(consumed).toBe(false); + }); + + it('consumes tcp_open_ack with a reverse-range id (high half) even when no matching handle exists', () => { + const { switchboard } = makeSwitchboard(); + const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: AGENT_REVERSE_ID_BASE + 1, ok: true }); + // Owned id-space; switchboard signals it consumed the frame even when + // the matching reverse handle was already torn down (race-safe). + expect(consumed).toBe(true); + }); + + it('tcp_close on a forward stream destroys the socket and drops the entry', async () => { + const { port, server, firstConn } = await startEchoServer(); + try { + const { switchboard } = makeSwitchboard({ + resolve: async () => ({ ok: true, host: '127.0.0.1', port }), + }); + switchboard.handleJsonFrame({ t: 'tcp_open', s: 11, stack: 's', service: 'svc', port: 80 }); + const sock = await firstConn; + await new Promise((r) => setTimeout(r, 25)); + expect(switchboard.tcpStreamCount()).toBe(1); + + const closed = new Promise((r) => sock.once('close', () => r())); + switchboard.handleJsonFrame({ t: 'tcp_close', s: 11 }); + await closed; + expect(switchboard.tcpStreamCount()).toBe(0); + } finally { + server.close(); + } + }); +}); diff --git a/backend/src/__tests__/pilot-agent-mesh-resolve.test.ts b/backend/src/__tests__/pilot-agent-mesh-resolve.test.ts index e0733e81..a5628bc3 100644 --- a/backend/src/__tests__/pilot-agent-mesh-resolve.test.ts +++ b/backend/src/__tests__/pilot-agent-mesh-resolve.test.ts @@ -1,53 +1,36 @@ /** - * F7 regression: pilot agent's mesh dial path no longer denies based on the - * pilot's local `mesh_stacks` table. Central is the sole authority for - * mesh opt-in (state lives in central's SQLite); the pilot resolves the - * target container by Compose labels and dials it directly. The tunnel JWT - * authenticates the caller, so per-stack gating on the pilot would only - * deny legitimate central-issued dials whenever Phase D's central-only - * state model is in effect. + * Compose-label container resolver shared by the pilot agent and the + * proxy-mode WS handler. The resolver runs the conventional compose name + * fast path first (`--1`), then falls back to a + * label-filtered `listContainers` call. The deterministic IP preference + * lives in `pickContainerIp` and is exercised here through the wrapper. + * + * Central is the sole authority for mesh opt-in (state lives in central's + * SQLite); the tunnel WS authentication is the trust boundary, so this + * resolver has no per-stack gating. */ import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; const listContainersMock = vi.fn(); +const inspectMock = vi.fn(); +const getContainerMock = vi.fn(() => ({ inspect: inspectMock })); vi.mock('dockerode', () => { function Docker(this: unknown) { - (this as { listContainers: typeof listContainersMock }).listContainers = listContainersMock; + const self = this as { listContainers: typeof listContainersMock; getContainer: typeof getContainerMock }; + self.listContainers = listContainersMock; + self.getContainer = getContainerMock; } return { default: Docker }; }); let tmpDir: string; -let PilotAgent: typeof import('../pilot/agent').PilotAgent; -let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; - -interface ResolveResult { - ok: boolean; - host?: string; - port?: number; - err?: string; -} - -function makeAgent(): import('../pilot/agent').PilotAgent { - return new PilotAgent({ - primaryUrl: 'http://primary.invalid', - loopbackPort: 1, - initialToken: 'irrelevant', - enrolling: false, - }); -} - -function callResolve(agent: import('../pilot/agent').PilotAgent, stack: string, service: string, port: number): Promise { - const fn = (agent as unknown as { resolveMeshTarget: (s: string, sv: string, p: number) => Promise }).resolveMeshTarget.bind(agent); - return fn(stack, service, port); -} +let resolveByComposeLabels: typeof import('../mesh/tcpStreamSwitchboard').resolveByComposeLabels; beforeAll(async () => { tmpDir = await setupTestDb(); - ({ PilotAgent } = await import('../pilot/agent')); - ({ DatabaseService } = await import('../services/DatabaseService')); + ({ resolveByComposeLabels } = await import('../mesh/tcpStreamSwitchboard')); }); afterAll(() => { @@ -56,32 +39,41 @@ afterAll(() => { afterEach(() => { listContainersMock.mockReset(); - DatabaseService.getInstance().getDb().prepare('DELETE FROM mesh_stacks').run(); + inspectMock.mockReset(); + getContainerMock.mockClear(); }); -describe('PilotAgent.resolveMeshTarget (F7: trust central)', () => { - it('returns the container IP when mesh_stacks is empty (central is the gate)', async () => { - listContainersMock.mockResolvedValue([ - { NetworkSettings: { Networks: { sencho_mesh: { IPAddress: '172.30.0.5' } } } }, - ]); +describe('resolveByComposeLabels (mesh target resolver)', () => { + it('uses the conventional name fast path when it succeeds', async () => { + inspectMock.mockResolvedValueOnce({ + NetworkSettings: { Networks: { sencho_mesh: { IPAddress: '172.30.0.5' } } }, + }); - const agent = makeAgent(); - const result = await callResolve(agent, 'audit-mesh-pilot', 'echo', 9001); + const result = await resolveByComposeLabels('audit-mesh-pilot', 'echo', 9001); - expect(result.err).not.toBe('denied'); expect(result.ok).toBe(true); + if (!result.ok) throw new Error('narrowing'); expect(result.host).toBe('172.30.0.5'); expect(result.port).toBe(9001); + // Fast path hit; the label-filtered listContainers call was not needed. + expect(getContainerMock).toHaveBeenCalledWith('audit-mesh-pilot-echo-1'); + expect(listContainersMock).not.toHaveBeenCalled(); }); - it('queries dockerode with the Compose project + service label filter', async () => { - listContainersMock.mockResolvedValue([ - { NetworkSettings: { Networks: { bridge: { IPAddress: '10.0.0.7' } } } }, - ]); + it('falls back to a label-filtered listContainers when the fast path returns no IP', async () => { + // Fast inspect returns a container with no usable IP. + inspectMock.mockResolvedValueOnce({ NetworkSettings: { Networks: {} } }); + listContainersMock.mockResolvedValueOnce([{ Id: 'abc' }]); + // Fallback inspect resolves the IP. + inspectMock.mockResolvedValueOnce({ + NetworkSettings: { Networks: { bridge: { IPAddress: '10.0.0.7' } } }, + }); - const agent = makeAgent(); - await callResolve(agent, 'api', 'db', 5432); + const result = await resolveByComposeLabels('api', 'db', 5432); + expect(result.ok).toBe(true); + if (!result.ok) throw new Error('narrowing'); + expect(result.host).toBe('10.0.0.7'); expect(listContainersMock).toHaveBeenCalledTimes(1); const args = listContainersMock.mock.calls[0][0] as { filters: { label: string[] } }; expect(args.filters.label).toEqual([ @@ -90,23 +82,44 @@ describe('PilotAgent.resolveMeshTarget (F7: trust central)', () => { ]); }); - it('returns no_target (not denied) when dockerode finds no matching container', async () => { - listContainersMock.mockResolvedValue([]); + it('returns no_target when both the fast path and label fallback find nothing', async () => { + inspectMock.mockResolvedValueOnce(null); + listContainersMock.mockResolvedValueOnce([]); - const agent = makeAgent(); - const result = await callResolve(agent, 'missing', 'svc', 8080); + const result = await resolveByComposeLabels('missing', 'svc', 8080); expect(result.ok).toBe(false); + if (result.ok) throw new Error('narrowing'); expect(result.err).toBe('no_target'); }); - it('returns agent_error (not denied) when dockerode throws', async () => { - listContainersMock.mockRejectedValue(new Error('docker daemon unreachable')); + it('returns agent_error when dockerode rejects on the listContainers call', async () => { + inspectMock.mockResolvedValueOnce(null); + listContainersMock.mockRejectedValueOnce(new Error('docker daemon unreachable')); - const agent = makeAgent(); - const result = await callResolve(agent, 'api', 'db', 5432); + const result = await resolveByComposeLabels('api', 'db', 5432); expect(result.ok).toBe(false); + if (result.ok) throw new Error('narrowing'); expect(result.err).toBe('agent_error'); }); + + it('prefers the compose default network over any other attached network', async () => { + // Container is attached to both bridge and stack_default; pickContainerIp + // must return the stack_default IP regardless of object key order. + inspectMock.mockResolvedValueOnce({ + NetworkSettings: { + Networks: { + bridge: { IPAddress: '10.0.0.7' }, + api_default: { IPAddress: '172.30.0.42' }, + }, + }, + }); + + const result = await resolveByComposeLabels('api', 'db', 5432); + + expect(result.ok).toBe(true); + if (!result.ok) throw new Error('narrowing'); + expect(result.host).toBe('172.30.0.42'); + }); }); diff --git a/backend/src/__tests__/pilot-agent-reverse-stream.test.ts b/backend/src/__tests__/pilot-agent-reverse-stream.test.ts index 84afd321..b380d1f0 100644 --- a/backend/src/__tests__/pilot-agent-reverse-stream.test.ts +++ b/backend/src/__tests__/pilot-agent-reverse-stream.test.ts @@ -1,10 +1,12 @@ /** - * Phase B: PilotAgent's outbound reverse mesh stream. The agent's - * MeshForwarder hands off cross-node connections to PilotAgent via - * `openMeshTcpStream`, which sends a `tcp_open_reverse` frame and returns - * a stream handle. The test exercises the wire shape and the lifecycle - * dispatchers (open / data / close) without needing a real WebSocket or - * the full Sencho boot. + * Reverse mesh streams: the switchboard's `openReverseStream` is the + * outbound side of the protocol. Exercises wire-shape (the JSON + * `tcp_open_reverse` frame and the binary `TcpData` envelope) and the + * stream-handle lifecycle dispatchers (open / data / close). + * + * Both the pilot agent and the proxy-mode WS handler delegate to the + * switchboard's `openReverseStream`; the lifecycle assertions here cover + * the shared logic both callers depend on. */ import { afterAll, beforeAll, describe, expect, it } from 'vitest'; import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; @@ -17,15 +19,19 @@ import { } from '../pilot/protocol'; let tmpDir: string; -let PilotAgent: typeof import('../pilot/agent').PilotAgent; -let ReverseTcpStreamHandle: typeof import('../pilot/agent').ReverseTcpStreamHandle; +let attachTcpStreamSwitchboard: typeof import('../mesh/tcpStreamSwitchboard').attachTcpStreamSwitchboard; +let ReverseTcpStreamHandle: typeof import('../mesh/tcpStreamSwitchboard').ReverseTcpStreamHandle; interface CapturedSend { raw: string | Buffer; binary: boolean; } -function makeMockAgent(): { agent: import('../pilot/agent').PilotAgent; sent: CapturedSend[]; mockWs: { readyState: number; send: (data: unknown, opts?: { binary?: boolean }) => void } } { +function makeSwitchboard(): { + switchboard: ReturnType; + sent: CapturedSend[]; + mockWs: { readyState: number; send: (data: unknown, opts?: { binary?: boolean }) => void }; +} { const sent: CapturedSend[] = []; const mockWs = { readyState: 1, // WebSocket.OPEN @@ -37,32 +43,29 @@ function makeMockAgent(): { agent: import('../pilot/agent').PilotAgent; sent: Ca }); }, }; - const agent = new PilotAgent({ - primaryUrl: 'http://primary.invalid', - loopbackPort: 1, - initialToken: 'irrelevant', - enrolling: false, + // The switchboard accepts the `ws` type structurally; the cast is + // narrow and only used in this test harness. + const switchboard = attachTcpStreamSwitchboard({ + ws: mockWs as unknown as import('ws').WebSocket, + resolveTarget: async () => ({ ok: false, err: 'no_target' }), + logLabel: 'Test', }); - // Inject the mock ws into the private slot. The agent's - // openMeshTcpStream checks readyState and calls send(); the mock - // captures both for assertions without needing real connect/handshake. - (agent as unknown as { ws: typeof mockWs }).ws = mockWs; - return { agent, sent, mockWs }; + return { switchboard, sent, mockWs }; } beforeAll(async () => { tmpDir = await setupTestDb(); - ({ PilotAgent, ReverseTcpStreamHandle } = await import('../pilot/agent')); + ({ attachTcpStreamSwitchboard, ReverseTcpStreamHandle } = await import('../mesh/tcpStreamSwitchboard')); }); afterAll(() => { cleanupTestDb(tmpDir); }); -describe('PilotAgent.openMeshTcpStream (Phase B)', () => { +describe('TcpStreamSwitchboard.openReverseStream', () => { it('allocates an id in the agent-reverse range and emits a tcp_open_reverse frame with the target', () => { - const { agent, sent } = makeMockAgent(); - const handle = agent.openMeshTcpStream({ + const { switchboard, sent } = makeSwitchboard(); + const handle = switchboard.openReverseStream({ nodeId: 12, stack: 'api', service: 'db', @@ -84,15 +87,15 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => { }); it('returns null when the tunnel is not OPEN', () => { - const { agent, mockWs } = makeMockAgent(); + const { switchboard, mockWs } = makeSwitchboard(); mockWs.readyState = 0; // CONNECTING - const handle = agent.openMeshTcpStream({ nodeId: 1, stack: 'a', service: 'b', port: 1 }); + const handle = switchboard.openReverseStream({ nodeId: 1, stack: 'a', service: 'b', port: 1 }); expect(handle).toBeNull(); }); - it('handle.write encodes a TcpData binary frame with the agent-allocated streamId', () => { - const { agent, sent } = makeMockAgent(); - const handle = agent.openMeshTcpStream({ nodeId: 2, stack: 's', service: 'svc', port: 80 }); + it('handle.write encodes a TcpData binary frame with the allocated streamId', () => { + const { switchboard, sent } = makeSwitchboard(); + const handle = switchboard.openReverseStream({ nodeId: 2, stack: 's', service: 'svc', port: 80 }); if (!handle) throw new Error('handle should exist'); sent.length = 0; // clear the open frame @@ -106,10 +109,11 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => { expect(decoded.payload.toString()).toBe('hello'); }); - it('handle.end sends a tcp_close JSON frame and removes the stream from the agent map', () => { - const { agent, sent } = makeMockAgent(); - const handle = agent.openMeshTcpStream({ nodeId: 3, stack: 's', service: 'svc', port: 80 }); + it('handle.end sends a tcp_close JSON frame and drops the stream count', () => { + const { switchboard, sent } = makeSwitchboard(); + const handle = switchboard.openReverseStream({ nodeId: 3, stack: 's', service: 'svc', port: 80 }); if (!handle) throw new Error('handle should exist'); + expect(switchboard.tcpStreamCount()).toBe(1); sent.length = 0; handle.end(); @@ -121,31 +125,25 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => { if (decoded.t !== 'tcp_close') throw new Error('narrowing'); expect(decoded.s).toBe(handle.streamId); - const internalMap = (agent as unknown as { reverseTcpStreams: Map }).reverseTcpStreams; - expect(internalMap.has(handle.streamId)).toBe(false); + expect(switchboard.tcpStreamCount()).toBe(0); }); it('inbound tcp_open_ack {ok: true} fires the open event on the matching handle', () => { - const { agent, mockWs } = makeMockAgent(); - const handle = agent.openMeshTcpStream({ nodeId: 4, stack: 's', service: 'svc', port: 80 }); + const { switchboard } = makeSwitchboard(); + const handle = switchboard.openReverseStream({ nodeId: 4, stack: 's', service: 'svc', port: 80 }); if (!handle) throw new Error('handle should exist'); let opened = false; handle.on('open', () => { opened = true; }); - // Simulate an inbound ack frame: invoke the agent's private json- - // dispatch path with a synthetic frame. - const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent); - onTcpOpenAckReverse({ t: 'tcp_open_ack', s: handle.streamId, ok: true }); - + const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: handle.streamId, ok: true }); + expect(consumed).toBe(true); expect(opened).toBe(true); - // Sanity: mockWs's readyState wasn't touched. - expect(mockWs.readyState).toBe(1); }); it('inbound tcp_open_ack {ok: false} emits error and close, then drops the handle', () => { - const { agent } = makeMockAgent(); - const handle = agent.openMeshTcpStream({ nodeId: 5, stack: 's', service: 'svc', port: 80 }); + const { switchboard } = makeSwitchboard(); + const handle = switchboard.openReverseStream({ nodeId: 5, stack: 's', service: 'svc', port: 80 }); if (!handle) throw new Error('handle should exist'); let errMessage: string | undefined; @@ -153,46 +151,60 @@ describe('PilotAgent.openMeshTcpStream (Phase B)', () => { handle.on('error', (err: Error) => { errMessage = err.message; }); handle.on('close', () => { closed = true; }); - const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent); - onTcpOpenAckReverse({ t: 'tcp_open_ack', s: handle.streamId, ok: false, err: 'unreachable' }); + switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: handle.streamId, ok: false, err: 'unreachable' }); expect(errMessage).toBe('unreachable'); expect(closed).toBe(true); - const internalMap = (agent as unknown as { reverseTcpStreams: Map }).reverseTcpStreams; - expect(internalMap.has(handle.streamId)).toBe(false); + expect(switchboard.tcpStreamCount()).toBe(0); }); - it('forward-direction tcp_open_ack ids (low half) do not match reverse handles', () => { - const { agent } = makeMockAgent(); - const handle = agent.openMeshTcpStream({ nodeId: 6, stack: 's', service: 'svc', port: 80 }); + it('forward-direction tcp_open_ack ids (low half) are not consumed by the switchboard', () => { + const { switchboard } = makeSwitchboard(); + const handle = switchboard.openReverseStream({ nodeId: 6, stack: 's', service: 'svc', port: 80 }); if (!handle) throw new Error('handle should exist'); let opened = false; handle.on('open', () => { opened = true; }); - // Primary-direction id (< AGENT_REVERSE_ID_BASE) must not bleed into - // the reverse map. - const onTcpOpenAckReverse = (agent as unknown as { onTcpOpenAckReverse: (frame: unknown) => void }).onTcpOpenAckReverse.bind(agent); - onTcpOpenAckReverse({ t: 'tcp_open_ack', s: 5, ok: true }); - + const consumed = switchboard.handleJsonFrame({ t: 'tcp_open_ack', s: 5, ok: true }); + // Low-half ack is left for the caller to route (it's the + // primary-allocated forward-stream ack, irrelevant to the + // switchboard's reverse map). + expect(consumed).toBe(false); expect(opened).toBe(false); }); it('inbound TcpData binary frame (encoded against the wire) emits data on the handle', () => { - const { agent } = makeMockAgent(); - const handle = agent.openMeshTcpStream({ nodeId: 7, stack: 's', service: 'svc', port: 80 }); + const { switchboard } = makeSwitchboard(); + const handle = switchboard.openReverseStream({ nodeId: 7, stack: 's', service: 'svc', port: 80 }); if (!handle) throw new Error('handle should exist'); const received: Buffer[] = []; handle.on('data', (chunk: Buffer) => received.push(chunk)); - // Drive the agent's binary-frame handler with a wire-encoded TcpData - // frame to also exercise the routing branch added in Phase B. const buf = encodeBinaryFrame(BinaryFrameType.TcpData, handle.streamId, Buffer.from('echo!')); const decoded = decodeBinaryFrame(buf); - const handleBinaryFrame = (agent as unknown as { handleBinaryFrame: (frame: unknown) => void }).handleBinaryFrame.bind(agent); - handleBinaryFrame(decoded); - + const consumed = switchboard.handleBinaryFrame(decoded); + expect(consumed).toBe(true); expect(Buffer.concat(received).toString()).toBe('echo!'); }); + + it('cleanup() emits error + close on every outstanding reverse handle and resets the count', () => { + const { switchboard } = makeSwitchboard(); + const h1 = switchboard.openReverseStream({ nodeId: 8, stack: 's', service: 'a', port: 1 })!; + const h2 = switchboard.openReverseStream({ nodeId: 9, stack: 's', service: 'b', port: 2 })!; + const closed: number[] = []; + const errors: string[] = []; + h1.on('close', () => closed.push(h1.streamId)); + h2.on('close', () => closed.push(h2.streamId)); + h1.on('error', (e: Error) => errors.push(e.message)); + h2.on('error', (e: Error) => errors.push(e.message)); + + switchboard.cleanup('mock disconnect'); + + expect(closed).toContain(h1.streamId); + expect(closed).toContain(h2.streamId); + expect(errors).toEqual(expect.arrayContaining(['mock disconnect', 'mock disconnect'])); + expect(switchboard.tcpStreamCount()).toBe(0); + }); }); diff --git a/backend/src/__tests__/pilot-tunnel-manager-ensure.test.ts b/backend/src/__tests__/pilot-tunnel-manager-ensure.test.ts new file mode 100644 index 00000000..6091c907 --- /dev/null +++ b/backend/src/__tests__/pilot-tunnel-manager-ensure.test.ts @@ -0,0 +1,112 @@ +/** + * PilotTunnelManager.ensureBridge: dispatcher used by MeshService to + * resolve a mesh-capable bridge for any nodeId without caring whether the + * node runs in pilot-agent mode (long-lived tunnel) or Distributed API + * mode (Phase C on-demand proxy tunnel). + * + * Covers: + * - returns the existing bridge when one is registered. + * - delegates to MeshProxyTunnelDialer.ensureBridge when none exists. + * - registerProxyBridge refuses to shadow a pre-existing bridge. + */ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; + +let tmpDir: string; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; +let PilotTunnelManager: typeof import('../services/PilotTunnelManager').PilotTunnelManager; +let MeshProxyTunnelDialer: typeof import('../services/MeshProxyTunnelDialer').MeshProxyTunnelDialer; +let PilotTunnelBridge: typeof import('../services/PilotTunnelBridge').PilotTunnelBridge; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService } = await import('../services/DatabaseService')); + ({ PilotTunnelManager } = await import('../services/PilotTunnelManager')); + ({ MeshProxyTunnelDialer } = await import('../services/MeshProxyTunnelDialer')); + ({ PilotTunnelBridge } = await import('../services/PilotTunnelBridge')); +}); + +afterAll(() => { + cleanupTestDb(tmpDir); +}); + +beforeEach(() => { + DatabaseService.getInstance().getDb().prepare("DELETE FROM nodes WHERE name LIKE 'pilot-mgr-test-%'").run(); + MeshProxyTunnelDialer.resetForTest(0); // idle-close disabled; clears any prior bridges + // Drop any bridges held over from a prior test in the same file. + type BridgeMap = Map; + const bridges = (PilotTunnelManager.getInstance() as unknown as { bridges: BridgeMap }).bridges; + bridges.clear(); +}); + +function makeFakeBridge(nodeId: number): { bridge: import('../services/PilotTunnelBridge').PilotTunnelBridge; close: () => void } { + // Bridges hold a WS reference but only the surface we touch in this + // test (openTcpStream, getBufferedAmount, close emission) ever runs. + const mockWs = { + readyState: 1, + bufferedAmount: 0, + on() { return this; }, + off() { return this; }, + once() { return this; }, + send() { /* ignore */ }, + close() { /* ignore */ }, + } as unknown as import('ws').WebSocket; + const bridge = new PilotTunnelBridge(nodeId, mockWs); + return { bridge, close: () => bridge.close(1000, 'test cleanup') }; +} + +describe('PilotTunnelManager.ensureBridge', () => { + it('returns the existing bridge when one is registered for the node', async () => { + const mgr = PilotTunnelManager.getInstance(); + const { bridge, close } = makeFakeBridge(101); + try { + mgr.registerProxyBridge(101, bridge); + const handle = await mgr.ensureBridge(101); + expect(handle).toBe(bridge); + } finally { + close(); + } + }); + + it('delegates to MeshProxyTunnelDialer when no bridge is registered', async () => { + const mgr = PilotTunnelManager.getInstance(); + // No bridge, no node configured -> dialer returns null and caches no_target. + const handle = await mgr.ensureBridge(404); + expect(handle).toBeNull(); + const cached = MeshProxyTunnelDialer.getInstance().getRecentFailure(404); + expect(cached?.code).toBe('no_target'); + }); + + it('registerProxyBridge refuses to shadow a node that already has a bridge', () => { + const mgr = PilotTunnelManager.getInstance(); + const { bridge: bridgeA, close: closeA } = makeFakeBridge(202); + const { bridge: bridgeB, close: closeB } = makeFakeBridge(202); + try { + mgr.registerProxyBridge(202, bridgeA); + expect(() => mgr.registerProxyBridge(202, bridgeB)).toThrow(/already registered/); + expect(mgr.getBridge(202)).toBe(bridgeA); + } finally { + closeA(); + closeB(); + } + }); + + it('registerProxyBridge emits proxy-bridge-up and proxy-bridge-down on lifecycle', async () => { + const mgr = PilotTunnelManager.getInstance(); + const up: number[] = []; + const down: number[] = []; + mgr.on('proxy-bridge-up', (id: number) => up.push(id)); + mgr.on('proxy-bridge-down', (id: number) => down.push(id)); + + const { bridge } = makeFakeBridge(303); + mgr.registerProxyBridge(303, bridge); + expect(up).toContain(303); + + bridge.close(1000, 'test teardown'); + await new Promise((resolve) => setImmediate(resolve)); + expect(down).toContain(303); + + mgr.off('proxy-bridge-up', () => {}); + mgr.off('proxy-bridge-down', () => {}); + }); +}); diff --git a/backend/src/mesh/containerLookup.ts b/backend/src/mesh/containerLookup.ts new file mode 100644 index 00000000..b2123384 --- /dev/null +++ b/backend/src/mesh/containerLookup.ts @@ -0,0 +1,84 @@ +/** + * Shared Compose-container lookup for mesh routing. Both + * `MeshService.resolveContainerIp` (same-node fast path) and the + * `TcpStreamSwitchboard` (inbound `tcp_open` handler) need to translate + * a stack/service pair to a single deterministic IPv4 address; their + * earlier implementations diverged on the network-preference rule, + * which caused flaky resolution on containers attached to multiple + * networks (compose `_default` plus `sencho_mesh`, for example). + * + * `pickContainerIp` is the canonical preference rule. `lookupContainerIp` + * runs the conventional-name fast path first (`--1`) + * before falling back to a compose-label container list. + */ +interface ContainerInspectInfo { + NetworkSettings?: { + Networks?: Record; + IPAddress?: string; + }; +} + +interface ContainerListItem { + Id: string; +} + +interface DockerodeLike { + getContainer(id: string): { inspect(): Promise }; + listContainers(opts: unknown): Promise; +} + +/** + * Pick a deterministic IPv4 for a container. Prefer the compose default + * network (`_default` or any network named `_*`), then any + * other attached network, then the legacy `NetworkSettings.IPAddress`. + * Without this preference order, `Object.values(Networks)` ordering on + * multi-network containers varies across daemon versions and can make + * same-node forwarding flaky on a redeploy. + */ +export function pickContainerIp(stackName: string, info: ContainerInspectInfo): string | null { + const networks = info.NetworkSettings?.Networks ?? {}; + const composeDefault = networks[`${stackName}_default`]; + if (composeDefault?.IPAddress) return composeDefault.IPAddress; + for (const [name, net] of Object.entries(networks)) { + if (name.startsWith(`${stackName}_`) && net?.IPAddress) return net.IPAddress; + } + for (const net of Object.values(networks)) { + if (net?.IPAddress) return net.IPAddress; + } + return info.NetworkSettings?.IPAddress || null; +} + +/** + * Resolve a stack + service to a container IPv4. Tries the conventional + * compose name first, then falls back to a label-filtered list. Returns + * null when no container matches or the matching container has no IP on + * any attached network. + * + * Errors propagate. Callers that only need a single string-or-null + * surface should wrap with a try/catch (`MeshService.resolveContainerIp` + * does this); callers that need to distinguish "Docker errored" from + * "no container matched" use the thrown error to pick the right code. + */ +export async function lookupContainerIp( + docker: DockerodeLike, + stack: string, + service: string, +): Promise { + const conventional = `${stack}-${service}-1`; + const fast = await docker.getContainer(conventional).inspect().catch(() => null); + const fastIp = fast ? pickContainerIp(stack, fast) : null; + if (fastIp) return fastIp; + + const containers = (await docker.listContainers({ + all: true, + filters: { + label: [ + `com.docker.compose.project=${stack}`, + `com.docker.compose.service=${service}`, + ], + }, + })) as ContainerListItem[]; + if (containers.length === 0) return null; + const info = await docker.getContainer(containers[0].Id).inspect().catch(() => null); + return info ? pickContainerIp(stack, info) : null; +} diff --git a/backend/src/mesh/tcpStreamSwitchboard.ts b/backend/src/mesh/tcpStreamSwitchboard.ts new file mode 100644 index 00000000..cf46f780 --- /dev/null +++ b/backend/src/mesh/tcpStreamSwitchboard.ts @@ -0,0 +1,415 @@ +import net from 'net'; +import { EventEmitter } from 'events'; +import WebSocket from 'ws'; +import { + AGENT_REVERSE_ID_BASE, + BinaryFrameType, + DecodedBinaryFrame, + JsonFrame, + MAX_STREAMS_PER_TUNNEL, + MeshErrCode, + STREAM_IDLE_TIMEOUT_MS, + StreamIdAllocator, + encodeBinaryFrame, + encodeJsonFrame, +} from '../pilot/protocol'; +import { lookupContainerIp } from './containerLookup'; +import { sanitizeForLog } from '../utils/safeLog'; +import { isDebugEnabled } from '../utils/debug'; + +/** + * Sencho Mesh TCP stream switchboard. + * + * Owns the "agent side" of the mesh frame protocol: accept `tcp_open`, + * resolve a target by Compose labels, dial a local container, and splice + * bytes via `TcpData` frames in both directions. Also owns the outbound + * `tcp_open_reverse` allocator so the local MeshForwarder can emit reverse + * streams when its meshed containers dial cross-node aliases. + * + * Single source of truth for the protocol's "agent" behavior. Two callers: + * + * - `backend/src/pilot/agent.ts` — pilot-mode WS client. Mesh ws is the + * long-lived pilot tunnel; the agent threads its HTTP and WebSocket + * stream count into `extraStreamCount` so the per-tunnel cap is shared. + * + * - `backend/src/websocket/meshProxyTunnel.ts` — proxy-mode WS server. + * Mesh ws is a short-lived central-initiated tunnel that only carries + * TCP frames; `extraStreamCount` defaults to 0. + * + * The switchboard does not touch authentication or lifecycle; callers + * authenticate the WS upgrade themselves and decide when to call + * `cleanup()` on disconnect. State held here (`tcpStreams`, + * `reverseTcpStreams`, `reverseStreamIds`, per-stream idle timers) is + * scoped to a single WS instance — recreate the switchboard on reconnect. + */ +const MESH_CONNECT_TIMEOUT_MS = 10_000; + +export type MeshResolveResult = + | { ok: true; host: string; port: number } + | { ok: false; err: MeshErrCode }; + +export type ResolveTarget = (stack: string, service: string, port: number) => Promise; + +export interface SwitchboardCtx { + /** Per-connection WebSocket. The switchboard sends frames on this and never reassigns it; recreate the switchboard on reconnect. */ + ws: WebSocket; + /** Resolve `stack`+`service`+`port` to a TCP target. Implementations typically query Dockerode by Compose labels. */ + resolveTarget: ResolveTarget; + /** Non-mesh streams sharing the per-tunnel cap (e.g., the pilot's HTTP + WS multiplex). 0 for pure-TCP tunnels. */ + extraStreamCount?: () => number; + /** Diagnostic prefix for log lines emitted from this switchboard. */ + logLabel?: string; +} + +interface ForwardTcpStream { + socket: net.Socket; + accepted: boolean; +} + +/** + * Handle returned by `openReverseStream` to MeshService. Mirrors the + * surface of `PilotTunnelBridge.TcpStream` (write/end/destroy + + * 'open'/'data'/'error'/'close' events) so MeshService.openCrossNode can + * splice bytes against it without caring whether the underlying tunnel is + * a pilot tunnel or a proxy-mode tunnel. + */ +export class ReverseTcpStreamHandle extends EventEmitter { + public readonly streamId: number; + private readonly sendData: (streamId: number, payload: Buffer) => void; + private readonly sendClose: (streamId: number) => void; + private closed = false; + + constructor( + streamId: number, + sendData: (streamId: number, payload: Buffer) => void, + sendClose: (streamId: number) => void, + ) { + super(); + this.streamId = streamId; + this.sendData = sendData; + this.sendClose = sendClose; + } + + public write(chunk: Buffer): boolean { + if (this.closed) return false; + this.sendData(this.streamId, chunk); + return true; + } + + public end(): void { + if (this.closed) return; + this.closed = true; + this.sendClose(this.streamId); + } + + public destroy(): void { this.end(); } + + /** @internal Called by the switchboard on inbound `tcp_open_ack { ok: true }`. */ + public _dispatchOpen(): void { this.emit('open'); } + /** @internal Called by the switchboard on inbound `TcpData` for this stream. */ + public _dispatchData(chunk: Buffer): void { this.emit('data', chunk); } + /** @internal Called by the switchboard on tunnel-side error or rejection. */ + public _dispatchError(err: Error): void { this.emit('error', err); } + /** @internal Called by the switchboard on tunnel-side close. */ + public _dispatchClose(): void { + if (this.closed) return; + this.closed = true; + this.emit('close'); + } +} + +export class TcpStreamSwitchboard { + private readonly ctx: SwitchboardCtx; + private readonly logLabel: string; + private readonly tcpStreams = new Map(); + private readonly reverseTcpStreams = new Map(); + private reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE); + private readonly idleTimers = new Map(); + + constructor(ctx: SwitchboardCtx) { + this.ctx = ctx; + this.logLabel = ctx.logLabel || 'Mesh'; + } + + /** Active mesh stream count (forward + reverse) owned by this switchboard. */ + public tcpStreamCount(): number { + return this.tcpStreams.size + this.reverseTcpStreams.size; + } + + private totalStreamCount(): number { + return this.tcpStreamCount() + (this.ctx.extraStreamCount?.() ?? 0); + } + + private refreshIdleTimer(streamId: number): void { + const existing = this.idleTimers.get(streamId); + if (existing) clearTimeout(existing); + const timer = setTimeout(() => this.onStreamIdle(streamId), STREAM_IDLE_TIMEOUT_MS); + this.idleTimers.set(streamId, timer); + } + + private clearIdleTimer(streamId: number): void { + const timer = this.idleTimers.get(streamId); + if (timer) { + clearTimeout(timer); + this.idleTimers.delete(streamId); + } + } + + private onStreamIdle(streamId: number): void { + this.idleTimers.delete(streamId); + const ws = this.ctx.ws; + const fwd = this.tcpStreams.get(streamId); + if (fwd) { + try { fwd.socket.destroy(); } catch { /* ignore */ } + this.tcpStreams.delete(streamId); + try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ } + return; + } + const reverse = this.reverseTcpStreams.get(streamId); + if (reverse) { + this.reverseTcpStreams.delete(streamId); + reverse._dispatchError(new Error('mesh idle timeout')); + reverse._dispatchClose(); + try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ } + } + } + + /** + * Dispatch a JSON frame. Returns true when the frame matched one of the + * TCP-related types and was handled; false otherwise so the caller's + * outer dispatcher can route HTTP / WS / control frames. + */ + public handleJsonFrame(frame: JsonFrame): boolean { + switch (frame.t) { + case 'tcp_open': + void this.onTcpOpen(frame); + return true; + case 'tcp_open_ack': + if (frame.s < AGENT_REVERSE_ID_BASE) return false; + this.onTcpOpenAckReverse(frame); + return true; + case 'tcp_close': + this.onTcpClose(frame.s); + return true; + default: + return false; + } + } + + /** + * Dispatch a binary frame. Returns true iff the frame is `TcpData` (the + * only binary type owned by the switchboard). + */ + public handleBinaryFrame(frame: DecodedBinaryFrame): boolean { + if (frame.type !== BinaryFrameType.TcpData) return false; + if (frame.streamId >= AGENT_REVERSE_ID_BASE) { + const reverse = this.reverseTcpStreams.get(frame.streamId); + if (!reverse) return true; + reverse._dispatchData(frame.payload); + this.refreshIdleTimer(frame.streamId); + return true; + } + const fwd = this.tcpStreams.get(frame.streamId); + if (!fwd) return true; + try { fwd.socket.write(frame.payload); } catch { /* ignore */ } + this.refreshIdleTimer(frame.streamId); + return true; + } + + private async onTcpOpen(frame: { s: number; stack: string; service: string; port: number }): Promise { + const ws = this.ctx.ws; + if (this.totalStreamCount() >= MAX_STREAMS_PER_TUNNEL) { + try { + ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: 'agent_error' })); + } catch { /* ignore */ } + return; + } + + const target = await this.ctx.resolveTarget(frame.stack, frame.service, frame.port); + if (!target.ok) { + try { + ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: target.err })); + } catch { /* ignore */ } + return; + } + + const socket = net.createConnection({ host: target.host, port: target.port }); + socket.setTimeout(MESH_CONNECT_TIMEOUT_MS); + const entry: ForwardTcpStream = { socket, accepted: false }; + this.tcpStreams.set(frame.s, entry); + this.refreshIdleTimer(frame.s); + + const sendAck = (ok: boolean, err?: MeshErrCode) => { + try { ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok, err })); } catch { /* ignore */ } + }; + + socket.once('connect', () => { + entry.accepted = true; + socket.setTimeout(0); + sendAck(true); + this.refreshIdleTimer(frame.s); + }); + socket.on('data', (chunk: Buffer) => { + try { + ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, frame.s, chunk), { binary: true }); + } catch { /* ignore */ } + this.refreshIdleTimer(frame.s); + }); + socket.on('timeout', () => { + if (entry.accepted) return; + entry.accepted = true; + sendAck(false, 'unreachable'); + this.tcpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); + try { socket.destroy(); } catch { /* ignore */ } + }); + socket.on('error', (err) => { + if (!entry.accepted) { + entry.accepted = true; + sendAck(false, 'unreachable'); + this.tcpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); + return; + } + if (isDebugEnabled()) { + console.warn(`[${this.logLabel}:diag] tcp stream error:`, sanitizeForLog(err.message)); + } + if (this.tcpStreams.delete(frame.s)) { + this.clearIdleTimer(frame.s); + try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ } + } + }); + socket.on('close', () => { + if (this.tcpStreams.delete(frame.s)) { + this.clearIdleTimer(frame.s); + try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ } + } + }); + } + + private onTcpClose(streamId: number): void { + if (streamId >= AGENT_REVERSE_ID_BASE) { + const handle = this.reverseTcpStreams.get(streamId); + if (!handle) return; + this.reverseTcpStreams.delete(streamId); + this.clearIdleTimer(streamId); + handle._dispatchClose(); + return; + } + const entry = this.tcpStreams.get(streamId); + if (!entry) return; + this.tcpStreams.delete(streamId); + this.clearIdleTimer(streamId); + try { entry.socket.destroy(); } catch { /* ignore */ } + } + + private onTcpOpenAckReverse(frame: { s: number; ok: boolean; err?: MeshErrCode }): void { + const handle = this.reverseTcpStreams.get(frame.s); + if (!handle) return; + if (frame.ok) { + handle._dispatchOpen(); + this.refreshIdleTimer(frame.s); + } else { + this.reverseTcpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); + handle._dispatchError(new Error(frame.err ?? 'tcp_open_reverse rejected')); + handle._dispatchClose(); + } + } + + /** + * Allocate a reverse stream id, send `tcp_open_reverse`, and return a + * handle MeshService can splice bytes through. Returns null if the WS + * is not OPEN or the per-tunnel cap is reached. + */ + public openReverseStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null { + const ws = this.ctx.ws; + if (ws.readyState !== WebSocket.OPEN) return null; + if (this.totalStreamCount() >= MAX_STREAMS_PER_TUNNEL) return null; + const streamId = this.reverseStreamIds.allocate(); + const handle = new ReverseTcpStreamHandle( + streamId, + (sid, payload) => { + if (ws.readyState !== WebSocket.OPEN) return; + try { ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, sid, payload), { binary: true }); } catch { /* ignore */ } + this.refreshIdleTimer(sid); + }, + (sid) => { + if (!this.reverseTcpStreams.has(sid)) return; + this.reverseTcpStreams.delete(sid); + this.clearIdleTimer(sid); + if (ws.readyState !== WebSocket.OPEN) return; + try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: sid })); } catch { /* ignore */ } + }, + ); + this.reverseTcpStreams.set(streamId, handle); + this.refreshIdleTimer(streamId); + try { + ws.send(encodeJsonFrame({ + t: 'tcp_open_reverse', + s: streamId, + targetNodeId: target.nodeId, + stack: target.stack, + service: target.service, + port: target.port, + })); + } catch (err) { + this.reverseTcpStreams.delete(streamId); + this.clearIdleTimer(streamId); + handle._dispatchError(err as Error); + handle._dispatchClose(); + return null; + } + return handle; + } + + /** + * Tear down all stream state. Call on WS disconnect; the next + * connection should construct a fresh switchboard. + */ + public cleanup(reason = 'mesh tunnel closed'): void { + for (const [, entry] of this.tcpStreams) { + try { entry.socket.destroy(); } catch { /* ignore */ } + } + this.tcpStreams.clear(); + for (const [, handle] of this.reverseTcpStreams) { + try { + handle._dispatchError(new Error(reason)); + handle._dispatchClose(); + } catch { /* ignore */ } + } + this.reverseTcpStreams.clear(); + // Reset the allocator so a long-lived caller that reconnects many + // times doesn't drift up the id range and approach the wrap point. + this.reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE); + for (const [, timer] of this.idleTimers) clearTimeout(timer); + this.idleTimers.clear(); + } +} + +export function attachTcpStreamSwitchboard(ctx: SwitchboardCtx): TcpStreamSwitchboard { + return new TcpStreamSwitchboard(ctx); +} + +/** + * Resolve a Compose-managed container's IP address by stack name + service + * name. Honors the deterministic network preference shared with + * `MeshService.resolveContainerIp` so a redeploy that adds or reorders + * networks does not flip which IP is returned. Used by both the pilot + * agent and the proxy-mode WS handler. + */ +export async function resolveByComposeLabels(stack: string, service: string, port: number): Promise { + try { + const dockerodeMod = await import('dockerode'); + const Docker = (dockerodeMod as { default: new (opts?: unknown) => Parameters[0] }).default; + const docker = new Docker(); + const ip = await lookupContainerIp(docker, stack, service); + if (!ip) return { ok: false, err: 'no_target' }; + return { ok: true, host: ip, port }; + } catch (err) { + if (isDebugEnabled()) { + console.warn('[Mesh:diag] resolveByComposeLabels failed:', sanitizeForLog((err as Error).message)); + } + return { ok: false, err: 'agent_error' }; + } +} diff --git a/backend/src/pilot/agent.ts b/backend/src/pilot/agent.ts index bbc70bb1..4419ab4a 100644 --- a/backend/src/pilot/agent.ts +++ b/backend/src/pilot/agent.ts @@ -1,22 +1,17 @@ import fs from 'fs'; -import net from 'net'; import path from 'path'; import http from 'http'; -import { EventEmitter } from 'events'; import jwt from 'jsonwebtoken'; import WebSocket from 'ws'; import { getSenchoVersion } from '../services/CapabilityRegistry'; import { DatabaseService } from '../services/DatabaseService'; import { NodeRegistry } from '../services/NodeRegistry'; import { - AGENT_REVERSE_ID_BASE, BinaryFrameType, MAX_FRAME_SIZE_BYTES, MAX_STREAMS_PER_TUNNEL, - MeshErrCode, PROTOCOL_VERSION, STREAM_IDLE_TIMEOUT_MS, - StreamIdAllocator, decodeBinaryFrame, decodeJsonFrame, encodeBinaryFrame, @@ -24,7 +19,14 @@ import { wsDataToBuffer, wsDataToString, } from './protocol'; +import { + ReverseTcpStreamHandle, + TcpStreamSwitchboard, + attachTcpStreamSwitchboard, + resolveByComposeLabels, +} from '../mesh/tcpStreamSwitchboard'; import { sanitizeForLog } from '../utils/safeLog'; +import { httpUrlToWs } from '../utils/wsUrl'; import { isDebugEnabled } from '../utils/debug'; const RECONNECT_MIN_MS = 1_000; @@ -90,11 +92,9 @@ export class PilotAgent { private reconnectTimer?: NodeJS.Timeout; private readonly httpStreams = new Map(); private readonly wsStreams = new Map(); - private readonly tcpStreams = new Map(); - /** Reverse mesh streams the agent itself initiated via `tcp_open_reverse`. Keyed on the agent-allocated id. Disjoint from `tcpStreams` because those are primary-allocated and live in the lower id half. */ - private readonly reverseTcpStreams = new Map(); - /** Allocator is recreated on every disconnect (`cleanupAfterDisconnect`) so a long-lived agent that reconnects many times doesn't drift up the id range. */ - private reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE); + /** Per-connection mesh frame handler. Created on `connect()`, cleaned up on disconnect. */ + private switchboard: TcpStreamSwitchboard | null = null; + /** Idle timers for HTTP and WebSocket streams only; mesh streams keep their own timers inside the switchboard. */ private readonly idleTimers = new Map(); private shuttingDown = false; private readonly agentVersion: string; @@ -170,7 +170,7 @@ export class PilotAgent { private connect(): void { if (this.shuttingDown) return; - const wsUrl = this.options.primaryUrl.replace(/^http/, 'ws').replace(/\/$/, '') + '/api/pilot/tunnel'; + const wsUrl = httpUrlToWs(this.options.primaryUrl) + '/api/pilot/tunnel'; const ws = new WebSocket(wsUrl, { headers: { Authorization: `Bearer ${this.token}`, @@ -180,13 +180,19 @@ export class PilotAgent { maxPayload: MAX_FRAME_SIZE_BYTES, // Self-signed deployments can supply an internal CA bundle via // SENCHO_PILOT_CA_FILE; rejectUnauthorized stays true. There is - // intentionally no env var to disable TLS verification — that + // intentionally no env var to disable TLS verification, which // would defeat the entire trust model of the tunnel credential. // The bundle is read once at agent construction (this.customCa); // rotate by restarting the container. ...(this.customCa ? { ca: this.customCa } : {}), }); this.ws = ws; + this.switchboard = attachTcpStreamSwitchboard({ + ws, + resolveTarget: resolveByComposeLabels, + extraStreamCount: () => this.httpStreams.size + this.wsStreams.size, + logLabel: 'Pilot', + }); ws.on('open', () => { // Backoff intentionally NOT reset here: a TCP-level connect that @@ -234,27 +240,16 @@ export class PilotAgent { try { ws.close(1006, 'tunnel closed'); } catch { /* ignore */ } } this.wsStreams.clear(); - for (const [, stream] of this.tcpStreams) { - try { stream.socket.destroy(); } catch { /* ignore */ } + if (this.switchboard) { + this.switchboard.cleanup('pilot tunnel closed'); + this.switchboard = null; } - this.tcpStreams.clear(); - for (const [, handle] of this.reverseTcpStreams) { - try { - handle._dispatchError(new Error('pilot tunnel closed')); - handle._dispatchClose(); - } catch { /* ignore */ } - } - this.reverseTcpStreams.clear(); - // Reset the reverse allocator so a long-lived agent that - // reconnects many times doesn't drift up the id range and - // approach the wrap point unnecessarily. - this.reverseStreamIds = new StreamIdAllocator(AGENT_REVERSE_ID_BASE); for (const [, timer] of this.idleTimers) clearTimeout(timer); this.idleTimers.clear(); } private streamCount(): number { - return this.httpStreams.size + this.wsStreams.size + this.tcpStreams.size + this.reverseTcpStreams.size; + return this.httpStreams.size + this.wsStreams.size + (this.switchboard?.tcpStreamCount() ?? 0); } private refreshIdleTimer(streamId: number): void { @@ -291,25 +286,6 @@ export class PilotAgent { if (ws) { try { ws.send(encodeJsonFrame({ t: 'ws_close', s: streamId, code: 1001, reason: 'idle' })); } catch { /* ignore */ } } - return; - } - const tcpEntry = this.tcpStreams.get(streamId); - if (tcpEntry) { - try { tcpEntry.socket.destroy(); } catch { /* ignore */ } - this.tcpStreams.delete(streamId); - if (ws) { - try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ } - } - return; - } - const reverseEntry = this.reverseTcpStreams.get(streamId); - if (reverseEntry) { - this.reverseTcpStreams.delete(streamId); - reverseEntry._dispatchError(new Error('agent idle timeout')); - reverseEntry._dispatchClose(); - if (ws) { - try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: streamId })); } catch { /* ignore */ } - } } } @@ -346,6 +322,7 @@ export class PilotAgent { private handleJsonFrame(frame: ReturnType): void { const ws = this.ws; if (!ws) return; + if (this.switchboard?.handleJsonFrame(frame)) return; switch (frame.t) { case 'hello': { if (frame.version !== PROTOCOL_VERSION) { @@ -374,9 +351,6 @@ export class PilotAgent { case 'ws_open': this.onWsOpen(frame); break; case 'ws_msg_text': this.onWsMsgText(frame.s, frame.data); break; case 'ws_close': this.onWsClose(frame.s, frame.code, frame.reason); break; - case 'tcp_open': this.onTcpOpen(frame); break; - case 'tcp_open_ack': this.onTcpOpenAckReverse(frame); break; - case 'tcp_close': this.onTcpClose(frame.s); break; default: // Other frame types are primary-bound only; agent ignores. break; @@ -384,6 +358,7 @@ export class PilotAgent { } private handleBinaryFrame(frame: ReturnType): void { + if (this.switchboard?.handleBinaryFrame(frame)) return; switch (frame.type) { case BinaryFrameType.HttpReqBody: { const entry = this.httpStreams.get(frame.streamId); @@ -399,20 +374,6 @@ export class PilotAgent { this.refreshIdleTimer(frame.streamId); break; } - case BinaryFrameType.TcpData: { - if (frame.streamId >= AGENT_REVERSE_ID_BASE) { - const reverse = this.reverseTcpStreams.get(frame.streamId); - if (!reverse) return; - reverse._dispatchData(frame.payload); - this.refreshIdleTimer(frame.streamId); - return; - } - const stream = this.tcpStreams.get(frame.streamId); - if (!stream) return; - try { stream.socket.write(frame.payload); } catch { /* ignore */ } - this.refreshIdleTimer(frame.streamId); - break; - } default: break; } @@ -560,272 +521,27 @@ export class PilotAgent { this.clearIdleTimer(streamId); } - // --- Sencho Mesh TCP dispatch (tunnel -> Compose service container) --- + // --- Sencho Mesh TCP dispatch --- // - // Central is the sole authority for mesh opt-in (state lives in central's - // SQLite mesh_stacks table). The pilot resolves a target by Compose - // container labels and dials directly. The tunnel JWT (scope - // 'pilot_tunnel') gates the WS upgrade itself, so any tcp_open frame on - // an open tunnel is trusted to originate from central. - - private async onTcpOpen(frame: Extract, { t: 'tcp_open' }>): Promise { - const ws = this.ws; - if (!ws) return; - if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) { - try { - ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: 'agent_error' })); - } catch { /* ignore */ } - return; - } - - const target = await this.resolveMeshTarget(frame.stack, frame.service, frame.port); - if (!target.ok) { - try { - ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok: false, err: target.err })); - } catch { /* ignore */ } - return; - } - - const socket = net.createConnection({ host: target.host, port: target.port }); - socket.setTimeout(MESH_CONNECT_TIMEOUT_MS); - const entry: MeshTcpStream = { socket, accepted: false }; - this.tcpStreams.set(frame.s, entry); - this.refreshIdleTimer(frame.s); - - const sendAck = (ok: boolean, err?: MeshErrCode) => { - try { ws.send(encodeJsonFrame({ t: 'tcp_open_ack', s: frame.s, ok, err })); } catch { /* ignore */ } - }; - - socket.once('connect', () => { - entry.accepted = true; - socket.setTimeout(0); - sendAck(true); - this.refreshIdleTimer(frame.s); - }); - socket.on('data', (chunk: Buffer) => { - try { - ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, frame.s, chunk), { binary: true }); - } catch { /* ignore */ } - this.refreshIdleTimer(frame.s); - }); - socket.on('timeout', () => { - if (entry.accepted) return; - entry.accepted = true; - sendAck(false, 'unreachable'); - this.tcpStreams.delete(frame.s); - this.clearIdleTimer(frame.s); - try { socket.destroy(); } catch { /* ignore */ } - }); - socket.on('error', (err) => { - if (!entry.accepted) { - entry.accepted = true; - sendAck(false, 'unreachable'); - this.tcpStreams.delete(frame.s); - this.clearIdleTimer(frame.s); - return; - } - console.warn('[Pilot] tcp stream error:', sanitizeForLog(err.message)); - if (this.tcpStreams.delete(frame.s)) { - this.clearIdleTimer(frame.s); - try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ } - } - }); - socket.on('close', () => { - if (this.tcpStreams.delete(frame.s)) { - this.clearIdleTimer(frame.s); - try { ws.send(encodeJsonFrame({ t: 'tcp_close', s: frame.s })); } catch { /* ignore */ } - } - }); - } - - private onTcpClose(streamId: number): void { - // Reverse stream (agent-initiated): primary is closing the - // upstream half. Tear down the local socket the agent's - // MeshForwarder handed us via openMeshTcpStream. - if (streamId >= AGENT_REVERSE_ID_BASE) { - const handle = this.reverseTcpStreams.get(streamId); - if (!handle) return; - this.reverseTcpStreams.delete(streamId); - this.clearIdleTimer(streamId); - handle._dispatchClose(); - return; - } - const entry = this.tcpStreams.get(streamId); - if (!entry) return; - this.tcpStreams.delete(streamId); - this.clearIdleTimer(streamId); - try { entry.socket.destroy(); } catch { /* ignore */ } - } + // Mesh frame handling lives in `backend/src/mesh/tcpStreamSwitchboard.ts` + // and is shared with the proxy-mode WS handler. The agent owns a + // switchboard per WS connection and delegates the public reverse-dial + // surface used by MeshService. /** - * Inbound `tcp_open_ack` for an agent-initiated reverse stream. The - * primary acknowledges (or rejects) the dial; emit 'open' or 'error' on - * the local handle so MeshService's splice setup can either start - * piping bytes or tear down the source socket. - */ - private onTcpOpenAckReverse(frame: Extract, { t: 'tcp_open_ack' }>): void { - if (frame.s < AGENT_REVERSE_ID_BASE) return; // forward-direction acks are primary-bound; ignore. - const handle = this.reverseTcpStreams.get(frame.s); - if (!handle) return; - if (frame.ok) { - handle._dispatchOpen(); - this.refreshIdleTimer(frame.s); - } else { - this.reverseTcpStreams.delete(frame.s); - this.clearIdleTimer(frame.s); - handle._dispatchError(new Error(frame.err ?? 'tcp_open_reverse rejected')); - handle._dispatchClose(); - } - } - - /** - * Public entry point for the agent's mesh forwarder. Allocates a - * reverse stream id, sends `tcp_open_reverse`, and returns a handle - * MeshService can splice bytes through. Returns null if the tunnel is - * not currently open or the per-tunnel stream cap is reached. + * Allocates a reverse stream id, sends `tcp_open_reverse`, and returns + * a handle MeshService can splice bytes through. Returns null if the + * tunnel is not currently open or the per-tunnel stream cap is + * reached. Delegates to the per-connection switchboard. */ public openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null { const ws = this.ws; if (!ws || ws.readyState !== WebSocket.OPEN) return null; if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) return null; - const streamId = this.reverseStreamIds.allocate(); - const handle = new ReverseTcpStreamHandle( - streamId, - (sid, payload) => { - if (this.ws?.readyState !== WebSocket.OPEN) return; - try { this.ws.send(encodeBinaryFrame(BinaryFrameType.TcpData, sid, payload), { binary: true }); } catch { /* ignore */ } - this.refreshIdleTimer(sid); - }, - (sid) => { - if (!this.reverseTcpStreams.has(sid)) return; - this.reverseTcpStreams.delete(sid); - this.clearIdleTimer(sid); - if (this.ws?.readyState !== WebSocket.OPEN) return; - try { this.ws.send(encodeJsonFrame({ t: 'tcp_close', s: sid })); } catch { /* ignore */ } - }, - ); - this.reverseTcpStreams.set(streamId, handle); - this.refreshIdleTimer(streamId); - try { - ws.send(encodeJsonFrame({ - t: 'tcp_open_reverse', - s: streamId, - targetNodeId: target.nodeId, - stack: target.stack, - service: target.service, - port: target.port, - })); - } catch (err) { - this.reverseTcpStreams.delete(streamId); - this.clearIdleTimer(streamId); - handle._dispatchError(err as Error); - handle._dispatchClose(); - return null; - } - return handle; - } - - /** - * Resolves a mesh target by Compose container labels and returns the - * container's first usable IP. Central has already validated that the - * target stack is opted in before issuing the dial; the tunnel JWT - * (scope 'pilot_tunnel') authenticates the caller, so this handler - * does no per-stack gating of its own. - */ - private async resolveMeshTarget( - stack: string, - service: string, - port: number, - ): Promise { - try { - const dockerodeMod = await import('dockerode'); - const Docker = (dockerodeMod as { default: new (opts?: unknown) => { listContainers: (opts?: unknown) => Promise } }).default; - const docker = new Docker(); - const containers = (await docker.listContainers({ - filters: { label: [`com.docker.compose.project=${stack}`, `com.docker.compose.service=${service}`] }, - })) as Array<{ NetworkSettings?: { Networks?: Record } }>; - for (const c of containers) { - const networks = c.NetworkSettings?.Networks ?? {}; - for (const net of Object.values(networks)) { - if (net.IPAddress) return { ok: true, host: net.IPAddress, port }; - } - } - return { ok: false, err: 'no_target' }; - } catch (err) { - console.warn('[Pilot] resolveMeshTarget failed:', sanitizeForLog((err as Error).message)); - return { ok: false, err: 'agent_error' }; - } + return this.switchboard?.openReverseStream(target) ?? null; } } -const MESH_CONNECT_TIMEOUT_MS = 10_000; - -interface MeshTcpStream { - socket: net.Socket; - accepted: boolean; -} - -/** - * Handle returned by `PilotAgent.openMeshTcpStream` to MeshService. Mirrors - * the surface of `PilotTunnelBridge.TcpStream` (write/end/destroy + - * 'open'/'data'/'error'/'close' events) so MeshService.openCrossNode can - * splice bytes against it without caring whether it's running on central - * or on a pilot. The agent owns the per-stream WS plumbing through the - * `sendData` and `sendClose` callbacks; this class is a thin EventEmitter - * facade. - */ -export class ReverseTcpStreamHandle extends EventEmitter { - public readonly streamId: number; - private readonly sendData: (streamId: number, payload: Buffer) => void; - private readonly sendClose: (streamId: number) => void; - private closed = false; - - constructor( - streamId: number, - sendData: (streamId: number, payload: Buffer) => void, - sendClose: (streamId: number) => void, - ) { - super(); - this.streamId = streamId; - this.sendData = sendData; - this.sendClose = sendClose; - } - - public write(chunk: Buffer): boolean { - if (this.closed) return false; - this.sendData(this.streamId, chunk); - return true; - } - - public end(): void { - if (this.closed) return; - this.closed = true; - this.sendClose(this.streamId); - } - - public destroy(): void { this.end(); } - - /** @internal Called by PilotAgent on inbound `tcp_open_ack { ok: true }`. */ - public _dispatchOpen(): void { this.emit('open'); } - - /** @internal Called by PilotAgent on inbound `TcpData` for this stream. */ - public _dispatchData(chunk: Buffer): void { this.emit('data', chunk); } - - /** @internal Called by PilotAgent on tunnel-side error or rejection. */ - public _dispatchError(err: Error): void { this.emit('error', err); } - - /** @internal Called by PilotAgent on tunnel-side close. */ - public _dispatchClose(): void { - if (this.closed) return; - this.closed = true; - this.emit('close'); - } -} - -type MeshResolveResult = - | { ok: true; host: string; port: number } - | { ok: false; err: MeshErrCode }; - /** * Read the persisted long-lived tunnel token from disk if present. ENOENT is * the normal first-boot case and stays silent. Any other error class diff --git a/backend/src/services/MeshProxyTunnelDialer.ts b/backend/src/services/MeshProxyTunnelDialer.ts new file mode 100644 index 00000000..50c1472c --- /dev/null +++ b/backend/src/services/MeshProxyTunnelDialer.ts @@ -0,0 +1,371 @@ +import { EventEmitter } from 'events'; +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 { redactSensitiveText, sanitizeForLog } from '../utils/safeLog'; +import { httpUrlToWs } from '../utils/wsUrl'; +import { isDebugEnabled } from '../utils/debug'; +import { PilotMetrics } from './PilotMetrics'; +import type { MeshActivityType } from './MeshService'; + +/** + * Central-side dialer for proxy-mode mesh tunnels. + * + * Today's pilot tunnels are agent-initiated: the pilot dials central and + * the resulting WS is registered in `PilotTunnelManager`. Proxy-mode + * remotes do not dial; central reaches them via the existing HTTP proxy + * using the long-lived `api_token`. To carry streaming TCP mesh traffic to + * a proxy-mode remote, central opens a WebSocket to the remote's new + * `/api/mesh/proxy-tunnel` endpoint on demand and registers the resulting + * `PilotTunnelBridge` in the manager under the same nodeId. The rest of + * the mesh code path (alias dispatch, openTcpStream, reverse-stream relay) + * is mode-agnostic from there. + * + * Lifecycle: + * - `ensureBridge(nodeId)` dials if not connected; concurrent callers + * dedupe through an in-flight promise map. + * - A periodic check tears down bridges that have seen no active streams + * for `SENCHO_MESH_PROXY_TUNNEL_IDLE_MS` (default 5 minutes). Setting + * the env var to `0` disables idle close (tunnel persists until the + * remote drops it or central shuts down). + * - Recent failures are cached for 60 seconds so a misconfigured remote + * does not cause a continuous redial storm; `MeshService.getStatus` + * consults the cache to surface a `reachableReason` to the UI. + */ +const DEFAULT_IDLE_TTL_MS = 5 * 60 * 1000; +const IDLE_CHECK_INTERVAL_MS = 60 * 1000; +const HANDSHAKE_TIMEOUT_MS = 15_000; +const FAILURE_CACHE_TTL_MS = 60 * 1000; + +export type DialFailureCode = + | 'no_target' + | 'endpoint_not_found' + | 'auth_failed' + | 'tls_failed' + | 'network_error'; + +/** + * Activity-log reason. Wider than `DialFailureCode` because some failures + * share a wire-level code (e.g., `network_error`) but deserve a more + * specific label in the operator-facing log to distinguish a network- + * layer failure from a post-handshake bridge failure or a manager + * rejection. + */ +type DialFailureReason = DialFailureCode | 'bridge_start_failed' | 'manager_rejected'; + +type ProxyTunnelEvent = 'open.ok' | 'open.fail' | 'close'; + +const ACTIVITY_TYPE: Record = { + 'open.ok': 'proxy-tunnel.open.ok', + 'open.fail': 'proxy-tunnel.open.fail', + 'close': 'proxy-tunnel.close', +}; + +export interface DialFailure { + code: DialFailureCode; + message?: string; + ts: number; +} + +export class MeshProxyTunnelDialer extends EventEmitter { + private static instance: MeshProxyTunnelDialer | null = null; + + private readonly bridges = new Map(); + private readonly inflight = new Map>(); + private readonly idleSince = new Map(); + private readonly recentFailures = new Map(); + private readonly idleTtlMs: number; + private idleCheckTimer: NodeJS.Timeout | null = null; + private stopped = false; + + private constructor(idleTtlOverrideMs?: number) { + super(); + if (typeof idleTtlOverrideMs === 'number') { + this.idleTtlMs = idleTtlOverrideMs; + } else { + const raw = process.env.SENCHO_MESH_PROXY_TUNNEL_IDLE_MS; + const parsed = raw === undefined ? Number.NaN : Number(raw); + this.idleTtlMs = Number.isFinite(parsed) && parsed >= 0 ? parsed : DEFAULT_IDLE_TTL_MS; + } + this.startIdleCheck(); + } + + public static getInstance(): MeshProxyTunnelDialer { + if (!this.instance) this.instance = new MeshProxyTunnelDialer(); + return this.instance; + } + + /** Test hook: reset the singleton with an optional idle-TTL override. */ + public static resetForTest(idleTtlOverrideMs?: number): MeshProxyTunnelDialer { + if (this.instance) this.instance.stop(); + this.instance = new MeshProxyTunnelDialer(idleTtlOverrideMs); + return this.instance; + } + + public hasBridge(nodeId: number): boolean { + return this.bridges.has(nodeId); + } + + public getBridge(nodeId: number): MeshTunnelHandle | null { + return this.bridges.get(nodeId) ?? null; + } + + public getRecentFailure(nodeId: number): DialFailure | null { + const entry = this.recentFailures.get(nodeId); + if (!entry) return null; + if (Date.now() - entry.ts > FAILURE_CACHE_TTL_MS) { + this.recentFailures.delete(nodeId); + return null; + } + return entry; + } + + /** + * Dial-if-needed. Concurrent callers dedupe through `inflight`; a + * cached recent failure short-circuits the dial so a misconfigured + * remote does not see one upgrade attempt per cross-node TCP open. + */ + public async ensureBridge(nodeId: number): Promise { + const existing = this.bridges.get(nodeId); + if (existing) { + this.idleSince.set(nodeId, Date.now()); + return existing; + } + if (this.getRecentFailure(nodeId)) return null; + const inflight = this.inflight.get(nodeId); + if (inflight) return inflight; + const dial = this.dial(nodeId).finally(() => { + this.inflight.delete(nodeId); + }); + this.inflight.set(nodeId, dial); + return dial; + } + + /** Force-close a bridge (e.g., on node deletion or scope change). */ + public closeBridge(nodeId: number, reason = 'closed by dialer'): void { + const bridge = this.bridges.get(nodeId); + if (!bridge) return; + this.bridges.delete(nodeId); + this.idleSince.delete(nodeId); + try { bridge.close(1000, reason); } catch { /* ignore */ } + } + + /** Stop the idle-check timer and tear down every open bridge. */ + public stop(): void { + this.stopped = true; + if (this.idleCheckTimer) { + clearInterval(this.idleCheckTimer); + this.idleCheckTimer = null; + } + for (const [nodeId, bridge] of this.bridges) { + this.bridges.delete(nodeId); + try { bridge.close(1000, 'dialer shutdown'); } catch { /* ignore */ } + } + this.idleSince.clear(); + this.recentFailures.clear(); + this.inflight.clear(); + } + + /** Test hook: count active bridges. */ + public get activeBridgeCount(): number { + return this.bridges.size; + } + + private async dial(nodeId: number): Promise { + const target = NodeRegistry.getInstance().getProxyTarget(nodeId); + if (!target || !target.apiToken) { + this.recordFailure(nodeId, 'no_target', 'no proxy target configured'); + return null; + } + + if (this.stopped) return null; + const dialStartedAt = Date.now(); + if (isDebugEnabled()) { + console.log(`[MeshProxyDialer:diag] dialing node=${nodeId} url=${sanitizeForLog(target.apiUrl)}`.replace(/[\n\r]/g, '')); + } + + const wsUrl = httpUrlToWs(target.apiUrl) + '/api/mesh/proxy-tunnel'; + let ws: WebSocket; + try { + ws = new WebSocket(wsUrl, { + headers: { Authorization: `Bearer ${target.apiToken}` }, + handshakeTimeout: HANDSHAKE_TIMEOUT_MS, + maxPayload: MAX_FRAME_SIZE_BYTES, + }); + } catch (err) { + this.recordFailure(nodeId, 'network_error', (err as Error).message); + return null; + } + + try { + await this.awaitOpen(ws); + } catch (err) { + // `awaitOpen` removes the 'error' listener on reject; calling + // `close()` on a still-CONNECTING socket emits a tail 'error' + // ('WebSocket was closed before the connection was established') + // that would otherwise propagate as an unhandled exception. + ws.on('error', () => { /* swallow tail error */ }); + try { ws.close(); } catch { /* ignore */ } + const failure = classifyDialError(err); + this.recordFailure(nodeId, failure.code, failure.message); + return null; + } + + if (this.stopped) { + try { ws.close(1001, 'dialer shutdown'); } catch { /* ignore */ } + return null; + } + + const bridge = new PilotTunnelBridge(nodeId, ws); + try { + await bridge.start(); + } catch (err) { + this.recordFailure(nodeId, 'network_error', (err as Error).message, 'bridge_start_failed'); + try { bridge.close(1011, 'bridge start failed'); } catch { /* ignore */ } + return null; + } + + if (this.stopped) { + try { bridge.close(1001, 'dialer shutdown'); } catch { /* ignore */ } + return null; + } + + try { + PilotTunnelManager.getInstance().registerProxyBridge(nodeId, bridge); + } catch (err) { + // Cap hit or a pilot tunnel concurrently claimed this nodeId. + this.recordFailure(nodeId, 'network_error', (err as Error).message, 'manager_rejected'); + try { bridge.close(1013, 'manager rejected'); } catch { /* ignore */ } + return null; + } + + // 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.idleSince.set(nodeId, Date.now()); + this.recentFailures.delete(nodeId); + void this.logActivity(nodeId, 'open.ok', {}); + this.emit('proxy-bridge-up', nodeId); + if (isDebugEnabled()) { + console.log(`[MeshProxyDialer:diag] dial ok node=${nodeId} elapsedMs=${Date.now() - dialStartedAt}`.replace(/[\n\r]/g, '')); + } + return bridge; + } + + private awaitOpen(ws: WebSocket): Promise { + return new Promise((resolve, reject) => { + const cleanup = () => { + 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(); + const err = new Error(`upgrade failed: HTTP ${res.statusCode}`) as Error & { httpStatus?: number }; + err.httpStatus = res.statusCode ?? 0; + try { res.resume(); } catch { /* ignore */ } + reject(err); + }); + }); + } + + /** + * Cache a dial failure and emit at most one activity-log entry per + * cache window per `(nodeId, code)`. The dedupe bounds operator log + * noise when a meshed container retries cross-node TCP opens against + * a misconfigured remote. + */ + private recordFailure(nodeId: number, code: DialFailureCode, rawMessage?: string, reasonOverride?: DialFailureReason): void { + const message = rawMessage ? sanitizeForLog(redactSensitiveText(rawMessage)) : undefined; + const previous = this.recentFailures.get(nodeId); + const reason = reasonOverride ?? code; + const isFresh = !previous + || Date.now() - previous.ts > FAILURE_CACHE_TTL_MS + || previous.code !== code; + this.recentFailures.set(nodeId, { code, message, ts: Date.now() }); + PilotMetrics.increment('proxy_dials_failed'); + if (isFresh) { + void this.logActivity(nodeId, 'open.fail', message ? { reason, message } : { reason }); + } + if (isDebugEnabled()) { + console.warn(`[MeshProxyDialer:diag] dial failure node=${nodeId} code=${code} reason=${reason}${message ? ` message=${message}` : ''}`.replace(/[\n\r]/g, '')); + } + } + + private startIdleCheck(): void { + if (this.idleCheckTimer || this.stopped) return; + if (this.idleTtlMs <= 0) return; // 0 disables idle close + this.idleCheckTimer = setInterval(() => this.runIdleCheck(), IDLE_CHECK_INTERVAL_MS); + this.idleCheckTimer.unref?.(); + } + + private runIdleCheck(): void { + const now = Date.now(); + for (const [nodeId, bridge] of this.bridges) { + if (bridge.getActiveStreamCount() > 0) { + this.idleSince.set(nodeId, now); + continue; + } + 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); + if (isDebugEnabled()) { + console.log(`[MeshProxyDialer:diag] idle close node=${nodeId} idleMs=${now - last}`); + } + } + } + } + + private async logActivity( + nodeId: number, + event: ProxyTunnelEvent, + details: Record, + ): Promise { + try { + // Lazy import to avoid a circular dependency with MeshService. + const { MeshService } = await import('./MeshService'); + const message = typeof details.message === 'string' + ? details.message + : `proxy tunnel ${event} for node ${nodeId}`; + MeshService.getInstance().logActivity({ + source: 'mesh', + level: event === 'open.fail' ? 'error' : 'info', + type: ACTIVITY_TYPE[event], + nodeId, + message, + details, + }); + } catch { + // Activity logging is best-effort; never let it propagate. + } + } +} + +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; + if (httpStatus === 404) return { code: 'endpoint_not_found', message: 'remote does not expose /api/mesh/proxy-tunnel' }; + if (httpStatus === 401 || httpStatus === 403) return { code: 'auth_failed', message: 'api token rejected by remote' }; + const tlsCodes = new Set(['CERT_HAS_EXPIRED', 'UNABLE_TO_VERIFY_LEAF_SIGNATURE', 'DEPTH_ZERO_SELF_SIGNED_CERT', 'SELF_SIGNED_CERT_IN_CHAIN']); + const errno = (err as NodeJS.ErrnoException).code; + if (errno && tlsCodes.has(errno)) return { code: 'tls_failed', message }; + return { code: 'network_error', message }; +} diff --git a/backend/src/services/MeshService.ts b/backend/src/services/MeshService.ts index b57ffb54..c9e4b466 100644 --- a/backend/src/services/MeshService.ts +++ b/backend/src/services/MeshService.ts @@ -12,7 +12,9 @@ import { PROXY_TIER_HEADER, PROXY_VARIANT_HEADER } from './license-headers'; import { MeshForwarder, type MeshForwarderHost } from './MeshForwarder'; import { NodeRegistry } from './NodeRegistry'; import { PilotTunnelManager } from './PilotTunnelManager'; +import { MeshProxyTunnelDialer, type DialFailureCode } from './MeshProxyTunnelDialer'; import { generateOverrideYaml, MeshAlias, SENCHO_MESH_NETWORK } from './MeshComposeOverride'; +import { lookupContainerIp } from '../mesh/containerLookup'; import { sanitizeForLog } from '../utils/safeLog'; import { isPathWithinBase, isValidStackName } from '../utils/validation'; import { PORT as SENCHO_LISTEN_PORT } from '../helpers/constants'; @@ -24,6 +26,14 @@ const PROBE_TIMEOUT_MS = 5_000; const SLOW_PROBE_THRESHOLD_MS = 500; const DEFAULT_MESH_SUBNET = '172.30.0.0/24'; +const REACHABLE_REASON: Record = { + auth_failed: 'api token rejected (scope must be full-admin)', + endpoint_not_found: 'remote does not support proxy mesh', + tls_failed: 'TLS handshake failed', + no_target: 'proxy target missing', + network_error: 'remote unreachable', +}; + /** * Returns the static IPv4 address Sencho will pin itself to on the mesh * Docker network: ` + 2`. The Docker daemon assigns @@ -60,7 +70,8 @@ export type MeshActivityType = | 'mesh.enable' | 'mesh.disable' | 'mesh.override.preserved' | 'probe.ok' | 'probe.fail' - | 'forwarder.listen' | 'forwarder.unlisten' | 'forwarder.error'; + | 'forwarder.listen' | 'forwarder.unlisten' | 'forwarder.error' + | 'proxy-tunnel.open.ok' | 'proxy-tunnel.open.fail' | 'proxy-tunnel.close'; export interface MeshActivityEvent { ts: number; @@ -105,13 +116,38 @@ export interface MeshRegenSummary { reason?: string; } +/** + * How a node participates in mesh routing right now: + * - `local`: the Sencho serving this request. + * - `pilot`: a remote with a pilot agent. Live-tunnel state is captured + * separately in `pilotConnected`. + * - `proxy`: a remote that central reaches via the long-lived api_token. + * Mesh tunnels are opened on demand by `MeshProxyTunnelDialer`; the + * operator sees no badge while the configuration is sound. + * - `unreachable`: configuration or runtime problem keeps mesh traffic + * from flowing. `reachableReason` carries an actionable hint. + */ +export type MeshReachableMode = 'local' | 'pilot' | 'proxy' | 'unreachable'; + export interface MeshNodeStatus { nodeId: number; nodeName: string; enabled: boolean; /** Forwarder state for the LOCAL node (the Sencho instance answering this request). Always `null` for any non-local node — fetching the remote forwarder state requires a cross-node call which lands in Phase B. */ localForwarderListening: boolean | null; + /** + * True iff a pilot tunnel is currently registered for this node. Only + * meaningful when `reachableMode === 'pilot'`. Kept for diagnostic + * surfaces; the Routing tab badge logic consumes `reachableMode` and + * ignores this field for proxy / local nodes. + * TODO: collapse into `reachableMode` (introduce `pilot_offline` value) + * once no remaining caller reads `pilotConnected` directly. + */ pilotConnected: boolean; + /** Canonical reachability classification consumed by the Routing tab. */ + reachableMode: MeshReachableMode; + /** Short, operator-facing reason when `reachableMode === 'unreachable'`. Null otherwise. */ + reachableReason: string | null; optedInStacks: string[]; activeStreamCount: number; } @@ -1284,7 +1320,7 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { if (target.nodeId === selfNodeId) { await this.openSameNode(target, src); } else { - this.openCrossNode(target, src); + await this.openCrossNode(target, src); } } @@ -1340,28 +1376,9 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { * central. */ public async resolveContainerIp(target: { stack: string; service: string }): Promise { + const docker = DockerController.getInstance().getDocker() as unknown as Parameters[0]; try { - const docker = DockerController.getInstance().getDocker(); - // Compose default container name pattern; -1 is the first replica. - const conventionalName = `${target.stack}-${target.service}-1`; - const info = await docker.getContainer(conventionalName).inspect().catch(() => null); - const fromInspect = info ? this.extractContainerIp(target.stack, info) : null; - if (fromInspect) return fromInspect; - // Fallback: filter by compose labels in case of a non-conventional - // container name (operator overrode `container_name` or compose - // project). - const containers = await docker.listContainers({ - all: true, - filters: { - label: [ - `com.docker.compose.project=${target.stack}`, - `com.docker.compose.service=${target.service}`, - ], - }, - }); - if (containers.length === 0) return null; - const fallbackInfo = await docker.getContainer(containers[0].Id).inspect().catch(() => null); - return fallbackInfo ? this.extractContainerIp(target.stack, fallbackInfo) : null; + return await lookupContainerIp(docker, target.stack, target.service); } catch (err) { console.warn('[MeshService] container IP lookup failed:', sanitizeForLog((err as Error).message)); return null; @@ -1369,43 +1386,39 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { } /** - * Pick a deterministic IP. Prefer the compose default network - * (`_default` or any network whose name starts with `_`), - * then any other declared network, then the legacy bridge `IPAddress`. - * Without this preference order, `Object.values(Networks)` ordering on - * containers attached to multiple networks varies across daemon - * versions and can make same-node forwarding flaky on a redeploy. - */ - private extractContainerIp( - stackName: string, - info: { NetworkSettings?: { Networks?: Record; IPAddress?: string } }, - ): string | null { - const networks = info.NetworkSettings?.Networks ?? {}; - const composeDefault = networks[`${stackName}_default`]; - if (composeDefault?.IPAddress) return composeDefault.IPAddress; - for (const [name, net] of Object.entries(networks)) { - if (name.startsWith(`${stackName}_`) && net?.IPAddress) return net.IPAddress; - } - for (const net of Object.values(networks)) { - if (net?.IPAddress) return net.IPAddress; - } - return info.NetworkSettings?.IPAddress || null; - } - - /** - * Pluggable reverse dialer. Set by `PilotAgent` on a pilot host; left - * null on a central host. When set, `openCrossNode` routes outbound - * mesh dials through the agent's `tcp_open_reverse` path; when unset, - * `openCrossNode` uses the central-side `PilotTunnelManager.getBridge` - * directly. Lets the same MeshService code work on both sides. + * Pluggable reverse dialer. Set by `PilotAgent` on a pilot host or by + * the proxy-mode WS handler when a `/api/mesh/proxy-tunnel` upgrade + * comes in; left null on a central host with no inbound mesh tunnel. + * When set, `openCrossNode` routes outbound mesh dials through the + * dialer's `tcp_open_reverse` path; when unset, `openCrossNode` uses + * the central-side `PilotTunnelManager.getBridge` directly. Lets the + * same MeshService code work on both sides. */ private reverseDialer: ReverseMeshDialer | null = null; - public setReverseDialer(dialer: ReverseMeshDialer | null): void { + /** + * Install or clear the reverse dialer. Supports compare-and-swap via + * the optional `expected` argument so a caller (e.g., the proxy-mode + * WS handler) can install on a null slot and uninstall only if its + * own dialer is still the active one. Without the `expected` arg the + * operation is unconditional (used by the pilot agent at boot). + * + * Returns true when the swap happened, false when the CAS rejected + * because `expected` did not match the current dialer. + */ + public setReverseDialer(dialer: ReverseMeshDialer | null, expected?: ReverseMeshDialer | null): boolean { + if (expected !== undefined && this.reverseDialer !== expected) return false; + // Loud warn instead of silent overwrite: pilot and proxy modes are + // mutually exclusive by topology, so this branch indicates a + // misconfigured deployment rather than an expected race. + if (expected === undefined && dialer !== null && this.reverseDialer !== null && this.reverseDialer !== dialer) { + console.warn('[MeshService] reverse dialer overwritten without CAS; pilot/proxy mode race or duplicate install'); + } this.reverseDialer = dialer; + return true; } - private dialMeshTcpStream(target: MeshTarget): MeshTcpStreamLike | null { + private async dialMeshTcpStream(target: MeshTarget): Promise { if (this.reverseDialer) { return this.reverseDialer.openMeshTcpStream({ nodeId: target.nodeId, @@ -1415,13 +1428,16 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { }); } const ptm = PilotTunnelManager.getInstance(); - if (!ptm.hasActiveTunnel(target.nodeId)) return null; - const bridge = ptm.getBridge(target.nodeId); + // ensureBridge resolves an existing pilot tunnel, an existing + // proxy-mode tunnel, or dials a fresh proxy-mode tunnel on demand. + // Returns null for unreachable nodes (no api_url/api_token, scope + // insufficient, remote pre-Phase-C, network error). + const bridge = await ptm.ensureBridge(target.nodeId); if (!bridge) return null; return bridge.openTcpStream({ stack: target.stack, service: target.service, port: target.port }); } - private openCrossNode(target: MeshTarget, src: net.Socket): void { + private async openCrossNode(target: MeshTarget, src: net.Socket): Promise { // Log every dispatch entry. Same-node logs route.resolve.ok on // its TCP `connect` event; cross-node only logs route.resolve.ok // once tcp_open_ack arrives from the agent. Without this entry @@ -1433,14 +1449,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { message: `cross-node dispatch to ${target.alias} on node ${target.nodeId}`, }); - const tcpStream = this.dialMeshTcpStream(target); + const tcpStream = await this.dialMeshTcpStream(target); if (!tcpStream) { this.logActivity({ source: 'pilot', level: 'error', type: 'tunnel.fail', nodeId: target.nodeId, alias: target.alias, message: this.reverseDialer ? `cannot open reverse mesh stream to node ${target.nodeId}` - : `no active pilot tunnel to node ${target.nodeId}`, + : `no mesh tunnel reachable for node ${target.nodeId}`, }); try { src.destroy(); } catch { /* ignore */ } return; @@ -1597,19 +1613,45 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { // --- Diagnostics --- /** - * Whether mesh traffic to this node can flow. Local nodes are always - * reachable because mesh uses the same-node fast path on localhost. - * Remote nodes are reachable only when a pilot tunnel is registered. The - * literal `hasActiveTunnel(localNodeId)` would always be false (local - * nodes do not establish tunnels to themselves), so a direct call would - * render every local alias as `tunnel down` in the UI even on a working - * route. + * Whether mesh CAN route to a node based on current configuration. + * Distinct from "live tunnel up right now": for proxy-mode remotes + * the tunnel is opened on demand, so a caller that demanded a + * current tunnel would misreport every idle proxy-mode route as + * `tunnel down`. Live pilot-tunnel state is surfaced separately. */ - private isMeshReachable(nodeId: number): boolean { - const node = DatabaseService.getInstance().getNode(nodeId); + private isNodeMeshConfigured(node: ReturnType): boolean { if (!node) return false; if (node.type !== 'remote') return true; - return PilotTunnelManager.getInstance().hasActiveTunnel(nodeId); + if (node.mode === 'pilot_agent') return true; + if (node.mode === 'proxy') return !!node.api_url && !!node.api_token; + return false; + } + + /** + * Compute the reachability classification surfaced in `MeshNodeStatus` + * and consumed by the Routing tab. See `MeshReachableMode` for the + * meaning of each value. Callers pass in the already-fetched node row + * (and the local nodeId) so this helper is free of DB I/O when + * `getStatus` iterates the fleet. + */ + private computeReachable(node: ReturnType, localNodeId: number): { mode: MeshReachableMode; reason: string | null } { + if (!node) return { mode: 'unreachable', reason: 'unknown node' }; + if (node.id === localNodeId || node.type !== 'remote') return { mode: 'local', reason: null }; + if (node.mode === 'pilot_agent') return { mode: 'pilot', reason: null }; + if (node.mode === 'proxy') { + if (!node.api_url) return { mode: 'unreachable', reason: 'api_url not set' }; + if (!node.api_token) return { mode: 'unreachable', reason: 'api token missing' }; + // Recent-failure cache surfaces the last failed dial so the + // operator sees a clear reason without triggering a redial + // storm. + const failure = MeshProxyTunnelDialer.getInstance().getRecentFailure(node.id); + if (failure) { + const reason = REACHABLE_REASON[failure.code] ?? failure.message ?? 'remote unreachable'; + return { mode: 'unreachable', reason }; + } + return { mode: 'proxy', reason: null }; + } + return { mode: 'unreachable', reason: 'unknown node mode' }; } public async getRouteDiagnostic(alias: string): Promise { @@ -1621,14 +1663,20 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { return { alias, target: null, pilot: { connected: false, lastSeen: null }, lastError, lastProbeMs, state: 'not authorized' }; } - const pilotConnected = this.isMeshReachable(target.nodeId); const node = DatabaseService.getInstance().getNode(target.nodeId); + const routable = this.isNodeMeshConfigured(node); + // Pilot-mode routes surface live tunnel state; proxy-mode routes + // fall back to `routable` because the tunnel is opened on demand + // and a quiescent state is normal. + const pilotLive = node?.type === 'remote' && node.mode === 'pilot_agent' + ? PilotTunnelManager.getInstance().hasActiveTunnel(target.nodeId) + : routable; const lastSeen = node?.pilot_last_seen ?? null; const optedIn = DatabaseService.getInstance().isMeshStackEnabled(target.nodeId, target.stackName); let state: MeshRouteDiagnostic['state']; if (!optedIn) state = 'not authorized'; - else if (!pilotConnected) state = 'tunnel down'; + else if (!routable || !pilotLive) state = 'tunnel down'; else if (lastError && Date.now() - lastError.ts < 60_000) state = 'unreachable'; else if (lastProbeMs !== null && lastProbeMs > SLOW_PROBE_THRESHOLD_MS) state = 'degraded'; else state = 'healthy'; @@ -1642,7 +1690,7 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { port: target.port, alias, }, - pilot: { connected: pilotConnected, lastSeen }, + pilot: { connected: pilotLive, lastSeen }, lastError, lastProbeMs, state, @@ -1689,15 +1737,26 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { const nodes = db.getNodes(); const localNodeId = NodeRegistry.getInstance().getDefaultNodeId(); const localListening = this.isLocalForwarderActive(); - return nodes.map((node) => ({ - nodeId: node.id, - nodeName: node.name, - enabled: db.getNodeMeshEnabled(node.id), - localForwarderListening: node.id === localNodeId ? localListening : null, - pilotConnected: this.isMeshReachable(node.id), - optedInStacks: db.listMeshStacks(node.id).map((s) => s.stack_name), - activeStreamCount: this.activeStreams.size, - })); + const ptm = PilotTunnelManager.getInstance(); + return nodes.map((node) => { + const reach = this.computeReachable(node, localNodeId); + return { + nodeId: node.id, + nodeName: node.name, + enabled: db.getNodeMeshEnabled(node.id), + localForwarderListening: node.id === localNodeId ? localListening : null, + // `pilotConnected` stays at its original meaning: a pilot + // tunnel is currently registered for this node. The + // Routing tab now derives badge state from `reachableMode` + // and reads `pilotConnected` only for the pilot-offline + // sub-state. + pilotConnected: node.type !== 'remote' || ptm.hasActiveTunnel(node.id), + reachableMode: reach.mode, + reachableReason: reach.reason, + optedInStacks: db.listMeshStacks(node.id).map((s) => s.stack_name), + activeStreamCount: this.activeStreams.size, + }; + }); } /** True when the local Sencho's forwarder is started and bound to at least one alias port. */ @@ -1731,7 +1790,7 @@ export class MeshError extends Error { /** * Common surface of an outbound mesh TCP stream as MeshService consumes * it. Both the central-side `PilotTunnelBridge.TcpStream` and the - * pilot-side `ReverseTcpStreamHandle` (from `pilot/agent.ts`) implement + * pilot-side `ReverseTcpStreamHandle` (from `mesh/tcpStreamSwitchboard.ts`) implement * this shape structurally so MeshService.openCrossNode can splice bytes * against either without caring which side initiated the stream. */ diff --git a/backend/src/services/PilotMetrics.ts b/backend/src/services/PilotMetrics.ts index 42e5976f..012fc33e 100644 --- a/backend/src/services/PilotMetrics.ts +++ b/backend/src/services/PilotMetrics.ts @@ -16,6 +16,12 @@ interface Counters { tunnels_rejected_capacity: number; enroll_acks: number; frame_decode_errors: number; + /** Successful Distributed API mesh proxy-tunnel registrations. Counterpart to `tunnels_total` for pilot-mode tunnels; the two together cover every mesh-capable bridge the manager ever accepted. */ + proxy_bridges_total: number; + /** Failed proxy-tunnel dial attempts (all reasons). Pair with `proxy_bridges_total` for an attempt/success ratio; pair with `proxy_idle_closes` for retention. */ + 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; } class PilotMetricsImpl { @@ -25,6 +31,9 @@ class PilotMetricsImpl { tunnels_rejected_capacity: 0, enroll_acks: 0, frame_decode_errors: 0, + proxy_bridges_total: 0, + proxy_dials_failed: 0, + proxy_idle_closes: 0, }; public increment(name: K): void { diff --git a/backend/src/services/PilotTunnelBridge.ts b/backend/src/services/PilotTunnelBridge.ts index 8eb3fd5b..d2d3e9cc 100644 --- a/backend/src/services/PilotTunnelBridge.ts +++ b/backend/src/services/PilotTunnelBridge.ts @@ -203,6 +203,8 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle public getLoopbackUrl(): string { return this.loopbackUrl; } public getConnectedAt(): number { return this.connectedAt; } public getBufferedAmount(): number { return this.tunnelWs.bufferedAmount; } + /** Total active stream count across HTTP, WebSocket, forward TCP, and reverse TCP. Used by the proxy-tunnel dialer to detect idle bridges eligible for teardown. */ + public getActiveStreamCount(): number { return this.streams.size; } public isOpen(): boolean { return !this.closed && this.tunnelWs.readyState === WebSocket.OPEN; } /** @@ -823,7 +825,10 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle private async acceptReverseRelay(s: number, targetNodeId: number, target: { stack: string; service: string; port: number }): Promise { const { PilotTunnelManager } = await import('./PilotTunnelManager'); - const targetBridge = PilotTunnelManager.getInstance().getBridge(targetNodeId); + // ensureBridge resolves either an existing pilot tunnel or a fresh + // proxy-mode tunnel dialed on demand, so proxy <-> proxy relays + // succeed even when neither remote has a pilot agent. + const targetBridge = await PilotTunnelManager.getInstance().ensureBridge(targetNodeId); if (!targetBridge) { this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' }); return; diff --git a/backend/src/services/PilotTunnelManager.ts b/backend/src/services/PilotTunnelManager.ts index 57e4195f..c16339cb 100644 --- a/backend/src/services/PilotTunnelManager.ts +++ b/backend/src/services/PilotTunnelManager.ts @@ -34,17 +34,36 @@ export class PilotTunnelCapacityError extends Error { } /** - * PilotTunnelManager: singleton registry of active pilot tunnels. + * PilotTunnelManager: singleton registry of active mesh-capable bridges. * - * Each enrolled pilot-agent node holds one outbound WebSocket to the primary. - * For every such tunnel we spin up a local loopback HTTP server that demuxes - * requests into frames. Remote-proxy code paths (http-proxy-middleware and the - * WebSocket upgrade handler) can then treat pilot nodes identically to standard - * proxy nodes by pointing at the loopback URL. + * Two flavors of bridge live in the same `bridges` map, keyed by nodeId: + * + * - **Pilot-agent tunnels** (the original use case): long-lived, + * agent-initiated. The pilot dials central; `registerTunnel` accepts + * the WS, starts a loopback HTTP server, and emits `tunnel-up` so + * downstream observers (capability cache, status badges) refresh. A + * pilot-agent bridge stays open for the agent's lifetime and supports + * HTTP, WebSocket, and TCP multiplexing. + * + * - **Proxy-mode tunnels** (Phase C): short-lived, central-initiated. + * `ensureBridge` delegates to `MeshProxyTunnelDialer`, which opens a + * WebSocket to the remote's `/api/mesh/proxy-tunnel` endpoint using + * the long-lived `api_token`. Carries only TCP mesh frames; the + * loopback HTTP server is left running on the bridge but unused + * because proxy-mode HTTP traffic flows through the existing + * `remoteNodeProxy`. Idle close after a configurable TTL. + * + * Mesh dispatch is mode-agnostic: `MeshService.dialMeshTcpStream` awaits + * `ensureBridge(nodeId)` and consumes the resulting `MeshTunnelHandle`. * * Events: - * - 'tunnel-up' (nodeId: number) after a tunnel is accepted - * - 'tunnel-down' (nodeId: number) after a tunnel closes (for any reason) + * - 'tunnel-up' (nodeId) when a pilot-agent tunnel is accepted (NOT + * emitted for proxy-mode bridges, which are opened on demand and + * should not trigger pilot-specific listeners like the F9 capability + * cache invalidation). + * - 'tunnel-down' (nodeId) when a pilot-agent tunnel closes. + * - 'proxy-bridge-up' / 'proxy-bridge-down' (nodeId) for observability + * on proxy-mode bridge lifecycle. No current consumer. */ export class PilotTunnelManager extends EventEmitter { private static instance: PilotTunnelManager; @@ -168,6 +187,60 @@ export class PilotTunnelManager extends EventEmitter { return this.bridges.get(nodeId) ?? null; } + /** + * Dial-if-needed: return the existing pilot or proxy bridge, or open + * a new proxy-mode bridge on demand. Used by `MeshService` so cross- + * node TCP dispatch works for both pilot-agent remotes (long-lived + * tunnel) and proxy-mode remotes (on-demand tunnel) without any + * mode-specific branching at the call site. + * + * Returns null if the node has no active pilot tunnel AND cannot be + * dialed as a proxy-mode remote (missing api_url / api_token, scope + * insufficient, remote offline, or remote pre-Phase-C). + */ + public async ensureBridge(nodeId: number): Promise { + const existing = this.bridges.get(nodeId); + if (existing) return existing; + // Lazy import to avoid a cycle: MeshProxyTunnelDialer imports + // PilotTunnelBridge, which imports PilotTunnelManager via the + // existing tcp_open_reverse relay path. + const { MeshProxyTunnelDialer } = await import('./MeshProxyTunnelDialer'); + return MeshProxyTunnelDialer.getInstance().ensureBridge(nodeId); + } + + /** + * Register a central-initiated proxy-mode bridge for an existing + * remote. Distinct from `registerTunnel`: skips the pilot-only side + * effects (DB node-status update, `pilot_last_seen` write, + * `tunnel-up` event, replacement of any prior pilot tunnel). Still + * honors the hard tunnel cap so a dial storm cannot exhaust gateway + * memory. + * + * Throws `PilotTunnelCapacityError` when the cap is reached. + */ + public registerProxyBridge(nodeId: number, bridge: PilotTunnelBridge): void { + const existing = this.bridges.get(nodeId); + if (existing) { + // A pilot tunnel for this node already exists. Proxy bridges + // should not silently shadow them; refuse the registration so + // the dialer can surface a clear error. + throw new Error(`pilot tunnel already registered for node ${nodeId}; proxy bridge refused`); + } + 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.emit('proxy-bridge-down', nodeId); + } + }); + this.bridges.set(nodeId, bridge); + PilotMetrics.increment('proxy_bridges_total'); + this.emit('proxy-bridge-up', nodeId); + } + /** * Force-close a tunnel (e.g., on node deletion). */ diff --git a/backend/src/utils/wsUrl.ts b/backend/src/utils/wsUrl.ts new file mode 100644 index 00000000..53e778fc --- /dev/null +++ b/backend/src/utils/wsUrl.ts @@ -0,0 +1,15 @@ +/** + * Convert an HTTP(S) base URL to a WebSocket URL by upgrading the scheme. + * Maps `http://` to `ws://` and `https://` to `wss://`; trims a trailing + * slash so the caller can append a path. Other schemes pass through + * unchanged because the caller may already have a `ws://` or `wss://` + * URL. + * + * Existed previously as inline regex replacements in several places that + * either dropped the trailing slash unevenly or used `replace(/^http/, 'ws')` + * which silently downgrades `https://` to `ws://` (cleartext). This helper + * is the single correct form. + */ +export function httpUrlToWs(baseUrl: string): string { + return baseUrl.replace(/\/$/, '').replace(/^https?/, (m) => (m === 'https' ? 'wss' : 'ws')); +} diff --git a/backend/src/websocket/meshProxyTunnel.ts b/backend/src/websocket/meshProxyTunnel.ts new file mode 100644 index 00000000..e595347c --- /dev/null +++ b/backend/src/websocket/meshProxyTunnel.ts @@ -0,0 +1,137 @@ +import type { IncomingMessage } from 'http'; +import type { Duplex } from 'stream'; +import { WebSocketServer, type WebSocket } from 'ws'; +import { MAX_FRAME_SIZE_BYTES, decodeBinaryFrame, decodeJsonFrame, wsDataToBuffer, wsDataToString } from '../pilot/protocol'; +import { + TcpStreamSwitchboard, + attachTcpStreamSwitchboard, + resolveByComposeLabels, + type ReverseTcpStreamHandle, +} from '../mesh/tcpStreamSwitchboard'; +import { sanitizeForLog } from '../utils/safeLog'; +import { isDebugEnabled } from '../utils/debug'; +import { rejectUpgrade as reject } from './reject'; + +/** + * Mesh proxy-tunnel ingress. + * + * The remote side of a Phase C proxy-mode mesh tunnel. Central dials + * `WSS /api/mesh/proxy-tunnel` using the long-lived `api_token` + * as a Bearer credential; this handler upgrades the connection and wires + * the shared `TcpStreamSwitchboard` to handle `tcp_open` / `tcp_open_ack` + * / `tcp_open_reverse` / `tcp_close` and `TcpData` frames. + * + * Auth + scope gating happens in `upgradeHandler.ts` before this handler + * runs (require `full-admin` api_token scope). The handler itself trusts + * the upgrade; the WS credential is the only trust boundary. + * + * Bidirectional: when the tunnel opens, the handler registers itself as + * the reverse-dialer on the local MeshService so meshed containers on + * this Sencho can dial cross-node aliases via `tcp_open_reverse` over + * the same WS. On disconnect the reverse-dialer registration is + * cleared. + */ +const wss = new WebSocketServer({ noServer: true, maxPayload: MAX_FRAME_SIZE_BYTES }); + +interface SwitchboardReverseDialer { + openMeshTcpStream(target: { nodeId: number; stack: string; service: string; port: number }): ReverseTcpStreamHandle | null; +} + +export async function handleMeshProxyTunnel(req: IncomingMessage, socket: Duplex, head: Buffer): Promise { + // Mesh service is only available on the central deployment (SENCHO_MODE + // unset or 'central'). Pilot-mode Sencho receives mesh traffic via the + // pilot tunnel and has no use for the proxy-mode WS path. + if (process.env.SENCHO_MODE === 'pilot') { + return reject(socket, 404, 'Not Found'); + } + + await new Promise((resolve) => { + wss.handleUpgrade(req, socket as Parameters[1], head, (ws) => { + void attachSwitchboard(ws).finally(resolve); + }); + }); +} + +async function attachSwitchboard(ws: WebSocket): Promise { + let switchboard: TcpStreamSwitchboard | null = null; + let meshServiceCleanup: (() => void) | null = null; + + try { + switchboard = attachTcpStreamSwitchboard({ + ws, + resolveTarget: resolveByComposeLabels, + logLabel: 'MeshProxy', + }); + + // Register a reverse dialer so this side's MeshForwarder can dial + // cross-node aliases via `tcp_open_reverse` over the same WS. The + // CAS swap refuses to overwrite a dialer that another caller + // (a concurrent proxy-tunnel upgrade, or a pilot agent in a + // misconfigured deployment) has already installed. + const { MeshService } = await import('../services/MeshService'); + const meshService = MeshService.getInstance(); + const localSwitchboard = switchboard; + const localDialer: SwitchboardReverseDialer = { + openMeshTcpStream(target) { + return localSwitchboard.openReverseStream(target); + }, + }; + const installed = meshService.setReverseDialer(localDialer, null); + if (!installed) { + console.warn('[MeshProxy] reverse dialer already installed; rejecting concurrent tunnel'); + try { ws.close(1013, 'reverse dialer already installed'); } catch { /* ignore */ } + switchboard.cleanup('reverse dialer already installed'); + switchboard = null; + return; + } + meshServiceCleanup = () => { + meshService.setReverseDialer(null, localDialer); + }; + } catch (err) { + if (isDebugEnabled()) { + console.warn('[MeshProxy:diag] failed to attach switchboard:', sanitizeForLog((err as Error).message)); + } + try { ws.close(1011, 'switchboard attach failed'); } catch { /* ignore */ } + return; + } + + const onMessage = (data: unknown, isBinary: boolean): void => { + if (!switchboard) return; + try { + if (isBinary) { + const buf = wsDataToBuffer(data); + if (!buf) return; + switchboard.handleBinaryFrame(decodeBinaryFrame(buf)); + } else { + const text = wsDataToString(data); + if (text == null) return; + switchboard.handleJsonFrame(decodeJsonFrame(text)); + } + } catch (err) { + if (isDebugEnabled()) { + console.warn('[MeshProxy:diag] malformed frame:', sanitizeForLog((err as Error).message)); + } + } + }; + + const teardown = (): void => { + ws.off('message', onMessage); + if (switchboard) { + switchboard.cleanup('mesh proxy-tunnel closed'); + switchboard = null; + } + if (meshServiceCleanup) { + meshServiceCleanup(); + meshServiceCleanup = null; + } + }; + + ws.on('message', onMessage); + ws.once('close', teardown); + ws.once('error', (err) => { + if (isDebugEnabled()) { + console.warn('[MeshProxy:diag] ws error:', sanitizeForLog(err.message)); + } + teardown(); + }); +} diff --git a/backend/src/websocket/upgradeHandler.ts b/backend/src/websocket/upgradeHandler.ts index 256466a0..a32cbcb4 100644 --- a/backend/src/websocket/upgradeHandler.ts +++ b/backend/src/websocket/upgradeHandler.ts @@ -7,6 +7,7 @@ import { DatabaseService, type UserRole } from '../services/DatabaseService'; import { NodeRegistry } from '../services/NodeRegistry'; import { COOKIE_NAME } from '../helpers/constants'; import { handlePilotTunnel } from './pilotTunnel'; +import { handleMeshProxyTunnel } from './meshProxyTunnel'; import { handleNotificationsWs } from './notifications'; import { handleRemoteForwarder } from './remoteForwarder'; import { handleLogsWs } from './logs'; @@ -32,11 +33,12 @@ function parseCookies(req: IncomingMessage): Record { * 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. `/ws/notifications` local -> handleNotificationsWs - * 5. remote nodeId path -> handleRemoteForwarder - * 6. `/api/stacks/:name/logs` -> handleLogsWs - * 7. `/api/system/host-console` -> handleHostConsoleWs - * 8. fallback -> handleGenericWs (`/ws` exec + stats) + * 4. `/api/mesh/proxy-tunnel` -> handleMeshProxyTunnel (requires full-admin api_token scope) + * 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) */ export function attachUpgrade( server: http.Server, @@ -120,6 +122,18 @@ export function attachUpgrade( } } + // Mesh proxy-tunnel ingress: a sibling Sencho is dialing this node + // to carry mesh TCP traffic. Require an api_token Bearer with the + // full-admin scope; mesh manipulates traffic and must not be + // reachable under a session cookie or a node_proxy JWT. + if (pathname === '/api/mesh/proxy-tunnel') { + if (wsApiTokenScope !== 'full-admin') { + return reject(socket, 403, 'Forbidden'); + } + await handleMeshProxyTunnel(req, socket, head); + return; + } + const nodeIdParam = parsedUrl.searchParams.get('nodeId'); const nodeId = nodeIdParam ? parseInt(nodeIdParam, 10) : NodeRegistry.getInstance().getDefaultNodeId(); const node = NodeRegistry.getInstance().getNode(nodeId); diff --git a/docs/features/sencho-mesh.mdx b/docs/features/sencho-mesh.mdx index 42d8895a..bd4ac3c4 100644 --- a/docs/features/sencho-mesh.mdx +++ b/docs/features/sencho-mesh.mdx @@ -1,22 +1,23 @@ --- title: Sencho Mesh -description: Connect containers across nodes by hostname over the Pilot tunnel, no VPN, no firewall changes, no extra ports. +description: Connect containers across nodes by hostname over an authenticated WebSocket tunnel, no VPN, no firewall changes, no extra ports. --- Sencho Mesh requires a Sencho **Admiral** license. Skipper and Community Edition do not include this feature. -Sencho Mesh makes a multi-node fleet feel like one machine. Opt a stack into the mesh and its services become reachable from any other meshed stack on the fleet by a stable hostname. Traffic rides the existing Pilot tunnel, so there are no new ports to open and no separate VPN to manage. +Sencho Mesh makes a multi-node fleet feel like one machine. Opt a stack into the mesh and its services become reachable from any other meshed stack on the fleet by a stable hostname. Traffic rides an authenticated WebSocket tunnel between Sencho instances, so there are no new ports to open and no separate VPN to manage. + +Mesh works with any remote mode: + +- **Pilot Agent** nodes carry mesh traffic over the long-lived agent tunnel. +- **Distributed API** nodes carry mesh traffic over a short-lived tunnel that central opens on demand using the node's API token. Central tears the tunnel down after five minutes of idle, so a node that sees no mesh traffic costs nothing extra to keep configured. - Mesh rides on the Pilot tunnel. See Pilot Agent for how to enroll a node into your fleet. + Pilot Agent is the easiest remote mode for nodes that cannot expose an inbound port. See Pilot Agent for how to enroll a node into your fleet. - - Mesh today requires nodes that connect to your primary in **Pilot Agent** mode. Support for Distributed API (proxy-mode) remotes is on the roadmap. - - ## How it works Each Sencho instance creates an internal Docker bridge network called `sencho_mesh` (default subnet `172.30.0.0/24`) on first boot and pins itself to a static IP on that network. When you opt a stack into the mesh, Sencho: @@ -47,7 +48,7 @@ Four guarantees: 1. **Only opted-in services are reachable.** A stack reaches another stack's services through the mesh only if both stacks have explicitly opted in. The Pilot agent on the target node refuses any request for a non-opted service. 2. **Aliases are not internet-reachable.** Sencho listens on an internal Docker network; nothing about the mesh exposes new ports beyond the host's existing firewall posture. -3. **Traffic is encrypted in transit.** Cross-node bytes ride the existing Pilot WSS tunnel. +3. **Traffic is encrypted in transit.** Cross-node bytes ride an authenticated WSS tunnel between Sencho instances. Distributed API nodes additionally require an API token with the **full-admin** scope; tokens with narrower scopes cannot carry mesh traffic. 4. **Tier-gated and audit-logged.** Only Admiral users can configure the mesh. Enable, disable, opt-in, and opt-out events write durable rows to the audit log with the actor's identity. What the mesh does **not** do: @@ -99,7 +100,7 @@ A few things are deliberately out of scope for the first release: - **One alias per TCP port across the fleet.** If two stacks expose the same port (for example two Postgres instances on 5432), only the first can be added to the mesh. The opt-in sheet shows a clear inline error if the second tries. - **Sencho's API port is reserved.** A meshed service exposing port 1852 is rejected at opt-in to prevent collision with the Sencho UI / API listener. -- **Pilot-to-pilot routing rides through central.** Mesh works for traffic between the central node and its pilots in either direction, and pilot-to-pilot via central relay. Direct peer-to-peer pilot tunnels are not supported. +- **Remote-to-remote routing rides through central.** Mesh works for traffic between central and any remote in either direction, and between two remotes via central relay. Direct peer-to-peer tunnels between remotes are not supported. - **Stream pool shared with the Pilot tunnel.** Each Pilot tunnel multiplexes up to 1024 concurrent streams covering HTTP, WebSocket, and mesh TCP traffic. A heavy mesh workload counts against the same ceiling as ordinary fleet API traffic. See the [Pilot Agent](/features/pilot-agent) resource-limit notes for the full picture. - **No TLS termination, no blue/green cutover.** Layer 7 features land in a follow-up. @@ -137,4 +138,13 @@ A few things are deliberately out of scope for the first release: `172.30.0.0/24` is the default range for the `sencho_mesh` network and is rare in most setups. If it collides with an existing VPN, VLAN, or Docker network, set `SENCHO_MESH_SUBNET` on the Sencho service to a free `/24` and restart the container. Each node can be configured independently. + + + Central could not open a mesh tunnel to this node. The badge tooltip shows the specific reason. While a node is in this state the **mesh toggle and Add stack to mesh action on the node card are disabled** so a redeploy is not triggered against an unreachable target. Common causes: + + - `api token rejected (scope must be full-admin)` — the API token configured for this node is restricted. Open the remote Sencho, generate a token with the **full-admin** scope, and update the node's credentials in **Settings → Nodes**. + - `remote does not support proxy mesh` — the remote Sencho is on a version that predates proxy-mode mesh. Update the remote and the badge clears on the next refresh. + - `TLS handshake failed` — the remote serves a certificate Node's default trust store does not accept. Use a certificate issued by a trusted authority on the remote. + - `api_url not set` or `api token missing` — the node was added without credentials. Edit the node in **Settings → Nodes** and supply the URL and token. + diff --git a/frontend/src/components/NodeManager.tsx b/frontend/src/components/NodeManager.tsx index f76171ff..e8660591 100644 --- a/frontend/src/components/NodeManager.tsx +++ b/frontend/src/components/NodeManager.tsx @@ -356,7 +356,7 @@ export function NodeManager() { ]} />

- Pilot Agent requires only outbound HTTPS from the remote host. Distributed API Proxy requires the remote host to expose an inbound port. + Pilot Agent requires only outbound HTTPS from the remote host. Distributed API Proxy requires the remote host to expose an inbound port. Sencho Mesh works with both modes.

)} diff --git a/frontend/src/components/fleet/RoutingNodeCard.tsx b/frontend/src/components/fleet/RoutingNodeCard.tsx index 47f207d4..13ef27a8 100644 --- a/frontend/src/components/fleet/RoutingNodeCard.tsx +++ b/frontend/src/components/fleet/RoutingNodeCard.tsx @@ -11,7 +11,6 @@ import { meshRouteStateFor, meshRouteStateTokens } from './meshRouteState'; interface Props { status: MeshNodeStatus; aliases: MeshAlias[]; - isLocal: boolean; onAddStack: () => void; onShowDiagnostics: () => void; onShowAlias: (alias: string) => void; @@ -20,7 +19,7 @@ interface Props { } export function RoutingNodeCard({ - status, aliases, isLocal, onAddStack, onShowDiagnostics, onShowAlias, onTestUpstream, onChanged, + status, aliases, onAddStack, onShowDiagnostics, onShowAlias, onTestUpstream, onChanged, }: Props) { const [toggling, setToggling] = useState(false); const [testingAlias, setTestingAlias] = useState(null); @@ -55,21 +54,29 @@ export function RoutingNodeCard({
{status.nodeName} - {isLocal && ( + {status.reachableMode === 'local' && ( ★ Local )} - {!status.pilotConnected && status.nodeId !== -1 && !isLocal && ( + {status.reachableMode === 'pilot' && !status.pilotConnected && ( pilot offline )} + {status.reachableMode === 'unreachable' && ( + + unreachable + + )}
{ void toggleEnabled(next); }} />
- {!status.pilotConnected && !isLocal && ( + {status.reachableMode === 'unreachable' && status.reachableReason && ( +
+ {status.reachableReason} +
+ )} + {status.reachableMode === 'pilot' && !status.pilotConnected && (
- Sencho Mesh requires the pilot agent on this node. + Pilot tunnel is not connected. Mesh traffic resumes when the agent reconnects.
)} @@ -129,7 +141,11 @@ export function RoutingNodeCard({ ); })} - diff --git a/frontend/src/components/fleet/RoutingTab.tsx b/frontend/src/components/fleet/RoutingTab.tsx index 9a118462..2fbcc7d0 100644 --- a/frontend/src/components/fleet/RoutingTab.tsx +++ b/frontend/src/components/fleet/RoutingTab.tsx @@ -60,7 +60,14 @@ export function RoutingTab() { const totalAliases = aliases.length; const meshedNodes = status.filter((s) => s.enabled).length; - const onlineNodes = status.filter((s) => s.pilotConnected).length; + // A node is "reachable for mesh" when it is local, a pilot with an + // active tunnel, or a Distributed API remote with valid credentials + // (central can dial the on-demand proxy tunnel as needed). + const reachableNodes = status.filter((s) => ( + s.reachableMode === 'local' + || (s.reachableMode === 'pilot' && s.pilotConnected) + || s.reachableMode === 'proxy' + )).length; if (loading) { return ( @@ -86,7 +93,7 @@ export function RoutingTab() { if (meshedNodes === 0) { return (
- setActivityOpen(true)} /> + setActivityOpen(true)} />
Mesh containers across nodes
@@ -100,7 +107,6 @@ export function RoutingTab() { key={s.nodeId} status={s} aliases={aliases} - isLocal={s.nodeId === status[0]?.nodeId && s.pilotConnected} onAddStack={() => setOptInNode({ id: s.nodeId, name: s.nodeName })} onShowDiagnostics={() => setDiagnosticsNode({ id: s.nodeId, name: s.nodeName })} onShowAlias={(alias) => setRouteDetailAlias(alias)} @@ -123,14 +129,13 @@ export function RoutingTab() { return (
- setActivityOpen(true)} /> + setActivityOpen(true)} />
{status.map((s) => ( setOptInNode({ id: s.nodeId, name: s.nodeName })} onShowDiagnostics={() => setDiagnosticsNode({ id: s.nodeId, name: s.nodeName })} onShowAlias={(alias) => setRouteDetailAlias(alias)} @@ -150,10 +155,10 @@ export function RoutingTab() { ); } -function RoutingMasthead({ meshedNodes, onlineNodes, totalAliases, onShowActivity }: { - meshedNodes: number; onlineNodes: number; totalAliases: number; onShowActivity: () => void; +function RoutingMasthead({ meshedNodes, reachableNodes, totalAliases, onShowActivity }: { + meshedNodes: number; reachableNodes: number; totalAliases: number; onShowActivity: () => void; }) { - const stateWord = meshedNodes === 0 ? 'unmeshed' : meshedNodes < onlineNodes ? 'partial' : 'meshed'; + const stateWord = meshedNodes === 0 ? 'unmeshed' : meshedNodes < reachableNodes ? 'partial' : 'meshed'; return (
@@ -161,7 +166,7 @@ function RoutingMasthead({ meshedNodes, onlineNodes, totalAliases, onShowActivit
meshed
-
{meshedNodes}/{onlineNodes}
+
{meshedNodes}/{reachableNodes}
aliases
diff --git a/frontend/src/types/mesh.ts b/frontend/src/types/mesh.ts index 8dad9e01..729ee0d1 100644 --- a/frontend/src/types/mesh.ts +++ b/frontend/src/types/mesh.ts @@ -9,13 +9,20 @@ export interface MeshAlias { port: number; } +export type MeshReachableMode = 'local' | 'pilot' | 'proxy' | 'unreachable'; + export interface MeshNodeStatus { nodeId: number; nodeName: string; enabled: boolean; - /** Forwarder state for the LOCAL Sencho instance only. `null` for non-local nodes — cross-node forwarder state lands in Phase B. */ + /** Forwarder state for the LOCAL Sencho instance only. `null` for non-local nodes; cross-node forwarder state lands in Phase B. */ localForwarderListening: boolean | null; + /** True iff a pilot tunnel is currently registered. Only meaningful when `reachableMode === 'pilot'`. */ pilotConnected: boolean; + /** How this node participates in mesh routing. Drives Routing tab badge state. */ + reachableMode: MeshReachableMode; + /** Operator-facing reason when `reachableMode === 'unreachable'`. Null otherwise. */ + reachableReason: string | null; optedInStacks: string[]; activeStreamCount: number; }