Compare commits

...

12 Commits

Author SHA1 Message Date
f82d44164c 19.6.16
Some checks failed
Default (tags) / security (push) Successful in 1m20s
Default (tags) / test (push) Failing after 29m31s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2025-07-03 03:17:35 +00:00
2a4ed38f6b update logs 2025-07-03 02:54:56 +00:00
bb2c82b44a 19.6.15
Some checks failed
Default (tags) / security (push) Successful in 1m22s
Default (tags) / test (push) Failing after 29m38s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2025-07-03 02:45:30 +00:00
dddcf8dec4 improve logging 2025-07-03 02:45:08 +00:00
8d7213e91b 19.6.14
Some checks failed
Default (tags) / security (push) Successful in 1m24s
Default (tags) / test (push) Failing after 29m37s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2025-07-03 02:33:04 +00:00
5d011ba84c better logging 2025-07-03 02:32:17 +00:00
67aff4bb30 19.6.13
Some checks failed
Default (tags) / security (push) Successful in 1m25s
Default (tags) / test (push) Failing after 29m5s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2025-06-23 15:42:39 +00:00
3857d2670f fix(metrics): fix metrics 2025-06-23 15:42:04 +00:00
4587940f38 19.6.12
Some checks failed
Default (tags) / security (push) Successful in 1m28s
Default (tags) / test (push) Failing after 29m14s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2025-06-23 13:19:56 +00:00
82ca0381e9 fix(metrics): fix metrics 2025-06-23 13:19:39 +00:00
7bf15e72f9 19.6.11
Some checks failed
Default (tags) / security (push) Successful in 1m29s
Default (tags) / test (push) Failing after 29m11s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2025-06-23 13:07:46 +00:00
caa15e539e fix(metrics): fix metrics 2025-06-23 13:07:30 +00:00
22 changed files with 1895 additions and 501 deletions

View File

@ -1,5 +1,5 @@
{
"expiryDate": "2025-09-21T08:37:03.077Z",
"issueDate": "2025-06-23T08:37:03.077Z",
"savedAt": "2025-06-23T08:37:03.078Z"
"expiryDate": "2025-10-01T02:31:27.435Z",
"issueDate": "2025-07-03T02:31:27.435Z",
"savedAt": "2025-07-03T02:31:27.435Z"
}

View File

@ -1,6 +1,6 @@
{
"name": "@push.rocks/smartproxy",
"version": "19.6.10",
"version": "19.6.16",
"private": false,
"description": "A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.",
"main": "dist_ts/index.js",

View File

@ -2,6 +2,21 @@
## Byte Tracking and Metrics
### Throughput Drift Issue (Fixed)
**Problem**: Throughput numbers were gradually increasing over time for long-lived connections.
**Root Cause**: The `byRoute()` and `byIP()` methods were dividing cumulative total bytes (since connection start) by the window duration, causing rates to appear higher as connections aged:
- Hour 1: 1GB total / 60s = 17 MB/s ✓
- Hour 2: 2GB total / 60s = 34 MB/s ✗ (appears doubled!)
- Hour 3: 3GB total / 60s = 50 MB/s ✗ (keeps rising!)
**Solution**: Implemented dedicated ThroughputTracker instances for each route and IP address:
- Each route and IP gets its own throughput tracker with per-second sampling
- Samples are taken every second and stored in a circular buffer
- Rate calculations use actual samples within the requested window
- Default window is now 1 second for real-time accuracy
### What Gets Counted (Network Interface Throughput)
The byte tracking is designed to match network interface throughput (what Unifi/network monitoring tools show):
@ -39,12 +54,19 @@ The byte tracking is designed to match network interface throughput (what Unifi/
### Metrics Architecture
The metrics system has three layers:
The metrics system has multiple layers:
1. **Connection Records** (`record.bytesReceived/bytesSent`): Track total bytes per connection
2. **ThroughputTracker**: Accumulates bytes between samples for rate calculations (bytes/second)
3. **connectionByteTrackers**: Track bytes per connection with timestamps for per-route/IP metrics
2. **Global ThroughputTracker**: Accumulates bytes between samples for overall rate calculations
3. **Per-Route ThroughputTrackers**: Dedicated tracker for each route with per-second sampling
4. **Per-IP ThroughputTrackers**: Dedicated tracker for each IP with per-second sampling
5. **connectionByteTrackers**: Track cumulative bytes and metadata for active connections
Total byte counts come from connection records only, preventing double counting.
Key features:
- All throughput trackers sample every second (1Hz)
- Each tracker maintains a circular buffer of samples (default: 1 hour retention)
- Rate calculations are accurate for any requested window (default: 1 second)
- All byte counting happens exactly once at the data flow point
- Unused route/IP trackers are automatically cleaned up when connections close
### Understanding "High" Byte Counts
@ -120,4 +142,157 @@ Keep-alive connections receive special treatment based on `keepAliveTreatment` s
The system supports both receiving and sending PROXY protocol:
- **Receiving**: Automatically detected from trusted proxy IPs (configured in `proxyIPs`)
- **Sending**: Enabled per-route or globally via `sendProxyProtocol` setting
- Real client IP is preserved and used for all connection tracking and security checks
- Real client IP is preserved and used for all connection tracking and security checks
## Metrics and Throughput Calculation
The metrics system tracks throughput using per-second sampling:
1. **Byte Recording**: Bytes are recorded as data flows through connections
2. **Sampling**: Every second, accumulated bytes are stored as a sample
3. **Rate Calculation**: Throughput is calculated by summing bytes over a time window
4. **Per-Route/IP Tracking**: Separate ThroughputTracker instances for each route and IP
Key implementation details:
- Bytes are recorded in the bidirectional forwarding callbacks
- The instant() method returns throughput over the last 1 second
- The recent() method returns throughput over the last 10 seconds
- Custom windows can be specified for different averaging periods
### Throughput Spikes Issue
There's a fundamental difference between application-layer and network-layer throughput:
**Application Layer (what we measure)**:
- Bytes are recorded when delivered to/from the application
- Large chunks can arrive "instantly" due to kernel/Node.js buffering
- Shows spikes when buffers are flushed (e.g., 20MB in 1 second = 160 Mbit/s)
**Network Layer (what Unifi shows)**:
- Actual packet flow through the network interface
- Limited by physical network speed (e.g., 20 Mbit/s)
- Data transfers over time, not in bursts
The spikes occur because:
1. Data flows over network at 20 Mbit/s (takes 8 seconds for 20MB)
2. Kernel/Node.js buffers this incoming data
3. When buffer is flushed, application receives large chunk at once
4. We record entire chunk in current second, creating artificial spike
**Potential Solutions**:
1. Use longer window for "instant" measurements (e.g., 5 seconds instead of 1)
2. Track socket write backpressure to estimate actual network flow
3. Implement bandwidth estimation based on connection duration
4. Accept that application-layer != network-layer throughput
## Connection Limiting
### Per-IP Connection Limits
- SmartProxy tracks connections per IP address in the SecurityManager
- Default limit is 100 connections per IP (configurable via `maxConnectionsPerIP`)
- Connection rate limiting is also enforced (default 300 connections/minute per IP)
- HttpProxy has been enhanced to also enforce per-IP limits when forwarding from SmartProxy
### Route-Level Connection Limits
- Routes can define `security.maxConnections` to limit connections per route
- ConnectionManager tracks connections by route ID using a separate Map
- Limits are enforced in RouteConnectionHandler before forwarding
- Connection is tracked when route is matched: `trackConnectionByRoute(routeId, connectionId)`
### HttpProxy Integration
- When SmartProxy forwards to HttpProxy for TLS termination, it sends a `CLIENT_IP:<ip>\r\n` header
- HttpProxy parses this header to track the real client IP, not the localhost IP
- This ensures per-IP limits are enforced even for forwarded connections
- The header is parsed in the connection handler before any data processing
### Memory Optimization
- Periodic cleanup runs every 60 seconds to remove:
- IPs with no active connections
- Expired rate limit timestamps (older than 1 minute)
- Prevents memory accumulation from many unique IPs over time
- Cleanup is automatic and runs in background with `unref()` to not keep process alive
### Connection Cleanup Queue
- Cleanup queue processes connections in batches to prevent overwhelming the system
- Race condition prevention using `isProcessingCleanup` flag
- Try-finally block ensures flag is always reset even if errors occur
- New connections added during processing are queued for next batch
### Important Implementation Notes
- Always use `NodeJS.Timeout` type instead of `NodeJS.Timer` for interval/timeout references
- IPv4/IPv6 normalization is handled (e.g., `::ffff:127.0.0.1` and `127.0.0.1` are treated as the same IP)
- Connection limits are checked before route matching to prevent DoS attacks
- SharedSecurityManager supports checking route-level limits via optional parameter
## Log Deduplication
To reduce log spam during high-traffic scenarios or attacks, SmartProxy implements log deduplication for repetitive events:
### How It Works
- Similar log events are batched and aggregated over a 5-second window
- Instead of logging each event individually, a summary is emitted
- Events are grouped by type and deduplicated by key (e.g., IP address, reason)
### Deduplicated Event Types
1. **Connection Rejections** (`connection-rejected`):
- Groups by rejection reason (global-limit, route-limit, etc.)
- Example: "Rejected 150 connections (reasons: global-limit: 100, route-limit: 50)"
2. **IP Rejections** (`ip-rejected`):
- Groups by IP address
- Shows top offenders with rejection counts and reasons
- Example: "Rejected 500 connections from 10 IPs (top offenders: 192.168.1.100 (200x, rate-limit), ...)"
3. **Connection Cleanups** (`connection-cleanup`):
- Groups by cleanup reason (normal, timeout, error, zombie, etc.)
- Example: "Cleaned up 250 connections (reasons: normal: 200, timeout: 30, error: 20)"
4. **IP Tracking Cleanup** (`ip-cleanup`):
- Summarizes periodic IP cleanup operations
- Example: "IP tracking cleanup: removed 50 entries across 5 cleanup cycles"
### Configuration
- Default flush interval: 5 seconds
- Maximum batch size: 100 events (triggers immediate flush)
- Global periodic flush: Every 10 seconds (ensures logs are emitted regularly)
- Process exit handling: Logs are flushed on SIGINT/SIGTERM
### Benefits
- Reduces log volume during attacks or high traffic
- Provides better overview of patterns (e.g., which IPs are attacking)
- Improves log readability and analysis
- Prevents log storage overflow
- Maintains detailed information in aggregated form
### Log Output Examples
Instead of hundreds of individual logs:
```
Connection rejected
Connection rejected
Connection rejected
... (repeated 500 times)
```
You'll see:
```
[SUMMARY] Rejected 500 connections from 10 IPs in 5s (rate-limit: 350, per-ip-limit: 150) (top offenders: 192.168.1.100 (200x, rate-limit), 10.0.0.1 (150x, per-ip-limit))
```
Instead of:
```
Connection terminated: ::ffff:127.0.0.1 (client_closed). Active: 266
Connection terminated: ::ffff:127.0.0.1 (client_closed). Active: 265
... (repeated 266 times)
```
You'll see:
```
[SUMMARY] 266 HttpProxy connections terminated in 5s (reasons: client_closed: 266, activeConnections: 0)
```
### Rapid Event Handling
- During attacks or high-volume scenarios, logs are flushed more frequently
- If 50+ events occur within 1 second, immediate flush is triggered
- Prevents memory buildup during flooding attacks
- Maintains real-time visibility during incidents

View File

@ -1,364 +1,53 @@
# SmartProxy Metrics Improvement Plan
# SmartProxy Connection Limiting Improvements Plan
## Overview
Command to re-read CLAUDE.md: `cat /home/philkunz/.claude/CLAUDE.md`
The current `getThroughputRate()` implementation calculates cumulative throughput over a 60-second window rather than providing an actual rate, making metrics misleading for monitoring systems. This plan outlines a comprehensive redesign of the metrics system to provide accurate, time-series based metrics suitable for production monitoring.
## Issues Identified
## 1. Core Issues with Current Implementation
1. **HttpProxy Bypass**: Connections forwarded to HttpProxy for TLS termination only check global limits, not per-IP limits
2. **Missing Route-Level Connection Enforcement**: Routes can define `security.maxConnections` but it's never enforced
3. **Cleanup Queue Race Condition**: New connections can be added to cleanup queue while processing
4. **IP Tracking Memory Optimization**: IP entries remain in map even without active connections
- **Cumulative vs Rate**: Current method accumulates all bytes from connections in the last minute rather than calculating actual throughput rate
- **No Time-Series Data**: Cannot track throughput changes over time
- **Inaccurate Estimates**: Attempting to estimate rates for older connections is fundamentally flawed
- **No Sliding Windows**: Cannot provide different time window views (1s, 10s, 60s, etc.)
- **Limited Granularity**: Only provides a single 60-second view
## Implementation Steps
## 2. Proposed Architecture
### 1. Fix HttpProxy Per-IP Validation ✓
- [x] Pass IP information to HttpProxy when forwarding connections
- [x] Add per-IP validation in HttpProxy connection handler
- [x] Ensure connection tracking is consistent between SmartProxy and HttpProxy
### A. Time-Series Throughput Tracking
### 2. Implement Route-Level Connection Limits ✓
- [x] Add connection count tracking per route in ConnectionManager
- [x] Update SharedSecurityManager.isAllowed() to check route-specific maxConnections
- [x] Add route connection limit validation in route-connection-handler.ts
```typescript
interface IThroughputSample {
timestamp: number;
bytesIn: number;
bytesOut: number;
}
### 3. Fix Cleanup Queue Race Condition ✓
- [x] Implement proper queue snapshotting before processing
- [x] Ensure new connections added during processing aren't missed
- [x] Add proper synchronization for cleanup operations
class ThroughputTracker {
private samples: IThroughputSample[] = [];
private readonly MAX_SAMPLES = 3600; // 1 hour at 1 sample/second
private lastSampleTime: number = 0;
private accumulatedBytesIn: number = 0;
private accumulatedBytesOut: number = 0;
// Called on every data transfer
public recordBytes(bytesIn: number, bytesOut: number): void {
this.accumulatedBytesIn += bytesIn;
this.accumulatedBytesOut += bytesOut;
}
// Called periodically (every second)
public takeSample(): void {
const now = Date.now();
// Record accumulated bytes since last sample
this.samples.push({
timestamp: now,
bytesIn: this.accumulatedBytesIn,
bytesOut: this.accumulatedBytesOut
});
// Reset accumulators
this.accumulatedBytesIn = 0;
this.accumulatedBytesOut = 0;
// Trim old samples
const cutoff = now - 3600000; // 1 hour
this.samples = this.samples.filter(s => s.timestamp > cutoff);
}
// Get rate over specified window
public getRate(windowSeconds: number): { bytesInPerSec: number; bytesOutPerSec: number } {
const now = Date.now();
const windowStart = now - (windowSeconds * 1000);
const relevantSamples = this.samples.filter(s => s.timestamp > windowStart);
if (relevantSamples.length === 0) {
return { bytesInPerSec: 0, bytesOutPerSec: 0 };
}
const totalBytesIn = relevantSamples.reduce((sum, s) => sum + s.bytesIn, 0);
const totalBytesOut = relevantSamples.reduce((sum, s) => sum + s.bytesOut, 0);
const actualWindow = (now - relevantSamples[0].timestamp) / 1000;
return {
bytesInPerSec: Math.round(totalBytesIn / actualWindow),
bytesOutPerSec: Math.round(totalBytesOut / actualWindow)
};
}
}
```
### 4. Optimize IP Tracking Memory Usage ✓
- [x] Add periodic cleanup for IPs with no active connections
- [x] Implement expiry for rate limit timestamps
- [x] Add memory-efficient data structures for IP tracking
### B. Connection-Level Byte Tracking
### 5. Add Comprehensive Tests ✓
- [x] Test per-IP limits with HttpProxy forwarding
- [x] Test route-level connection limits
- [x] Test cleanup queue edge cases
- [x] Test memory usage with many unique IPs
```typescript
// In ConnectionRecord, add:
interface IConnectionRecord {
// ... existing fields ...
// Byte counters with timestamps
bytesReceivedHistory: Array<{ timestamp: number; bytes: number }>;
bytesSentHistory: Array<{ timestamp: number; bytes: number }>;
// For efficiency, could use circular buffer
lastBytesReceivedUpdate: number;
lastBytesSentUpdate: number;
}
```
### 6. Log Deduplication for High-Volume Scenarios ✓
- [x] Implement LogDeduplicator utility for batching similar events
- [x] Add deduplication for connection rejections, terminations, and cleanups
- [x] Include rejection reasons in IP rejection summaries
- [x] Provide aggregated summaries with meaningful context
### C. Enhanced Metrics Interface
## Notes
```typescript
interface IMetrics {
// Connection metrics
connections: {
active(): number;
total(): number;
byRoute(): Map<string, number>;
byIP(): Map<string, number>;
topIPs(limit?: number): Array<{ ip: string; count: number }>;
};
// Throughput metrics (bytes per second)
throughput: {
instant(): { in: number; out: number }; // Last 1 second
recent(): { in: number; out: number }; // Last 10 seconds
average(): { in: number; out: number }; // Last 60 seconds
custom(seconds: number): { in: number; out: number };
history(seconds: number): Array<{ timestamp: number; in: number; out: number }>;
byRoute(windowSeconds?: number): Map<string, { in: number; out: number }>;
byIP(windowSeconds?: number): Map<string, { in: number; out: number }>;
};
// Request metrics
requests: {
perSecond(): number;
perMinute(): number;
total(): number;
};
// Cumulative totals
totals: {
bytesIn(): number;
bytesOut(): number;
connections(): number;
};
// Performance metrics
percentiles: {
connectionDuration(): { p50: number; p95: number; p99: number };
bytesTransferred(): {
in: { p50: number; p95: number; p99: number };
out: { p50: number; p95: number; p99: number };
};
};
}
```
## 3. Implementation Plan
### Current Status
- **Phase 1**: ~90% complete (core functionality implemented, tests need fixing)
- **Phase 2**: ~60% complete (main features done, percentiles pending)
- **Phase 3**: ~40% complete (basic optimizations in place)
- **Phase 4**: 0% complete (export formats not started)
### Phase 1: Core Throughput Tracking (Week 1)
- [x] Implement `ThroughputTracker` class
- [x] Integrate byte recording into socket data handlers
- [x] Add periodic sampling (1-second intervals)
- [x] Update `getThroughputRate()` to use time-series data (replaced with new clean API)
- [ ] Add unit tests for throughput tracking
### Phase 2: Enhanced Metrics (Week 2)
- [x] Add configurable time windows (1s, 10s, 60s, 5m, etc.)
- [ ] Implement percentile calculations
- [x] Add route-specific and IP-specific throughput tracking
- [x] Create historical data access methods
- [ ] Add integration tests
### Phase 3: Performance Optimization (Week 3)
- [x] Use circular buffers for efficiency
- [ ] Implement data aggregation for longer time windows
- [x] Add configurable retention periods
- [ ] Optimize memory usage
- [ ] Add performance benchmarks
### Phase 4: Export Formats (Week 4)
- [ ] Add Prometheus metric format with proper metric types
- [ ] Add StatsD format support
- [ ] Add JSON export with metadata
- [ ] Create OpenMetrics compatibility
- [ ] Add documentation and examples
## 4. Key Design Decisions
### A. Sampling Strategy
- **1-second samples** for fine-grained data
- **Aggregate to 1-minute** for longer retention
- **Keep 1 hour** of second-level data
- **Keep 24 hours** of minute-level data
### B. Memory Management
- **Circular buffers** for fixed memory usage
- **Configurable retention** periods
- **Lazy aggregation** for older data
- **Efficient data structures** (typed arrays for samples)
### C. Performance Considerations
- **Batch updates** during high throughput
- **Debounced calculations** for expensive metrics
- **Cached results** with TTL
- **Worker thread** option for heavy calculations
## 5. Configuration Options
```typescript
interface IMetricsConfig {
enabled: boolean;
// Sampling configuration
sampleIntervalMs: number; // Default: 1000 (1 second)
retentionSeconds: number; // Default: 3600 (1 hour)
// Performance tuning
enableDetailedTracking: boolean; // Per-connection byte history
enablePercentiles: boolean; // Calculate percentiles
cacheResultsMs: number; // Cache expensive calculations
// Export configuration
prometheusEnabled: boolean;
prometheusPath: string; // Default: /metrics
prometheusPrefix: string; // Default: smartproxy_
}
```
## 6. Example Usage
```typescript
const proxy = new SmartProxy({
metrics: {
enabled: true,
sampleIntervalMs: 1000,
enableDetailedTracking: true
}
});
// Get metrics instance
const metrics = proxy.getMetrics();
// Connection metrics
console.log(`Active connections: ${metrics.connections.active()}`);
console.log(`Total connections: ${metrics.connections.total()}`);
// Throughput metrics
const instant = metrics.throughput.instant();
console.log(`Current: ${instant.in} bytes/sec in, ${instant.out} bytes/sec out`);
const recent = metrics.throughput.recent(); // Last 10 seconds
const average = metrics.throughput.average(); // Last 60 seconds
// Custom time window
const custom = metrics.throughput.custom(30); // Last 30 seconds
// Historical data for graphing
const history = metrics.throughput.history(300); // Last 5 minutes
history.forEach(point => {
console.log(`${new Date(point.timestamp)}: ${point.in} bytes/sec in, ${point.out} bytes/sec out`);
});
// Top routes by throughput
const routeThroughput = metrics.throughput.byRoute(60);
routeThroughput.forEach((stats, route) => {
console.log(`Route ${route}: ${stats.in} bytes/sec in, ${stats.out} bytes/sec out`);
});
// Request metrics
console.log(`RPS: ${metrics.requests.perSecond()}`);
console.log(`RPM: ${metrics.requests.perMinute()}`);
// Totals
console.log(`Total bytes in: ${metrics.totals.bytesIn()}`);
console.log(`Total bytes out: ${metrics.totals.bytesOut()}`);
```
## 7. Prometheus Export Example
```
# HELP smartproxy_throughput_bytes_per_second Current throughput in bytes per second
# TYPE smartproxy_throughput_bytes_per_second gauge
smartproxy_throughput_bytes_per_second{direction="in",window="1s"} 1234567
smartproxy_throughput_bytes_per_second{direction="out",window="1s"} 987654
smartproxy_throughput_bytes_per_second{direction="in",window="10s"} 1134567
smartproxy_throughput_bytes_per_second{direction="out",window="10s"} 887654
# HELP smartproxy_bytes_total Total bytes transferred
# TYPE smartproxy_bytes_total counter
smartproxy_bytes_total{direction="in"} 123456789
smartproxy_bytes_total{direction="out"} 98765432
# HELP smartproxy_active_connections Current number of active connections
# TYPE smartproxy_active_connections gauge
smartproxy_active_connections 42
# HELP smartproxy_connection_duration_seconds Connection duration in seconds
# TYPE smartproxy_connection_duration_seconds histogram
smartproxy_connection_duration_seconds_bucket{le="0.1"} 100
smartproxy_connection_duration_seconds_bucket{le="1"} 500
smartproxy_connection_duration_seconds_bucket{le="10"} 800
smartproxy_connection_duration_seconds_bucket{le="+Inf"} 850
smartproxy_connection_duration_seconds_sum 4250
smartproxy_connection_duration_seconds_count 850
```
## 8. Migration Strategy
### Breaking Changes
- Completely replace the old metrics API with the new clean design
- Remove all `get*` prefixed methods in favor of grouped properties
- Use simple `{ in, out }` objects instead of verbose property names
- Provide clear migration guide in documentation
### Implementation Approach
1. ✅ Create new `ThroughputTracker` class for time-series data
2. ✅ Implement new `IMetrics` interface with clean API
3. ✅ Replace `MetricsCollector` implementation entirely
4. ✅ Update all references to use new API
5. ⚠️ Add comprehensive tests for accuracy validation (partial)
### Additional Refactoring Completed
- Refactored all SmartProxy components to use cleaner dependency pattern
- Components now receive only `SmartProxy` instance instead of individual dependencies
- Access to other components via `this.smartProxy.componentName`
- Significantly simplified constructor signatures across the codebase
## 9. Success Metrics
- **Accuracy**: Throughput metrics accurate within 1% of actual
- **Performance**: < 1% CPU overhead for metrics collection
- **Memory**: < 10MB memory usage for 1 hour of data
- **Latency**: < 1ms to retrieve any metric
- **Reliability**: No metrics data loss under load
## 10. Future Enhancements
### Phase 5: Advanced Analytics
- Anomaly detection for traffic patterns
- Predictive analytics for capacity planning
- Correlation analysis between routes
- Real-time alerting integration
### Phase 6: Distributed Metrics
- Metrics aggregation across multiple proxies
- Distributed time-series storage
- Cross-proxy analytics
- Global dashboard support
## 11. Risks and Mitigations
### Risk: Memory Usage
- **Mitigation**: Circular buffers and configurable retention
- **Monitoring**: Track memory usage per metric type
### Risk: Performance Impact
- **Mitigation**: Efficient data structures and caching
- **Testing**: Load test with metrics enabled/disabled
### Risk: Data Accuracy
- **Mitigation**: Atomic operations and proper synchronization
- **Validation**: Compare with external monitoring tools
## Conclusion
This plan transforms SmartProxy's metrics from a basic cumulative system to a comprehensive, time-series based monitoring solution suitable for production environments. The phased approach ensures minimal disruption while delivering immediate value through accurate throughput measurements.
- All connection limiting is now consistent across SmartProxy and HttpProxy
- Route-level limits provide additional granular control
- Memory usage is optimized for high-traffic scenarios
- Comprehensive test coverage ensures reliability
- Log deduplication reduces spam during attacks or high-traffic periods
- IP rejection summaries now include rejection reasons in main message

View File

@ -0,0 +1,299 @@
import { expect, tap } from '@git.zone/tstest/tapbundle';
import * as net from 'net';
import { SmartProxy } from '../ts/proxies/smart-proxy/index.js';
import { HttpProxy } from '../ts/proxies/http-proxy/index.js';
let testServer: net.Server;
let smartProxy: SmartProxy;
let httpProxy: HttpProxy;
const TEST_SERVER_PORT = 5100;
const PROXY_PORT = 5101;
const HTTP_PROXY_PORT = 5102;
// Track all created servers and connections for cleanup
const allServers: net.Server[] = [];
const allProxies: (SmartProxy | HttpProxy)[] = [];
const activeConnections: net.Socket[] = [];
// Helper: Creates a test TCP server
function createTestServer(port: number): Promise<net.Server> {
return new Promise((resolve) => {
const server = net.createServer((socket) => {
socket.on('data', (data) => {
socket.write(`Echo: ${data.toString()}`);
});
socket.on('error', () => {});
});
server.listen(port, 'localhost', () => {
console.log(`[Test Server] Listening on localhost:${port}`);
allServers.push(server);
resolve(server);
});
});
}
// Helper: Creates multiple concurrent connections
async function createConcurrentConnections(
port: number,
count: number,
fromIP?: string
): Promise<net.Socket[]> {
const connections: net.Socket[] = [];
const promises: Promise<net.Socket>[] = [];
for (let i = 0; i < count; i++) {
promises.push(
new Promise((resolve, reject) => {
const client = new net.Socket();
const timeout = setTimeout(() => {
client.destroy();
reject(new Error(`Connection ${i} timeout`));
}, 5000);
client.connect(port, 'localhost', () => {
clearTimeout(timeout);
activeConnections.push(client);
connections.push(client);
resolve(client);
});
client.on('error', (err) => {
clearTimeout(timeout);
reject(err);
});
})
);
}
await Promise.all(promises);
return connections;
}
// Helper: Clean up connections
function cleanupConnections(connections: net.Socket[]): void {
connections.forEach(conn => {
if (!conn.destroyed) {
conn.destroy();
}
});
}
tap.test('Setup test environment', async () => {
testServer = await createTestServer(TEST_SERVER_PORT);
// Create SmartProxy with low connection limits for testing
smartProxy = new SmartProxy({
routes: [{
name: 'test-route',
match: {
ports: PROXY_PORT
},
action: {
type: 'forward',
target: {
host: 'localhost',
port: TEST_SERVER_PORT
}
},
security: {
maxConnections: 5 // Low limit for testing
}
}],
maxConnectionsPerIP: 3, // Low per-IP limit
connectionRateLimitPerMinute: 10, // Low rate limit
defaults: {
security: {
maxConnections: 10 // Low global limit
}
}
});
await smartProxy.start();
allProxies.push(smartProxy);
});
tap.test('Per-IP connection limits', async () => {
// Test that we can create up to the per-IP limit
const connections1 = await createConcurrentConnections(PROXY_PORT, 3);
expect(connections1.length).toEqual(3);
// Try to create one more connection - should fail
try {
await createConcurrentConnections(PROXY_PORT, 1);
expect.fail('Should not allow more than 3 connections per IP');
} catch (err) {
expect(err.message).toInclude('ECONNRESET');
}
// Clean up first set of connections
cleanupConnections(connections1);
await new Promise(resolve => setTimeout(resolve, 100));
// Should be able to create new connections after cleanup
const connections2 = await createConcurrentConnections(PROXY_PORT, 2);
expect(connections2.length).toEqual(2);
cleanupConnections(connections2);
});
tap.test('Route-level connection limits', async () => {
// Create multiple connections up to route limit
const connections = await createConcurrentConnections(PROXY_PORT, 5);
expect(connections.length).toEqual(5);
// Try to exceed route limit
try {
await createConcurrentConnections(PROXY_PORT, 1);
expect.fail('Should not allow more than 5 connections for this route');
} catch (err) {
expect(err.message).toInclude('ECONNRESET');
}
cleanupConnections(connections);
});
tap.test('Connection rate limiting', async () => {
// Create connections rapidly
const connections: net.Socket[] = [];
// Create 10 connections rapidly (at rate limit)
for (let i = 0; i < 10; i++) {
try {
const conn = await createConcurrentConnections(PROXY_PORT, 1);
connections.push(...conn);
// Small delay to avoid per-IP limit
if (connections.length >= 3) {
cleanupConnections(connections.splice(0, 3));
await new Promise(resolve => setTimeout(resolve, 50));
}
} catch (err) {
// Expected to fail at some point due to rate limit
expect(i).toBeGreaterThan(0);
break;
}
}
cleanupConnections(connections);
});
tap.test('HttpProxy per-IP validation', async () => {
// Create HttpProxy
httpProxy = new HttpProxy({
port: HTTP_PROXY_PORT,
maxConnectionsPerIP: 2,
connectionRateLimitPerMinute: 10,
routes: []
});
await httpProxy.start();
allProxies.push(httpProxy);
// Update SmartProxy to use HttpProxy for TLS termination
await smartProxy.stop();
smartProxy = new SmartProxy({
routes: [{
name: 'https-route',
match: {
ports: PROXY_PORT + 10
},
action: {
type: 'forward',
target: {
host: 'localhost',
port: TEST_SERVER_PORT
},
tls: {
mode: 'terminate'
}
}
}],
useHttpProxy: [PROXY_PORT + 10],
httpProxyPort: HTTP_PROXY_PORT,
maxConnectionsPerIP: 3
});
await smartProxy.start();
// Test that HttpProxy enforces its own per-IP limits
const connections = await createConcurrentConnections(PROXY_PORT + 10, 2);
expect(connections.length).toEqual(2);
// Should reject additional connections
try {
await createConcurrentConnections(PROXY_PORT + 10, 1);
expect.fail('HttpProxy should enforce per-IP limits');
} catch (err) {
expect(err.message).toInclude('ECONNRESET');
}
cleanupConnections(connections);
});
tap.test('IP tracking cleanup', async (tools) => {
// Create and close many connections from different IPs
const connections: net.Socket[] = [];
for (let i = 0; i < 5; i++) {
const conn = await createConcurrentConnections(PROXY_PORT, 1);
connections.push(...conn);
}
// Close all connections
cleanupConnections(connections);
// Wait for cleanup interval (set to 60s in production, but we'll check immediately)
await tools.delayFor(100);
// Verify that IP tracking has been cleaned up
const securityManager = (smartProxy as any).securityManager;
const ipCount = (securityManager.connectionsByIP as Map<string, any>).size;
// Should have no IPs tracked after cleanup
expect(ipCount).toEqual(0);
});
tap.test('Cleanup queue race condition handling', async () => {
// Create many connections concurrently to trigger batched cleanup
const promises: Promise<net.Socket[]>[] = [];
for (let i = 0; i < 20; i++) {
promises.push(createConcurrentConnections(PROXY_PORT, 1).catch(() => []));
}
const results = await Promise.all(promises);
const allConnections = results.flat();
// Close all connections rapidly
allConnections.forEach(conn => conn.destroy());
// Give cleanup queue time to process
await new Promise(resolve => setTimeout(resolve, 500));
// Verify all connections were cleaned up
const connectionManager = (smartProxy as any).connectionManager;
const remainingConnections = connectionManager.getConnectionCount();
expect(remainingConnections).toEqual(0);
});
tap.test('Cleanup and shutdown', async () => {
// Clean up any remaining connections
cleanupConnections(activeConnections);
activeConnections.length = 0;
// Stop all proxies
for (const proxy of allProxies) {
await proxy.stop();
}
allProxies.length = 0;
// Close all test servers
for (const server of allServers) {
await new Promise<void>((resolve) => {
server.close(() => resolve());
});
}
allServers.length = 0;
});
tap.start();

View File

@ -0,0 +1,120 @@
import { expect, tap } from '@git.zone/tstest/tapbundle';
import { SecurityManager } from '../ts/proxies/http-proxy/security-manager.js';
import { createLogger } from '../ts/proxies/http-proxy/models/types.js';
let securityManager: SecurityManager;
const logger = createLogger('error'); // Quiet logger for tests
tap.test('Setup HttpProxy SecurityManager', async () => {
securityManager = new SecurityManager(logger, [], 3, 10); // Low limits for testing
});
tap.test('HttpProxy IP connection tracking', async () => {
const testIP = '10.0.0.1';
// Track connections
securityManager.trackConnectionByIP(testIP, 'http-conn1');
securityManager.trackConnectionByIP(testIP, 'http-conn2');
expect(securityManager.getConnectionCountByIP(testIP)).toEqual(2);
// Validate IP should pass
let result = securityManager.validateIP(testIP);
expect(result.allowed).toBeTrue();
// Add one more to reach limit
securityManager.trackConnectionByIP(testIP, 'http-conn3');
// Should now reject new connections
result = securityManager.validateIP(testIP);
expect(result.allowed).toBeFalse();
expect(result.reason).toInclude('Maximum connections per IP (3) exceeded');
// Remove a connection
securityManager.removeConnectionByIP(testIP, 'http-conn1');
// Should allow connections again
result = securityManager.validateIP(testIP);
expect(result.allowed).toBeTrue();
// Clean up
securityManager.removeConnectionByIP(testIP, 'http-conn2');
securityManager.removeConnectionByIP(testIP, 'http-conn3');
});
tap.test('HttpProxy connection rate limiting', async () => {
const testIP = '10.0.0.2';
// Make 10 connections rapidly (at rate limit)
for (let i = 0; i < 10; i++) {
const result = securityManager.validateIP(testIP);
expect(result.allowed).toBeTrue();
// Track the connection to simulate real usage
securityManager.trackConnectionByIP(testIP, `rate-conn${i}`);
}
// 11th connection should be rate limited
const result = securityManager.validateIP(testIP);
expect(result.allowed).toBeFalse();
expect(result.reason).toInclude('Connection rate limit (10/min) exceeded');
// Clean up
for (let i = 0; i < 10; i++) {
securityManager.removeConnectionByIP(testIP, `rate-conn${i}`);
}
});
tap.test('HttpProxy CLIENT_IP header handling', async () => {
// This tests the scenario where SmartProxy forwards the real client IP
const realClientIP = '203.0.113.1';
const proxyIP = '127.0.0.1';
// Simulate SmartProxy tracking the real client IP
securityManager.trackConnectionByIP(realClientIP, 'forwarded-conn1');
securityManager.trackConnectionByIP(realClientIP, 'forwarded-conn2');
securityManager.trackConnectionByIP(realClientIP, 'forwarded-conn3');
// Real client IP should be at limit
let result = securityManager.validateIP(realClientIP);
expect(result.allowed).toBeFalse();
// But proxy IP should still be allowed
result = securityManager.validateIP(proxyIP);
expect(result.allowed).toBeTrue();
// Clean up
securityManager.removeConnectionByIP(realClientIP, 'forwarded-conn1');
securityManager.removeConnectionByIP(realClientIP, 'forwarded-conn2');
securityManager.removeConnectionByIP(realClientIP, 'forwarded-conn3');
});
tap.test('HttpProxy automatic cleanup', async (tools) => {
const testIP = '10.0.0.3';
// Create and immediately remove connections
for (let i = 0; i < 5; i++) {
securityManager.trackConnectionByIP(testIP, `cleanup-conn${i}`);
securityManager.removeConnectionByIP(testIP, `cleanup-conn${i}`);
}
// Add rate limit entries
for (let i = 0; i < 5; i++) {
securityManager.validateIP(testIP);
}
// Wait a bit (cleanup runs every 60 seconds in production)
// For testing, we'll just verify the cleanup logic works
await tools.delayFor(100);
// Manually trigger cleanup (in production this happens automatically)
(securityManager as any).performIpCleanup();
// IP should be cleaned up
expect(securityManager.getConnectionCountByIP(testIP)).toEqual(0);
});
tap.test('Cleanup HttpProxy SecurityManager', async () => {
securityManager.clearIPTracking();
});
tap.start();

View File

@ -0,0 +1,112 @@
import { expect, tap } from '@git.zone/tstest/tapbundle';
import { LogDeduplicator } from '../ts/core/utils/log-deduplicator.js';
let deduplicator: LogDeduplicator;
tap.test('Setup log deduplicator', async () => {
deduplicator = new LogDeduplicator(1000); // 1 second flush interval for testing
});
tap.test('Connection rejection deduplication', async (tools) => {
// Simulate multiple connection rejections
for (let i = 0; i < 10; i++) {
deduplicator.log(
'connection-rejected',
'warn',
'Connection rejected',
{ reason: 'global-limit', component: 'test' },
'global-limit'
);
}
for (let i = 0; i < 5; i++) {
deduplicator.log(
'connection-rejected',
'warn',
'Connection rejected',
{ reason: 'route-limit', component: 'test' },
'route-limit'
);
}
// Force flush
deduplicator.flush('connection-rejected');
// The logs should have been aggregated
// (Can't easily test the actual log output, but we can verify the mechanism works)
expect(deduplicator).toBeInstanceOf(LogDeduplicator);
});
tap.test('IP rejection deduplication', async (tools) => {
// Simulate rejections from multiple IPs
const ips = ['192.168.1.100', '192.168.1.101', '192.168.1.100', '10.0.0.1'];
const reasons = ['per-ip-limit', 'rate-limit', 'per-ip-limit', 'global-limit'];
for (let i = 0; i < ips.length; i++) {
deduplicator.log(
'ip-rejected',
'warn',
`Connection rejected from ${ips[i]}`,
{ remoteIP: ips[i], reason: reasons[i] },
ips[i]
);
}
// Add more rejections from the same IP
for (let i = 0; i < 20; i++) {
deduplicator.log(
'ip-rejected',
'warn',
'Connection rejected from 192.168.1.100',
{ remoteIP: '192.168.1.100', reason: 'rate-limit' },
'192.168.1.100'
);
}
// Force flush
deduplicator.flush('ip-rejected');
// Verify the deduplicator exists and works
expect(deduplicator).toBeInstanceOf(LogDeduplicator);
});
tap.test('Connection cleanup deduplication', async (tools) => {
// Simulate various cleanup events
const reasons = ['normal', 'timeout', 'error', 'normal', 'zombie'];
for (const reason of reasons) {
for (let i = 0; i < 5; i++) {
deduplicator.log(
'connection-cleanup',
'info',
`Connection cleanup: ${reason}`,
{ connectionId: `conn-${i}`, reason },
reason
);
}
}
// Wait for automatic flush
await tools.delayFor(1500);
// Verify deduplicator is working
expect(deduplicator).toBeInstanceOf(LogDeduplicator);
});
tap.test('Automatic periodic flush', async (tools) => {
// Add some events
deduplicator.log('test-event', 'info', 'Test message', {}, 'test');
// Wait for automatic flush (should happen within 2x flush interval = 2 seconds)
await tools.delayFor(2500);
// Events should have been flushed automatically
expect(deduplicator).toBeInstanceOf(LogDeduplicator);
});
tap.test('Cleanup deduplicator', async () => {
deduplicator.cleanup();
expect(deduplicator).toBeInstanceOf(LogDeduplicator);
});
tap.start();

View File

@ -0,0 +1,159 @@
import { expect, tap } from '@git.zone/tstest/tapbundle';
import { SharedSecurityManager } from '../ts/core/utils/shared-security-manager.js';
import type { IRouteConfig, IRouteContext } from '../ts/proxies/smart-proxy/models/route-types.js';
let securityManager: SharedSecurityManager;
tap.test('Setup SharedSecurityManager', async () => {
securityManager = new SharedSecurityManager({
maxConnectionsPerIP: 5,
connectionRateLimitPerMinute: 10,
cleanupIntervalMs: 1000 // 1 second for faster testing
});
});
tap.test('IP connection tracking', async () => {
const testIP = '192.168.1.100';
// Track multiple connections
securityManager.trackConnectionByIP(testIP, 'conn1');
securityManager.trackConnectionByIP(testIP, 'conn2');
securityManager.trackConnectionByIP(testIP, 'conn3');
// Verify connection count
expect(securityManager.getConnectionCountByIP(testIP)).toEqual(3);
// Remove a connection
securityManager.removeConnectionByIP(testIP, 'conn2');
expect(securityManager.getConnectionCountByIP(testIP)).toEqual(2);
// Remove remaining connections
securityManager.removeConnectionByIP(testIP, 'conn1');
securityManager.removeConnectionByIP(testIP, 'conn3');
expect(securityManager.getConnectionCountByIP(testIP)).toEqual(0);
});
tap.test('Per-IP connection limits validation', async () => {
const testIP = '192.168.1.101';
// Track connections up to limit
for (let i = 1; i <= 5; i++) {
securityManager.trackConnectionByIP(testIP, `conn${i}`);
const result = securityManager.validateIP(testIP);
expect(result.allowed).toBeTrue();
}
// Verify we're at the limit
expect(securityManager.getConnectionCountByIP(testIP)).toEqual(5);
// Next connection should be rejected
const result = securityManager.validateIP(testIP);
expect(result.allowed).toBeFalse();
expect(result.reason).toInclude('Maximum connections per IP');
// Clean up
for (let i = 1; i <= 5; i++) {
securityManager.removeConnectionByIP(testIP, `conn${i}`);
}
});
tap.test('Connection rate limiting', async () => {
const testIP = '192.168.1.102';
// Make connections at the rate limit
for (let i = 0; i < 10; i++) {
const result = securityManager.validateIP(testIP);
expect(result.allowed).toBeTrue();
securityManager.trackConnectionByIP(testIP, `conn${i}`);
}
// Next connection should exceed rate limit
const result = securityManager.validateIP(testIP);
expect(result.allowed).toBeFalse();
expect(result.reason).toInclude('Connection rate limit');
// Clean up connections
for (let i = 0; i < 10; i++) {
securityManager.removeConnectionByIP(testIP, `conn${i}`);
}
});
tap.test('Route-level connection limits', async () => {
const route: IRouteConfig = {
name: 'test-route',
match: { ports: 443 },
action: { type: 'forward', target: { host: 'localhost', port: 8080 } },
security: {
maxConnections: 3
}
};
const context: IRouteContext = {
port: 443,
clientIp: '192.168.1.103',
serverIp: '0.0.0.0',
timestamp: Date.now(),
connectionId: 'test-conn'
};
// Test with connection counts below limit
expect(securityManager.isAllowed(route, context, 0)).toBeTrue();
expect(securityManager.isAllowed(route, context, 2)).toBeTrue();
// Test at limit
expect(securityManager.isAllowed(route, context, 3)).toBeFalse();
// Test above limit
expect(securityManager.isAllowed(route, context, 5)).toBeFalse();
});
tap.test('IPv4/IPv6 normalization', async () => {
const ipv4 = '127.0.0.1';
const ipv4Mapped = '::ffff:127.0.0.1';
// Track connection with IPv4
securityManager.trackConnectionByIP(ipv4, 'conn1');
// Both representations should show the same connection
expect(securityManager.getConnectionCountByIP(ipv4)).toEqual(1);
expect(securityManager.getConnectionCountByIP(ipv4Mapped)).toEqual(1);
// Track another connection with IPv6 representation
securityManager.trackConnectionByIP(ipv4Mapped, 'conn2');
// Both should show 2 connections
expect(securityManager.getConnectionCountByIP(ipv4)).toEqual(2);
expect(securityManager.getConnectionCountByIP(ipv4Mapped)).toEqual(2);
// Clean up
securityManager.removeConnectionByIP(ipv4, 'conn1');
securityManager.removeConnectionByIP(ipv4Mapped, 'conn2');
});
tap.test('Automatic cleanup of expired data', async (tools) => {
const testIP = '192.168.1.104';
// Track a connection and then remove it
securityManager.trackConnectionByIP(testIP, 'temp-conn');
securityManager.removeConnectionByIP(testIP, 'temp-conn');
// Add some rate limit entries (they expire after 1 minute)
for (let i = 0; i < 5; i++) {
securityManager.validateIP(testIP);
}
// Wait for cleanup interval (set to 1 second in our test)
await tools.delayFor(1500);
// The IP should be cleaned up since it has no connections
// Note: We can't directly check the internal map, but we can verify
// that a new connection is allowed (fresh rate limit)
const result = securityManager.validateIP(testIP);
expect(result.allowed).toBeTrue();
});
tap.test('Cleanup SharedSecurityManager', async () => {
securityManager.clearIPTracking();
});
tap.start();

View File

@ -0,0 +1,370 @@
import { logger } from './logger.js';
interface ILogEvent {
level: 'info' | 'warn' | 'error' | 'debug';
message: string;
data?: any;
count: number;
firstSeen: number;
lastSeen: number;
}
interface IAggregatedEvent {
key: string;
events: Map<string, ILogEvent>;
flushTimer?: NodeJS.Timeout;
}
/**
* Log deduplication utility to reduce log spam for repetitive events
*/
export class LogDeduplicator {
private globalFlushTimer?: NodeJS.Timeout;
private aggregatedEvents: Map<string, IAggregatedEvent> = new Map();
private flushInterval: number = 5000; // 5 seconds
private maxBatchSize: number = 100;
private rapidEventThreshold: number = 50; // Flush early if this many events in 1 second
private lastRapidCheck: number = Date.now();
constructor(flushInterval?: number) {
if (flushInterval) {
this.flushInterval = flushInterval;
}
// Set up global periodic flush to ensure logs are emitted regularly
this.globalFlushTimer = setInterval(() => {
this.flushAll();
}, this.flushInterval * 2); // Flush everything every 2x the normal interval
if (this.globalFlushTimer.unref) {
this.globalFlushTimer.unref();
}
}
/**
* Log a deduplicated event
* @param key - Aggregation key (e.g., 'connection-rejected', 'cleanup-batch')
* @param level - Log level
* @param message - Log message template
* @param data - Additional data
* @param dedupeKey - Deduplication key within the aggregation (e.g., IP address, reason)
*/
public log(
key: string,
level: 'info' | 'warn' | 'error' | 'debug',
message: string,
data?: any,
dedupeKey?: string
): void {
const eventKey = dedupeKey || message;
const now = Date.now();
if (!this.aggregatedEvents.has(key)) {
this.aggregatedEvents.set(key, {
key,
events: new Map(),
flushTimer: undefined
});
}
const aggregated = this.aggregatedEvents.get(key)!;
if (aggregated.events.has(eventKey)) {
const event = aggregated.events.get(eventKey)!;
event.count++;
event.lastSeen = now;
if (data) {
event.data = { ...event.data, ...data };
}
} else {
aggregated.events.set(eventKey, {
level,
message,
data,
count: 1,
firstSeen: now,
lastSeen: now
});
}
// Check for rapid events (many events in short time)
const totalEvents = Array.from(aggregated.events.values()).reduce((sum, e) => sum + e.count, 0);
// If we're getting flooded with events, flush more frequently
if (now - this.lastRapidCheck < 1000 && totalEvents >= this.rapidEventThreshold) {
this.flush(key);
this.lastRapidCheck = now;
} else if (aggregated.events.size >= this.maxBatchSize) {
// Check if we should flush due to size
this.flush(key);
} else if (!aggregated.flushTimer) {
// Schedule flush
aggregated.flushTimer = setTimeout(() => {
this.flush(key);
}, this.flushInterval);
if (aggregated.flushTimer.unref) {
aggregated.flushTimer.unref();
}
}
// Update rapid check time
if (now - this.lastRapidCheck >= 1000) {
this.lastRapidCheck = now;
}
}
/**
* Flush aggregated events for a specific key
*/
public flush(key: string): void {
const aggregated = this.aggregatedEvents.get(key);
if (!aggregated || aggregated.events.size === 0) {
return;
}
if (aggregated.flushTimer) {
clearTimeout(aggregated.flushTimer);
aggregated.flushTimer = undefined;
}
// Emit aggregated log based on the key
switch (key) {
case 'connection-rejected':
this.flushConnectionRejections(aggregated);
break;
case 'connection-cleanup':
this.flushConnectionCleanups(aggregated);
break;
case 'connection-terminated':
this.flushConnectionTerminations(aggregated);
break;
case 'ip-rejected':
this.flushIPRejections(aggregated);
break;
default:
this.flushGeneric(aggregated);
}
// Clear events
aggregated.events.clear();
}
/**
* Flush all pending events
*/
public flushAll(): void {
for (const key of this.aggregatedEvents.keys()) {
this.flush(key);
}
}
private flushConnectionRejections(aggregated: IAggregatedEvent): void {
const totalCount = Array.from(aggregated.events.values()).reduce((sum, e) => sum + e.count, 0);
const byReason = new Map<string, number>();
for (const [, event] of aggregated.events) {
const reason = event.data?.reason || 'unknown';
byReason.set(reason, (byReason.get(reason) || 0) + event.count);
}
const reasonSummary = Array.from(byReason.entries())
.sort((a, b) => b[1] - a[1])
.map(([reason, count]) => `${reason}: ${count}`)
.join(', ');
const duration = Date.now() - Math.min(...Array.from(aggregated.events.values()).map(e => e.firstSeen));
logger.log('warn', `[SUMMARY] Rejected ${totalCount} connections in ${Math.round(duration/1000)}s`, {
reasons: reasonSummary,
uniqueIPs: aggregated.events.size,
component: 'connection-dedup'
});
}
private flushConnectionCleanups(aggregated: IAggregatedEvent): void {
const totalCount = Array.from(aggregated.events.values()).reduce((sum, e) => sum + e.count, 0);
const byReason = new Map<string, number>();
for (const [, event] of aggregated.events) {
const reason = event.data?.reason || 'normal';
byReason.set(reason, (byReason.get(reason) || 0) + event.count);
}
const reasonSummary = Array.from(byReason.entries())
.sort((a, b) => b[1] - a[1])
.slice(0, 5) // Top 5 reasons
.map(([reason, count]) => `${reason}: ${count}`)
.join(', ');
logger.log('info', `Cleaned up ${totalCount} connections`, {
reasons: reasonSummary,
duration: Date.now() - Math.min(...Array.from(aggregated.events.values()).map(e => e.firstSeen)),
component: 'connection-dedup'
});
}
private flushConnectionTerminations(aggregated: IAggregatedEvent): void {
const totalCount = Array.from(aggregated.events.values()).reduce((sum, e) => sum + e.count, 0);
const byReason = new Map<string, number>();
const byIP = new Map<string, number>();
let lastActiveCount = 0;
for (const [, event] of aggregated.events) {
const reason = event.data?.reason || 'unknown';
const ip = event.data?.remoteIP || 'unknown';
byReason.set(reason, (byReason.get(reason) || 0) + event.count);
// Track by IP
if (ip !== 'unknown') {
byIP.set(ip, (byIP.get(ip) || 0) + event.count);
}
// Track the last active connection count
if (event.data?.activeConnections !== undefined) {
lastActiveCount = event.data.activeConnections;
}
}
const reasonSummary = Array.from(byReason.entries())
.sort((a, b) => b[1] - a[1])
.slice(0, 5) // Top 5 reasons
.map(([reason, count]) => `${reason}: ${count}`)
.join(', ');
// Show top IPs if there are many different ones
let ipInfo = '';
if (byIP.size > 3) {
const topIPs = Array.from(byIP.entries())
.sort((a, b) => b[1] - a[1])
.slice(0, 3)
.map(([ip, count]) => `${ip} (${count})`)
.join(', ');
ipInfo = `, from ${byIP.size} IPs (top: ${topIPs})`;
} else if (byIP.size > 0) {
ipInfo = `, IPs: ${Array.from(byIP.keys()).join(', ')}`;
}
const duration = Date.now() - Math.min(...Array.from(aggregated.events.values()).map(e => e.firstSeen));
// Special handling for localhost connections (HttpProxy)
const localhostCount = byIP.get('::ffff:127.0.0.1') || 0;
if (localhostCount > 0 && byIP.size === 1) {
// All connections are from localhost (HttpProxy)
logger.log('info', `[SUMMARY] ${totalCount} HttpProxy connections terminated in ${Math.round(duration/1000)}s`, {
reasons: reasonSummary,
activeConnections: lastActiveCount,
component: 'connection-dedup'
});
} else {
logger.log('info', `[SUMMARY] ${totalCount} connections terminated in ${Math.round(duration/1000)}s`, {
reasons: reasonSummary,
activeConnections: lastActiveCount,
uniqueReasons: byReason.size,
...(ipInfo ? { ips: ipInfo } : {}),
component: 'connection-dedup'
});
}
}
private flushIPRejections(aggregated: IAggregatedEvent): void {
const byIP = new Map<string, { count: number; reasons: Set<string> }>();
const allReasons = new Map<string, number>();
for (const [ip, event] of aggregated.events) {
if (!byIP.has(ip)) {
byIP.set(ip, { count: 0, reasons: new Set() });
}
const ipData = byIP.get(ip)!;
ipData.count += event.count;
if (event.data?.reason) {
ipData.reasons.add(event.data.reason);
// Track overall reason counts
allReasons.set(event.data.reason, (allReasons.get(event.data.reason) || 0) + event.count);
}
}
// Create reason summary
const reasonSummary = Array.from(allReasons.entries())
.sort((a, b) => b[1] - a[1])
.map(([reason, count]) => `${reason}: ${count}`)
.join(', ');
// Log top offenders
const topOffenders = Array.from(byIP.entries())
.sort((a, b) => b[1].count - a[1].count)
.slice(0, 10)
.map(([ip, data]) => `${ip} (${data.count}x, ${Array.from(data.reasons).join('/')})`)
.join(', ');
const totalRejections = Array.from(byIP.values()).reduce((sum, data) => sum + data.count, 0);
const duration = Date.now() - Math.min(...Array.from(aggregated.events.values()).map(e => e.firstSeen));
logger.log('warn', `[SUMMARY] Rejected ${totalRejections} connections from ${byIP.size} IPs in ${Math.round(duration/1000)}s (${reasonSummary})`, {
topOffenders,
component: 'ip-dedup'
});
}
private flushGeneric(aggregated: IAggregatedEvent): void {
const totalCount = Array.from(aggregated.events.values()).reduce((sum, e) => sum + e.count, 0);
const level = aggregated.events.values().next().value?.level || 'info';
// Special handling for IP cleanup events
if (aggregated.key === 'ip-cleanup') {
const totalCleaned = Array.from(aggregated.events.values()).reduce((sum, e) => {
return sum + (e.data?.cleanedIPs || 0) + (e.data?.cleanedRateLimits || 0);
}, 0);
if (totalCleaned > 0) {
logger.log(level as any, `IP tracking cleanup: removed ${totalCleaned} entries across ${totalCount} cleanup cycles`, {
duration: Date.now() - Math.min(...Array.from(aggregated.events.values()).map(e => e.firstSeen)),
component: 'log-dedup'
});
}
} else {
logger.log(level as any, `${aggregated.key}: ${totalCount} events`, {
uniqueEvents: aggregated.events.size,
duration: Date.now() - Math.min(...Array.from(aggregated.events.values()).map(e => e.firstSeen)),
component: 'log-dedup'
});
}
}
/**
* Cleanup and stop deduplication
*/
public cleanup(): void {
this.flushAll();
if (this.globalFlushTimer) {
clearInterval(this.globalFlushTimer);
this.globalFlushTimer = undefined;
}
for (const aggregated of this.aggregatedEvents.values()) {
if (aggregated.flushTimer) {
clearTimeout(aggregated.flushTimer);
}
}
this.aggregatedEvents.clear();
}
}
// Global instance for connection-related log deduplication
export const connectionLogDeduplicator = new LogDeduplicator(5000); // 5 second batches
// Ensure logs are flushed on process exit
process.on('beforeExit', () => {
connectionLogDeduplicator.flushAll();
});
process.on('SIGINT', () => {
connectionLogDeduplicator.cleanup();
process.exit(0);
});
process.on('SIGTERM', () => {
connectionLogDeduplicator.cleanup();
process.exit(0);
});

View File

@ -152,9 +152,10 @@ export class SharedSecurityManager {
*
* @param route - The route to check
* @param context - The request context
* @param routeConnectionCount - Current connection count for this route (optional)
* @returns Whether access is allowed
*/
public isAllowed(route: IRouteConfig, context: IRouteContext): boolean {
public isAllowed(route: IRouteConfig, context: IRouteContext, routeConnectionCount?: number): boolean {
if (!route.security) {
return true; // No security restrictions
}
@ -165,6 +166,14 @@ export class SharedSecurityManager {
return false;
}
// --- Route-level connection limit ---
if (route.security.maxConnections !== undefined && routeConnectionCount !== undefined) {
if (routeConnectionCount >= route.security.maxConnections) {
this.logger?.debug?.(`Route connection limit (${route.security.maxConnections}) exceeded for route ${route.name || 'unnamed'}`);
return false;
}
}
// --- Rate limiting ---
if (route.security.rateLimit?.enabled && !this.isWithinRateLimit(route, context)) {
this.logger?.debug?.(`Rate limit exceeded for route ${route.name || 'unnamed'}`);
@ -304,6 +313,20 @@ export class SharedSecurityManager {
// Clean up rate limits
cleanupExpiredRateLimits(this.rateLimits, this.logger);
// Clean up IP connection tracking
let cleanedIPs = 0;
for (const [ip, info] of this.connectionsByIP.entries()) {
// Remove IPs with no active connections and no recent timestamps
if (info.connections.size === 0 && info.timestamps.length === 0) {
this.connectionsByIP.delete(ip);
cleanedIPs++;
}
}
if (cleanedIPs > 0 && this.logger?.debug) {
this.logger.debug(`Cleaned up ${cleanedIPs} IPs with no active connections`);
}
// IP filter cache doesn't need cleanup (tied to routes)
}

View File

@ -17,6 +17,8 @@ import { WebSocketHandler } from './websocket-handler.js';
import { HttpRouter } from '../../routing/router/index.js';
import { cleanupSocket } from '../../core/utils/socket-utils.js';
import { FunctionCache } from './function-cache.js';
import { SecurityManager } from './security-manager.js';
import { connectionLogDeduplicator } from '../../core/utils/log-deduplicator.js';
/**
* HttpProxy provides a reverse proxy with TLS termination, WebSocket support,
@ -43,6 +45,7 @@ export class HttpProxy implements IMetricsTracker {
private router = new HttpRouter(); // Unified HTTP router
private routeManager: RouteManager;
private functionCache: FunctionCache;
private securityManager: SecurityManager;
// State tracking
public socketMap = new plugins.lik.ObjectMap<plugins.net.Socket>();
@ -113,6 +116,14 @@ export class HttpProxy implements IMetricsTracker {
maxCacheSize: this.options.functionCacheSize || 1000,
defaultTtl: this.options.functionCacheTtl || 5000
});
// Initialize security manager
this.securityManager = new SecurityManager(
this.logger,
[],
this.options.maxConnectionsPerIP || 100,
this.options.connectionRateLimitPerMinute || 300
);
// Initialize other components
this.certificateManager = new CertificateManager(this.options);
@ -269,14 +280,113 @@ export class HttpProxy implements IMetricsTracker {
*/
private setupConnectionTracking(): void {
this.httpsServer.on('connection', (connection: plugins.net.Socket) => {
// Check if max connections reached
let remoteIP = connection.remoteAddress || '';
const connectionId = Math.random().toString(36).substring(2, 15);
const isFromSmartProxy = this.options.portProxyIntegration && connection.remoteAddress?.includes('127.0.0.1');
// For SmartProxy connections, wait for CLIENT_IP header
if (isFromSmartProxy) {
let headerBuffer = Buffer.alloc(0);
let headerParsed = false;
const parseHeader = (data: Buffer) => {
if (headerParsed) return data;
headerBuffer = Buffer.concat([headerBuffer, data]);
const headerStr = headerBuffer.toString();
const headerEnd = headerStr.indexOf('\r\n');
if (headerEnd !== -1) {
const header = headerStr.substring(0, headerEnd);
if (header.startsWith('CLIENT_IP:')) {
remoteIP = header.substring(10); // Extract IP after "CLIENT_IP:"
this.logger.debug(`Extracted client IP from SmartProxy: ${remoteIP}`);
}
headerParsed = true;
// Store the real IP on the connection
(connection as any)._realRemoteIP = remoteIP;
// Validate the real IP
const ipValidation = this.securityManager.validateIP(remoteIP);
if (!ipValidation.allowed) {
connectionLogDeduplicator.log(
'ip-rejected',
'warn',
`HttpProxy connection rejected (via SmartProxy)`,
{ remoteIP, reason: ipValidation.reason, component: 'http-proxy' },
remoteIP
);
connection.destroy();
return null;
}
// Track connection by real IP
this.securityManager.trackConnectionByIP(remoteIP, connectionId);
// Return remaining data after header
return headerBuffer.slice(headerEnd + 2);
}
return null;
};
// Override the first data handler to parse header
const originalEmit = connection.emit;
connection.emit = function(event: string, ...args: any[]) {
if (event === 'data' && !headerParsed) {
const remaining = parseHeader(args[0]);
if (remaining && remaining.length > 0) {
// Call original emit with remaining data
return originalEmit.apply(connection, ['data', remaining]);
} else if (headerParsed) {
// Header parsed but no remaining data
return true;
}
// Header not complete yet, suppress this data event
return true;
}
return originalEmit.apply(connection, [event, ...args]);
} as any;
} else {
// Direct connection - validate immediately
const ipValidation = this.securityManager.validateIP(remoteIP);
if (!ipValidation.allowed) {
connectionLogDeduplicator.log(
'ip-rejected',
'warn',
`HttpProxy connection rejected`,
{ remoteIP, reason: ipValidation.reason, component: 'http-proxy' },
remoteIP
);
connection.destroy();
return;
}
// Track connection by IP
this.securityManager.trackConnectionByIP(remoteIP, connectionId);
}
// Then check global max connections
if (this.socketMap.getArray().length >= this.options.maxConnections) {
this.logger.warn(`Max connections (${this.options.maxConnections}) reached, rejecting new connection`);
connectionLogDeduplicator.log(
'connection-rejected',
'warn',
'HttpProxy max connections reached',
{
reason: 'global-limit',
currentConnections: this.socketMap.getArray().length,
maxConnections: this.options.maxConnections,
component: 'http-proxy'
},
'http-proxy-global-limit'
);
connection.destroy();
return;
}
// Add connection to tracking
// Add connection to tracking with metadata
(connection as any)._connectionId = connectionId;
(connection as any)._remoteIP = remoteIP;
this.socketMap.add(connection);
this.connectedClients = this.socketMap.getArray().length;
@ -284,12 +394,12 @@ export class HttpProxy implements IMetricsTracker {
const localPort = connection.localPort || 0;
const remotePort = connection.remotePort || 0;
// If this connection is from a SmartProxy (usually indicated by it coming from localhost)
if (this.options.portProxyIntegration && connection.remoteAddress?.includes('127.0.0.1')) {
// If this connection is from a SmartProxy
if (isFromSmartProxy) {
this.portProxyConnections++;
this.logger.debug(`New connection from SmartProxy (local: ${localPort}, remote: ${remotePort})`);
this.logger.debug(`New connection from SmartProxy for client ${remoteIP} (local: ${localPort}, remote: ${remotePort})`);
} else {
this.logger.debug(`New direct connection (local: ${localPort}, remote: ${remotePort})`);
this.logger.debug(`New direct connection from ${remoteIP} (local: ${localPort}, remote: ${remotePort})`);
}
// Setup connection cleanup handlers
@ -298,12 +408,19 @@ export class HttpProxy implements IMetricsTracker {
this.socketMap.remove(connection);
this.connectedClients = this.socketMap.getArray().length;
// Remove IP tracking
const connId = (connection as any)._connectionId;
const connIP = (connection as any)._realRemoteIP || (connection as any)._remoteIP;
if (connId && connIP) {
this.securityManager.removeConnectionByIP(connIP, connId);
}
// If this was a SmartProxy connection, decrement the counter
if (this.options.portProxyIntegration && connection.remoteAddress?.includes('127.0.0.1')) {
this.portProxyConnections--;
}
this.logger.debug(`Connection closed. ${this.connectedClients} connections remaining`);
this.logger.debug(`Connection closed from ${connIP || 'unknown'}. ${this.connectedClients} connections remaining`);
}
};
@ -480,6 +597,9 @@ export class HttpProxy implements IMetricsTracker {
// Certificate management cleanup is handled by SmartCertManager
// Flush any pending deduplicated logs
connectionLogDeduplicator.flushAll();
// Close the HTTPS server
return new Promise((resolve) => {
this.httpsServer.close(() => {

View File

@ -45,6 +45,10 @@ export interface IHttpProxyOptions {
// Direct route configurations
routes?: IRouteConfig[];
// Rate limiting and security
maxConnectionsPerIP?: number; // Maximum simultaneous connections from a single IP
connectionRateLimitPerMinute?: number; // Max new connections per minute from a single IP
}
/**

View File

@ -14,7 +14,14 @@ export class SecurityManager {
// Store rate limits per route and key
private rateLimits: Map<string, Map<string, { count: number, expiry: number }>> = new Map();
constructor(private logger: ILogger, private routes: IRouteConfig[] = []) {}
// Connection tracking by IP
private connectionsByIP: Map<string, Set<string>> = new Map();
private connectionRateByIP: Map<string, number[]> = new Map();
constructor(private logger: ILogger, private routes: IRouteConfig[] = [], private maxConnectionsPerIP: number = 100, private connectionRateLimitPerMinute: number = 300) {
// Start periodic cleanup for connection tracking
this.startPeriodicIpCleanup();
}
/**
* Update the routes configuration
@ -295,4 +302,132 @@ export class SecurityManager {
return false;
}
}
/**
* Get connections count by IP
*/
public getConnectionCountByIP(ip: string): number {
return this.connectionsByIP.get(ip)?.size || 0;
}
/**
* Check and update connection rate for an IP
* @returns true if within rate limit, false if exceeding limit
*/
public checkConnectionRate(ip: string): boolean {
const now = Date.now();
const minute = 60 * 1000;
if (!this.connectionRateByIP.has(ip)) {
this.connectionRateByIP.set(ip, [now]);
return true;
}
// Get timestamps and filter out entries older than 1 minute
const timestamps = this.connectionRateByIP.get(ip)!.filter((time) => now - time < minute);
timestamps.push(now);
this.connectionRateByIP.set(ip, timestamps);
// Check if rate exceeds limit
return timestamps.length <= this.connectionRateLimitPerMinute;
}
/**
* Track connection by IP
*/
public trackConnectionByIP(ip: string, connectionId: string): void {
if (!this.connectionsByIP.has(ip)) {
this.connectionsByIP.set(ip, new Set());
}
this.connectionsByIP.get(ip)!.add(connectionId);
}
/**
* Remove connection tracking for an IP
*/
public removeConnectionByIP(ip: string, connectionId: string): void {
if (this.connectionsByIP.has(ip)) {
const connections = this.connectionsByIP.get(ip)!;
connections.delete(connectionId);
if (connections.size === 0) {
this.connectionsByIP.delete(ip);
}
}
}
/**
* Check if IP should be allowed considering connection rate and max connections
* @returns Object with result and reason
*/
public validateIP(ip: string): { allowed: boolean; reason?: string } {
// Check connection count limit
if (this.getConnectionCountByIP(ip) >= this.maxConnectionsPerIP) {
return {
allowed: false,
reason: `Maximum connections per IP (${this.maxConnectionsPerIP}) exceeded`
};
}
// Check connection rate limit
if (!this.checkConnectionRate(ip)) {
return {
allowed: false,
reason: `Connection rate limit (${this.connectionRateLimitPerMinute}/min) exceeded`
};
}
return { allowed: true };
}
/**
* Clears all IP tracking data (for shutdown)
*/
public clearIPTracking(): void {
this.connectionsByIP.clear();
this.connectionRateByIP.clear();
}
/**
* Start periodic cleanup of IP tracking data
*/
private startPeriodicIpCleanup(): void {
// Clean up IP tracking data every minute
setInterval(() => {
this.performIpCleanup();
}, 60000).unref();
}
/**
* Perform cleanup of expired IP data
*/
private performIpCleanup(): void {
const now = Date.now();
const minute = 60 * 1000;
let cleanedRateLimits = 0;
let cleanedIPs = 0;
// Clean up expired rate limit timestamps
for (const [ip, timestamps] of this.connectionRateByIP.entries()) {
const validTimestamps = timestamps.filter(time => now - time < minute);
if (validTimestamps.length === 0) {
this.connectionRateByIP.delete(ip);
cleanedRateLimits++;
} else if (validTimestamps.length < timestamps.length) {
this.connectionRateByIP.set(ip, validTimestamps);
}
}
// Clean up IPs with no active connections
for (const [ip, connections] of this.connectionsByIP.entries()) {
if (connections.size === 0) {
this.connectionsByIP.delete(ip);
cleanedIPs++;
}
}
if (cleanedRateLimits > 0 || cleanedIPs > 0) {
this.logger.debug(`IP cleanup: removed ${cleanedIPs} IPs and ${cleanedRateLimits} rate limits`);
}
}
}

View File

@ -1,6 +1,7 @@
import * as plugins from '../../plugins.js';
import type { IConnectionRecord } from './models/interfaces.js';
import { logger } from '../../core/utils/logger.js';
import { connectionLogDeduplicator } from '../../core/utils/log-deduplicator.js';
import { LifecycleComponent } from '../../core/utils/lifecycle-component.js';
import { cleanupSocket } from '../../core/utils/socket-utils.js';
import { WrappedSocket } from '../../core/models/wrapped-socket.js';
@ -26,6 +27,10 @@ export class ConnectionManager extends LifecycleComponent {
// Cleanup queue for batched processing
private cleanupQueue: Set<string> = new Set();
private cleanupTimer: NodeJS.Timeout | null = null;
private isProcessingCleanup: boolean = false;
// Route-level connection tracking
private connectionsByRoute: Map<string, Set<string>> = new Map();
constructor(
private smartProxy: SmartProxy
@ -56,11 +61,19 @@ export class ConnectionManager extends LifecycleComponent {
public createConnection(socket: plugins.net.Socket | WrappedSocket): IConnectionRecord | null {
// Enforce connection limit
if (this.connectionRecords.size >= this.maxConnections) {
logger.log('warn', `Connection limit reached (${this.maxConnections}). Rejecting new connection.`, {
currentConnections: this.connectionRecords.size,
maxConnections: this.maxConnections,
component: 'connection-manager'
});
// Use deduplicated logging for connection limit
connectionLogDeduplicator.log(
'connection-rejected',
'warn',
'Global connection limit reached',
{
reason: 'global-limit',
currentConnections: this.connectionRecords.size,
maxConnections: this.maxConnections,
component: 'connection-manager'
},
'global-limit'
);
socket.destroy();
return null;
}
@ -165,18 +178,53 @@ export class ConnectionManager extends LifecycleComponent {
return this.connectionRecords.size;
}
/**
* Track connection by route
*/
public trackConnectionByRoute(routeId: string, connectionId: string): void {
if (!this.connectionsByRoute.has(routeId)) {
this.connectionsByRoute.set(routeId, new Set());
}
this.connectionsByRoute.get(routeId)!.add(connectionId);
}
/**
* Remove connection tracking for a route
*/
public removeConnectionByRoute(routeId: string, connectionId: string): void {
if (this.connectionsByRoute.has(routeId)) {
const connections = this.connectionsByRoute.get(routeId)!;
connections.delete(connectionId);
if (connections.size === 0) {
this.connectionsByRoute.delete(routeId);
}
}
}
/**
* Get connection count by route
*/
public getConnectionCountByRoute(routeId: string): number {
return this.connectionsByRoute.get(routeId)?.size || 0;
}
/**
* Initiates cleanup once for a connection
*/
public initiateCleanupOnce(record: IConnectionRecord, reason: string = 'normal'): void {
if (this.smartProxy.settings.enableDetailedLogging) {
logger.log('info', `Connection cleanup initiated`, {
// Use deduplicated logging for cleanup events
connectionLogDeduplicator.log(
'connection-cleanup',
'info',
`Connection cleanup: ${reason}`,
{
connectionId: record.id,
remoteIP: record.remoteIP,
reason,
component: 'connection-manager'
});
}
},
reason
);
if (record.incomingTerminationReason == null) {
record.incomingTerminationReason = reason;
@ -200,10 +248,10 @@ export class ConnectionManager extends LifecycleComponent {
this.cleanupQueue.add(connectionId);
// Process immediately if queue is getting large
if (this.cleanupQueue.size >= this.cleanupBatchSize) {
// Process immediately if queue is getting large and not already processing
if (this.cleanupQueue.size >= this.cleanupBatchSize && !this.isProcessingCleanup) {
this.processCleanupQueue();
} else if (!this.cleanupTimer) {
} else if (!this.cleanupTimer && !this.isProcessingCleanup) {
// Otherwise, schedule batch processing
this.cleanupTimer = this.setTimeout(() => {
this.processCleanupQueue();
@ -215,27 +263,40 @@ export class ConnectionManager extends LifecycleComponent {
* Process the cleanup queue in batches
*/
private processCleanupQueue(): void {
// Prevent concurrent processing
if (this.isProcessingCleanup) {
return;
}
this.isProcessingCleanup = true;
if (this.cleanupTimer) {
this.clearTimeout(this.cleanupTimer);
this.cleanupTimer = null;
}
const toCleanup = Array.from(this.cleanupQueue).slice(0, this.cleanupBatchSize);
// Remove only the items we're processing, not the entire queue!
for (const connectionId of toCleanup) {
this.cleanupQueue.delete(connectionId);
const record = this.connectionRecords.get(connectionId);
if (record) {
this.cleanupConnection(record, record.incomingTerminationReason || 'normal');
try {
// Take a snapshot of items to process
const toCleanup = Array.from(this.cleanupQueue).slice(0, this.cleanupBatchSize);
// Remove only the items we're processing from the queue
for (const connectionId of toCleanup) {
this.cleanupQueue.delete(connectionId);
const record = this.connectionRecords.get(connectionId);
if (record) {
this.cleanupConnection(record, record.incomingTerminationReason || 'normal');
}
}
} finally {
// Always reset the processing flag
this.isProcessingCleanup = false;
// Check if more items were added while we were processing
if (this.cleanupQueue.size > 0) {
this.cleanupTimer = this.setTimeout(() => {
this.processCleanupQueue();
}, 10);
}
}
// If there are more in queue, schedule next batch
if (this.cleanupQueue.size > 0) {
this.cleanupTimer = this.setTimeout(() => {
this.processCleanupQueue();
}, 10);
}
}
@ -252,6 +313,11 @@ export class ConnectionManager extends LifecycleComponent {
// Track connection termination
this.smartProxy.securityManager.removeConnectionByIP(record.remoteIP, record.id);
// Remove from route tracking
if (record.routeId) {
this.removeConnectionByRoute(record.routeId, record.id);
}
// Remove from metrics tracking
if (this.smartProxy.metricsCollector) {
this.smartProxy.metricsCollector.removeConnection(record.id);
@ -335,23 +401,34 @@ export class ConnectionManager extends LifecycleComponent {
// Remove the record from the tracking map
this.connectionRecords.delete(record.id);
// Log connection details
// Use deduplicated logging for connection termination
if (this.smartProxy.settings.enableDetailedLogging) {
logger.log('info',
`Connection terminated: ${record.remoteIP}:${record.localPort} (${reason}) - ` +
`${plugins.prettyMs(duration)}, IN: ${record.bytesReceived}B, OUT: ${record.bytesSent}B`,
logData
// For detailed logging, include more info but still deduplicate by IP+reason
connectionLogDeduplicator.log(
'connection-terminated',
'info',
`Connection terminated: ${record.remoteIP}:${record.localPort}`,
{
...logData,
duration_ms: duration,
bytesIn: record.bytesReceived,
bytesOut: record.bytesSent
},
`${record.remoteIP}-${reason}`
);
} else {
logger.log('info',
`Connection terminated: ${record.remoteIP} (${reason}). Active: ${this.connectionRecords.size}`,
// For normal logging, deduplicate by termination reason
connectionLogDeduplicator.log(
'connection-terminated',
'info',
`Connection terminated`,
{
connectionId: record.id,
remoteIP: record.remoteIP,
reason,
activeConnections: this.connectionRecords.size,
component: 'connection-manager'
}
},
reason // Group by termination reason
);
}
}

View File

@ -121,6 +121,11 @@ export class HttpProxyBridge {
proxySocket.on('error', reject);
});
// Send client IP information header first (custom protocol)
// Format: "CLIENT_IP:<ip>\r\n"
const clientIPHeader = Buffer.from(`CLIENT_IP:${record.remoteIP}\r\n`);
proxySocket.write(clientIPHeader);
// Send initial chunk if present
if (initialChunk) {
// Count the initial chunk bytes

View File

@ -15,6 +15,8 @@ import { logger } from '../../core/utils/logger.js';
export class MetricsCollector implements IMetrics {
// Throughput tracking
private throughputTracker: ThroughputTracker;
private routeThroughputTrackers = new Map<string, ThroughputTracker>();
private ipThroughputTrackers = new Map<string, ThroughputTracker>();
// Request tracking
private requestTimestamps: number[] = [];
@ -119,78 +121,28 @@ export class MetricsCollector implements IMetrics {
return this.throughputTracker.getHistory(seconds);
},
byRoute: (windowSeconds: number = 60): Map<string, IThroughputData> => {
byRoute: (windowSeconds: number = 1): Map<string, IThroughputData> => {
const routeThroughput = new Map<string, IThroughputData>();
const now = Date.now();
const windowStart = now - (windowSeconds * 1000);
// Aggregate bytes by route with proper time calculation
const routeData = new Map<string, { bytesIn: number; bytesOut: number; totalDuration: number }>();
for (const [_, tracker] of this.connectionByteTrackers) {
// Only include connections that were active within the window
if (tracker.lastUpdate > windowStart || tracker.startTime > windowStart) {
// Calculate the actual duration this connection was active within the window
const connectionStart = Math.max(tracker.startTime, windowStart);
const connectionEnd = tracker.lastUpdate;
const durationInWindow = (connectionEnd - connectionStart) / 1000; // Convert to seconds
if (durationInWindow > 0) {
const current = routeData.get(tracker.routeName) || { bytesIn: 0, bytesOut: 0, totalDuration: 0 };
current.bytesIn += tracker.bytesIn;
current.bytesOut += tracker.bytesOut;
current.totalDuration += durationInWindow;
routeData.set(tracker.routeName, current);
}
}
}
// Convert to rates (bytes per second)
for (const [route, data] of routeData) {
if (data.totalDuration > 0) {
routeThroughput.set(route, {
in: Math.round(data.bytesIn / data.totalDuration),
out: Math.round(data.bytesOut / data.totalDuration)
});
// Get throughput from each route's dedicated tracker
for (const [route, tracker] of this.routeThroughputTrackers) {
const rate = tracker.getRate(windowSeconds);
if (rate.in > 0 || rate.out > 0) {
routeThroughput.set(route, rate);
}
}
return routeThroughput;
},
byIP: (windowSeconds: number = 60): Map<string, IThroughputData> => {
byIP: (windowSeconds: number = 1): Map<string, IThroughputData> => {
const ipThroughput = new Map<string, IThroughputData>();
const now = Date.now();
const windowStart = now - (windowSeconds * 1000);
// Aggregate bytes by IP with proper time calculation
const ipData = new Map<string, { bytesIn: number; bytesOut: number; totalDuration: number }>();
for (const [_, tracker] of this.connectionByteTrackers) {
// Only include connections that were active within the window
if (tracker.lastUpdate > windowStart || tracker.startTime > windowStart) {
// Calculate the actual duration this connection was active within the window
const connectionStart = Math.max(tracker.startTime, windowStart);
const connectionEnd = tracker.lastUpdate;
const durationInWindow = (connectionEnd - connectionStart) / 1000; // Convert to seconds
if (durationInWindow > 0) {
const current = ipData.get(tracker.remoteIP) || { bytesIn: 0, bytesOut: 0, totalDuration: 0 };
current.bytesIn += tracker.bytesIn;
current.bytesOut += tracker.bytesOut;
current.totalDuration += durationInWindow;
ipData.set(tracker.remoteIP, current);
}
}
}
// Convert to rates (bytes per second)
for (const [ip, data] of ipData) {
if (data.totalDuration > 0) {
ipThroughput.set(ip, {
in: Math.round(data.bytesIn / data.totalDuration),
out: Math.round(data.bytesOut / data.totalDuration)
});
// Get throughput from each IP's dedicated tracker
for (const [ip, tracker] of this.ipThroughputTrackers) {
const rate = tracker.getRate(windowSeconds);
if (rate.in > 0 || rate.out > 0) {
ipThroughput.set(ip, rate);
}
}
@ -323,6 +275,22 @@ export class MetricsCollector implements IMetrics {
tracker.bytesIn += bytesIn;
tracker.bytesOut += bytesOut;
tracker.lastUpdate = Date.now();
// Update per-route throughput tracker
let routeTracker = this.routeThroughputTrackers.get(tracker.routeName);
if (!routeTracker) {
routeTracker = new ThroughputTracker(this.retentionSeconds);
this.routeThroughputTrackers.set(tracker.routeName, routeTracker);
}
routeTracker.recordBytes(bytesIn, bytesOut);
// Update per-IP throughput tracker
let ipTracker = this.ipThroughputTrackers.get(tracker.remoteIP);
if (!ipTracker) {
ipTracker = new ThroughputTracker(this.retentionSeconds);
this.ipThroughputTrackers.set(tracker.remoteIP, ipTracker);
}
ipTracker.recordBytes(bytesIn, bytesOut);
}
}
@ -343,8 +311,19 @@ export class MetricsCollector implements IMetrics {
// Start periodic sampling
this.samplingInterval = setInterval(() => {
// Sample global throughput
this.throughputTracker.takeSample();
// Sample per-route throughput
for (const [_, tracker] of this.routeThroughputTrackers) {
tracker.takeSample();
}
// Sample per-IP throughput
for (const [_, tracker] of this.ipThroughputTrackers) {
tracker.takeSample();
}
// Clean up old connection trackers (connections closed more than 5 minutes ago)
const cutoff = Date.now() - 300000;
for (const [id, tracker] of this.connectionByteTrackers) {
@ -352,6 +331,22 @@ export class MetricsCollector implements IMetrics {
this.connectionByteTrackers.delete(id);
}
}
// Clean up unused route trackers
const activeRoutes = new Set(Array.from(this.connectionByteTrackers.values()).map(t => t.routeName));
for (const [route, _] of this.routeThroughputTrackers) {
if (!activeRoutes.has(route)) {
this.routeThroughputTrackers.delete(route);
}
}
// Clean up unused IP trackers
const activeIPs = new Set(Array.from(this.connectionByteTrackers.values()).map(t => t.remoteIP));
for (const [ip, _] of this.ipThroughputTrackers) {
if (!activeIPs.has(ip)) {
this.ipThroughputTrackers.delete(ip);
}
}
}, this.sampleIntervalMs);
// Subscribe to new connections

View File

@ -165,6 +165,7 @@ export interface IConnectionRecord {
tlsHandshakeComplete: boolean; // Whether the TLS handshake is complete
hasReceivedInitialData: boolean; // Whether initial data has been received
routeConfig?: IRouteConfig; // Associated route config for this connection
routeId?: string; // ID of the route this connection is associated with
// Target information (for dynamic port/host mapping)
targetHost?: string; // Resolved target host

View File

@ -49,8 +49,8 @@ export interface IMetrics {
average(): IThroughputData; // Last 60 seconds
custom(seconds: number): IThroughputData;
history(seconds: number): Array<IThroughputHistoryPoint>;
byRoute(windowSeconds?: number): Map<string, IThroughputData>;
byIP(windowSeconds?: number): Map<string, IThroughputData>;
byRoute(windowSeconds?: number): Map<string, IThroughputData>; // Default: 1 second
byIP(windowSeconds?: number): Map<string, IThroughputData>; // Default: 1 second
};
// Request metrics

View File

@ -1,6 +1,7 @@
import * as plugins from '../../plugins.js';
import type { IConnectionRecord, ISmartProxyOptions } from './models/interfaces.js';
import { logger } from '../../core/utils/logger.js';
import { connectionLogDeduplicator } from '../../core/utils/log-deduplicator.js';
// Route checking functions have been removed
import type { IRouteConfig, IRouteAction } from './models/route-types.js';
import type { IRouteContext } from '../../core/models/route-context.js';
@ -89,7 +90,13 @@ export class RouteConnectionHandler {
// Note: For wrapped sockets, this will use the underlying socket IP until PROXY protocol is parsed
const ipValidation = this.smartProxy.securityManager.validateIP(wrappedSocket.remoteAddress || '');
if (!ipValidation.allowed) {
logger.log('warn', `Connection rejected`, { remoteIP: wrappedSocket.remoteAddress, reason: ipValidation.reason, component: 'route-handler' });
connectionLogDeduplicator.log(
'ip-rejected',
'warn',
`Connection rejected from ${wrappedSocket.remoteAddress}`,
{ remoteIP: wrappedSocket.remoteAddress, reason: ipValidation.reason, component: 'route-handler' },
wrappedSocket.remoteAddress
);
cleanupSocket(wrappedSocket.socket, `rejected-${ipValidation.reason}`, { immediate: true });
return;
}
@ -563,12 +570,20 @@ export class RouteConnectionHandler {
);
if (!isIPAllowed) {
logger.log('warn', `IP ${remoteIP} blocked by route security for route ${route.name || 'unnamed'} (connection: ${connectionId})`, {
connectionId,
remoteIP,
routeName: route.name || 'unnamed',
component: 'route-handler'
});
// Deduplicated logging for route IP blocks
connectionLogDeduplicator.log(
'ip-rejected',
'warn',
`IP blocked by route security`,
{
connectionId,
remoteIP,
routeName: route.name || 'unnamed',
reason: 'route-ip-blocked',
component: 'route-handler'
},
remoteIP
);
socket.end();
this.smartProxy.connectionManager.cleanupConnection(record, 'route_ip_blocked');
return;
@ -577,14 +592,28 @@ export class RouteConnectionHandler {
// Check max connections per route
if (route.security.maxConnections !== undefined) {
// TODO: Implement per-route connection tracking
// For now, log that this feature is not yet implemented
if (this.smartProxy.settings.enableDetailedLogging) {
logger.log('warn', `Route ${route.name} has maxConnections=${route.security.maxConnections} configured but per-route connection limits are not yet implemented`, {
connectionId,
routeName: route.name,
component: 'route-handler'
});
const routeId = route.id || route.name || 'unnamed';
const currentConnections = this.smartProxy.connectionManager.getConnectionCountByRoute(routeId);
if (currentConnections >= route.security.maxConnections) {
// Deduplicated logging for route connection limits
connectionLogDeduplicator.log(
'connection-rejected',
'warn',
`Route connection limit reached`,
{
connectionId,
routeName: route.name,
currentConnections,
maxConnections: route.security.maxConnections,
reason: 'route-limit',
component: 'route-handler'
},
`route-limit-${route.name}`
);
socket.end();
this.smartProxy.connectionManager.cleanupConnection(record, 'route_connection_limit');
return;
}
}
@ -642,6 +671,10 @@ export class RouteConnectionHandler {
// Store the route config in the connection record for metrics and other uses
record.routeConfig = route;
record.routeId = route.id || route.name || 'unnamed';
// Track connection by route
this.smartProxy.connectionManager.trackConnectionByRoute(record.routeId, record.id);
// Check if this route uses NFTables for forwarding
if (action.forwardingEngine === 'nftables') {
@ -960,6 +993,10 @@ export class RouteConnectionHandler {
// Store the route config in the connection record for metrics and other uses
record.routeConfig = route;
record.routeId = route.id || route.name || 'unnamed';
// Track connection by route
this.smartProxy.connectionManager.trackConnectionByRoute(record.routeId, record.id);
if (!route.action.socketHandler) {
logger.log('error', 'socket-handler action missing socketHandler function', {

View File

@ -1,5 +1,7 @@
import * as plugins from '../../plugins.js';
import type { SmartProxy } from './smart-proxy.js';
import { logger } from '../../core/utils/logger.js';
import { connectionLogDeduplicator } from '../../core/utils/log-deduplicator.js';
/**
* Handles security aspects like IP tracking, rate limiting, and authorization
@ -7,8 +9,12 @@ import type { SmartProxy } from './smart-proxy.js';
export class SecurityManager {
private connectionsByIP: Map<string, Set<string>> = new Map();
private connectionRateByIP: Map<string, number[]> = new Map();
private cleanupInterval: NodeJS.Timeout | null = null;
constructor(private smartProxy: SmartProxy) {}
constructor(private smartProxy: SmartProxy) {
// Start periodic cleanup every 60 seconds
this.startPeriodicCleanup();
}
/**
* Get connections count by IP
@ -164,7 +170,76 @@ export class SecurityManager {
* Clears all IP tracking data (for shutdown)
*/
public clearIPTracking(): void {
if (this.cleanupInterval) {
clearInterval(this.cleanupInterval);
this.cleanupInterval = null;
}
this.connectionsByIP.clear();
this.connectionRateByIP.clear();
}
/**
* Start periodic cleanup of expired data
*/
private startPeriodicCleanup(): void {
this.cleanupInterval = setInterval(() => {
this.performCleanup();
}, 60000); // Run every minute
// Unref the timer so it doesn't keep the process alive
if (this.cleanupInterval.unref) {
this.cleanupInterval.unref();
}
}
/**
* Perform cleanup of expired rate limits and empty IP entries
*/
private performCleanup(): void {
const now = Date.now();
const minute = 60 * 1000;
let cleanedRateLimits = 0;
let cleanedIPs = 0;
// Clean up expired rate limit timestamps
for (const [ip, timestamps] of this.connectionRateByIP.entries()) {
const validTimestamps = timestamps.filter(time => now - time < minute);
if (validTimestamps.length === 0) {
// No valid timestamps, remove the IP entry
this.connectionRateByIP.delete(ip);
cleanedRateLimits++;
} else if (validTimestamps.length < timestamps.length) {
// Some timestamps expired, update with valid ones
this.connectionRateByIP.set(ip, validTimestamps);
}
}
// Clean up IPs with no active connections
for (const [ip, connections] of this.connectionsByIP.entries()) {
if (connections.size === 0) {
this.connectionsByIP.delete(ip);
cleanedIPs++;
}
}
// Log cleanup stats if anything was cleaned
if (cleanedRateLimits > 0 || cleanedIPs > 0) {
if (this.smartProxy.settings.enableDetailedLogging) {
connectionLogDeduplicator.log(
'ip-cleanup',
'debug',
'IP tracking cleanup completed',
{
cleanedRateLimits,
cleanedIPs,
remainingIPs: this.connectionsByIP.size,
remainingRateLimits: this.connectionRateByIP.size,
component: 'security-manager'
},
'periodic-cleanup'
);
}
}
}
}

View File

@ -1,5 +1,6 @@
import * as plugins from '../../plugins.js';
import { logger } from '../../core/utils/logger.js';
import { connectionLogDeduplicator } from '../../core/utils/log-deduplicator.js';
// Importing required components
import { ConnectionManager } from './connection-manager.js';
@ -515,6 +516,9 @@ export class SmartProxy extends plugins.EventEmitter {
// Stop metrics collector
this.metricsCollector.stop();
// Flush any pending deduplicated logs
connectionLogDeduplicator.flushAll();
logger.log('info', 'SmartProxy shutdown complete.');
}

View File

@ -65,24 +65,18 @@ export class ThroughputTracker {
return { in: 0, out: 0 };
}
// Sum bytes in the window
// Calculate total bytes in window
const totalBytesIn = relevantSamples.reduce((sum, s) => sum + s.bytesIn, 0);
const totalBytesOut = relevantSamples.reduce((sum, s) => sum + s.bytesOut, 0);
// Calculate actual window duration (might be less than requested if not enough data)
const actualWindowSeconds = Math.min(
windowSeconds,
(now - relevantSamples[0].timestamp) / 1000
);
// Avoid division by zero
if (actualWindowSeconds === 0) {
return { in: 0, out: 0 };
}
// Use actual number of seconds covered by samples for accurate rate
const oldestSampleTime = relevantSamples[0].timestamp;
const newestSampleTime = relevantSamples[relevantSamples.length - 1].timestamp;
const actualSeconds = Math.max(1, (newestSampleTime - oldestSampleTime) / 1000 + 1);
return {
in: Math.round(totalBytesIn / actualWindowSeconds),
out: Math.round(totalBytesOut / actualWindowSeconds)
in: Math.round(totalBytesIn / actualSeconds),
out: Math.round(totalBytesOut / actualSeconds)
};
}