import { WebSocket, WebSocketServer } from 'ws' import { validBrowserContext, validUnsafeRequest } from './oauth.mjs' import { equalSecret } from './sessions.mjs' const MAX_REQUEST_BYTES = 1024 * 1024 const MAX_EVENT_BYTES = 1024 * 1024 const API_TIMEOUT_MILLISECONDS = 30_000 export class ApiProxy { constructor(config, oauth, fetchImplementation = fetch) { this.config = config this.oauth = oauth this.fetch = fetchImplementation this.webSockets = new WebSocketServer({ noServer: true, maxPayload: MAX_EVENT_BYTES, }) } async route(request, response, url) { if (!url.pathname.startsWith('/bff/api/')) return false if (!['GET', 'POST'].includes(request.method)) { response.writeHead(405, { Allow: 'GET, POST' }) response.end() return true } const loaded = await this.oauth.authenticated(request) if (loaded === null) { response.writeHead(401, { 'Cache-Control': 'no-store' }) response.end() return true } if ( request.method !== 'GET' && !validUnsafeRequest(request, this.config, loaded.session.csrfToken) ) { requestEvent('csrf_rejected', { origin_matches: request.headers.origin === this.config.publicOrigin, fetch_site: request.headers['sec-fetch-site'] ?? 'absent', csrf_present: typeof request.headers['x-remora-csrf'] === 'string', csrf_matches: equalSecret( request.headers['x-remora-csrf'], loaded.session.csrfToken, ), }) response.writeHead(403, { 'Cache-Control': 'no-store' }) response.end() return true } const suffix = url.pathname.slice('/bff/api/'.length) if (suffix.startsWith('/')) { response.writeHead(400, { 'Cache-Control': 'no-store' }) response.end() return true } const upstream = new URL(suffix, this.config.apiUpstream) upstream.search = url.search if ( upstream.origin !== this.config.apiUpstream.origin || !upstream.pathname.startsWith(this.config.apiUpstream.pathname) ) { response.writeHead(400, { 'Cache-Control': 'no-store' }) response.end() return true } const headers = { Accept: request.headers.accept || 'application/json', Authorization: `Bearer ${loaded.session.accessToken}`, } if (request.method === 'POST' && suffix === 'event-tickets') headers.Origin = this.config.publicOrigin copyHeader(request, headers, 'content-type') copyHeader(request, headers, 'idempotency-key') copyHeader(request, headers, 'traceparent') let body try { body = request.method === 'GET' ? undefined : await requestBody(request) } catch (error) { if (!(error instanceof RequestBodyTooLarge)) throw error response.writeHead(413, { 'Cache-Control': 'no-store' }) response.end() return true } const upstreamResponse = await this.fetch(upstream, { method: request.method, headers, body, redirect: 'manual', signal: AbortSignal.timeout(API_TIMEOUT_MILLISECONDS), }) const responseHeaders = { 'Cache-Control': 'no-store' } for (const name of ['content-type', 'traceparent']) if (upstreamResponse.headers.has(name)) responseHeaders[name] = upstreamResponse.headers.get(name) response.writeHead(upstreamResponse.status, responseHeaders) response.end(Buffer.from(await upstreamResponse.arrayBuffer())) return true } async upgrade(request, socket, head) { const url = new URL(request.url || '/', this.config.publicOrigin) if (url.pathname !== '/bff/events') return false const ticket = url.searchParams.get('ticket') const cursor = url.searchParams.get('cursor') const loaded = await this.oauth.authenticated(request, false) if ( loaded === null || ticket === null || cursor === null || !validBrowserContext(request, this.config) ) { socketEvent('websocket_rejected') rejectUpgrade(socket, 401) return true } const target = new URL(this.config.eventsUpstream) target.search = new URLSearchParams({ ticket, cursor }).toString() this.webSockets.handleUpgrade(request, socket, head, (browser) => { const upstream = new WebSocket(target, { handshakeTimeout: 15_000, maxPayload: MAX_EVENT_BYTES, origin: this.config.publicOrigin, }) upstream.on('open', () => socketEvent('websocket_upstream_open')) upstream.on('unexpected-response', (_request, response) => socketEvent('websocket_upstream_rejected', response.statusCode), ) upstream.on('error', () => socketEvent('websocket_upstream_error')) relay(browser, upstream) }) return true } close() { for (const browser of this.webSockets.clients) { browser.close(1012, 'service restart') const timeout = setTimeout(() => { if (browser.readyState !== WebSocket.CLOSED) browser.terminate() }, 1_000) timeout.unref() } this.webSockets.close() } } function copyHeader(request, headers, name) { const value = request.headers[name] if (typeof value === 'string') headers[name] = value } async function requestBody(request) { const chunks = [] let size = 0 for await (const chunk of request) { size += chunk.length if (size > MAX_REQUEST_BYTES) throw new RequestBodyTooLarge() chunks.push(chunk) } return Buffer.concat(chunks) } class RequestBodyTooLarge extends Error {} function relay(browser, upstream) { upstream.on('open', () => undefined) browser.on('message', (data, binary) => { if (upstream.readyState === WebSocket.OPEN) upstream.send(data, { binary }) }) upstream.on('message', (data, binary) => { if (browser.readyState === WebSocket.OPEN) browser.send(data, { binary }) }) const closeBrowser = (code = 1011, reason = '') => { if (browser.readyState === WebSocket.OPEN) browser.close(code, reason) } const closeUpstream = (code = 1011, reason = '') => { if (upstream.readyState === WebSocket.OPEN) upstream.close(code, reason) else if (upstream.readyState === WebSocket.CONNECTING) upstream.terminate() } upstream.on('close', closeBrowser) browser.on('close', closeUpstream) upstream.on('error', () => closeBrowser()) browser.on('error', () => closeUpstream()) } function rejectUpgrade(socket, status) { socket.end(`HTTP/1.1 ${status} Unauthorized\r\nConnection: close\r\n\r\n`) } function socketEvent(event, status) { requestEvent(event, status === undefined ? {} : { status }) } function requestEvent(event, details) { process.stdout.write( `${JSON.stringify({ timestamp: new Date().toISOString(), level: event.endsWith('open') ? 'info' : 'warn', component: 'remora-web', event, ...details, })}\n`, ) }