From f5a52e44dc0a3e0647db08719818db6a6ea11a68 Mon Sep 17 00:00:00 2001 From: Anso Date: Thu, 7 May 2026 23:45:16 -0400 Subject: [PATCH] refactor(pilot): narrow Mesh handle, prune dead code, add replay-route test (#982) * refactor(pilot): narrow PilotTunnelManager.getBridge to MeshTunnelHandle The manager handed out the entire PilotTunnelBridge to its only consumer, MeshService, which let any current or future caller reach into transport internals: loopback URL, per-stream maps, the close API, the underscored _writeTcpData / _closeTcpStream, the diagnostic helpers. None of that was load-bearing for Mesh. Introduce a MeshTunnelHandle interface alongside TcpStream in PilotTunnelBridge.ts that exposes only the two methods Mesh actually calls (openTcpStream and getBufferedAmount), have PilotTunnelBridge declare implements MeshTunnelHandle, and change getBridge to return the interface. MeshService continues to compile unchanged because its existing call sites only touch the narrowed surface. A future alternative transport (a stub for tests, a different routing strategy) now has a one-method-and-a-getter contract to satisfy instead of the full bridge. * refactor(pilot): drop unused tunnel manager and bridge surface Three public methods predating the hardening pass had zero callers across the entire codebase (verified via grep across backend, frontend, e2e): - PilotTunnelManager.touch(nodeId): never invoked. The pilot_last_seen timestamp is updated by the manager itself on registerTunnel and by the persistence layer on heartbeat events. - PilotTunnelManager.listActive(): never invoked. The metrics endpoint added in PR #979 returns its own per-node breakdown via getMetricsSnapshot, which is the canonical observability surface. - PilotTunnelBridge.listTcpStreams() and the supporting TcpStreamSummary interface: never invoked. The Mesh diagnostics sheet that would have consumed it is not wired and would use getMetricsSnapshot if/when it ships. Removing them tightens the public surface and prevents accidental new dependencies on speculative future-proofing. * test(pilot): cover enrollment replay rejection at the route layer The DB-level test in pilot-enrollment.test.ts already verifies that consumePilotEnrollment is one-shot. That guards the persistence layer but not the route handler: a refactor of handlePilotTunnel that swaps the order of consume vs upgrade, or that grants the WebSocket upgrade before checking the consume result, would silently break the security invariant while DB-layer tests stay green. Drive handlePilotTunnel directly with a stub IncomingMessage and Duplex socket. Six cases: - Already-consumed enrollment row -> 401. - Token whose hash matches no row -> 401. - Row whose expires_at has passed -> 401. - Missing Authorization header -> 401. - JWT signed with a wrong secret -> 401. - pilot_tunnel JWT for an unknown node -> 404. The stub captures HTTP/1.1 status writes so the test asserts the exact rejection lands on the wire, not just that the function returned without throwing. * docs(debug): warn future committers off per-frame isDebugEnabled calls Code-review feedback on PR #979 noted that isDebugEnabled is fine in the cadences it has today (per-tunnel, per-request, error paths) but is fragile against a future commit that drops it into a per-frame WebSocket loop. The function does a try/catch + Node require cache lookup + a method call into DatabaseService on every invocation; acceptable at hundreds of calls per second, expensive at thousands. Add a comment block above the function spelling out the acceptable and unacceptable cadences and the snapshot-outside-the-loop mitigation, so the next person to add a diag log there sees the constraint. * fix(pilot): address PR A code-review findings Code review on this branch surfaced four items: - Em-dash directive (CLAUDE.md Directive 18) violations in four comment sites; replaced with colons or restructured. - debug.ts perf comment claimed try/catch frame setup as the cost driver. V8 inlines those; the load-bearing cost is the require lookup and singleton dispatch. Reword. - Test file header described coverage as 'replay rejection at the route layer' but the file now also covers missing-header, wrong-secret, expired-row, never-stored, and unknown-node rejection paths. Widen the JSDoc and the describe label to match. - Stub-cast comment in the replay test now explicitly lists the IncomingMessage and Duplex surface the stub satisfies, so a future commit that grows handlePilotTunnel's surface (rate limiting, socket options) updates both stubs instead of silently no-opping against them. No behavior change. --- .../__tests__/pilot-enrollment-replay.test.ts | 231 ++++++++++++++++++ backend/src/services/PilotTunnelBridge.ts | 36 ++- backend/src/services/PilotTunnelManager.ts | 28 +-- backend/src/utils/debug.ts | 12 + 4 files changed, 262 insertions(+), 45 deletions(-) create mode 100644 backend/src/__tests__/pilot-enrollment-replay.test.ts diff --git a/backend/src/__tests__/pilot-enrollment-replay.test.ts b/backend/src/__tests__/pilot-enrollment-replay.test.ts new file mode 100644 index 00000000..ec82eae1 --- /dev/null +++ b/backend/src/__tests__/pilot-enrollment-replay.test.ts @@ -0,0 +1,231 @@ +/** + * Route-level coverage for the rejection paths inside handlePilotTunnel. + * + * The headline invariant is "a pilot_enroll JWT is one-shot at the upgrade + * handler, not just at the DB layer". pilot-enrollment.test.ts already + * verifies the DB layer. That is necessary but not sufficient: a future + * refactor of handlePilotTunnel that re-orders mint and consume, or that + * grants the WebSocket upgrade before checking the consume result, would + * silently break the invariant while the DB-layer test stays green. + * + * This file drives handlePilotTunnel directly with a stub IncomingMessage + * and Duplex socket. Each test exercises a different early-return branch + * (missing header, wrong-secret JWT, never-stored row, expired row, + * already-consumed row, unknown node) and asserts the rejection lands on + * the wire with the right HTTP status before any upgrade attempt. + */ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { EventEmitter } from 'events'; +import type { IncomingMessage } from 'http'; +import type { Duplex } from 'stream'; +import crypto from 'crypto'; +import jwt from 'jsonwebtoken'; +import { WebSocketServer } from 'ws'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; +import { handlePilotTunnel } from '../websocket/pilotTunnel'; + +let tmpDir: string; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; +let pilotTunnelWss: WebSocketServer; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService } = await import('../services/DatabaseService')); + pilotTunnelWss = new WebSocketServer({ noServer: true }); +}); + +afterAll(() => { + pilotTunnelWss.close(); + cleanupTestDb(tmpDir); +}); + +interface StubSocket extends EventEmitter { + writes: string[]; + destroyed: boolean; + write(chunk: string): boolean; + destroy(): void; +} + +function makeStubSocket(): StubSocket { + const sock = new EventEmitter() as StubSocket; + sock.writes = []; + sock.destroyed = false; + sock.write = (chunk: string) => { sock.writes.push(chunk); return true; }; + sock.destroy = () => { sock.destroyed = true; }; + return sock; +} + +function makeStubReq(authHeader: string | undefined, agentVersion = 'test-1.0'): IncomingMessage { + const headers: Record = { 'x-sencho-agent-version': agentVersion }; + if (authHeader !== undefined) headers['authorization'] = authHeader; + // Cast through unknown: handlePilotTunnel only reads `headers` from + // req, and on the reject path only calls socket.write(string) and + // socket.destroy() on the Duplex. Anything else we add to the handler + // (e.g. reading req.socket.remoteAddress for rate-limiting) silently + // no-ops against the stub, so update both stubs when the surface grows. + return { headers } as unknown as IncomingMessage; +} + +/** + * Mint a real pilot_enroll JWT against the test DB's auth secret. Mirrors + * mintPilotEnrollment in routes/nodes.ts but inlined to avoid pulling the + * full Express request shape into the test. + */ +function mintEnroll(nodeId: number, ttlSeconds = 15 * 60): { token: string; hash: string } { + const db = DatabaseService.getInstance(); + const jwtSecret = db.getGlobalSettings().auth_jwt_secret; + if (!jwtSecret) throw new Error('test DB has no auth_jwt_secret'); + const token = jwt.sign( + { scope: 'pilot_enroll', nodeId, enrollNonce: crypto.randomUUID() }, + jwtSecret, + { expiresIn: ttlSeconds }, + ); + const hash = crypto.createHash('sha256').update(token).digest('hex'); + db.createPilotEnrollment(nodeId, hash, Date.now() + ttlSeconds * 1000); + return { token, hash }; +} + +let nodeId: number; + +beforeEach(() => { + // Fresh pilot-mode node per test so prior enrollments do not bleed. + nodeId = DatabaseService.getInstance().addNode({ + name: `pilot-replay-${Date.now()}-${Math.random()}`, + type: 'remote', + mode: 'pilot_agent', + compose_dir: '/tmp/x', + is_default: false, + api_url: '', + api_token: '', + }); +}); + +describe('handlePilotTunnel: rejection paths in the upgrade handler', () => { + it('rejects an enrollment token whose row was already consumed', async () => { + const { token, hash } = mintEnroll(nodeId); + + // Simulate that a prior successful enrollment consumed the row. + const firstConsume = DatabaseService.getInstance().consumePilotEnrollment(hash); + expect(firstConsume).toBeDefined(); + expect(firstConsume?.node_id).toBe(nodeId); + + // Now drive the upgrade handler with the same token. The route + // must reject with 401 before granting the WebSocket upgrade. + const socket = makeStubSocket(); + await handlePilotTunnel( + makeStubReq(`Bearer ${token}`), + socket as unknown as Duplex, + Buffer.alloc(0), + pilotTunnelWss, + ); + + expect(socket.destroyed).toBe(true); + expect(socket.writes.length).toBeGreaterThan(0); + expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/); + }); + + it('rejects an enrollment token whose row was never created', async () => { + // Hand-rolled token whose hash matches no row in pilot_enrollments. + const db = DatabaseService.getInstance(); + const jwtSecret = db.getGlobalSettings().auth_jwt_secret; + const orphan = jwt.sign( + { scope: 'pilot_enroll', nodeId, enrollNonce: 'never-stored' }, + jwtSecret, + { expiresIn: 60 }, + ); + + const socket = makeStubSocket(); + await handlePilotTunnel( + makeStubReq(`Bearer ${orphan}`), + socket as unknown as Duplex, + Buffer.alloc(0), + pilotTunnelWss, + ); + + expect(socket.destroyed).toBe(true); + expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/); + }); + + it('rejects an enrollment token whose row has expired', async () => { + const db = DatabaseService.getInstance(); + const jwtSecret = db.getGlobalSettings().auth_jwt_secret; + // JWT TTL still valid (60 s) but the DB row's expires_at is in the + // past. consumePilotEnrollment filters on expires_at > now, so this + // tests the DB-side window, not the JWT-side one. + const token = jwt.sign( + { scope: 'pilot_enroll', nodeId, enrollNonce: 'expired-row' }, + jwtSecret, + { expiresIn: 60 }, + ); + const hash = crypto.createHash('sha256').update(token).digest('hex'); + db.createPilotEnrollment(nodeId, hash, Date.now() - 1_000); + + const socket = makeStubSocket(); + await handlePilotTunnel( + makeStubReq(`Bearer ${token}`), + socket as unknown as Duplex, + Buffer.alloc(0), + pilotTunnelWss, + ); + + expect(socket.destroyed).toBe(true); + expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/); + }); + + it('rejects a missing Authorization header before touching the DB', async () => { + const socket = makeStubSocket(); + await handlePilotTunnel( + makeStubReq(undefined), + socket as unknown as Duplex, + Buffer.alloc(0), + pilotTunnelWss, + ); + + expect(socket.destroyed).toBe(true); + expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/); + }); + + it('rejects a JWT signed with a wrong secret', async () => { + const wrongToken = jwt.sign( + { scope: 'pilot_enroll', nodeId, enrollNonce: 'wrong-secret' }, + 'this-is-not-the-real-secret', + { expiresIn: 60 }, + ); + + const socket = makeStubSocket(); + await handlePilotTunnel( + makeStubReq(`Bearer ${wrongToken}`), + socket as unknown as Duplex, + Buffer.alloc(0), + pilotTunnelWss, + ); + + expect(socket.destroyed).toBe(true); + expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 401 Unauthorized/); + }); + + it('rejects a pilot_tunnel-scoped JWT for an unknown node', async () => { + // pilot_tunnel scope is the long-lived credential persisted on the + // agent. A pilot_tunnel JWT for a node that has been deleted (or + // never existed) must be rejected at the upgrade handler: the + // node lookup is the gate, not the JWT signature alone. + const db = DatabaseService.getInstance(); + const jwtSecret = db.getGlobalSettings().auth_jwt_secret; + const ghost = jwt.sign( + { scope: 'pilot_tunnel', nodeId: 99_999_999 }, + jwtSecret, + { expiresIn: '365d' }, + ); + + const socket = makeStubSocket(); + await handlePilotTunnel( + makeStubReq(`Bearer ${ghost}`), + socket as unknown as Duplex, + Buffer.alloc(0), + pilotTunnelWss, + ); + + expect(socket.destroyed).toBe(true); + expect(socket.writes[0]).toMatch(/^HTTP\/1\.1 404 /); + }); +}); diff --git a/backend/src/services/PilotTunnelBridge.ts b/backend/src/services/PilotTunnelBridge.ts index d4ef2656..16a1d53a 100644 --- a/backend/src/services/PilotTunnelBridge.ts +++ b/backend/src/services/PilotTunnelBridge.ts @@ -59,13 +59,6 @@ interface TcpStreamState extends StreamMeta { type StreamState = HttpStreamState | WsStreamState | TcpStreamState; -export interface TcpStreamSummary { - streamId: number; - bytesIn: number; - bytesOut: number; - openedAt: number; -} - /** * Sencho Mesh TCP stream handle. EventEmitter-based duplex-like surface that * MeshService consumes to bridge a local socket to a Compose service on the @@ -105,6 +98,19 @@ export class TcpStream extends EventEmitter { } } +/** + * Narrow surface that Sencho Mesh consumes from a registered pilot tunnel. + * `PilotTunnelManager.getBridge` returns this interface (not the concrete + * bridge) so MeshService cannot reach into transport internals: close code, + * loopback URL, the per-stream maps, the underscored helpers, etc. + * Keep this set minimal: it is the contract a future alternative transport + * (a stub for tests, a different routing strategy) would have to implement. + */ +export interface MeshTunnelHandle { + openTcpStream(target: { stack: string; service: string; port: number }): TcpStream | null; + getBufferedAmount(): number; +} + /** * Per-tunnel bridge: hosts a loopback HTTP server that demuxes requests into * wire frames sent over the pilot WebSocket, and remuxes response frames back @@ -114,7 +120,7 @@ export class TcpStream extends EventEmitter { * as just another remote target, so HTTP and WebSocket proxy logic, header * stripping/injection, and license-tier propagation all work unchanged. */ -export class PilotTunnelBridge extends EventEmitter { +export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle { private readonly tunnelWs: WebSocket; private readonly loopback: HttpServer; private readonly wsUpgradeServer: WebSocketServer; @@ -231,20 +237,6 @@ export class PilotTunnelBridge extends EventEmitter { this.sendJson({ t: 'tcp_close', s: streamId }); } - /** - * Snapshot of active TCP streams for the diagnostics sheet. Cheap; called - * on demand by MeshService. - */ - public listTcpStreams(): TcpStreamSummary[] { - const out: TcpStreamSummary[] = []; - for (const [streamId, s] of this.streams) { - if (s.kind === 'tcp') { - out.push({ streamId, bytesIn: s.bytesIn, bytesOut: s.bytesOut, openedAt: s.openedAt }); - } - } - return out; - } - public close(code = 1000, reason = 'closed by primary'): void { if (this.closed) return; this.closed = true; diff --git a/backend/src/services/PilotTunnelManager.ts b/backend/src/services/PilotTunnelManager.ts index 48a2d849..57e4195f 100644 --- a/backend/src/services/PilotTunnelManager.ts +++ b/backend/src/services/PilotTunnelManager.ts @@ -1,6 +1,6 @@ import { EventEmitter } from 'events'; import WebSocket from 'ws'; -import { PilotTunnelBridge } from './PilotTunnelBridge'; +import { PilotTunnelBridge, type MeshTunnelHandle } from './PilotTunnelBridge'; import { DatabaseService } from './DatabaseService'; import { PilotCloseCode } from '../pilot/protocol'; import { isDebugEnabled } from '../utils/debug'; @@ -160,11 +160,11 @@ export class PilotTunnelManager extends EventEmitter { } /** - * Direct accessor for the per-node bridge. Returns null when no tunnel is - * registered. Used by Sencho Mesh to open TCP streams without going - * through the loopback HTTP path. + * Per-node tunnel handle, returned to MeshService. Returns the bridge + * narrowed to the MeshTunnelHandle surface so callers cannot reach + * into transport internals (loopback URL, per-stream maps, close API). */ - public getBridge(nodeId: number): PilotTunnelBridge | null { + public getBridge(nodeId: number): MeshTunnelHandle | null { return this.bridges.get(nodeId) ?? null; } @@ -178,22 +178,4 @@ export class PilotTunnelManager extends EventEmitter { this.bridges.delete(nodeId); } - /** - * Snapshot of currently active tunnels. - */ - public listActive(): Array<{ nodeId: number; loopbackUrl: string; connectedAt: number }> { - return Array.from(this.bridges.entries()).map(([nodeId, bridge]) => ({ - nodeId, - loopbackUrl: bridge.getLoopbackUrl(), - connectedAt: bridge.getConnectedAt(), - })); - } - - /** - * Record an application-level heartbeat from the agent. - */ - public touch(nodeId: number): void { - if (!this.bridges.has(nodeId)) return; - DatabaseService.getInstance().updateNode(nodeId, { pilot_last_seen: Date.now() }); - } } diff --git a/backend/src/utils/debug.ts b/backend/src/utils/debug.ts index 2317d3b2..704f7613 100644 --- a/backend/src/utils/debug.ts +++ b/backend/src/utils/debug.ts @@ -2,6 +2,18 @@ * Shared diagnostic logging gate. Reads `developer_mode` from * DatabaseService, which caches the global_settings snapshot internally * and invalidates on write, so hot-path callers can query freely. + * + * Acceptable: error paths, per-request paths, per-tunnel paths, + * once-per-stream paths. Each call costs one Map lookup in Node's + * require cache, one method call into the cached settings object, and + * a property access. Fine for these cadences. + * + * NOT acceptable: per-frame paths in steady state (WebSocket message + * loops, container stats streams, log-tail demuxers). The require + * lookup and singleton dispatch add up at thousands of calls per + * second. If you need diagnostic logging in a per-frame loop, + * snapshot the result once outside the loop and read the snapshot + * inside. */ export function isDebugEnabled(): boolean {