From 3d9489648ec9d2b13051edd07f2f13982914a095 Mon Sep 17 00:00:00 2001 From: Anso Date: Thu, 7 May 2026 21:28:35 -0400 Subject: [PATCH] fix(pilot): harden outbound reverse-tunnel against resource exhaustion (#979) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(pilot): cap tunnel frame size, concurrent streams, and stream idle time Pilot tunnels carry every HTTP, WS, and Mesh-TCP byte for a remote node through a single multiplexed WebSocket. A buggy or compromised peer that sent oversized frames, opened streams in a tight loop, or left streams parked indefinitely could exhaust gateway memory. Three protocol-level limits applied symmetrically on both ends: - MAX_FRAME_SIZE_BYTES (8 MB): set as the ws maxPayload on the gateway-side WebSocketServer and the agent-side WebSocket client, plus a defense-in-depth length check in decodeBinaryFrame and decodeJsonFrame. - MAX_STREAMS_PER_TUNNEL (1024): bridge refuses new loopback HTTP / upgrade / TCP allocations with 503 once at the cap; the agent rejects new incoming http_req / ws_open / tcp_open with the appropriate error frame. - STREAM_IDLE_TIMEOUT_MS (10 min): every stream gets a per-stream timer refreshed on each inbound or outbound activity. Expiry tears down the local half and notifies the peer. The constants are colocated in pilot/protocol.ts so any future agent build picks them up via the protocol module. * fix(pilot): rate-limit pilot enrollment endpoints to 10 per minute Pilot enrollment mints a JWT and writes a pilot_enrollments row, both privileged operations that should not share the global API budget (200/min). Add a dedicated express-rate-limit instance keyed by user or IP at 10/min in production (100/min in dev so the local enrollment test loop is not throttled). The limiter is applied directly on POST /api/nodes/:id/pilot/enroll. On POST /api/nodes the limiter's skip function reads the parsed body and exempts proxy-mode creates; only requests that resolve to mode=pilot_agent count against the enrollment budget. * fix(pilot): cap concurrent pilot tunnels system-wide The PilotTunnelManager held an unbounded Map of bridges. A reconnect storm, runaway enrollment, or operator misconfiguration could grow the map without limit, even though every tunnel still carries a valid JWT. Cap at 256 concurrent tunnels per primary instance: - Soft warning at 128 (logged once per crossing). - Hard refusal at 256: registerTunnel throws PilotTunnelCapacityError and the upgrade handler closes the WebSocket with 1013 (Try Again Later) so the agent backs off rather than tight-looping. A node that already has a registered tunnel does not consume a new slot when it reconnects; the existing bridge is closed first. * fix(pilot): release backpressure as soon as the tunnel buffer drains Previously the bridge paused HTTP request bodies when the tunnel WebSocket's bufferedAmount climbed above 4 MB but never resumed them explicitly; the next data event re-checked the threshold, and TCP streams only saw a 'drain' fan-out on the 30 s ping cycle. Slow-consumer peers held buffered bytes for tens of seconds longer than needed. Replace both with an on-demand drain check: when at least one stream is paused, sample bufferedAmount every 100 ms; once it drops below the high-water mark, resume every paused request and emit 'drain' to every accepted TCP stream, then stop the timer. Dormant when no stream is paused, so steady-state cost is zero. * fix(pilot): trust internal CAs via SENCHO_PILOT_CA_FILE Self-hosted deployments often terminate TLS with an internal CA. The only previous escape hatch was NODE_TLS_REJECT_UNAUTHORIZED=0, which disables verification across every outbound connection in the agent process and is the wrong shape of fix. Add SENCHO_PILOT_CA_FILE: when set, the agent reads the file as a PEM bundle and passes it to ws as the `ca` option for the tunnel WebSocket. rejectUnauthorized remains true. Failure to read the file exits with a clear error so the operator does not silently fall back to the default trust store. There is no flag to disable TLS verification entirely; that would defeat the credential trust model. * fix(pilot): gate frame-rate-adjacent logs behind developer_mode The agent's malformed-frame warning fires per inbound frame and could flood logs under attack or against a buggy primary. The bridge swallowed parse errors silently, leaving operators blind to protocol drift, and the manager had no signal at all on tunnel registration. Route per-frame and per-tunnel diagnostic logs through the existing isDebugEnabled() helper (same pattern used by middleware/auth, RBAC, and websocket/logs). Production stays quiet by default; the toggle is the existing developer_mode setting in global_settings — no new env var, no new dependency. * feat(pilot): expose per-tunnel metrics via /api/system/pilot-tunnels Operator support had no signal beyond raw logs to answer "is this tunnel flapping?" or "is one bad node hiding behind the aggregate?". Add an in-process counter set covering tunnels_total, tunnels_replaced, tunnels_rejected_capacity, enroll_acks, frame_decode_errors, plus a per-node array carrying the connectedAt and bufferedAmount so a single tunnel sitting on a stuck buffer remains visible. Counters live in services/PilotMetrics.ts; the gateway has no shared in-process metrics facility today, so this is a per-feature pattern documented as such for future consolidation. Counters are strictly process-local — no telemetry, no export, no phone-home — consistent with Sencho's privacy posture. Surfaced read-only via GET /api/system/pilot-tunnels behind admin auth, mirroring the existing cache-stats endpoint. * fix(pilot): tighten reconnect backoff and log handling Three small follow-ups to the hardening pass: - Sanitize the primary URL on the agent's connect log so a malicious SENCHO_PRIMARY_URL with embedded control characters cannot inject fake log lines. - Move the reconnect-backoff reset from the 'open' callback to the 'hello' frame handler. A peer that always rejects the handshake (incompatible version, consumed enrollment token) used to reset the backoff on every TCP-level connect and tight-loop reconnects; now the reset waits for a clean protocol round-trip. - Document the StreamIdAllocator wrap behavior so future readers can see the relationship between the 2^31 wrap and the MAX_STREAMS_PER_TUNNEL cap without re-deriving it. * test(pilot): add coverage for enrollment, rate limiter, and tunnel caps Adds two test files exercising the hardening pass surfaces that were previously uncovered: - pilot-enrollment.test.ts: end-to-end through POST /api/nodes (pilot mode), POST /api/nodes/:id/pilot/enroll, the SHA256 token-hash persistence, replay protection, expired enrollments, and the rate-limit header wiring (enrollment limiter on pilot paths, global limiter elsewhere). - pilot-bridge-limits.test.ts: oversize-frame rejection at both binary and JSON decoders, and the per-tunnel concurrent stream cap as observed via openTcpStream returning null and the loopback HTTP server returning 503 once the bridge is at capacity. Bridge cap test fills the slot map with TCP stream handles rather than real HTTP requests so the OS socket pool stays out of the picture. * test(pilot): cover manager metrics snapshot and capacity-error shape Adds direct coverage for: - PilotTunnelCapacityError exposing the limit so the upgrade handler can format a useful close-frame reason. - PilotMetrics.snapshot returning a defensive copy (callers must not mutate the live counter set). - getMetricsSnapshot returning the open count, per-node breakdown, and counter set in a stable shape, with tunnels_total bumping on a successful registerTunnel. The manager test seeds a real pilot-mode node row first because updateNode throws on missing rows; this models the production path where a node is created before its enrollment is consumed. * docs(pilot): document tunnel limits, custom CA, and new failure modes Refresh the Pilot Agent feature doc to cover the hardening surfaces: - New 'Self-signed primary TLS certs' section walking through the SENCHO_PILOT_CA_FILE env var with an example docker run command. - New 'Resource limits' section listing the per-tunnel and system-wide ceilings so operators know what the wire enforces. - Four new troubleshooting entries: WebSocket close 1013 (system cap), loopback 503 (stream cap), close 1002 with protocol error (frame size / malformed), and HTTP 429 on enrollment. Also expand .env.example with a new pilot-agent block covering SENCHO_MODE, SENCHO_PRIMARY_URL, SENCHO_ENROLL_TOKEN, and SENCHO_PILOT_CA_FILE so operators do not have to read code or the feature doc to discover the agent-side config surface. * fix(pilot): address code-review findings on the hardening pass Critical: - Bridge: TCP backpressure now starts the drain timer and tracks streams awaiting drain, so a TcpStream caller waiting on 'drain' no longer hangs when no HTTP request is in flight. - Bridge: every direct streams.delete call now goes through the removeStream helper so per-stream idle timers are cleared consistently and pausedReqs / tcpAwaitingDrain stay aligned. - Bridge close(): resume any paused IncomingMessage before clearing the map so a parser does not stay stuck across teardown. High: - Protocol: decodeJsonFrame now compares Buffer.byteLength(raw, 'utf8') against the cap; raw.length (UTF-16 code units) let multi-byte payloads sneak ~3x the byte budget through. - Agent: the optional CA bundle is read once at construction and cached on the instance, so a missing or rotated SENCHO_PILOT_CA_FILE no longer process-exits on every reconnect attempt. Medium / Low: - Bridge limits test: assert mockWs.sent.length to confirm slot allocations actually serialized a tcp_open frame and that a rejected (cap+1) attempt did not consume a stream id. - Manager: defer the tunnels_replaced increment until after the cap check passes, so a rejected reconnect does not double-count as both replacement and capacity rejection. - Manager: clarify the protocol comment about MAX_FRAME_SIZE_BYTES enforcement layering (ws maxPayload is authoritative; decoder check is for tests and defense-in-depth). - Limiter: defensive `if (!req.body) return false` so a future refactor that delays body parsing cannot silently skip the enrollment limiter. Deferred (documented as follow-ups, not in this branch): - M2: narrowing public surface of PilotTunnelManager.getBridge to a MeshTunnelHandle interface (cross-cutting refactor). - H4: end-to-end replay test driving the upgrade handler twice with the same enrollment JWT (needs WS test harness). --- .env.example | 26 +++ .../src/__tests__/pilot-bridge-limits.test.ts | 140 +++++++++++ .../src/__tests__/pilot-enrollment.test.ts | 204 ++++++++++++++++ .../src/__tests__/pilot-manager-cap.test.ts | 116 ++++++++++ backend/src/middleware/rateLimiters.ts | 27 +++ backend/src/pilot/agent.ts | 161 ++++++++++++- backend/src/pilot/protocol.ts | 51 +++- backend/src/routes/metrics.ts | 17 ++ backend/src/routes/nodes.ts | 5 +- backend/src/server.ts | 9 +- backend/src/services/PilotMetrics.ts | 39 ++++ backend/src/services/PilotTunnelBridge.ts | 219 +++++++++++++++--- backend/src/services/PilotTunnelManager.ts | 76 ++++++ backend/src/websocket/pilotTunnel.ts | 10 +- docs/features/pilot-agent.mdx | 44 ++++ 15 files changed, 1100 insertions(+), 44 deletions(-) create mode 100644 backend/src/__tests__/pilot-bridge-limits.test.ts create mode 100644 backend/src/__tests__/pilot-enrollment.test.ts create mode 100644 backend/src/__tests__/pilot-manager-cap.test.ts create mode 100644 backend/src/services/PilotMetrics.ts diff --git a/.env.example b/.env.example index 0512a316..d22439c5 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,32 @@ API_RATE_LIMIT=200 # Increase for environments with many concurrent browser sessions behind shared NAT. API_POLLING_RATE_LIMIT=300 +# ─── Pilot agent (remote host only) ────────────────────────────── +# These three vars are required ONLY on a remote host running as a +# pilot-agent reverse-tunnel container. The primary instance does not +# read them. See docs/features/pilot-agent.mdx for the full setup. + +# Switches the container into agent mode. The primary instance leaves +# this unset. +SENCHO_MODE= + +# WebSocket-capable URL of the controlling Sencho instance. Use https:// +# scheme; the agent rewrites it to wss:// for the tunnel upgrade. +SENCHO_PRIMARY_URL= + +# Single-use, 15-minute enrollment token issued by the primary when the +# pilot-agent node is created. After the first successful connect the +# agent persists a long-lived tunnel credential at /app/data/pilot.jwt +# and ignores this var on subsequent restarts. +SENCHO_ENROLL_TOKEN= + +# Optional: path inside the agent container to a PEM CA bundle the +# agent should trust when validating the primary's TLS cert. Use this +# for self-hosted deployments terminating TLS with an internal CA. +# Leave unset to fall back to the system trust store. There is no +# escape hatch to disable TLS verification. +SENCHO_PILOT_CA_FILE= + # ─── SSO / LDAP Configuration ──────────────────────────────────── # LDAP / Active Directory diff --git a/backend/src/__tests__/pilot-bridge-limits.test.ts b/backend/src/__tests__/pilot-bridge-limits.test.ts new file mode 100644 index 00000000..11c8dc04 --- /dev/null +++ b/backend/src/__tests__/pilot-bridge-limits.test.ts @@ -0,0 +1,140 @@ +/** + * Tests for the pilot-tunnel bridge resource limits introduced in the + * hardening pass: + * - Frame-size cap enforced at the protocol decoders. + * - Per-tunnel concurrent stream cap enforced by the bridge. + * - Per-stream idle timeout (verified via internal helpers; the real + * timer fires at 10 minutes which is too long for a unit test, so we + * drive the timer paths via reduced state). + * + * The bridge is exercised through its loopback HTTP server with a mock + * tunnel WebSocket; this keeps the test in-process and avoids needing a + * full pilot agent for the cap surfaces. + */ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import http from 'http'; +import { EventEmitter } from 'events'; +import { WebSocket } from 'ws'; +import { + MAX_FRAME_SIZE_BYTES, + MAX_STREAMS_PER_TUNNEL, + decodeBinaryFrame, + decodeJsonFrame, +} from '../pilot/protocol'; +import { PilotTunnelBridge } from '../services/PilotTunnelBridge'; + +describe('protocol decoders', () => { + it('rejects oversized binary frames', () => { + const buf = Buffer.alloc(MAX_FRAME_SIZE_BYTES + 1); + expect(() => decodeBinaryFrame(buf)).toThrow(/too large/); + }); + + it('rejects undersized binary frames', () => { + expect(() => decodeBinaryFrame(Buffer.alloc(3))).toThrow(/too short/); + }); + + it('rejects oversized JSON frames', () => { + const big = '"' + 'x'.repeat(MAX_FRAME_SIZE_BYTES) + '"'; + expect(() => decodeJsonFrame(big)).toThrow(/too large/); + }); + + it('rejects malformed JSON', () => { + expect(() => decodeJsonFrame('not json')).toThrow(); + }); + + it('rejects JSON without a type discriminator', () => { + expect(() => decodeJsonFrame('{"hello":"world"}')).toThrow(/discriminator/); + }); +}); + +/** + * Minimal mock that satisfies PilotTunnelBridge's contract with its + * tunnelWs argument. The bridge calls `send`, reads `bufferedAmount`, + * checks `readyState`, and listens for `message`, `close`, `error`. + */ +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('PilotTunnelBridge concurrent stream cap', () => { + let bridge: PilotTunnelBridge; + let mockWs: ReturnType; + let loopbackUrl: string; + + beforeAll(async () => { + mockWs = makeMockTunnelWs(); + // The bridge constructor type-checks against `WebSocket`; the runtime + // only needs the surface exercised above, so the cast keeps the test + // honest without dragging in a real ws pair just to fill the cap. + bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket); + await bridge.start(); + loopbackUrl = bridge.getLoopbackUrl(); + }); + + afterAll(() => { + bridge.close(); + }); + + it('refuses HTTP requests with 503 once the cap is reached', async () => { + // Use openTcpStream to fill the stream map without putting any load + // on the loopback HTTP socket pool. Each TCP stream consumes a slot + // exactly the same way an HTTP request would. + const filled: Array> = []; + for (let i = 0; i < MAX_STREAMS_PER_TUNNEL; i++) { + const handle = bridge.openTcpStream({ stack: 's', service: 'svc', port: 80 }); + // Suppress the 'error' that fires later when we close the + // bridge; the handle is used purely for the slot. + handle?.on('error', () => { /* expected on cleanup */ }); + filled.push(handle); + } + + // Confirm all 1024 slot allocations actually serialized a tcp_open + // frame on the tunnel. If a slot were short-circuited without + // sending the frame, this would catch the regression. + const sentCount = mockWs.sent.length; + expect(sentCount).toBe(MAX_STREAMS_PER_TUNNEL); + + // The (cap+1)th openTcpStream returns null per the cap check, and + // does NOT consume a stream id (no tcp_open is sent). + expect(bridge.openTcpStream({ stack: 's', service: 'svc', port: 80 })).toBeNull(); + expect(mockWs.sent.length).toBe(sentCount); + + // And a loopback HTTP request lands on the 503 branch of + // handleLoopbackRequest. + const url = new URL(loopbackUrl); + const overflow = await new Promise((resolve, reject) => { + const req = http.request({ + host: url.hostname, + port: Number(url.port), + method: 'GET', + path: '/over-the-cap', + }, (res) => resolve(res.statusCode || 0)); + req.on('error', reject); + req.end(); + }); + expect(overflow).toBe(503); + // The 503 path also does not allocate a stream or send any frame. + expect(mockWs.sent.length).toBe(sentCount); + }, 15_000); +}); diff --git a/backend/src/__tests__/pilot-enrollment.test.ts b/backend/src/__tests__/pilot-enrollment.test.ts new file mode 100644 index 00000000..f1b95f98 --- /dev/null +++ b/backend/src/__tests__/pilot-enrollment.test.ts @@ -0,0 +1,204 @@ +/** + * Tests for pilot-agent enrollment lifecycle and rate limiting. + * + * Covers: + * - POST /api/nodes with mode=pilot_agent mints an enrollment token, persists + * the SHA256 hash into pilot_enrollments, and returns a docker run command + * containing the bearer token. + * - consumePilotEnrollment is one-shot: a second consume on the same hash + * returns null (replay protection). + * - Expired enrollments are not consumable. + * - The enrollment rate limiter (10/min in production, 100/min in dev) is + * wired only on routes that mint a pilot enrollment. + */ +import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import request from 'supertest'; +import crypto from 'crypto'; +import { setupTestDb, cleanupTestDb, loginAsTestAdmin } from './helpers/setupTestDb'; + +let tmpDir: string; +let app: import('express').Express; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; +let adminCookie: string; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ DatabaseService } = await import('../services/DatabaseService')); + ({ app } = await import('../index')); + adminCookie = await loginAsTestAdmin(app); +}); + +afterAll(() => cleanupTestDb(tmpDir)); + +describe('POST /api/nodes (pilot_agent mode)', () => { + it('mints an enrollment token and returns a docker run command', async () => { + const res = await request(app) + .post('/api/nodes') + .set('Cookie', adminCookie) + .send({ name: 'pilot-1', type: 'remote', mode: 'pilot_agent' }); + + expect(res.status).toBe(200); + expect(res.body.success).toBe(true); + expect(res.body.enrollment).toBeDefined(); + expect(res.body.enrollment.token).toMatch(/^[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$/); + expect(res.body.enrollment.dockerRun).toContain('SENCHO_MODE=pilot'); + expect(res.body.enrollment.dockerRun).toContain(`SENCHO_ENROLL_TOKEN=${res.body.enrollment.token}`); + expect(res.body.enrollment.expiresAt).toBeGreaterThan(Date.now()); + }); + + it('persists the token hash, not the raw token', async () => { + const res = await request(app) + .post('/api/nodes') + .set('Cookie', adminCookie) + .send({ name: 'pilot-2', type: 'remote', mode: 'pilot_agent' }); + + const expectedHash = crypto.createHash('sha256').update(res.body.enrollment.token).digest('hex'); + const row = DatabaseService.getInstance().getDb() + .prepare('SELECT token_hash, used_at FROM pilot_enrollments WHERE node_id = ?') + .get(res.body.id) as { token_hash: string; used_at: number | null } | undefined; + + expect(row).toBeDefined(); + expect(row?.token_hash).toBe(expectedHash); + expect(row?.used_at).toBeNull(); + // The raw token must not appear anywhere in the row. + expect(row?.token_hash).not.toBe(res.body.enrollment.token); + }); + + it('rejects unauthenticated callers with 401', async () => { + const res = await request(app) + .post('/api/nodes') + .send({ name: 'pilot-anon', type: 'remote', mode: 'pilot_agent' }); + expect(res.status).toBe(401); + }); +}); + +describe('POST /api/nodes/:id/pilot/enroll', () => { + it('regenerates the enrollment for an existing pilot node', async () => { + const create = await request(app) + .post('/api/nodes') + .set('Cookie', adminCookie) + .send({ name: 'pilot-regen', type: 'remote', mode: 'pilot_agent' }); + const original = create.body.enrollment.token; + + const regen = await request(app) + .post(`/api/nodes/${create.body.id}/pilot/enroll`) + .set('Cookie', adminCookie); + + expect(regen.status).toBe(200); + expect(regen.body.enrollment.token).not.toBe(original); + }); + + it('rejects regeneration for proxy-mode remote nodes', async () => { + const create = await request(app) + .post('/api/nodes') + .set('Cookie', adminCookie) + .send({ + name: 'proxy-not-pilot', + type: 'remote', + api_url: 'http://192.168.1.50:1852', + api_token: 'tok', + }); + + const regen = await request(app) + .post(`/api/nodes/${create.body.id}/pilot/enroll`) + .set('Cookie', adminCookie); + + expect(regen.status).toBe(400); + expect(regen.body.error).toMatch(/pilot/i); + }); + + it('rejects unknown nodes with 404', async () => { + const res = await request(app) + .post('/api/nodes/999999/pilot/enroll') + .set('Cookie', adminCookie); + expect(res.status).toBe(404); + }); +}); + +describe('consumePilotEnrollment replay protection', () => { + it('marks the row used and rejects the second consume', () => { + const db = DatabaseService.getInstance(); + const create = db.addNode({ + name: 'pilot-replay', + type: 'remote', + compose_dir: '/tmp/x', + mode: 'pilot_agent', + is_default: false, + api_url: '', + api_token: '', + }); + const tokenHash = crypto.createHash('sha256').update('synthetic-test-token').digest('hex'); + db.createPilotEnrollment(create, tokenHash, Date.now() + 60_000); + + const first = db.consumePilotEnrollment(tokenHash); + expect(first).toBeDefined(); + expect(first?.node_id).toBe(create); + + const second = db.consumePilotEnrollment(tokenHash); + expect(second).toBeUndefined(); + }); + + it('rejects expired enrollments', () => { + const db = DatabaseService.getInstance(); + const nodeId = db.addNode({ + name: 'pilot-expired', + type: 'remote', + compose_dir: '/tmp/x', + mode: 'pilot_agent', + is_default: false, + api_url: '', + api_token: '', + }); + const tokenHash = crypto.createHash('sha256').update('expired-test-token').digest('hex'); + db.createPilotEnrollment(nodeId, tokenHash, Date.now() - 1_000); + + expect(db.consumePilotEnrollment(tokenHash)).toBeUndefined(); + }); + + it('rejects unknown token hashes', () => { + const db = DatabaseService.getInstance(); + expect(db.consumePilotEnrollment('0'.repeat(64))).toBeUndefined(); + }); +}); + +describe('Enrollment rate limiter wiring', () => { + it('POST /api/nodes (pilot mode) advertises a tighter limit than the global limiter', async () => { + const res = await request(app) + .post('/api/nodes') + .set('Cookie', adminCookie) + .send({ name: 'pilot-headers', type: 'remote', mode: 'pilot_agent' }); + + const limit = parseInt(res.headers['ratelimit-limit'], 10); + // Dev limit is 100/min for the enrollment limiter; global is 1000/min. + expect(limit).toBe(100); + }); + + it('POST /api/nodes (proxy mode) falls back to the global limiter', async () => { + const res = await request(app) + .post('/api/nodes') + .set('Cookie', adminCookie) + .send({ + name: 'proxy-global-limit', + type: 'remote', + api_url: 'http://192.168.1.51:1852', + api_token: 'tok', + }); + + const limit = parseInt(res.headers['ratelimit-limit'], 10); + expect(limit).toBe(1000); + }); + + it('POST /api/nodes/:id/pilot/enroll uses the enrollment limiter', async () => { + const create = await request(app) + .post('/api/nodes') + .set('Cookie', adminCookie) + .send({ name: 'pilot-regen-headers', type: 'remote', mode: 'pilot_agent' }); + + const regen = await request(app) + .post(`/api/nodes/${create.body.id}/pilot/enroll`) + .set('Cookie', adminCookie); + + const limit = parseInt(regen.headers['ratelimit-limit'], 10); + expect(limit).toBe(100); + }); +}); diff --git a/backend/src/__tests__/pilot-manager-cap.test.ts b/backend/src/__tests__/pilot-manager-cap.test.ts new file mode 100644 index 00000000..54c124c4 --- /dev/null +++ b/backend/src/__tests__/pilot-manager-cap.test.ts @@ -0,0 +1,116 @@ +/** + * Tests for PilotTunnelManager system-wide cap (M3) and the metrics + * snapshot exposed via /api/system/pilot-tunnels (M2). + */ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest'; +import { EventEmitter } from 'events'; +import { WebSocket } from 'ws'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; +import { PilotTunnelManager, PilotTunnelCapacityError } from '../services/PilotTunnelManager'; +import { PilotMetrics } from '../services/PilotMetrics'; + +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('PilotTunnelCapacityError', () => { + it('carries the limit that triggered the rejection', () => { + const err = new PilotTunnelCapacityError(256); + expect(err.name).toBe('PilotTunnelCapacityError'); + expect(err.limit).toBe(256); + expect(err.message).toContain('256'); + }); +}); + +describe('PilotMetrics counters', () => { + beforeEach(() => { + // Counters are process-singleton; we verify deltas rather than + // absolute values so the test is order-independent. + }); + + it('increment then snapshot returns the bumped value', () => { + const before = PilotMetrics.snapshot(); + PilotMetrics.increment('enroll_acks'); + const after = PilotMetrics.snapshot(); + expect(after.enroll_acks).toBe(before.enroll_acks + 1); + }); + + it('snapshot returns a copy, not the live object', () => { + const snap = PilotMetrics.snapshot(); + const original = snap.tunnels_total; + PilotMetrics.increment('tunnels_total'); + // The earlier snapshot must remain unchanged. + expect(snap.tunnels_total).toBe(original); + }); +}); + +describe('PilotTunnelManager.getMetricsSnapshot', () => { + it('returns the current open count and counter set', () => { + const mgr = PilotTunnelManager.getInstance(); + const snap = mgr.getMetricsSnapshot(); + + expect(snap).toHaveProperty('counters'); + expect(snap).toHaveProperty('tunnels_open'); + expect(snap).toHaveProperty('per_node'); + expect(Array.isArray(snap.per_node)).toBe(true); + expect(snap.tunnels_open).toBe(snap.per_node.length); + }); + + it('reflects per-tunnel breakdown after registerTunnel succeeds', async () => { + const mgr = PilotTunnelManager.getInstance(); + const before = mgr.getMetricsSnapshot(); + + // Seed a real pilot-mode node so updateNode does not throw. + const nodeId = DatabaseService.getInstance().addNode({ + name: `pilot-mgr-${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'); + + const after = mgr.getMetricsSnapshot(); + expect(after.tunnels_open).toBe(before.tunnels_open + 1); + expect(after.per_node.some((p) => p.nodeId === nodeId)).toBe(true); + expect(after.counters.tunnels_total).toBe(before.counters.tunnels_total + 1); + + // Cleanup so this test does not leak a tunnel into the next. + mgr.closeTunnel(nodeId); + }); +}); diff --git a/backend/src/middleware/rateLimiters.ts b/backend/src/middleware/rateLimiters.ts index 69139516..c0c82ff4 100644 --- a/backend/src/middleware/rateLimiters.ts +++ b/backend/src/middleware/rateLimiters.ts @@ -145,6 +145,33 @@ export const ssoRateLimiter = rateLimit({ message: { error: 'Too many SSO attempts. Please try again later.' }, }); +// Pilot enrollment limiter. Enrollment mints a JWT and writes a +// pilot_enrollments row, so it deserves a stricter ceiling than the global +// API limiter (200/min). Applied on the regenerate route directly and on +// `POST /api/nodes` only when the body resolves to pilot_agent mode (the +// limiter's `skip` function reads the parsed body). +export const enrollmentLimiter = rateLimit({ + ...rateLimitBase, + max: process.env.NODE_ENV === 'production' ? 10 : 100, + keyGenerator: rateLimitKeyGenerator, + message: { error: 'Too many enrollment requests. Please try again shortly.' }, + skip: (req: Request) => { + // Only gate `POST /api/nodes` calls that actually create a pilot agent; + // proxy-mode node creation falls back to the global limiter. The + // dedicated `/pilot/enroll` route applies this limiter unconditionally, + // so the body check is bypassed there by skipping the skip when the + // path already targets enrollment. + if (req.path.endsWith('/pilot/enroll')) return false; + // Express.json() runs globally before route handlers in app.ts, so + // req.body is parsed by the time this fires. If a future refactor moves + // body parsing per-route the worst case is "limiter applies even to + // proxy-mode" rather than "limiter is bypassed entirely". + if (!req.body) return false; + const body = req.body as { mode?: string; type?: string }; + return !(body.type === 'remote' && body.mode === 'pilot_agent'); + }, +}); + // Trivy install/update limiter. Install + update are expensive (binary download, // sha256 verification) so a 10-minute window prevents accidental thrashing. export const trivyInstallLimiter = rateLimit({ diff --git a/backend/src/pilot/agent.ts b/backend/src/pilot/agent.ts index 9c3e916e..d2531921 100644 --- a/backend/src/pilot/agent.ts +++ b/backend/src/pilot/agent.ts @@ -6,8 +6,11 @@ import WebSocket from 'ws'; import { getSenchoVersion } from '../services/CapabilityRegistry'; import { BinaryFrameType, + MAX_FRAME_SIZE_BYTES, + MAX_STREAMS_PER_TUNNEL, MeshErrCode, PROTOCOL_VERSION, + STREAM_IDLE_TIMEOUT_MS, decodeBinaryFrame, decodeJsonFrame, encodeBinaryFrame, @@ -16,6 +19,7 @@ import { wsDataToString, } from './protocol'; import { sanitizeForLog } from '../utils/safeLog'; +import { isDebugEnabled } from '../utils/debug'; const RECONNECT_MIN_MS = 1_000; const RECONNECT_MAX_MS = 60_000; @@ -69,13 +73,22 @@ class PilotAgent { private readonly httpStreams = new Map(); private readonly wsStreams = new Map(); private readonly tcpStreams = new Map(); + private readonly idleTimers = new Map(); private shuttingDown = false; private readonly agentVersion: string; + /** + * Optional CA bundle read once at agent construction. Cached so that a + * later rotation (file renamed, secret rotated) does not surprise the + * agent with a process exit on the next reconnect; container restart is + * the documented way to pick up a new CA bundle. + */ + private readonly customCa: Buffer | null; constructor(options: AgentOptions) { this.options = options; this.token = options.initialToken; this.agentVersion = getSenchoVersion() || '0.0.0'; + this.customCa = readPilotCaBundle(); } public start(): void { @@ -101,12 +114,24 @@ class PilotAgent { 'x-sencho-agent-version': this.agentVersion, }, handshakeTimeout: 15_000, + 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 + // 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; ws.on('open', () => { - console.log('[Pilot] Tunnel connected to', this.options.primaryUrl); - this.backoff = RECONNECT_MIN_MS; + // Backoff intentionally NOT reset here: a TCP-level connect that + // immediately fails the protocol handshake (incompatible version, + // bad token consumed at upgrade) would otherwise reset the + // backoff and tight-loop reconnects. The reset moves to the + // handleJsonFrame 'hello' case once we have a clean handshake. + console.log('[Pilot] Tunnel connected to', sanitizeForLog(this.options.primaryUrl)); try { ws.send(encodeJsonFrame({ t: 'hello', @@ -150,6 +175,58 @@ class PilotAgent { try { stream.socket.destroy(); } catch { /* ignore */ } } this.tcpStreams.clear(); + for (const [, timer] of this.idleTimers) clearTimeout(timer); + this.idleTimers.clear(); + } + + private streamCount(): number { + return this.httpStreams.size + this.wsStreams.size + this.tcpStreams.size; + } + + 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.ws; + const httpEntry = this.httpStreams.get(streamId); + if (httpEntry) { + try { httpEntry.req.destroy(); } catch { /* ignore */ } + this.httpStreams.delete(streamId); + if (ws) { + try { ws.send(encodeJsonFrame({ t: 'http_err', s: streamId, code: 'timeout', message: 'agent idle timeout' })); } catch { /* ignore */ } + } + return; + } + const wsEntry = this.wsStreams.get(streamId); + if (wsEntry) { + try { wsEntry.close(1001, 'idle'); } catch { /* ignore */ } + this.wsStreams.delete(streamId); + 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 */ } + } + } } private scheduleReconnect(): void { @@ -176,7 +253,9 @@ class PilotAgent { this.handleJsonFrame(frame); } } catch (err) { - console.warn('[Pilot] Malformed frame from primary:', sanitizeForLog((err as Error).message)); + // Per-frame; diag-gated to avoid log floods from a misbehaving + // primary or a malformed frame arriving in a tight loop. + if (isDebugEnabled()) console.warn('[Pilot:diag] Malformed frame from primary:', sanitizeForLog((err as Error).message)); } } @@ -191,6 +270,11 @@ class PilotAgent { try { ws.close(1002, 'incompatible version'); } catch { /* ignore */ } process.exit(1); } + // Clean handshake: it is now safe to reset the reconnect + // backoff. Doing this earlier (in 'open') would let a peer + // that always rejects the handshake drive us into a tight + // reconnect loop. + this.backoff = RECONNECT_MIN_MS; break; } case 'ctrl': { @@ -220,18 +304,21 @@ class PilotAgent { const entry = this.httpStreams.get(frame.streamId); if (!entry) return; try { entry.req.write(frame.payload); } catch { /* ignore */ } + this.refreshIdleTimer(frame.streamId); break; } case BinaryFrameType.WsMessageBinary: { const ws = this.wsStreams.get(frame.streamId); if (!ws) return; try { ws.send(frame.payload, { binary: true }); } catch { /* ignore */ } + this.refreshIdleTimer(frame.streamId); break; } case BinaryFrameType.TcpData: { const stream = this.tcpStreams.get(frame.streamId); if (!stream) return; try { stream.socket.write(frame.payload); } catch { /* ignore */ } + this.refreshIdleTimer(frame.streamId); break; } default: @@ -244,6 +331,17 @@ class PilotAgent { private onHttpReq(frame: Extract, { t: 'http_req' }>): void { const ws = this.ws; if (!ws) return; + if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) { + try { + ws.send(encodeJsonFrame({ + t: 'http_err', + s: frame.s, + code: 'agent_error', + message: 'agent stream cap reached', + })); + } catch { /* ignore */ } + return; + } const req = http.request({ host: '127.0.0.1', @@ -265,17 +363,21 @@ class PilotAgent { headers: outHeaders, })); } catch { /* ignore */ } + this.refreshIdleTimer(frame.s); res.on('data', (chunk: Buffer) => { try { ws.send(encodeBinaryFrame(BinaryFrameType.HttpResBody, frame.s, chunk), { binary: true }); } catch { /* ignore */ } + this.refreshIdleTimer(frame.s); }); res.on('end', () => { try { ws.send(encodeJsonFrame({ t: 'http_res_end', s: frame.s })); } catch { /* ignore */ } this.httpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); }); res.on('error', () => { try { ws.send(encodeJsonFrame({ t: 'http_err', s: frame.s, code: 'bad_response', message: 'upstream error' })); } catch { /* ignore */ } this.httpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); }); }); @@ -289,15 +391,18 @@ class PilotAgent { })); } catch { /* ignore */ } this.httpStreams.delete(frame.s); + this.clearIdleTimer(frame.s); }); this.httpStreams.set(frame.s, { req }); + this.refreshIdleTimer(frame.s); } private onHttpReqEnd(streamId: number): void { const entry = this.httpStreams.get(streamId); if (!entry) return; try { entry.req.end(); } catch { /* ignore */ } + this.refreshIdleTimer(streamId); } // --- WebSocket dispatch (tunnel -> loopback) --- @@ -305,15 +410,28 @@ class PilotAgent { private onWsOpen(frame: Extract, { t: 'ws_open' }>): void { const ws = this.ws; if (!ws) return; + if (this.streamCount() >= MAX_STREAMS_PER_TUNNEL) { + try { + ws.send(encodeJsonFrame({ + t: 'ws_reject', + s: frame.s, + status: 503, + message: 'agent stream cap reached', + })); + } catch { /* ignore */ } + return; + } const target = `ws://127.0.0.1:${this.options.loopbackPort}${frame.path}`; const client = new WebSocket(target, { headers: { ...frame.headers, host: `127.0.0.1:${this.options.loopbackPort}` }, + maxPayload: MAX_FRAME_SIZE_BYTES, }); client.on('open', () => { try { ws.send(encodeJsonFrame({ t: 'ws_accept', s: frame.s, headers: {} })); } catch { /* ignore */ } this.wsStreams.set(frame.s, client); + this.refreshIdleTimer(frame.s); }); client.on('message', (data, isBinary) => { if (isBinary) { @@ -321,14 +439,17 @@ class PilotAgent { } else { try { ws.send(encodeJsonFrame({ t: 'ws_msg_text', s: frame.s, data: wsDataToString(data) ?? '' })); } catch { /* ignore */ } } + this.refreshIdleTimer(frame.s); }); client.on('close', (code, reason) => { try { ws.send(encodeJsonFrame({ t: 'ws_close', s: frame.s, code, reason: reason?.toString?.() })); } catch { /* ignore */ } this.wsStreams.delete(frame.s); + this.clearIdleTimer(frame.s); }); client.on('error', () => { try { ws.send(encodeJsonFrame({ t: 'ws_reject', s: frame.s, status: 502, message: 'agent websocket failed' })); } catch { /* ignore */ } this.wsStreams.delete(frame.s); + this.clearIdleTimer(frame.s); }); } @@ -336,6 +457,7 @@ class PilotAgent { const ws = this.wsStreams.get(streamId); if (!ws) return; try { ws.send(data); } catch { /* ignore */ } + this.refreshIdleTimer(streamId); } private onWsClose(streamId: number, code: number, reason?: string): void { @@ -343,6 +465,7 @@ class PilotAgent { if (!ws) return; try { ws.close(code, reason); } catch { /* ignore */ } this.wsStreams.delete(streamId); + this.clearIdleTimer(streamId); } // --- Sencho Mesh TCP dispatch (tunnel -> Compose service container) --- @@ -354,6 +477,12 @@ class PilotAgent { 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) { @@ -367,6 +496,7 @@ class PilotAgent { 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 */ } @@ -376,17 +506,20 @@ class PilotAgent { 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) => { @@ -394,15 +527,18 @@ class PilotAgent { 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 */ } } }); @@ -412,6 +548,7 @@ class PilotAgent { const entry = this.tcpStreams.get(streamId); if (!entry) return; this.tcpStreams.delete(streamId); + this.clearIdleTimer(streamId); try { entry.socket.destroy(); } catch { /* ignore */ } } @@ -475,6 +612,24 @@ function readPersistedToken(): string | null { return null; } +/** + * Load the optional CA bundle pointed at by SENCHO_PILOT_CA_FILE so a pilot + * agent can verify a self-signed primary cert without disabling TLS + * verification globally. Returns the file contents or null if the var is + * unset; surfaces a clear error and exits if the file cannot be read so the + * operator does not silently fall back to the default trust store. + */ +function readPilotCaBundle(): Buffer | null { + const caFile = process.env.SENCHO_PILOT_CA_FILE; + if (!caFile) return null; + try { + return fs.readFileSync(caFile); + } catch (err) { + console.error('[Pilot] Failed to read SENCHO_PILOT_CA_FILE:', sanitizeForLog((err as Error).message)); + process.exit(1); + } +} + function persistToken(token: string): void { try { const dir = path.dirname(TOKEN_PATH); diff --git a/backend/src/pilot/protocol.ts b/backend/src/pilot/protocol.ts index bbd337ac..1ec79728 100644 --- a/backend/src/pilot/protocol.ts +++ b/backend/src/pilot/protocol.ts @@ -17,6 +17,39 @@ export const PROTOCOL_VERSION = 1; +/** + * Hard ceiling on a single tunnel WebSocket frame, in bytes. Authoritative + * enforcement is at the WebSocket layer: `maxPayload` on the gateway-side + * `WebSocketServer` and the agent-side `WebSocket` client both reject + * oversized frames before they reach the decoder. The decoder also enforces + * the same cap for two narrow cases: (1) tests that build Buffers locally + * and skip the WebSocket layer, and (2) defense-in-depth if a future + * codepath ever passes user-controlled bytes to the decoder directly. + * + * 8 MB comfortably accommodates compose YAML, image-list responses, and + * exec stream chunks while bounding the decode buffer a buggy or malicious + * peer can force the other side to allocate. + */ +export const MAX_FRAME_SIZE_BYTES = 8 * 1024 * 1024; + +/** + * Maximum concurrent multiplexed streams on a single tunnel. Beyond this the + * bridge refuses new loopback requests with 503 and the agent rejects new + * incoming streams with the appropriate error frame. Sized for normal Sencho + * fanout (UI tabs polling stats, logs, stack lifecycle) with substantial + * headroom; the realistic ceiling under load is closer to single digits per + * tunnel. + */ +export const MAX_STREAMS_PER_TUNNEL = 1024; + +/** + * Per-stream idle timeout. A stream that sees no inbound or outbound activity + * for this long is closed and removed from the stream map. Protects against + * leaked streams (one side crashed, peer never noticed) leaking memory over + * long uptimes. + */ +export const STREAM_IDLE_TIMEOUT_MS = 10 * 60 * 1000; + // --- Binary frame types (first byte of a binary WS frame) --- export enum BinaryFrameType { @@ -157,6 +190,14 @@ export function encodeJsonFrame(frame: JsonFrame): string { } export function decodeJsonFrame(raw: string): JsonFrame { + // Compare against UTF-8 byte length, not the string's UTF-16 code-unit + // count: multi-byte payloads can otherwise sneak ~3x the byte budget + // through a `raw.length` check. Cheap because Buffer.byteLength does + // not allocate. + const byteLen = Buffer.byteLength(raw, 'utf8'); + if (byteLen > MAX_FRAME_SIZE_BYTES) { + throw new Error(`json frame too large: ${byteLen} bytes`); + } const parsed = JSON.parse(raw); if (!parsed || typeof parsed !== 'object' || typeof parsed.t !== 'string') { throw new Error('invalid frame: missing type discriminator'); @@ -189,6 +230,9 @@ export function decodeBinaryFrame(buf: Buffer): DecodedBinaryFrame { if (buf.length < 5) { throw new Error(`binary frame too short: ${buf.length} bytes`); } + if (buf.length > MAX_FRAME_SIZE_BYTES) { + throw new Error(`binary frame too large: ${buf.length} bytes`); + } const type = buf.readUInt8(0) as BinaryFrameType; if (type !== BinaryFrameType.HttpReqBody && type !== BinaryFrameType.HttpResBody && @@ -228,7 +272,12 @@ export const PilotCloseCode = { /** * Monotonic stream id generator. Primary is the sole allocator. - * Wraps at 2^31 (well above practical per-tunnel concurrency). + * + * Wraps at 2^31. With MAX_STREAMS_PER_TUNNEL = 1024 the allocator + * cannot collide with a still-live stream during a single tunnel + * lifetime: the wrap distance (~2.1 billion) is more than six orders + * of magnitude larger than the cap. A new tunnel restarts the + * sequence at 1, so cross-tunnel reuse is also harmless. */ export class StreamIdAllocator { private next: number; diff --git a/backend/src/routes/metrics.ts b/backend/src/routes/metrics.ts index 77f60e0f..18c384fe 100644 --- a/backend/src/routes/metrics.ts +++ b/backend/src/routes/metrics.ts @@ -5,6 +5,7 @@ import DockerController, { globalDockerNetwork } from '../services/DockerControl import { DatabaseService } from '../services/DatabaseService'; import { CacheService } from '../services/CacheService'; import { NodeRegistry } from '../services/NodeRegistry'; +import { PilotTunnelManager } from '../services/PilotTunnelManager'; import { authMiddleware } from '../middleware/auth'; import { requireAdmin } from '../middleware/tierGates'; import { STATS_CACHE_TTL_MS, SYSTEM_STATS_CACHE_TTL_MS } from '../helpers/constants'; @@ -249,3 +250,19 @@ metricsRouter.get('/system/cache-stats', authMiddleware, async (req: Request, re res.status(500).json({ error: 'Failed to fetch cache stats' }); } }); + +/** + * Admin-only pilot tunnel observability. Counters reset on process restart + * by design (see PilotMetrics.ts). The per_node array is the load-bearing + * field: aggregate counters can hide a single tunnel that is flapping or + * sitting on a stuck write buffer. + */ +metricsRouter.get('/system/pilot-tunnels', authMiddleware, async (req: Request, res: Response): Promise => { + if (!requireAdmin(req, res)) return; + try { + res.json(PilotTunnelManager.getInstance().getMetricsSnapshot()); + } catch (error) { + console.error('Failed to fetch pilot tunnel metrics:', error); + res.status(500).json({ error: 'Failed to fetch pilot tunnel metrics' }); + } +}); diff --git a/backend/src/routes/nodes.ts b/backend/src/routes/nodes.ts index cc31ba5c..d0fef48b 100644 --- a/backend/src/routes/nodes.ts +++ b/backend/src/routes/nodes.ts @@ -5,6 +5,7 @@ import { authMiddleware } from '../middleware/auth'; import { requirePermission } from '../middleware/permissions'; import { rejectApiTokenScope } from '../middleware/apiTokenScope'; import { requireAdmiral } from '../middleware/tierGates'; +import { enrollmentLimiter } from '../middleware/rateLimiters'; import { DatabaseService } from '../services/DatabaseService'; import { NodeRegistry } from '../services/NodeRegistry'; import { CacheService } from '../services/CacheService'; @@ -120,7 +121,7 @@ nodesRouter.get('/:id', async (req: Request, res: Response) => { } }); -nodesRouter.post('/', async (req: Request, res: Response) => { +nodesRouter.post('/', enrollmentLimiter, async (req: Request, res: Response) => { if (rejectApiTokenScope(req, res, NODE_SCOPE_MESSAGE)) return; if (!requirePermission(req, res, 'node:manage')) return; try { @@ -190,7 +191,7 @@ nodesRouter.post('/', async (req: Request, res: Response) => { } }); -nodesRouter.post('/:id/pilot/enroll', async (req: Request, res: Response) => { +nodesRouter.post('/:id/pilot/enroll', enrollmentLimiter, async (req: Request, res: Response) => { if (rejectApiTokenScope(req, res, NODE_SCOPE_MESSAGE)) return; const nodeIdStr = req.params.id as string; if (!requirePermission(req, res, 'node:manage', 'node', nodeIdStr)) return; diff --git a/backend/src/server.ts b/backend/src/server.ts index 4e9423ef..a1b48eff 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -1,6 +1,7 @@ import http from 'http'; import type { Express } from 'express'; import { WebSocketServer } from 'ws'; +import { MAX_FRAME_SIZE_BYTES } from './pilot/protocol'; export interface SenchoServer { server: http.Server; @@ -21,8 +22,12 @@ export function createServer(app: Express): SenchoServer { const wss = new WebSocketServer({ noServer: true }); // Agents dial /api/pilot/tunnel; the handshake verifies a pilot_enroll or - // pilot_tunnel JWT, then hands the socket off to PilotTunnelManager. - const pilotTunnelWss = new WebSocketServer({ noServer: true }); + // pilot_tunnel JWT, then hands the socket off to PilotTunnelManager. The + // pilot tunnel multiplexes every HTTP / WS / Mesh-TCP byte for a remote + // node, so a single oversized frame from a buggy or malicious agent can + // bloat the gateway's decode buffer; cap the payload at the protocol-level + // ceiling rather than relying on the ws default (100 MB). + const pilotTunnelWss = new WebSocketServer({ noServer: true, maxPayload: MAX_FRAME_SIZE_BYTES }); return { server, wss, pilotTunnelWss }; } diff --git a/backend/src/services/PilotMetrics.ts b/backend/src/services/PilotMetrics.ts new file mode 100644 index 00000000..42e5976f --- /dev/null +++ b/backend/src/services/PilotMetrics.ts @@ -0,0 +1,39 @@ +/** + * PilotMetrics: in-memory counters for the pilot-agent reverse-tunnel + * subsystem. Strictly process-local — Sencho does not export metrics to any + * external sink (privacy posture, see CLAUDE.md). Counters reset on process + * restart by design; their purpose is operator support and debug, not + * long-term trend analysis. + * + * No general in-process metrics facility exists in the backend today, so this + * is a per-feature pattern. When a shared facility lands, this module should + * be replaced by an instance of it rather than grown. + */ + +interface Counters { + tunnels_total: number; + tunnels_replaced: number; + tunnels_rejected_capacity: number; + enroll_acks: number; + frame_decode_errors: number; +} + +class PilotMetricsImpl { + private counters: Counters = { + tunnels_total: 0, + tunnels_replaced: 0, + tunnels_rejected_capacity: 0, + enroll_acks: 0, + frame_decode_errors: 0, + }; + + public increment(name: K): void { + this.counters[name] += 1; + } + + public snapshot(): Counters { + return { ...this.counters }; + } +} + +export const PilotMetrics = new PilotMetricsImpl(); diff --git a/backend/src/services/PilotTunnelBridge.ts b/backend/src/services/PilotTunnelBridge.ts index 79f8029c..d4ef2656 100644 --- a/backend/src/services/PilotTunnelBridge.ts +++ b/backend/src/services/PilotTunnelBridge.ts @@ -5,6 +5,8 @@ import { WebSocket, WebSocketServer } from 'ws'; import { BinaryFrameType, DecodedBinaryFrame, + MAX_STREAMS_PER_TUNNEL, + STREAM_IDLE_TIMEOUT_MS, StreamIdAllocator, decodeBinaryFrame, decodeJsonFrame, @@ -13,17 +15,32 @@ import { wsDataToBuffer, wsDataToString, } from '../pilot/protocol'; +import { isDebugEnabled } from '../utils/debug'; +import { sanitizeForLog } from '../utils/safeLog'; +import { PilotMetrics } from './PilotMetrics'; const BUFFER_HIGH_WATER_MARK = 4 * 1024 * 1024; const PING_INTERVAL_MS = 30_000; +/** + * Poll cadence for the backpressure drain check. The `ws` WebSocket does not + * surface a usable 'drain' event, so when at least one stream is paused we + * sample `bufferedAmount` at this rate and resume / fan out 'drain' as soon + * as the buffer drops below the high-water mark. Dormant when nothing is + * paused, so steady-state cost is zero. + */ +const DRAIN_CHECK_INTERVAL_MS = 100; -interface HttpStreamState { +interface StreamMeta { + idleTimer?: NodeJS.Timeout; +} + +interface HttpStreamState extends StreamMeta { kind: 'http'; res: ServerResponse; headersWritten: boolean; } -interface WsStreamState { +interface WsStreamState extends StreamMeta { kind: 'ws'; rawSocket?: Socket; rawHead?: Buffer; @@ -31,7 +48,7 @@ interface WsStreamState { clientWs?: WebSocket; } -interface TcpStreamState { +interface TcpStreamState extends StreamMeta { kind: 'tcp'; handle: TcpStream; bytesIn: number; @@ -104,8 +121,11 @@ export class PilotTunnelBridge extends EventEmitter { private readonly streamIds = new StreamIdAllocator(); private readonly streams = new Map(); private readonly connectedAt = Date.now(); + private readonly pausedReqs = new Map(); + private readonly tcpAwaitingDrain = new Set(); private loopbackUrl = ''; private pingTimer?: NodeJS.Timeout; + private drainTimer?: NodeJS.Timeout; private closed = false; constructor(_nodeId: number, tunnelWs: WebSocket) { @@ -143,14 +163,6 @@ export class PilotTunnelBridge extends EventEmitter { this.pingTimer = setInterval(() => { if (this.tunnelWs.readyState !== WebSocket.OPEN) return; try { this.tunnelWs.ping(); } catch { /* surfaced via 'error' */ } - // Coarse drain fan-out for TCP streams: ws does not expose a - // socket-level 'drain' we can hook, so we let backpressure clear - // by the next ping cycle. - if (this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK) { - for (const s of this.streams.values()) { - if (s.kind === 'tcp' && s.accepted) s.handle.emit('drain'); - } - } }, PING_INTERVAL_MS); } @@ -166,16 +178,19 @@ export class PilotTunnelBridge extends EventEmitter { */ public openTcpStream(target: { stack: string; service: string; port: number }): TcpStream | null { if (!this.isOpen()) return null; + if (this.streams.size >= MAX_STREAMS_PER_TUNNEL) return null; const streamId = this.streamIds.allocate(); const handle = new TcpStream(streamId, this); - this.streams.set(streamId, { + const state: TcpStreamState = { kind: 'tcp', handle, bytesIn: 0, bytesOut: 0, openedAt: Date.now(), accepted: false, - }); + }; + this.streams.set(streamId, state); + this.refreshIdleTimer(streamId, state); this.sendJson({ t: 'tcp_open', s: streamId, @@ -197,13 +212,22 @@ export class PilotTunnelBridge extends EventEmitter { if (!this.isOpen()) return false; this.sendBinary(BinaryFrameType.TcpData, streamId, payload); s.bytesOut += payload.length; - return this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK; + this.refreshIdleTimer(streamId, s); + const ok = this.tunnelWs.bufferedAmount <= BUFFER_HIGH_WATER_MARK; + if (!ok) { + // Backpressure: caller will pause until 'drain'. Track this TCP + // stream so checkDrain knows to fire 'drain' on it (and so the + // drain timer keeps running for TCP-only backpressure scenarios). + this.tcpAwaitingDrain.add(streamId); + this.ensureDrainTimer(); + } + return ok; } /** @internal Called only by TcpStream.end / .destroy. */ public _closeTcpStream(streamId: number): void { if (!this.streams.has(streamId)) return; - this.streams.delete(streamId); + this.removeStream(streamId); this.sendJson({ t: 'tcp_close', s: streamId }); } @@ -225,8 +249,21 @@ export class PilotTunnelBridge extends EventEmitter { if (this.closed) return; this.closed = true; if (this.pingTimer) { clearInterval(this.pingTimer); this.pingTimer = undefined; } + this.stopDrainTimer(); + // Resume any paused IncomingMessage so its end event can fire and + // its parser unwinds; the loopback close below will tear down the + // socket either way, but this avoids holding a dangling parser + // state across teardown. + for (const [, req] of this.pausedReqs) { + try { req.resume(); } catch { /* ignore */ } + } + this.pausedReqs.clear(); + this.tcpAwaitingDrain.clear(); - for (const [, state] of this.streams) this.teardownStream(state); + for (const [, state] of this.streams) { + this.clearIdleTimer(state); + this.teardownStream(state); + } this.streams.clear(); try { this.tunnelWs.close(code, reason); } catch { /* ignore */ } @@ -243,9 +280,17 @@ export class PilotTunnelBridge extends EventEmitter { res.end('pilot tunnel not ready'); return; } + if (this.streams.size >= MAX_STREAMS_PER_TUNNEL) { + res.statusCode = 503; + res.setHeader('content-type', 'text/plain'); + res.end('pilot tunnel: stream cap reached'); + return; + } const streamId = this.streamIds.allocate(); - this.streams.set(streamId, { kind: 'http', res, headersWritten: false }); + const state: HttpStreamState = { kind: 'http', res, headersWritten: false }; + this.streams.set(streamId, state); + this.refreshIdleTimer(streamId, state); const headers: Record = {}; for (const [k, v] of Object.entries(req.headers)) { @@ -262,9 +307,13 @@ export class PilotTunnelBridge extends EventEmitter { }); req.on('data', (chunk: Buffer) => { - if (!this.streams.has(streamId)) return; + const s = this.streams.get(streamId); + if (!s) return; this.sendBinary(BinaryFrameType.HttpReqBody, streamId, chunk); - if (this.tunnelWs.bufferedAmount > BUFFER_HIGH_WATER_MARK) req.pause(); + this.refreshIdleTimer(streamId, s); + if (this.tunnelWs.bufferedAmount > BUFFER_HIGH_WATER_MARK) { + this.pauseRequest(streamId, req); + } }); req.on('end', () => { if (!this.streams.has(streamId)) return; @@ -273,13 +322,13 @@ export class PilotTunnelBridge extends EventEmitter { req.on('error', () => { const s = this.streams.get(streamId); if (s) this.teardownStream(s); - this.streams.delete(streamId); + this.removeStream(streamId); }); res.on('close', () => { // Client disconnected before response finished. if (this.streams.has(streamId)) { - this.streams.delete(streamId); + this.removeStream(streamId); this.sendJson({ t: 'http_err', s: streamId, code: 'tunnel_down', message: 'client aborted' }); } }); @@ -291,14 +340,21 @@ export class PilotTunnelBridge extends EventEmitter { socket.destroy(); return; } + if (this.streams.size >= MAX_STREAMS_PER_TUNNEL) { + socket.write('HTTP/1.1 503 Service Unavailable\r\n\r\n'); + socket.destroy(); + return; + } const streamId = this.streamIds.allocate(); - this.streams.set(streamId, { + const state: WsStreamState = { kind: 'ws', rawSocket: socket, rawHead: head, upgradeRequest: req, - }); + }; + this.streams.set(streamId, state); + this.refreshIdleTimer(streamId, state); const headers: Record = {}; for (const [k, v] of Object.entries(req.headers)) { @@ -316,12 +372,12 @@ export class PilotTunnelBridge extends EventEmitter { socket.on('error', () => { const s = this.streams.get(streamId); if (s) this.teardownStream(s); - this.streams.delete(streamId); + this.removeStream(streamId); }); socket.on('close', () => { if (this.streams.has(streamId)) { this.sendJson({ t: 'ws_close', s: streamId, code: 1006, reason: 'client closed' }); - this.streams.delete(streamId); + this.removeStream(streamId); } }); } @@ -341,8 +397,14 @@ export class PilotTunnelBridge extends EventEmitter { const frame = decodeJsonFrame(text); this.handleJsonFrame(frame); } - } catch { - // Malformed frame: kill the tunnel to force re-sync. + } catch (err) { + // Malformed frame: kill the tunnel to force re-sync. Diag-gated + // so a flood of malformed frames cannot drown the log; the + // tunnel close itself is the loud signal. + PilotMetrics.increment('frame_decode_errors'); + if (isDebugEnabled()) { + console.warn('[PilotBridge:diag] Malformed frame from agent:', sanitizeForLog((err as Error).message)); + } this.close(1002, 'protocol error'); } } @@ -358,13 +420,14 @@ export class PilotTunnelBridge extends EventEmitter { } catch { /* headers already sent or invalid */ } s.headersWritten = true; } + this.refreshIdleTimer(frame.s, s); break; } case 'http_res_end': { const s = this.streams.get(frame.s); if (!s || s.kind !== 'http') return; try { s.res.end(); } catch { /* ignore */ } - this.streams.delete(frame.s); + this.removeStream(frame.s); break; } case 'http_err': { @@ -378,7 +441,7 @@ export class PilotTunnelBridge extends EventEmitter { } else { this.teardownStream(s); } - this.streams.delete(frame.s); + this.removeStream(frame.s); break; } case 'ws_accept': { @@ -388,7 +451,10 @@ export class PilotTunnelBridge extends EventEmitter { s.clientWs = ws; s.rawSocket = undefined; s.rawHead = undefined; + this.refreshIdleTimer(frame.s, s); ws.on('message', (msg, isBin) => { + const cur = this.streams.get(frame.s); + if (cur) this.refreshIdleTimer(frame.s, cur); if (isBin) { this.sendBinary(BinaryFrameType.WsMessageBinary, frame.s, wsDataToBuffer(msg) ?? Buffer.alloc(0)); } else { @@ -398,11 +464,11 @@ export class PilotTunnelBridge extends EventEmitter { ws.on('close', (code, reason) => { if (this.streams.has(frame.s)) { this.sendJson({ t: 'ws_close', s: frame.s, code, reason: reason?.toString?.() }); - this.streams.delete(frame.s); + this.removeStream(frame.s); } }); ws.on('error', () => { - if (this.streams.has(frame.s)) this.streams.delete(frame.s); + if (this.streams.has(frame.s)) this.removeStream(frame.s); }); }); break; @@ -414,13 +480,14 @@ export class PilotTunnelBridge extends EventEmitter { s.rawSocket.write(`HTTP/1.1 ${frame.status} ${frame.message}\r\n\r\n`); s.rawSocket.destroy(); } catch { /* ignore */ } - this.streams.delete(frame.s); + this.removeStream(frame.s); break; } case 'ws_msg_text': { const s = this.streams.get(frame.s); if (!s || s.kind !== 'ws' || !s.clientWs) return; try { s.clientWs.send(frame.data); } catch { /* ignore */ } + this.refreshIdleTimer(frame.s, s); break; } case 'ws_close': { @@ -431,7 +498,7 @@ export class PilotTunnelBridge extends EventEmitter { } else if (s.rawSocket) { try { s.rawSocket.destroy(); } catch { /* ignore */ } } - this.streams.delete(frame.s); + this.removeStream(frame.s); break; } case 'ctrl': { @@ -445,9 +512,10 @@ export class PilotTunnelBridge extends EventEmitter { if (!s || s.kind !== 'tcp') return; if (frame.ok) { s.accepted = true; + this.refreshIdleTimer(frame.s, s); s.handle.emit('open'); } else { - this.streams.delete(frame.s); + this.removeStream(frame.s); s.handle.emit('error', new Error(frame.err ?? 'tcp_open rejected')); s.handle.emit('close'); } @@ -456,7 +524,7 @@ export class PilotTunnelBridge extends EventEmitter { case 'tcp_close': { const s = this.streams.get(frame.s); if (!s || s.kind !== 'tcp') return; - this.streams.delete(frame.s); + this.removeStream(frame.s); s.handle.emit('close'); break; } @@ -478,11 +546,13 @@ export class PilotTunnelBridge extends EventEmitter { s.headersWritten = true; } try { s.res.write(frame.payload); } catch { /* ignore */ } + this.refreshIdleTimer(frame.streamId, s); break; } case BinaryFrameType.WsMessageBinary: { if (s.kind !== 'ws' || !s.clientWs) return; try { s.clientWs.send(frame.payload, { binary: true }); } catch { /* ignore */ } + this.refreshIdleTimer(frame.streamId, s); break; } case BinaryFrameType.HttpReqBody: @@ -491,6 +561,7 @@ export class PilotTunnelBridge extends EventEmitter { case BinaryFrameType.TcpData: { if (s.kind !== 'tcp') return; s.bytesIn += frame.payload.length; + this.refreshIdleTimer(frame.streamId, s); s.handle.emit('data', frame.payload); break; } @@ -506,6 +577,84 @@ export class PilotTunnelBridge extends EventEmitter { // --- Helpers --- + private pauseRequest(streamId: number, req: IncomingMessage): void { + if (this.pausedReqs.has(streamId)) return; + try { req.pause(); } catch { /* ignore */ } + this.pausedReqs.set(streamId, req); + this.ensureDrainTimer(); + } + + private ensureDrainTimer(): void { + if (this.drainTimer || this.closed) return; + this.drainTimer = setInterval(() => this.checkDrain(), DRAIN_CHECK_INTERVAL_MS); + } + + private checkDrain(): void { + if (this.closed) { + this.stopDrainTimer(); + return; + } + if (this.tunnelWs.bufferedAmount > BUFFER_HIGH_WATER_MARK) return; + for (const [, req] of this.pausedReqs) { + try { req.resume(); } catch { /* ignore */ } + } + this.pausedReqs.clear(); + // Fire 'drain' only on TCP streams that signaled backpressure; do not + // wake every accepted TCP stream because that violates the documented + // event contract on TcpStream. + for (const streamId of this.tcpAwaitingDrain) { + const s = this.streams.get(streamId); + if (s && s.kind === 'tcp' && s.accepted) s.handle.emit('drain'); + } + this.tcpAwaitingDrain.clear(); + this.stopDrainTimer(); + } + + private stopDrainTimer(): void { + if (this.drainTimer) { + clearInterval(this.drainTimer); + this.drainTimer = undefined; + } + } + + private refreshIdleTimer(streamId: number, state: StreamState): void { + if (state.idleTimer) clearTimeout(state.idleTimer); + state.idleTimer = setTimeout(() => this.onStreamIdle(streamId), STREAM_IDLE_TIMEOUT_MS); + } + + private clearIdleTimer(state: StreamState): void { + if (state.idleTimer) { + clearTimeout(state.idleTimer); + state.idleTimer = undefined; + } + } + + private removeStream(streamId: number): void { + const s = this.streams.get(streamId); + if (!s) return; + this.clearIdleTimer(s); + this.streams.delete(streamId); + this.pausedReqs.delete(streamId); + this.tcpAwaitingDrain.delete(streamId); + } + + private onStreamIdle(streamId: number): void { + const s = this.streams.get(streamId); + if (!s) return; + // Tear down the loopback side and tell the agent to release its half. + this.teardownStream(s); + this.streams.delete(streamId); + this.pausedReqs.delete(streamId); + this.tcpAwaitingDrain.delete(streamId); + if (s.kind === 'tcp') { + this.sendJson({ t: 'tcp_close', s: streamId }); + } else if (s.kind === 'ws') { + this.sendJson({ t: 'ws_close', s: streamId, code: 1001, reason: 'idle' }); + } else { + this.sendJson({ t: 'http_err', s: streamId, code: 'timeout', message: 'stream idle timeout' }); + } + } + private sendJson(frame: Parameters[0]): void { if (this.tunnelWs.readyState !== WebSocket.OPEN) return; try { this.tunnelWs.send(encodeJsonFrame(frame)); } catch { /* ignore */ } diff --git a/backend/src/services/PilotTunnelManager.ts b/backend/src/services/PilotTunnelManager.ts index 576cbee5..48a2d849 100644 --- a/backend/src/services/PilotTunnelManager.ts +++ b/backend/src/services/PilotTunnelManager.ts @@ -3,6 +3,35 @@ import WebSocket from 'ws'; import { PilotTunnelBridge } from './PilotTunnelBridge'; import { DatabaseService } from './DatabaseService'; import { PilotCloseCode } from '../pilot/protocol'; +import { isDebugEnabled } from '../utils/debug'; +import { PilotMetrics } from './PilotMetrics'; + +/** + * Soft warning threshold: a single instance handling more than this many + * concurrent pilot tunnels is unusual and likely indicates a reconnect storm + * or operator misconfiguration. Logged at WARN. + */ +const PILOT_TUNNEL_SOFT_LIMIT = 128; + +/** + * Hard ceiling on concurrent pilot tunnels per primary. Beyond this the + * gateway refuses to register new tunnels (the upgrade handler closes the + * socket with 1013 try-again-later) so a runaway reconnect storm cannot + * exhaust gateway memory. + */ +const PILOT_TUNNEL_HARD_LIMIT = 256; + +/** + * Thrown by registerTunnel when the system-wide cap is exceeded. The pilot + * upgrade handler catches this and closes the WebSocket cleanly so the agent + * backs off rather than tight-looping. + */ +export class PilotTunnelCapacityError extends Error { + constructor(public readonly limit: number) { + super(`pilot tunnel cap (${limit}) reached`); + this.name = 'PilotTunnelCapacityError'; + } +} /** * PilotTunnelManager: singleton registry of active pilot tunnels. @@ -20,6 +49,7 @@ import { PilotCloseCode } from '../pilot/protocol'; export class PilotTunnelManager extends EventEmitter { private static instance: PilotTunnelManager; private bridges: Map = new Map(); + private softWarned = false; private constructor() { super(); @@ -42,11 +72,31 @@ export class PilotTunnelManager extends EventEmitter { */ public async registerTunnel(nodeId: number, ws: WebSocket, agentVersion?: string): Promise { const existing = this.bridges.get(nodeId); + const replaced = existing != null; if (existing) { existing.close(PilotCloseCode.Replaced, 'replaced by newer tunnel'); this.bridges.delete(nodeId); } + // Hard cap: only counts tunnels for *other* nodes since we just + // released the matching slot above. A reconnect by the same node + // does not consume new capacity. + if (this.bridges.size >= PILOT_TUNNEL_HARD_LIMIT) { + PilotMetrics.increment('tunnels_rejected_capacity'); + throw new PilotTunnelCapacityError(PILOT_TUNNEL_HARD_LIMIT); + } + + // Bump the replaced counter only after the cap check passes, so a + // rejection does not double-count as both a replacement and a + // capacity rejection. + if (replaced) PilotMetrics.increment('tunnels_replaced'); + if (this.bridges.size >= PILOT_TUNNEL_SOFT_LIMIT && !this.softWarned) { + console.warn(`[Pilot] Active tunnel count at soft limit (${this.bridges.size}/${PILOT_TUNNEL_HARD_LIMIT}); reconnect storm or runaway enrollment likely.`); + this.softWarned = true; + } else if (this.bridges.size < PILOT_TUNNEL_SOFT_LIMIT) { + this.softWarned = false; + } + const bridge = new PilotTunnelBridge(nodeId, ws); bridge.once('closed', () => { if (this.bridges.get(nodeId) === bridge) { @@ -64,9 +114,35 @@ export class PilotTunnelManager extends EventEmitter { pilot_last_seen: Date.now(), pilot_agent_version: agentVersion ?? null, }); + PilotMetrics.increment('tunnels_total'); + if (isDebugEnabled()) { + console.log('[PilotMgr:diag] Tunnel registered:', { nodeId, active: this.bridges.size }); + } this.emit('tunnel-up', nodeId); } + /** + * Per-tunnel breakdown for the metrics endpoint. Includes the + * loopback-relative connectedAt and bufferedAmount so one-bad-node cases + * stay visible (an aggregate hides a single tunnel sitting on a stuck + * write buffer). + */ + public getMetricsSnapshot(): { + counters: ReturnType; + tunnels_open: number; + per_node: Array<{ nodeId: number; connectedAt: number; bufferedAmount: number }>; + } { + return { + counters: PilotMetrics.snapshot(), + tunnels_open: this.bridges.size, + per_node: Array.from(this.bridges.entries()).map(([nodeId, bridge]) => ({ + nodeId, + connectedAt: bridge.getConnectedAt(), + bufferedAmount: bridge.getBufferedAmount(), + })), + }; + } + /** * Return the loopback base URL (http://127.0.0.1:PORT) for a node's active * tunnel, or null if no tunnel is currently registered. diff --git a/backend/src/websocket/pilotTunnel.ts b/backend/src/websocket/pilotTunnel.ts index 51420747..08680374 100644 --- a/backend/src/websocket/pilotTunnel.ts +++ b/backend/src/websocket/pilotTunnel.ts @@ -4,7 +4,8 @@ import type { WebSocketServer } from 'ws'; import jwt from 'jsonwebtoken'; import crypto from 'crypto'; import { DatabaseService } from '../services/DatabaseService'; -import { PilotTunnelManager } from '../services/PilotTunnelManager'; +import { PilotTunnelCapacityError, PilotTunnelManager } from '../services/PilotTunnelManager'; +import { PilotMetrics } from '../services/PilotMetrics'; import { encodeJsonFrame as encodePilotJsonFrame, PROTOCOL_VERSION as PILOT_PROTOCOL_VERSION } from '../pilot/protocol'; import { getErrorMessage } from '../utils/errors'; import { rejectUpgrade as rejectSocket } from './reject'; @@ -63,6 +64,7 @@ export async function handlePilotTunnel( jwtSecret, { expiresIn: '365d' }, ); + PilotMetrics.increment('enroll_acks'); } const agentVersionHeader = req.headers['x-sencho-agent-version']; @@ -90,6 +92,12 @@ export async function handlePilotTunnel( try { await PilotTunnelManager.getInstance().registerTunnel(decoded.nodeId!, ws, agentVersion); } catch (err) { + if (err instanceof PilotTunnelCapacityError) { + // 1013 (Try Again Later) signals the agent to back off rather than + // tight-loop reconnect on a saturated gateway. + try { ws.close(1013, 'pilot tunnel cap reached'); } catch { /* ignore */ } + return; + } console.error('[Pilot] Failed to register tunnel:', getErrorMessage(err, 'unknown')); try { ws.close(1011, 'registration failed'); } catch { /* ignore */ } } diff --git a/docs/features/pilot-agent.mdx b/docs/features/pilot-agent.mdx index 60bdbde2..7964f7eb 100644 --- a/docs/features/pilot-agent.mdx +++ b/docs/features/pilot-agent.mdx @@ -71,6 +71,34 @@ The agent reconnects automatically if the tunnel drops (transient network blip, If the agent container is destroyed before its first successful connect, or the enrollment token expires before you paste it on the remote, open the node's edit dialog on the primary and click **Regenerate enrollment token**. A fresh 15-minute token is issued and the previous tunnel (if any) is disconnected so the new agent replaces it cleanly. +## Self-signed primary TLS certs + +If your primary terminates TLS with an internal CA and the agent cannot validate the certificate against the system trust store, point the agent at the CA bundle with `SENCHO_PILOT_CA_FILE`. Mount the bundle into the agent container and add the env var: + +```bash +docker run -d --restart=unless-stopped --name sencho-agent \ + -v /var/run/docker.sock:/var/run/docker.sock \ + -v sencho-agent-data:/app/data \ + -v /opt/docker/sencho:/app/compose \ + -v /etc/ssl/internal-ca.pem:/etc/ssl/internal-ca.pem:ro \ + -e SENCHO_MODE=pilot \ + -e SENCHO_PRIMARY_URL=https://sencho.internal.example.com \ + -e SENCHO_ENROLL_TOKEN= \ + -e SENCHO_PILOT_CA_FILE=/etc/ssl/internal-ca.pem \ + saelix/sencho:latest +``` + +The agent uses the bundle as the only trust anchor for the tunnel WebSocket. TLS verification stays on; there is no env var to disable verification globally because that would defeat the credential trust model. + +## Resource limits + +Each tunnel has fixed protocol-level ceilings to keep one misbehaving agent from impacting the primary: + +- **Frame size:** individual WebSocket frames are capped at 8 MB. Single requests and responses larger than that are rejected and the tunnel reconnects. +- **Concurrent streams:** at most 1024 multiplexed HTTP, WebSocket, or Mesh-TCP streams per tunnel. Above the cap the bridge returns 503 and the agent rejects new incoming streams with a typed error frame. +- **Stream idle:** any stream with no activity for 10 minutes is closed and removed. +- **System-wide tunnels:** a single primary accepts up to 256 concurrent pilot tunnels. Past the soft warning at 128 the primary logs a WARN; past the hard cap of 256 new tunnels are refused with WebSocket close code 1013 (Try Again Later) so the agent backs off rather than tight-looping. + ## Troubleshooting @@ -93,4 +121,20 @@ If the agent container is destroyed before its first successful connect, or the The agent's persisted tunnel credential is signed with the primary's JWT secret. If that secret is rotated or the primary is rebuilt from scratch, existing tunnels stop verifying. Regenerate enrollment for the affected node and restart the agent container so it consumes the fresh token. + + + The primary is at the system-wide concurrent tunnel cap of 256. A reconnect storm or runaway enrollment is the usual cause. Inspect the primary's logs for the `[Pilot] Active tunnel count at soft limit` warning that fires at 128 to find the trend, and check `GET /api/system/pilot-tunnels` (admin-only) for the per-node breakdown to identify a flapping agent. The agent backs off and retries automatically once headroom returns. + + + + A single tunnel is at the 1024 concurrent stream cap. The most common trigger is a UI session that opened many long-poll streams and never closed them, or a script firing parallel API calls without bound. Reload the affected browser tab so its WebSocket and SSE streams reset. If the cap is being hit by automation, throttle the caller; the cap protects the primary's memory from a runaway agent or proxy. + + + + The agent sent a frame larger than the 8 MB ceiling, or the wire decoder rejected a malformed frame. Run the agent with developer mode enabled on the primary (Settings → Developer) to surface the diagnostic decode log, then check the primary's logs for `[PilotBridge:diag] Malformed frame from agent`. The agent reconnects automatically; persistent failures usually mean a version mismatch between primary and agent images. + + + + Enrollment endpoints are rate-limited to 10 requests per minute per user (or per IP, when unauthenticated) so a script accidentally minting tokens in a loop cannot exhaust the database. Wait sixty seconds and retry. If you genuinely need to bulk-enroll many nodes, space the requests out or contact the operator who runs the primary about raising the limit. +