Compare commits

...

10 Commits

4 changed files with 133 additions and 27 deletions

View File

@ -1,5 +1,36 @@
# Changelog # Changelog
## 2025-02-22 - 3.10.1 - fix(PortProxy)
Improve socket cleanup logic to prevent potential resource leaks
- Updated socket cleanup in PortProxy to ensure sockets are forcefully destroyed if not already destroyed.
## 2025-02-22 - 3.10.0 - feat(smartproxy.portproxy)
Enhance PortProxy with detailed connection statistics and termination tracking
- Added tracking of termination statistics for incoming and outgoing connections
- Enhanced logging to include detailed termination statistics
- Introduced helpers to update and log termination stats
- Retained detailed connection duration and active connection logging
## 2025-02-22 - 3.9.4 - fix(PortProxy)
Ensure proper cleanup on connection rejection in PortProxy
- Added cleanup calls after socket end in connection rejection scenarios within PortProxy
## 2025-02-21 - 3.9.3 - fix(PortProxy)
Fix handling of optional outgoing socket in PortProxy
- Refactored the cleanUpSockets function to correctly handle cases where the outgoing socket may be undefined.
- Ensured correct handling of socket events with non-null assertions where applicable.
- Improved robustness in connection establishment and cleanup processes.
## 2025-02-21 - 3.9.2 - fix(PortProxy)
Improve timeout handling for port proxy connections
- Added console logging for both incoming and outgoing side timeouts in the PortProxy class.
- Updated the timeout event handlers to ensure proper cleanup of connections.
## 2025-02-21 - 3.9.1 - fix(dependencies) ## 2025-02-21 - 3.9.1 - fix(dependencies)
Ensure correct ordering of dependencies and improve logging format. Ensure correct ordering of dependencies and improve logging format.

View File

@ -1,6 +1,6 @@
{ {
"name": "@push.rocks/smartproxy", "name": "@push.rocks/smartproxy",
"version": "3.9.1", "version": "3.10.1",
"private": false, "private": false,
"description": "a proxy for handling high workloads of proxying", "description": "a proxy for handling high workloads of proxying",
"main": "dist_ts/index.js", "main": "dist_ts/index.js",

View File

@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@push.rocks/smartproxy', name: '@push.rocks/smartproxy',
version: '3.9.1', version: '3.10.1',
description: 'a proxy for handling high workloads of proxying' description: 'a proxy for handling high workloads of proxying'
} }

View File

@ -123,6 +123,15 @@ export class PortProxy {
private outgoingConnectionTimes: Map<plugins.net.Socket, number> = new Map(); private outgoingConnectionTimes: Map<plugins.net.Socket, number> = new Map();
private connectionLogger: NodeJS.Timeout | null = null; private connectionLogger: NodeJS.Timeout | null = null;
// Overall termination statistics
private terminationStats: {
incoming: Record<string, number>;
outgoing: Record<string, number>;
} = {
incoming: {},
outgoing: {},
};
constructor(settings: IProxySettings) { constructor(settings: IProxySettings) {
this.settings = { this.settings = {
...settings, ...settings,
@ -130,16 +139,24 @@ export class PortProxy {
}; };
} }
// Helper to update termination stats.
private incrementTerminationStat(side: 'incoming' | 'outgoing', reason: string): void {
if (!this.terminationStats[side][reason]) {
this.terminationStats[side][reason] = 1;
} else {
this.terminationStats[side][reason]++;
}
}
public async start() { public async start() {
const cleanUpSockets = (from: plugins.net.Socket, to: plugins.net.Socket) => { // Adjusted cleanUpSockets: forcefully destroy both sockets if they haven't been destroyed.
from.end(); const cleanUpSockets = (from: plugins.net.Socket, to?: plugins.net.Socket) => {
to.end(); if (!from.destroyed) {
from.removeAllListeners();
to.removeAllListeners();
from.unpipe();
to.unpipe();
from.destroy(); from.destroy();
}
if (to && !to.destroyed) {
to.destroy(); to.destroy();
}
}; };
const normalizeIP = (ip: string): string[] => { const normalizeIP = (ip: string): string[] => {
@ -180,6 +197,10 @@ export class PortProxy {
// Flag to detect if we've received the first data chunk. // Flag to detect if we've received the first data chunk.
let initialDataReceived = false; let initialDataReceived = false;
// Local termination reason trackers for each side.
let incomingTermReason: string | null = null;
let outgoingTermReason: string | null = null;
// Immediately attach an error handler to catch early errors. // Immediately attach an error handler to catch early errors.
socket.on('error', (err: Error) => { socket.on('error', (err: Error) => {
if (!initialDataReceived) { if (!initialDataReceived) {
@ -189,12 +210,12 @@ export class PortProxy {
} }
}); });
// Flag to ensure cleanup happens only once. // Ensure cleanup happens only once.
let connectionClosed = false; let connectionClosed = false;
const cleanupOnce = () => { const cleanupOnce = () => {
if (!connectionClosed) { if (!connectionClosed) {
connectionClosed = true; connectionClosed = true;
cleanUpSockets(socket, to); cleanUpSockets(socket, to || undefined);
this.incomingConnectionTimes.delete(socket); this.incomingConnectionTimes.delete(socket);
if (to) { if (to) {
this.outgoingConnectionTimes.delete(to); this.outgoingConnectionTimes.delete(to);
@ -206,20 +227,39 @@ export class PortProxy {
} }
}; };
let to: plugins.net.Socket; // Outgoing connection placeholder.
let to: plugins.net.Socket | null = null;
// Handle errors by recording termination reason and cleaning up.
const handleError = (side: 'incoming' | 'outgoing') => (err: Error) => { const handleError = (side: 'incoming' | 'outgoing') => (err: Error) => {
const code = (err as any).code; const code = (err as any).code;
let reason = 'error';
if (code === 'ECONNRESET') { if (code === 'ECONNRESET') {
reason = 'econnreset';
console.log(`ECONNRESET on ${side} side from ${remoteIP}: ${err.message}`); console.log(`ECONNRESET on ${side} side from ${remoteIP}: ${err.message}`);
} else { } else {
console.log(`Error on ${side} side from ${remoteIP}: ${err.message}`); console.log(`Error on ${side} side from ${remoteIP}: ${err.message}`);
} }
if (side === 'incoming' && incomingTermReason === null) {
incomingTermReason = reason;
this.incrementTerminationStat('incoming', reason);
} else if (side === 'outgoing' && outgoingTermReason === null) {
outgoingTermReason = reason;
this.incrementTerminationStat('outgoing', reason);
}
cleanupOnce(); cleanupOnce();
}; };
// Handle close events. If no termination reason was recorded, mark as "normal".
const handleClose = (side: 'incoming' | 'outgoing') => () => { const handleClose = (side: 'incoming' | 'outgoing') => () => {
console.log(`Connection closed on ${side} side from ${remoteIP}`); console.log(`Connection closed on ${side} side from ${remoteIP}`);
if (side === 'incoming' && incomingTermReason === null) {
incomingTermReason = 'normal';
this.incrementTerminationStat('incoming', 'normal');
} else if (side === 'outgoing' && outgoingTermReason === null) {
outgoingTermReason = 'normal';
this.incrementTerminationStat('outgoing', 'normal');
}
cleanupOnce(); cleanupOnce();
}; };
@ -232,16 +272,31 @@ export class PortProxy {
if (!domainConfig) { if (!domainConfig) {
console.log(`Connection rejected: No matching domain config for ${serverName} from ${remoteIP}`); console.log(`Connection rejected: No matching domain config for ${serverName} from ${remoteIP}`);
socket.end(); socket.end();
if (incomingTermReason === null) {
incomingTermReason = 'rejected';
this.incrementTerminationStat('incoming', 'rejected');
}
cleanupOnce();
return; return;
} }
if (!isAllowed(remoteIP, domainConfig.allowedIPs)) { if (!isAllowed(remoteIP, domainConfig.allowedIPs)) {
console.log(`Connection rejected: IP ${remoteIP} not allowed for domain ${serverName}`); console.log(`Connection rejected: IP ${remoteIP} not allowed for domain ${serverName}`);
socket.end(); socket.end();
if (incomingTermReason === null) {
incomingTermReason = 'rejected';
this.incrementTerminationStat('incoming', 'rejected');
}
cleanupOnce();
return; return;
} }
} else if (!isDefaultAllowed && !serverName) { } else if (!isDefaultAllowed && !serverName) {
console.log(`Connection rejected: No SNI and IP ${remoteIP} not in default allowed list`); console.log(`Connection rejected: No SNI and IP ${remoteIP} not in default allowed list`);
socket.end(); socket.end();
if (incomingTermReason === null) {
incomingTermReason = 'rejected';
this.incrementTerminationStat('incoming', 'rejected');
}
cleanupOnce();
return; return;
} else { } else {
console.log(`Connection allowed: IP ${remoteIP} is in default allowed list`); console.log(`Connection allowed: IP ${remoteIP} is in default allowed list`);
@ -262,8 +317,9 @@ export class PortProxy {
// Establish outgoing connection. // Establish outgoing connection.
to = plugins.net.connect(connectionOptions); to = plugins.net.connect(connectionOptions);
// Record start time for the outgoing connection. if (to) {
this.outgoingConnectionTimes.set(to, Date.now()); this.outgoingConnectionTimes.set(to, Date.now());
}
console.log(`Connection established: ${remoteIP} -> ${targetHost}:${this.settings.toPort}${serverName ? ` (SNI: ${serverName})` : ''}`); console.log(`Connection established: ${remoteIP} -> ${targetHost}:${this.settings.toPort}${serverName ? ` (SNI: ${serverName})` : ''}`);
// Push back the initial chunk if provided. // Push back the initial chunk if provided.
@ -271,25 +327,38 @@ export class PortProxy {
socket.unshift(initialChunk); socket.unshift(initialChunk);
} }
socket.setTimeout(120000); socket.setTimeout(120000);
socket.pipe(to); socket.pipe(to!);
to.pipe(socket); to!.pipe(socket);
// Attach error and close handlers for both sockets. // Attach event handlers for both sockets.
socket.on('error', handleError('incoming')); socket.on('error', handleError('incoming'));
to.on('error', handleError('outgoing')); to!.on('error', handleError('outgoing'));
socket.on('close', handleClose('incoming')); socket.on('close', handleClose('incoming'));
to.on('close', handleClose('outgoing')); to!.on('close', handleClose('outgoing'));
socket.on('timeout', handleError('incoming')); socket.on('timeout', () => {
to.on('timeout', handleError('outgoing')); console.log(`Timeout on incoming side from ${remoteIP}`);
if (incomingTermReason === null) {
incomingTermReason = 'timeout';
this.incrementTerminationStat('incoming', 'timeout');
}
cleanupOnce();
});
to!.on('timeout', () => {
console.log(`Timeout on outgoing side from ${remoteIP}`);
if (outgoingTermReason === null) {
outgoingTermReason = 'timeout';
this.incrementTerminationStat('outgoing', 'timeout');
}
cleanupOnce();
});
socket.on('end', handleClose('incoming')); socket.on('end', handleClose('incoming'));
to.on('end', handleClose('outgoing')); to!.on('end', handleClose('outgoing'));
}; };
// For SNI-enabled connections, peek at the first chunk. // For SNI-enabled connections, peek at the first chunk.
if (this.settings.sniEnabled) { if (this.settings.sniEnabled) {
socket.once('data', (chunk: Buffer) => { socket.once('data', (chunk: Buffer) => {
initialDataReceived = true; initialDataReceived = true;
// Try to extract the server name from the ClientHello.
const serverName = extractSNI(chunk) || ''; const serverName = extractSNI(chunk) || '';
console.log(`Received connection from ${remoteIP} with SNI: ${serverName}`); console.log(`Received connection from ${remoteIP} with SNI: ${serverName}`);
setupConnection(serverName, chunk); setupConnection(serverName, chunk);
@ -300,6 +369,11 @@ export class PortProxy {
if (!this.settings.defaultAllowedIPs || !isAllowed(remoteIP, this.settings.defaultAllowedIPs)) { if (!this.settings.defaultAllowedIPs || !isAllowed(remoteIP, this.settings.defaultAllowedIPs)) {
console.log(`Connection rejected: IP ${remoteIP} not allowed for non-SNI connection`); console.log(`Connection rejected: IP ${remoteIP} not allowed for non-SNI connection`);
socket.end(); socket.end();
if (incomingTermReason === null) {
incomingTermReason = 'rejected';
this.incrementTerminationStat('incoming', 'rejected');
}
cleanupOnce();
return; return;
} }
setupConnection(''); setupConnection('');
@ -312,7 +386,8 @@ export class PortProxy {
console.log(`PortProxy -> OK: Now listening on port ${this.settings.fromPort}${this.settings.sniEnabled ? ' (SNI passthrough enabled)' : ''}`); console.log(`PortProxy -> OK: Now listening on port ${this.settings.fromPort}${this.settings.sniEnabled ? ' (SNI passthrough enabled)' : ''}`);
}); });
// Log active connection count and longest running connections every 10 seconds. // Log active connection count, longest running connection durations,
// and termination statistics every 10 seconds.
this.connectionLogger = setInterval(() => { this.connectionLogger = setInterval(() => {
const now = Date.now(); const now = Date.now();
let maxIncoming = 0; let maxIncoming = 0;
@ -329,7 +404,7 @@ export class PortProxy {
maxOutgoing = duration; maxOutgoing = duration;
} }
} }
console.log(`(Interval Log) Active connections: ${this.activeConnections.size}. Longest running incoming: ${plugins.prettyMs(maxIncoming)}, outgoing: ${plugins.prettyMs(maxOutgoing)}`); console.log(`(Interval Log) Active connections: ${this.activeConnections.size}. Longest running incoming: ${plugins.prettyMs(maxIncoming)}, outgoing: ${plugins.prettyMs(maxOutgoing)}. Termination stats (incoming): ${JSON.stringify(this.terminationStats.incoming)}, (outgoing): ${JSON.stringify(this.terminationStats.outgoing)}`);
}, 10000); }, 10000);
} }