feat(desktop): proxy realtime requests with runtime headers

2026-06-30 01:21:42 +03:00
parent 359c73fcf3
commit c10930dfd0
9 changed files with 724 additions and 7 deletions
@@ -0,0 +1,279 @@
+import { WebSocket, WebSocketServer } from 'ws';
+
+const PROXY_SSE_PATH = '/api/openchamber/realtime-proxy/sse';
+const PROXY_WS_PATH = '/api/openchamber/realtime-proxy/ws';
+
+const isAllowedSsePath = (pathname) => {
+  return pathname === '/api/event'
+    || pathname === '/api/global/event'
+    || pathname === '/api/openchamber/events'
+    || pathname === '/api/notifications/stream'
+    || /^\/api\/terminal\/[^/]+\/stream$/.test(pathname);
+};
+
+const isAllowedWebSocketPath = (pathname) => {
+  return pathname === '/api/event/ws'
+    || pathname === '/api/global/event/ws'
+    || pathname === '/api/terminal/ws';
+};
+
+const normalizeBaseUrl = (value) => {
+  if (typeof value !== 'string') return '';
+  return value.trim().replace(/\/+$/, '');
+};
+
+const sanitizeHeaders = (headers) => {
+  if (!headers || typeof headers !== 'object') return {};
+  const next = {};
+  for (const [rawName, rawValue] of Object.entries(headers)) {
+    const name = typeof rawName === 'string' ? rawName.trim() : '';
+    const value = typeof rawValue === 'string' ? rawValue.trim() : '';
+    if (!name || !value || /[\r\n:]/.test(name) || /[\r\n]/.test(value)) continue;
+    if (name.toLowerCase() === 'authorization') continue;
+    next[name] = value;
+  }
+  return next;
+};
+
+const hasHeaders = (headers) => Object.keys(headers).length > 0;
+
+const getTargetParam = (req) => {
+  let raw = typeof req.query?.url === 'string' ? req.query.url : '';
+  if (!raw) {
+    try {
+      raw = new URL(req.url || '/', 'http://127.0.0.1').searchParams.get('url') || '';
+    } catch {
+      raw = '';
+    }
+  }
+  if (!raw) return null;
+  try {
+    return new URL(raw);
+  } catch {
+    return null;
+  }
+};
+
+const urlsMatchRuntime = (target, apiBaseUrl) => {
+  const base = normalizeBaseUrl(apiBaseUrl);
+  if (!base) return false;
+  try {
+    const baseUrl = new URL(base);
+    const targetForCompare = new URL(target.toString());
+    if (targetForCompare.protocol === 'ws:') targetForCompare.protocol = 'http:';
+    if (targetForCompare.protocol === 'wss:') targetForCompare.protocol = 'https:';
+    return targetForCompare.origin === baseUrl.origin;
+  } catch {
+    return false;
+  }
+};
+
+const protocolMatchesProxyType = (target, type) => {
+  if (type === 'ws') return target.protocol === 'ws:' || target.protocol === 'wss:';
+  return target.protocol === 'http:' || target.protocol === 'https:';
+};
+
+const pathMatchesProxyType = (target, type) => {
+  return type === 'ws' ? isAllowedWebSocketPath(target.pathname) : isAllowedSsePath(target.pathname);
+};
+
+const resolveProxyTarget = (req, getDesktopRuntimeConfig, type) => {
+  const config = typeof getDesktopRuntimeConfig === 'function' ? getDesktopRuntimeConfig() : null;
+  const requestHeaders = sanitizeHeaders(config?.requestHeaders);
+  const apiBaseUrl = normalizeBaseUrl(config?.apiBaseUrl);
+  const target = getTargetParam(req);
+  if (!target || !apiBaseUrl || !hasHeaders(requestHeaders)) return null;
+  if (!protocolMatchesProxyType(target, type)) return null;
+  if (!pathMatchesProxyType(target, type)) return null;
+  if (!urlsMatchRuntime(target, apiBaseUrl)) return null;
+  return { target, requestHeaders };
+};
+
+const safeHeader = (headers, name) => {
+  const value = headers?.[name.toLowerCase()];
+  if (Array.isArray(value)) return value.find((item) => typeof item === 'string' && item.trim()) || '';
+  return typeof value === 'string' ? value.trim() : '';
+};
+
+const buildSseRequestHeaders = (req, requestHeaders) => {
+  const headers = {};
+  const accept = safeHeader(req.headers, 'accept');
+  const lastEventId = safeHeader(req.headers, 'last-event-id');
+  if (accept) headers.Accept = accept;
+  if (lastEventId) headers['Last-Event-ID'] = lastEventId;
+  return { ...headers, ...requestHeaders };
+};
+
+const rejectWebSocketUpgrade = (socket, statusCode, message) => {
+  socket.write(`HTTP/1.1 ${statusCode} ${message}\r\nConnection: close\r\n\r\n`);
+  socket.destroy();
+};
+
+export const buildRealtimeProxySseUrl = (localOrigin, targetUrl) => {
+  const url = new URL(PROXY_SSE_PATH, localOrigin);
+  url.searchParams.set('url', targetUrl);
+  return url.toString();
+};
+
+export const buildRealtimeProxyWsUrl = (localOrigin, targetUrl) => {
+  const url = new URL(PROXY_WS_PATH, localOrigin);
+  url.searchParams.set('url', targetUrl);
+  url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
+  return url.toString();
+};
+
+export const attachRealtimeProxy = ({ app, server, getDesktopRuntimeConfig, getUiAuthController, isRequestOriginAllowed }) => {
+  if (!app || !server || typeof getDesktopRuntimeConfig !== 'function') {
+    return { stop: () => {} };
+  }
+
+  const originAllowed = async (req) => {
+    if (typeof isRequestOriginAllowed !== 'function') return false;
+    try {
+      return await isRequestOriginAllowed(req);
+    } catch {
+      return false;
+    }
+  };
+
+  const ensureAuthenticated = async (req, res) => {
+    const controller = typeof getUiAuthController === 'function' ? getUiAuthController() : null;
+    if (typeof controller?.ensureSessionToken !== 'function') return false;
+    const response = res || { setHeader: () => {} };
+    const token = await controller.ensureSessionToken(req, response);
+    return Boolean(token);
+  };
+
+  app.get(PROXY_SSE_PATH, async (req, res) => {
+    if (!await ensureAuthenticated(req, res)) {
+      res.status(401).json({ error: 'UI authentication required' });
+      return;
+    }
+    if (!await originAllowed(req)) {
+      res.status(403).json({ error: 'Realtime proxy origin is not allowed' });
+      return;
+    }
+    const resolved = resolveProxyTarget(req, getDesktopRuntimeConfig, 'sse');
+    if (!resolved) {
+      res.status(404).json({ error: 'Realtime proxy is unavailable' });
+      return;
+    }
+
+    const abort = new AbortController();
+    req.on('close', () => abort.abort());
+    try {
+      const response = await fetch(resolved.target.toString(), {
+        headers: buildSseRequestHeaders(req, resolved.requestHeaders),
+        signal: abort.signal,
+      });
+      if (!response.ok || !response.body) {
+        res.status(response.status || 502).end();
+        return;
+      }
+
+      res.status(response.status);
+      res.setHeader('Content-Type', response.headers.get('content-type') || 'text/event-stream');
+      res.setHeader('Cache-Control', response.headers.get('cache-control') || 'no-cache');
+      res.setHeader('Connection', 'keep-alive');
+
+      for await (const chunk of response.body) {
+        if (abort.signal.aborted) break;
+        res.write(chunk);
+      }
+      res.end();
+    } catch (error) {
+      if (!abort.signal.aborted && !res.headersSent) {
+        res.status(502).json({ error: error instanceof Error ? error.message : 'Realtime proxy failed' });
+      } else if (!res.destroyed) {
+        res.end();
+      }
+    }
+  });
+
+  const wsServer = new WebSocketServer({ noServer: true });
+
+  wsServer.on('connection', (client, request) => {
+    const resolved = resolveProxyTarget(request, getDesktopRuntimeConfig, 'ws');
+    if (!resolved) {
+      client.close(1008, 'Realtime proxy is unavailable');
+      return;
+    }
+
+    const upstream = new WebSocket(resolved.target.toString(), {
+      headers: resolved.requestHeaders,
+    });
+    const pending = [];
+
+    const flush = () => {
+      while (pending.length > 0 && upstream.readyState === WebSocket.OPEN) {
+        const [data, isBinary] = pending.shift();
+        upstream.send(data, { binary: isBinary });
+      }
+    };
+
+    client.on('message', (data, isBinary) => {
+      if (upstream.readyState === WebSocket.OPEN) {
+        upstream.send(data, { binary: isBinary });
+        return;
+      }
+      if (upstream.readyState === WebSocket.CONNECTING) {
+        pending.push([data, isBinary]);
+      }
+    });
+    upstream.on('open', flush);
+    upstream.on('message', (data, isBinary) => {
+      if (client.readyState === WebSocket.OPEN) {
+        client.send(data, { binary: isBinary });
+      }
+    });
+    upstream.on('close', (code, reason) => {
+      if (client.readyState === WebSocket.OPEN || client.readyState === WebSocket.CONNECTING) {
+        client.close(code || 1000, reason);
+      }
+    });
+    upstream.on('error', () => {
+      if (client.readyState === WebSocket.OPEN || client.readyState === WebSocket.CONNECTING) {
+        client.close(1011, 'Realtime proxy upstream error');
+      }
+    });
+    client.on('close', () => {
+      if (upstream.readyState === WebSocket.OPEN || upstream.readyState === WebSocket.CONNECTING) {
+        upstream.close();
+      }
+    });
+  });
+
+  const upgradeHandler = (req, socket, head) => {
+    const pathname = (() => {
+      try { return new URL(req.url || '/', 'http://127.0.0.1').pathname; } catch { return ''; }
+    })();
+    if (pathname !== PROXY_WS_PATH) return;
+    void ensureAuthenticated(req, null).then((authenticated) => {
+      if (!authenticated) {
+        rejectWebSocketUpgrade(socket, 401, 'Unauthorized');
+        return;
+      }
+      void originAllowed(req).then((allowed) => {
+        if (!allowed) {
+          rejectWebSocketUpgrade(socket, 403, 'Forbidden');
+          return;
+        }
+        wsServer.handleUpgrade(req, socket, head, (ws) => {
+          wsServer.emit('connection', ws, req);
+        });
+      }).catch(() => {
+        rejectWebSocketUpgrade(socket, 403, 'Forbidden');
+      });
+    }).catch(() => {
+      rejectWebSocketUpgrade(socket, 401, 'Unauthorized');
+    });
+  };
+
+  server.on('upgrade', upgradeHandler);
+  return {
+    stop: () => {
+      server.off('upgrade', upgradeHandler);
+      wsServer.close();
+    },
+  };
+};