diff --git a/packages/core/src/builtins/newznab/addon.ts b/packages/core/src/builtins/newznab/addon.ts index 0a6c34e3..e9277ba2 100644 --- a/packages/core/src/builtins/newznab/addon.ts +++ b/packages/core/src/builtins/newznab/addon.ts @@ -1,6 +1,6 @@ import { z } from 'zod'; import { ParsedId } from '../../utils/id-parser.js'; -import { constants, createLogger } from '../../utils/index.js'; +import { constants, createLogger, Env } from '../../utils/index.js'; import { Torrent, NZB } from '../../debrid/index.js'; import { SearchMetadata } from '../base/debrid.js'; import { createHash } from 'crypto'; @@ -10,6 +10,7 @@ import { NabAddonConfigSchema, NabAddonConfig, } from '../base/nab/addon.js'; +import { BuiltinProxy, createProxy } from '../../proxy/index.js'; const logger = createLogger('newznab'); @@ -19,15 +20,20 @@ class NewznabApi extends BaseNabApi<'newznab'> { } } +export const NewznabAddonConfigSchema = NabAddonConfigSchema.extend({ + proxyAuth: z.string(), +}); +export type NewznabAddonConfig = z.infer; + // Addon class -export class NewznabAddon extends BaseNabAddon { +export class NewznabAddon extends BaseNabAddon { readonly name = 'Newznab'; readonly version = '1.0.0'; readonly id = 'newznab'; readonly logger = logger; readonly api: NewznabApi; - constructor(userData: NabAddonConfig, clientIp?: string) { - super(userData, NabAddonConfigSchema, clientIp); + constructor(userData: NewznabAddonConfig, clientIp?: string) { + super(userData, NewznabAddonConfigSchema, clientIp); if ( !userData.services.find((s) => s.id === constants.TORBOX_SERVICE) || userData.services.length > 1 @@ -75,6 +81,32 @@ export class NewznabAddon extends BaseNabAddon { type: 'usenet', }); } + + if (this.userData.proxyAuth) { + try { + BuiltinProxy.validateAuth(this.userData.proxyAuth); + } catch (error) { + throw new Error('Invalid AIOStreams Proxy Auth Credentials'); + } + const proxy = createProxy({ + id: constants.BUILTIN_SERVICE, + url: Env.BASE_URL, + credentials: this.userData.proxyAuth, + }); + const urlsToProxy = nzbs.map((nzb) => nzb.nzb); + const proxiedUrls = await proxy.generateUrls( + urlsToProxy.map((url) => ({ + url, + filename: url.split('/').pop(), + })) + ); + if (!proxiedUrls) { + throw new Error('Failed to proxy NZBs'); + } + for (let i = 0; i < nzbs.length; i++) { + nzbs[i].nzb = proxiedUrls[i]; + } + } return nzbs; } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3920442a..a78a9c35 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -5,6 +5,7 @@ export * from './parser/index.js'; export * from './formatters/index.js'; export * from './transformers/index.js'; export * from './debrid/index.js'; +export * from './proxy/index.js'; export { TorBoxSearchAddon, GDriveAddon, diff --git a/packages/core/src/presets/newznab.ts b/packages/core/src/presets/newznab.ts index b7affecc..68406564 100644 --- a/packages/core/src/presets/newznab.ts +++ b/packages/core/src/presets/newznab.ts @@ -38,6 +38,14 @@ export class NewznabPreset extends BuiltinAddonPreset { required: false, default: '/api', }, + { + id: 'proxyAuth', + name: 'AIOStreams Proxy Auth', + description: + 'If you want to proxy the NZBs through AIOStreams, provide a username:password pair from the `BUILTIN_PROXY_AUTH` environment variable.', + type: 'password', + required: false, + }, { id: 'timeout', name: 'Timeout', @@ -150,6 +158,7 @@ export class NewznabPreset extends BuiltinAddonPreset { url: options.newznabUrl, apiPath: options.apiPath, apiKey: options.apiKey, + proxyAuth: options.proxyAuth, forceQuerySearch: options.forceQuerySearch ?? false, }; diff --git a/packages/core/src/proxy/base.ts b/packages/core/src/proxy/base.ts index cc952e93..a4721503 100644 --- a/packages/core/src/proxy/base.ts +++ b/packages/core/src/proxy/base.ts @@ -1,5 +1,11 @@ import { StreamProxyConfig } from '../db/schemas.js'; -import { Cache, createLogger, maskSensitiveInfo, Env } from '../utils/index.js'; +import { + Cache, + createLogger, + maskSensitiveInfo, + Env, + constants, +} from '../utils/index.js'; const logger = createLogger('proxy'); const cache = Cache.getInstance('publicIp'); @@ -14,7 +20,7 @@ export interface ProxyStream { } type ValidatedStreamProxyConfig = StreamProxyConfig & { - id: 'mediaflow' | 'stremthru'; + id: 'mediaflow' | 'stremthru' | 'builtin'; url: string; credentials: string; }; @@ -25,6 +31,9 @@ export abstract class BaseProxy { /^(10\.|127\.|172\.(1[6-9]|2[0-9]|3[01])\.|192\.168\.)/; constructor(config: StreamProxyConfig) { + if (config.id === constants.BUILTIN_SERVICE) { + config.url = Env.BASE_URL; + } if (!config.id || !config.credentials || !config.url) { throw new Error('Proxy configuration is missing'); } diff --git a/packages/core/src/proxy/builtin.ts b/packages/core/src/proxy/builtin.ts new file mode 100644 index 00000000..caf1380b --- /dev/null +++ b/packages/core/src/proxy/builtin.ts @@ -0,0 +1,168 @@ +import { BaseProxy, ProxyStream } from './base.js'; +import { + createLogger, + maskSensitiveInfo, + Env, + makeRequest, + encryptString, + Cache, +} from '../utils/index.js'; +import path from 'path'; + +const logger = createLogger('builtin'); + +export class BuiltinProxy extends BaseProxy { + public static validateAuth(auth: string): { + username: string; + password: string; + admin: boolean; + } { + const [username, password] = auth.split(':'); + if (!username || !password) { + throw new Error('Invalid credentials'); + } + + if ( + Env.BUILTIN_PROXY_AUTH?.has(username) && + Env.BUILTIN_PROXY_AUTH?.get(username) !== password + ) { + throw new Error('Invalid credentials'); + } + + return { + username, + password, + admin: + Env.BUILTIN_PROXY_ADMINS && Env.BUILTIN_PROXY_ADMINS.length > 0 + ? Env.BUILTIN_PROXY_ADMINS.includes(username) + : true, + }; + } + + protected override generateProxyUrl(endpoint: string): URL { + return new URL(endpoint); + } + + protected override getPublicIpEndpoint(): string { + return ''; + } + + protected override getPublicIpFromResponse(data: any): string | null { + return null; + } + + protected override getHeaders(): Record { + return {}; + } + + public override async getPublicIp(): Promise { + BuiltinProxy.validateAuth(this.config.credentials); + + const response = await makeRequest('https://checkip.amazonaws.com', { + method: 'GET', + timeout: 5000, + }); + + return response.text(); + } + + protected override async generateStreamUrls( + streams: ProxyStream[] + ): Promise { + const auth = BuiltinProxy.validateAuth(this.config.credentials); + return streams.map((stream) => { + const encryptedAuth = encryptString( + JSON.stringify({ + username: auth.username, + password: auth.password, + }) + ); + const encryptedData = encryptString( + JSON.stringify({ + url: stream.url, + filename: stream.filename, + requestHeaders: stream.headers?.request, + responseHeaders: stream.headers?.response, + }) + ); + return `${Env.BASE_URL}/api/v1/proxy/${encryptedAuth.data}.${encryptedData.data}/${encodeURIComponent(stream.filename ?? '')}`; + }); + } +} + +export class BuiltinProxyStats { + private activeConnections = Cache.getInstance< + string, + { ip: string; url: string; filename?: string; timestamp: number }[] + >('bproxy:stats', 10000, 'sql'); + + constructor() {} + + public async getAllActiveConnections(): Promise< + Map< + string, + { ip: string; url: string; filename?: string; timestamp: number }[] + > + > { + const users = Env.BUILTIN_PROXY_AUTH?.keys(); + + // create a map of users and their active connections + const connections = new Map< + string, + { ip: string; url: string; filename?: string; timestamp: number }[] + >(); + for (const user of users ?? []) { + connections.set(user, await this.getActiveConnections(user)); + } + return connections; + } + + public async getActiveConnections( + user: string + ): Promise< + { ip: string; url: string; filename?: string; timestamp: number }[] + > { + return (await this.activeConnections.get(user)) ?? []; + } + + public async addActiveConnection( + user: string, + ip: string, + url: string, + timestamp: number, + filename?: string + ) { + logger.debug(`[${user}] Adding active connection`, { + ip, + url, + filename, + timestamp, + }); + + const existingConnections = (await this.activeConnections.get(user)) ?? []; + const connectionKey = `${ip}:${url}`; + + // Filter out any existing connections with the same IP+filename combination + const filteredConnections = existingConnections.filter((conn) => { + return `${conn.ip}:${conn.url}` !== connectionKey; + }); + + // Add the new connection (which will be the most recent for this IP+filename) + const updatedConnections = [ + ...filteredConnections, + { ip, url, filename, timestamp }, + ]; + + await this.activeConnections.set(user, updatedConnections, 1 * 60 * 60); + } + + public async removeActiveConnection(user: string, ip: string, url: string) { + await this.activeConnections.set( + user, + ((await this.activeConnections.get(user)) ?? []).filter( + (connection) => connection.ip !== ip && connection.url !== url + ), + 24 * 60 * 60 + ); + } +} diff --git a/packages/core/src/proxy/index.ts b/packages/core/src/proxy/index.ts index a648074d..68235864 100644 --- a/packages/core/src/proxy/index.ts +++ b/packages/core/src/proxy/index.ts @@ -1,4 +1,5 @@ export * from './base.js'; +export * from './builtin.js'; export * from './mediaflow.js'; export * from './stremthru.js'; @@ -7,6 +8,7 @@ import { BaseProxy } from './base.js'; import { MediaFlowProxy } from './mediaflow.js'; import { StremThruProxy } from './stremthru.js'; import { StreamProxyConfig } from '../db/schemas.js'; +import { BuiltinProxy } from './builtin.js'; export function createProxy(config: StreamProxyConfig): BaseProxy { switch (config.id) { @@ -14,6 +16,8 @@ export function createProxy(config: StreamProxyConfig): BaseProxy { return new MediaFlowProxy(config); case constants.STREMTHRU_SERVICE: return new StremThruProxy(config); + case constants.BUILTIN_SERVICE: + return new BuiltinProxy(config); default: throw new Error(`Unknown proxy type: ${config.id}`); } diff --git a/packages/core/src/utils/config.ts b/packages/core/src/utils/config.ts index 47e1acf9..cf100947 100644 --- a/packages/core/src/utils/config.ts +++ b/packages/core/src/utils/config.ts @@ -902,6 +902,9 @@ async function validateProxy( if (!proxy.id) { throw new Error('Proxy ID is required'); } + if (proxy.id === constants.BUILTIN_SERVICE) { + proxy.url = Env.BASE_URL; + } if (!proxy.url) { throw new Error('Proxy URL is required'); } diff --git a/packages/core/src/utils/constants.ts b/packages/core/src/utils/constants.ts index a0c02ab6..78dc4200 100644 --- a/packages/core/src/utils/constants.ts +++ b/packages/core/src/utils/constants.ts @@ -21,6 +21,7 @@ export enum ErrorCode { METHOD_NOT_ALLOWED = 'METHOD_NOT_ALLOWED', RATE_LIMIT_EXCEEDED = 'RATE_LIMIT_EXCEEDED', BAD_REQUEST = 'BAD_REQUEST', + UNAUTHORIZED = 'UNAUTHORIZED', } interface ErrorDetails { @@ -89,6 +90,10 @@ export const ErrorMap: Record = { statusCode: 400, message: 'Bad request', }, + [ErrorCode.UNAUTHORIZED]: { + statusCode: 401, + message: 'Unauthorized', + }, }; export class APIError extends Error { @@ -231,8 +236,13 @@ export type BuiltinServiceId = (typeof BUILTIN_SUPPORTED_SERVICES)[number]; export const MEDIAFLOW_SERVICE = 'mediaflow' as const; export const STREMTHRU_SERVICE = 'stremthru' as const; +export const BUILTIN_SERVICE = 'builtin' as const; -export const PROXY_SERVICES = [MEDIAFLOW_SERVICE, STREMTHRU_SERVICE] as const; +export const PROXY_SERVICES = [ + MEDIAFLOW_SERVICE, + STREMTHRU_SERVICE, + BUILTIN_SERVICE, +] as const; export type ProxyServiceId = (typeof PROXY_SERVICES)[number]; export const PROXY_SERVICE_DETAILS: Record< @@ -258,7 +268,14 @@ export const PROXY_SERVICE_DETAILS: Record< description: '[StremThru](https://github.com/MunifTanjim/stremthru) is a feature packed companion to Stremio which also offers a HTTP proxy, written in Go.', credentialDescription: - 'A valid credential for your StremThru instance, defined in the `STREMTHRU_PROXY_AUTH` environment variable.', + 'A valid username:password pair for your StremThru instance, defined in the `STREMTHRU_PROXY_AUTH` environment variable.', + }, + [BUILTIN_SERVICE]: { + id: BUILTIN_SERVICE, + name: 'Builtin Proxy', + description: 'A proxy service that is built into the core of AIOStreams', + credentialDescription: + 'A valid username:password pair for this AIOStreams instance, defined in the `BUILTIN_PROXY_AUTH` environment variable.', }, }; diff --git a/packages/core/src/utils/env.ts b/packages/core/src/utils/env.ts index cef4631f..03c14c55 100644 --- a/packages/core/src/utils/env.ts +++ b/packages/core/src/utils/env.ts @@ -200,6 +200,24 @@ const readonly = makeValidator((x) => { return x; }); +const proxyAuth = makeValidator((x) => { + if (typeof x !== 'string') { + throw new EnvError('Proxy auth must be a string'); + } + // comma separated list of username:password + const userMap: Map = new Map(); + x.split(',').forEach((x) => { + const [username, password] = x.split(':'); + if (!username || !password) { + throw new EnvError( + 'Proxy auth must be a comma separated list of username:password pairs' + ); + } + userMap.set(username, password); + }); + return userMap; +}); + const boolOrList = makeValidator((x) => { if (typeof x !== 'string') { return undefined; @@ -1589,6 +1607,15 @@ export const Env = cleanEnv(process.env, { desc: 'Default AStream user agent', }), + BUILTIN_PROXY_AUTH: proxyAuth({ + default: undefined, + desc: 'Builtin proxy auth', + }), + BUILTIN_PROXY_ADMINS: commaSeparated({ + default: undefined, + desc: 'Comma separated list of admin usernames. If not set, all users are admins.', + }), + BUILTIN_STREMTHRU_URL: url({ default: 'https://stremthru.13377001.xyz', desc: 'Builtin StremThru URL', diff --git a/packages/core/src/utils/http.ts b/packages/core/src/utils/http.ts index 44fde8c3..08d3a41f 100644 --- a/packages/core/src/utils/http.ts +++ b/packages/core/src/utils/http.ts @@ -138,7 +138,7 @@ export async function makeRequest(url: string, options: RequestOptions) { } const proxyAgents = new Map(); -function getProxyAgent(proxyUrl: string): Dispatcher | undefined { +export function getProxyAgent(proxyUrl: string): Dispatcher | undefined { if (!proxyUrl) { return undefined; } @@ -163,7 +163,7 @@ function getProxyAgent(proxyUrl: string): Dispatcher | undefined { return proxyAgent; } -function shouldProxy(url: URL): { +export function shouldProxy(url: URL): { useProxy: boolean; proxyIndex: number; } { diff --git a/packages/core/src/utils/logger.ts b/packages/core/src/utils/logger.ts index f9ac42ef..4bc24da1 100644 --- a/packages/core/src/utils/logger.ts +++ b/packages/core/src/utils/logger.ts @@ -151,7 +151,11 @@ export const getTimeTakenSincePoint = (point: number) => { const duration = timeNow - point; if (duration < 1000) { return `${duration.toFixed(2)}ms`; - } else { + } else if (duration < 60000) { return `${(duration / 1000).toFixed(2)}s`; + } else if (duration < 3600000) { + return `${(duration / 60000).toFixed(0)}m`; + } else { + return `${(duration / 3600000).toFixed(0)}h`; } }; diff --git a/packages/core/src/utils/startup.ts b/packages/core/src/utils/startup.ts index 1f5b4e61..b78c37f3 100644 --- a/packages/core/src/utils/startup.ts +++ b/packages/core/src/utils/startup.ts @@ -528,6 +528,34 @@ const logStartupInfo = () => { ); }); + logSection('BUILT-IN PROXY', '🔧', () => { + if (Env.BUILTIN_PROXY_AUTH) { + logKeyValue('Status:', '✅ Configured'); + const users = Array.from(Env.BUILTIN_PROXY_AUTH.keys()); + if (users.length === 0) { + logKeyValue('Users:', '❌ None'); + } else { + logKeyValue('Users:', ''); + for (const user of users) { + const password = Env.BUILTIN_PROXY_AUTH.get(user); + const masked = + password && password.length > 0 + ? '*'.repeat(Math.max(4, Math.min(password.length, 12))) + : '❌ None'; + logKeyValue(` → ${user}:`, masked, ' '); + } + } + logKeyValue( + 'Admins:', + Env.BUILTIN_PROXY_ADMINS + ? `${Env.BUILTIN_PROXY_ADMINS.join(', ')}` + : '⚠️ All users' + ); + } else { + logKeyValue('Status:', '❌ None'); + } + }); + logSection('BUILT-IN ADDONS', '🔧', () => { // Torznab logKeyValue('*znab:', ''); diff --git a/packages/frontend/src/components/menu/proxy.tsx b/packages/frontend/src/components/menu/proxy.tsx index a977420d..458b433b 100644 --- a/packages/frontend/src/components/menu/proxy.tsx +++ b/packages/frontend/src/components/menu/proxy.tsx @@ -138,55 +138,59 @@ function Content() { )} -
- { - setUserData((prev) => ({ - ...prev, - proxy: { ...prev.proxy, url: v }, - })); - }} - placeholder="Enter proxy URL" - disabled={isUrlForced || !userData.proxy?.enabled} - /> -

- The URL of your hosted proxy service. -

-
+ {userData.proxy?.id !== 'builtin' && ( +
+ { + setUserData((prev) => ({ + ...prev, + proxy: { ...prev.proxy, url: v }, + })); + }} + placeholder="Enter proxy URL" + disabled={isUrlForced || !userData.proxy?.enabled} + /> +

+ The URL of your hosted proxy service. +

+
+ )} -
- { - setUserData((prev) => ({ - ...prev, - proxy: { ...prev.proxy, publicUrl: v }, - })); - }} - placeholder="Enter proxy public URL" - disabled={isPublicUrlForced || !userData.proxy?.enabled} - /> -

- The public URL of your hosted proxy service. Provide this only if - you want to use a local URL for requests but a publicly accessible - URL is needed for streams. e.g. setting http:// - {userData.proxy?.id - ? userData.proxy.id === 'stremthru' - ? 'stremthru:8080' - : 'mediaflow-proxy:8888' - : 'mediaflow-proxy:8888'} - as the URL above but then using https:// - {userData.proxy?.id - ? userData.proxy.id === 'stremthru' - ? 'stremthru.yourdomain.com' - : 'mediaflow-proxy.yourdomain.com' - : 'mediaflow-proxy.yourdomain.com'} - as the public URL. -

-
+ {userData.proxy?.id !== 'builtin' && ( +
+ { + setUserData((prev) => ({ + ...prev, + proxy: { ...prev.proxy, publicUrl: v }, + })); + }} + placeholder="Enter proxy public URL" + disabled={isPublicUrlForced || !userData.proxy?.enabled} + /> +

+ The public URL of your hosted proxy service. Provide this only + if you want to use a local URL for requests but a publicly + accessible URL is needed for streams. e.g. setting http:// + {userData.proxy?.id + ? userData.proxy.id === 'stremthru' + ? 'stremthru:8080' + : 'mediaflow-proxy:8888' + : 'mediaflow-proxy:8888'} + as the URL above but then using https:// + {userData.proxy?.id + ? userData.proxy.id === 'stremthru' + ? 'stremthru.yourdomain.com' + : 'mediaflow-proxy.yourdomain.com' + : 'mediaflow-proxy.yourdomain.com'} + as the public URL. +

+
+ )}
{ + // only show stats to admin users + try { + const { auth: authQuery } = z + .object({ auth: z.string() }) + .parse(req.query); + const auth = BuiltinProxy.validateAuth(authQuery); + if (!auth.admin) { + throw new APIError( + constants.ErrorCode.UNAUTHORIZED, + undefined, + 'Invalid auth' + ); + } + } catch (error) { + if (error instanceof APIError) { + next(error); + } else { + next( + new APIError( + constants.ErrorCode.UNAUTHORIZED, + undefined, + 'Invalid auth' + ) + ); + } + } + + try { + const allConnections = await proxyStats.getAllActiveConnections(); + + // Convert Map to a more JSON-friendly format + const stats = { + timestamp: new Date().toISOString(), + totalUsers: allConnections.size, + activeConnections: Object.fromEntries( + Array.from(allConnections.entries()).map(([user, connections]) => [ + user, + connections.map((conn) => ({ + ...conn, + timestamp: new Date(conn.timestamp).toISOString(), + relativeTimestamp: `${getTimeTakenSincePoint(conn.timestamp)} ago`, + })), + ]) + ), + summary: { + totalActiveConnections: Array.from(allConnections.values()).reduce( + (total, connections) => total + connections.length, + 0 + ), + usersWithActiveConnections: Array.from( + allConnections.entries() + ).filter(([_, connections]) => connections.length > 0).length, + }, + }; + + res.json(stats); + } catch (error) { + logger.error('Failed to get proxy stats', { + error: error instanceof Error ? error.message : String(error), + }); + next(error); + } + } +); + +router.all( + '/:encryptedAuthAndData{/:filename}', + async (req: Request, res: Response, next: NextFunction) => { + const startTime = Date.now(); + const requestId = Math.random().toString(36).substring(7); + let upstreamResponse: Dispatcher.ResponseData | undefined; + let auth: { username: string; password: string } | undefined; + let data: z.infer | undefined; + let clientIp: string | undefined; + + try { + // decrypt and authenticate the request + const { encryptedAuthAndData } = req.params; + const [encryptedAuth, encryptedData] = encryptedAuthAndData.split('.'); + const filename = req.params.filename as string | undefined; + + const { data: rawData } = decryptString(encryptedData); + const { data: rawAuth } = decryptString(encryptedAuth); + + if (!rawData || !rawAuth) { + logger.error(`[${requestId}] Decryption failed`); + throw new APIError( + constants.ErrorCode.ENCRYPTION_ERROR, + undefined, + 'Could not decrypt data or auth' + ); + } + + data = ProxyDataSchema.parse(JSON.parse(rawData)); + auth = ProxyAuthSchema.parse(JSON.parse(rawAuth)); + + if ( + !Env.BUILTIN_PROXY_AUTH?.has(auth.username) || + Env.BUILTIN_PROXY_AUTH?.get(auth.username) !== auth.password + ) { + logger.warn(`[${requestId}] Authentication failed`, { + username: auth.username, + }); + throw new APIError( + constants.ErrorCode.UNAUTHORIZED, + undefined, + 'Invalid auth' + ); + } + + // Track the active connection + clientIp = req.ip || req.connection.remoteAddress || 'unknown'; + const timestamp = Date.now(); + proxyStats.addActiveConnection( + auth.username, + clientIp, + data.url, + timestamp, + filename + ); + + // prepare and execute upstream request + const { host, ...clientHeaders } = req.headers; + + const isBodyRequest = + req.method === 'POST' || req.method === 'PUT' || req.method === 'PATCH'; + + const upstreamStartTime = Date.now(); + const urlObj = new URL(data.url); + if (Env.BASE_URL && urlObj.origin === Env.BASE_URL) { + const internalUrl = new URL(Env.INTERNAL_URL); + urlObj.protocol = internalUrl.protocol; + urlObj.host = internalUrl.host; + urlObj.port = internalUrl.port; + } + + if (Env.REQUEST_URL_MAPPINGS) { + for (const [key, value] of Object.entries(Env.REQUEST_URL_MAPPINGS)) { + if (urlObj.origin === key) { + const mappedUrl = new URL(value); + urlObj.protocol = mappedUrl.protocol; + urlObj.host = mappedUrl.host; + urlObj.port = mappedUrl.port; + break; + } + } + } + const { useProxy, proxyIndex } = shouldProxy(urlObj); + const proxyAgent = useProxy + ? getProxyAgent(Env.ADDON_PROXY![proxyIndex]) + : undefined; + upstreamResponse = await request(data.url, { + method: req.method as Dispatcher.HttpMethod, + headers: { ...clientHeaders, ...data.requestHeaders }, + dispatcher: proxyAgent, + body: isBodyRequest ? req : undefined, + bodyTimeout: 0, + headersTimeout: 0, + }); + const upstreamDuration = getTimeTakenSincePoint(upstreamStartTime); + + logger.debug(`[${requestId}] Serving upstream response`, { + username: auth.username, + targetUrl: data.url, + statusCode: upstreamResponse.statusCode, + upstreamDuration, + }); + + // forward upstream response to client + res.set(upstreamResponse.headers); + if (data.responseHeaders) { + res.set(data.responseHeaders); + } + res.status(upstreamResponse.statusCode); + + if (req.method === 'HEAD') { + res.end(); + } else { + await pipeline(upstreamResponse.body, res); + } + logger.debug(`[${requestId}] Proxy connection closed`, { + username: auth.username, + }); + } catch (error) { + const totalDuration = Date.now() - startTime; + + // Remove the active connection tracking on error + if (auth && clientIp && data) { + proxyStats + .removeActiveConnection(auth.username, clientIp, data.url) + .catch((statsError) => + logger.warn( + `[${requestId}] Failed to remove connection from stats on error`, + { error: statsError } + ) + ); + } + + if (upstreamResponse) { + upstreamResponse.body.destroy(); + } + + if ( + (error as NodeJS.ErrnoException)?.code !== 'ERR_STREAM_PREMATURE_CLOSE' + ) { + logger.error(`[${requestId}] Proxy request failed`, { + error: error instanceof Error ? error.message : String(error), + durationMs: totalDuration, + upstreamStatusCode: upstreamResponse?.statusCode, + }); + next(error); + } else { + logger.debug(`[${requestId}] Client disconnected (premature close)`, { + durationMs: totalDuration, + }); + } + } + } +); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index aa90958a..937388fc 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -295,6 +295,9 @@ importers: rate-limit-redis: specifier: ^4.2.2 version: 4.2.2(express-rate-limit@8.0.1(express@5.1.0)) + undici: + specifier: ^7.13.0 + version: 7.13.0 zod: specifier: ^4.1.5 version: 4.1.5