diff --git a/backend/src/__tests__/mesh-service.test.ts b/backend/src/__tests__/mesh-service.test.ts index 303a0a86..62aa77ec 100644 --- a/backend/src/__tests__/mesh-service.test.ts +++ b/backend/src/__tests__/mesh-service.test.ts @@ -1303,6 +1303,77 @@ describe('MeshService.openCrossNode (BUG-4)', () => { expect(activeStreams.size).toBe(0); }); + + // Pins the ordering contract for protocols that send immediately after + // connect (HTTP, TLS, Redis, Postgres): src bytes arriving between + // dial and tcp_open_ack must reach the upstream in order, ahead of any + // post-ack writes. Without local buffering the first packet would race + // the ack on the wire and land on a stream the agent has not yet dialed. + it('buffers src bytes until tcpStream opens, then flushes them in order before any post-open writes', async () => { + const svc = MeshService.getInstance(); + const target: MeshTarget = { + nodeId: 14, stack: 'audit-mesh-pilot', service: 'echo', + port: 9001, alias: 'echo.audit-mesh-pilot.sencho-pilot-test.sencho', + }; + const fakeStream = makeFakeStream(60); + vi.spyOn( + svc as unknown as { dialMeshTcpStream: (t: MeshTarget) => MeshTcpStreamLike | null }, + 'dialMeshTcpStream', + ).mockReturnValue(fakeStream); + + const fakeSrc = makeEmittingSocket(); + await (svc as unknown as { openCrossNode: (t: MeshTarget, s: unknown) => Promise }) + .openCrossNode(target, fakeSrc); + + // Push data before the stream is open. Nothing should be written + // through to the tunnel. + fakeSrc.emit('data', Buffer.from('GET /healthz HTTP/1.1\r\n')); + fakeSrc.emit('data', Buffer.from('Host: echo.local\r\n\r\n')); + expect(fakeStream.write).not.toHaveBeenCalled(); + + // Open the stream. The pre-open buffer must flush in order, ahead + // of any post-open writes. + fakeStream.emit('open'); + + const writeCalls = (fakeStream.write as unknown as { mock: { calls: Buffer[][] } }).mock.calls; + expect(writeCalls.length).toBe(2); + expect(writeCalls[0][0].toString()).toBe('GET /healthz HTTP/1.1\r\n'); + expect(writeCalls[1][0].toString()).toBe('Host: echo.local\r\n\r\n'); + + // Post-open data writes through directly (still in order). + fakeSrc.emit('data', Buffer.from('post-open')); + expect(writeCalls.length).toBe(3); + expect(writeCalls[2][0].toString()).toBe('post-open'); + + // Cleanup so the open-timer does not keep the worker alive. + fakeStream.emit('close'); + }); + + it('tears down both sockets when buffered bytes exceed STREAM_PENDING_DATA_MAX_BYTES before open', async () => { + const { STREAM_PENDING_DATA_MAX_BYTES } = await import('../pilot/protocol'); + const svc = MeshService.getInstance(); + const target: MeshTarget = { + nodeId: 14, stack: 'audit-mesh-pilot', service: 'echo', + port: 9001, alias: 'echo.audit-mesh-pilot.sencho-pilot-test.sencho', + }; + const fakeStream = makeFakeStream(61); + vi.spyOn( + svc as unknown as { dialMeshTcpStream: (t: MeshTarget) => MeshTcpStreamLike | null }, + 'dialMeshTcpStream', + ).mockReturnValue(fakeStream); + + const fakeSrc = makeEmittingSocket(); + await (svc as unknown as { openCrossNode: (t: MeshTarget, s: unknown) => Promise }) + .openCrossNode(target, fakeSrc); + + // Overflow the per-stream cap in one shot. The handler must destroy + // both sockets and not call write on the stream. + fakeSrc.emit('data', Buffer.alloc(STREAM_PENDING_DATA_MAX_BYTES + 1)); + + expect(fakeStream.write).not.toHaveBeenCalled(); + expect(fakeSrc.destroy).toHaveBeenCalled(); + expect(fakeStream.destroy).toHaveBeenCalled(); + }); }); describe('MeshService.openCrossNode without reverseDialer', () => { diff --git a/backend/src/__tests__/pilot-close-tunnel.test.ts b/backend/src/__tests__/pilot-close-tunnel.test.ts new file mode 100644 index 00000000..097e3141 --- /dev/null +++ b/backend/src/__tests__/pilot-close-tunnel.test.ts @@ -0,0 +1,117 @@ +/** + * Regression for PilotTunnelManager.closeTunnel lifecycle parity. + * + * The natural-disconnect path (bridge's `'closed'` event) writes + * `nodes.status='offline'` and emits `tunnel-down` for pilot bridges, or + * emits `proxy-bridge-down` for proxy bridges. Explicit `closeTunnel` calls + * (enrollment regenerate, node deletion) must run the same cleanup so the + * UI does not keep showing a Pilot node as Online after the operator has + * intentionally torn its session down. + */ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { EventEmitter } from 'events'; +import { WebSocket } from 'ws'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; +import { PilotTunnelManager } from '../services/PilotTunnelManager'; + +let tmpDir: string; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService } = await import('../services/DatabaseService')); +}); + +afterAll(() => cleanupTestDb(tmpDir)); + +function makeMockTunnelWs(): EventEmitter & { + sent: unknown[]; + readyState: number; + bufferedAmount: number; + send: (data: unknown) => void; + ping: () => void; + close: () => void; +} { + const ws = new EventEmitter() as EventEmitter & { + sent: unknown[]; + readyState: number; + bufferedAmount: number; + send: (data: unknown) => void; + ping: () => void; + close: () => void; + }; + ws.sent = []; + ws.readyState = WebSocket.OPEN; + ws.bufferedAmount = 0; + ws.send = (data: unknown) => { ws.sent.push(data); }; + ws.ping = () => { /* no-op */ }; + ws.close = () => { ws.readyState = WebSocket.CLOSED; ws.emit('close'); }; + return ws; +} + +describe('PilotTunnelManager.closeTunnel lifecycle parity', () => { + it('marks the node offline and emits tunnel-down when closing a pilot bridge', async () => { + const mgr = PilotTunnelManager.getInstance(); + const nodeId = DatabaseService.getInstance().addNode({ + name: `pilot-close-${Date.now()}`, + type: 'remote', + mode: 'pilot_agent', + compose_dir: '/tmp/x', + is_default: false, + api_url: '', + api_token: '', + }); + + const ws = makeMockTunnelWs(); + await mgr.registerTunnel(nodeId, ws as unknown as WebSocket, 'test-1.0.0'); + + // Confirm the registration write actually landed before we measure + // the close-side delta; otherwise the assertion would pass for the + // wrong reason on a fresh node that defaulted to status=null. + expect(DatabaseService.getInstance().getNode(nodeId)?.status).toBe('online'); + + let tunnelDownNodeId: number | null = null; + const onTunnelDown = (id: number): void => { tunnelDownNodeId = id; }; + mgr.once('tunnel-down', onTunnelDown); + + mgr.closeTunnel(nodeId); + + expect(DatabaseService.getInstance().getNode(nodeId)?.status).toBe('offline'); + expect(tunnelDownNodeId).toBe(nodeId); + expect(mgr.hasActiveTunnel(nodeId)).toBe(false); + }); + + it('does not double-emit tunnel-down when the bridge close fires after closeTunnel', async () => { + const mgr = PilotTunnelManager.getInstance(); + const nodeId = DatabaseService.getInstance().addNode({ + name: `pilot-no-double-${Date.now()}`, + type: 'remote', + mode: 'pilot_agent', + compose_dir: '/tmp/x', + is_default: false, + api_url: '', + api_token: '', + }); + + const ws = makeMockTunnelWs(); + await mgr.registerTunnel(nodeId, ws as unknown as WebSocket, 'test-1.0.0'); + + let emitCount = 0; + const onTunnelDown = (id: number): void => { + if (id === nodeId) emitCount += 1; + }; + mgr.on('tunnel-down', onTunnelDown); + + // closeTunnel deletes the map entry, then calls bridge.close() which + // synchronously emits 'closed' on the mock ws. The bridge's + // `bridges.get(nodeId) === bridge` check inside its 'closed' handler + // must short-circuit because we already deleted the entry, so this + // produces exactly one tunnel-down emission. + mgr.closeTunnel(nodeId); + // Yield a tick for any deferred listeners. + await new Promise((r) => setTimeout(r, 10)); + + mgr.off('tunnel-down', onTunnelDown); + expect(emitCount).toBe(1); + }); +}); diff --git a/backend/src/__tests__/upgrade-order.test.ts b/backend/src/__tests__/upgrade-order.test.ts index c5a729ae..b2088ecd 100644 --- a/backend/src/__tests__/upgrade-order.test.ts +++ b/backend/src/__tests__/upgrade-order.test.ts @@ -342,6 +342,45 @@ describe('WebSocket upgrade dispatch order', () => { }); }); + describe('remote dispatch via NodeRegistry.getProxyTarget', () => { + // Pilot-mode nodes carry empty `api_url` and `api_token` by design and + // expose their API on a per-tunnel loopback bridge that + // NodeRegistry.getProxyTarget resolves. The dispatch ladder must route + // through that abstraction so pilot and proxy modes share one code + // path, and must reject upgrades whose target is unresolvable rather + // than serving gateway-local data for a request that named a remote + // node. + + let pilotNodeId: number; + + beforeAll(async () => { + const { DatabaseService } = await import('../services/DatabaseService'); + pilotNodeId = DatabaseService.getInstance().addNode({ + name: `upgrade-order-pilot-${Date.now()}`, + type: 'remote', + mode: 'pilot_agent', + compose_dir: '/tmp/x', + is_default: false, + api_url: '', + api_token: '', + }); + }); + + it('rejects a WS upgrade to a pilot-mode node with no active tunnel with HTTP 503 (not a fall-through to local handlers)', async () => { + const ws = connect(`/api/stacks/anything/logs?nodeId=${pilotNodeId}`, { cookie: sessionCookie }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('unexpected'); + if (outcome.kind === 'unexpected') expect(outcome.status).toBe(503); + }); + + it('rejects /ws?nodeId= with HTTP 503 instead of dispatching to the local generic handler', async () => { + const ws = connect(`/ws?nodeId=${pilotNodeId}`, { cookie: sessionCookie }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('unexpected'); + if (outcome.kind === 'unexpected') expect(outcome.status).toBe(503); + }); + }); + it('dispatches /api/pilot/tunnel to the pilot handler (rejects non-pilot bearer before path-based dispatch)', async () => { // A plain session cookie is a valid *user* JWT but not a pilot JWT. The // pilot handler runs first (before the shared cookie/Bearer auth) and diff --git a/backend/src/services/MeshService.ts b/backend/src/services/MeshService.ts index 6a577a4d..ae5862f1 100644 --- a/backend/src/services/MeshService.ts +++ b/backend/src/services/MeshService.ts @@ -15,6 +15,7 @@ import { PilotTunnelManager } from './PilotTunnelManager'; import { MeshProxyTunnelDialer, type DialFailureCode } from './MeshProxyTunnelDialer'; import { generateOverrideYaml, MeshAlias, SENCHO_MESH_NETWORK } from './MeshComposeOverride'; import { lookupContainerIp } from '../mesh/containerLookup'; +import { STREAM_PENDING_DATA_MAX_BYTES } from '../pilot/protocol'; import { sanitizeForLog } from '../utils/safeLog'; import { isPathWithinBase, isValidStackName } from '../utils/validation'; import { PORT as SENCHO_LISTEN_PORT } from '../helpers/constants'; @@ -1893,6 +1894,15 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { const record = this.registerActiveStream(target.alias, tcpStream.streamId); const t0 = Date.now(); + // Hold src bytes until tcp_open_ack arrives. Writing through to the + // tunnel before 'open' fires races the first packet ahead of the + // ack, which breaks protocols that send immediately after connect + // (HTTP, TLS, Redis, Postgres). Cap matches the bridge's reservation + // cap so a misbehaving source cannot exhaust gateway memory. + let tcpOpen = false; + const pending: Buffer[] = []; + let pendingBytes = 0; + // Timer guards against the agent never returning a tcp_open_ack // (broken pilot, frame dropped, dial stuck after handshake). // Without this the stream sits forever and the operator sees @@ -1929,6 +1939,14 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { message: `cross-node connect to ${target.alias}`, }); this.routeLatencyMap.set(target.alias, Date.now() - t0); + // Flush any bytes that arrived before tcp_open_ack so the upstream + // sees them in order, ahead of anything that lands post-ack. + for (const buf of pending) { + try { tcpStream.write(buf); } catch { /* ignore */ } + } + pending.length = 0; + pendingBytes = 0; + tcpOpen = true; }); tcpStream.on('data', (chunk: Buffer) => { record.bytesIn += chunk.length; @@ -1949,7 +1967,17 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { }); src.on('data', (chunk: Buffer) => { record.bytesOut += chunk.length; - tcpStream.write(chunk); + if (tcpOpen) { + tcpStream.write(chunk); + return; + } + if (pendingBytes + chunk.length > STREAM_PENDING_DATA_MAX_BYTES) { + try { src.destroy(); } catch { /* ignore */ } + try { tcpStream.destroy(); } catch { /* ignore */ } + return; + } + pending.push(Buffer.from(chunk)); + pendingBytes += chunk.length; }); src.on('end', () => tcpStream.end()); src.on('close', () => { cleanupRecord(); try { tcpStream.destroy(); } catch { /* ignore */ } }); diff --git a/backend/src/services/PilotTunnelManager.ts b/backend/src/services/PilotTunnelManager.ts index 36d89b37..cea8888e 100644 --- a/backend/src/services/PilotTunnelManager.ts +++ b/backend/src/services/PilotTunnelManager.ts @@ -295,14 +295,28 @@ export class PilotTunnelManager extends EventEmitter { } /** - * Force-close a tunnel (e.g., on node deletion). + * Force-close a tunnel (e.g., on node deletion, enrollment regenerate). + * + * Mirrors the cleanup the bridge's natural `'closed'` event handler runs + * so explicit closure and natural disconnect produce identical state: + * pilot tunnels write `nodes.status='offline'` and emit `tunnel-down`; + * proxy bridges emit `proxy-bridge-down`. The maps are cleared before + * `bridge.close()` so the natural handler's `=== bridge` check + * short-circuits and we do not double-emit. */ public closeTunnel(nodeId: number, code = 1000, reason = 'closed by primary'): void { const bridge = this.bridges.get(nodeId); if (!bridge) return; - bridge.close(code, reason); + const kind = this.bridgeKinds.get(nodeId); this.bridges.delete(nodeId); this.bridgeKinds.delete(nodeId); + if (kind === 'pilot') { + DatabaseService.getInstance().updateNodeStatus(nodeId, 'offline'); + this.emit('tunnel-down', nodeId); + } else if (kind === 'proxy') { + this.emit('proxy-bridge-down', nodeId); + } + bridge.close(code, reason); } } diff --git a/backend/src/websocket/remoteForwarder.ts b/backend/src/websocket/remoteForwarder.ts index 4e1388b5..8b7a28fd 100644 --- a/backend/src/websocket/remoteForwarder.ts +++ b/backend/src/websocket/remoteForwarder.ts @@ -1,6 +1,5 @@ import type { IncomingMessage } from 'http'; import type { Duplex } from 'stream'; -import type { Node } from '../services/DatabaseService'; import { PROXY_TIER_HEADER, PROXY_VARIANT_HEADER } from '../services/license-headers'; import { LicenseService } from '../services/LicenseService'; import { wsProxyServer } from '../proxy/websocketProxy'; @@ -13,35 +12,41 @@ import { rejectUpgrade as reject } from './reject'; * api_token never reaches an interactive terminal (the remote's upgrade * handler rejects node_proxy tokens on those paths). * - * The caller must have already established that `node.type === 'remote'` - * and that `api_url` + `api_token` are present. + * Target resolution lives in the caller (parallel to the HTTP proxy in + * `proxy/remoteNodeProxy.ts`), so this function works uniformly for + * proxy-mode remotes (api_url + api_token) and pilot-mode remotes (loopback + * URL + empty token). When the target carries an empty token, the loopback + * bridge sits on 127.0.0.1 and demuxes onto an already-authenticated tunnel: + * we skip the console-token exchange and do not inject a Bearer. */ export async function handleRemoteForwarder( req: IncomingMessage, socket: Duplex, head: Buffer, - opts: { node: Node; pathname: string }, + opts: { pathname: string; target: { apiUrl: string; apiToken: string } }, ): Promise { - const { node, pathname } = opts; - // Guaranteed non-null by caller; assert here so the rest of the function is nullsafe. - if (!node.api_url || !node.api_token) return reject(socket, 503, 'Service Unavailable'); + const { pathname, target } = opts; + if (!target.apiUrl) return reject(socket, 503, 'Service Unavailable'); - const wsTarget = node.api_url.replace(/\/$/, '').replace(/^https?/, (m) => m === 'https' ? 'wss' : 'ws'); + const wsTarget = target.apiUrl.replace(/\/$/, '').replace(/^https?/, (m) => m === 'https' ? 'wss' : 'ws'); + const isPilotLoopback = target.apiToken === ''; // Interactive console paths (host console / container exec) are guarded on // the remote by an isProxyToken check that rejects the long-lived api_token. // Exchange it for a short-lived console_session token before forwarding so // the remote allows the connection while keeping the guard intact for - // direct api_token access. + // direct api_token access. Pilot loopback targets skip this: there is no + // long-lived api_token to exchange, and host-console is disabled on pilot + // mode at the capability registry anyway. const isInteractiveConsolePath = pathname === '/api/system/host-console' || pathname === '/ws'; - let bearerTokenForProxy = node.api_token; - if (isInteractiveConsolePath) { + let bearerTokenForProxy = target.apiToken; + if (isInteractiveConsolePath && !isPilotLoopback) { try { const consoleHeaders = LicenseService.getInstance().getProxyHeaders(); - const tokenRes = await fetch(`${node.api_url.replace(/\/$/, '')}/api/system/console-token`, { + const tokenRes = await fetch(`${target.apiUrl.replace(/\/$/, '')}/api/system/console-token`, { method: 'POST', headers: { - 'Authorization': `Bearer ${node.api_token}`, + 'Authorization': `Bearer ${target.apiToken}`, [PROXY_TIER_HEADER]: consoleHeaders.tier, [PROXY_VARIANT_HEADER]: consoleHeaders.variant || '', }, @@ -58,11 +63,18 @@ export async function handleRemoteForwarder( } } - req.headers['authorization'] = `Bearer ${bearerTokenForProxy}`; + if (isPilotLoopback) { + // The loopback bridge does not authenticate inbound traffic; an + // Authorization header inherited from the browser would only confuse + // the agent. Strip it explicitly. + delete req.headers['authorization']; + } else { + req.headers['authorization'] = `Bearer ${bearerTokenForProxy}`; + } delete req.headers['x-node-id']; // Strip the browser's session cookie: signed by this instance's JWT secret // and would fail verification on the remote. Auth is handled exclusively - // via the Bearer token. + // via the Bearer token (or, for pilot loopback, the tunnel itself). delete req.headers['cookie']; const fwdHeaders = LicenseService.getInstance().getProxyHeaders(); req.headers[PROXY_TIER_HEADER] = fwdHeaders.tier; diff --git a/backend/src/websocket/upgradeHandler.ts b/backend/src/websocket/upgradeHandler.ts index 3c74e340..d16809f2 100644 --- a/backend/src/websocket/upgradeHandler.ts +++ b/backend/src/websocket/upgradeHandler.ts @@ -182,8 +182,18 @@ export function attachUpgrade( return; } - if (node && node.type === 'remote' && node.api_url && node.api_token) { - await handleRemoteForwarder(req, socket, head, { node, pathname }); + if (node && node.type === 'remote') { + // Resolve the proxy target through NodeRegistry so pilot-mode nodes + // (empty api_url + api_token, loopback bridge instead) and proxy-mode + // nodes share one dispatch path. Mirrors proxy/remoteNodeProxy.ts. + const target = NodeRegistry.getInstance().getProxyTarget(nodeId); + if (!target) { + // Pilot tunnel disconnected, or proxy-mode node missing credentials. + // Reject the upgrade cleanly; falling through to local handlers would + // serve gateway-local data for a request that named a remote node. + return reject(socket, 503, 'Service Unavailable'); + } + await handleRemoteForwarder(req, socket, head, { pathname, target }); return; } diff --git a/docs/features/pilot-agent.mdx b/docs/features/pilot-agent.mdx index fb51151d..048ee087 100644 --- a/docs/features/pilot-agent.mdx +++ b/docs/features/pilot-agent.mdx @@ -49,7 +49,7 @@ Conceptually, the agent reverses the usual client/server direction. 3. For every request the control instance needs to make against that node (listing containers, deploying a stack, streaming logs, opening a console, forwarding mesh traffic), frames are multiplexed over that single connection. 4. On the agent side, those frames are demultiplexed and replayed locally against the host's Docker socket and filesystem. -Inside the control instance, the tunnel terminates at a **loopback bridge**: a tiny HTTP server on `127.0.0.1:` per active tunnel. Every other Sencho feature (the existing HTTP proxy, WebSocket forwarder, mesh dialer, license-tier propagation) treats the bridge as just another remote URL. That is why a pilot-mode node behaves identically to a proxy-mode node from the operator's point of view. Once enrolled, it shows up in the same Nodes table, the same node switcher, the same dashboard. +Inside the control instance, the tunnel terminates at a **loopback bridge**: a tiny HTTP server on `127.0.0.1:` per active tunnel. Every other Sencho feature (the existing HTTP proxy, WebSocket forwarder, mesh dialer, license-tier propagation) treats the bridge as just another remote URL. Once enrolled, the pilot node shows up in the same Nodes table, the same node switcher, and the same dashboard as a proxy-mode node, with the same shipped capabilities except where pilot mode intentionally narrows the surface (see below). Settings · System · Nodes table showing a Local row and a Pilot Agent row. The pilot row has Mode badge 'Pilot Agent', Endpoint 'tunnel (seen 45m ago)', and Status badge 'Online'. @@ -58,7 +58,7 @@ Inside the control instance, the tunnel terminates at a **loopback bridge**: a t What rides through the tunnel: - HTTP requests (every API call and stream). -- WebSockets (live logs, host console, container exec, notifications). +- WebSockets (live logs, container exec, notifications). - Sencho Mesh TCP streams (cross-node service-to-service traffic). What does NOT ride through the tunnel: @@ -67,6 +67,13 @@ What does NOT ride through the tunnel: - The agent serves no UI of its own. Operators always work from the control instance. - The agent's internal `/api/health` endpoint is reachable only from the tunnel's loopback bridge, not from outside the agent container. +### Intentional non-parity with proxy mode + +Pilot mode narrows the operator surface in a couple of places where parity with proxy mode would be misleading: + +- **Host console is not available on pilot nodes.** A pilot agent runs as a container with no useful shell into the underlying host, so the in-app host console is disabled at the capability level. If you need a host shell on a remote, use proxy mode for that node, or SSH in directly. +- **Self-update on pilot nodes recreates the agent container in place via the compose file the operator wrote during enrollment.** This is the same code path proxy-mode nodes use; it works as long as the host runs the compose file from the same directory across updates. + ## Prerequisites | Requirement | Detail |