diff --git a/config.default.ts b/config.default.ts index 31ad4bc5..bf5f98b9 100644 --- a/config.default.ts +++ b/config.default.ts @@ -233,6 +233,7 @@ export const config: NodelinkConfig = { baseCapacity: 500, refillRatePerSecond: 100, maxConcurrentSockets: 25, + maxProxySockets: 1024, ipv6SubnetMask: 64, blockScoreThreshold: 30, blockDurationMs: 300000, diff --git a/dist/config.default.js b/dist/config.default.js index c8b50dbf..794c106c 100644 --- a/dist/config.default.js +++ b/dist/config.default.js @@ -225,6 +225,7 @@ export const config = { baseCapacity: 500, refillRatePerSecond: 100, maxConcurrentSockets: 25, + maxProxySockets: 1024, ipv6SubnetMask: 64, blockScoreThreshold: 30, blockDurationMs: 300000, diff --git a/dist/src/api/index.js b/dist/src/api/index.js index 83e31be7..22a4cea6 100644 --- a/dist/src/api/index.js +++ b/dist/src/api/index.js @@ -10,6 +10,7 @@ import fs from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { PATH_VERSION } from '../constants.js'; +import { isLoopbackRequest } from '../utils/clientAddress.js'; import { logger, sendErrorResponse, sendResponse, verifyMethod } from '../utils.js'; const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); @@ -171,10 +172,11 @@ async function requestHandler(nodelink, req, res) { } nodelink.statsManager.incrementApiRequest(parsedUrl.pathname); const trace = parsedUrl.searchParams.get('trace') === 'true'; - const remoteAddress = req.socket?.remoteAddress ?? 'unknown'; - const remotePort = req.socket?.remotePort; - const isInternal = ['127.0.0.1', '::1', 'localhost'].includes(remoteAddress); - const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}:${remotePort ?? 'unknown'})`; + const remoteAddress = nodelink.admissionManager.resolveClientAddress(req) ?? 'unknown'; + const isProxied = nodelink.admissionManager.isTrustedProxy(req.socket?.remoteAddress); + const remotePort = isProxied ? undefined : req.socket?.remotePort; + const isInternal = isLoopbackRequest(req.socket?.remoteAddress, req.headers); + const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}${remotePort ? `:${remotePort}` : ''})`; const requestId = `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; const originalEnd = res.end.bind(res); res.end = (...args) => { diff --git a/dist/src/api/profiler.file.js b/dist/src/api/profiler.file.js index 69d9788a..abab6340 100644 --- a/dist/src/api/profiler.file.js +++ b/dist/src/api/profiler.file.js @@ -1,12 +1,8 @@ import fsPromises from 'node:fs/promises'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; +import { isLoopbackRequest } from '../utils/clientAddress.js'; import { sendErrorResponse, sendResponse } from '../utils.js'; -/** - * Loopback addresses allowed to access the profiler endpoints when external - * access is disabled. - */ -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']); /** * Returns whether the provided body value is a plain object record. * @@ -210,8 +206,8 @@ async function handler(nodelink, req, res, _sendResponse, parsedUrl) { sendErrorResponse(req, res, 403, 'Forbidden', 'Profiler endpoint is disabled.', parsedUrl.pathname); return; } - const remoteAddress = req.socket?.remoteAddress ?? ''; - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers); + if (!endpointConfig.allowExternalPatch && !isLocal) { sendErrorResponse(req, res, 403, 'Forbidden', 'External access to profiler file endpoint is blocked.', parsedUrl.pathname); return; } diff --git a/dist/src/api/profiler.js b/dist/src/api/profiler.js index 55161364..bb0326ab 100644 --- a/dist/src/api/profiler.js +++ b/dist/src/api/profiler.js @@ -3,8 +3,8 @@ import fsPromises from 'node:fs/promises'; import inspector from 'node:inspector'; import os from 'node:os'; import v8 from 'node:v8'; +import { isLoopbackRequest } from '../utils/clientAddress.js'; import { sendErrorResponse, sendResponse } from '../utils.js'; -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']); const { NODELINK_PROFILER_DIR: profilerDirectoryEnv } = process.env; const profilerBaseDir = profilerDirectoryEnv || '.profiles'; let activeMasterCpu = null; @@ -190,8 +190,8 @@ function validateAccess(nodelink, req, suppliedCode) { if (!endpointConfig.patchEnabled) { return { ok: false, error: 'Profiler endpoint is disabled.' }; } - const remoteAddress = req.socket?.remoteAddress || ''; - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers); + if (!endpointConfig.allowExternalPatch && !isLocal) { return { ok: false, error: 'External access to profiler endpoint is blocked.' diff --git a/dist/src/api/profiler.ui.js b/dist/src/api/profiler.ui.js index 0e8abbe9..283bd267 100644 --- a/dist/src/api/profiler.ui.js +++ b/dist/src/api/profiler.ui.js @@ -1,8 +1,8 @@ import fs from 'node:fs'; import path from 'node:path'; import { fileURLToPath } from 'node:url'; +import { isLoopbackRequest } from '../utils/clientAddress.js'; import { sendErrorResponse } from '../utils.js'; -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']); const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); /** @@ -2366,8 +2366,8 @@ async function handler(nodelink, req, res, _sendResponse, parsedUrl) { if (!endpointConfig.patchEnabled) { return sendErrorResponse(req, res, 403, 'Forbidden', 'Profiler endpoint is disabled.', parsedUrl.pathname); } - const remoteAddress = req.socket?.remoteAddress || ''; - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers); + if (!endpointConfig.allowExternalPatch && !isLocal) { return sendErrorResponse(req, res, 403, 'Forbidden', 'External access to profiler UI is blocked.', parsedUrl.pathname); } const code = parsedUrl.searchParams.get('code'); diff --git a/dist/src/api/workers.js b/dist/src/api/workers.js index 1c42917e..6924d5b7 100644 --- a/dist/src/api/workers.js +++ b/dist/src/api/workers.js @@ -1,9 +1,5 @@ +import { isLoopbackRequest } from '../utils/clientAddress.js'; import { sendErrorResponse, sendResponse } from '../utils.js'; -/** - * Loopback addresses allowed to access the workers patch endpoint when - * external patching is disabled. - */ -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']); /** * Creates a strongly typed runtime view for the workers endpoint. * @@ -151,8 +147,8 @@ function handlePatch(nodelink, req, res, parsedUrl) { sendErrorResponse(req, res, 403, 'Forbidden', 'Workers patch endpoint is disabled.', parsedUrl.pathname); return; } - const remoteAddress = req.socket?.remoteAddress ?? ''; - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers); + if (!endpointConfig.allowExternalPatch && !isLocal) { sendErrorResponse(req, res, 403, 'Forbidden', 'External access to the workers patch endpoint is blocked.', parsedUrl.pathname); return; } diff --git a/dist/src/managers/admissionManager.js b/dist/src/managers/admissionManager.js index c33e5753..a5002053 100644 --- a/dist/src/managers/admissionManager.js +++ b/dist/src/managers/admissionManager.js @@ -1,3 +1,4 @@ +import { normalizeAddress, resolveClientAddress, TrustedProxyList } from '../utils/clientAddress.js'; import { logger } from '../utils.js'; const DEFAULT_CONFIG = { enabled: true, @@ -38,6 +39,7 @@ const DEFAULT_CONFIG = { baseCapacity: 500, refillRatePerSecond: 100, maxConcurrentSockets: 25, + maxProxySockets: 1024, ipv6SubnetMask: 64, blockScoreThreshold: 30, blockDurationMs: 300000, @@ -90,6 +92,8 @@ export default class AdmissionManager { guildStates; sessionStates; ipStates; + proxySockets; + trustedProxies; globalState; cleanupInterval; constructor(nodelink, config) { @@ -98,6 +102,8 @@ export default class AdmissionManager { this.guildStates = new Map(); this.sessionStates = new Map(); this.ipStates = new Map(); + this.proxySockets = new Map(); + this.trustedProxies = this._buildTrustedProxies(); const now = performance.now(); this.globalState = { tokens: 300, @@ -207,6 +213,92 @@ export default class AdmissionManager { } return this._buildAllowedDecision('ip', 100, 100); } + /** + * Resolves the originating client address, honoring forwarding headers only + * from trusted proxies. Unlike admission keys, the result is not subnet-masked. + * @param req - Raw or shimmed API request. + */ + resolveClientAddress(req) { + return resolveClientAddress(req.socket?.remoteAddress, req.headers, this.trustedProxies); + } + /** + * Checks whether a TCP peer is a configured trusted proxy. + * @param rawAddress - Peer address. + */ + isTrustedProxy(rawAddress) { + return this.trustedProxies.contains(rawAddress); + } + /** + * Admits a new TCP connection. Trusted proxies share one aggregate pool, since + * per-client bans and limits are enforced per request once headers are known. + * Other peers are checked against their own block state and socket pool. + * @param rawAddress - Peer address. + * @returns Whether the connection may proceed; call releaseConnection on close. + */ + admitConnection(rawAddress) { + if (!this.trustedProxies.contains(rawAddress)) { + if (this.isIpBlocked(rawAddress)) + return false; + return this.incrementActiveSockets(rawAddress); + } + const proxy = normalizeAddress(rawAddress); + const active = this.proxySockets.get(proxy) ?? 0; + const maxSockets = this.config.ip.maxProxySockets; + if (active >= maxSockets) { + logger('warn', 'AdmissionManager', `Trusted proxy ${proxy} exceeded aggregate socket pool (${active}/${maxSockets}). Dropping connection.`); + return false; + } + this.proxySockets.set(proxy, active + 1); + return true; + } + /** + * Releases a connection admitted by admitConnection. + * @param rawAddress - Peer address. + */ + releaseConnection(rawAddress) { + if (!this.trustedProxies.contains(rawAddress)) { + this.decrementActiveSockets(rawAddress); + return; + } + const proxy = normalizeAddress(rawAddress); + const next = (this.proxySockets.get(proxy) ?? 0) - 1; + if (next > 0) { + this.proxySockets.set(proxy, next); + } + else { + this.proxySockets.delete(proxy); + } + } + /** + * Reserves socket capacity for a WebSocket upgrade. Proxied clients share the + * proxy's TCP pool, so their own per-client pool is charged here. + * @param peerAddress - TCP peer address. + * @param clientKey - Resolved client admission key (AdmissionContext.ip). + * @param includePeer - Also admit the peer connection itself, for runtimes + * without a TCP-level connection hook (Bun). + * @returns An idempotent release callback, or null if capacity is exhausted. + */ + reserveUpgrade(peerAddress, clientKey, includePeer) { + const releases = []; + const release = () => { + while (releases.length > 0) { + releases.pop()?.(); + } + }; + if (includePeer) { + if (!this.admitConnection(peerAddress)) + return null; + releases.push(() => this.releaseConnection(peerAddress)); + } + if (clientKey && this.trustedProxies.contains(peerAddress)) { + if (!this.incrementActiveSockets(clientKey)) { + release(); + return null; + } + releases.push(() => this.decrementActiveSockets(clientKey)); + } + return release; + } /** * Tracks an incoming TCP socket. Returns false if IP connection pool is exhausted. * @param rawAddress - Remote IP address. @@ -878,26 +970,32 @@ export default class AdmissionManager { return match?.[1] ?? null; } /** - * Resolves remote IP with edge proxy header precedence. + * Resolves the admission key for a request's client address. * @internal */ _resolveIp(req) { - const socketAddress = req.socket?.remoteAddress; - const trustProxyEnabled = this.config.trustProxy === true; - if (!trustProxyEnabled) { - return this._normalizeIp(socketAddress); + return this._normalizeIp(this.resolveClientAddress(req)); + } + /** + * Builds the trusted proxy matcher and reports misconfiguration. + * @internal + */ + _buildTrustedProxies() { + const { trustProxy, trustedProxies } = this.config; + if (!trustProxy) { + if (trustedProxies.length > 0) { + logger('warn', 'AdmissionManager', 'admission.trustedProxies is set but admission.trustProxy is false; forwarding headers are ignored.'); + } + return new TrustedProxyList(); + } + const list = new TrustedProxyList(trustedProxies); + for (const entry of list.invalidEntries) { + logger('warn', 'AdmissionManager', `Ignoring invalid admission.trustedProxies entry: ${entry}`); + } + if (list.size === 0) { + logger('warn', 'AdmissionManager', 'admission.trustProxy is enabled but admission.trustedProxies has no valid entries; forwarding headers are ignored.'); } - const headers = req.headers; - const cfConnectingIp = this._getHeader(headers, 'cf-connecting-ip'); - const trueClientIp = this._getHeader(headers, 'true-client-ip'); - const xRealIp = this._getHeader(headers, 'x-real-ip'); - const forwardedFor = this._getHeader(headers, 'x-forwarded-for'); - const candidate = cfConnectingIp ?? - trueClientIp ?? - xRealIp ?? - forwardedFor?.split(',')?.[0]?.trim() ?? - socketAddress; - return this._normalizeIp(candidate); + return list; } /** * Normalizes IP and applies IPv6 /64 subnet mask. diff --git a/dist/src/server/bunServer.js b/dist/src/server/bunServer.js index 606ab3ba..ddabf608 100644 --- a/dist/src/server/bunServer.js +++ b/dist/src/server/bunServer.js @@ -1,4 +1,5 @@ import { EventEmitter } from 'node:events'; +import { isLoopbackRequest } from '../utils/clientAddress.js'; import { logger, parseClient, verifyDiscordID } from '../utils.js'; const VOICE_PATH_RE = /^\/v4\/websocket\/voice\/([A-Za-z0-9]+)\/?$/; const LIVE_PATH_RE = /^\/v4\/websocket\/youtube\/live\/([^/]+)\/?$/; @@ -134,7 +135,7 @@ export function createBunServer(context, getRequestHandler) { : url.pathname; if (pathname === '/v4/profiler/socket') { const remoteAddress = server.requestIP(req)?.address || 'unknown'; - const isInternal = /^(::1|localhost|127\.0\.0\.1)/.test(remoteAddress); + const isInternal = isLoopbackRequest(remoteAddress, Object.fromEntries(req.headers)); const endpoint = context.options.cluster?.endpoint || {}; const patchEnabled = endpoint.patchEnabled === true; const allowExternalPatch = endpoint.allowExternalPatch === true; @@ -182,7 +183,15 @@ export function createBunServer(context, getRequestHandler) { const liveMatch = pathname.match(LIVE_PATH_RE); const isMainWs = pathname === '/v4/websocket'; if (isMainWs || voiceMatch || liveMatch) { - const remoteAddress = server.requestIP(req)?.address || 'unknown'; + const peerAddress = server.requestIP(req)?.address || 'unknown'; + const upgradeReqShim = { + method: req.method, + url: req.url, + headers: Object.fromEntries(req.headers), + socket: { remoteAddress: peerAddress } + }; + const remoteAddress = context.admissionManager.resolveClientAddress(upgradeReqShim) ?? + peerAddress; const clientAddress = `[External] (${remoteAddress})`; const isIpBlocked = context.admissionManager.isIpBlocked(remoteAddress); if (isIpBlocked) { @@ -191,12 +200,6 @@ export function createBunServer(context, getRequestHandler) { statusText: 'Forbidden' }); } - const upgradeReqShim = { - method: req.method, - url: req.url, - headers: Object.fromEntries(req.headers), - socket: { remoteAddress } - }; const admissionContext = context.admissionManager.resolveContext(upgradeReqShim, url); const admissionDecision = context.admissionManager.admit(admissionContext); if (!admissionDecision.allowed) { @@ -284,6 +287,18 @@ export function createBunServer(context, getRequestHandler) { logger('warn', 'Server', `Session-ID provided by ${clientAddress} does not exist or is not resumable: ${sessionId}, creating a new session`); sessionId = null; } + /* INFO: Bun has no TCP connection hook, so peer and per-client capacity are reserved per upgrade */ + const releaseCapacity = context.admissionManager.reserveUpgrade(peerAddress, admissionContext.ip, true); + if (!releaseCapacity) { + return new Response('Too many concurrent connections.', { + status: 429, + statusText: 'Too Many Requests', + headers: { + 'Nodelink-Api-Version': '4', + IamNodelink: 'true' + } + }); + } const success = server.upgrade(req, { data: { clientInfo, @@ -293,11 +308,13 @@ export function createBunServer(context, getRequestHandler) { url: req.url, pathname, eventName, - routeId + routeId, + releaseCapacity } }); if (success) return undefined; + releaseCapacity(); return new Response('WebSocket upgrade failed', { status: 400, headers: { @@ -475,6 +492,7 @@ export function createBunServer(context, getRequestHandler) { wrapper._handleMessage(message); }, close(ws, code, reason) { + ws.data?.releaseCapacity?.(); const wrapper = ws.data?.wrapper; if (!wrapper) { logger('debug', 'WebSocket', `Bun close received without wrapper (code: ${code}, remote: ${ws.data?.remoteAddress || 'unknown'})`); diff --git a/dist/src/server/bunServer.test.js b/dist/src/server/bunServer.test.js new file mode 100644 index 00000000..e9887365 --- /dev/null +++ b/dist/src/server/bunServer.test.js @@ -0,0 +1,112 @@ +import assert from 'node:assert/strict'; +import { EventEmitter } from 'node:events'; +import test from 'node:test'; +import AdmissionManager from '../managers/admissionManager.js'; +import { createBunServer } from './bunServer.js'; +/* INFO: Bun-only regression; run with `bun test src/server/bunServer.test.ts` */ +const skip = typeof Bun === 'undefined' ? 'requires the Bun runtime' : false; +const PASSWORD = 'test-password'; +async function startBunServer(config) { + const options = { + server: { host: '127.0.0.1', port: 0, password: PASSWORD }, + playback: { voiceReceive: { enabled: false } }, + cluster: {} + }; + const admission = new AdmissionManager({ options }, config); + const socketBus = new EventEmitter(); + const context = { + options, + admissionManager: admission, + socket: socketBus, + sessions: { isResumable: () => false } + }; + const server = createBunServer(context, () => Promise.reject(new Error('REST is not used in this test'))); + return { + url: `ws://127.0.0.1:${server.port}/v4/websocket`, + socketBus, + close: async () => { + admission.destroy(); + await server.stop(true); + } + }; +} +function connect(url, forwardedFor) { + const headers = { + Authorization: PASSWORD, + 'Client-Name': 'bun-release-test/1.0.0', + 'User-Id': '123456789012345678' + }; + if (forwardedFor) + headers['X-Forwarded-For'] = forwardedFor; + return new Promise((resolve) => { + const ws = new WebSocket(url, { headers }); + ws.addEventListener('open', () => resolve(ws), { once: true }); + ws.addEventListener('error', () => resolve(null), { once: true }); + }); +} +async function closeAll(sockets) { + await Promise.all(sockets.map((ws) => new Promise((resolve) => { + if (!ws || ws.readyState === WebSocket.CLOSED) + return resolve(null); + ws.addEventListener('close', resolve, { once: true }); + ws.close(1000, 'done'); + }))); + /* INFO: The server-side close handler releases capacity after the client sees the close */ + await new Promise((resolve) => setTimeout(resolve, 50)); +} +test('Bun caps direct peers and releases slots on close', { + skip +}, async (t) => { + const harness = await startBunServer({ + ip: { maxConcurrentSockets: 2 } + }); + t.after(() => harness.close()); + for (let round = 0; round < 3; round++) { + const open = [await connect(harness.url), await connect(harness.url)]; + assert.ok(open[0] && open[1], `round ${round}: both connections open`); + assert.equal(await connect(harness.url), null, 'third is over the limit'); + await closeAll(open); + } +}); +test('Bun caps proxied clients and the proxy pool, releasing both', { + skip +}, async (t) => { + const harness = await startBunServer({ + trustProxy: true, + trustedProxies: ['127.0.0.1'], + ip: { + maxConcurrentSockets: 2, + maxProxySockets: 3 + } + }); + t.after(() => harness.close()); + for (let round = 0; round < 3; round++) { + const a1 = await connect(harness.url, '198.51.100.1'); + const a2 = await connect(harness.url, '198.51.100.1'); + assert.ok(a1 && a2, `round ${round}: client A gets two sockets`); + assert.equal(await connect(harness.url, '198.51.100.1'), null); + const b1 = await connect(harness.url, '198.51.100.2'); + assert.ok(b1, 'client B is unaffected by client A'); + assert.equal(await connect(harness.url, '198.51.100.3'), null, 'proxy aggregate pool is full'); + await closeAll([a1, a2, b1]); + } +}); +test('Bun releases slots when the server closes the socket', { + skip +}, async (t) => { + const harness = await startBunServer({ + ip: { maxConcurrentSockets: 1 } + }); + t.after(() => harness.close()); + harness.socketBus.on('/v4/websocket', (wrapper) => { + setTimeout(() => wrapper.close(), 10); + }); + for (let round = 0; round < 3; round++) { + const ws = await connect(harness.url); + assert.ok(ws, `round ${round}: slot available after server close`); + await new Promise((resolve) => ws.readyState === WebSocket.CLOSED + ? resolve(null) + : ws.addEventListener('close', resolve, { once: true })); + await new Promise((resolve) => setTimeout(resolve, 50)); + } +}); diff --git a/dist/src/server/clientIsolation.test.js b/dist/src/server/clientIsolation.test.js new file mode 100644 index 00000000..9be631cc --- /dev/null +++ b/dist/src/server/clientIsolation.test.js @@ -0,0 +1,166 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import WebSocketServer from '@performanc/pwsl-server'; +import requestHandler from '../api/index.js'; +import AdmissionManager from '../managers/admissionManager.js'; +import { createHttpServer } from './httpServer.js'; +const PASSWORD = 'test-password'; +async function startServer(config) { + const options = { + server: { password: PASSWORD }, + playback: { voiceReceive: { enabled: false } }, + cluster: {}, + api: {} + }; + const admission = new AdmissionManager({ options }, config); + const socketBus = new WebSocketServer(); + const context = { + options, + admissionManager: admission, + socket: socketBus, + sessions: { isResumable: () => false }, + pluginManager: { callHook: () => { } }, + statsManager: { + incrementApiRequest: () => { }, + recordHttpRequestDuration: () => { } + }, + extensions: { middlewares: [], routes: [] } + }; + const server = createHttpServer(context, () => Promise.resolve(requestHandler)); + /* INFO: Upgraded sockets are not closed by closeAllConnections, so a failing test would hang teardown */ + const sockets = new Set(); + server.on('connection', (socket) => sockets.add(socket)); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const { port } = server.address(); + return { + url: `ws://127.0.0.1:${port}/v4/websocket`, + restUrl: `http://127.0.0.1:${port}/v4/nodelink-test-route`, + socketBus, + close: async () => { + admission.destroy(); + for (const socket of sockets) + socket.destroy(); + await new Promise((resolve) => server.close(() => resolve())); + } + }; +} +function connect(url, forwardedFor, password = PASSWORD) { + const headers = { + Authorization: password, + 'Client-Name': 'release-test/1.0.0', + 'User-Id': '123456789012345678' + }; + if (forwardedFor) + headers['X-Forwarded-For'] = forwardedFor; + return new Promise((resolve) => { + const ws = new WebSocket(url, { headers }); + ws.addEventListener('open', () => resolve(ws), { once: true }); + ws.addEventListener('error', () => resolve(null), { once: true }); + }); +} +async function closeClient(ws) { + if (ws.readyState === WebSocket.CLOSED) + return; + const closed = new Promise((resolve) => ws.addEventListener('close', resolve, { once: true })); + ws.close(1000, 'done'); + await closed; +} +/* INFO: The server releases on its own close handling, which can trail the client's close event */ +async function connectEventually(url, forwardedFor) { + for (let attempt = 0; attempt < 20; attempt++) { + const ws = await connect(url, forwardedFor); + if (ws) + return ws; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + return null; +} +test('direct peers get their socket slots back after normal closes', async (t) => { + const harness = await startServer({ + ip: { maxConcurrentSockets: 2 } + }); + t.after(() => harness.close()); + for (let round = 0; round < 3; round++) { + const first = await connectEventually(harness.url); + const second = await connectEventually(harness.url); + assert.ok(first && second, `round ${round}: both connections open`); + assert.equal(await connect(harness.url), null, 'third is over the limit'); + await closeClient(first); + await closeClient(second); + } +}); +test('proxied clients release both per-client and proxy pools', async (t) => { + const harness = await startServer({ + trustProxy: true, + trustedProxies: ['127.0.0.1'], + ip: { + maxConcurrentSockets: 2, + maxProxySockets: 3 + } + }); + t.after(() => harness.close()); + for (let round = 0; round < 3; round++) { + const a1 = await connectEventually(harness.url, '198.51.100.1'); + const a2 = await connectEventually(harness.url, '198.51.100.1'); + assert.ok(a1 && a2, `round ${round}: client A gets two sockets`); + assert.equal(await connect(harness.url, '198.51.100.1'), null, 'client A is capped per client'); + const b1 = await connectEventually(harness.url, '198.51.100.2'); + assert.ok(b1, 'client B is unaffected by client A'); + assert.equal(await connect(harness.url, '198.51.100.3'), null, 'proxy aggregate pool is full'); + await closeClient(a1); + await closeClient(a2); + await closeClient(b1); + } +}); +test('server-side destroy releases the slot', async (t) => { + const harness = await startServer({ + ip: { maxConcurrentSockets: 1 } + }); + t.after(() => harness.close()); + harness.socketBus.on('/v4/websocket', (ws) => { + setTimeout(() => ws.destroy(), 10); + }); + for (let round = 0; round < 3; round++) { + const ws = await connectEventually(harness.url); + assert.ok(ws, `round ${round}: slot available after server destroy`); + await new Promise((resolve) => ws.readyState === WebSocket.CLOSED + ? resolve(null) + : ws.addEventListener('close', resolve, { once: true })); + } +}); +async function restStatus(url, forwardedFor, password = PASSWORD) { + const response = await fetch(url, { + headers: { Authorization: password, 'X-Forwarded-For': forwardedFor } + }); + await response.body?.cancel(); + return response.status; +} +const proxyConfig = { + trustProxy: true, + trustedProxies: ['127.0.0.1'] +}; +test('REST auth failures behind a proxy ban only the offending client', async (t) => { + const harness = await startServer(proxyConfig); + t.after(() => harness.close()); + for (let attempt = 0; attempt < 5; attempt++) { + assert.equal(await restStatus(harness.restUrl, '198.51.100.66', 'wrong'), 401); + } + assert.equal(await restStatus(harness.restUrl, '198.51.100.66'), 403); + assert.equal(await restStatus(harness.restUrl, '6.6.6.6, 198.51.100.66'), 403, 'a spoofed leftmost hop does not lift the ban'); + assert.equal(await restStatus(harness.restUrl, '198.51.100.10'), 404); + const bystander = await connectEventually(harness.url, '198.51.100.10'); + assert.ok(bystander, 'other clients behind the proxy still connect'); + await closeClient(bystander); +}); +test('WebSocket auth failures behind a proxy ban only the offending client', async (t) => { + const harness = await startServer(proxyConfig); + t.after(() => harness.close()); + for (let attempt = 0; attempt < 5; attempt++) { + assert.equal(await connect(harness.url, '198.51.100.66', 'wrong'), null); + } + assert.equal(await connect(harness.url, '198.51.100.66'), null); + assert.equal(await restStatus(harness.restUrl, '198.51.100.66'), 403); + const bystander = await connectEventually(harness.url, '198.51.100.10'); + assert.ok(bystander, 'other clients behind the proxy still connect'); + await closeClient(bystander); +}); diff --git a/dist/src/server/httpServer.js b/dist/src/server/httpServer.js index f52b2b82..a1342602 100644 --- a/dist/src/server/httpServer.js +++ b/dist/src/server/httpServer.js @@ -1,6 +1,7 @@ import http from 'node:http'; import process from 'node:process'; import { logger } from '../utils.js'; +import { trackSocketRelease } from './socketRelease.js'; import { handleHttpUpgrade } from './wsRouter.js'; /* INFO: Creates and configures native Node.js HTTP server with socket pool guards, DoS defense, and upgrade routing */ function createHttpServer(nodelink, getRequestHandler) { @@ -25,19 +26,12 @@ function createHttpServer(nodelink, getRequestHandler) { /* INFO: Guard all incoming sockets against DoS blocks, connection floods, and reset errors */ server.on('connection', (socket) => { const remoteAddress = socket.remoteAddress; - const isIpBlocked = nodelink.admissionManager.isIpBlocked(remoteAddress); - if (isIpBlocked) { + const connectionAllowed = nodelink.admissionManager.admitConnection(remoteAddress); + if (!connectionAllowed) { socket.destroy(); return; } - const socketAllowed = nodelink.admissionManager.incrementActiveSockets(remoteAddress); - if (!socketAllowed) { - socket.destroy(); - return; - } - socket.on('close', () => { - nodelink.admissionManager.decrementActiveSockets(remoteAddress); - }); + trackSocketRelease(socket, () => nodelink.admissionManager.releaseConnection(remoteAddress)); socket.on('error', (err) => { const isBenign = err?.code === 'EPIPE' || err?.code === 'ECONNRESET'; if (isBenign) diff --git a/dist/src/server/socketRelease.js b/dist/src/server/socketRelease.js new file mode 100644 index 00000000..9b16f138 --- /dev/null +++ b/dist/src/server/socketRelease.js @@ -0,0 +1,51 @@ +const pendingReleases = new WeakMap(); +/** + * Registers a capacity release to run once when the socket goes away. + * The TCP 'close' event is only a fallback: PWSL strips every socket + * listener when a WebSocket closes, so upgraded sockets must also be + * bound with bindWebSocketRelease. + * @param socket - TCP socket holding the capacity. + * @param release - Callback that returns the capacity. + * @public + */ +export function trackSocketRelease(socket, release) { + const releases = pendingReleases.get(socket); + if (releases) { + releases.push(release); + return; + } + pendingReleases.set(socket, [release]); + socket.once('close', () => releaseSocket(socket)); +} +/** + * Runs and clears every pending release for the socket. Idempotent. + * @param socket - TCP socket holding the capacity. + * @internal + */ +function releaseSocket(socket) { + const releases = pendingReleases.get(socket); + if (!releases) + return; + pendingReleases.delete(socket); + for (const release of releases) { + release(); + } +} +/** + * Releases the socket's capacity when its WebSocket ends. PWSL emits 'close' + * for peer-initiated and transport closes, but its destroy() (used for + * protocol errors and server-side teardown) does not, so it is wrapped too. + * @param ws - WebSocket connection created by the upgrade. + * @param socket - Underlying TCP socket. + * @public + */ +export function bindWebSocketRelease(ws, socket) { + ws.once('close', () => releaseSocket(socket)); + const destroy = ws.destroy; + if (typeof destroy !== 'function') + return; + ws.destroy = function () { + releaseSocket(socket); + return destroy.call(this); + }; +} diff --git a/dist/src/server/wsRouter.js b/dist/src/server/wsRouter.js index e16b043d..4aff3559 100644 --- a/dist/src/server/wsRouter.js +++ b/dist/src/server/wsRouter.js @@ -2,17 +2,13 @@ import { Buffer } from 'node:buffer'; import crypto from 'node:crypto'; import { URL } from 'node:url'; import { attachProfilerSocket } from '../api/profiler.socket.js'; +import { isLoopbackRequest } from '../utils/clientAddress.js'; import { decodeTrack, logger, parseClient, verifyDiscordID } from '../utils.js'; +import { bindWebSocketRelease, trackSocketRelease } from './socketRelease.js'; import { handleClientWebSocket } from './wsSession.js'; const VOICE_PATH_RE = /^\/v4\/websocket\/voice\/([A-Za-z0-9]+)\/?$/; const LIVE_PATH_RE = /^\/v4\/websocket\/youtube\/live\/([^/]+)\/?$/; const DISCORD_SNOWFLAKE_RE = /^\d{17,20}$/; -const INTERNAL_IPS = new Set([ - '127.0.0.1', - '::1', - '::ffff:127.0.0.1', - 'localhost' -]); function _getHeader(headers, name) { const value = headers[name]; return Array.isArray(value) ? value[0] : value; @@ -51,18 +47,27 @@ function handleHttpUpgrade(context, request, socket, head) { return; logger('debug', 'Server', `Upgrade socket error: ${err.message}`); }); - const remoteAddress = request.socket.remoteAddress || 'unknown'; - const remotePort = request.socket.remotePort || 0; - const isInternal = INTERNAL_IPS.has(remoteAddress); - const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}:${remotePort})`; + const admission = context.admissionManager; + const peerAddress = request.socket.remoteAddress; + const remoteAddress = admission.resolveClientAddress(request) ?? 'unknown'; + const isProxied = admission.isTrustedProxy(peerAddress); + const remotePort = isProxied ? 0 : request.socket.remotePort || 0; + const isInternal = isLoopbackRequest(peerAddress, request.headers); + const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}${remotePort ? `:${remotePort}` : ''})`; const url = new URL(request.url || '/', 'http://localhost'); const pathname = url.pathname; - const admissionContext = context.admissionManager.resolveContext(request, url); - const admissionDecision = context.admissionManager.admit(admissionContext); + const admissionContext = admission.resolveContext(request, url); + const admissionDecision = admission.admit(admissionContext); if (!admissionDecision.allowed) { _rejectUpgrade(socket, admissionDecision.status, 'Too Many Requests', admissionDecision.message); return; } + const releaseUpgrade = admission.reserveUpgrade(peerAddress, admissionContext.ip, false); + if (!releaseUpgrade) { + _rejectUpgrade(socket, 429, 'Too Many Requests', 'Too many concurrent connections.'); + return; + } + trackSocketRelease(socket, releaseUpgrade); if (pathname === '/v4/profiler/socket') { _handleProfilerUpgrade(context, request, socket, head, clientAddress, isInternal); return; @@ -96,6 +101,7 @@ function _handleProfilerUpgrade(context, request, socket, head, clientAddress, i logger('info', 'ProfilerSocket', `Profiler socket connected from ${clientAddress} | URL: ${request.url}`); const wsServer = context.socket; wsServer?.handleUpgrade(request, socket, head, null, (ws) => { + bindWebSocketRelease(ws, socket); context.socket?.emit('/v4/profiler/socket', ws, request, { name: 'ProfilerUI', version: '1' }, null, null); }); } @@ -107,7 +113,7 @@ function _handleGatewayUpgrade(context, request, socket, head, pathname, clientA const authHeader = _getHeader(request.headers, 'authorization'); const isAuthorized = _isAuthorized(authHeader, context.options.server?.password ?? ''); if (!isAuthorized) { - context.admissionManager.recordAuthFailure(request.socket.remoteAddress); + context.admissionManager.recordAuthFailure(context.admissionManager.resolveClientAddress(request)); reject(401, 'Unauthorized', 'Invalid password provided.'); return; } @@ -152,6 +158,7 @@ function _handleGatewayUpgrade(context, request, socket, head, pathname, clientA } const wsServer = context.socket; wsServer?.handleUpgrade(request, socket, head, null, (ws) => { + bindWebSocketRelease(ws, socket); context.socket?.emit(eventName, ws, request, clientInfo, sessionId, routeId); }); } diff --git a/dist/src/utils/clientAddress.js b/dist/src/utils/clientAddress.js new file mode 100644 index 00000000..42ea16dd --- /dev/null +++ b/dist/src/utils/clientAddress.js @@ -0,0 +1,150 @@ +import net from 'node:net'; +/** + * Matches peer addresses against an explicit list of trusted proxy IPs/CIDRs. + * IPv4 and IPv6 rules are kept in separate lists so an address only matches + * rules of its own family. Invalid entries are reported through + * `invalidEntries` and never match. + * @public + */ +export class TrustedProxyList { + ipv4 = new net.BlockList(); + ipv6 = new net.BlockList(); + invalidEntries = []; + size = 0; + constructor(entries = []) { + for (const entry of entries) { + if (this.add(entry)) { + this.size += 1; + } + else { + this.invalidEntries.push(entry); + } + } + } + /** + * Checks whether an address belongs to a trusted proxy. + * @param rawAddress - Peer address (IPv4, IPv6 or IPv4-mapped IPv6). + */ + contains(rawAddress) { + if (this.size === 0) + return false; + const address = normalizeAddress(rawAddress); + if (!address) + return false; + return net.isIP(address) === 4 + ? this.ipv4.check(address, 'ipv4') + : this.ipv6.check(address, 'ipv6'); + } + add(entry) { + const [rawAddress, rawPrefix, ...rest] = entry.trim().split('/'); + if (rest.length > 0) + return false; + const address = normalizeAddress(rawAddress); + if (!address) + return false; + const isIpv4 = net.isIP(address) === 4; + const bits = isIpv4 ? 32 : 128; + const prefix = rawPrefix === undefined ? bits : Number(rawPrefix); + const isValidPrefix = rawPrefix === undefined || + (/^\d+$/.test(rawPrefix) && prefix >= 0 && prefix <= bits); + if (!isValidPrefix) + return false; + if (isIpv4) { + this.ipv4.addSubnet(address, prefix, 'ipv4'); + } + else { + this.ipv6.addSubnet(address, prefix, 'ipv6'); + } + return true; + } +} +/** + * Strips IPv4-mapped prefixes, brackets and ports from an address. + * Returns null when the result is not a valid IP literal. + * @public + */ +export function normalizeAddress(rawAddress) { + if (!rawAddress) + return null; + let address = rawAddress.trim(); + if (!address) + return null; + const bracketed = address.match(/^\[([^\]]+)\](?::\d+)?$/); + if (bracketed?.[1]) { + address = bracketed[1]; + } + else if (/^[\d.]+:\d+$/.test(address)) { + address = address.slice(0, address.lastIndexOf(':')); + } + const zoneIndex = address.indexOf('%'); + if (zoneIndex !== -1) { + address = address.slice(0, zoneIndex); + } + const mapped = address.match(/^::ffff:(\d+\.\d+\.\d+\.\d+)$/i); + if (mapped?.[1]) { + address = mapped[1]; + } + return net.isIP(address) === 0 ? null : address.toLowerCase(); +} +/** + * Resolves the originating client address of a request. + * + * Forwarding headers are honored only when the TCP peer is a trusted proxy. + * X-Forwarded-For is walked from the nearest hop (right) to the farthest + * (left), skipping trusted proxies; the first untrusted hop is the client. + * X-Real-IP is used only when the trusted peer sent no X-Forwarded-For. + * + * @param peerAddress - TCP socket remote address. + * @param headers - Request headers. + * @param trustedProxies - Trusted proxy list. + * @returns The client address, or null if the peer address is unusable. + * @public + */ +export function resolveClientAddress(peerAddress, headers, trustedProxies) { + const peer = normalizeAddress(peerAddress); + if (!peer) + return null; + if (!headers || !trustedProxies.contains(peer)) + return peer; + const forwardedFor = readHeader(headers, 'x-forwarded-for'); + const hops = forwardedFor + ? forwardedFor.split(',') + : [readHeader(headers, 'x-real-ip') ?? '']; + let client = peer; + for (let index = hops.length - 1; index >= 0; index--) { + const hop = normalizeAddress(hops[index]); + if (!hop) + break; + client = hop; + if (!trustedProxies.contains(hop)) + break; + } + return client; +} +const FORWARDING_HEADERS = [ + 'x-forwarded-for', + 'x-real-ip', + 'forwarded', + 'cf-connecting-ip', + 'true-client-ip' +]; +/** + * Checks whether a request originates from this host. A loopback peer that + * carries forwarding headers is a local reverse proxy relaying an outside + * client, so it is not considered local. + * @param peerAddress - TCP socket remote address. + * @param headers - Request headers. + * @public + */ +export function isLoopbackRequest(peerAddress, headers) { + const peer = normalizeAddress(peerAddress); + const isLoopbackPeer = peer === '::1' || Boolean(peer?.startsWith('127.')); + if (!isLoopbackPeer) + return false; + return !FORWARDING_HEADERS.some((name) => headers && readHeader(headers, name) !== undefined); +} +function readHeader(headers, name) { + const raw = headers[name]; + const value = Array.isArray(raw) ? raw.join(',') : raw; + return value?.trim() || undefined; +} diff --git a/dist/src/utils/clientAddress.test.js b/dist/src/utils/clientAddress.test.js new file mode 100644 index 00000000..388608df --- /dev/null +++ b/dist/src/utils/clientAddress.test.js @@ -0,0 +1,83 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import AdmissionManager from '../managers/admissionManager.js'; +import { isLoopbackRequest, normalizeAddress, resolveClientAddress, TrustedProxyList } from './clientAddress.js'; +const proxies = new TrustedProxyList(['127.0.0.1', '10.0.0.0/8', 'fd00::/8']); +test('normalizeAddress strips mapping, brackets and ports', () => { + assert.equal(normalizeAddress('::ffff:203.0.113.7'), '203.0.113.7'); + assert.equal(normalizeAddress('203.0.113.7:5123'), '203.0.113.7'); + assert.equal(normalizeAddress('[2001:db8::1]:443'), '2001:db8::1'); + assert.equal(normalizeAddress('fe80::1%eth0'), 'fe80::1'); + assert.equal(normalizeAddress('unknown'), null); + assert.equal(normalizeAddress(''), null); +}); +test('TrustedProxyList matches IPv4/IPv6 CIDRs and rejects bad entries', () => { + const list = new TrustedProxyList([ + '192.168.0.0/16', + '2001:db8::/32', + '1.2.3.4/33', + 'nope' + ]); + assert.equal(list.size, 2); + assert.deepEqual(list.invalidEntries, ['1.2.3.4/33', 'nope']); + assert.equal(list.contains('192.168.44.1'), true); + assert.equal(list.contains('::ffff:192.168.44.1'), true); + assert.equal(list.contains('192.169.0.1'), false); + assert.equal(list.contains('2001:db8:ffff::1'), true); + assert.equal(list.contains('2001:db9::1'), false); + assert.equal(new TrustedProxyList(['0.0.0.0/0']).contains('8.8.8.8'), true); +}); +test('TrustedProxyList only matches rules of the same address family', () => { + const ipv6Only = new TrustedProxyList(['::ffff:0:0/96', '::/0']); + assert.equal(ipv6Only.contains('10.0.0.1'), false); + assert.equal(ipv6Only.contains('::ffff:10.0.0.1'), false); + assert.equal(ipv6Only.contains('2001:db8::1'), true); + const ipv4Only = new TrustedProxyList(['0.0.0.0/0']); + assert.equal(ipv4Only.contains('2001:db8::1'), false); + assert.equal(ipv4Only.contains('::ffff:10.0.0.1'), true); +}); +test('forwarding headers from untrusted peers are ignored', () => { + const client = resolveClientAddress('203.0.113.7', { 'x-forwarded-for': '198.51.100.1', 'x-real-ip': '198.51.100.2' }, proxies); + assert.equal(client, '203.0.113.7'); +}); +test('X-Forwarded-For is walked from the nearest hop', () => { + // A client-supplied leftmost entry must not override what the proxy appended. + assert.equal(resolveClientAddress('127.0.0.1', { 'x-forwarded-for': '6.6.6.6, 198.51.100.1' }, proxies), '198.51.100.1'); + // Trusted intermediate hops are skipped. + assert.equal(resolveClientAddress('10.0.0.1', { 'x-forwarded-for': '198.51.100.1, 10.2.0.9' }, proxies), '198.51.100.1'); + // When every hop is trusted, the farthest one is the client. + assert.equal(resolveClientAddress('10.0.0.1', { 'x-forwarded-for': '10.9.9.9, 10.2.0.9' }, proxies), '10.9.9.9'); + // A malformed hop stops the walk at the last trusted address. + assert.equal(resolveClientAddress('10.0.0.1', { 'x-forwarded-for': '198.51.100.1, garbage' }, proxies), '10.0.0.1'); + assert.equal(resolveClientAddress('::ffff:127.0.0.1', { 'x-forwarded-for': '[2001:db8::5]:8080' }, proxies), '2001:db8::5'); +}); +test('X-Real-IP is used only without X-Forwarded-For', () => { + assert.equal(resolveClientAddress('127.0.0.1', { 'x-real-ip': '198.51.100.3' }, proxies), '198.51.100.3'); + assert.equal(resolveClientAddress('127.0.0.1', {}, proxies), '127.0.0.1'); +}); +test('isLoopbackRequest rejects proxied loopback traffic', () => { + assert.equal(isLoopbackRequest('127.0.0.1', {}), true); + assert.equal(isLoopbackRequest('::ffff:127.0.0.1', {}), true); + assert.equal(isLoopbackRequest('::1', {}), true); + assert.equal(isLoopbackRequest('127.0.0.1', { 'x-forwarded-for': '198.51.100.1' }), false); + assert.equal(isLoopbackRequest('127.0.0.1', { forwarded: 'for=1.2.3.4' }), false); + assert.equal(isLoopbackRequest('198.51.100.1', {}), false); +}); +test('trustProxy without trustedProxies ignores forwarding headers', (t) => { + const nodelink = { + options: { server: { password: 'test-password' } } + }; + const admission = new AdmissionManager(nodelink, { + trustProxy: true, + trustedProxies: [] + }); + t.after(() => admission.destroy()); + const request = { + method: 'GET', + url: '/v4/info', + headers: { 'x-forwarded-for': '198.51.100.66' }, + socket: { remoteAddress: '::ffff:127.0.0.1' } + }; + assert.equal(admission.resolveClientAddress(request), '127.0.0.1'); + assert.equal(admission.isTrustedProxy('127.0.0.1'), false); +}); diff --git a/src/api/index.ts b/src/api/index.ts index a257ff60..475dd2db 100644 --- a/src/api/index.ts +++ b/src/api/index.ts @@ -12,6 +12,7 @@ import type { ApiRouteDefinition, ApiRouteModule } from '../typings/api/api.types.ts' +import { isLoopbackRequest } from '../utils/clientAddress.ts' import { logger, sendErrorResponse, @@ -229,10 +230,14 @@ async function requestHandler( nodelink.statsManager.incrementApiRequest(parsedUrl.pathname) const trace = parsedUrl.searchParams.get('trace') === 'true' - const remoteAddress = req.socket?.remoteAddress ?? 'unknown' - const remotePort = req.socket?.remotePort - const isInternal = ['127.0.0.1', '::1', 'localhost'].includes(remoteAddress) - const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}:${remotePort ?? 'unknown'})` + const remoteAddress = + nodelink.admissionManager.resolveClientAddress(req) ?? 'unknown' + const isProxied = nodelink.admissionManager.isTrustedProxy( + req.socket?.remoteAddress + ) + const remotePort = isProxied ? undefined : req.socket?.remotePort + const isInternal = isLoopbackRequest(req.socket?.remoteAddress, req.headers) + const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}${remotePort ? `:${remotePort}` : ''})` const requestId = `${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}` const originalEnd = res.end.bind(res) diff --git a/src/api/profiler.file.ts b/src/api/profiler.file.ts index 9c4175a9..e00545b2 100644 --- a/src/api/profiler.file.ts +++ b/src/api/profiler.file.ts @@ -8,14 +8,9 @@ import type { ApiRouteModule, ApiSendResponse } from '../typings/api/api.types.ts' +import { isLoopbackRequest } from '../utils/clientAddress.ts' import { sendErrorResponse, sendResponse } from '../utils.ts' -/** - * Loopback addresses allowed to access the profiler endpoints when external - * access is disabled. - */ -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']) - /** * JSON-compatible object used for request payload narrowing. */ @@ -364,8 +359,8 @@ async function handler( return } - const remoteAddress = req.socket?.remoteAddress ?? '' - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers) + if (!endpointConfig.allowExternalPatch && !isLocal) { sendErrorResponse( req, res, diff --git a/src/api/profiler.ts b/src/api/profiler.ts index 1aeb5c03..9f8f02ab 100644 --- a/src/api/profiler.ts +++ b/src/api/profiler.ts @@ -10,9 +10,9 @@ import type { ApiRouteModule, ApiSendResponse } from '../typings/api/api.types.ts' +import { isLoopbackRequest } from '../utils/clientAddress.ts' import { sendErrorResponse, sendResponse } from '../utils.ts' -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']) const { NODELINK_PROFILER_DIR: profilerDirectoryEnv } = process.env const profilerBaseDir = profilerDirectoryEnv || '.profiles' @@ -1153,8 +1153,8 @@ function validateAccess( return { ok: false, error: 'Profiler endpoint is disabled.' } } - const remoteAddress = req.socket?.remoteAddress || '' - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers) + if (!endpointConfig.allowExternalPatch && !isLocal) { return { ok: false, error: 'External access to profiler endpoint is blocked.' diff --git a/src/api/profiler.ui.ts b/src/api/profiler.ui.ts index d7ccdf3c..b70c623c 100644 --- a/src/api/profiler.ui.ts +++ b/src/api/profiler.ui.ts @@ -8,9 +8,9 @@ import type { ApiRouteModule, ApiSendResponse } from '../typings/api/api.types.ts' +import { isLoopbackRequest } from '../utils/clientAddress.ts' import { sendErrorResponse } from '../utils.ts' -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']) const __filename = fileURLToPath(import.meta.url) const __dirname = path.dirname(__filename) @@ -2418,8 +2418,8 @@ async function handler( ) } - const remoteAddress = req.socket?.remoteAddress || '' - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers) + if (!endpointConfig.allowExternalPatch && !isLocal) { return sendErrorResponse( req, res, diff --git a/src/api/workers.ts b/src/api/workers.ts index 45c822a0..f4329ecb 100644 --- a/src/api/workers.ts +++ b/src/api/workers.ts @@ -9,14 +9,9 @@ import type { WorkerMetricsEntry, WorkerMetricsPayload } from '../typings/api/stats.types.ts' +import { isLoopbackRequest } from '../utils/clientAddress.ts' import { sendErrorResponse, sendResponse } from '../utils.ts' -/** - * Loopback addresses allowed to access the workers patch endpoint when - * external patching is disabled. - */ -const LOOPBACKS = new Set(['127.0.0.1', '::1', '::ffff:127.0.0.1']) - /** * Request payload accepted by the workers patch endpoint. */ @@ -341,8 +336,8 @@ function handlePatch( return } - const remoteAddress = req.socket?.remoteAddress ?? '' - if (!endpointConfig.allowExternalPatch && !LOOPBACKS.has(remoteAddress)) { + const isLocal = isLoopbackRequest(req.socket?.remoteAddress, req.headers) + if (!endpointConfig.allowExternalPatch && !isLocal) { sendErrorResponse( req, res, diff --git a/src/managers/admissionManager.ts b/src/managers/admissionManager.ts index 97af6351..3dfb026c 100644 --- a/src/managers/admissionManager.ts +++ b/src/managers/admissionManager.ts @@ -10,6 +10,11 @@ import type { SessionAdmissionState } from '../typings/admission/admission.types.ts' import type { ApiRequest } from '../typings/api/api.types.ts' +import { + normalizeAddress, + resolveClientAddress, + TrustedProxyList +} from '../utils/clientAddress.ts' import { logger } from '../utils.ts' type NodelinkServerLike = import('../index.ts').default @@ -53,6 +58,7 @@ const DEFAULT_CONFIG: AdmissionConfig = { baseCapacity: 500, refillRatePerSecond: 100, maxConcurrentSockets: 25, + maxProxySockets: 1024, ipv6SubnetMask: 64, blockScoreThreshold: 30, blockDurationMs: 300000, @@ -107,6 +113,8 @@ export default class AdmissionManager { private readonly guildStates: Map private readonly sessionStates: Map private readonly ipStates: Map + private readonly proxySockets: Map + private readonly trustedProxies: TrustedProxyList private readonly globalState: { tokens: number lastRefill: number @@ -126,6 +134,8 @@ export default class AdmissionManager { this.guildStates = new Map() this.sessionStates = new Map() this.ipStates = new Map() + this.proxySockets = new Map() + this.trustedProxies = this._buildTrustedProxies() const now = performance.now() this.globalState = { @@ -266,6 +276,112 @@ export default class AdmissionManager { return this._buildAllowedDecision('ip', 100, 100) } + /** + * Resolves the originating client address, honoring forwarding headers only + * from trusted proxies. Unlike admission keys, the result is not subnet-masked. + * @param req - Raw or shimmed API request. + */ + resolveClientAddress(req: ApiRequest): string | null { + return resolveClientAddress( + req.socket?.remoteAddress, + req.headers, + this.trustedProxies + ) + } + + /** + * Checks whether a TCP peer is a configured trusted proxy. + * @param rawAddress - Peer address. + */ + isTrustedProxy(rawAddress?: string | null): boolean { + return this.trustedProxies.contains(rawAddress) + } + + /** + * Admits a new TCP connection. Trusted proxies share one aggregate pool, since + * per-client bans and limits are enforced per request once headers are known. + * Other peers are checked against their own block state and socket pool. + * @param rawAddress - Peer address. + * @returns Whether the connection may proceed; call releaseConnection on close. + */ + admitConnection(rawAddress?: string | null): boolean { + if (!this.trustedProxies.contains(rawAddress)) { + if (this.isIpBlocked(rawAddress)) return false + return this.incrementActiveSockets(rawAddress) + } + + const proxy = normalizeAddress(rawAddress) as string + const active = this.proxySockets.get(proxy) ?? 0 + const maxSockets = this.config.ip.maxProxySockets + if (active >= maxSockets) { + logger( + 'warn', + 'AdmissionManager', + `Trusted proxy ${proxy} exceeded aggregate socket pool (${active}/${maxSockets}). Dropping connection.` + ) + return false + } + + this.proxySockets.set(proxy, active + 1) + return true + } + + /** + * Releases a connection admitted by admitConnection. + * @param rawAddress - Peer address. + */ + releaseConnection(rawAddress?: string | null): void { + if (!this.trustedProxies.contains(rawAddress)) { + this.decrementActiveSockets(rawAddress) + return + } + + const proxy = normalizeAddress(rawAddress) as string + const next = (this.proxySockets.get(proxy) ?? 0) - 1 + if (next > 0) { + this.proxySockets.set(proxy, next) + } else { + this.proxySockets.delete(proxy) + } + } + + /** + * Reserves socket capacity for a WebSocket upgrade. Proxied clients share the + * proxy's TCP pool, so their own per-client pool is charged here. + * @param peerAddress - TCP peer address. + * @param clientKey - Resolved client admission key (AdmissionContext.ip). + * @param includePeer - Also admit the peer connection itself, for runtimes + * without a TCP-level connection hook (Bun). + * @returns An idempotent release callback, or null if capacity is exhausted. + */ + reserveUpgrade( + peerAddress: string | null | undefined, + clientKey: string | null, + includePeer: boolean + ): (() => void) | null { + const releases: Array<() => void> = [] + const release = (): void => { + while (releases.length > 0) { + releases.pop()?.() + } + } + + if (includePeer) { + if (!this.admitConnection(peerAddress)) return null + releases.push(() => this.releaseConnection(peerAddress)) + } + + if (clientKey && this.trustedProxies.contains(peerAddress)) { + if (!this.incrementActiveSockets(clientKey)) { + release() + return null + } + releases.push(() => this.decrementActiveSockets(clientKey)) + } + + return release + } + /** * Tracks an incoming TCP socket. Returns false if IP connection pool is exhausted. * @param rawAddress - Remote IP address. @@ -1151,31 +1267,47 @@ export default class AdmissionManager { } /** - * Resolves remote IP with edge proxy header precedence. + * Resolves the admission key for a request's client address. * @internal */ private _resolveIp(req: ApiRequest): string | null { - const socketAddress = req.socket?.remoteAddress - - const trustProxyEnabled = this.config.trustProxy === true - if (!trustProxyEnabled) { - return this._normalizeIp(socketAddress) - } + return this._normalizeIp(this.resolveClientAddress(req)) + } - const headers = req.headers - const cfConnectingIp = this._getHeader(headers, 'cf-connecting-ip') - const trueClientIp = this._getHeader(headers, 'true-client-ip') - const xRealIp = this._getHeader(headers, 'x-real-ip') - const forwardedFor = this._getHeader(headers, 'x-forwarded-for') + /** + * Builds the trusted proxy matcher and reports misconfiguration. + * @internal + */ + private _buildTrustedProxies(): TrustedProxyList { + const { trustProxy, trustedProxies } = this.config - const candidate = - cfConnectingIp ?? - trueClientIp ?? - xRealIp ?? - forwardedFor?.split(',')?.[0]?.trim() ?? - socketAddress + if (!trustProxy) { + if (trustedProxies.length > 0) { + logger( + 'warn', + 'AdmissionManager', + 'admission.trustedProxies is set but admission.trustProxy is false; forwarding headers are ignored.' + ) + } + return new TrustedProxyList() + } - return this._normalizeIp(candidate) + const list = new TrustedProxyList(trustedProxies) + for (const entry of list.invalidEntries) { + logger( + 'warn', + 'AdmissionManager', + `Ignoring invalid admission.trustedProxies entry: ${entry}` + ) + } + if (list.size === 0) { + logger( + 'warn', + 'AdmissionManager', + 'admission.trustProxy is enabled but admission.trustedProxies has no valid entries; forwarding headers are ignored.' + ) + } + return list } /** diff --git a/src/server/bunServer.test.ts b/src/server/bunServer.test.ts new file mode 100644 index 00000000..38e2bea0 --- /dev/null +++ b/src/server/bunServer.test.ts @@ -0,0 +1,148 @@ +import assert from 'node:assert/strict' +import { EventEmitter } from 'node:events' +import test from 'node:test' + +import AdmissionManager from '../managers/admissionManager.ts' +import type { AdmissionConfig } from '../typings/admission/admission.types.ts' +import { type BunServerContext, createBunServer } from './bunServer.ts' + +/* INFO: Bun-only regression; run with `bun test src/server/bunServer.test.ts` */ +const skip = typeof Bun === 'undefined' ? 'requires the Bun runtime' : false + +const PASSWORD = 'test-password' + +async function startBunServer(config: Partial) { + const options = { + server: { host: '127.0.0.1', port: 0, password: PASSWORD }, + playback: { voiceReceive: { enabled: false } }, + cluster: {} + } + const admission = new AdmissionManager( + { options } as unknown as ConstructorParameters[0], + config + ) + const socketBus = new EventEmitter() + const context = { + options, + admissionManager: admission, + socket: socketBus, + sessions: { isResumable: () => false } + } as unknown as BunServerContext + + const server = createBunServer(context, () => + Promise.reject(new Error('REST is not used in this test')) + ) + + return { + url: `ws://127.0.0.1:${server.port}/v4/websocket`, + socketBus, + close: async () => { + admission.destroy() + await server.stop(true) + } + } +} + +function connect( + url: string, + forwardedFor?: string +): Promise { + const headers: Record = { + Authorization: PASSWORD, + 'Client-Name': 'bun-release-test/1.0.0', + 'User-Id': '123456789012345678' + } + if (forwardedFor) headers['X-Forwarded-For'] = forwardedFor + + return new Promise((resolve) => { + const ws = new WebSocket(url, { headers } as unknown as string[]) + ws.addEventListener('open', () => resolve(ws), { once: true }) + ws.addEventListener('error', () => resolve(null), { once: true }) + }) +} + +async function closeAll(sockets: Array): Promise { + await Promise.all( + sockets.map( + (ws) => + new Promise((resolve) => { + if (!ws || ws.readyState === WebSocket.CLOSED) return resolve(null) + ws.addEventListener('close', resolve, { once: true }) + ws.close(1000, 'done') + }) + ) + ) + /* INFO: The server-side close handler releases capacity after the client sees the close */ + await new Promise((resolve) => setTimeout(resolve, 50)) +} + +test('Bun caps direct peers and releases slots on close', { + skip +}, async (t) => { + const harness = await startBunServer({ + ip: { maxConcurrentSockets: 2 } as AdmissionConfig['ip'] + }) + t.after(() => harness.close()) + + for (let round = 0; round < 3; round++) { + const open = [await connect(harness.url), await connect(harness.url)] + assert.ok(open[0] && open[1], `round ${round}: both connections open`) + assert.equal(await connect(harness.url), null, 'third is over the limit') + await closeAll(open) + } +}) + +test('Bun caps proxied clients and the proxy pool, releasing both', { + skip +}, async (t) => { + const harness = await startBunServer({ + trustProxy: true, + trustedProxies: ['127.0.0.1'], + ip: { + maxConcurrentSockets: 2, + maxProxySockets: 3 + } as AdmissionConfig['ip'] + }) + t.after(() => harness.close()) + + for (let round = 0; round < 3; round++) { + const a1 = await connect(harness.url, '198.51.100.1') + const a2 = await connect(harness.url, '198.51.100.1') + assert.ok(a1 && a2, `round ${round}: client A gets two sockets`) + assert.equal(await connect(harness.url, '198.51.100.1'), null) + + const b1 = await connect(harness.url, '198.51.100.2') + assert.ok(b1, 'client B is unaffected by client A') + assert.equal( + await connect(harness.url, '198.51.100.3'), + null, + 'proxy aggregate pool is full' + ) + + await closeAll([a1, a2, b1]) + } +}) + +test('Bun releases slots when the server closes the socket', { + skip +}, async (t) => { + const harness = await startBunServer({ + ip: { maxConcurrentSockets: 1 } as AdmissionConfig['ip'] + }) + t.after(() => harness.close()) + + harness.socketBus.on('/v4/websocket', (wrapper: { close: () => void }) => { + setTimeout(() => wrapper.close(), 10) + }) + + for (let round = 0; round < 3; round++) { + const ws = await connect(harness.url) + assert.ok(ws, `round ${round}: slot available after server close`) + await new Promise((resolve) => + ws.readyState === WebSocket.CLOSED + ? resolve(null) + : ws.addEventListener('close', resolve, { once: true }) + ) + await new Promise((resolve) => setTimeout(resolve, 50)) + } +}) diff --git a/src/server/bunServer.ts b/src/server/bunServer.ts index 45ebe80e..16a8dea2 100644 --- a/src/server/bunServer.ts +++ b/src/server/bunServer.ts @@ -10,6 +10,7 @@ import type { SessionSocket } from '../typings/index.types.ts' import type { ClientInfo, ReqShim } from '../typings/shared.types.ts' +import { isLoopbackRequest } from '../utils/clientAddress.ts' import { logger, parseClient, verifyDiscordID } from '../utils.ts' const VOICE_PATH_RE = /^\/v4\/websocket\/voice\/([A-Za-z0-9]+)\/?$/ @@ -197,7 +198,10 @@ export function createBunServer( if (pathname === '/v4/profiler/socket') { const remoteAddress = server.requestIP(req)?.address || 'unknown' - const isInternal = /^(::1|localhost|127\.0\.0\.1)/.test(remoteAddress) + const isInternal = isLoopbackRequest( + remoteAddress, + Object.fromEntries(req.headers) + ) const endpoint = context.options.cluster?.endpoint || {} const patchEnabled = endpoint.patchEnabled === true const allowExternalPatch = endpoint.allowExternalPatch === true @@ -251,7 +255,16 @@ export function createBunServer( const isMainWs = pathname === '/v4/websocket' if (isMainWs || voiceMatch || liveMatch) { - const remoteAddress = server.requestIP(req)?.address || 'unknown' + const peerAddress = server.requestIP(req)?.address || 'unknown' + const upgradeReqShim: ApiRequest = { + method: req.method, + url: req.url, + headers: Object.fromEntries(req.headers), + socket: { remoteAddress: peerAddress } + } + const remoteAddress = + context.admissionManager.resolveClientAddress(upgradeReqShim) ?? + peerAddress const clientAddress = `[External] (${remoteAddress})` const isIpBlocked = context.admissionManager.isIpBlocked(remoteAddress) @@ -262,13 +275,6 @@ export function createBunServer( }) } - const upgradeReqShim: ApiRequest = { - method: req.method, - url: req.url, - headers: Object.fromEntries(req.headers), - socket: { remoteAddress } - } - const admissionContext = context.admissionManager.resolveContext( upgradeReqShim, url @@ -375,6 +381,23 @@ export function createBunServer( sessionId = null } + /* INFO: Bun has no TCP connection hook, so peer and per-client capacity are reserved per upgrade */ + const releaseCapacity = context.admissionManager.reserveUpgrade( + peerAddress, + admissionContext.ip, + true + ) + if (!releaseCapacity) { + return new Response('Too many concurrent connections.', { + status: 429, + statusText: 'Too Many Requests', + headers: { + 'Nodelink-Api-Version': '4', + IamNodelink: 'true' + } + }) + } + const success = server.upgrade(req, { data: { clientInfo, @@ -384,11 +407,13 @@ export function createBunServer( url: req.url, pathname, eventName, - routeId + routeId, + releaseCapacity } }) if (success) return undefined + releaseCapacity() return new Response('WebSocket upgrade failed', { status: 400, headers: { @@ -618,6 +643,7 @@ export function createBunServer( wrapper._handleMessage(message) }, close(ws: ServerWebSocket, code: number, reason: string) { + ws.data?.releaseCapacity?.() const wrapper = ws.data?.wrapper if (!wrapper) { logger( diff --git a/src/server/clientIsolation.test.ts b/src/server/clientIsolation.test.ts new file mode 100644 index 00000000..1cd4b081 --- /dev/null +++ b/src/server/clientIsolation.test.ts @@ -0,0 +1,237 @@ +import assert from 'node:assert/strict' +import type http from 'node:http' +import type { AddressInfo, Socket } from 'node:net' +import test from 'node:test' +import WebSocketServer from '@performanc/pwsl-server' + +import requestHandler from '../api/index.ts' +import type NodelinkServer from '../index.ts' +import AdmissionManager from '../managers/admissionManager.ts' +import type { AdmissionConfig } from '../typings/admission/admission.types.ts' +import { createHttpServer } from './httpServer.ts' + +const PASSWORD = 'test-password' + +interface Harness { + url: string + restUrl: string + socketBus: WebSocketServer + close: () => Promise +} + +async function startServer(config: Partial): Promise { + const options = { + server: { password: PASSWORD }, + playback: { voiceReceive: { enabled: false } }, + cluster: {}, + api: {} + } + const admission = new AdmissionManager( + { options } as unknown as ConstructorParameters[0], + config + ) + const socketBus = new WebSocketServer() + const context = { + options, + admissionManager: admission, + socket: socketBus, + sessions: { isResumable: () => false }, + pluginManager: { callHook: () => {} }, + statsManager: { + incrementApiRequest: () => {}, + recordHttpRequestDuration: () => {} + }, + extensions: { middlewares: [], routes: [] } + } as unknown as NodelinkServer + + const server = createHttpServer(context, () => + Promise.resolve(requestHandler) + ) as http.Server + /* INFO: Upgraded sockets are not closed by closeAllConnections, so a failing test would hang teardown */ + const sockets = new Set() + server.on('connection', (socket: Socket) => sockets.add(socket)) + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)) + const { port } = server.address() as AddressInfo + + return { + url: `ws://127.0.0.1:${port}/v4/websocket`, + restUrl: `http://127.0.0.1:${port}/v4/nodelink-test-route`, + socketBus, + close: async () => { + admission.destroy() + for (const socket of sockets) socket.destroy() + await new Promise((resolve) => server.close(() => resolve())) + } + } +} + +function connect( + url: string, + forwardedFor?: string, + password = PASSWORD +): Promise { + const headers: Record = { + Authorization: password, + 'Client-Name': 'release-test/1.0.0', + 'User-Id': '123456789012345678' + } + if (forwardedFor) headers['X-Forwarded-For'] = forwardedFor + + return new Promise((resolve) => { + const ws = new WebSocket(url, { headers } as unknown as string[]) + ws.addEventListener('open', () => resolve(ws), { once: true }) + ws.addEventListener('error', () => resolve(null), { once: true }) + }) +} + +async function closeClient(ws: WebSocket): Promise { + if (ws.readyState === WebSocket.CLOSED) return + const closed = new Promise((resolve) => + ws.addEventListener('close', resolve, { once: true }) + ) + ws.close(1000, 'done') + await closed +} + +/* INFO: The server releases on its own close handling, which can trail the client's close event */ +async function connectEventually( + url: string, + forwardedFor?: string +): Promise { + for (let attempt = 0; attempt < 20; attempt++) { + const ws = await connect(url, forwardedFor) + if (ws) return ws + await new Promise((resolve) => setTimeout(resolve, 50)) + } + return null +} + +test('direct peers get their socket slots back after normal closes', async (t) => { + const harness = await startServer({ + ip: { maxConcurrentSockets: 2 } as AdmissionConfig['ip'] + }) + t.after(() => harness.close()) + + for (let round = 0; round < 3; round++) { + const first = await connectEventually(harness.url) + const second = await connectEventually(harness.url) + assert.ok(first && second, `round ${round}: both connections open`) + assert.equal(await connect(harness.url), null, 'third is over the limit') + + await closeClient(first) + await closeClient(second) + } +}) + +test('proxied clients release both per-client and proxy pools', async (t) => { + const harness = await startServer({ + trustProxy: true, + trustedProxies: ['127.0.0.1'], + ip: { + maxConcurrentSockets: 2, + maxProxySockets: 3 + } as AdmissionConfig['ip'] + }) + t.after(() => harness.close()) + + for (let round = 0; round < 3; round++) { + const a1 = await connectEventually(harness.url, '198.51.100.1') + const a2 = await connectEventually(harness.url, '198.51.100.1') + assert.ok(a1 && a2, `round ${round}: client A gets two sockets`) + assert.equal( + await connect(harness.url, '198.51.100.1'), + null, + 'client A is capped per client' + ) + + const b1 = await connectEventually(harness.url, '198.51.100.2') + assert.ok(b1, 'client B is unaffected by client A') + assert.equal( + await connect(harness.url, '198.51.100.3'), + null, + 'proxy aggregate pool is full' + ) + + await closeClient(a1) + await closeClient(a2) + await closeClient(b1) + } +}) + +test('server-side destroy releases the slot', async (t) => { + const harness = await startServer({ + ip: { maxConcurrentSockets: 1 } as AdmissionConfig['ip'] + }) + t.after(() => harness.close()) + + harness.socketBus.on('/v4/websocket', (ws: { destroy: () => void }) => { + setTimeout(() => ws.destroy(), 10) + }) + + for (let round = 0; round < 3; round++) { + const ws = await connectEventually(harness.url) + assert.ok(ws, `round ${round}: slot available after server destroy`) + await new Promise((resolve) => + ws.readyState === WebSocket.CLOSED + ? resolve(null) + : ws.addEventListener('close', resolve, { once: true }) + ) + } +}) + +async function restStatus( + url: string, + forwardedFor: string, + password = PASSWORD +): Promise { + const response = await fetch(url, { + headers: { Authorization: password, 'X-Forwarded-For': forwardedFor } + }) + await response.body?.cancel() + return response.status +} + +const proxyConfig: Partial = { + trustProxy: true, + trustedProxies: ['127.0.0.1'] +} + +test('REST auth failures behind a proxy ban only the offending client', async (t) => { + const harness = await startServer(proxyConfig) + t.after(() => harness.close()) + + for (let attempt = 0; attempt < 5; attempt++) { + assert.equal( + await restStatus(harness.restUrl, '198.51.100.66', 'wrong'), + 401 + ) + } + + assert.equal(await restStatus(harness.restUrl, '198.51.100.66'), 403) + assert.equal( + await restStatus(harness.restUrl, '6.6.6.6, 198.51.100.66'), + 403, + 'a spoofed leftmost hop does not lift the ban' + ) + assert.equal(await restStatus(harness.restUrl, '198.51.100.10'), 404) + + const bystander = await connectEventually(harness.url, '198.51.100.10') + assert.ok(bystander, 'other clients behind the proxy still connect') + await closeClient(bystander) +}) + +test('WebSocket auth failures behind a proxy ban only the offending client', async (t) => { + const harness = await startServer(proxyConfig) + t.after(() => harness.close()) + + for (let attempt = 0; attempt < 5; attempt++) { + assert.equal(await connect(harness.url, '198.51.100.66', 'wrong'), null) + } + + assert.equal(await connect(harness.url, '198.51.100.66'), null) + assert.equal(await restStatus(harness.restUrl, '198.51.100.66'), 403) + + const bystander = await connectEventually(harness.url, '198.51.100.10') + assert.ok(bystander, 'other clients behind the proxy still connect') + await closeClient(bystander) +}) diff --git a/src/server/httpServer.ts b/src/server/httpServer.ts index 3b2d50c3..424983c5 100644 --- a/src/server/httpServer.ts +++ b/src/server/httpServer.ts @@ -5,6 +5,7 @@ import process from 'node:process' import type NodelinkServer from '../index.ts' import type { ApiNodelinkServer } from '../typings/api/api.types.ts' import { logger } from '../utils.ts' +import { trackSocketRelease } from './socketRelease.ts' import { handleHttpUpgrade } from './wsRouter.ts' type RequestHandlerType = typeof import('../api/index.ts').default @@ -43,22 +44,16 @@ function createHttpServer( server.on('connection', (socket: NetSocket) => { const remoteAddress = socket.remoteAddress - const isIpBlocked = nodelink.admissionManager.isIpBlocked(remoteAddress) - if (isIpBlocked) { + const connectionAllowed = + nodelink.admissionManager.admitConnection(remoteAddress) + if (!connectionAllowed) { socket.destroy() return } - const socketAllowed = - nodelink.admissionManager.incrementActiveSockets(remoteAddress) - if (!socketAllowed) { - socket.destroy() - return - } - - socket.on('close', () => { - nodelink.admissionManager.decrementActiveSockets(remoteAddress) - }) + trackSocketRelease(socket, () => + nodelink.admissionManager.releaseConnection(remoteAddress) + ) socket.on('error', (err: NodeJS.ErrnoException) => { const isBenign = err?.code === 'EPIPE' || err?.code === 'ECONNRESET' diff --git a/src/server/socketRelease.ts b/src/server/socketRelease.ts new file mode 100644 index 00000000..172eaf61 --- /dev/null +++ b/src/server/socketRelease.ts @@ -0,0 +1,65 @@ +import type { EventEmitter } from 'node:events' +import type { Socket as NetSocket } from 'node:net' + +const pendingReleases = new WeakMap void>>() + +/** + * Registers a capacity release to run once when the socket goes away. + * The TCP 'close' event is only a fallback: PWSL strips every socket + * listener when a WebSocket closes, so upgraded sockets must also be + * bound with bindWebSocketRelease. + * @param socket - TCP socket holding the capacity. + * @param release - Callback that returns the capacity. + * @public + */ +export function trackSocketRelease( + socket: NetSocket, + release: () => void +): void { + const releases = pendingReleases.get(socket) + if (releases) { + releases.push(release) + return + } + + pendingReleases.set(socket, [release]) + socket.once('close', () => releaseSocket(socket)) +} + +/** + * Runs and clears every pending release for the socket. Idempotent. + * @param socket - TCP socket holding the capacity. + * @internal + */ +function releaseSocket(socket: NetSocket): void { + const releases = pendingReleases.get(socket) + if (!releases) return + + pendingReleases.delete(socket) + for (const release of releases) { + release() + } +} + +/** + * Releases the socket's capacity when its WebSocket ends. PWSL emits 'close' + * for peer-initiated and transport closes, but its destroy() (used for + * protocol errors and server-side teardown) does not, so it is wrapped too. + * @param ws - WebSocket connection created by the upgrade. + * @param socket - Underlying TCP socket. + * @public + */ +export function bindWebSocketRelease( + ws: EventEmitter & { destroy?: () => void }, + socket: NetSocket +): void { + ws.once('close', () => releaseSocket(socket)) + + const destroy = ws.destroy + if (typeof destroy !== 'function') return + + ws.destroy = function (this: unknown) { + releaseSocket(socket) + return destroy.call(this) + } +} diff --git a/src/server/wsRouter.ts b/src/server/wsRouter.ts index 9df53a34..ad12fede 100644 --- a/src/server/wsRouter.ts +++ b/src/server/wsRouter.ts @@ -9,20 +9,15 @@ import { attachProfilerSocket } from '../api/profiler.socket.ts' import type NodelinkServer from '../index.ts' import type { RequestShim, SessionSocket } from '../typings/index.types.ts' import type { ClientInfo } from '../typings/shared.types.ts' +import { isLoopbackRequest } from '../utils/clientAddress.ts' import { decodeTrack, logger, parseClient, verifyDiscordID } from '../utils.ts' +import { bindWebSocketRelease, trackSocketRelease } from './socketRelease.ts' import { handleClientWebSocket } from './wsSession.ts' const VOICE_PATH_RE = /^\/v4\/websocket\/voice\/([A-Za-z0-9]+)\/?$/ const LIVE_PATH_RE = /^\/v4\/websocket\/youtube\/live\/([^/]+)\/?$/ const DISCORD_SNOWFLAKE_RE = /^\d{17,20}$/ -const INTERNAL_IPS = new Set([ - '127.0.0.1', - '::1', - '::ffff:127.0.0.1', - 'localhost' -]) - function _getHeader( headers: http.IncomingHttpHeaders, name: string @@ -80,16 +75,19 @@ function handleHttpUpgrade( logger('debug', 'Server', `Upgrade socket error: ${err.message}`) }) - const remoteAddress = request.socket.remoteAddress || 'unknown' - const remotePort = request.socket.remotePort || 0 - const isInternal = INTERNAL_IPS.has(remoteAddress) - const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}:${remotePort})` + const admission = context.admissionManager + const peerAddress = request.socket.remoteAddress + const remoteAddress = admission.resolveClientAddress(request) ?? 'unknown' + const isProxied = admission.isTrustedProxy(peerAddress) + const remotePort = isProxied ? 0 : request.socket.remotePort || 0 + const isInternal = isLoopbackRequest(peerAddress, request.headers) + const clientAddress = `${isInternal ? '[Internal]' : '[External]'} (${remoteAddress}${remotePort ? `:${remotePort}` : ''})` const url = new URL(request.url || '/', 'http://localhost') const pathname = url.pathname - const admissionContext = context.admissionManager.resolveContext(request, url) - const admissionDecision = context.admissionManager.admit(admissionContext) + const admissionContext = admission.resolveContext(request, url) + const admissionDecision = admission.admit(admissionContext) if (!admissionDecision.allowed) { _rejectUpgrade( socket, @@ -100,6 +98,22 @@ function handleHttpUpgrade( return } + const releaseUpgrade = admission.reserveUpgrade( + peerAddress, + admissionContext.ip, + false + ) + if (!releaseUpgrade) { + _rejectUpgrade( + socket, + 429, + 'Too Many Requests', + 'Too many concurrent connections.' + ) + return + } + trackSocketRelease(socket, releaseUpgrade) + if (pathname === '/v4/profiler/socket') { _handleProfilerUpgrade( context, @@ -161,6 +175,7 @@ function _handleProfilerUpgrade( const wsServer = context.socket as WebSocketServer wsServer?.handleUpgrade(request, socket, head, null, (ws) => { + bindWebSocketRelease(ws, socket) context.socket?.emit( '/v4/profiler/socket', ws as SessionSocket, @@ -195,7 +210,9 @@ function _handleGatewayUpgrade( context.options.server?.password ?? '' ) if (!isAuthorized) { - context.admissionManager.recordAuthFailure(request.socket.remoteAddress) + context.admissionManager.recordAuthFailure( + context.admissionManager.resolveClientAddress(request) + ) reject(401, 'Unauthorized', 'Invalid password provided.') return } @@ -254,6 +271,7 @@ function _handleGatewayUpgrade( const wsServer = context.socket as WebSocketServer wsServer?.handleUpgrade(request, socket, head, null, (ws) => { + bindWebSocketRelease(ws, socket) context.socket?.emit( eventName, ws as SessionSocket, diff --git a/src/typings/admission/admission.types.ts b/src/typings/admission/admission.types.ts index 48e7caf8..64963e9a 100644 --- a/src/typings/admission/admission.types.ts +++ b/src/typings/admission/admission.types.ts @@ -368,9 +368,15 @@ export interface IpAdmissionConfig { /** Refill rate: tokens added per second. */ refillRatePerSecond: number - /** Maximum concurrent TCP connections per IP. */ + /** + * Maximum concurrent TCP connections per client IP. Behind a trusted proxy + * this applies to WebSocket upgrades of each forwarded client. + */ maxConcurrentSockets: number + /** Maximum concurrent TCP connections from a single trusted proxy (aggregate). */ + maxProxySockets: number + /** Subnet mask for IPv6 grouping (e.g. 64). */ ipv6SubnetMask: number @@ -415,10 +421,13 @@ export interface AdmissionConfig { /** Master toggle for the entire admission manager. */ enabled: boolean - /** Whether to trust proxy headers (CF-Connecting-IP, etc.). */ + /** + * Whether to resolve client IPs from X-Forwarded-For / X-Real-IP. + * Only honored for peers listed in trustedProxies. + */ trustProxy: boolean - /** Trusted proxy IP list. */ + /** Trusted reverse proxy IPs or CIDR blocks (e.g. '127.0.0.1', '10.0.0.0/8'). */ trustedProxies: string[] /** Guild execution layer config. */ diff --git a/src/typings/index.types.ts b/src/typings/index.types.ts index 385af904..f11a039e 100644 --- a/src/typings/index.types.ts +++ b/src/typings/index.types.ts @@ -66,6 +66,12 @@ export interface BunSocketData { * @internal */ routeId?: string | null + + /** + * Returns the socket capacity reserved for this upgrade; idempotent + * @internal + */ + releaseCapacity?: () => void } /** diff --git a/src/utils/clientAddress.test.ts b/src/utils/clientAddress.test.ts new file mode 100644 index 00000000..8885a41e --- /dev/null +++ b/src/utils/clientAddress.test.ts @@ -0,0 +1,149 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import AdmissionManager from '../managers/admissionManager.ts' +import type { ApiRequest } from '../typings/api/api.types.ts' +import { + isLoopbackRequest, + normalizeAddress, + resolveClientAddress, + TrustedProxyList +} from './clientAddress.ts' + +const proxies = new TrustedProxyList(['127.0.0.1', '10.0.0.0/8', 'fd00::/8']) + +test('normalizeAddress strips mapping, brackets and ports', () => { + assert.equal(normalizeAddress('::ffff:203.0.113.7'), '203.0.113.7') + assert.equal(normalizeAddress('203.0.113.7:5123'), '203.0.113.7') + assert.equal(normalizeAddress('[2001:db8::1]:443'), '2001:db8::1') + assert.equal(normalizeAddress('fe80::1%eth0'), 'fe80::1') + assert.equal(normalizeAddress('unknown'), null) + assert.equal(normalizeAddress(''), null) +}) + +test('TrustedProxyList matches IPv4/IPv6 CIDRs and rejects bad entries', () => { + const list = new TrustedProxyList([ + '192.168.0.0/16', + '2001:db8::/32', + '1.2.3.4/33', + 'nope' + ]) + assert.equal(list.size, 2) + assert.deepEqual(list.invalidEntries, ['1.2.3.4/33', 'nope']) + assert.equal(list.contains('192.168.44.1'), true) + assert.equal(list.contains('::ffff:192.168.44.1'), true) + assert.equal(list.contains('192.169.0.1'), false) + assert.equal(list.contains('2001:db8:ffff::1'), true) + assert.equal(list.contains('2001:db9::1'), false) + assert.equal(new TrustedProxyList(['0.0.0.0/0']).contains('8.8.8.8'), true) +}) + +test('TrustedProxyList only matches rules of the same address family', () => { + const ipv6Only = new TrustedProxyList(['::ffff:0:0/96', '::/0']) + assert.equal(ipv6Only.contains('10.0.0.1'), false) + assert.equal(ipv6Only.contains('::ffff:10.0.0.1'), false) + assert.equal(ipv6Only.contains('2001:db8::1'), true) + + const ipv4Only = new TrustedProxyList(['0.0.0.0/0']) + assert.equal(ipv4Only.contains('2001:db8::1'), false) + assert.equal(ipv4Only.contains('::ffff:10.0.0.1'), true) +}) + +test('forwarding headers from untrusted peers are ignored', () => { + const client = resolveClientAddress( + '203.0.113.7', + { 'x-forwarded-for': '198.51.100.1', 'x-real-ip': '198.51.100.2' }, + proxies + ) + assert.equal(client, '203.0.113.7') +}) + +test('X-Forwarded-For is walked from the nearest hop', () => { + // A client-supplied leftmost entry must not override what the proxy appended. + assert.equal( + resolveClientAddress( + '127.0.0.1', + { 'x-forwarded-for': '6.6.6.6, 198.51.100.1' }, + proxies + ), + '198.51.100.1' + ) + // Trusted intermediate hops are skipped. + assert.equal( + resolveClientAddress( + '10.0.0.1', + { 'x-forwarded-for': '198.51.100.1, 10.2.0.9' }, + proxies + ), + '198.51.100.1' + ) + // When every hop is trusted, the farthest one is the client. + assert.equal( + resolveClientAddress( + '10.0.0.1', + { 'x-forwarded-for': '10.9.9.9, 10.2.0.9' }, + proxies + ), + '10.9.9.9' + ) + // A malformed hop stops the walk at the last trusted address. + assert.equal( + resolveClientAddress( + '10.0.0.1', + { 'x-forwarded-for': '198.51.100.1, garbage' }, + proxies + ), + '10.0.0.1' + ) + assert.equal( + resolveClientAddress( + '::ffff:127.0.0.1', + { 'x-forwarded-for': '[2001:db8::5]:8080' }, + proxies + ), + '2001:db8::5' + ) +}) + +test('X-Real-IP is used only without X-Forwarded-For', () => { + assert.equal( + resolveClientAddress('127.0.0.1', { 'x-real-ip': '198.51.100.3' }, proxies), + '198.51.100.3' + ) + assert.equal(resolveClientAddress('127.0.0.1', {}, proxies), '127.0.0.1') +}) + +test('isLoopbackRequest rejects proxied loopback traffic', () => { + assert.equal(isLoopbackRequest('127.0.0.1', {}), true) + assert.equal(isLoopbackRequest('::ffff:127.0.0.1', {}), true) + assert.equal(isLoopbackRequest('::1', {}), true) + assert.equal( + isLoopbackRequest('127.0.0.1', { 'x-forwarded-for': '198.51.100.1' }), + false + ) + assert.equal( + isLoopbackRequest('127.0.0.1', { forwarded: 'for=1.2.3.4' }), + false + ) + assert.equal(isLoopbackRequest('198.51.100.1', {}), false) +}) + +test('trustProxy without trustedProxies ignores forwarding headers', (t) => { + const nodelink = { + options: { server: { password: 'test-password' } } + } as unknown as ConstructorParameters[0] + const admission = new AdmissionManager(nodelink, { + trustProxy: true, + trustedProxies: [] + }) + t.after(() => admission.destroy()) + + const request = { + method: 'GET', + url: '/v4/info', + headers: { 'x-forwarded-for': '198.51.100.66' }, + socket: { remoteAddress: '::ffff:127.0.0.1' } + } as ApiRequest + assert.equal(admission.resolveClientAddress(request), '127.0.0.1') + assert.equal(admission.isTrustedProxy('127.0.0.1'), false) +}) diff --git a/src/utils/clientAddress.ts b/src/utils/clientAddress.ts new file mode 100644 index 00000000..4e47fbc0 --- /dev/null +++ b/src/utils/clientAddress.ts @@ -0,0 +1,170 @@ +import net from 'node:net' + +type HeaderBag = Record + +/** + * Matches peer addresses against an explicit list of trusted proxy IPs/CIDRs. + * IPv4 and IPv6 rules are kept in separate lists so an address only matches + * rules of its own family. Invalid entries are reported through + * `invalidEntries` and never match. + * @public + */ +export class TrustedProxyList { + private readonly ipv4 = new net.BlockList() + private readonly ipv6 = new net.BlockList() + readonly invalidEntries: string[] = [] + readonly size: number = 0 + + constructor(entries: readonly string[] = []) { + for (const entry of entries) { + if (this.add(entry)) { + this.size += 1 + } else { + this.invalidEntries.push(entry) + } + } + } + + /** + * Checks whether an address belongs to a trusted proxy. + * @param rawAddress - Peer address (IPv4, IPv6 or IPv4-mapped IPv6). + */ + contains(rawAddress?: string | null): boolean { + if (this.size === 0) return false + + const address = normalizeAddress(rawAddress) + if (!address) return false + + return net.isIP(address) === 4 + ? this.ipv4.check(address, 'ipv4') + : this.ipv6.check(address, 'ipv6') + } + + private add(entry: string): boolean { + const [rawAddress, rawPrefix, ...rest] = entry.trim().split('/') + if (rest.length > 0) return false + + const address = normalizeAddress(rawAddress) + if (!address) return false + + const isIpv4 = net.isIP(address) === 4 + const bits = isIpv4 ? 32 : 128 + const prefix = rawPrefix === undefined ? bits : Number(rawPrefix) + const isValidPrefix = + rawPrefix === undefined || + (/^\d+$/.test(rawPrefix) && prefix >= 0 && prefix <= bits) + if (!isValidPrefix) return false + + if (isIpv4) { + this.ipv4.addSubnet(address, prefix, 'ipv4') + } else { + this.ipv6.addSubnet(address, prefix, 'ipv6') + } + return true + } +} + +/** + * Strips IPv4-mapped prefixes, brackets and ports from an address. + * Returns null when the result is not a valid IP literal. + * @public + */ +export function normalizeAddress(rawAddress?: string | null): string | null { + if (!rawAddress) return null + let address = rawAddress.trim() + if (!address) return null + + const bracketed = address.match(/^\[([^\]]+)\](?::\d+)?$/) + if (bracketed?.[1]) { + address = bracketed[1] + } else if (/^[\d.]+:\d+$/.test(address)) { + address = address.slice(0, address.lastIndexOf(':')) + } + + const zoneIndex = address.indexOf('%') + if (zoneIndex !== -1) { + address = address.slice(0, zoneIndex) + } + + const mapped = address.match(/^::ffff:(\d+\.\d+\.\d+\.\d+)$/i) + if (mapped?.[1]) { + address = mapped[1] + } + + return net.isIP(address) === 0 ? null : address.toLowerCase() +} + +/** + * Resolves the originating client address of a request. + * + * Forwarding headers are honored only when the TCP peer is a trusted proxy. + * X-Forwarded-For is walked from the nearest hop (right) to the farthest + * (left), skipping trusted proxies; the first untrusted hop is the client. + * X-Real-IP is used only when the trusted peer sent no X-Forwarded-For. + * + * @param peerAddress - TCP socket remote address. + * @param headers - Request headers. + * @param trustedProxies - Trusted proxy list. + * @returns The client address, or null if the peer address is unusable. + * @public + */ +export function resolveClientAddress( + peerAddress: string | null | undefined, + headers: HeaderBag | undefined, + trustedProxies: TrustedProxyList +): string | null { + const peer = normalizeAddress(peerAddress) + if (!peer) return null + if (!headers || !trustedProxies.contains(peer)) return peer + + const forwardedFor = readHeader(headers, 'x-forwarded-for') + const hops = forwardedFor + ? forwardedFor.split(',') + : [readHeader(headers, 'x-real-ip') ?? ''] + + let client = peer + for (let index = hops.length - 1; index >= 0; index--) { + const hop = normalizeAddress(hops[index]) + if (!hop) break + + client = hop + if (!trustedProxies.contains(hop)) break + } + + return client +} + +const FORWARDING_HEADERS = [ + 'x-forwarded-for', + 'x-real-ip', + 'forwarded', + 'cf-connecting-ip', + 'true-client-ip' +] + +/** + * Checks whether a request originates from this host. A loopback peer that + * carries forwarding headers is a local reverse proxy relaying an outside + * client, so it is not considered local. + * @param peerAddress - TCP socket remote address. + * @param headers - Request headers. + * @public + */ +export function isLoopbackRequest( + peerAddress: string | null | undefined, + headers: HeaderBag | undefined +): boolean { + const peer = normalizeAddress(peerAddress) + const isLoopbackPeer = peer === '::1' || Boolean(peer?.startsWith('127.')) + if (!isLoopbackPeer) return false + + return !FORWARDING_HEADERS.some( + (name) => headers && readHeader(headers, name) !== undefined + ) +} + +function readHeader(headers: HeaderBag, name: string): string | undefined { + const raw = headers[name] + const value = Array.isArray(raw) ? raw.join(',') : raw + return value?.trim() || undefined +}