From 146434d95a6f09620f8431f87233cba0964cc9c5 Mon Sep 17 00:00:00 2001 From: JustEvil <71156616+EvilG-MC@users.noreply.github.com> Date: Sat, 10 Oct 2026 08:02:01 -0600 Subject: [PATCH 1/6] fix: isolate clients behind trusted reverse proxies Authentication bans and connection limits keyed on the TCP peer, so behind a reverse proxy every client shared the proxy's address. One client failing auth jailed the proxy for 15 minutes and NodeLink reset all of its connections, which the proxy surfaced as HTTP 502. Enabling trustProxy did not help: the socket guards ignored it, and when enabled it trusted forwarding headers from any peer, so identities were spoofable. - Resolve the client address only through admission.trustedProxies (IPs/CIDRs), walking X-Forwarded-For from the nearest hop and ignoring forwarded identities from untrusted peers - Apply auth bans and REST/WebSocket upgrade limits per client - Give trusted proxies a separate aggregate TCP pool (admission.ip.maxProxySockets) and release capacity on close - Treat loopback requests carrying forwarding headers as external in the profiler and workers endpoints --- config.default.ts | 1 + src/api/index.ts | 13 +- src/api/profiler.file.ts | 11 +- src/api/profiler.ts | 6 +- src/api/profiler.ui.ts | 6 +- src/api/workers.ts | 11 +- src/managers/admissionManager.ts | 133 +++++++++++-- src/server/bunServer.ts | 24 ++- src/server/httpServer.ts | 14 +- src/server/wsRouter.ts | 42 ++-- src/typings/admission/admission.types.ts | 15 +- src/utils/clientAddress.test.ts | 198 +++++++++++++++++++ src/utils/clientAddress.ts | 238 +++++++++++++++++++++++ 13 files changed, 631 insertions(+), 81 deletions(-) create mode 100644 src/utils/clientAddress.test.ts create mode 100644 src/utils/clientAddress.ts diff --git a/config.default.ts b/config.default.ts index 31ad4bc5a..bf5f98b92 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/src/api/index.ts b/src/api/index.ts index a257ff601..475dd2db4 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 9c4175a9f..e00545b2d 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 1aeb5c035..9f8f02ab4 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 d7ccdf3c4..b70c623c2 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 45c822a09..f4329ecbf 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 97af63510..f34cb0936 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,75 @@ 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) + } + } + /** * Tracks an incoming TCP socket. Returns false if IP connection pool is exhausted. * @param rawAddress - Remote IP address. @@ -1151,31 +1230,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.ts b/src/server/bunServer.ts index 45ebe80e9..118bafbc1 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 diff --git a/src/server/httpServer.ts b/src/server/httpServer.ts index 3b2d50c31..ba9ac3fa1 100644 --- a/src/server/httpServer.ts +++ b/src/server/httpServer.ts @@ -43,21 +43,15 @@ function createHttpServer( server.on('connection', (socket: NetSocket) => { const remoteAddress = socket.remoteAddress - const isIpBlocked = nodelink.admissionManager.isIpBlocked(remoteAddress) - if (isIpBlocked) { - socket.destroy() - return - } - - const socketAllowed = - nodelink.admissionManager.incrementActiveSockets(remoteAddress) - if (!socketAllowed) { + const connectionAllowed = + nodelink.admissionManager.admitConnection(remoteAddress) + if (!connectionAllowed) { socket.destroy() return } socket.on('close', () => { - nodelink.admissionManager.decrementActiveSockets(remoteAddress) + nodelink.admissionManager.releaseConnection(remoteAddress) }) socket.on('error', (err: NodeJS.ErrnoException) => { diff --git a/src/server/wsRouter.ts b/src/server/wsRouter.ts index 9df53a344..dd760779e 100644 --- a/src/server/wsRouter.ts +++ b/src/server/wsRouter.ts @@ -9,6 +9,7 @@ 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 { handleClientWebSocket } from './wsSession.ts' @@ -16,13 +17,6 @@ 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 +74,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 +97,21 @@ function handleHttpUpgrade( return } + /* INFO: Proxied clients share the proxy's TCP pool, so cap each client's upgraded sockets here */ + if (isProxied && admissionContext.ip) { + const clientIp = admissionContext.ip + if (!admission.incrementActiveSockets(clientIp)) { + _rejectUpgrade( + socket, + 429, + 'Too Many Requests', + 'Too many concurrent connections.' + ) + return + } + socket.once('close', () => admission.decrementActiveSockets(clientIp)) + } + if (pathname === '/v4/profiler/socket') { _handleProfilerUpgrade( context, @@ -195,7 +207,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 } diff --git a/src/typings/admission/admission.types.ts b/src/typings/admission/admission.types.ts index 48e7caf82..64963e9a9 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/utils/clientAddress.test.ts b/src/utils/clientAddress.test.ts new file mode 100644 index 000000000..f2a1fc4f2 --- /dev/null +++ b/src/utils/clientAddress.test.ts @@ -0,0 +1,198 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import AdmissionManager from '../managers/admissionManager.ts' +import type { AdmissionConfig } from '../typings/admission/admission.types.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('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) +}) + +function createAdmission(config: Partial): AdmissionManager { + const nodelink = { + options: { server: { password: 'test-password' } } + } as unknown as ConstructorParameters[0] + return new AdmissionManager(nodelink, config) +} + +function proxiedRequest(clientIp: string): ApiRequest { + return { + method: 'GET', + url: '/v4/info', + headers: { 'x-forwarded-for': clientIp }, + socket: { remoteAddress: '::ffff:127.0.0.1' } + } as ApiRequest +} + +test('auth bans behind a trusted proxy apply per client', (t) => { + const admission = createAdmission({ + trustProxy: true, + trustedProxies: ['127.0.0.1'] + }) + t.after(() => admission.destroy()) + + const attacker = proxiedRequest('198.51.100.66') + const bystander = proxiedRequest('198.51.100.10') + + for (let attempt = 0; attempt < 5; attempt++) { + admission.recordAuthFailure(admission.resolveClientAddress(attacker)) + } + + const url = new URL('http://localhost/v4/info') + const attackerDecision = admission.admit( + admission.resolveContext(attacker, url) + ) + const bystanderDecision = admission.admit( + admission.resolveContext(bystander, url) + ) + + assert.equal(attackerDecision.allowed, false) + assert.equal(attackerDecision.status, 403) + assert.equal(bystanderDecision.allowed, true) + + // The proxy itself keeps accepting connections. + assert.equal(admission.admitConnection('::ffff:127.0.0.1'), true) + admission.releaseConnection('::ffff:127.0.0.1') +}) + +test('trusted proxies use an aggregate pool that is released on close', (t) => { + const admission = createAdmission({ + trustProxy: true, + trustedProxies: ['127.0.0.1'], + ip: { maxConcurrentSockets: 2, maxProxySockets: 3 } as AdmissionConfig['ip'] + }) + t.after(() => admission.destroy()) + + const proxy = '::ffff:127.0.0.1' + for (let index = 0; index < 3; index++) { + assert.equal(admission.admitConnection(proxy), true) + } + assert.equal(admission.admitConnection(proxy), false) + + admission.releaseConnection(proxy) + assert.equal(admission.admitConnection(proxy), true) + + // Direct peers keep their per-IP limit. + assert.equal(admission.admitConnection('203.0.113.7'), true) + assert.equal(admission.admitConnection('203.0.113.7'), true) + assert.equal(admission.admitConnection('203.0.113.7'), false) +}) + +test('trustProxy without trustedProxies ignores forwarding headers', (t) => { + const admission = createAdmission({ trustProxy: true, trustedProxies: [] }) + t.after(() => admission.destroy()) + + const request = proxiedRequest('198.51.100.66') + 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 000000000..3d9ac3f71 --- /dev/null +++ b/src/utils/clientAddress.ts @@ -0,0 +1,238 @@ +import net from 'node:net' + +type HeaderBag = Record + +interface AddressRange { + family: 4 | 6 + base: bigint + mask: bigint +} + +/** + * Matches peer addresses against an explicit list of trusted proxy IPs/CIDRs. + * Invalid entries are reported through `invalidEntries` and never match. + * @public + */ +export class TrustedProxyList { + private readonly ranges: AddressRange[] + readonly invalidEntries: string[] + + constructor(entries: readonly string[] = []) { + this.ranges = [] + this.invalidEntries = [] + + for (const entry of entries) { + const range = parseRange(entry) + if (range) { + this.ranges.push(range) + } else { + this.invalidEntries.push(entry) + } + } + } + + get size(): number { + return this.ranges.length + } + + /** + * 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.ranges.length === 0) return false + + const address = normalizeAddress(rawAddress) + if (!address) return false + + const parsed = parseAddress(address) + if (!parsed) return false + + return this.ranges.some( + (range) => + range.family === parsed.family && + (parsed.value & range.mask) === range.base + ) + } +} + +/** + * 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 +} + +function parseRange(entry: string): AddressRange | null { + const [rawAddress, rawPrefix, ...rest] = entry.trim().split('/') + if (rest.length > 0) return null + + const address = normalizeAddress(rawAddress) + if (!address) return null + + const parsed = parseAddress(address) + if (!parsed) return null + + const bits = parsed.family === 4 ? 32 : 128 + const prefix = rawPrefix === undefined ? bits : Number(rawPrefix) + const isValidPrefix = + rawPrefix === undefined || + (/^\d+$/.test(rawPrefix) && prefix >= 0 && prefix <= bits) + if (!isValidPrefix) return null + + const hostBits = BigInt(bits - prefix) + const full = (1n << BigInt(bits)) - 1n + const mask = (full >> hostBits) << hostBits + + return { family: parsed.family, base: parsed.value & mask, mask } +} + +function parseAddress( + address: string +): { family: 4 | 6; value: bigint } | null { + const family = net.isIP(address) + + if (family === 4) { + const value = address + .split('.') + .reduce((acc, octet) => (acc << 8n) | BigInt(Number(octet)), 0n) + return { family: 4, value } + } + + if (family === 6) { + const groups = expandIpv6(address) + if (!groups) return null + const value = groups.reduce( + (acc, group) => (acc << 16n) | BigInt(group), + 0n + ) + return { family: 6, value } + } + + return null +} + +function expandIpv6(address: string): number[] | null { + let source = address + const embeddedIpv4 = source.match(/(\d+\.\d+\.\d+\.\d+)$/) + if (embeddedIpv4?.[1]) { + const octets = embeddedIpv4[1].split('.').map(Number) + const high = ((octets[0] ?? 0) << 8) | (octets[1] ?? 0) + const low = ((octets[2] ?? 0) << 8) | (octets[3] ?? 0) + source = `${source.slice(0, -embeddedIpv4[1].length)}${high.toString(16)}:${low.toString(16)}` + } + + const [head = '', tail] = source.split('::') + const headGroups = head ? head.split(':') : [] + const tailGroups = tail ? tail.split(':') : [] + const missing = 8 - headGroups.length - tailGroups.length + if (tail === undefined ? missing !== 0 : missing < 0) return null + + const groups = [ + ...headGroups, + ...new Array(tail === undefined ? 0 : missing).fill('0'), + ...tailGroups + ].map((group) => Number.parseInt(group, 16)) + + return groups.length === 8 && groups.every((g) => g >= 0 && g <= 0xffff) + ? groups + : null +} From 17786c9873f48c297eaa071700f1ce9e99c03666 Mon Sep 17 00:00:00 2001 From: JustEvil <71156616+EvilG-MC@users.noreply.github.com> Date: Sat, 10 Oct 2026 08:02:07 -0600 Subject: [PATCH 2/6] update: compile proxy client isolation --- dist/config.default.js | 1 + dist/src/api/index.js | 10 +- dist/src/api/profiler.file.js | 10 +- dist/src/api/profiler.js | 6 +- dist/src/api/profiler.ui.js | 6 +- dist/src/api/workers.js | 10 +- dist/src/managers/admissionManager.js | 100 ++++++++++--- dist/src/server/bunServer.js | 19 +-- dist/src/server/httpServer.js | 11 +- dist/src/server/wsRouter.js | 33 +++-- dist/src/utils/clientAddress.js | 194 ++++++++++++++++++++++++++ dist/src/utils/clientAddress.test.js | 117 ++++++++++++++++ 12 files changed, 448 insertions(+), 69 deletions(-) create mode 100644 dist/src/utils/clientAddress.js create mode 100644 dist/src/utils/clientAddress.test.js diff --git a/dist/config.default.js b/dist/config.default.js index c8b50dbfc..794c106ce 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 83e31be7b..22a4cea63 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 69d9788ab..abab63405 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 551613647..bb0326abf 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 0e8abbe9a..283bd2672 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 1c42917ef..6924d5b7d 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 c33e5753e..db43f8c4c 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,62 @@ 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); + } + } /** * Tracks an incoming TCP socket. Returns false if IP connection pool is exhausted. * @param rawAddress - Remote IP address. @@ -878,26 +940,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 606ab3ba6..6e8e98136 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) { diff --git a/dist/src/server/httpServer.js b/dist/src/server/httpServer.js index f52b2b82d..f4deb3375 100644 --- a/dist/src/server/httpServer.js +++ b/dist/src/server/httpServer.js @@ -25,18 +25,13 @@ 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) { - socket.destroy(); - return; - } - const socketAllowed = nodelink.admissionManager.incrementActiveSockets(remoteAddress); - if (!socketAllowed) { + const connectionAllowed = nodelink.admissionManager.admitConnection(remoteAddress); + if (!connectionAllowed) { socket.destroy(); return; } socket.on('close', () => { - nodelink.admissionManager.decrementActiveSockets(remoteAddress); + nodelink.admissionManager.releaseConnection(remoteAddress); }); socket.on('error', (err) => { const isBenign = err?.code === 'EPIPE' || err?.code === 'ECONNRESET'; diff --git a/dist/src/server/wsRouter.js b/dist/src/server/wsRouter.js index e16b043da..d497d0816 100644 --- a/dist/src/server/wsRouter.js +++ b/dist/src/server/wsRouter.js @@ -2,17 +2,12 @@ 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 { 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 +46,30 @@ 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; } + /* INFO: Proxied clients share the proxy's TCP pool, so cap each client's upgraded sockets here */ + if (isProxied && admissionContext.ip) { + const clientIp = admissionContext.ip; + if (!admission.incrementActiveSockets(clientIp)) { + _rejectUpgrade(socket, 429, 'Too Many Requests', 'Too many concurrent connections.'); + return; + } + socket.once('close', () => admission.decrementActiveSockets(clientIp)); + } if (pathname === '/v4/profiler/socket') { _handleProfilerUpgrade(context, request, socket, head, clientAddress, isInternal); return; @@ -107,7 +114,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; } diff --git a/dist/src/utils/clientAddress.js b/dist/src/utils/clientAddress.js new file mode 100644 index 000000000..429f21cc1 --- /dev/null +++ b/dist/src/utils/clientAddress.js @@ -0,0 +1,194 @@ +import net from 'node:net'; +/** + * Matches peer addresses against an explicit list of trusted proxy IPs/CIDRs. + * Invalid entries are reported through `invalidEntries` and never match. + * @public + */ +export class TrustedProxyList { + ranges; + invalidEntries; + constructor(entries = []) { + this.ranges = []; + this.invalidEntries = []; + for (const entry of entries) { + const range = parseRange(entry); + if (range) { + this.ranges.push(range); + } + else { + this.invalidEntries.push(entry); + } + } + } + get size() { + return this.ranges.length; + } + /** + * Checks whether an address belongs to a trusted proxy. + * @param rawAddress - Peer address (IPv4, IPv6 or IPv4-mapped IPv6). + */ + contains(rawAddress) { + if (this.ranges.length === 0) + return false; + const address = normalizeAddress(rawAddress); + if (!address) + return false; + const parsed = parseAddress(address); + if (!parsed) + return false; + return this.ranges.some((range) => range.family === parsed.family && + (parsed.value & range.mask) === range.base); + } +} +/** + * 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; +} +function parseRange(entry) { + const [rawAddress, rawPrefix, ...rest] = entry.trim().split('/'); + if (rest.length > 0) + return null; + const address = normalizeAddress(rawAddress); + if (!address) + return null; + const parsed = parseAddress(address); + if (!parsed) + return null; + const bits = parsed.family === 4 ? 32 : 128; + const prefix = rawPrefix === undefined ? bits : Number(rawPrefix); + const isValidPrefix = rawPrefix === undefined || + (/^\d+$/.test(rawPrefix) && prefix >= 0 && prefix <= bits); + if (!isValidPrefix) + return null; + const hostBits = BigInt(bits - prefix); + const full = (1n << BigInt(bits)) - 1n; + const mask = (full >> hostBits) << hostBits; + return { family: parsed.family, base: parsed.value & mask, mask }; +} +function parseAddress(address) { + const family = net.isIP(address); + if (family === 4) { + const value = address + .split('.') + .reduce((acc, octet) => (acc << 8n) | BigInt(Number(octet)), 0n); + return { family: 4, value }; + } + if (family === 6) { + const groups = expandIpv6(address); + if (!groups) + return null; + const value = groups.reduce((acc, group) => (acc << 16n) | BigInt(group), 0n); + return { family: 6, value }; + } + return null; +} +function expandIpv6(address) { + let source = address; + const embeddedIpv4 = source.match(/(\d+\.\d+\.\d+\.\d+)$/); + if (embeddedIpv4?.[1]) { + const octets = embeddedIpv4[1].split('.').map(Number); + const high = ((octets[0] ?? 0) << 8) | (octets[1] ?? 0); + const low = ((octets[2] ?? 0) << 8) | (octets[3] ?? 0); + source = `${source.slice(0, -embeddedIpv4[1].length)}${high.toString(16)}:${low.toString(16)}`; + } + const [head = '', tail] = source.split('::'); + const headGroups = head ? head.split(':') : []; + const tailGroups = tail ? tail.split(':') : []; + const missing = 8 - headGroups.length - tailGroups.length; + if (tail === undefined ? missing !== 0 : missing < 0) + return null; + const groups = [ + ...headGroups, + ...new Array(tail === undefined ? 0 : missing).fill('0'), + ...tailGroups + ].map((group) => Number.parseInt(group, 16)); + return groups.length === 8 && groups.every((g) => g >= 0 && g <= 0xffff) + ? groups + : null; +} diff --git a/dist/src/utils/clientAddress.test.js b/dist/src/utils/clientAddress.test.js new file mode 100644 index 000000000..497fe2f5c --- /dev/null +++ b/dist/src/utils/clientAddress.test.js @@ -0,0 +1,117 @@ +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('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); +}); +function createAdmission(config) { + const nodelink = { + options: { server: { password: 'test-password' } } + }; + return new AdmissionManager(nodelink, config); +} +function proxiedRequest(clientIp) { + return { + method: 'GET', + url: '/v4/info', + headers: { 'x-forwarded-for': clientIp }, + socket: { remoteAddress: '::ffff:127.0.0.1' } + }; +} +test('auth bans behind a trusted proxy apply per client', (t) => { + const admission = createAdmission({ + trustProxy: true, + trustedProxies: ['127.0.0.1'] + }); + t.after(() => admission.destroy()); + const attacker = proxiedRequest('198.51.100.66'); + const bystander = proxiedRequest('198.51.100.10'); + for (let attempt = 0; attempt < 5; attempt++) { + admission.recordAuthFailure(admission.resolveClientAddress(attacker)); + } + const url = new URL('http://localhost/v4/info'); + const attackerDecision = admission.admit(admission.resolveContext(attacker, url)); + const bystanderDecision = admission.admit(admission.resolveContext(bystander, url)); + assert.equal(attackerDecision.allowed, false); + assert.equal(attackerDecision.status, 403); + assert.equal(bystanderDecision.allowed, true); + // The proxy itself keeps accepting connections. + assert.equal(admission.admitConnection('::ffff:127.0.0.1'), true); + admission.releaseConnection('::ffff:127.0.0.1'); +}); +test('trusted proxies use an aggregate pool that is released on close', (t) => { + const admission = createAdmission({ + trustProxy: true, + trustedProxies: ['127.0.0.1'], + ip: { maxConcurrentSockets: 2, maxProxySockets: 3 } + }); + t.after(() => admission.destroy()); + const proxy = '::ffff:127.0.0.1'; + for (let index = 0; index < 3; index++) { + assert.equal(admission.admitConnection(proxy), true); + } + assert.equal(admission.admitConnection(proxy), false); + admission.releaseConnection(proxy); + assert.equal(admission.admitConnection(proxy), true); + // Direct peers keep their per-IP limit. + assert.equal(admission.admitConnection('203.0.113.7'), true); + assert.equal(admission.admitConnection('203.0.113.7'), true); + assert.equal(admission.admitConnection('203.0.113.7'), false); +}); +test('trustProxy without trustedProxies ignores forwarding headers', (t) => { + const admission = createAdmission({ trustProxy: true, trustedProxies: [] }); + t.after(() => admission.destroy()); + const request = proxiedRequest('198.51.100.66'); + assert.equal(admission.resolveClientAddress(request), '127.0.0.1'); + assert.equal(admission.isTrustedProxy('127.0.0.1'), false); +}); From 46f50e64eb811171522880bd5d701eec56ff8b4a Mon Sep 17 00:00:00 2001 From: JustEvil <71156616+EvilG-MC@users.noreply.github.com> Date: Sat, 10 Oct 2026 08:19:52 -0600 Subject: [PATCH 3/6] fix: release socket capacity when websockets close PWSL calls removeAllListeners() on the TCP socket right after ending or destroying it, so 'close' listeners never ran for upgraded sockets. Every closed WebSocket kept its per-IP slot, and with this branch also its per-client and trusted proxy slots, until clients were refused. - Track releases per socket and run them once, from the WebSocket 'close' event, its destroy() (protocol errors and server teardown emit no 'close') and TCP 'close' as a fallback - Reserve peer and per-client capacity for Bun upgrades, which have no TCP connection hook, and release it on failed upgrade or close --- src/managers/admissionManager.ts | 37 +++++++ src/server/bunServer.ts | 22 +++- src/server/httpServer.ts | 5 +- src/server/socketRelease.test.ts | 173 +++++++++++++++++++++++++++++++ src/server/socketRelease.ts | 65 ++++++++++++ src/server/wsRouter.ts | 30 +++--- src/typings/index.types.ts | 6 ++ 7 files changed, 322 insertions(+), 16 deletions(-) create mode 100644 src/server/socketRelease.test.ts create mode 100644 src/server/socketRelease.ts diff --git a/src/managers/admissionManager.ts b/src/managers/admissionManager.ts index f34cb0936..3dfb026cd 100644 --- a/src/managers/admissionManager.ts +++ b/src/managers/admissionManager.ts @@ -345,6 +345,43 @@ export default class AdmissionManager { } } + /** + * 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. diff --git a/src/server/bunServer.ts b/src/server/bunServer.ts index 118bafbc1..16a8dea20 100644 --- a/src/server/bunServer.ts +++ b/src/server/bunServer.ts @@ -381,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, @@ -390,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: { @@ -624,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/httpServer.ts b/src/server/httpServer.ts index ba9ac3fa1..424983c5e 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 @@ -50,9 +51,9 @@ function createHttpServer( return } - socket.on('close', () => { + 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.test.ts b/src/server/socketRelease.test.ts new file mode 100644 index 000000000..4b5ccd9dd --- /dev/null +++ b/src/server/socketRelease.test.ts @@ -0,0 +1,173 @@ +import assert from 'node:assert/strict' +import type http from 'node:http' +import type { AddressInfo } from 'node:net' +import test from 'node:test' +import WebSocketServer from '@performanc/pwsl-server' + +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' +import { setupWebSocketEvents } from './wsRouter.ts' + +const PASSWORD = 'test-password' + +interface Harness { + url: string + admission: AdmissionManager + socketBus: WebSocketServer + close: () => Promise +} + +async function startServer(config: Partial): Promise { + const options = { + server: { password: PASSWORD }, + playback: { voiceReceive: { enabled: false } }, + cluster: {} + } + 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: () => {} } + } as unknown as NodelinkServer + + setupWebSocketEvents(context) + socketBus.removeAllListeners('/v4/websocket') + + const server = createHttpServer(context, () => + Promise.reject(new Error('REST is not used in this test')) + ) as http.Server + 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`, + admission, + socketBus, + close: async () => { + admission.destroy() + server.closeAllConnections() + await new Promise((resolve) => server.close(() => resolve())) + } + } +} + +function connect( + url: string, + forwardedFor?: string +): 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 }) + ) + } +}) diff --git a/src/server/socketRelease.ts b/src/server/socketRelease.ts new file mode 100644 index 000000000..a9f4bdac7 --- /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. + * @public + */ +export 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 dd760779e..ad12fede5 100644 --- a/src/server/wsRouter.ts +++ b/src/server/wsRouter.ts @@ -11,6 +11,7 @@ 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]+)\/?$/ @@ -97,20 +98,21 @@ function handleHttpUpgrade( return } - /* INFO: Proxied clients share the proxy's TCP pool, so cap each client's upgraded sockets here */ - if (isProxied && admissionContext.ip) { - const clientIp = admissionContext.ip - if (!admission.incrementActiveSockets(clientIp)) { - _rejectUpgrade( - socket, - 429, - 'Too Many Requests', - 'Too many concurrent connections.' - ) - return - } - socket.once('close', () => admission.decrementActiveSockets(clientIp)) + 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( @@ -173,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, @@ -268,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/index.types.ts b/src/typings/index.types.ts index 385af9040..f11a039e9 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 } /** From 83e843343abf4282f99a48feb1c6b41d8a5a93f8 Mon Sep 17 00:00:00 2001 From: JustEvil <71156616+EvilG-MC@users.noreply.github.com> Date: Sat, 10 Oct 2026 08:20:17 -0600 Subject: [PATCH 4/6] update: compile websocket socket release --- dist/src/managers/admissionManager.js | 30 +++++++ dist/src/server/bunServer.js | 17 +++- dist/src/server/httpServer.js | 5 +- dist/src/server/socketRelease.js | 51 +++++++++++ dist/src/server/socketRelease.test.js | 122 ++++++++++++++++++++++++++ dist/src/server/wsRouter.js | 16 ++-- 6 files changed, 229 insertions(+), 12 deletions(-) create mode 100644 dist/src/server/socketRelease.js create mode 100644 dist/src/server/socketRelease.test.js diff --git a/dist/src/managers/admissionManager.js b/dist/src/managers/admissionManager.js index db43f8c4c..a50020537 100644 --- a/dist/src/managers/admissionManager.js +++ b/dist/src/managers/admissionManager.js @@ -269,6 +269,36 @@ export default class AdmissionManager { 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. diff --git a/dist/src/server/bunServer.js b/dist/src/server/bunServer.js index 6e8e98136..ddabf6085 100644 --- a/dist/src/server/bunServer.js +++ b/dist/src/server/bunServer.js @@ -287,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, @@ -296,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: { @@ -478,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/httpServer.js b/dist/src/server/httpServer.js index f4deb3375..a13426023 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) { @@ -30,9 +31,7 @@ function createHttpServer(nodelink, getRequestHandler) { socket.destroy(); return; } - socket.on('close', () => { - nodelink.admissionManager.releaseConnection(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 000000000..f5a5e92f6 --- /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. + * @public + */ +export 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/socketRelease.test.js b/dist/src/server/socketRelease.test.js new file mode 100644 index 000000000..22a008231 --- /dev/null +++ b/dist/src/server/socketRelease.test.js @@ -0,0 +1,122 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import WebSocketServer from '@performanc/pwsl-server'; +import AdmissionManager from '../managers/admissionManager.js'; +import { createHttpServer } from './httpServer.js'; +import { setupWebSocketEvents } from './wsRouter.js'; +const PASSWORD = 'test-password'; +async function startServer(config) { + const options = { + server: { password: PASSWORD }, + playback: { voiceReceive: { enabled: false } }, + cluster: {} + }; + const admission = new AdmissionManager({ options }, config); + const socketBus = new WebSocketServer(); + const context = { + options, + admissionManager: admission, + socket: socketBus, + sessions: { isResumable: () => false }, + pluginManager: { callHook: () => { } } + }; + setupWebSocketEvents(context); + socketBus.removeAllListeners('/v4/websocket'); + const server = createHttpServer(context, () => Promise.reject(new Error('REST is not used in this test'))); + 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`, + admission, + socketBus, + close: async () => { + admission.destroy(); + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + } + }; +} +function connect(url, forwardedFor) { + 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 })); + } +}); diff --git a/dist/src/server/wsRouter.js b/dist/src/server/wsRouter.js index d497d0816..4aff35595 100644 --- a/dist/src/server/wsRouter.js +++ b/dist/src/server/wsRouter.js @@ -4,6 +4,7 @@ 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\/([^/]+)\/?$/; @@ -61,15 +62,12 @@ function handleHttpUpgrade(context, request, socket, head) { _rejectUpgrade(socket, admissionDecision.status, 'Too Many Requests', admissionDecision.message); return; } - /* INFO: Proxied clients share the proxy's TCP pool, so cap each client's upgraded sockets here */ - if (isProxied && admissionContext.ip) { - const clientIp = admissionContext.ip; - if (!admission.incrementActiveSockets(clientIp)) { - _rejectUpgrade(socket, 429, 'Too Many Requests', 'Too many concurrent connections.'); - return; - } - socket.once('close', () => admission.decrementActiveSockets(clientIp)); + 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; @@ -103,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); }); } @@ -159,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); }); } From 08286424798cb210a512f0ad95359a3a52aea59b Mon Sep 17 00:00:00 2001 From: JustEvil <71156616+EvilG-MC@users.noreply.github.com> Date: Sat, 10 Oct 2026 08:44:15 -0600 Subject: [PATCH 5/6] improve: simplify proxy matching and test real transports - Match trusted proxies with net.BlockList instead of a hand-rolled CIDR parser, keeping IPv4 and IPv6 rules in separate lists so an address only matches rules of its own family - Keep releaseSocket private to socketRelease.ts - Drive the auth isolation tests through the real REST handler and WebSocket upgrade instead of feeding admission the client IP - Drop the duplicated aggregate pool unit test and the unused session handlers from the transport harness, and destroy upgraded sockets on teardown so a failing test cannot hang - Add a Bun-only regression for upgrade capacity (bun test) --- src/server/bunServer.test.ts | 148 ++++++++++++++++++ ...elease.test.ts => clientIsolation.test.ts} | 90 +++++++++-- src/server/socketRelease.ts | 4 +- src/utils/clientAddress.test.ts | 87 +++------- src/utils/clientAddress.ts | 138 +++++----------- 5 files changed, 281 insertions(+), 186 deletions(-) create mode 100644 src/server/bunServer.test.ts rename src/server/{socketRelease.test.ts => clientIsolation.test.ts} (65%) diff --git a/src/server/bunServer.test.ts b/src/server/bunServer.test.ts new file mode 100644 index 000000000..38e2bea02 --- /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/socketRelease.test.ts b/src/server/clientIsolation.test.ts similarity index 65% rename from src/server/socketRelease.test.ts rename to src/server/clientIsolation.test.ts index 4b5ccd9dd..1cd4b0818 100644 --- a/src/server/socketRelease.test.ts +++ b/src/server/clientIsolation.test.ts @@ -1,20 +1,20 @@ import assert from 'node:assert/strict' import type http from 'node:http' -import type { AddressInfo } from 'node:net' +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' -import { setupWebSocketEvents } from './wsRouter.ts' const PASSWORD = 'test-password' interface Harness { url: string - admission: AdmissionManager + restUrl: string socketBus: WebSocketServer close: () => Promise } @@ -23,7 +23,8 @@ async function startServer(config: Partial): Promise { const options = { server: { password: PASSWORD }, playback: { voiceReceive: { enabled: false } }, - cluster: {} + cluster: {}, + api: {} } const admission = new AdmissionManager( { options } as unknown as ConstructorParameters[0], @@ -35,25 +36,30 @@ async function startServer(config: Partial): Promise { admissionManager: admission, socket: socketBus, sessions: { isResumable: () => false }, - pluginManager: { callHook: () => {} } + pluginManager: { callHook: () => {} }, + statsManager: { + incrementApiRequest: () => {}, + recordHttpRequestDuration: () => {} + }, + extensions: { middlewares: [], routes: [] } } as unknown as NodelinkServer - setupWebSocketEvents(context) - socketBus.removeAllListeners('/v4/websocket') - const server = createHttpServer(context, () => - Promise.reject(new Error('REST is not used in this test')) + 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`, - admission, + restUrl: `http://127.0.0.1:${port}/v4/nodelink-test-route`, socketBus, close: async () => { admission.destroy() - server.closeAllConnections() + for (const socket of sockets) socket.destroy() await new Promise((resolve) => server.close(() => resolve())) } } @@ -61,10 +67,11 @@ async function startServer(config: Partial): Promise { function connect( url: string, - forwardedFor?: string + forwardedFor?: string, + password = PASSWORD ): Promise { const headers: Record = { - Authorization: PASSWORD, + Authorization: password, 'Client-Name': 'release-test/1.0.0', 'User-Id': '123456789012345678' } @@ -171,3 +178,60 @@ test('server-side destroy releases the slot', async (t) => { ) } }) + +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/socketRelease.ts b/src/server/socketRelease.ts index a9f4bdac7..172eaf610 100644 --- a/src/server/socketRelease.ts +++ b/src/server/socketRelease.ts @@ -29,9 +29,9 @@ export function trackSocketRelease( /** * Runs and clears every pending release for the socket. Idempotent. * @param socket - TCP socket holding the capacity. - * @public + * @internal */ -export function releaseSocket(socket: NetSocket): void { +function releaseSocket(socket: NetSocket): void { const releases = pendingReleases.get(socket) if (!releases) return diff --git a/src/utils/clientAddress.test.ts b/src/utils/clientAddress.test.ts index f2a1fc4f2..8885a41ee 100644 --- a/src/utils/clientAddress.test.ts +++ b/src/utils/clientAddress.test.ts @@ -2,7 +2,6 @@ import assert from 'node:assert/strict' import test from 'node:test' import AdmissionManager from '../managers/admissionManager.ts' -import type { AdmissionConfig } from '../typings/admission/admission.types.ts' import type { ApiRequest } from '../typings/api/api.types.ts' import { isLoopbackRequest, @@ -39,6 +38,17 @@ test('TrustedProxyList matches IPv4/IPv6 CIDRs and rejects bad entries', () => { 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', @@ -118,81 +128,22 @@ test('isLoopbackRequest rejects proxied loopback traffic', () => { assert.equal(isLoopbackRequest('198.51.100.1', {}), false) }) -function createAdmission(config: Partial): AdmissionManager { +test('trustProxy without trustedProxies ignores forwarding headers', (t) => { const nodelink = { options: { server: { password: 'test-password' } } } as unknown as ConstructorParameters[0] - return new AdmissionManager(nodelink, config) -} + const admission = new AdmissionManager(nodelink, { + trustProxy: true, + trustedProxies: [] + }) + t.after(() => admission.destroy()) -function proxiedRequest(clientIp: string): ApiRequest { - return { + const request = { method: 'GET', url: '/v4/info', - headers: { 'x-forwarded-for': clientIp }, + headers: { 'x-forwarded-for': '198.51.100.66' }, socket: { remoteAddress: '::ffff:127.0.0.1' } } as ApiRequest -} - -test('auth bans behind a trusted proxy apply per client', (t) => { - const admission = createAdmission({ - trustProxy: true, - trustedProxies: ['127.0.0.1'] - }) - t.after(() => admission.destroy()) - - const attacker = proxiedRequest('198.51.100.66') - const bystander = proxiedRequest('198.51.100.10') - - for (let attempt = 0; attempt < 5; attempt++) { - admission.recordAuthFailure(admission.resolveClientAddress(attacker)) - } - - const url = new URL('http://localhost/v4/info') - const attackerDecision = admission.admit( - admission.resolveContext(attacker, url) - ) - const bystanderDecision = admission.admit( - admission.resolveContext(bystander, url) - ) - - assert.equal(attackerDecision.allowed, false) - assert.equal(attackerDecision.status, 403) - assert.equal(bystanderDecision.allowed, true) - - // The proxy itself keeps accepting connections. - assert.equal(admission.admitConnection('::ffff:127.0.0.1'), true) - admission.releaseConnection('::ffff:127.0.0.1') -}) - -test('trusted proxies use an aggregate pool that is released on close', (t) => { - const admission = createAdmission({ - trustProxy: true, - trustedProxies: ['127.0.0.1'], - ip: { maxConcurrentSockets: 2, maxProxySockets: 3 } as AdmissionConfig['ip'] - }) - t.after(() => admission.destroy()) - - const proxy = '::ffff:127.0.0.1' - for (let index = 0; index < 3; index++) { - assert.equal(admission.admitConnection(proxy), true) - } - assert.equal(admission.admitConnection(proxy), false) - - admission.releaseConnection(proxy) - assert.equal(admission.admitConnection(proxy), true) - - // Direct peers keep their per-IP limit. - assert.equal(admission.admitConnection('203.0.113.7'), true) - assert.equal(admission.admitConnection('203.0.113.7'), true) - assert.equal(admission.admitConnection('203.0.113.7'), false) -}) - -test('trustProxy without trustedProxies ignores forwarding headers', (t) => { - const admission = createAdmission({ trustProxy: true, trustedProxies: [] }) - t.after(() => admission.destroy()) - - const request = proxiedRequest('198.51.100.66') 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 index 3d9ac3f71..4e47fbc0d 100644 --- a/src/utils/clientAddress.ts +++ b/src/utils/clientAddress.ts @@ -2,57 +2,65 @@ import net from 'node:net' type HeaderBag = Record -interface AddressRange { - family: 4 | 6 - base: bigint - mask: bigint -} - /** * Matches peer addresses against an explicit list of trusted proxy IPs/CIDRs. - * Invalid entries are reported through `invalidEntries` and never match. + * 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 ranges: AddressRange[] - readonly invalidEntries: string[] + private readonly ipv4 = new net.BlockList() + private readonly ipv6 = new net.BlockList() + readonly invalidEntries: string[] = [] + readonly size: number = 0 constructor(entries: readonly string[] = []) { - this.ranges = [] - this.invalidEntries = [] - for (const entry of entries) { - const range = parseRange(entry) - if (range) { - this.ranges.push(range) + if (this.add(entry)) { + this.size += 1 } else { this.invalidEntries.push(entry) } } } - get size(): number { - return this.ranges.length - } - /** * 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.ranges.length === 0) return false + if (this.size === 0) return false const address = normalizeAddress(rawAddress) if (!address) return false - const parsed = parseAddress(address) - if (!parsed) 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 - return this.ranges.some( - (range) => - range.family === parsed.family && - (parsed.value & range.mask) === range.base - ) + 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 } } @@ -160,79 +168,3 @@ function readHeader(headers: HeaderBag, name: string): string | undefined { const value = Array.isArray(raw) ? raw.join(',') : raw return value?.trim() || undefined } - -function parseRange(entry: string): AddressRange | null { - const [rawAddress, rawPrefix, ...rest] = entry.trim().split('/') - if (rest.length > 0) return null - - const address = normalizeAddress(rawAddress) - if (!address) return null - - const parsed = parseAddress(address) - if (!parsed) return null - - const bits = parsed.family === 4 ? 32 : 128 - const prefix = rawPrefix === undefined ? bits : Number(rawPrefix) - const isValidPrefix = - rawPrefix === undefined || - (/^\d+$/.test(rawPrefix) && prefix >= 0 && prefix <= bits) - if (!isValidPrefix) return null - - const hostBits = BigInt(bits - prefix) - const full = (1n << BigInt(bits)) - 1n - const mask = (full >> hostBits) << hostBits - - return { family: parsed.family, base: parsed.value & mask, mask } -} - -function parseAddress( - address: string -): { family: 4 | 6; value: bigint } | null { - const family = net.isIP(address) - - if (family === 4) { - const value = address - .split('.') - .reduce((acc, octet) => (acc << 8n) | BigInt(Number(octet)), 0n) - return { family: 4, value } - } - - if (family === 6) { - const groups = expandIpv6(address) - if (!groups) return null - const value = groups.reduce( - (acc, group) => (acc << 16n) | BigInt(group), - 0n - ) - return { family: 6, value } - } - - return null -} - -function expandIpv6(address: string): number[] | null { - let source = address - const embeddedIpv4 = source.match(/(\d+\.\d+\.\d+\.\d+)$/) - if (embeddedIpv4?.[1]) { - const octets = embeddedIpv4[1].split('.').map(Number) - const high = ((octets[0] ?? 0) << 8) | (octets[1] ?? 0) - const low = ((octets[2] ?? 0) << 8) | (octets[3] ?? 0) - source = `${source.slice(0, -embeddedIpv4[1].length)}${high.toString(16)}:${low.toString(16)}` - } - - const [head = '', tail] = source.split('::') - const headGroups = head ? head.split(':') : [] - const tailGroups = tail ? tail.split(':') : [] - const missing = 8 - headGroups.length - tailGroups.length - if (tail === undefined ? missing !== 0 : missing < 0) return null - - const groups = [ - ...headGroups, - ...new Array(tail === undefined ? 0 : missing).fill('0'), - ...tailGroups - ].map((group) => Number.parseInt(group, 16)) - - return groups.length === 8 && groups.every((g) => g >= 0 && g <= 0xffff) - ? groups - : null -} From 4b45029a7ae182e2a87db4d073801e4a66b0019e Mon Sep 17 00:00:00 2001 From: JustEvil <71156616+EvilG-MC@users.noreply.github.com> Date: Sat, 10 Oct 2026 08:44:55 -0600 Subject: [PATCH 6/6] update: compile simplified proxy matching and tests --- dist/src/server/bunServer.test.js | 112 ++++++++++++++++++ ...elease.test.js => clientIsolation.test.js} | 64 ++++++++-- dist/src/server/socketRelease.js | 4 +- dist/src/utils/clientAddress.js | 112 ++++++------------ dist/src/utils/clientAddress.test.js | 68 +++-------- 5 files changed, 219 insertions(+), 141 deletions(-) create mode 100644 dist/src/server/bunServer.test.js rename dist/src/server/{socketRelease.test.js => clientIsolation.test.js} (62%) diff --git a/dist/src/server/bunServer.test.js b/dist/src/server/bunServer.test.js new file mode 100644 index 000000000..e98873650 --- /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/socketRelease.test.js b/dist/src/server/clientIsolation.test.js similarity index 62% rename from dist/src/server/socketRelease.test.js rename to dist/src/server/clientIsolation.test.js index 22a008231..9be631cc0 100644 --- a/dist/src/server/socketRelease.test.js +++ b/dist/src/server/clientIsolation.test.js @@ -1,15 +1,16 @@ 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'; -import { setupWebSocketEvents } from './wsRouter.js'; const PASSWORD = 'test-password'; async function startServer(config) { const options = { server: { password: PASSWORD }, playback: { voiceReceive: { enabled: false } }, - cluster: {} + cluster: {}, + api: {} }; const admission = new AdmissionManager({ options }, config); const socketBus = new WebSocketServer(); @@ -18,27 +19,34 @@ async function startServer(config) { admissionManager: admission, socket: socketBus, sessions: { isResumable: () => false }, - pluginManager: { callHook: () => { } } + pluginManager: { callHook: () => { } }, + statsManager: { + incrementApiRequest: () => { }, + recordHttpRequestDuration: () => { } + }, + extensions: { middlewares: [], routes: [] } }; - setupWebSocketEvents(context); - socketBus.removeAllListeners('/v4/websocket'); - const server = createHttpServer(context, () => Promise.reject(new Error('REST is not used in this test'))); + 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`, - admission, + restUrl: `http://127.0.0.1:${port}/v4/nodelink-test-route`, socketBus, close: async () => { admission.destroy(); - server.closeAllConnections(); + for (const socket of sockets) + socket.destroy(); await new Promise((resolve) => server.close(() => resolve())); } }; } -function connect(url, forwardedFor) { +function connect(url, forwardedFor, password = PASSWORD) { const headers = { - Authorization: PASSWORD, + Authorization: password, 'Client-Name': 'release-test/1.0.0', 'User-Id': '123456789012345678' }; @@ -120,3 +128,39 @@ test('server-side destroy releases the slot', async (t) => { : 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/socketRelease.js b/dist/src/server/socketRelease.js index f5a5e92f6..9b16f138f 100644 --- a/dist/src/server/socketRelease.js +++ b/dist/src/server/socketRelease.js @@ -20,9 +20,9 @@ export function trackSocketRelease(socket, release) { /** * Runs and clears every pending release for the socket. Idempotent. * @param socket - TCP socket holding the capacity. - * @public + * @internal */ -export function releaseSocket(socket) { +function releaseSocket(socket) { const releases = pendingReleases.get(socket); if (!releases) return; diff --git a/dist/src/utils/clientAddress.js b/dist/src/utils/clientAddress.js index 429f21cc1..42ea16dda 100644 --- a/dist/src/utils/clientAddress.js +++ b/dist/src/utils/clientAddress.js @@ -1,43 +1,61 @@ import net from 'node:net'; /** * Matches peer addresses against an explicit list of trusted proxy IPs/CIDRs. - * Invalid entries are reported through `invalidEntries` and never match. + * 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 { - ranges; - invalidEntries; + ipv4 = new net.BlockList(); + ipv6 = new net.BlockList(); + invalidEntries = []; + size = 0; constructor(entries = []) { - this.ranges = []; - this.invalidEntries = []; for (const entry of entries) { - const range = parseRange(entry); - if (range) { - this.ranges.push(range); + if (this.add(entry)) { + this.size += 1; } else { this.invalidEntries.push(entry); } } } - get size() { - return this.ranges.length; - } /** * Checks whether an address belongs to a trusted proxy. * @param rawAddress - Peer address (IPv4, IPv6 or IPv4-mapped IPv6). */ contains(rawAddress) { - if (this.ranges.length === 0) + 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 parsed = parseAddress(address); - if (!parsed) + 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; - return this.ranges.some((range) => range.family === parsed.family && - (parsed.value & range.mask) === range.base); + if (isIpv4) { + this.ipv4.addSubnet(address, prefix, 'ipv4'); + } + else { + this.ipv6.addSubnet(address, prefix, 'ipv6'); + } + return true; } } /** @@ -130,65 +148,3 @@ function readHeader(headers, name) { const value = Array.isArray(raw) ? raw.join(',') : raw; return value?.trim() || undefined; } -function parseRange(entry) { - const [rawAddress, rawPrefix, ...rest] = entry.trim().split('/'); - if (rest.length > 0) - return null; - const address = normalizeAddress(rawAddress); - if (!address) - return null; - const parsed = parseAddress(address); - if (!parsed) - return null; - const bits = parsed.family === 4 ? 32 : 128; - const prefix = rawPrefix === undefined ? bits : Number(rawPrefix); - const isValidPrefix = rawPrefix === undefined || - (/^\d+$/.test(rawPrefix) && prefix >= 0 && prefix <= bits); - if (!isValidPrefix) - return null; - const hostBits = BigInt(bits - prefix); - const full = (1n << BigInt(bits)) - 1n; - const mask = (full >> hostBits) << hostBits; - return { family: parsed.family, base: parsed.value & mask, mask }; -} -function parseAddress(address) { - const family = net.isIP(address); - if (family === 4) { - const value = address - .split('.') - .reduce((acc, octet) => (acc << 8n) | BigInt(Number(octet)), 0n); - return { family: 4, value }; - } - if (family === 6) { - const groups = expandIpv6(address); - if (!groups) - return null; - const value = groups.reduce((acc, group) => (acc << 16n) | BigInt(group), 0n); - return { family: 6, value }; - } - return null; -} -function expandIpv6(address) { - let source = address; - const embeddedIpv4 = source.match(/(\d+\.\d+\.\d+\.\d+)$/); - if (embeddedIpv4?.[1]) { - const octets = embeddedIpv4[1].split('.').map(Number); - const high = ((octets[0] ?? 0) << 8) | (octets[1] ?? 0); - const low = ((octets[2] ?? 0) << 8) | (octets[3] ?? 0); - source = `${source.slice(0, -embeddedIpv4[1].length)}${high.toString(16)}:${low.toString(16)}`; - } - const [head = '', tail] = source.split('::'); - const headGroups = head ? head.split(':') : []; - const tailGroups = tail ? tail.split(':') : []; - const missing = 8 - headGroups.length - tailGroups.length; - if (tail === undefined ? missing !== 0 : missing < 0) - return null; - const groups = [ - ...headGroups, - ...new Array(tail === undefined ? 0 : missing).fill('0'), - ...tailGroups - ].map((group) => Number.parseInt(group, 16)); - return groups.length === 8 && groups.every((g) => g >= 0 && g <= 0xffff) - ? groups - : null; -} diff --git a/dist/src/utils/clientAddress.test.js b/dist/src/utils/clientAddress.test.js index 497fe2f5c..388608dfd 100644 --- a/dist/src/utils/clientAddress.test.js +++ b/dist/src/utils/clientAddress.test.js @@ -27,6 +27,15 @@ test('TrustedProxyList matches IPv4/IPv6 CIDRs and rejects bad entries', () => { 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'); @@ -54,64 +63,21 @@ test('isLoopbackRequest rejects proxied loopback traffic', () => { assert.equal(isLoopbackRequest('127.0.0.1', { forwarded: 'for=1.2.3.4' }), false); assert.equal(isLoopbackRequest('198.51.100.1', {}), false); }); -function createAdmission(config) { +test('trustProxy without trustedProxies ignores forwarding headers', (t) => { const nodelink = { options: { server: { password: 'test-password' } } }; - return new AdmissionManager(nodelink, config); -} -function proxiedRequest(clientIp) { - return { + const admission = new AdmissionManager(nodelink, { + trustProxy: true, + trustedProxies: [] + }); + t.after(() => admission.destroy()); + const request = { method: 'GET', url: '/v4/info', - headers: { 'x-forwarded-for': clientIp }, + headers: { 'x-forwarded-for': '198.51.100.66' }, socket: { remoteAddress: '::ffff:127.0.0.1' } }; -} -test('auth bans behind a trusted proxy apply per client', (t) => { - const admission = createAdmission({ - trustProxy: true, - trustedProxies: ['127.0.0.1'] - }); - t.after(() => admission.destroy()); - const attacker = proxiedRequest('198.51.100.66'); - const bystander = proxiedRequest('198.51.100.10'); - for (let attempt = 0; attempt < 5; attempt++) { - admission.recordAuthFailure(admission.resolveClientAddress(attacker)); - } - const url = new URL('http://localhost/v4/info'); - const attackerDecision = admission.admit(admission.resolveContext(attacker, url)); - const bystanderDecision = admission.admit(admission.resolveContext(bystander, url)); - assert.equal(attackerDecision.allowed, false); - assert.equal(attackerDecision.status, 403); - assert.equal(bystanderDecision.allowed, true); - // The proxy itself keeps accepting connections. - assert.equal(admission.admitConnection('::ffff:127.0.0.1'), true); - admission.releaseConnection('::ffff:127.0.0.1'); -}); -test('trusted proxies use an aggregate pool that is released on close', (t) => { - const admission = createAdmission({ - trustProxy: true, - trustedProxies: ['127.0.0.1'], - ip: { maxConcurrentSockets: 2, maxProxySockets: 3 } - }); - t.after(() => admission.destroy()); - const proxy = '::ffff:127.0.0.1'; - for (let index = 0; index < 3; index++) { - assert.equal(admission.admitConnection(proxy), true); - } - assert.equal(admission.admitConnection(proxy), false); - admission.releaseConnection(proxy); - assert.equal(admission.admitConnection(proxy), true); - // Direct peers keep their per-IP limit. - assert.equal(admission.admitConnection('203.0.113.7'), true); - assert.equal(admission.admitConnection('203.0.113.7'), true); - assert.equal(admission.admitConnection('203.0.113.7'), false); -}); -test('trustProxy without trustedProxies ignores forwarding headers', (t) => { - const admission = createAdmission({ trustProxy: true, trustedProxies: [] }); - t.after(() => admission.destroy()); - const request = proxiedRequest('198.51.100.66'); assert.equal(admission.resolveClientAddress(request), '127.0.0.1'); assert.equal(admission.isTrustedProxy('127.0.0.1'), false); });