mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-08-20 15:22:59 +00:00
refactor(backend): extract remote proxy, WebSocket upgrade handler, and server factory (phase 3) (#733)
Phase 3 of the index.ts refactor. Pulls the remote HTTP/WS proxy plumbing,
the WebSocket upgrade dispatcher, and the http/WSS construction out of the
monolith. index.ts drops roughly 620 lines.
New modules:
- proxy/websocketProxy.ts: shared httpProxy.createProxyServer singleton
(used by both the HTTP proxy middleware and the remote WS forwarder)
- proxy/remoteNodeProxy.ts: createRemoteProxyMiddleware() factory; consumes
the isProxyExemptPath helper instead of open-coding the prefix list
- server.ts: createServer(app) returns { server, wss, pilotTunnelWss }
- services/FleetUpdateTrackerService.ts: singleton wrapping the in-flight
fleet update tracker Map with create()/resolve() helpers
- helpers/consoleSession.ts: mintConsoleSession(), isConsoleSessionScope()
- websocket/upgradeHandler.ts: attachUpgrade(server, deps) dispatcher that
runs the manual cookie/JWT verify and delegates to sub-handlers
- websocket/pilotTunnel.ts: handlePilotTunnel (pilot_enroll consumption and
pilot_tunnel registration)
- websocket/notifications.ts: /ws/notifications local subscriber
- websocket/remoteForwarder.ts: remote-node WS proxy with console_session
token exchange for interactive paths
- websocket/logs.ts: /api/stacks/:name/logs supervisor stream
- websocket/hostConsole.ts: /api/system/host-console PTY, Admiral-gated
- websocket/generic.ts: /ws exec + streamStats action dispatch, owns the
terminalWs single-instance reference
- websocket/reject.ts: shared rejectUpgrade helper (replaces five copies)
Service extension:
- NotificationService: setBroadcaster(fn) replaced by subscribe(ws) that
returns an unsubscriber; broadcastToSubscribers is now internal. Subscriber
set lives on the service rather than in index.ts.
Wiring in index.ts:
- const app = createApp() already in place from Phase 2
- const { server, wss, pilotTunnelWss } = createServer(app)
- attachUpgrade(server, { wss, pilotTunnelWss })
- app.use('/api/', createRemoteProxyMiddleware())
- /api/system/console-token route now uses mintConsoleSession()
- deploy/down/update routes read the streaming target via getTerminalWs()
(return type is WebSocket | undefined so the || undefined fallback is gone)
Code review fixes: five duplicated reject helpers collapsed into
websocket/reject.ts; dropped the createTracker/resolveTracker bind
aliases in index.ts so call sites go through the service directly;
removed em dashes; replaced req.url! with req.url || '/'.
This commit is contained in:
Anso
@@ -0,0 +1,121 @@
|
||||
import type { IncomingMessage } from 'http';
|
||||
import type { Duplex } from 'stream';
|
||||
import WebSocket, { WebSocketServer } from 'ws';
|
||||
import DockerController from '../services/DockerController';
|
||||
import { DatabaseService } from '../services/DatabaseService';
|
||||
import { NodeRegistry } from '../services/NodeRegistry';
|
||||
import { isDebugEnabled } from '../utils/debug';
|
||||
import { rejectUpgrade as reject } from './reject';
|
||||
|
||||
/**
|
||||
* Module-scope singleton: the most recent WebSocket to send
|
||||
* `{action: 'connectTerminal'}` receives streaming output from any subsequent
|
||||
* compose deploy/down/update. Routes that want to echo compose progress read
|
||||
* the current value via `getTerminalWs()`.
|
||||
*
|
||||
* Intentionally single-instance. If multiple clients connect, the last one
|
||||
* wins. This matches pre-refactor behavior; race-hardening is a separate
|
||||
* concern.
|
||||
*/
|
||||
let terminalWs: WebSocket | undefined;
|
||||
|
||||
export function getTerminalWs(): WebSocket | undefined {
|
||||
return terminalWs;
|
||||
}
|
||||
|
||||
interface GenericContext {
|
||||
decoded: { scope?: string; username?: string; tv?: number };
|
||||
isProxyToken: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle the generic `/ws` upgrade: terminal (container exec) and streaming
|
||||
* stats. Gates node-proxy tokens and non-admin users; the subsequent
|
||||
* connection handler processes `{action: ...}` messages from the client.
|
||||
*/
|
||||
export function handleGenericWs(
|
||||
req: IncomingMessage,
|
||||
socket: Duplex,
|
||||
head: Buffer,
|
||||
wss: WebSocketServer,
|
||||
ctx: GenericContext,
|
||||
): void {
|
||||
const { decoded, isProxyToken } = ctx;
|
||||
|
||||
if (isProxyToken) return reject(socket, 403, 'Forbidden');
|
||||
|
||||
// Admin enforcement: container exec requires admin role.
|
||||
// console_session tokens are already admin-gated at creation time.
|
||||
// API tokens reaching this point have full-admin scope (read-only /
|
||||
// deploy-only are blocked by the upgrade handler's scope gate).
|
||||
if (!decoded.scope) {
|
||||
const execUser = decoded.username ? DatabaseService.getInstance().getUserByUsername(decoded.username) : undefined;
|
||||
if (!execUser) {
|
||||
console.warn('[Exec] User account not found:', decoded.username);
|
||||
return reject(socket, 401, 'Unauthorized');
|
||||
}
|
||||
if (decoded.tv !== undefined && execUser.token_version !== decoded.tv) {
|
||||
console.warn('[Exec] Session invalidated (token version mismatch):', decoded.username);
|
||||
return reject(socket, 401, 'Unauthorized');
|
||||
}
|
||||
if (execUser.role !== 'admin') {
|
||||
console.warn('[Exec] Non-admin user rejected:', decoded.username);
|
||||
return reject(socket, 403, 'Forbidden');
|
||||
}
|
||||
}
|
||||
|
||||
if (isDebugEnabled()) {
|
||||
console.debug('[Exec:diag] WS upgrade for exec path', {
|
||||
username: decoded.username,
|
||||
scope: decoded.scope || 'user-session',
|
||||
});
|
||||
}
|
||||
wss.handleUpgrade(req, socket, head, (ws) => {
|
||||
wss.emit('connection', ws, req);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Wire up the `connection` handler on the main wss. Processes `{action}`
|
||||
* messages for `connectTerminal` (captures the ws for deploy-output
|
||||
* streaming), `streamStats`, and `execContainer`. `{type}` messages (input,
|
||||
* resize, ping) are handled by per-session listeners registered inside
|
||||
* `execContainer`'s closure.
|
||||
*/
|
||||
export function attachGenericConnectionHandlers(wss: WebSocketServer): void {
|
||||
wss.on('connection', (ws) => {
|
||||
console.log('WebSocket connected');
|
||||
|
||||
ws.on('message', (message) => {
|
||||
try {
|
||||
const data = JSON.parse(message.toString());
|
||||
if (!data.action) return;
|
||||
|
||||
if (data.action === 'connectTerminal') {
|
||||
terminalWs = ws;
|
||||
} else if (data.action === 'streamStats') {
|
||||
const requestedId = data.nodeId ? parseInt(data.nodeId, 10) : NodeRegistry.getInstance().getDefaultNodeId();
|
||||
// When a WS is proxied from a gateway to this remote instance, the
|
||||
// nodeId in the message belongs to the gateway's DB and won't
|
||||
// resolve locally. Fall back to local.
|
||||
let nodeId = requestedId;
|
||||
try { NodeRegistry.getInstance().getDocker(requestedId); } catch { nodeId = NodeRegistry.getInstance().getDefaultNodeId(); }
|
||||
DockerController.getInstance(nodeId).streamStats(data.containerId, ws).catch((err: Error) => {
|
||||
console.error('[WS] streamStats error:', err.message);
|
||||
if (ws.readyState === WebSocket.OPEN) ws.close();
|
||||
});
|
||||
} else if (data.action === 'execContainer') {
|
||||
const requestedId = data.nodeId ? parseInt(data.nodeId, 10) : NodeRegistry.getInstance().getDefaultNodeId();
|
||||
let nodeId = requestedId;
|
||||
try { NodeRegistry.getInstance().getDocker(requestedId); } catch { nodeId = NodeRegistry.getInstance().getDefaultNodeId(); }
|
||||
DockerController.getInstance(nodeId).execContainer(data.containerId, ws).catch((err: Error) => {
|
||||
console.error('[WS] execContainer error:', err.message);
|
||||
if (ws.readyState === WebSocket.OPEN) ws.close();
|
||||
});
|
||||
}
|
||||
} catch {
|
||||
// Malformed JSON - ignore silently
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
import type { IncomingMessage } from 'http';
|
||||
import type { Duplex } from 'stream';
|
||||
import WebSocket, { WebSocketServer } from 'ws';
|
||||
import path from 'path';
|
||||
import { FileSystemService } from '../services/FileSystemService';
|
||||
import { NodeRegistry } from '../services/NodeRegistry';
|
||||
import { HostTerminalService } from '../services/HostTerminalService';
|
||||
import {
|
||||
LicenseService,
|
||||
isLicenseTier,
|
||||
isLicenseVariant,
|
||||
normalizeTier,
|
||||
normalizeVariant,
|
||||
PROXY_TIER_HEADER,
|
||||
PROXY_VARIANT_HEADER,
|
||||
} from '../services/LicenseService';
|
||||
import { ROLE_PERMISSIONS, type PermissionAction } from '../middleware/permissions';
|
||||
import type { UserRole } from '../services/DatabaseService';
|
||||
import { getErrorMessage } from '../utils/errors';
|
||||
import { rejectUpgrade as reject } from './reject';
|
||||
|
||||
interface HostConsoleContext {
|
||||
nodeId: number;
|
||||
decoded: { scope?: string; username?: string };
|
||||
isProxyToken: boolean;
|
||||
wsResolvedUser: { username: string; role: UserRole; token_version: number } | undefined;
|
||||
stackParam: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle `/api/system/host-console` WebSocket upgrades.
|
||||
*
|
||||
* Enforces three gates before spawning the host PTY:
|
||||
* 1. Machine-credential rejection: node_proxy tokens cannot reach an
|
||||
* interactive host shell.
|
||||
* 2. RBAC: user session tokens require the `system:console` permission.
|
||||
* console_session tokens are pre-gated at issuance (see
|
||||
* `routes/console.ts`) and skip this check.
|
||||
* 3. License: host console requires paid + admiral. For console_session
|
||||
* tokens the tier/variant is trusted from the gateway-supplied headers;
|
||||
* otherwise the local LicenseService is consulted.
|
||||
*/
|
||||
export function handleHostConsoleWs(
|
||||
req: IncomingMessage,
|
||||
socket: Duplex,
|
||||
head: Buffer,
|
||||
ctx: HostConsoleContext,
|
||||
): void {
|
||||
const { nodeId, decoded, isProxyToken, wsResolvedUser, stackParam } = ctx;
|
||||
|
||||
if (isProxyToken) return reject(socket, 403, 'Forbidden');
|
||||
|
||||
const isConsoleSession = decoded.scope === 'console_session';
|
||||
if (!isConsoleSession) {
|
||||
const userRole = wsResolvedUser?.role;
|
||||
const consolePermission: PermissionAction = 'system:console';
|
||||
if (!userRole || !ROLE_PERMISSIONS[userRole]?.includes(consolePermission)) {
|
||||
console.log('[HostConsole] Access denied: insufficient permissions', {
|
||||
username: wsResolvedUser?.username || decoded.username,
|
||||
role: userRole,
|
||||
});
|
||||
return reject(socket, 403, 'Forbidden');
|
||||
}
|
||||
}
|
||||
|
||||
const consoleTierHeader = req.headers[PROXY_TIER_HEADER] as string | undefined;
|
||||
const consoleVariantHeader = req.headers[PROXY_VARIANT_HEADER] as string | undefined;
|
||||
const ls = LicenseService.getInstance();
|
||||
const consoleTier = (isConsoleSession && isLicenseTier(consoleTierHeader))
|
||||
? normalizeTier(consoleTierHeader)
|
||||
: ls.getTier();
|
||||
const consoleVariant = (isConsoleSession && consoleVariantHeader !== undefined && isLicenseVariant(consoleVariantHeader))
|
||||
? normalizeVariant(consoleVariantHeader)
|
||||
: ls.getVariant();
|
||||
if (consoleTier !== 'paid' || consoleVariant !== 'admiral') {
|
||||
return reject(socket, 403, 'Forbidden');
|
||||
}
|
||||
|
||||
const consoleUsername = wsResolvedUser?.username || decoded.username || 'console_session';
|
||||
console.log('[HostConsole] WebSocket upgrade accepted', {
|
||||
username: consoleUsername,
|
||||
nodeId,
|
||||
stack: stackParam || '(root)',
|
||||
});
|
||||
|
||||
const hostConsoleWss = new WebSocketServer({ noServer: true });
|
||||
hostConsoleWss.handleUpgrade(req, socket, head, (ws) => {
|
||||
hostConsoleWss.close();
|
||||
let targetDirectory = '';
|
||||
try {
|
||||
const baseDir = FileSystemService.getInstance(nodeId).getBaseDir();
|
||||
if (stackParam) {
|
||||
const resolved = path.resolve(baseDir, stackParam);
|
||||
if (!resolved.startsWith(path.resolve(baseDir))) {
|
||||
ws.send('Error: Invalid stack path\r\n');
|
||||
ws.close();
|
||||
return;
|
||||
}
|
||||
targetDirectory = resolved;
|
||||
} else {
|
||||
targetDirectory = baseDir;
|
||||
}
|
||||
} catch {
|
||||
targetDirectory = FileSystemService.getInstance(NodeRegistry.getInstance().getDefaultNodeId()).getBaseDir();
|
||||
}
|
||||
try {
|
||||
HostTerminalService.spawnTerminal(ws, targetDirectory, consoleUsername);
|
||||
} catch (error) {
|
||||
console.error('[HostConsole] Unhandled spawn error:', { user: consoleUsername, error: getErrorMessage(error, 'unknown') });
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
ws.send('Error: Failed to start terminal session.\r\n');
|
||||
ws.close();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
import type { IncomingMessage } from 'http';
|
||||
import type { Duplex } from 'stream';
|
||||
import WebSocket, { WebSocketServer } from 'ws';
|
||||
import { ComposeService } from '../services/ComposeService';
|
||||
import { isValidStackName } from '../utils/validation';
|
||||
import { isDebugEnabled } from '../utils/debug';
|
||||
import { getErrorMessage } from '../utils/errors';
|
||||
|
||||
/**
|
||||
* Handle `/api/stacks/:stackName/logs` WebSocket upgrades. Streams the
|
||||
* supervisor log loop for the given stack on the caller's node context.
|
||||
*/
|
||||
export function handleLogsWs(
|
||||
req: IncomingMessage,
|
||||
socket: Duplex,
|
||||
head: Buffer,
|
||||
opts: { nodeId: number; stackName: string },
|
||||
): void {
|
||||
const { nodeId, stackName } = opts;
|
||||
const logsWss = new WebSocketServer({ noServer: true });
|
||||
logsWss.handleUpgrade(req, socket, head, (ws) => {
|
||||
// Close the per-connection server immediately after the upgrade completes.
|
||||
// The wss instance is only needed to negotiate the handshake; keeping it
|
||||
// open accumulates listeners and allocates memory for every connection.
|
||||
logsWss.close();
|
||||
if (!isValidStackName(stackName)) {
|
||||
ws.send('Error: Invalid stack name\r\n');
|
||||
ws.close();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
if (isDebugEnabled()) console.debug('[Stacks:debug] WS log stream opened', { stackName, nodeId });
|
||||
ws.on('close', () => {
|
||||
if (isDebugEnabled()) console.debug('[Stacks:debug] WS log stream closed', { stackName, nodeId });
|
||||
});
|
||||
ComposeService.getInstance(nodeId).streamLogs(stackName, ws);
|
||||
} catch (error) {
|
||||
console.error('[Stacks] Failed to stream logs:', error);
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
ws.send(`Error streaming logs: ${getErrorMessage(error, 'unknown')}\n`);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
import type { IncomingMessage } from 'http';
|
||||
import type { Duplex } from 'stream';
|
||||
import { WebSocketServer } from 'ws';
|
||||
import { NotificationService } from '../services/NotificationService';
|
||||
|
||||
/**
|
||||
* Accept a `/ws/notifications` upgrade, register the resulting socket as a
|
||||
* NotificationService subscriber, and wire up cleanup on close/error.
|
||||
*
|
||||
* The per-connection `WebSocketServer` exists only to negotiate the handshake
|
||||
* and is closed immediately afterward to avoid accumulating listeners.
|
||||
*/
|
||||
export function handleNotificationsWs(
|
||||
req: IncomingMessage,
|
||||
socket: Duplex,
|
||||
head: Buffer,
|
||||
): void {
|
||||
const notifWss = new WebSocketServer({ noServer: true });
|
||||
notifWss.handleUpgrade(req, socket, head, (ws) => {
|
||||
notifWss.close();
|
||||
const unsubscribe = NotificationService.getInstance().subscribe(ws);
|
||||
ws.on('close', unsubscribe);
|
||||
ws.on('error', () => {
|
||||
unsubscribe();
|
||||
ws.terminate();
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
import type { IncomingMessage } from 'http';
|
||||
import type { Duplex } from 'stream';
|
||||
import type { WebSocketServer } from 'ws';
|
||||
import jwt from 'jsonwebtoken';
|
||||
import crypto from 'crypto';
|
||||
import { DatabaseService } from '../services/DatabaseService';
|
||||
import { PilotTunnelManager } from '../services/PilotTunnelManager';
|
||||
import { encodeJsonFrame as encodePilotJsonFrame, PROTOCOL_VERSION as PILOT_PROTOCOL_VERSION } from '../pilot/protocol';
|
||||
import { getErrorMessage } from '../utils/errors';
|
||||
import { rejectUpgrade as rejectSocket } from './reject';
|
||||
|
||||
/**
|
||||
* Handle an inbound pilot-agent tunnel upgrade. Accepts either:
|
||||
* - pilot_enroll (15m, one-time): consume the enrollment row, mint a
|
||||
* long-lived pilot_tunnel token, send it back in a ctrl enroll_ack frame.
|
||||
* - pilot_tunnel (365d): accept the socket directly.
|
||||
*
|
||||
* In both cases the accepted WebSocket is handed to `PilotTunnelManager`.
|
||||
* Handled independently of user/session auth because these are machine
|
||||
* credentials and carry no cookies.
|
||||
*/
|
||||
export async function handlePilotTunnel(
|
||||
req: IncomingMessage,
|
||||
socket: Duplex,
|
||||
head: Buffer,
|
||||
pilotTunnelWss: WebSocketServer,
|
||||
): Promise<void> {
|
||||
const authHeader = req.headers['authorization'];
|
||||
const header = Array.isArray(authHeader) ? authHeader[0] : authHeader;
|
||||
const token = header?.startsWith('Bearer ') ? header.slice(7) : null;
|
||||
if (!token) return rejectSocket(socket, 401, 'Unauthorized');
|
||||
|
||||
const db = DatabaseService.getInstance();
|
||||
const jwtSecret = db.getGlobalSettings().auth_jwt_secret;
|
||||
if (!jwtSecret) return rejectSocket(socket, 500, 'Internal Server Error');
|
||||
|
||||
let decoded: { scope?: string; nodeId?: number; enrollNonce?: string };
|
||||
try {
|
||||
decoded = jwt.verify(token, jwtSecret) as typeof decoded;
|
||||
} catch {
|
||||
return rejectSocket(socket, 401, 'Unauthorized');
|
||||
}
|
||||
|
||||
if (decoded.scope !== 'pilot_enroll' && decoded.scope !== 'pilot_tunnel') {
|
||||
return rejectSocket(socket, 403, 'Forbidden');
|
||||
}
|
||||
if (typeof decoded.nodeId !== 'number') return rejectSocket(socket, 400, 'Bad Request');
|
||||
|
||||
const node = db.getNode(decoded.nodeId);
|
||||
if (!node || node.type !== 'remote' || node.mode !== 'pilot_agent') {
|
||||
return rejectSocket(socket, 404, 'Not Found');
|
||||
}
|
||||
|
||||
let mintedTunnelToken: string | null = null;
|
||||
if (decoded.scope === 'pilot_enroll') {
|
||||
const tokenHash = crypto.createHash('sha256').update(token).digest('hex');
|
||||
const row = db.consumePilotEnrollment(tokenHash);
|
||||
if (!row || row.node_id !== decoded.nodeId) {
|
||||
return rejectSocket(socket, 401, 'Unauthorized');
|
||||
}
|
||||
mintedTunnelToken = jwt.sign(
|
||||
{ scope: 'pilot_tunnel', nodeId: decoded.nodeId },
|
||||
jwtSecret,
|
||||
{ expiresIn: '365d' },
|
||||
);
|
||||
}
|
||||
|
||||
const agentVersionHeader = req.headers['x-sencho-agent-version'];
|
||||
const agentVersion = Array.isArray(agentVersionHeader) ? agentVersionHeader[0] : agentVersionHeader;
|
||||
|
||||
pilotTunnelWss.handleUpgrade(req, socket, head, async (ws) => {
|
||||
try {
|
||||
ws.send(encodePilotJsonFrame({
|
||||
t: 'hello',
|
||||
version: PILOT_PROTOCOL_VERSION,
|
||||
role: 'primary',
|
||||
}));
|
||||
if (mintedTunnelToken) {
|
||||
ws.send(encodePilotJsonFrame({
|
||||
t: 'ctrl',
|
||||
op: 'enroll_ack',
|
||||
payload: { token: mintedTunnelToken, nodeId: decoded.nodeId },
|
||||
}));
|
||||
}
|
||||
} catch {
|
||||
try { ws.close(1011, 'hello failed'); } catch { /* ignore */ }
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
await PilotTunnelManager.getInstance().registerTunnel(decoded.nodeId!, ws, agentVersion);
|
||||
} catch (err) {
|
||||
console.error('[Pilot] Failed to register tunnel:', getErrorMessage(err, 'unknown'));
|
||||
try { ws.close(1011, 'registration failed'); } catch { /* ignore */ }
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
import type { Duplex } from 'stream';
|
||||
|
||||
/**
|
||||
* Write an HTTP status line and destroy the socket. Used by every WebSocket
|
||||
* handler to reject an upgrade before a successful handshake. Errors during
|
||||
* write/destroy are intentionally swallowed: the socket is already being
|
||||
* torn down and nothing downstream can recover.
|
||||
*/
|
||||
export function rejectUpgrade(socket: Duplex, status: number, message: string): void {
|
||||
try { socket.write(`HTTP/1.1 ${status} ${message}\r\n\r\n`); } catch { /* ignore */ }
|
||||
try { socket.destroy(); } catch { /* ignore */ }
|
||||
}
|
||||
@@ -0,0 +1,80 @@
|
||||
import type { IncomingMessage } from 'http';
|
||||
import type { Duplex } from 'stream';
|
||||
import type { Node } from '../services/DatabaseService';
|
||||
import {
|
||||
LicenseService,
|
||||
PROXY_TIER_HEADER,
|
||||
PROXY_VARIANT_HEADER,
|
||||
} from '../services/LicenseService';
|
||||
import { wsProxyServer } from '../proxy/websocketProxy';
|
||||
import { getErrorMessage } from '../utils/errors';
|
||||
import { rejectUpgrade as reject } from './reject';
|
||||
|
||||
/**
|
||||
* Forward a WebSocket upgrade to a remote Sencho instance. Handles the
|
||||
* console_session token exchange for interactive paths so the long-lived
|
||||
* api_token never reaches an interactive terminal (the remote's upgrade
|
||||
* handler rejects node_proxy tokens on those paths).
|
||||
*
|
||||
* The caller must have already established that `node.type === 'remote'`
|
||||
* and that `api_url` + `api_token` are present.
|
||||
*/
|
||||
export async function handleRemoteForwarder(
|
||||
req: IncomingMessage,
|
||||
socket: Duplex,
|
||||
head: Buffer,
|
||||
opts: { node: Node; pathname: string },
|
||||
): Promise<void> {
|
||||
const { node, pathname } = opts;
|
||||
// Guaranteed non-null by caller; assert here so the rest of the function is nullsafe.
|
||||
if (!node.api_url || !node.api_token) return reject(socket, 503, 'Service Unavailable');
|
||||
|
||||
const wsTarget = node.api_url.replace(/\/$/, '').replace(/^https?/, (m) => m === 'https' ? 'wss' : 'ws');
|
||||
|
||||
// Interactive console paths (host console / container exec) are guarded on
|
||||
// the remote by an isProxyToken check that rejects the long-lived api_token.
|
||||
// Exchange it for a short-lived console_session token before forwarding so
|
||||
// the remote allows the connection while keeping the guard intact for
|
||||
// direct api_token access.
|
||||
const isInteractiveConsolePath = pathname === '/api/system/host-console' || pathname === '/ws';
|
||||
let bearerTokenForProxy = node.api_token;
|
||||
if (isInteractiveConsolePath) {
|
||||
try {
|
||||
const ls = LicenseService.getInstance();
|
||||
const tokenRes = await fetch(`${node.api_url.replace(/\/$/, '')}/api/system/console-token`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Authorization': `Bearer ${node.api_token}`,
|
||||
[PROXY_TIER_HEADER]: ls.getTier(),
|
||||
[PROXY_VARIANT_HEADER]: ls.getVariant() || '',
|
||||
},
|
||||
});
|
||||
if (!tokenRes.ok) {
|
||||
console.error(`[WS Proxy] Remote console-token request failed: ${tokenRes.status}`);
|
||||
return reject(socket, 502, 'Bad Gateway');
|
||||
}
|
||||
const data = await tokenRes.json() as { token?: string };
|
||||
if (typeof data.token === 'string') bearerTokenForProxy = data.token;
|
||||
} catch (e) {
|
||||
console.error('[WS Proxy] Failed to fetch remote console token:', getErrorMessage(e, 'unknown'));
|
||||
return reject(socket, 502, 'Bad Gateway');
|
||||
}
|
||||
}
|
||||
|
||||
req.headers['authorization'] = `Bearer ${bearerTokenForProxy}`;
|
||||
delete req.headers['x-node-id'];
|
||||
// Strip the browser's session cookie: signed by this instance's JWT secret
|
||||
// and would fail verification on the remote. Auth is handled exclusively
|
||||
// via the Bearer token.
|
||||
delete req.headers['cookie'];
|
||||
const wsLs = LicenseService.getInstance();
|
||||
req.headers[PROXY_TIER_HEADER] = wsLs.getTier();
|
||||
req.headers[PROXY_VARIANT_HEADER] = wsLs.getVariant() || '';
|
||||
// Strip nodeId from the forwarded URL so the remote treats the request as
|
||||
// local. The remote has no record of the gateway's nodeId; leaving it would
|
||||
// trigger nodeContext's 404 branch.
|
||||
const fwdUrl = new URL(req.url || '/', `http://${req.headers.host || 'localhost'}`);
|
||||
fwdUrl.searchParams.delete('nodeId');
|
||||
req.url = fwdUrl.pathname + (fwdUrl.searchParams.toString() ? `?${fwdUrl.searchParams.toString()}` : '');
|
||||
wsProxyServer.ws(req, socket, head, { target: wsTarget });
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
import type http from 'http';
|
||||
import type { IncomingMessage } from 'http';
|
||||
import type { WebSocketServer } from 'ws';
|
||||
import jwt from 'jsonwebtoken';
|
||||
import crypto from 'crypto';
|
||||
import { DatabaseService, type UserRole } from '../services/DatabaseService';
|
||||
import { NodeRegistry } from '../services/NodeRegistry';
|
||||
import { COOKIE_NAME } from '../helpers/constants';
|
||||
import { handlePilotTunnel } from './pilotTunnel';
|
||||
import { handleNotificationsWs } from './notifications';
|
||||
import { handleRemoteForwarder } from './remoteForwarder';
|
||||
import { handleLogsWs } from './logs';
|
||||
import { handleHostConsoleWs } from './hostConsole';
|
||||
import { handleGenericWs, attachGenericConnectionHandlers } from './generic';
|
||||
import { rejectUpgrade as reject } from './reject';
|
||||
|
||||
function parseCookies(req: IncomingMessage): Record<string, string> {
|
||||
const header = req.headers.cookie || '';
|
||||
return Object.fromEntries(
|
||||
header
|
||||
.split(';')
|
||||
.map((c) => c.trim().split('='))
|
||||
.filter(([k, v]) => k && v),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach the upgrade dispatcher to the HTTP server and wire the generic
|
||||
* `connection` handler on the main wss.
|
||||
*
|
||||
* Dispatch order (first match wins):
|
||||
* 1. `/api/pilot/tunnel` -> handlePilotTunnel (own auth, own wss)
|
||||
* 2. shared cookie/Bearer auth + JWT verify (rejects unauthenticated)
|
||||
* 3. API token scope gate (read-only / deploy-only restricted to logs + notifications)
|
||||
* 4. `/ws/notifications` local -> handleNotificationsWs
|
||||
* 5. remote nodeId path -> handleRemoteForwarder
|
||||
* 6. `/api/stacks/:name/logs` -> handleLogsWs
|
||||
* 7. `/api/system/host-console` -> handleHostConsoleWs
|
||||
* 8. fallback -> handleGenericWs (`/ws` exec + stats)
|
||||
*/
|
||||
export function attachUpgrade(
|
||||
server: http.Server,
|
||||
deps: { wss: WebSocketServer; pilotTunnelWss: WebSocketServer },
|
||||
): void {
|
||||
const { wss, pilotTunnelWss } = deps;
|
||||
|
||||
attachGenericConnectionHandlers(wss);
|
||||
|
||||
server.on('upgrade', async (req, socket, head) => {
|
||||
// Pilot-agent tunnel ingress: machine credentials, no cookies.
|
||||
try {
|
||||
const reqUrl = new URL(req.url || '/', `http://${req.headers.host || 'localhost'}`);
|
||||
if (reqUrl.pathname === '/api/pilot/tunnel') {
|
||||
await handlePilotTunnel(req, socket, head, pilotTunnelWss);
|
||||
return;
|
||||
}
|
||||
} catch {
|
||||
// URL parse error falls through and will be rejected below.
|
||||
}
|
||||
|
||||
const cookies = parseCookies(req);
|
||||
const cookieToken = cookies[COOKIE_NAME];
|
||||
const authHeader = req.headers['authorization'] as string | undefined;
|
||||
const bearerToken = authHeader?.startsWith('Bearer ') ? authHeader.slice(7) : null;
|
||||
// Prefer Bearer over cookie: node-to-node proxy upgrades carry a Bearer
|
||||
// token and must not be shadowed by a browser cookie signed with a
|
||||
// different instance's JWT secret.
|
||||
const token = bearerToken || cookieToken;
|
||||
|
||||
if (!token) return reject(socket, 401, 'Unauthorized');
|
||||
|
||||
try {
|
||||
const settings = DatabaseService.getInstance().getGlobalSettings();
|
||||
const jwtSecret = settings.auth_jwt_secret;
|
||||
if (!jwtSecret) throw new Error('No JWT secret');
|
||||
const decoded = jwt.verify(token, jwtSecret) as { username?: string; scope?: string; role?: string; tv?: number };
|
||||
|
||||
// Node proxy tokens are machine-to-machine credentials and must never be
|
||||
// granted interactive terminal access (host console or container exec).
|
||||
const isProxyToken = decoded.scope === 'node_proxy';
|
||||
|
||||
let wsApiTokenScope: string | null = null;
|
||||
if (decoded.scope === 'api_token') {
|
||||
const tokenHash = crypto.createHash('sha256').update(token).digest('hex');
|
||||
const apiToken = DatabaseService.getInstance().getApiTokenByHash(tokenHash);
|
||||
if (!apiToken || apiToken.revoked_at) return reject(socket, 401, 'Unauthorized');
|
||||
if (apiToken.expires_at && apiToken.expires_at < Date.now()) return reject(socket, 401, 'Unauthorized');
|
||||
DatabaseService.getInstance().updateApiTokenLastUsed(apiToken.id);
|
||||
wsApiTokenScope = apiToken.scope;
|
||||
}
|
||||
|
||||
// For user session tokens (no scope), resolve against DB for up-to-date
|
||||
// role and token_version checks. Scoped tokens (api_token, node_proxy,
|
||||
// console_session) skip this: they are validated by their own logic
|
||||
// above or by the gateway that issued them.
|
||||
let wsResolvedUser: { username: string; role: UserRole; token_version: number } | undefined;
|
||||
if (!decoded.scope && decoded.username) {
|
||||
const dbUser = DatabaseService.getInstance().getUserByUsername(decoded.username);
|
||||
if (!dbUser) return reject(socket, 401, 'Unauthorized');
|
||||
if (decoded.tv !== undefined && dbUser.token_version !== decoded.tv) {
|
||||
console.log('[Auth] WS session rejected: token version mismatch for:', decoded.username);
|
||||
return reject(socket, 401, 'Unauthorized');
|
||||
}
|
||||
wsResolvedUser = {
|
||||
username: dbUser.username,
|
||||
role: dbUser.role as UserRole,
|
||||
token_version: dbUser.token_version,
|
||||
};
|
||||
}
|
||||
|
||||
const parsedUrl = new URL(req.url || '', `http://${req.headers.host || 'localhost'}`);
|
||||
const pathname = parsedUrl.pathname;
|
||||
|
||||
// Gate WebSocket paths by API token scope
|
||||
if (wsApiTokenScope) {
|
||||
const isLogPath = /^\/api\/stacks\/[^/]+\/logs$/.test(pathname);
|
||||
const isNotifPath = pathname === '/ws/notifications';
|
||||
if (wsApiTokenScope === 'read-only' || wsApiTokenScope === 'deploy-only') {
|
||||
if (!isLogPath && !isNotifPath) return reject(socket, 403, 'Forbidden');
|
||||
}
|
||||
}
|
||||
|
||||
const nodeIdParam = parsedUrl.searchParams.get('nodeId');
|
||||
const nodeId = nodeIdParam ? parseInt(nodeIdParam, 10) : NodeRegistry.getInstance().getDefaultNodeId();
|
||||
const node = NodeRegistry.getInstance().getNode(nodeId);
|
||||
|
||||
// Notification push channel: local only when no remote nodeId is
|
||||
// specified. When a remote nodeId is provided, fall through to the
|
||||
// forwarder so the browser subscribes to that remote node's push stream.
|
||||
if (pathname === '/ws/notifications' && (!node || node.type !== 'remote')) {
|
||||
handleNotificationsWs(req, socket, head);
|
||||
return;
|
||||
}
|
||||
|
||||
if (node && node.type === 'remote' && node.api_url && node.api_token) {
|
||||
await handleRemoteForwarder(req, socket, head, { node, pathname });
|
||||
return;
|
||||
}
|
||||
|
||||
const logsMatch = pathname.match(/^\/api\/stacks\/([^/]+)\/logs$/);
|
||||
if (logsMatch) {
|
||||
handleLogsWs(req, socket, head, { nodeId, stackName: decodeURIComponent(logsMatch[1]) });
|
||||
return;
|
||||
}
|
||||
|
||||
if (pathname.startsWith('/api/system/host-console')) {
|
||||
handleHostConsoleWs(req, socket, head, {
|
||||
nodeId,
|
||||
decoded,
|
||||
isProxyToken,
|
||||
wsResolvedUser,
|
||||
stackParam: parsedUrl.searchParams.get('stack'),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
handleGenericWs(req, socket, head, wss, { decoded, isProxyToken });
|
||||
} catch {
|
||||
return reject(socket, 401, 'Unauthorized');
|
||||
}
|
||||
});
|
||||
}
|
||||
Reference in New Issue
Block a user