diff --git a/.env.example b/.env.example index 3ee937fd..251dfa9a 100644 --- a/.env.example +++ b/.env.example @@ -120,9 +120,9 @@ SSO_CALLBACK_URL= # live regardless. This flag controls UI discovery only. SENCHO_EXPERIMENTAL=false -# Idle teardown (in ms) for on-demand mesh tunnels that central opens -# to Distributed API (proxy-mode) remotes. After this many milliseconds -# of zero active mesh streams, central closes the WebSocket; the next -# mesh dial re-opens it. Defaults to 5 minutes. Set to 0 to keep -# tunnels open until the remote drops them or central shuts down. -SENCHO_MESH_PROXY_TUNNEL_IDLE_MS=300000 +# Idle teardown for the persistent mesh proxy tunnel central opens to +# Distributed API (proxy-mode) remotes. The tunnel stays open for the +# life of the WebSocket. Set a positive number of milliseconds to +# enable idle teardown after that many ms of zero active mesh streams; +# the next mesh dial re-opens it. Default 0 (persistent). +SENCHO_MESH_PROXY_TUNNEL_IDLE_MS=0 diff --git a/backend/src/__tests__/mesh-inspect-remote.test.ts b/backend/src/__tests__/mesh-inspect-remote.test.ts index 2ccc6c57..38c82f04 100644 --- a/backend/src/__tests__/mesh-inspect-remote.test.ts +++ b/backend/src/__tests__/mesh-inspect-remote.test.ts @@ -148,4 +148,44 @@ describe('MeshService.inspectStackServices dispatch (C-3 fix)', () => { expect(fetchSpy).not.toHaveBeenCalled(); db.deleteNode(remoteNodeId); }); + + it('logs an offline-shaped warn (not a generic error) when proxyFetch throws MeshError no_target', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const remoteNodeId = db.addNode({ + name: 'inspect-remote-no-target', + type: 'remote', + mode: 'pilot_agent', + compose_dir: '/tmp', + is_default: false, + api_url: '', + api_token: '', + }); + + // getProxyTarget returns null -> proxyFetch raises MeshError('no_target'). + // The catch branch must recognise no_target alongside push_failed and + // emit console.warn (operator-friendly), not console.error (which the + // Routing tab surfaces as an unexpected fault). + vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue(null); + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => { /* swallow */ }); + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => { /* swallow */ }); + + const out = await (svc as unknown as { inspectStackServices: (n: number, s: string) => Promise }) + .inspectStackServices(remoteNodeId, 'audit-mesh-pilot'); + + expect(out).toEqual([]); + const warnedAboutNoTarget = warnSpy.mock.calls.some((args) => + String(args[0] ?? '').includes('inspectStackServices: unreachable') && + String(args[0] ?? '').includes('no_target'), + ); + expect(warnedAboutNoTarget).toBe(true); + // No "remote unreachable" error log: that path is reserved for + // unexpected exceptions, not the known no_target / push_failed pair. + const erroredAsUnreachable = errorSpy.mock.calls.some((args) => + String(args[0] ?? '').includes('inspectStackServices remote unreachable'), + ); + expect(erroredAsUnreachable).toBe(false); + + db.deleteNode(remoteNodeId); + }); }); diff --git a/backend/src/__tests__/mesh-remove-override-remote.test.ts b/backend/src/__tests__/mesh-remove-override-remote.test.ts new file mode 100644 index 00000000..abcda5f3 --- /dev/null +++ b/backend/src/__tests__/mesh-remove-override-remote.test.ts @@ -0,0 +1,128 @@ +/** + * Narrow contract test for `MeshService.removeOverrideFromNode` against a + * remote node. Pins the HTTP shape (method, path, headers) so a regression + * inside `removeOverrideFromNode` itself is caught even when callers like + * `disableForNode` are tested against a mock of the helper. + * + * Same fetch-spy pattern as `mesh-inspect-remote.test.ts`. + */ +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; +import { setupTestDb, cleanupTestDb } from './helpers/setupTestDb'; + +let tmpDir: string; +let MeshService: typeof import('../services/MeshService').MeshService; +let DatabaseService: typeof import('../services/DatabaseService').DatabaseService; +let NodeRegistry: typeof import('../services/NodeRegistry').NodeRegistry; + +beforeAll(async () => { + tmpDir = await setupTestDb(); + ({ MeshService } = await import('../services/MeshService')); + ({ DatabaseService } = await import('../services/DatabaseService')); + ({ NodeRegistry } = await import('../services/NodeRegistry')); +}); + +afterAll(() => { + vi.restoreAllMocks(); + cleanupTestDb(tmpDir); +}); + +afterEach(() => { + vi.restoreAllMocks(); +}); + +describe('MeshService.removeOverrideFromNode (remote dispatch)', () => { + it('issues DELETE /api/mesh/local-override/ with Authorization and tier headers', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const remoteNodeId = db.addNode({ + name: 'remove-override-remote-test', + type: 'remote', + mode: 'proxy', + compose_dir: '/tmp', + is_default: false, + api_url: 'https://remote.example.com:1852', + api_token: 'remote-tok', + }); + + vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue({ + apiUrl: 'https://remote.example.com:1852', + apiToken: 'remote-tok', + }); + + const fetchMock = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValue(new Response('ok', { status: 200 })); + + await svc.removeOverrideFromNode(remoteNodeId, 'sample-stack'); + + expect(fetchMock).toHaveBeenCalledTimes(1); + const call = fetchMock.mock.calls[0]; + expect(String(call[0])).toBe('https://remote.example.com:1852/api/mesh/local-override/sample-stack'); + const init = call[1] as { method: string; headers: Record }; + expect(init.method).toBe('DELETE'); + expect(init.headers['Authorization']).toBe('Bearer remote-tok'); + expect(init.headers).toHaveProperty('x-sencho-tier'); + expect(init.headers).toHaveProperty('x-sencho-variant'); + + db.deleteNode(remoteNodeId); + }); + + it('url-encodes the stack name in the path', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const remoteNodeId = db.addNode({ + name: 'remove-override-encode-test', + type: 'remote', + mode: 'proxy', + compose_dir: '/tmp', + is_default: false, + api_url: 'https://remote.example.com:1852', + api_token: 'remote-tok', + }); + + vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue({ + apiUrl: 'https://remote.example.com:1852', + apiToken: 'remote-tok', + }); + + const fetchMock = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValue(new Response('ok', { status: 200 })); + + await svc.removeOverrideFromNode(remoteNodeId, 'stack with space'); + + const call = fetchMock.mock.calls[0]; + expect(String(call[0])).toBe('https://remote.example.com:1852/api/mesh/local-override/stack%20with%20space'); + + db.deleteNode(remoteNodeId); + }); + + it('swallows network errors from the remote so the disable cascade can continue', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const remoteNodeId = db.addNode({ + name: 'remove-override-error-test', + type: 'remote', + mode: 'proxy', + compose_dir: '/tmp', + is_default: false, + api_url: 'https://remote.example.com:1852', + api_token: 'remote-tok', + }); + + vi.spyOn(NodeRegistry.getInstance(), 'getProxyTarget').mockReturnValue({ + apiUrl: 'https://remote.example.com:1852', + apiToken: 'remote-tok', + }); + + vi.spyOn(globalThis, 'fetch').mockRejectedValue(new Error('ECONNREFUSED')); + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => { /* silence */ }); + + // Must not throw: the remote being offline is a tolerable condition; + // the cascade upstream uses Promise.allSettled and continues. + await expect(svc.removeOverrideFromNode(remoteNodeId, 'sample-stack')).resolves.toBeUndefined(); + expect(warnSpy).toHaveBeenCalled(); + + db.deleteNode(remoteNodeId); + }); +}); diff --git a/backend/src/__tests__/mesh-service.test.ts b/backend/src/__tests__/mesh-service.test.ts index eab9896f..303a0a86 100644 --- a/backend/src/__tests__/mesh-service.test.ts +++ b/backend/src/__tests__/mesh-service.test.ts @@ -339,6 +339,118 @@ describe('MeshService.optInStack', () => { }); }); +describe('MeshService.disableForNode', () => { + it('redeploys affected stacks and cascades across the fleet on node disable', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const localNodeId = db.getNodes()[0].id; + const remoteNodeId = db.addNode({ + name: 'remote-pilot', type: 'remote', mode: 'pilot_agent', + compose_dir: '/tmp', is_default: false, api_url: '', api_token: '', + }); + db.setNodeMeshEnabled(localNodeId, true); + db.insertMeshStack(localNodeId, 'alpha', 'setup'); + db.insertMeshStack(localNodeId, 'beta', 'setup'); + db.insertMeshStack(remoteNodeId, 'gamma', 'setup'); + + const regenSpy = vi.spyOn( + svc as unknown as { regenerateOverridesAcrossFleet: (n?: number, s?: string) => Promise }, + 'regenerateOverridesAcrossFleet', + ).mockResolvedValue(undefined); + const cascadeSpy = vi.spyOn( + svc as unknown as { cascadeRecomposeAcrossFleet: (n: number | undefined, s: string | undefined, a: string) => void }, + 'cascadeRecomposeAcrossFleet', + ).mockImplementation(() => { /* noop */ }); + const triggerSpy = vi.spyOn(svc, 'triggerRedeploy').mockImplementation(() => { /* noop */ }); + const removeSpy = vi.spyOn(svc, 'removeOverrideFromNode').mockResolvedValue(undefined); + + await svc.disableForNode(localNodeId, 'tester'); + + expect(db.getNodeMeshEnabled(localNodeId)).toBe(false); + // Disabled-node rows are deleted before the cascade so listMeshStacks + // does not return them. The other node's row stays. + expect(db.listMeshStacks(localNodeId)).toEqual([]); + expect(db.listMeshStacks(remoteNodeId).map((s) => s.stack_name)).toEqual(['gamma']); + + expect(regenSpy).toHaveBeenCalledTimes(1); + // Cascade walks every remaining mesh_stacks row, no skip tuple. + expect(cascadeSpy).toHaveBeenCalledWith(undefined, undefined, 'tester'); + // Each disabled-node stack triggers a direct redeploy so its + // container detaches from sencho_mesh. Cascade is mocked, so the + // only triggerRedeploy calls come from the disable loop itself + // (twice, not three: gamma stays put on the remote node). + expect(triggerSpy).toHaveBeenCalledTimes(2); + expect(triggerSpy).toHaveBeenCalledWith(localNodeId, 'alpha', 'tester'); + expect(triggerSpy).toHaveBeenCalledWith(localNodeId, 'beta', 'tester'); + // disableForNode routes through removeOverrideFromNode (not the + // local-only removeStackOverride) so remote nodes also have their + // pushed override files deleted via DELETE /api/mesh/local-override. + expect(removeSpy).toHaveBeenCalledTimes(2); + expect(removeSpy).toHaveBeenCalledWith(localNodeId, 'alpha'); + expect(removeSpy).toHaveBeenCalledWith(localNodeId, 'beta'); + + db.deleteNode(remoteNodeId); + }); + + it('dispatches removeOverrideFromNode for each stack when the disabled node is remote', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const remoteNodeId = db.addNode({ + name: 'remote-pilot-disable', type: 'remote', mode: 'pilot_agent', + compose_dir: '/tmp', is_default: false, api_url: '', api_token: '', + }); + db.setNodeMeshEnabled(remoteNodeId, true); + db.insertMeshStack(remoteNodeId, 'delta', 'setup'); + db.insertMeshStack(remoteNodeId, 'epsilon', 'setup'); + + vi.spyOn( + svc as unknown as { regenerateOverridesAcrossFleet: (n?: number, s?: string) => Promise }, + 'regenerateOverridesAcrossFleet', + ).mockResolvedValue(undefined); + vi.spyOn( + svc as unknown as { cascadeRecomposeAcrossFleet: (n: number | undefined, s: string | undefined, a: string) => void }, + 'cascadeRecomposeAcrossFleet', + ).mockImplementation(() => { /* noop */ }); + vi.spyOn(svc, 'triggerRedeploy').mockImplementation(() => { /* noop */ }); + const removeSpy = vi.spyOn(svc, 'removeOverrideFromNode').mockResolvedValue(undefined); + + await svc.disableForNode(remoteNodeId, 'tester'); + + expect(db.getNodeMeshEnabled(remoteNodeId)).toBe(false); + // Remote-node disable must dispatch the remote-aware helper so the + // override file pushed earlier via applyLocalOverride is deleted + // on the peer via DELETE /api/mesh/local-override/:stack. + expect(removeSpy).toHaveBeenCalledTimes(2); + expect(removeSpy).toHaveBeenCalledWith(remoteNodeId, 'delta'); + expect(removeSpy).toHaveBeenCalledWith(remoteNodeId, 'epsilon'); + + db.deleteNode(remoteNodeId); + }); + + it('defaults the actor when none is supplied so legacy callers still log a non-empty actor', async () => { + const svc = MeshService.getInstance(); + const db = DatabaseService.getInstance(); + const localNodeId = db.getNodes()[0].id; + db.setNodeMeshEnabled(localNodeId, true); + db.insertMeshStack(localNodeId, 'solo', 'setup'); + + vi.spyOn(svc as unknown as { regenerateOverridesAcrossFleet: () => Promise }, 'regenerateOverridesAcrossFleet') + .mockResolvedValue(undefined); + const cascadeSpy = vi.spyOn( + svc as unknown as { cascadeRecomposeAcrossFleet: (n: number | undefined, s: string | undefined, a: string) => void }, + 'cascadeRecomposeAcrossFleet', + ).mockImplementation(() => { /* noop */ }); + const triggerSpy = vi.spyOn(svc, 'triggerRedeploy').mockImplementation(() => { /* noop */ }); + vi.spyOn(svc, 'removeOverrideFromNode').mockResolvedValue(undefined); + + await svc.disableForNode(localNodeId); + + expect(db.getNodeMeshEnabled(localNodeId)).toBe(false); + expect(cascadeSpy.mock.calls[0][2]).toBe('system:mesh.disable'); + expect(triggerSpy.mock.calls[0][2]).toBe('system:mesh.disable'); + }); +}); + describe('MeshService activity log', () => { it('keeps the most recent events under the 1000-cap', () => { const svc = MeshService.getInstance(); diff --git a/backend/src/__tests__/pilot-bridge-reverse.test.ts b/backend/src/__tests__/pilot-bridge-reverse.test.ts index 2c1c110c..d95c05fb 100644 --- a/backend/src/__tests__/pilot-bridge-reverse.test.ts +++ b/backend/src/__tests__/pilot-bridge-reverse.test.ts @@ -212,4 +212,84 @@ describe('PilotTunnelBridge handles tcp_open_reverse (Phase B)', () => { bridge.close(); vi.restoreAllMocks(); }); + + it('buffers TcpData arriving during ensureBridge and flushes it on target open (reverse-relay early-data fix)', async () => { + const mockWs = makeMockTunnelWs(); + const bridge = new PilotTunnelBridge(1, mockWs as unknown as WebSocket); + await bridge.start(); + + // Fake target stream: an EventEmitter with a write() method that + // captures bytes. We control when 'open' fires to widen the + // race window the fix guards. + const writes: Buffer[] = []; + const fakeTargetStream = new EventEmitter() as EventEmitter & { + write: (b: Buffer) => void; + }; + fakeTargetStream.write = (b: Buffer) => { writes.push(b); }; + + const fakeTargetBridge = { + openTcpStream: vi.fn(() => fakeTargetStream), + getBufferedAmount: () => 0, + }; + + const { PilotTunnelManager } = await import('../services/PilotTunnelManager'); + // Delay ensureBridge so a TcpData frame fired back-to-back with + // tcp_open_reverse lands while acceptReverseRelay is still + // waiting on the target bridge. Without the fix the bytes are + // dropped at the lookup-miss path in handleBinaryFrame. + vi.spyOn(PilotTunnelManager.getInstance(), 'ensureBridge') + .mockImplementation(() => new Promise((r) => setTimeout(() => r(fakeTargetBridge as never), 50))); + + const { NodeRegistry } = await import('../services/NodeRegistry'); + const localNodeId = NodeRegistry.getInstance().getDefaultNodeId(); + const remoteTargetNodeId = localNodeId + 99; + + const s = AGENT_REVERSE_ID_BASE + 7; + mockWs.emit('message', encodeJsonFrame({ + t: 'tcp_open_reverse', s, + targetNodeId: remoteTargetNodeId, stack: 'real', service: 'svc', port: 8080, + }), false); + // Immediately push the request bytes while ensureBridge is in flight. + mockWs.emit('message', + encodeBinaryFrame(BinaryFrameType.TcpData, s, Buffer.from('POST /hook HTTP/1.1\r\n\r\n')), + true); + + // Wait until acceptReverseRelay has actually swapped the reservation + // to kind:'reverse_relay' (and targetOpen is still false because the + // target hasn't emitted 'open'). Polling the bridge's stream state + // directly is more deterministic than polling openTcpStream.mock + // calls, since the state swap is synchronous after openTcpStream + // returns but the JS scheduler decides when the test sees it. + const bridgeStreams = (bridge as unknown as { streams: Map }).streams; + await waitFor(() => { + const cur = bridgeStreams.get(s); + return (cur?.kind === 'reverse_relay' && cur.targetOpen === false) ? true : undefined; + }); + // Second TcpData frame in the targetOpen===false window after the + // state swap. Without the reverse_relay branch's pendingData buffer, + // these bytes would be silently dropped. + mockWs.emit('message', + encodeBinaryFrame(BinaryFrameType.TcpData, s, Buffer.from('mid-window')), + true); + // Fire 'open' on the fake target so the relay flushes its pending buffer. + fakeTargetStream.emit('open'); + + const ack = await waitFor(() => findAck(mockWs, s)); + expect(ack.ok).toBe(true); + // Both pre-openTcpStream and post-state-swap buffered bytes were + // flushed exactly once, intact, in order. + await waitFor(() => (Buffer.concat(writes).toString().includes('mid-window')) ? true : undefined); + const combined = Buffer.concat(writes).toString(); + expect(combined).toBe('POST /hook HTTP/1.1\r\n\r\nmid-window'); + + // Subsequent post-open writes also flow. + mockWs.emit('message', + encodeBinaryFrame(BinaryFrameType.TcpData, s, Buffer.from('trailer')), + true); + await waitFor(() => (Buffer.concat(writes).toString().endsWith('trailer')) ? true : undefined); + expect(Buffer.concat(writes).toString()).toBe('POST /hook HTTP/1.1\r\n\r\nmid-windowtrailer'); + + bridge.close(); + vi.restoreAllMocks(); + }); }); diff --git a/backend/src/__tests__/upgrade-order.test.ts b/backend/src/__tests__/upgrade-order.test.ts index 06848f5b..c5a729ae 100644 --- a/backend/src/__tests__/upgrade-order.test.ts +++ b/backend/src/__tests__/upgrade-order.test.ts @@ -8,13 +8,14 @@ * product paths. This test pins each position by observing behavior that * could only originate from the expected handler. */ -import { describe, it, expect, beforeAll, afterAll } from 'vitest'; +import { describe, it, expect, beforeAll, afterAll, afterEach } from 'vitest'; import WebSocket from 'ws'; import jwt from 'jsonwebtoken'; import crypto from 'crypto'; import type { AddressInfo } from 'net'; import { setupTestDb, cleanupTestDb, TEST_USERNAME, TEST_JWT_SECRET } from './helpers/setupTestDb'; import { generateApiToken } from '../utils/apiTokenFormat'; +import { createTestApiToken } from './helpers/apiTokenTestHelper'; describe('WebSocket upgrade dispatch order', () => { let tmpDir: string; @@ -43,19 +44,35 @@ describe('WebSocket upgrade dispatch order', () => { const token = jwt.sign({ username: TEST_USERNAME }, TEST_JWT_SECRET, { expiresIn: '1m' }); sessionCookie = `sencho_token=${token}`; + + // Existing /api/mesh/proxy-tunnel scope tests assume the receiver + // license clears the Admiral check. Set the license to Admiral so + // the credential-only assertions below still hold; the dedicated + // license-gating describe block flips and restores per-test. + DatabaseService.getInstance().setSystemState('license_status', 'active'); + DatabaseService.getInstance().setSystemState('license_variant_type', 'admiral'); }); afterAll(async () => { + // Clear the Admiral state beforeAll set so this file does not leak + // license context into other tests sharing the same test DB. + const { DatabaseService } = await import('../services/DatabaseService'); + DatabaseService.getInstance().setSystemState('license_status', 'community'); + DatabaseService.getInstance().setSystemState('license_variant_type', ''); await new Promise((resolve, reject) => { server.close((err) => (err ? reject(err) : resolve())); }); cleanupTestDb(tmpDir); }); - function connect(pathAndQuery: string, opts: { cookie?: string; bearer?: string } = {}): WebSocket { + function connect( + pathAndQuery: string, + opts: { cookie?: string; bearer?: string; extraHeaders?: Record } = {}, + ): WebSocket { const headers: Record = {}; if (opts.cookie) headers['cookie'] = opts.cookie; if (opts.bearer) headers['authorization'] = `Bearer ${opts.bearer}`; + if (opts.extraHeaders) Object.assign(headers, opts.extraHeaders); return new WebSocket(`ws://127.0.0.1:${port}${pathAndQuery}`, { headers }); } @@ -218,6 +235,113 @@ describe('WebSocket upgrade dispatch order', () => { }); }); + describe('/api/mesh/proxy-tunnel Admiral entitlement gating', () => { + // Admiral entitlement on the WS data plane is decided against the + // *central's* asserted tier, matching the HTTP mesh routes + // (requireAdmiral in routes/mesh.ts reads req.proxyTier from forwarded + // headers off the node_proxy credential). The WS dispatcher trusts + // x-sencho-tier / x-sencho-variant only when the upgrade carries a + // node_proxy JWT; when no headers are present, or when the credential + // is a full-admin api_token (no central is asserting tier), it falls + // back to the receiver's own license. These tests pin both branches: + // (a) the trusted-header path accepts an Admiral central even on a + // Community receiver, and rejects a Community central on an + // Admiral receiver; + // (b) the local-fallback path keeps the receiver-license check intact + // for full-admin api_token upgrades and for header-less node_proxy + // upgrades. + + async function setLicense(status: string, variantType: string | null): Promise { + const { DatabaseService } = await import('../services/DatabaseService'); + DatabaseService.getInstance().setSystemState('license_status', status); + if (variantType === null) { + DatabaseService.getInstance().setSystemState('license_variant_type', ''); + } else { + DatabaseService.getInstance().setSystemState('license_variant_type', variantType); + } + } + + afterEach(async () => { + // Restore the Admiral state beforeAll established so subsequent + // tests in this file (and the proxy-tunnel scope block) keep passing. + await setLicense('active', 'admiral'); + }); + + it('rejects a node_proxy Bearer with HTTP 403 when no tier headers and the receiver license is community', async () => { + await setLicense('community', null); + const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + const ws = connect('/api/mesh/proxy-tunnel', { bearer: nodeProxyToken }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('unexpected'); + if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403); + }); + + it('rejects a node_proxy Bearer with HTTP 403 when no tier headers and the receiver license is paid but Skipper (not Admiral)', async () => { + await setLicense('active', 'skipper'); + const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + const ws = connect('/api/mesh/proxy-tunnel', { bearer: nodeProxyToken }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('unexpected'); + if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403); + }); + + it('rejects a full-admin api_token with HTTP 403 when the receiver license is community (forwarded headers ignored on api_token path)', async () => { + await setLicense('community', null); + const { DatabaseService } = await import('../services/DatabaseService'); + const adminId = DatabaseService.getInstance().getUserByUsername(TEST_USERNAME)!.id; + const rawToken = createTestApiToken({ + db: DatabaseService, + scope: 'full-admin', + userId: adminId, + name: `mesh-license-gate-${Date.now()}`, + }); + // Header is set but must be ignored: the full-admin api_token is a + // local-entitlement credential, not a node_proxy forwarder. + const ws = connect('/api/mesh/proxy-tunnel', { + bearer: rawToken, + extraHeaders: { 'x-sencho-tier': 'paid', 'x-sencho-variant': 'admiral' }, + }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('unexpected'); + if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403); + }); + + it('accepts a node_proxy Bearer asserting paid+admiral via forwarded headers even when the receiver license is community', async () => { + await setLicense('community', null); + const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + const ws = connect('/api/mesh/proxy-tunnel', { + bearer: nodeProxyToken, + extraHeaders: { 'x-sencho-tier': 'paid', 'x-sencho-variant': 'admiral' }, + }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('open'); + try { ws.terminate(); } catch { /* ignore */ } + }); + + it('rejects a node_proxy Bearer asserting community via forwarded headers even when the receiver license is Admiral', async () => { + // Receiver state is already Admiral (set by beforeAll / afterEach). + const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + const ws = connect('/api/mesh/proxy-tunnel', { + bearer: nodeProxyToken, + extraHeaders: { 'x-sencho-tier': 'community' }, + }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('unexpected'); + if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403); + }); + + it('rejects a node_proxy Bearer asserting paid+skipper via forwarded headers (paid but not Admiral)', async () => { + const nodeProxyToken = jwt.sign({ scope: 'node_proxy' }, TEST_JWT_SECRET, { expiresIn: '1m' }); + const ws = connect('/api/mesh/proxy-tunnel', { + bearer: nodeProxyToken, + extraHeaders: { 'x-sencho-tier': 'paid', 'x-sencho-variant': 'skipper' }, + }); + const outcome = await waitForOutcome(ws); + expect(outcome.kind).toBe('unexpected'); + if (outcome.kind === 'unexpected') expect(outcome.status).toBe(403); + }); + }); + it('dispatches /api/pilot/tunnel to the pilot handler (rejects non-pilot bearer before path-based dispatch)', async () => { // A plain session cookie is a valid *user* JWT but not a pilot JWT. The // pilot handler runs first (before the shared cookie/Bearer auth) and diff --git a/backend/src/routes/mesh.ts b/backend/src/routes/mesh.ts index 68c39c09..0753e869 100644 --- a/backend/src/routes/mesh.ts +++ b/backend/src/routes/mesh.ts @@ -85,7 +85,7 @@ meshRouter.post('/nodes/:nodeId/disable', async (req: Request, res: Response): P const nodeId = Number.parseInt(req.params.nodeId as string, 10); if (!Number.isFinite(nodeId)) { res.status(400).json({ error: 'Invalid node id' }); return; } try { - await MeshService.getInstance().disableForNode(nodeId); + await MeshService.getInstance().disableForNode(nodeId, actorFor(req)); res.json({ ok: true }); } catch (err) { res.status(500).json({ error: (err as Error).message }); diff --git a/backend/src/services/MeshForwarder.ts b/backend/src/services/MeshForwarder.ts index af6bca1e..c20d7039 100644 --- a/backend/src/services/MeshForwarder.ts +++ b/backend/src/services/MeshForwarder.ts @@ -2,19 +2,17 @@ import net from 'net'; import { sanitizeForLog } from '../utils/safeLog'; /** - * In-process mesh TCP forwarder. Owns per-port `net.Server` listeners on the - * host network and delegates accepted sockets to the host (MeshService) for - * resolve + splice. Replaces the separate `saelix/sencho-mesh` sidecar - * container that previously did this job over a control WebSocket. The - * resolve step is now a sync map lookup rather than a round-trip, so - * MeshForwarder is just a thin lifecycle layer; all routing + splicing - * lives on MeshService. + * In-process mesh TCP forwarder. Owns per-port `net.Server` listeners and + * delegates accepted sockets to the host (MeshService) for resolve + + * splice. All routing and splicing logic lives on MeshService; + * MeshForwarder is a thin lifecycle layer that opens and closes ports. * - * Sencho's container must run in `network_mode: host` (Linux) for the - * listeners to bind on the host's network where meshed containers' - * `extra_hosts: :host-gateway` entries point. Without host network - * mode, `net.createServer().listen(port)` lands inside the container's - * namespace and inbound traffic from peers never reaches it. + * Sencho runs in standard Docker bridge mode and attaches its own + * container to the shared `sencho_mesh` network at a stable IP. Meshed + * user containers are attached to the same network, so they reach the + * forwarder by the Sencho IP without `network_mode: host` or the + * `extra_hosts: :host-gateway` indirection an older sidecar + * design required. */ export interface MeshForwarderHost { @@ -50,9 +48,10 @@ export class MeshForwarder { const onListening = () => { server.removeListener('error', onError); resolve(); }; server.once('error', onError); server.once('listening', onListening); - // Bind on all interfaces. Under host network mode this is - // the host's own network; under bridge mode (mesh disabled - // at boot) this would be the container's namespace. + // Bind on all interfaces inside the Sencho container's + // networking namespace. The sencho_mesh bridge attaches + // both Sencho and meshed user containers, so peers reach + // this listener at Sencho's mesh-network IP. server.listen(port, '0.0.0.0'); }); this.listeners.set(port, server); diff --git a/backend/src/services/MeshProxyTunnelDialer.ts b/backend/src/services/MeshProxyTunnelDialer.ts index 079a0e7f..24c07e30 100644 --- a/backend/src/services/MeshProxyTunnelDialer.ts +++ b/backend/src/services/MeshProxyTunnelDialer.ts @@ -9,6 +9,8 @@ import { httpUrlToWs } from '../utils/wsUrl'; import { isDebugEnabled } from '../utils/debug'; import { PilotMetrics } from './PilotMetrics'; import type { MeshActivityType } from './MeshService'; +import { LicenseService } from './LicenseService'; +import { PROXY_TIER_HEADER, PROXY_VARIANT_HEADER } from './license-headers'; /** * Central-side dialer for proxy-mode mesh tunnels. @@ -234,10 +236,23 @@ export class MeshProxyTunnelDialer extends EventEmitter { // the peer falls back to its local DB default (always 1) and treats // cross-node aliases as same-node. const wsUrl = httpUrlToWs(target.apiUrl) + `/api/mesh/proxy-tunnel?nodeId=${nodeId}`; + // Forward central's tier and variant so the receiver enforces Admiral + // against the *central's* license (matching the HTTP mesh routes, + // which all gate on `requireAdmiral` against `req.proxyTier`). Without + // these the receiver falls back to its own local license, which would + // both reject Admiral centrals talking to Community remotes and let + // Community centrals dial locally-Admiral remotes. The headers are + // trusted on the receiver only when the WS carries a node_proxy / + // pilot_tunnel credential (see middleware/auth.ts:117-135). + const proxyHeaders = LicenseService.getInstance().getProxyHeaders(); let ws: WebSocket; try { ws = new WebSocket(wsUrl, { - headers: { Authorization: `Bearer ${target.apiToken}` }, + headers: { + Authorization: `Bearer ${target.apiToken}`, + [PROXY_TIER_HEADER]: proxyHeaders.tier, + [PROXY_VARIANT_HEADER]: proxyHeaders.variant || '', + }, handshakeTimeout: HANDSHAKE_TIMEOUT_MS, maxPayload: MAX_FRAME_SIZE_BYTES, }); diff --git a/backend/src/services/MeshService.ts b/backend/src/services/MeshService.ts index d030199a..6a577a4d 100644 --- a/backend/src/services/MeshService.ts +++ b/backend/src/services/MeshService.ts @@ -242,9 +242,8 @@ interface ActiveStreamRecord { /** * Sencho Mesh orchestrator. Owns: - * - in-process TCP forwarder (`MeshForwarder`) that binds host-network - * listeners on alias ports. Replaces the prior separate sidecar - * container; one container per node now. + * - in-process TCP forwarder (`MeshForwarder`) that binds per-alias + * listeners on Sencho's `sencho_mesh` bridge-network IP. * - opt-in / opt-out persistence and cascading override regeneration * - global alias aggregation (across the fleet via the existing HTTP * proxy chain, see `inspectStackServices`) @@ -258,8 +257,6 @@ interface ActiveStreamRecord { * Docker bridge network. Meshed user services join `sencho_mesh` so * that IP is reachable from inside their containers without any * host-firewall coordination. - * - cross-node mesh routing is central → pilot in this phase. Pilot → - * central and pilot ↔ pilot via central relay land in Phase B. */ export class MeshService extends EventEmitter implements MeshForwarderHost { private static instance: MeshService; @@ -904,15 +901,38 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { } } - public async disableForNode(nodeId: number): Promise { + public async disableForNode( + nodeId: number, + actor: string = 'system:mesh.disable', + ): Promise { DatabaseService.getInstance().setNodeMeshEnabled(nodeId, false); const stacks = DatabaseService.getInstance().listMeshStacks(nodeId); for (const s of stacks) { DatabaseService.getInstance().deleteMeshStack(nodeId, s.stack_name); - await this.removeStackOverride(nodeId, s.stack_name); } + // Dispatch DELETE /api/mesh/local-override/:stack for remote nodes + // (pilot or proxy) so the override file pushed earlier via + // applyLocalOverride is removed; falls back to local deletion for + // local nodes. Parallelize per the regenerateOverridesForNode + // rationale: each remote call is its own HTTP round-trip, so + // awaiting sequentially turns N stacks into N serialised DELETEs. + // `allSettled` so a single failure does not abort the others + // (removeOverrideFromNode already swallows errors internally). + await Promise.allSettled( + stacks.map((s) => this.removeOverrideFromNode(nodeId, s.stack_name)), + ); await this.refreshAliasCache(); await this.syncForwarderListeners(); + // Mirror optOutStack: regenerate every remaining node's override + // without the dropped aliases, recompose the rest of the fleet so + // their containers shed the stale extra_hosts, and redeploy the + // disabled node's own stacks so their containers detach from the + // sencho_mesh network and lose the alias entries they owned. + await this.regenerateOverridesAcrossFleet(); + this.cascadeRecomposeAcrossFleet(undefined, undefined, actor); + for (const s of stacks) { + this.triggerRedeploy(nodeId, s.stack_name, actor); + } this.logActivity({ source: 'mesh', level: 'info', type: 'mesh.disable', nodeId, message: `mesh disabled on node ${nodeId}`, @@ -1415,11 +1435,13 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { const body = await res.json() as { services?: Array<{ service: string; ports: number[] }> }; return body.services ?? []; } catch (err) { - // proxyFetch throws MeshError('push_failed') when getProxyTarget - // returns null (e.g. pilot tunnel offline). Treat the same as a - // non-OK response: empty list. - if (err instanceof MeshError && err.code === 'push_failed') { - console.warn(`[MeshService] inspectStackServices: no proxy target for node ${nodeId} (${sanitizeForLog(node.name)})`); + // proxyFetch throws MeshError('no_target') when getProxyTarget + // returns null (pilot tunnel offline, proxy bridge unreachable) + // and MeshError('push_failed') on a non-OK HTTP response. Treat + // both as a soft "no services to report" so the Routing tab does + // not surface a stack trace for a node that is simply offline. + if (err instanceof MeshError && (err.code === 'no_target' || err.code === 'push_failed')) { + console.warn(`[MeshService] inspectStackServices: unreachable node ${nodeId} (${sanitizeForLog(node.name)}): ${err.code}`); return []; } console.error('[MeshService] inspectStackServices remote unreachable:', sanitizeForLog((err as Error).message)); @@ -1690,12 +1712,12 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { /** * Same-node forward: dial the target container's bridge IP directly. - * Sencho runs in `network_mode: host` so it sees the docker bridge - * networks and can reach container IPs without going through any - * host-port publish. Looks up the container by Compose's - * `--` naming convention; falls back to a - * label-filtered listContainers if the conventional name is absent - * (e.g. when the operator overrode the project name). + * Sencho joins the `sencho_mesh` Docker bridge network alongside the + * meshed user containers, so it can reach their bridge IPs without + * going through any host-port publish. Looks up the container by + * Compose's `--` naming convention; falls + * back to a label-filtered listContainers if the conventional name + * is absent (e.g. when the operator overrode the project name). */ private async openSameNode(target: MeshTarget, src: net.Socket): Promise { const ip = await this.resolveContainerIp(target); @@ -2244,13 +2266,6 @@ export class MeshService extends EventEmitter implements MeshForwarderHost { private isLocalForwarderActive(): boolean { return this.started && this.forwarder.getListenerPorts().length > 0; } - - // mintSidecarToken / verifySidecarToken / spawnSidecar / stopSidecar / - // isSidecarRunning / handleSidecarResolve / sendSidecar / - // attachSidecarSocket are gone: the in-process MeshForwarder replaces - // the entire sidecar layer. Routing decisions happen via direct - // MeshService calls — no JWT minting, no separate container, no control - // WebSocket. See `docs/internal/architecture/mesh.md` for the new flow. } export type MeshErrorCode = diff --git a/backend/src/services/PilotTunnelBridge.ts b/backend/src/services/PilotTunnelBridge.ts index b45ff80c..5a998830 100644 --- a/backend/src/services/PilotTunnelBridge.ts +++ b/backend/src/services/PilotTunnelBridge.ts @@ -91,6 +91,16 @@ interface ReverseLocalTcpStreamState extends StreamMeta { interface ReverseRelayTcpStreamState extends StreamMeta { kind: 'reverse_relay'; target: TcpStream; + /** + * False between `acceptReverseRelay` swapping the reservation for this + * state and the target stream emitting `'open'`. `TcpData` frames + * arriving in that window are queued in `pendingData` rather than + * written into the target stream, whose own remote ack may not yet + * have landed. + */ + targetOpen: boolean; + pendingData: Buffer[]; + pendingBytes: number; bytesIn: number; bytesOut: number; } @@ -639,7 +649,23 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle } else if (s.kind === 'reverse_relay') { s.bytesIn += frame.payload.length; this.refreshIdleTimer(frame.streamId, s); - s.target.write(frame.payload); + if (s.targetOpen) { + s.target.write(frame.payload); + } else { + // Mirror the reverse_local pending-data path: between + // the state swap in acceptReverseRelay and the target + // stream's 'open' event, buffer up to the cap. Without + // this, a peer that sends body bytes immediately after + // tcp_open_reverse would drop them into a stream that + // is still waiting for its remote ack. + if (s.pendingBytes + frame.payload.length > STREAM_PENDING_DATA_MAX_BYTES) { + this.removeStream(frame.streamId); + this.sendJson({ t: 'tcp_close', s: frame.streamId }); + break; + } + s.pendingData.push(Buffer.from(frame.payload)); + s.pendingBytes += frame.payload.length; + } } break; } @@ -825,11 +851,12 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle if (targetNodeId === localNodeId) { await this.acceptReverseLocal(s, { stack, service, port }); } else { - // Relay path does not yet buffer early data (separate - // follow-up); drop the local-shaped reservation so the - // relay state machine starts from a clean slot. Any frames - // buffered up to this point are discarded with it. - if (this.streams.get(s) === reservation) this.removeStream(s); + // Leave the local-shaped reservation in place so any + // TcpData arriving while ensureBridge + openTcpStream + // resolve keeps buffering through the reverse_local + // branch in handleBinaryFrame. acceptReverseRelay + // transplants the buffered frames into the relay state + // and flushes them before the open ack. await this.acceptReverseRelay(s, targetNodeId, { stack, service, port }); } })().catch((err) => { @@ -938,25 +965,62 @@ export class PilotTunnelBridge extends EventEmitter implements MeshTunnelHandle } private async acceptReverseRelay(s: number, targetNodeId: number, target: { stack: string; service: string; port: number }): Promise { + // Pull the reverse_local-shaped reservation that handleTcpOpenReverse + // placed synchronously. Its pendingData buffer holds any TcpData + // frames that arrived while we were dispatching, which we transplant + // into the relay state so they can be flushed once the target stream + // opens. + const reservation = this.streams.get(s); + if (!reservation || reservation.kind !== 'reverse_local') return; + const { PilotTunnelManager } = await import('./PilotTunnelManager'); // ensureBridge resolves either an existing pilot tunnel or a fresh // proxy-mode tunnel dialed on demand, so proxy <-> proxy relays // succeed even when neither remote has a pilot agent. const targetBridge = await PilotTunnelManager.getInstance().ensureBridge(targetNodeId); if (!targetBridge) { + if (this.streams.get(s) === reservation) this.removeStream(s); this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' }); return; } + // Reservation may have been evicted under the pending-bytes cap + // while ensureBridge was resolving. If so, the peer has already + // been told via tcp_close to tear down; bail out cleanly. + if (this.streams.get(s) !== reservation) return; + const targetStream = targetBridge.openTcpStream(target); if (!targetStream) { + if (this.streams.get(s) === reservation) this.removeStream(s); this.sendJson({ t: 'tcp_open_ack', s, ok: false, err: 'unreachable' }); return; } - const state: ReverseRelayTcpStreamState = { kind: 'reverse_relay', target: targetStream, bytesIn: 0, bytesOut: 0 }; + const state: ReverseRelayTcpStreamState = { + kind: 'reverse_relay', + target: targetStream, + targetOpen: false, + pendingData: reservation.pendingData, + pendingBytes: reservation.pendingBytes, + bytesIn: reservation.bytesIn, + bytesOut: 0, + }; this.streams.set(s, state); this.refreshIdleTimer(s, state); targetStream.once('open', () => { + const cur = this.streams.get(s); + if (!cur || cur.kind !== 'reverse_relay') return; + // Flush buffered early data into the target stream BEFORE + // acking so the peer sees a clean "ack + data" sequence and + // the upstream receives the request body ahead of anything + // that arrives post-ack. + if (cur.pendingData.length > 0) { + for (const buf of cur.pendingData) { + try { cur.target.write(buf); } catch { /* ignore */ } + } + cur.pendingData = []; + cur.pendingBytes = 0; + } + cur.targetOpen = true; this.sendJson({ t: 'tcp_open_ack', s, ok: true }); }); targetStream.on('data', (chunk: Buffer) => { diff --git a/backend/src/websocket/upgradeHandler.ts b/backend/src/websocket/upgradeHandler.ts index 220e3abd..3c74e340 100644 --- a/backend/src/websocket/upgradeHandler.ts +++ b/backend/src/websocket/upgradeHandler.ts @@ -4,6 +4,7 @@ import type { WebSocketServer } from 'ws'; import jwt from 'jsonwebtoken'; import crypto from 'crypto'; import { DatabaseService, type UserRole } from '../services/DatabaseService'; +import { LicenseService } from '../services/LicenseService'; import { NodeRegistry } from '../services/NodeRegistry'; import { COOKIE_NAME } from '../helpers/constants'; import { handlePilotTunnel } from './pilotTunnel'; @@ -15,6 +16,8 @@ import { handleHostConsoleWs } from './hostConsole'; import { handleGenericWs, attachGenericConnectionHandlers } from './generic'; import { rejectUpgrade as reject } from './reject'; import { looksLikeApiToken, verifyApiTokenChecksum } from '../utils/apiTokenFormat'; +import { PROXY_TIER_HEADER, PROXY_VARIANT_HEADER } from '../services/license-headers'; +import { isLicenseTier, normalizeTier, isLicenseVariant, normalizeVariant } from '../services/license-normalize'; function parseCookies(req: IncomingMessage): Record { const header = req.headers.cookie || ''; @@ -138,10 +141,31 @@ export function attachUpgrade( // decoded scope is undefined (isProxyToken=false, wsApiTokenScope=null). // Restricted api_token scopes (read-only, deploy-only) are blocked // earlier by the scope gate above before this branch is reached. + // + // Admiral entitlement is decided against the *central's* license, not + // the receiver's, matching every HTTP mesh route in routes/mesh.ts that + // uses `requireAdmiral` / `effectiveTier`. On the node_proxy path the + // central forwards `x-sencho-tier` / `x-sencho-variant` and the WS + // dispatcher trusts them off the node_proxy credential (same rule as + // middleware/auth.ts:117-135 for HTTP). On the full-admin api_token + // path no central is asserting tier, so we fall back to the receiver's + // own license. Both produce paid+admiral or the upgrade is rejected. if (pathname === '/api/mesh/proxy-tunnel') { if (!isProxyToken && wsApiTokenScope !== 'full-admin') { return reject(socket, 403, 'Forbidden'); } + const license = LicenseService.getInstance(); + const tunnelTierHeader = req.headers[PROXY_TIER_HEADER] as string | undefined; + const tunnelVariantHeader = req.headers[PROXY_VARIANT_HEADER] as string | undefined; + const tunnelTier = isProxyToken && isLicenseTier(tunnelTierHeader) + ? normalizeTier(tunnelTierHeader) + : license.getTier(); + const tunnelVariant = isProxyToken && tunnelVariantHeader !== undefined && isLicenseVariant(tunnelVariantHeader) + ? normalizeVariant(tunnelVariantHeader) + : license.getVariant(); + if (tunnelTier !== 'paid' || tunnelVariant !== 'admiral') { + return reject(socket, 403, 'Forbidden'); + } await handleMeshProxyTunnel(req, socket, head); return; }