mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-08-13 04:06:59 +00:00
feat: pilot agent outbound-mode for remote nodes (#667)
* feat: pilot agent outbound-mode for remote nodes Adds a second mode for managing remote nodes: the agent dials an outbound WebSocket tunnel to the primary, so the remote host no longer needs an inbound port, a reachable URL, or its own TLS certificate. Works behind NAT, residential routers, and corporate firewalls. The primary multiplexes HTTP and WebSocket requests over a single tunnel via a hybrid JSON + binary frame protocol, bridged through a per-tunnel loopback server so existing proxy and upgrade handlers route pilot-mode nodes identically to proxy-mode ones. Enrollment uses a single-use 15-minute pilot_enroll JWT exchanged for a long-lived pilot_tunnel credential on first connect. Proxy mode continues to work unchanged and both modes are supported side-by-side. * test(e2e): switch to proxy mode before asserting api_url field Remote nodes default to Pilot Agent mode, which hides the api_url input. The SSRF-validation tests need proxy mode, so the helper now selects Distributed API Proxy after picking Remote type before asserting the field is visible. * fix(e2e): wire Combobox id prop so node-mode selector resolves The Combobox trigger button had no id, leaving its Label orphaned and making getByRole name-based lookups fail. Adding id to the primitive, passing id="node-mode" from NodeManager, and updating the E2E helper to use #node-mode fixes both the a11y regression and the CI timeout.
This commit is contained in:
Anso
@@ -26,16 +26,28 @@ export interface StackAlert {
|
||||
last_fired_at?: number;
|
||||
}
|
||||
|
||||
export type NodeMode = 'proxy' | 'pilot_agent';
|
||||
|
||||
export interface Node {
|
||||
id: number;
|
||||
name: string;
|
||||
type: 'local' | 'remote';
|
||||
mode: NodeMode;
|
||||
compose_dir: string;
|
||||
is_default: boolean;
|
||||
status: 'online' | 'offline' | 'unknown';
|
||||
created_at: number;
|
||||
api_url?: string;
|
||||
api_token?: string;
|
||||
pilot_last_seen?: number | null;
|
||||
pilot_agent_version?: string | null;
|
||||
}
|
||||
|
||||
export interface PilotEnrollment {
|
||||
node_id: number;
|
||||
token_hash: string;
|
||||
expires_at: number;
|
||||
used_at: number | null;
|
||||
}
|
||||
|
||||
export interface Label {
|
||||
@@ -483,6 +495,14 @@ export class DatabaseService {
|
||||
created_at INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS pilot_enrollments (
|
||||
node_id INTEGER PRIMARY KEY,
|
||||
token_hash TEXT NOT NULL,
|
||||
expires_at INTEGER NOT NULL,
|
||||
used_at INTEGER,
|
||||
FOREIGN KEY (node_id) REFERENCES nodes(id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS system_state (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
@@ -798,6 +818,11 @@ export class DatabaseService {
|
||||
maybeAddCol('nodes', 'api_url', "TEXT DEFAULT ''");
|
||||
maybeAddCol('nodes', 'api_token', "TEXT DEFAULT ''");
|
||||
|
||||
// Pilot Agent outbound-mode columns
|
||||
maybeAddCol('nodes', 'mode', "TEXT NOT NULL DEFAULT 'proxy'");
|
||||
maybeAddCol('nodes', 'pilot_last_seen', 'INTEGER');
|
||||
maybeAddCol('nodes', 'pilot_agent_version', 'TEXT');
|
||||
|
||||
// Scheduled operations migrations
|
||||
maybeAddCol('scheduled_task_runs', 'triggered_by', "TEXT NOT NULL DEFAULT 'scheduler'");
|
||||
maybeAddCol('scheduled_tasks', 'prune_targets', 'TEXT DEFAULT NULL');
|
||||
@@ -1278,36 +1303,42 @@ export class DatabaseService {
|
||||
return {
|
||||
...row,
|
||||
is_default: row.is_default === 1,
|
||||
mode: (row.mode === 'pilot_agent' ? 'pilot_agent' : 'proxy') as NodeMode,
|
||||
api_token: row.api_token ? crypto.decrypt(row.api_token) : '',
|
||||
pilot_last_seen: row.pilot_last_seen ?? null,
|
||||
pilot_agent_version: row.pilot_agent_version ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
private static readonly NODE_COLUMNS =
|
||||
'id, name, type, compose_dir, is_default, status, created_at, api_url, api_token, mode, pilot_last_seen, pilot_agent_version';
|
||||
|
||||
public getNodes(): Node[] {
|
||||
const stmt = this.db.prepare('SELECT id, name, type, compose_dir, is_default, status, created_at, api_url, api_token FROM nodes ORDER BY is_default DESC, name ASC');
|
||||
const stmt = this.db.prepare(`SELECT ${DatabaseService.NODE_COLUMNS} FROM nodes ORDER BY is_default DESC, name ASC`);
|
||||
return stmt.all().map((row: any) => this.decryptNodeRow(row));
|
||||
}
|
||||
|
||||
public getNode(id: number): Node | undefined {
|
||||
const stmt = this.db.prepare('SELECT id, name, type, compose_dir, is_default, status, created_at, api_url, api_token FROM nodes WHERE id = ?');
|
||||
const stmt = this.db.prepare(`SELECT ${DatabaseService.NODE_COLUMNS} FROM nodes WHERE id = ?`);
|
||||
const row = stmt.get(id) as any;
|
||||
if (!row) return undefined;
|
||||
return this.decryptNodeRow(row);
|
||||
}
|
||||
|
||||
public getDefaultNode(): Node | undefined {
|
||||
const stmt = this.db.prepare('SELECT id, name, type, compose_dir, is_default, status, created_at, api_url, api_token FROM nodes WHERE is_default = 1 LIMIT 1');
|
||||
const stmt = this.db.prepare(`SELECT ${DatabaseService.NODE_COLUMNS} FROM nodes WHERE is_default = 1 LIMIT 1`);
|
||||
const row = stmt.get() as any;
|
||||
if (!row) return undefined;
|
||||
return this.decryptNodeRow(row);
|
||||
}
|
||||
|
||||
public addNode(node: Omit<Node, 'id' | 'status' | 'created_at'>): number {
|
||||
public addNode(node: Omit<Node, 'id' | 'status' | 'created_at' | 'mode'> & { mode?: NodeMode }): number {
|
||||
if (node.is_default) {
|
||||
this.db.prepare('UPDATE nodes SET is_default = 0').run();
|
||||
}
|
||||
const crypto = CryptoService.getInstance();
|
||||
const stmt = this.db.prepare(
|
||||
'INSERT INTO nodes (name, type, compose_dir, is_default, status, created_at, api_url, api_token) VALUES (?, ?, ?, ?, ?, ?, ?, ?)'
|
||||
'INSERT INTO nodes (name, type, compose_dir, is_default, status, created_at, api_url, api_token, mode) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)'
|
||||
);
|
||||
const result = stmt.run(
|
||||
node.name,
|
||||
@@ -1317,7 +1348,8 @@ export class DatabaseService {
|
||||
'unknown',
|
||||
Date.now(),
|
||||
node.api_url || '',
|
||||
node.api_token ? crypto.encrypt(node.api_token) : ''
|
||||
node.api_token ? crypto.encrypt(node.api_token) : '',
|
||||
node.mode || 'proxy'
|
||||
);
|
||||
return result.lastInsertRowid as number;
|
||||
}
|
||||
@@ -1343,6 +1375,9 @@ export class DatabaseService {
|
||||
fields.push('api_token = ?');
|
||||
values.push(updates.api_token ? CryptoService.getInstance().encrypt(updates.api_token) : '');
|
||||
}
|
||||
if (updates.mode !== undefined) { fields.push('mode = ?'); values.push(updates.mode); }
|
||||
if (updates.pilot_last_seen !== undefined) { fields.push('pilot_last_seen = ?'); values.push(updates.pilot_last_seen); }
|
||||
if (updates.pilot_agent_version !== undefined) { fields.push('pilot_agent_version = ?'); values.push(updates.pilot_agent_version); }
|
||||
|
||||
if (fields.length === 0) return;
|
||||
|
||||
@@ -1370,6 +1405,43 @@ export class DatabaseService {
|
||||
this.db.prepare('UPDATE nodes SET status = ? WHERE id = ?').run(status, id);
|
||||
}
|
||||
|
||||
// --- Pilot enrollments ---
|
||||
|
||||
public getPilotEnrollment(nodeId: number): PilotEnrollment | undefined {
|
||||
const row = this.db.prepare(
|
||||
'SELECT node_id, token_hash, expires_at, used_at FROM pilot_enrollments WHERE node_id = ?'
|
||||
).get(nodeId) as PilotEnrollment | undefined;
|
||||
return row;
|
||||
}
|
||||
|
||||
public createPilotEnrollment(nodeId: number, tokenHash: string, expiresAt: number): void {
|
||||
this.db.prepare(
|
||||
`INSERT INTO pilot_enrollments (node_id, token_hash, expires_at, used_at)
|
||||
VALUES (?, ?, ?, NULL)
|
||||
ON CONFLICT(node_id) DO UPDATE SET
|
||||
token_hash = excluded.token_hash,
|
||||
expires_at = excluded.expires_at,
|
||||
used_at = NULL`
|
||||
).run(nodeId, tokenHash, expiresAt);
|
||||
}
|
||||
|
||||
public consumePilotEnrollment(tokenHash: string): PilotEnrollment | undefined {
|
||||
const now = Date.now();
|
||||
return this.db.transaction(() => {
|
||||
const row = this.db.prepare(
|
||||
`SELECT node_id, token_hash, expires_at, used_at FROM pilot_enrollments
|
||||
WHERE token_hash = ? AND used_at IS NULL AND expires_at > ?`
|
||||
).get(tokenHash, now) as PilotEnrollment | undefined;
|
||||
if (!row) return undefined;
|
||||
this.db.prepare('UPDATE pilot_enrollments SET used_at = ? WHERE node_id = ?').run(now, row.node_id);
|
||||
return { ...row, used_at: now };
|
||||
})();
|
||||
}
|
||||
|
||||
public deletePilotEnrollment(nodeId: number): void {
|
||||
this.db.prepare('DELETE FROM pilot_enrollments WHERE node_id = ?').run(nodeId);
|
||||
}
|
||||
|
||||
// --- Stack Update Status ---
|
||||
|
||||
public upsertStackUpdateStatus(nodeId: number, stackName: string, hasUpdate: boolean, checkedAt: number): void {
|
||||
|
||||
@@ -3,6 +3,7 @@ import axios from 'axios';
|
||||
import { EventEmitter } from 'events';
|
||||
import { DatabaseService, Node } from './DatabaseService';
|
||||
import { fetchRemoteMeta } from './CapabilityRegistry';
|
||||
import { PilotTunnelManager } from './PilotTunnelManager';
|
||||
|
||||
/**
|
||||
* NodeRegistry: Manages connections for multiple nodes.
|
||||
@@ -99,12 +100,22 @@ export class NodeRegistry extends EventEmitter {
|
||||
/**
|
||||
* Get the HTTP proxy target for a remote node.
|
||||
* Returns { apiUrl, apiToken } for use by the HTTP proxy middleware.
|
||||
*
|
||||
* Pilot-agent nodes resolve to the loopback URL of their active tunnel
|
||||
* bridge; the bridge strips the bearer token and re-authenticates
|
||||
* implicitly via the pre-verified tunnel socket.
|
||||
*/
|
||||
public getProxyTarget(nodeId: number): { apiUrl: string; apiToken: string } | null {
|
||||
const node = DatabaseService.getInstance().getNode(nodeId);
|
||||
if (!node || node.type !== 'remote' || !node.api_url || !node.api_token) {
|
||||
return null;
|
||||
if (!node || node.type !== 'remote') return null;
|
||||
|
||||
if (node.mode === 'pilot_agent') {
|
||||
const loopbackUrl = PilotTunnelManager.getInstance().getLoopbackUrl(nodeId);
|
||||
if (!loopbackUrl) return null;
|
||||
return { apiUrl: loopbackUrl, apiToken: '' };
|
||||
}
|
||||
|
||||
if (!node.api_url || !node.api_token) return null;
|
||||
return { apiUrl: node.api_url, apiToken: node.api_token };
|
||||
}
|
||||
|
||||
@@ -122,12 +133,48 @@ export class NodeRegistry extends EventEmitter {
|
||||
}
|
||||
|
||||
if (node.type === 'remote') {
|
||||
if (node.mode === 'pilot_agent') {
|
||||
return this.testPilotConnection(node);
|
||||
}
|
||||
return this.testRemoteConnection(node);
|
||||
}
|
||||
|
||||
return this.testLocalConnection(nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Check whether a pilot-agent node has an active tunnel. Does not call
|
||||
* any endpoint; tunnel liveness is tracked in-process by
|
||||
* PilotTunnelManager and via the JWT handshake that originally accepted
|
||||
* the agent.
|
||||
*/
|
||||
private async testPilotConnection(node: Node): Promise<{ success: boolean; error?: string; info?: any }> {
|
||||
const db = DatabaseService.getInstance();
|
||||
const active = PilotTunnelManager.getInstance().hasActiveTunnel(node.id);
|
||||
if (!active) {
|
||||
db.updateNodeStatus(node.id, 'offline');
|
||||
return { success: false, error: 'Pilot agent is not connected. Start the agent container or regenerate the enrollment token.' };
|
||||
}
|
||||
db.updateNodeStatus(node.id, 'online');
|
||||
return {
|
||||
success: true,
|
||||
info: {
|
||||
name: node.name,
|
||||
serverVersion: 'Pilot Agent',
|
||||
senchoVersion: node.pilot_agent_version ?? null,
|
||||
capabilities: [],
|
||||
os: 'Remote (tunnel)',
|
||||
architecture: 'Remote',
|
||||
containers: '-',
|
||||
containersRunning: '-',
|
||||
images: '-',
|
||||
memTotal: 0,
|
||||
cpus: '-',
|
||||
pilotLastSeen: node.pilot_last_seen ?? null,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
private async testLocalConnection(nodeId: number): Promise<{ success: boolean; error?: string; info?: any }> {
|
||||
const db = DatabaseService.getInstance();
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,385 @@
|
||||
import http, { IncomingMessage, Server as HttpServer, ServerResponse } from 'http';
|
||||
import { Socket } from 'net';
|
||||
import { EventEmitter } from 'events';
|
||||
import { WebSocket, WebSocketServer } from 'ws';
|
||||
import {
|
||||
BinaryFrameType,
|
||||
DecodedBinaryFrame,
|
||||
StreamIdAllocator,
|
||||
decodeBinaryFrame,
|
||||
decodeJsonFrame,
|
||||
encodeBinaryFrame,
|
||||
encodeJsonFrame,
|
||||
wsDataToBuffer,
|
||||
wsDataToString,
|
||||
} from '../pilot/protocol';
|
||||
|
||||
const BUFFER_HIGH_WATER_MARK = 4 * 1024 * 1024;
|
||||
const PING_INTERVAL_MS = 30_000;
|
||||
|
||||
interface HttpStreamState {
|
||||
kind: 'http';
|
||||
res: ServerResponse;
|
||||
headersWritten: boolean;
|
||||
}
|
||||
|
||||
interface WsStreamState {
|
||||
kind: 'ws';
|
||||
rawSocket?: Socket;
|
||||
rawHead?: Buffer;
|
||||
upgradeRequest: IncomingMessage;
|
||||
clientWs?: WebSocket;
|
||||
}
|
||||
|
||||
type StreamState = HttpStreamState | WsStreamState;
|
||||
|
||||
/**
|
||||
* Per-tunnel bridge: hosts a loopback HTTP server that demuxes requests into
|
||||
* wire frames sent over the pilot WebSocket, and remuxes response frames back
|
||||
* to the loopback caller.
|
||||
*
|
||||
* The primary's existing http-proxy-middleware setup treats the loopback URL
|
||||
* as just another remote target, so HTTP and WebSocket proxy logic, header
|
||||
* stripping/injection, and license-tier propagation all work unchanged.
|
||||
*/
|
||||
export class PilotTunnelBridge extends EventEmitter {
|
||||
private readonly tunnelWs: WebSocket;
|
||||
private readonly loopback: HttpServer;
|
||||
private readonly wsUpgradeServer: WebSocketServer;
|
||||
private readonly streamIds = new StreamIdAllocator();
|
||||
private readonly streams = new Map<number, StreamState>();
|
||||
private readonly connectedAt = Date.now();
|
||||
private loopbackUrl = '';
|
||||
private pingTimer?: NodeJS.Timeout;
|
||||
private closed = false;
|
||||
|
||||
constructor(_nodeId: number, tunnelWs: WebSocket) {
|
||||
super();
|
||||
this.tunnelWs = tunnelWs;
|
||||
this.loopback = http.createServer();
|
||||
this.wsUpgradeServer = new WebSocketServer({ noServer: true });
|
||||
|
||||
this.loopback.on('request', (req, res) => this.handleLoopbackRequest(req, res));
|
||||
this.loopback.on('upgrade', (req, socket, head) => this.handleLoopbackUpgrade(req, socket as Socket, head));
|
||||
this.loopback.on('clientError', (_err, socket) => {
|
||||
try { socket.destroy(); } catch { /* ignore */ }
|
||||
});
|
||||
|
||||
this.tunnelWs.on('message', (data, isBinary) => this.handleTunnelMessage(data, isBinary));
|
||||
this.tunnelWs.on('close', () => this.onTunnelClose());
|
||||
this.tunnelWs.on('error', () => this.onTunnelClose());
|
||||
}
|
||||
|
||||
public async start(): Promise<void> {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onError = (err: Error) => reject(err);
|
||||
this.loopback.once('error', onError);
|
||||
this.loopback.listen(0, '127.0.0.1', () => {
|
||||
const addr = this.loopback.address();
|
||||
if (!addr || typeof addr === 'string') {
|
||||
reject(new Error('loopback server returned unexpected address'));
|
||||
return;
|
||||
}
|
||||
this.loopbackUrl = `http://127.0.0.1:${addr.port}`;
|
||||
this.loopback.removeListener('error', onError);
|
||||
resolve();
|
||||
});
|
||||
});
|
||||
this.pingTimer = setInterval(() => {
|
||||
if (this.tunnelWs.readyState === WebSocket.OPEN) {
|
||||
try { this.tunnelWs.ping(); } catch { /* surfaced via 'error' */ }
|
||||
}
|
||||
}, PING_INTERVAL_MS);
|
||||
}
|
||||
|
||||
public getLoopbackUrl(): string { return this.loopbackUrl; }
|
||||
public getConnectedAt(): number { return this.connectedAt; }
|
||||
|
||||
public close(code = 1000, reason = 'closed by primary'): void {
|
||||
if (this.closed) return;
|
||||
this.closed = true;
|
||||
if (this.pingTimer) { clearInterval(this.pingTimer); this.pingTimer = undefined; }
|
||||
|
||||
for (const [, state] of this.streams) this.teardownStream(state);
|
||||
this.streams.clear();
|
||||
|
||||
try { this.tunnelWs.close(code, reason); } catch { /* ignore */ }
|
||||
try { this.loopback.close(); } catch { /* ignore */ }
|
||||
try { this.wsUpgradeServer.close(); } catch { /* ignore */ }
|
||||
this.emit('closed');
|
||||
}
|
||||
|
||||
// --- Loopback HTTP ingress ---
|
||||
|
||||
private handleLoopbackRequest(req: IncomingMessage, res: ServerResponse): void {
|
||||
if (this.closed || this.tunnelWs.readyState !== WebSocket.OPEN) {
|
||||
res.statusCode = 502;
|
||||
res.end('pilot tunnel not ready');
|
||||
return;
|
||||
}
|
||||
|
||||
const streamId = this.streamIds.allocate();
|
||||
this.streams.set(streamId, { kind: 'http', res, headersWritten: false });
|
||||
|
||||
const headers: Record<string, string> = {};
|
||||
for (const [k, v] of Object.entries(req.headers)) {
|
||||
if (typeof v === 'string') headers[k] = v;
|
||||
else if (Array.isArray(v)) headers[k] = v.join(', ');
|
||||
}
|
||||
|
||||
this.sendJson({
|
||||
t: 'http_req',
|
||||
s: streamId,
|
||||
method: req.method || 'GET',
|
||||
path: req.url || '/',
|
||||
headers,
|
||||
});
|
||||
|
||||
req.on('data', (chunk: Buffer) => {
|
||||
if (!this.streams.has(streamId)) return;
|
||||
this.sendBinary(BinaryFrameType.HttpReqBody, streamId, chunk);
|
||||
if (this.tunnelWs.bufferedAmount > BUFFER_HIGH_WATER_MARK) req.pause();
|
||||
});
|
||||
req.on('end', () => {
|
||||
if (!this.streams.has(streamId)) return;
|
||||
this.sendJson({ t: 'http_req_end', s: streamId });
|
||||
});
|
||||
req.on('error', () => {
|
||||
const s = this.streams.get(streamId);
|
||||
if (s) this.teardownStream(s);
|
||||
this.streams.delete(streamId);
|
||||
});
|
||||
|
||||
res.on('close', () => {
|
||||
// Client disconnected before response finished.
|
||||
if (this.streams.has(streamId)) {
|
||||
this.streams.delete(streamId);
|
||||
this.sendJson({ t: 'http_err', s: streamId, code: 'tunnel_down', message: 'client aborted' });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private handleLoopbackUpgrade(req: IncomingMessage, socket: Socket, head: Buffer): void {
|
||||
if (this.closed || this.tunnelWs.readyState !== WebSocket.OPEN) {
|
||||
socket.write('HTTP/1.1 502 Bad Gateway\r\n\r\n');
|
||||
socket.destroy();
|
||||
return;
|
||||
}
|
||||
|
||||
const streamId = this.streamIds.allocate();
|
||||
this.streams.set(streamId, {
|
||||
kind: 'ws',
|
||||
rawSocket: socket,
|
||||
rawHead: head,
|
||||
upgradeRequest: req,
|
||||
});
|
||||
|
||||
const headers: Record<string, string> = {};
|
||||
for (const [k, v] of Object.entries(req.headers)) {
|
||||
if (typeof v === 'string') headers[k] = v;
|
||||
else if (Array.isArray(v)) headers[k] = v.join(', ');
|
||||
}
|
||||
|
||||
this.sendJson({
|
||||
t: 'ws_open',
|
||||
s: streamId,
|
||||
path: req.url || '/',
|
||||
headers,
|
||||
});
|
||||
|
||||
socket.on('error', () => {
|
||||
const s = this.streams.get(streamId);
|
||||
if (s) this.teardownStream(s);
|
||||
this.streams.delete(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);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// --- Tunnel ingress (frames from agent) ---
|
||||
|
||||
private handleTunnelMessage(data: unknown, isBinary: boolean): void {
|
||||
if (this.closed) return;
|
||||
try {
|
||||
if (isBinary) {
|
||||
const buf = wsDataToBuffer(data);
|
||||
if (!buf) return;
|
||||
this.handleBinaryFrame(decodeBinaryFrame(buf));
|
||||
} else {
|
||||
const text = wsDataToString(data);
|
||||
if (text == null) return;
|
||||
const frame = decodeJsonFrame(text);
|
||||
this.handleJsonFrame(frame);
|
||||
}
|
||||
} catch {
|
||||
// Malformed frame: kill the tunnel to force re-sync.
|
||||
this.close(1002, 'protocol error');
|
||||
}
|
||||
}
|
||||
|
||||
private handleJsonFrame(frame: ReturnType<typeof decodeJsonFrame>): void {
|
||||
switch (frame.t) {
|
||||
case 'http_res': {
|
||||
const s = this.streams.get(frame.s);
|
||||
if (!s || s.kind !== 'http') return;
|
||||
if (!s.headersWritten) {
|
||||
try {
|
||||
s.res.writeHead(frame.status, frame.headers);
|
||||
} catch { /* headers already sent or invalid */ }
|
||||
s.headersWritten = true;
|
||||
}
|
||||
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);
|
||||
break;
|
||||
}
|
||||
case 'http_err': {
|
||||
const s = this.streams.get(frame.s);
|
||||
if (!s) return;
|
||||
if (s.kind === 'http' && !s.headersWritten) {
|
||||
try {
|
||||
s.res.writeHead(502, { 'content-type': 'text/plain' });
|
||||
s.res.end(`pilot tunnel error: ${frame.code} ${frame.message}`);
|
||||
} catch { /* ignore */ }
|
||||
} else {
|
||||
this.teardownStream(s);
|
||||
}
|
||||
this.streams.delete(frame.s);
|
||||
break;
|
||||
}
|
||||
case 'ws_accept': {
|
||||
const s = this.streams.get(frame.s);
|
||||
if (!s || s.kind !== 'ws' || !s.rawSocket || !s.rawHead) return;
|
||||
this.wsUpgradeServer.handleUpgrade(s.upgradeRequest, s.rawSocket, s.rawHead, (ws) => {
|
||||
s.clientWs = ws;
|
||||
s.rawSocket = undefined;
|
||||
s.rawHead = undefined;
|
||||
ws.on('message', (msg, isBin) => {
|
||||
if (isBin) {
|
||||
this.sendBinary(BinaryFrameType.WsMessageBinary, frame.s, wsDataToBuffer(msg) ?? Buffer.alloc(0));
|
||||
} else {
|
||||
this.sendJson({ t: 'ws_msg_text', s: frame.s, data: wsDataToString(msg) ?? '' });
|
||||
}
|
||||
});
|
||||
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);
|
||||
}
|
||||
});
|
||||
ws.on('error', () => {
|
||||
if (this.streams.has(frame.s)) this.streams.delete(frame.s);
|
||||
});
|
||||
});
|
||||
break;
|
||||
}
|
||||
case 'ws_reject': {
|
||||
const s = this.streams.get(frame.s);
|
||||
if (!s || s.kind !== 'ws' || !s.rawSocket) return;
|
||||
try {
|
||||
s.rawSocket.write(`HTTP/1.1 ${frame.status} ${frame.message}\r\n\r\n`);
|
||||
s.rawSocket.destroy();
|
||||
} catch { /* ignore */ }
|
||||
this.streams.delete(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 */ }
|
||||
break;
|
||||
}
|
||||
case 'ws_close': {
|
||||
const s = this.streams.get(frame.s);
|
||||
if (!s || s.kind !== 'ws') return;
|
||||
if (s.clientWs) {
|
||||
try { s.clientWs.close(frame.code, frame.reason); } catch { /* ignore */ }
|
||||
} else if (s.rawSocket) {
|
||||
try { s.rawSocket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
this.streams.delete(frame.s);
|
||||
break;
|
||||
}
|
||||
case 'ctrl': {
|
||||
// Primary-side bridge does not act on control ops today; the
|
||||
// upgrade handler consumes enroll_ack before registerTunnel is
|
||||
// called, and ping/pong are handled by the WS layer.
|
||||
break;
|
||||
}
|
||||
default:
|
||||
// Ignore unknown JSON frame types for forward compatibility.
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private handleBinaryFrame(frame: DecodedBinaryFrame): void {
|
||||
const s = this.streams.get(frame.streamId);
|
||||
if (!s) return;
|
||||
switch (frame.type) {
|
||||
case BinaryFrameType.HttpResBody: {
|
||||
if (s.kind !== 'http') return;
|
||||
if (!s.headersWritten) {
|
||||
// Agent sent body before headers; synthesize 200 so we don't drop data.
|
||||
try { s.res.writeHead(200); } catch { /* ignore */ }
|
||||
s.headersWritten = true;
|
||||
}
|
||||
try { s.res.write(frame.payload); } catch { /* ignore */ }
|
||||
break;
|
||||
}
|
||||
case BinaryFrameType.WsMessageBinary: {
|
||||
if (s.kind !== 'ws' || !s.clientWs) return;
|
||||
try { s.clientWs.send(frame.payload, { binary: true }); } catch { /* ignore */ }
|
||||
break;
|
||||
}
|
||||
case BinaryFrameType.HttpReqBody:
|
||||
// Agent never originates request bodies; ignore for defense-in-depth.
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private onTunnelClose(): void {
|
||||
if (this.closed) return;
|
||||
this.close(1006, 'tunnel closed');
|
||||
}
|
||||
|
||||
// --- Helpers ---
|
||||
|
||||
private sendJson(frame: Parameters<typeof encodeJsonFrame>[0]): void {
|
||||
if (this.tunnelWs.readyState !== WebSocket.OPEN) return;
|
||||
try { this.tunnelWs.send(encodeJsonFrame(frame)); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
private sendBinary(type: BinaryFrameType, streamId: number, payload: Buffer): void {
|
||||
if (this.tunnelWs.readyState !== WebSocket.OPEN) return;
|
||||
try { this.tunnelWs.send(encodeBinaryFrame(type, streamId, payload), { binary: true }); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
private teardownStream(state: StreamState): void {
|
||||
if (state.kind === 'http') {
|
||||
try {
|
||||
if (!state.headersWritten) {
|
||||
state.res.writeHead(502, { 'content-type': 'text/plain' });
|
||||
state.res.end('pilot tunnel closed');
|
||||
} else {
|
||||
state.res.end();
|
||||
}
|
||||
} catch { /* ignore */ }
|
||||
} else {
|
||||
if (state.clientWs) {
|
||||
try { state.clientWs.close(1011, 'tunnel closed'); } catch { /* ignore */ }
|
||||
} else if (state.rawSocket) {
|
||||
try { state.rawSocket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
import { EventEmitter } from 'events';
|
||||
import WebSocket from 'ws';
|
||||
import { PilotTunnelBridge } from './PilotTunnelBridge';
|
||||
import { DatabaseService } from './DatabaseService';
|
||||
import { PilotCloseCode } from '../pilot/protocol';
|
||||
|
||||
/**
|
||||
* PilotTunnelManager: singleton registry of active pilot tunnels.
|
||||
*
|
||||
* Each enrolled pilot-agent node holds one outbound WebSocket to the primary.
|
||||
* For every such tunnel we spin up a local loopback HTTP server that demuxes
|
||||
* requests into frames. Remote-proxy code paths (http-proxy-middleware and the
|
||||
* WebSocket upgrade handler) can then treat pilot nodes identically to standard
|
||||
* proxy nodes by pointing at the loopback URL.
|
||||
*
|
||||
* Events:
|
||||
* - 'tunnel-up' (nodeId: number) after a tunnel is accepted
|
||||
* - 'tunnel-down' (nodeId: number) after a tunnel closes (for any reason)
|
||||
*/
|
||||
export class PilotTunnelManager extends EventEmitter {
|
||||
private static instance: PilotTunnelManager;
|
||||
private bridges: Map<number, PilotTunnelBridge> = new Map();
|
||||
|
||||
private constructor() {
|
||||
super();
|
||||
this.setMaxListeners(50);
|
||||
}
|
||||
|
||||
public static getInstance(): PilotTunnelManager {
|
||||
if (!PilotTunnelManager.instance) {
|
||||
PilotTunnelManager.instance = new PilotTunnelManager();
|
||||
}
|
||||
return PilotTunnelManager.instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* Accept a newly handshaked pilot tunnel. Replaces any prior tunnel for the
|
||||
* same node (split-brain prevention): the previous bridge is closed
|
||||
* before the new one is installed.
|
||||
*
|
||||
* Resolves once the loopback HTTP server is listening.
|
||||
*/
|
||||
public async registerTunnel(nodeId: number, ws: WebSocket, agentVersion?: string): Promise<void> {
|
||||
const existing = this.bridges.get(nodeId);
|
||||
if (existing) {
|
||||
existing.close(PilotCloseCode.Replaced, 'replaced by newer tunnel');
|
||||
this.bridges.delete(nodeId);
|
||||
}
|
||||
|
||||
const bridge = new PilotTunnelBridge(nodeId, ws);
|
||||
bridge.once('closed', () => {
|
||||
if (this.bridges.get(nodeId) === bridge) {
|
||||
this.bridges.delete(nodeId);
|
||||
DatabaseService.getInstance().updateNodeStatus(nodeId, 'offline');
|
||||
this.emit('tunnel-down', nodeId);
|
||||
}
|
||||
});
|
||||
await bridge.start();
|
||||
|
||||
this.bridges.set(nodeId, bridge);
|
||||
const db = DatabaseService.getInstance();
|
||||
db.updateNodeStatus(nodeId, 'online');
|
||||
db.updateNode(nodeId, {
|
||||
pilot_last_seen: Date.now(),
|
||||
pilot_agent_version: agentVersion ?? null,
|
||||
});
|
||||
this.emit('tunnel-up', nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
public getLoopbackUrl(nodeId: number): string | null {
|
||||
const bridge = this.bridges.get(nodeId);
|
||||
return bridge ? bridge.getLoopbackUrl() : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* True if a tunnel for this node is registered and healthy.
|
||||
*/
|
||||
public hasActiveTunnel(nodeId: number): boolean {
|
||||
return this.bridges.has(nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Force-close a tunnel (e.g., on node deletion).
|
||||
*/
|
||||
public closeTunnel(nodeId: number, code = 1000, reason = 'closed by primary'): void {
|
||||
const bridge = this.bridges.get(nodeId);
|
||||
if (!bridge) return;
|
||||
bridge.close(code, reason);
|
||||
this.bridges.delete(nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Snapshot of currently active tunnels.
|
||||
*/
|
||||
public listActive(): Array<{ nodeId: number; loopbackUrl: string; connectedAt: number }> {
|
||||
return Array.from(this.bridges.entries()).map(([nodeId, bridge]) => ({
|
||||
nodeId,
|
||||
loopbackUrl: bridge.getLoopbackUrl(),
|
||||
connectedAt: bridge.getConnectedAt(),
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Record an application-level heartbeat from the agent.
|
||||
*/
|
||||
public touch(nodeId: number): void {
|
||||
if (!this.bridges.has(nodeId)) return;
|
||||
DatabaseService.getInstance().updateNode(nodeId, { pilot_last_seen: Date.now() });
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user