Compare commits

...

4 Commits

Author SHA1 Message Date
755c81c042 v25.7.9
Some checks failed
Default (tags) / security (push) Successful in 44s
Default (tags) / test (push) Failing after 4m0s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-02-21 13:27:55 +00:00
9368226ce0 fix(tests): use high non-privileged ports in tests to avoid conflicts and CI failures 2026-02-21 13:27:55 +00:00
d4739045cd feat: enhance HTTP/2 support by ensuring Host header is set and adding multiplexed request tests 2026-02-20 18:30:57 +00:00
9521f2e044 feat: add TCP keepalive options and connection pooling for improved performance
- Added `socket2` dependency for socket options.
- Introduced `keep_alive`, `keep_alive_initial_delay_ms`, and `max_connections` fields in `ConnectionConfig`.
- Implemented TCP keepalive settings in `TcpListenerManager` for both client and backend connections.
- Created a new `ConnectionPool` for managing idle HTTP/1.1 and HTTP/2 connections to reduce overhead.
- Enhanced TLS configuration to support ALPN for HTTP/2.
- Added performance tests for connection pooling, stability, and concurrent connections.
2026-02-20 18:16:09 +00:00
25 changed files with 1125 additions and 155 deletions

View File

@@ -1,5 +1,12 @@
# Changelog # Changelog
## 2026-02-21 - 25.7.9 - fix(tests)
use high non-privileged ports in tests to avoid conflicts and CI failures
- Updated multiple test files to use high-range, non-privileged ports instead of well-known or conflicting ports.
- Files changed: test/test.acme-http01-challenge.ts, test/test.connection-forwarding.ts, test/test.forwarding-regression.ts, test/test.http-port8080-forwarding.ts, test/test.port-mapping.ts, test/test.smartproxy.ts, test/test.socket-handler.ts.
- Notable port remappings: 8080/8081 -> 47730/47731 (and other proxy ports like 47710), 8443 -> 47711, 7001/7002 -> 47712/47713, 9090 -> 47721, 8181/8182 -> 47732/47733, 9999 -> 47780, TEST_PORT_START/PROXY_PORT_START -> 47750/48750, and TEST_SERVER_PORT/PROXY_PORT -> 47770/47771.
## 2026-02-19 - 25.7.8 - fix(no-changes) ## 2026-02-19 - 25.7.8 - fix(no-changes)
no changes detected; nothing to release no changes detected; nothing to release

View File

@@ -1,6 +1,6 @@
{ {
"name": "@push.rocks/smartproxy", "name": "@push.rocks/smartproxy",
"version": "25.7.8", "version": "25.7.9",
"private": false, "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.", "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", "main": "dist_ts/index.js",

16
rust/Cargo.lock generated
View File

@@ -509,7 +509,7 @@ dependencies = [
"hyper", "hyper",
"libc", "libc",
"pin-project-lite", "pin-project-lite",
"socket2", "socket2 0.6.2",
"tokio", "tokio",
"tower-service", "tower-service",
"tracing", "tracing",
@@ -971,6 +971,7 @@ dependencies = [
"rustproxy-metrics", "rustproxy-metrics",
"rustproxy-routing", "rustproxy-routing",
"rustproxy-security", "rustproxy-security",
"socket2 0.5.10",
"thiserror 2.0.18", "thiserror 2.0.18",
"tokio", "tokio",
"tokio-rustls", "tokio-rustls",
@@ -1019,6 +1020,7 @@ dependencies = [
"rustproxy-routing", "rustproxy-routing",
"serde", "serde",
"serde_json", "serde_json",
"socket2 0.5.10",
"thiserror 2.0.18", "thiserror 2.0.18",
"tokio", "tokio",
"tokio-rustls", "tokio-rustls",
@@ -1204,6 +1206,16 @@ version = "1.15.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
[[package]]
name = "socket2"
version = "0.5.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678"
dependencies = [
"libc",
"windows-sys 0.52.0",
]
[[package]] [[package]]
name = "socket2" name = "socket2"
version = "0.6.2" version = "0.6.2"
@@ -1329,7 +1341,7 @@ dependencies = [
"parking_lot", "parking_lot",
"pin-project-lite", "pin-project-lite",
"signal-hook-registry", "signal-hook-registry",
"socket2", "socket2 0.6.2",
"tokio-macros", "tokio-macros",
"windows-sys 0.61.2", "windows-sys 0.61.2",
] ]

View File

@@ -88,6 +88,9 @@ async-trait = "0.1"
# libc for uid checks # libc for uid checks
libc = "0.2" libc = "0.2"
# Socket-level options (keepalive, etc.)
socket2 = { version = "0.5", features = ["all"] }
# Internal crates # Internal crates
rustproxy-config = { path = "crates/rustproxy-config" } rustproxy-config = { path = "crates/rustproxy-config" }
rustproxy-routing = { path = "crates/rustproxy-routing" } rustproxy-routing = { path = "crates/rustproxy-routing" }

View File

@@ -208,6 +208,10 @@ pub struct RustProxyOptions {
#[serde(skip_serializing_if = "Option::is_none")] #[serde(skip_serializing_if = "Option::is_none")]
pub connection_rate_limit_per_minute: Option<u64>, pub connection_rate_limit_per_minute: Option<u64>,
/// Global maximum simultaneous connections (default: 100000)
#[serde(skip_serializing_if = "Option::is_none")]
pub max_connections: Option<u64>,
// ─── Keep-Alive Settings ───────────────────────────────────────── // ─── Keep-Alive Settings ─────────────────────────────────────────
/// How to treat keep-alive connections /// How to treat keep-alive connections
@@ -272,6 +276,7 @@ impl Default for RustProxyOptions {
enable_randomized_timeouts: None, enable_randomized_timeouts: None,
max_connections_per_ip: None, max_connections_per_ip: None,
connection_rate_limit_per_minute: None, connection_rate_limit_per_minute: None,
max_connections: None,
keep_alive_treatment: None, keep_alive_treatment: None,
keep_alive_inactivity_multiplier: None, keep_alive_inactivity_multiplier: None,
extended_keep_alive_lifetime: None, extended_keep_alive_lifetime: None,

View File

@@ -26,3 +26,4 @@ anyhow = { workspace = true }
arc-swap = { workspace = true } arc-swap = { workspace = true }
dashmap = { workspace = true } dashmap = { workspace = true }
tokio-util = { workspace = true } tokio-util = { workspace = true }
socket2 = { workspace = true }

View File

@@ -0,0 +1,188 @@
//! Backend connection pool for HTTP/1.1 and HTTP/2.
//!
//! Reuses idle keep-alive connections to avoid per-request TCP+TLS handshakes.
//! HTTP/2 connections are multiplexed (clone the sender for each request).
use std::sync::Arc;
use std::time::{Duration, Instant};
use bytes::Bytes;
use dashmap::DashMap;
use http_body_util::combinators::BoxBody;
use hyper::client::conn::{http1, http2};
use tracing::debug;
/// Maximum idle connections per backend key.
const MAX_IDLE_PER_KEY: usize = 16;
/// Default idle timeout — connections not used within this window are evicted.
const IDLE_TIMEOUT: Duration = Duration::from_secs(90);
/// Background eviction interval.
const EVICTION_INTERVAL: Duration = Duration::from_secs(30);
/// Identifies a unique backend endpoint.
#[derive(Clone, Debug, Hash, Eq, PartialEq)]
pub struct PoolKey {
pub host: String,
pub port: u16,
pub use_tls: bool,
pub h2: bool,
}
/// An idle HTTP/1.1 sender with a timestamp for eviction.
struct IdleH1 {
sender: http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
idle_since: Instant,
}
/// A pooled HTTP/2 sender (multiplexed, Clone-able).
struct PooledH2 {
sender: http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
#[allow(dead_code)] // Reserved for future age-based eviction
created_at: Instant,
}
/// Backend connection pool.
pub struct ConnectionPool {
/// HTTP/1.1 idle connections indexed by backend key.
h1_pool: Arc<DashMap<PoolKey, Vec<IdleH1>>>,
/// HTTP/2 multiplexed connections indexed by backend key.
h2_pool: Arc<DashMap<PoolKey, PooledH2>>,
/// Handle for the background eviction task.
eviction_handle: Option<tokio::task::JoinHandle<()>>,
}
impl ConnectionPool {
/// Create a new pool and start the background eviction task.
pub fn new() -> Self {
let h1_pool: Arc<DashMap<PoolKey, Vec<IdleH1>>> = Arc::new(DashMap::new());
let h2_pool: Arc<DashMap<PoolKey, PooledH2>> = Arc::new(DashMap::new());
let h1_clone = Arc::clone(&h1_pool);
let h2_clone = Arc::clone(&h2_pool);
let eviction_handle = tokio::spawn(async move {
Self::eviction_loop(h1_clone, h2_clone).await;
});
Self {
h1_pool,
h2_pool,
eviction_handle: Some(eviction_handle),
}
}
/// Try to check out an idle HTTP/1.1 sender for the given key.
/// Returns `None` if no usable idle connection exists.
pub fn checkout_h1(&self, key: &PoolKey) -> Option<http1::SendRequest<BoxBody<Bytes, hyper::Error>>> {
let mut entry = self.h1_pool.get_mut(key)?;
let idles = entry.value_mut();
while let Some(idle) = idles.pop() {
// Check if the connection is still alive and ready
if idle.idle_since.elapsed() < IDLE_TIMEOUT && idle.sender.is_ready() && !idle.sender.is_closed() {
debug!("Pool hit (h1): {}:{}", key.host, key.port);
return Some(idle.sender);
}
// Stale or closed — drop it
}
// Clean up empty entry
if idles.is_empty() {
drop(entry);
self.h1_pool.remove(key);
}
None
}
/// Return an HTTP/1.1 sender to the pool after the response body has been prepared.
/// The caller should NOT call this if the sender is closed or not ready.
pub fn checkin_h1(&self, key: PoolKey, sender: http1::SendRequest<BoxBody<Bytes, hyper::Error>>) {
if sender.is_closed() || !sender.is_ready() {
return; // Don't pool broken connections
}
let mut entry = self.h1_pool.entry(key).or_insert_with(Vec::new);
if entry.value().len() < MAX_IDLE_PER_KEY {
entry.value_mut().push(IdleH1 {
sender,
idle_since: Instant::now(),
});
}
// If at capacity, just drop the sender
}
/// Try to get a cloned HTTP/2 sender for the given key.
/// HTTP/2 senders are Clone-able (multiplexed), so we clone rather than remove.
pub fn checkout_h2(&self, key: &PoolKey) -> Option<http2::SendRequest<BoxBody<Bytes, hyper::Error>>> {
let entry = self.h2_pool.get(key)?;
let pooled = entry.value();
// Check if the h2 connection is still alive
if pooled.sender.is_closed() {
drop(entry);
self.h2_pool.remove(key);
return None;
}
if pooled.sender.is_ready() {
debug!("Pool hit (h2): {}:{}", key.host, key.port);
return Some(pooled.sender.clone());
}
None
}
/// Register an HTTP/2 sender in the pool. Since h2 is multiplexed,
/// only one sender per key is stored (it's Clone-able).
pub fn register_h2(&self, key: PoolKey, sender: http2::SendRequest<BoxBody<Bytes, hyper::Error>>) {
if sender.is_closed() {
return;
}
self.h2_pool.insert(key, PooledH2 {
sender,
created_at: Instant::now(),
});
}
/// Background eviction loop — runs every EVICTION_INTERVAL to remove stale connections.
async fn eviction_loop(
h1_pool: Arc<DashMap<PoolKey, Vec<IdleH1>>>,
h2_pool: Arc<DashMap<PoolKey, PooledH2>>,
) {
let mut interval = tokio::time::interval(EVICTION_INTERVAL);
loop {
interval.tick().await;
// Evict stale H1 connections
let mut empty_keys = Vec::new();
for mut entry in h1_pool.iter_mut() {
entry.value_mut().retain(|idle| {
idle.idle_since.elapsed() < IDLE_TIMEOUT && !idle.sender.is_closed()
});
if entry.value().is_empty() {
empty_keys.push(entry.key().clone());
}
}
for key in empty_keys {
h1_pool.remove(&key);
}
// Evict dead H2 connections
let mut dead_h2 = Vec::new();
for entry in h2_pool.iter() {
if entry.value().sender.is_closed() {
dead_h2.push(entry.key().clone());
}
}
for key in dead_h2 {
h2_pool.remove(&key);
}
}
}
}
impl Drop for ConnectionPool {
fn drop(&mut self) {
if let Some(handle) = self.eviction_handle.take() {
handle.abort();
}
}
}

View File

@@ -3,6 +3,7 @@
//! Hyper-based HTTP proxy service for RustProxy. //! Hyper-based HTTP proxy service for RustProxy.
//! Handles HTTP request parsing, route-based forwarding, and response filtering. //! Handles HTTP request parsing, route-based forwarding, and response filtering.
pub mod connection_pool;
pub mod counting_body; pub mod counting_body;
pub mod proxy_service; pub mod proxy_service;
pub mod request_filter; pub mod request_filter;
@@ -10,6 +11,7 @@ pub mod response_filter;
pub mod template; pub mod template;
pub mod upstream_selector; pub mod upstream_selector;
pub use connection_pool::*;
pub use counting_body::*; pub use counting_body::*;
pub use proxy_service::*; pub use proxy_service::*;
pub use template::*; pub use template::*;

View File

@@ -87,21 +87,20 @@ impl tokio::io::AsyncWrite for BackendStream {
} }
} }
/// Connect to a backend over TLS. Uses InsecureVerifier for internal backends /// Connect to a backend over TLS using the shared backend TLS config
/// with self-signed certs (same pattern as tls_handler::connect_tls). /// (from tls_handler). Session resumption is automatic.
async fn connect_tls_backend( async fn connect_tls_backend(
backend_tls_config: &Arc<rustls::ClientConfig>,
host: &str, host: &str,
port: u16, port: u16,
) -> Result<tokio_rustls::client::TlsStream<TcpStream>, Box<dyn std::error::Error + Send + Sync>> { ) -> Result<tokio_rustls::client::TlsStream<TcpStream>, Box<dyn std::error::Error + Send + Sync>> {
let _ = rustls::crypto::ring::default_provider().install_default(); let connector = tokio_rustls::TlsConnector::from(Arc::clone(backend_tls_config));
let config = rustls::ClientConfig::builder()
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureBackendVerifier))
.with_no_client_auth();
let connector = tokio_rustls::TlsConnector::from(Arc::new(config));
let stream = TcpStream::connect(format!("{}:{}", host, port)).await?; let stream = TcpStream::connect(format!("{}:{}", host, port)).await?;
stream.set_nodelay(true)?; stream.set_nodelay(true)?;
// Apply keepalive with 60s default
let _ = socket2::SockRef::from(&stream).set_tcp_keepalive(
&socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
);
let server_name = rustls::pki_types::ServerName::try_from(host.to_string())?; let server_name = rustls::pki_types::ServerName::try_from(host.to_string())?;
let tls_stream = connector.connect(server_name, stream).await?; let tls_stream = connector.connect(server_name, stream).await?;
@@ -109,56 +108,6 @@ async fn connect_tls_backend(
Ok(tls_stream) Ok(tls_stream)
} }
/// Insecure certificate verifier for backend TLS connections.
/// Internal backends may use self-signed certs.
#[derive(Debug)]
struct InsecureBackendVerifier;
impl rustls::client::danger::ServerCertVerifier for InsecureBackendVerifier {
fn verify_server_cert(
&self,
_end_entity: &rustls::pki_types::CertificateDer<'_>,
_intermediates: &[rustls::pki_types::CertificateDer<'_>],
_server_name: &rustls::pki_types::ServerName<'_>,
_ocsp_response: &[u8],
_now: rustls::pki_types::UnixTime,
) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
vec![
rustls::SignatureScheme::RSA_PKCS1_SHA256,
rustls::SignatureScheme::RSA_PKCS1_SHA384,
rustls::SignatureScheme::RSA_PKCS1_SHA512,
rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
rustls::SignatureScheme::ED25519,
rustls::SignatureScheme::RSA_PSS_SHA256,
rustls::SignatureScheme::RSA_PSS_SHA384,
rustls::SignatureScheme::RSA_PSS_SHA512,
]
}
}
/// HTTP proxy service that processes HTTP traffic. /// HTTP proxy service that processes HTTP traffic.
pub struct HttpProxyService { pub struct HttpProxyService {
route_manager: Arc<RouteManager>, route_manager: Arc<RouteManager>,
@@ -172,6 +121,10 @@ pub struct HttpProxyService {
request_counter: AtomicU64, request_counter: AtomicU64,
/// Cache of compiled URL rewrite regexes (keyed by pattern string). /// Cache of compiled URL rewrite regexes (keyed by pattern string).
regex_cache: DashMap<String, Regex>, regex_cache: DashMap<String, Regex>,
/// Shared backend TLS config for session resumption across connections.
backend_tls_config: Arc<rustls::ClientConfig>,
/// Backend connection pool for reusing keep-alive connections.
connection_pool: Arc<crate::connection_pool::ConnectionPool>,
} }
impl HttpProxyService { impl HttpProxyService {
@@ -184,6 +137,8 @@ impl HttpProxyService {
route_rate_limiters: Arc::new(DashMap::new()), route_rate_limiters: Arc::new(DashMap::new()),
request_counter: AtomicU64::new(0), request_counter: AtomicU64::new(0),
regex_cache: DashMap::new(), regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
} }
} }
@@ -201,9 +156,17 @@ impl HttpProxyService {
route_rate_limiters: Arc::new(DashMap::new()), route_rate_limiters: Arc::new(DashMap::new()),
request_counter: AtomicU64::new(0), request_counter: AtomicU64::new(0),
regex_cache: DashMap::new(), regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
} }
} }
/// Set the shared backend TLS config (enables session resumption).
/// Call this after construction to inject the shared config from tls_handler.
pub fn set_backend_tls_config(&mut self, config: Arc<rustls::ClientConfig>) {
self.backend_tls_config = config;
}
/// Handle an incoming HTTP connection on a plain TCP stream. /// Handle an incoming HTTP connection on a plain TCP stream.
pub async fn handle_connection( pub async fn handle_connection(
self: Arc<Self>, self: Arc<Self>,
@@ -217,8 +180,10 @@ impl HttpProxyService {
/// Handle an incoming HTTP connection on any IO type (plain TCP or TLS-terminated). /// Handle an incoming HTTP connection on any IO type (plain TCP or TLS-terminated).
/// ///
/// Uses HTTP/1.1 with upgrade support. Responds to graceful shutdown via the /// Uses `hyper_util::server::conn::auto::Builder` to auto-detect h1 vs h2
/// cancel token — in-flight requests complete, but no new requests are accepted. /// based on ALPN negotiation (TLS) or connection preface (h2c).
/// Supports HTTP/1.1 upgrades (WebSocket) and HTTP/2 CONNECT.
/// Responds to graceful shutdown via the cancel token.
pub async fn handle_io<I>( pub async fn handle_io<I>(
self: Arc<Self>, self: Arc<Self>,
stream: I, stream: I,
@@ -241,24 +206,23 @@ impl HttpProxyService {
} }
}); });
// Use http1::Builder with upgrades for WebSocket support // Auto-detect h1 vs h2 based on ALPN / connection preface.
let mut conn = hyper::server::conn::http1::Builder::new() // serve_connection_with_upgrades supports h1 Upgrade (WebSocket) and h2 CONNECT.
.keep_alive(true) let builder = hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
.serve_connection(io, service) let conn = builder.serve_connection_with_upgrades(io, service);
.with_upgrades(); // Pin on the heap — auto::UpgradeableConnection is !Unpin
let mut conn = Box::pin(conn);
// Use select to support graceful shutdown via cancellation token // Use select to support graceful shutdown via cancellation token
let conn_pin = std::pin::Pin::new(&mut conn);
tokio::select! { tokio::select! {
result = conn_pin => { result = conn.as_mut() => {
if let Err(e) = result { if let Err(e) = result {
debug!("HTTP connection error from {}: {}", peer_addr, e); debug!("HTTP connection error from {}: {}", peer_addr, e);
} }
} }
_ = cancel.cancelled() => { _ = cancel.cancelled() => {
// Graceful shutdown: let in-flight request finish, stop accepting new ones // Graceful shutdown: let in-flight request finish, stop accepting new ones
let conn_pin = std::pin::Pin::new(&mut conn); conn.as_mut().graceful_shutdown();
conn_pin.graceful_shutdown();
if let Err(e) = conn.await { if let Err(e) = conn.await {
debug!("HTTP connection error during shutdown from {}: {}", peer_addr, e); debug!("HTTP connection error during shutdown from {}: {}", peer_addr, e);
} }
@@ -280,7 +244,10 @@ impl HttpProxyService {
.map(|h| { .map(|h| {
// Strip port from host header // Strip port from host header
h.split(':').next().unwrap_or(h).to_string() h.split(':').next().unwrap_or(h).to_string()
}); })
// HTTP/2 uses :authority pseudo-header instead of Host;
// hyper maps it to the URI authority component
.or_else(|| req.uri().host().map(|h| h.to_string()));
let path = req.uri().path().to_string(); let path = req.uri().path().to_string();
let method = req.method().clone(); let method = req.method().clone();
@@ -433,11 +400,18 @@ impl HttpProxyService {
} }
} }
// Ensure Host header is set (HTTP/2 requests don't have Host; need it for h1 backends)
if !upstream_headers.contains_key("host") {
if let Some(ref h) = host {
if let Ok(val) = hyper::header::HeaderValue::from_str(h) {
upstream_headers.insert(hyper::header::HOST, val);
}
}
}
// Add standard reverse-proxy headers (X-Forwarded-*) // Add standard reverse-proxy headers (X-Forwarded-*)
{ {
let original_host = parts.headers.get("host") let original_host = host.as_deref().unwrap_or("");
.and_then(|h| h.to_str().ok())
.unwrap_or("");
let forwarded_proto = if route_match.route.action.tls.as_ref() let forwarded_proto = if route_match.route.action.tls.as_ref()
.map(|t| matches!(t.mode, .map(|t| matches!(t.mode,
rustproxy_config::TlsMode::Terminate rustproxy_config::TlsMode::Terminate
@@ -478,11 +452,32 @@ impl HttpProxyService {
} }
} }
// Connect to upstream with timeout (TLS if upstream.use_tls is set) // --- Connection pooling: try reusing an existing connection first ---
let pool_key = crate::connection_pool::PoolKey {
host: upstream.host.clone(),
port: upstream.port,
use_tls: upstream.use_tls,
h2: use_h2,
};
// Try pooled connection first (H2 only — H2 senders are Clone and multiplexed,
// so checkout doesn't consume request parts. For H1, we try pool inside forward_h1.)
if use_h2 {
if let Some(sender) = self.connection_pool.checkout_h2(&pool_key) {
let result = self.forward_h2_pooled(
sender, parts, body, upstream_headers, &upstream_path,
route_match.route, route_id, &ip_str, &pool_key,
).await;
self.upstream_selector.connection_ended(&upstream_key);
return result;
}
}
// Fresh connection path
let backend = if upstream.use_tls { let backend = if upstream.use_tls {
match tokio::time::timeout( match tokio::time::timeout(
self.connect_timeout, self.connect_timeout,
connect_tls_backend(&upstream.host, upstream.port), connect_tls_backend(&self.backend_tls_config, &upstream.host, upstream.port),
).await { ).await {
Ok(Ok(tls)) => BackendStream::Tls(tls), Ok(Ok(tls)) => BackendStream::Tls(tls),
Ok(Err(e)) => { Ok(Err(e)) => {
@@ -503,6 +498,9 @@ impl HttpProxyService {
).await { ).await {
Ok(Ok(s)) => { Ok(Ok(s)) => {
s.set_nodelay(true).ok(); s.set_nodelay(true).ok();
let _ = socket2::SockRef::from(&s).set_tcp_keepalive(
&socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
);
BackendStream::Plain(s) BackendStream::Plain(s)
} }
Ok(Err(e)) => { Ok(Err(e)) => {
@@ -521,17 +519,16 @@ impl HttpProxyService {
let io = TokioIo::new(backend); let io = TokioIo::new(backend);
let result = if use_h2 { let result = if use_h2 {
// HTTP/2 backend self.forward_h2(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str, &pool_key).await
self.forward_h2(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str).await
} else { } else {
// HTTP/1.1 backend (default) self.forward_h1(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str, &pool_key).await
self.forward_h1(io, parts, body, upstream_headers, &upstream_path, &upstream, route_match.route, route_id, &ip_str).await
}; };
self.upstream_selector.connection_ended(&upstream_key); self.upstream_selector.connection_ended(&upstream_key);
result result
} }
/// Forward request to backend via HTTP/1.1 with body streaming. /// Forward request to backend via HTTP/1.1 with body streaming.
/// Tries a pooled connection first; if unavailable, uses the fresh IO connection.
async fn forward_h1( async fn forward_h1(
&self, &self,
io: TokioIo<BackendStream>, io: TokioIo<BackendStream>,
@@ -543,8 +540,21 @@ impl HttpProxyService {
route: &rustproxy_config::RouteConfig, route: &rustproxy_config::RouteConfig,
route_id: Option<&str>, route_id: Option<&str>,
source_ip: &str, source_ip: &str,
pool_key: &crate::connection_pool::PoolKey,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> { ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
let (mut sender, conn) = match hyper::client::conn::http1::handshake(io).await { // Try pooled H1 connection first — avoids TCP+TLS handshake
if let Some(pooled_sender) = self.connection_pool.checkout_h1(pool_key) {
return self.forward_h1_with_sender(
pooled_sender, parts, body, upstream_headers, upstream_path,
route, route_id, source_ip, pool_key,
).await;
}
// Fresh connection: explicitly type the handshake with BoxBody for uniform pool type
let (sender, conn): (
hyper::client::conn::http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
hyper::client::conn::http1::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>>,
) = match hyper::client::conn::http1::handshake(io).await {
Ok(h) => h, Ok(h) => h,
Err(e) => { Err(e) => {
error!("Upstream handshake failed: {}", e); error!("Upstream handshake failed: {}", e);
@@ -558,16 +568,33 @@ impl HttpProxyService {
} }
}); });
self.forward_h1_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip, pool_key).await
}
/// Common H1 forwarding logic used by both fresh and pooled paths.
async fn forward_h1_with_sender(
&self,
mut sender: hyper::client::conn::http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
parts: hyper::http::request::Parts,
body: Incoming,
upstream_headers: hyper::HeaderMap,
upstream_path: &str,
route: &rustproxy_config::RouteConfig,
route_id: Option<&str>,
source_ip: &str,
pool_key: &crate::connection_pool::PoolKey,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
// Always use HTTP/1.1 for h1 backend connections (h2 incoming requests have version HTTP/2.0)
let mut upstream_req = Request::builder() let mut upstream_req = Request::builder()
.method(parts.method) .method(parts.method)
.uri(upstream_path) .uri(upstream_path)
.version(parts.version); .version(hyper::Version::HTTP_11);
if let Some(headers) = upstream_req.headers_mut() { if let Some(headers) = upstream_req.headers_mut() {
*headers = upstream_headers; *headers = upstream_headers;
} }
// Wrap the request body in CountingBody to track bytes_in // Wrap the request body in CountingBody then box it for the uniform pool type
let counting_req_body = CountingBody::new( let counting_req_body = CountingBody::new(
body, body,
Arc::clone(&self.metrics), Arc::clone(&self.metrics),
@@ -575,9 +602,9 @@ impl HttpProxyService {
Some(source_ip.to_string()), Some(source_ip.to_string()),
Direction::In, Direction::In,
); );
let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_req_body);
// Stream the request body through to upstream let upstream_req = upstream_req.body(boxed_body).unwrap();
let upstream_req = upstream_req.body(counting_req_body).unwrap();
let upstream_response = match sender.send_request(upstream_req).await { let upstream_response = match sender.send_request(upstream_req).await {
Ok(resp) => resp, Ok(resp) => resp,
@@ -587,10 +614,14 @@ impl HttpProxyService {
} }
}; };
// Return sender to pool (body streams lazily, sender is reusable once response head is received)
self.connection_pool.checkin_h1(pool_key.clone(), sender);
self.build_streaming_response(upstream_response, route, route_id, source_ip).await self.build_streaming_response(upstream_response, route, route_id, source_ip).await
} }
/// Forward request to backend via HTTP/2 with body streaming. /// Forward request to backend via HTTP/2 with body streaming (fresh connection).
/// Registers the h2 sender in the pool for future multiplexed requests.
async fn forward_h2( async fn forward_h2(
&self, &self,
io: TokioIo<BackendStream>, io: TokioIo<BackendStream>,
@@ -602,9 +633,14 @@ impl HttpProxyService {
route: &rustproxy_config::RouteConfig, route: &rustproxy_config::RouteConfig,
route_id: Option<&str>, route_id: Option<&str>,
source_ip: &str, source_ip: &str,
pool_key: &crate::connection_pool::PoolKey,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> { ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
let exec = hyper_util::rt::TokioExecutor::new(); let exec = hyper_util::rt::TokioExecutor::new();
let (mut sender, conn) = match hyper::client::conn::http2::handshake(exec, io).await { // Explicitly type the handshake with BoxBody for uniform pool type
let (sender, conn): (
hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
hyper::client::conn::http2::Connection<TokioIo<BackendStream>, BoxBody<Bytes, hyper::Error>, hyper_util::rt::TokioExecutor>,
) = match hyper::client::conn::http2::handshake(exec, io).await {
Ok(h) => h, Ok(h) => h,
Err(e) => { Err(e) => {
error!("HTTP/2 upstream handshake failed: {}", e); error!("HTTP/2 upstream handshake failed: {}", e);
@@ -618,6 +654,40 @@ impl HttpProxyService {
} }
}); });
// Register for multiplexed reuse
self.connection_pool.register_h2(pool_key.clone(), sender.clone());
self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip).await
}
/// Forward request using an existing (pooled) HTTP/2 sender.
async fn forward_h2_pooled(
&self,
sender: hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
parts: hyper::http::request::Parts,
body: Incoming,
upstream_headers: hyper::HeaderMap,
upstream_path: &str,
route: &rustproxy_config::RouteConfig,
route_id: Option<&str>,
source_ip: &str,
_pool_key: &crate::connection_pool::PoolKey,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
self.forward_h2_with_sender(sender, parts, body, upstream_headers, upstream_path, route, route_id, source_ip).await
}
/// Common H2 forwarding logic used by both fresh and pooled paths.
async fn forward_h2_with_sender(
&self,
mut sender: hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
parts: hyper::http::request::Parts,
body: Incoming,
upstream_headers: hyper::HeaderMap,
upstream_path: &str,
route: &rustproxy_config::RouteConfig,
route_id: Option<&str>,
source_ip: &str,
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
let mut upstream_req = Request::builder() let mut upstream_req = Request::builder()
.method(parts.method) .method(parts.method)
.uri(upstream_path); .uri(upstream_path);
@@ -626,7 +696,7 @@ impl HttpProxyService {
*headers = upstream_headers; *headers = upstream_headers;
} }
// Wrap the request body in CountingBody to track bytes_in // Wrap the request body in CountingBody then box it for the uniform pool type
let counting_req_body = CountingBody::new( let counting_req_body = CountingBody::new(
body, body,
Arc::clone(&self.metrics), Arc::clone(&self.metrics),
@@ -634,9 +704,9 @@ impl HttpProxyService {
Some(source_ip.to_string()), Some(source_ip.to_string()),
Direction::In, Direction::In,
); );
let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_req_body);
// Stream the request body through to upstream let upstream_req = upstream_req.body(boxed_body).unwrap();
let upstream_req = upstream_req.body(counting_req_body).unwrap();
let upstream_response = match sender.send_request(upstream_req).await { let upstream_response = match sender.send_request(upstream_req).await {
Ok(resp) => resp, Ok(resp) => resp,
@@ -723,7 +793,7 @@ impl HttpProxyService {
let mut upstream_stream: BackendStream = if upstream.use_tls { let mut upstream_stream: BackendStream = if upstream.use_tls {
match tokio::time::timeout( match tokio::time::timeout(
self.connect_timeout, self.connect_timeout,
connect_tls_backend(&upstream.host, upstream.port), connect_tls_backend(&self.backend_tls_config, &upstream.host, upstream.port),
).await { ).await {
Ok(Ok(tls)) => BackendStream::Tls(tls), Ok(Ok(tls)) => BackendStream::Tls(tls),
Ok(Err(e)) => { Ok(Err(e)) => {
@@ -744,6 +814,9 @@ impl HttpProxyService {
).await { ).await {
Ok(Ok(s)) => { Ok(Ok(s)) => {
s.set_nodelay(true).ok(); s.set_nodelay(true).ok();
let _ = socket2::SockRef::from(&s).set_tcp_keepalive(
&socket2::TcpKeepalive::new().with_time(std::time::Duration::from_secs(60))
);
BackendStream::Plain(s) BackendStream::Plain(s)
} }
Ok(Err(e)) => { Ok(Err(e)) => {
@@ -786,6 +859,7 @@ impl HttpProxyService {
// Copy all original headers (preserving the client's Host header). // Copy all original headers (preserving the client's Host header).
// Skip X-Forwarded-* since we set them ourselves below. // Skip X-Forwarded-* since we set them ourselves below.
let mut has_host_header = false;
for (name, value) in parts.headers.iter() { for (name, value) in parts.headers.iter() {
let name_str = name.as_str(); let name_str = name.as_str();
if name_str == "x-forwarded-for" if name_str == "x-forwarded-for"
@@ -794,13 +868,25 @@ impl HttpProxyService {
{ {
continue; continue;
} }
if name_str == "host" {
has_host_header = true;
}
raw_request.push_str(&format!("{}: {}\r\n", name, value.to_str().unwrap_or(""))); raw_request.push_str(&format!("{}: {}\r\n", name, value.to_str().unwrap_or("")));
} }
// HTTP/2 requests don't have Host header; add one from URI authority for h1 backends
let ws_host = parts.uri.host().map(|h| h.to_string());
if !has_host_header {
if let Some(ref h) = ws_host {
raw_request.push_str(&format!("host: {}\r\n", h));
}
}
// Add standard reverse-proxy headers (X-Forwarded-*) // Add standard reverse-proxy headers (X-Forwarded-*)
{ {
let original_host = parts.headers.get("host") let original_host = parts.headers.get("host")
.and_then(|h| h.to_str().ok()) .and_then(|h| h.to_str().ok())
.or(ws_host.as_deref())
.unwrap_or(""); .unwrap_or("");
let forwarded_proto = if route.action.tls.as_ref() let forwarded_proto = if route.action.tls.as_ref()
.map(|t| matches!(t.mode, .map(|t| matches!(t.mode,
@@ -1221,6 +1307,70 @@ fn guess_content_type(path: &std::path::Path) -> &'static str {
} }
} }
impl HttpProxyService {
/// Build a default backend TLS config with InsecureVerifier.
/// Used as fallback when no shared config is injected from tls_handler.
fn default_backend_tls_config() -> Arc<rustls::ClientConfig> {
let _ = rustls::crypto::ring::default_provider().install_default();
let config = rustls::ClientConfig::builder()
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureBackendVerifier))
.with_no_client_auth();
Arc::new(config)
}
}
/// Insecure certificate verifier for backend TLS connections (fallback only).
/// The production path uses the shared config from tls_handler which has the same
/// behavior but with session resumption across all outbound connections.
#[derive(Debug)]
struct InsecureBackendVerifier;
impl rustls::client::danger::ServerCertVerifier for InsecureBackendVerifier {
fn verify_server_cert(
&self,
_end_entity: &rustls::pki_types::CertificateDer<'_>,
_intermediates: &[rustls::pki_types::CertificateDer<'_>],
_server_name: &rustls::pki_types::ServerName<'_>,
_ocsp_response: &[u8],
_now: rustls::pki_types::UnixTime,
) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
Ok(rustls::client::danger::ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &rustls::pki_types::CertificateDer<'_>,
_dss: &rustls::DigitallySignedStruct,
) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
vec![
rustls::SignatureScheme::RSA_PKCS1_SHA256,
rustls::SignatureScheme::RSA_PKCS1_SHA384,
rustls::SignatureScheme::RSA_PKCS1_SHA512,
rustls::SignatureScheme::ECDSA_NISTP256_SHA256,
rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
rustls::SignatureScheme::ED25519,
rustls::SignatureScheme::RSA_PSS_SHA256,
rustls::SignatureScheme::RSA_PSS_SHA384,
rustls::SignatureScheme::RSA_PSS_SHA512,
]
}
}
impl Default for HttpProxyService { impl Default for HttpProxyService {
fn default() -> Self { fn default() -> Self {
Self { Self {
@@ -1231,6 +1381,8 @@ impl Default for HttpProxyService {
route_rate_limiters: Arc::new(DashMap::new()), route_rate_limiters: Arc::new(DashMap::new()),
request_counter: AtomicU64::new(0), request_counter: AtomicU64::new(0),
regex_cache: DashMap::new(), regex_cache: DashMap::new(),
backend_tls_config: Self::default_backend_tls_config(),
connection_pool: Arc::new(crate::connection_pool::ConnectionPool::new()),
} }
} }
} }

View File

@@ -23,3 +23,4 @@ rustls-pemfile = { workspace = true }
tokio-util = { workspace = true } tokio-util = { workspace = true }
serde = { workspace = true } serde = { workspace = true }
serde_json = { workspace = true } serde_json = { workspace = true }
socket2 = { workspace = true }

View File

@@ -11,6 +11,7 @@ pub mod tls_handler;
pub mod connection_record; pub mod connection_record;
pub mod connection_tracker; pub mod connection_tracker;
pub mod socket_relay; pub mod socket_relay;
pub mod socket_opts;
pub use tcp_listener::*; pub use tcp_listener::*;
pub use sni_parser::*; pub use sni_parser::*;
@@ -20,3 +21,4 @@ pub use tls_handler::*;
pub use connection_record::*; pub use connection_record::*;
pub use connection_tracker::*; pub use connection_tracker::*;
pub use socket_relay::*; pub use socket_relay::*;
pub use socket_opts::*;

View File

@@ -196,6 +196,7 @@ pub fn is_http(data: &[u8]) -> bool {
b"PATC", b"PATC",
b"OPTI", b"OPTI",
b"CONN", b"CONN",
b"PRI ", // HTTP/2 connection preface
]; ];
starts.iter().any(|s| data.starts_with(s)) starts.iter().any(|s| data.starts_with(s))
} }

View File

@@ -0,0 +1,19 @@
//! Socket-level options for TCP streams (keepalive, etc.).
//!
//! Uses `socket2::SockRef::from()` to borrow the raw fd without ownership transfer.
use std::io;
use std::time::Duration;
use tokio::net::TcpStream;
/// Apply TCP keepalive to a connected socket.
///
/// Enables SO_KEEPALIVE and sets the initial probe delay.
/// On Linux, also sets the interval between probes to the same value.
pub fn apply_keepalive(stream: &TcpStream, delay: Duration) -> io::Result<()> {
let sock_ref = socket2::SockRef::from(stream);
let ka = socket2::TcpKeepalive::new().with_time(delay);
#[cfg(target_os = "linux")]
let ka = ka.with_interval(delay);
sock_ref.set_tcp_keepalive(&ka)
}

View File

@@ -15,6 +15,7 @@ use crate::sni_parser;
use crate::forwarder; use crate::forwarder;
use crate::tls_handler; use crate::tls_handler;
use crate::connection_tracker::ConnectionTracker; use crate::connection_tracker::ConnectionTracker;
use crate::socket_opts;
/// RAII guard that decrements the active connection metric on drop. /// RAII guard that decrements the active connection metric on drop.
/// Ensures connection_closed is called on ALL exit paths — normal, error, or panic. /// Ensures connection_closed is called on ALL exit paths — normal, error, or panic.
@@ -87,6 +88,12 @@ pub struct ConnectionConfig {
/// Trusted IPs that may send PROXY protocol headers. /// Trusted IPs that may send PROXY protocol headers.
/// When non-empty, only connections from these IPs will have PROXY headers parsed. /// When non-empty, only connections from these IPs will have PROXY headers parsed.
pub proxy_ips: Vec<std::net::IpAddr>, pub proxy_ips: Vec<std::net::IpAddr>,
/// Enable TCP keepalive on sockets (default: true)
pub keep_alive: bool,
/// Initial delay before first keepalive probe (ms, default: 60000)
pub keep_alive_initial_delay_ms: u64,
/// Global maximum simultaneous connections (default: 100000)
pub max_connections: u64,
} }
impl Default for ConnectionConfig { impl Default for ConnectionConfig {
@@ -105,6 +112,9 @@ impl Default for ConnectionConfig {
accept_proxy_protocol: false, accept_proxy_protocol: false,
send_proxy_protocol: false, send_proxy_protocol: false,
proxy_ips: Vec::new(), proxy_ips: Vec::new(),
keep_alive: true,
keep_alive_initial_delay_ms: 60_000,
max_connections: 100_000,
} }
} }
} }
@@ -131,21 +141,26 @@ pub struct TcpListenerManager {
cancel_token: CancellationToken, cancel_token: CancellationToken,
/// Path to Unix domain socket for relaying socket-handler connections to TypeScript. /// Path to Unix domain socket for relaying socket-handler connections to TypeScript.
socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>, socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>,
/// Global connection semaphore — limits total simultaneous connections.
conn_semaphore: Arc<tokio::sync::Semaphore>,
} }
impl TcpListenerManager { impl TcpListenerManager {
pub fn new(route_manager: Arc<RouteManager>) -> Self { pub fn new(route_manager: Arc<RouteManager>) -> Self {
let metrics = Arc::new(MetricsCollector::new()); let metrics = Arc::new(MetricsCollector::new());
let conn_config = ConnectionConfig::default(); let conn_config = ConnectionConfig::default();
let http_proxy = Arc::new(HttpProxyService::with_connect_timeout( let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
Arc::clone(&route_manager), Arc::clone(&route_manager),
Arc::clone(&metrics), Arc::clone(&metrics),
std::time::Duration::from_millis(conn_config.connection_timeout_ms), std::time::Duration::from_millis(conn_config.connection_timeout_ms),
)); );
http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config());
let http_proxy = Arc::new(http_proxy_svc);
let conn_tracker = Arc::new(ConnectionTracker::new( let conn_tracker = Arc::new(ConnectionTracker::new(
conn_config.max_connections_per_ip, conn_config.max_connections_per_ip,
conn_config.connection_rate_limit_per_minute, conn_config.connection_rate_limit_per_minute,
)); ));
let max_conns = conn_config.max_connections as usize;
Self { Self {
listeners: HashMap::new(), listeners: HashMap::new(),
route_manager: Arc::new(ArcSwap::from(route_manager)), route_manager: Arc::new(ArcSwap::from(route_manager)),
@@ -157,21 +172,25 @@ impl TcpListenerManager {
conn_tracker, conn_tracker,
cancel_token: CancellationToken::new(), cancel_token: CancellationToken::new(),
socket_handler_relay: Arc::new(std::sync::RwLock::new(None)), socket_handler_relay: Arc::new(std::sync::RwLock::new(None)),
conn_semaphore: Arc::new(tokio::sync::Semaphore::new(max_conns)),
} }
} }
/// Create with a metrics collector. /// Create with a metrics collector.
pub fn with_metrics(route_manager: Arc<RouteManager>, metrics: Arc<MetricsCollector>) -> Self { pub fn with_metrics(route_manager: Arc<RouteManager>, metrics: Arc<MetricsCollector>) -> Self {
let conn_config = ConnectionConfig::default(); let conn_config = ConnectionConfig::default();
let http_proxy = Arc::new(HttpProxyService::with_connect_timeout( let mut http_proxy_svc = HttpProxyService::with_connect_timeout(
Arc::clone(&route_manager), Arc::clone(&route_manager),
Arc::clone(&metrics), Arc::clone(&metrics),
std::time::Duration::from_millis(conn_config.connection_timeout_ms), std::time::Duration::from_millis(conn_config.connection_timeout_ms),
)); );
http_proxy_svc.set_backend_tls_config(tls_handler::shared_backend_tls_config());
let http_proxy = Arc::new(http_proxy_svc);
let conn_tracker = Arc::new(ConnectionTracker::new( let conn_tracker = Arc::new(ConnectionTracker::new(
conn_config.max_connections_per_ip, conn_config.max_connections_per_ip,
conn_config.connection_rate_limit_per_minute, conn_config.connection_rate_limit_per_minute,
)); ));
let max_conns = conn_config.max_connections as usize;
Self { Self {
listeners: HashMap::new(), listeners: HashMap::new(),
route_manager: Arc::new(ArcSwap::from(route_manager)), route_manager: Arc::new(ArcSwap::from(route_manager)),
@@ -183,6 +202,7 @@ impl TcpListenerManager {
conn_tracker, conn_tracker,
cancel_token: CancellationToken::new(), cancel_token: CancellationToken::new(),
socket_handler_relay: Arc::new(std::sync::RwLock::new(None)), socket_handler_relay: Arc::new(std::sync::RwLock::new(None)),
conn_semaphore: Arc::new(tokio::sync::Semaphore::new(max_conns)),
} }
} }
@@ -192,6 +212,7 @@ impl TcpListenerManager {
config.max_connections_per_ip, config.max_connections_per_ip,
config.connection_rate_limit_per_minute, config.connection_rate_limit_per_minute,
)); ));
self.conn_semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_connections as usize));
self.conn_config = Arc::new(config); self.conn_config = Arc::new(config);
} }
@@ -247,11 +268,13 @@ impl TcpListenerManager {
let conn_tracker = Arc::clone(&self.conn_tracker); let conn_tracker = Arc::clone(&self.conn_tracker);
let cancel = self.cancel_token.clone(); let cancel = self.cancel_token.clone();
let relay = Arc::clone(&self.socket_handler_relay); let relay = Arc::clone(&self.socket_handler_relay);
let semaphore = Arc::clone(&self.conn_semaphore);
let handle = tokio::spawn(async move { let handle = tokio::spawn(async move {
Self::accept_loop( Self::accept_loop(
listener, port, route_manager_swap, metrics, tls_configs, listener, port, route_manager_swap, metrics, tls_configs,
shared_tls_acceptor, http_proxy, conn_config, conn_tracker, cancel, relay, shared_tls_acceptor, http_proxy, conn_config, conn_tracker, cancel, relay,
semaphore,
).await; ).await;
}); });
@@ -346,6 +369,7 @@ impl TcpListenerManager {
conn_tracker: Arc<ConnectionTracker>, conn_tracker: Arc<ConnectionTracker>,
cancel: CancellationToken, cancel: CancellationToken,
socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>, socket_handler_relay: Arc<std::sync::RwLock<Option<String>>>,
conn_semaphore: Arc<tokio::sync::Semaphore>,
) { ) {
loop { loop {
tokio::select! { tokio::select! {
@@ -358,10 +382,31 @@ impl TcpListenerManager {
Ok((stream, peer_addr)) => { Ok((stream, peer_addr)) => {
let ip = peer_addr.ip(); let ip = peer_addr.ip();
// Global connection limit — acquire semaphore permit with timeout
let permit = match tokio::time::timeout(
std::time::Duration::from_secs(5),
conn_semaphore.clone().acquire_owned(),
).await {
Ok(Ok(permit)) => permit,
Ok(Err(_)) => {
// Semaphore closed — shouldn't happen, but be safe
debug!("Connection semaphore closed, dropping connection from {}", peer_addr);
drop(stream);
continue;
}
Err(_) => {
// Timeout — global limit reached
debug!("Global connection limit reached, dropping connection from {}", peer_addr);
drop(stream);
continue;
}
};
// Check per-IP limits and rate limiting // Check per-IP limits and rate limiting
if !conn_tracker.try_accept(&ip) { if !conn_tracker.try_accept(&ip) {
debug!("Rejected connection from {} (per-IP limit or rate limit)", peer_addr); debug!("Rejected connection from {} (per-IP limit or rate limit)", peer_addr);
drop(stream); drop(stream);
drop(permit);
continue; continue;
} }
@@ -382,6 +427,8 @@ impl TcpListenerManager {
debug!("Accepted connection from {} on port {}", peer_addr, port); debug!("Accepted connection from {} on port {}", peer_addr, port);
tokio::spawn(async move { tokio::spawn(async move {
// Move permit into the task — auto-releases on drop
let _permit = permit;
let result = Self::handle_connection( let result = Self::handle_connection(
stream, port, peer_addr, rm, m, tc, sa, hp, cc, cn, sr, stream, port, peer_addr, rm, m, tc, sa, hp, cc, cn, sr,
).await; ).await;
@@ -418,6 +465,12 @@ impl TcpListenerManager {
use tokio::io::AsyncReadExt; use tokio::io::AsyncReadExt;
stream.set_nodelay(true)?; stream.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&stream, delay) {
debug!("Failed to set keepalive on client socket: {}", e);
}
}
// --- PROXY protocol: must happen BEFORE ip_str and fast path --- // --- PROXY protocol: must happen BEFORE ip_str and fast path ---
// Only parse PROXY headers from trusted proxy IPs (security). // Only parse PROXY headers from trusted proxy IPs (security).
@@ -543,6 +596,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()), Err(_) => return Err("Backend connection timeout".into()),
}; };
backend.set_nodelay(true)?; backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
// Send PROXY protocol header if configured // Send PROXY protocol header if configured
let should_send_proxy = conn_config.send_proxy_protocol let should_send_proxy = conn_config.send_proxy_protocol
@@ -778,6 +837,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()), Err(_) => return Err("Backend connection timeout".into()),
}; };
backend.set_nodelay(true)?; backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
// Send PROXY protocol header if configured // Send PROXY protocol header if configured
if let Some(ref header) = proxy_header { if let Some(ref header) = proxy_header {
@@ -857,6 +922,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()), Err(_) => return Err("Backend connection timeout".into()),
}; };
backend.set_nodelay(true)?; backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
let (tls_read, tls_write) = tokio::io::split(buf_stream); let (tls_read, tls_write) = tokio::io::split(buf_stream);
let (backend_read, backend_write) = tokio::io::split(backend); let (backend_read, backend_write) = tokio::io::split(backend);
@@ -944,6 +1015,12 @@ impl TcpListenerManager {
Err(_) => return Err("Backend connection timeout".into()), Err(_) => return Err("Backend connection timeout".into()),
}; };
backend.set_nodelay(true)?; backend.set_nodelay(true)?;
if conn_config.keep_alive {
let delay = std::time::Duration::from_millis(conn_config.keep_alive_initial_delay_ms);
if let Err(e) = socket_opts::apply_keepalive(&backend, delay) {
debug!("Failed to set keepalive on backend socket: {}", e);
}
}
// Send PROXY protocol header if configured // Send PROXY protocol header if configured
if let Some(ref header) = proxy_header { if let Some(ref header) = proxy_header {

View File

@@ -1,6 +1,6 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::io::BufReader; use std::io::BufReader;
use std::sync::Arc; use std::sync::{Arc, OnceLock};
use rustls::pki_types::{CertificateDer, PrivateKeyDer}; use rustls::pki_types::{CertificateDer, PrivateKeyDer};
use rustls::server::ResolvesServerCert; use rustls::server::ResolvesServerCert;
@@ -84,13 +84,16 @@ pub fn build_shared_tls_acceptor(resolver: CertResolver) -> Result<TlsAcceptor,
.with_no_client_auth() .with_no_client_auth()
.with_cert_resolver(Arc::new(resolver)); .with_cert_resolver(Arc::new(resolver));
// ALPN: advertise h2 and http/1.1 for client-facing HTTP/2 support
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
// Shared session cache — enables session ID resumption across connections // Shared session cache — enables session ID resumption across connections
config.session_storage = rustls::server::ServerSessionMemoryCache::new(4096); config.session_storage = rustls::server::ServerSessionMemoryCache::new(4096);
// Session ticket resumption (12-hour lifetime, Chacha20Poly1305 encrypted) // Session ticket resumption (12-hour lifetime, Chacha20Poly1305 encrypted)
config.ticketer = rustls::crypto::ring::Ticketer::new() config.ticketer = rustls::crypto::ring::Ticketer::new()
.map_err(|e| format!("Ticketer: {}", e))?; .map_err(|e| format!("Ticketer: {}", e))?;
info!("Built shared TLS config with session cache (4096) and ticket support"); info!("Built shared TLS config with session cache (4096), ticket support, and ALPN h2+http/1.1");
Ok(TlsAcceptor::from(Arc::new(config))) Ok(TlsAcceptor::from(Arc::new(config)))
} }
@@ -122,6 +125,9 @@ pub fn build_tls_acceptor_with_config(
.with_single_cert(certs, key)? .with_single_cert(certs, key)?
}; };
// ALPN: advertise h2 and http/1.1 for client-facing HTTP/2 support
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
// Apply session timeout if configured // Apply session timeout if configured
if let Some(route_tls) = tls_config { if let Some(route_tls) = tls_config {
if let Some(timeout_secs) = route_tls.session_timeout { if let Some(timeout_secs) = route_tls.session_timeout {
@@ -179,21 +185,40 @@ pub async fn accept_tls(
Ok(tls_stream) Ok(tls_stream)
} }
/// Get or create a shared backend TLS `ClientConfig`.
///
/// Uses `OnceLock` to ensure only one config is created across the entire process.
/// The built-in rustls `Resumption` (session tickets + session IDs) is enabled
/// by default, so all outbound backend connections share the same session cache.
static SHARED_CLIENT_CONFIG: OnceLock<Arc<rustls::ClientConfig>> = OnceLock::new();
pub fn shared_backend_tls_config() -> Arc<rustls::ClientConfig> {
SHARED_CLIENT_CONFIG.get_or_init(|| {
ensure_crypto_provider();
let config = rustls::ClientConfig::builder()
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureVerifier))
.with_no_client_auth();
info!("Built shared backend TLS client config with session resumption");
Arc::new(config)
}).clone()
}
/// Connect to a backend with TLS (for terminate-and-reencrypt mode). /// Connect to a backend with TLS (for terminate-and-reencrypt mode).
/// Uses the shared backend TLS config for session resumption.
pub async fn connect_tls( pub async fn connect_tls(
host: &str, host: &str,
port: u16, port: u16,
) -> Result<tokio_rustls::client::TlsStream<TcpStream>, Box<dyn std::error::Error + Send + Sync>> { ) -> Result<tokio_rustls::client::TlsStream<TcpStream>, Box<dyn std::error::Error + Send + Sync>> {
ensure_crypto_provider(); let config = shared_backend_tls_config();
let config = rustls::ClientConfig::builder() let connector = TlsConnector::from(config);
.dangerous()
.with_custom_certificate_verifier(Arc::new(InsecureVerifier))
.with_no_client_auth();
let connector = TlsConnector::from(Arc::new(config));
let stream = TcpStream::connect(format!("{}:{}", host, port)).await?; let stream = TcpStream::connect(format!("{}:{}", host, port)).await?;
stream.set_nodelay(true)?; stream.set_nodelay(true)?;
// Apply keepalive with 60s default (tls_handler doesn't have ConnectionConfig access)
if let Err(e) = crate::socket_opts::apply_keepalive(&stream, std::time::Duration::from_secs(60)) {
debug!("Failed to set keepalive on backend TLS socket: {}", e);
}
let server_name = rustls::pki_types::ServerName::try_from(host.to_string())?; let server_name = rustls::pki_types::ServerName::try_from(host.to_string())?;
let tls_stream = connector.connect(server_name, stream).await?; let tls_stream = connector.connect(server_name, stream).await?;

View File

@@ -221,6 +221,9 @@ impl RustProxy {
.iter() .iter()
.filter_map(|s| s.parse::<std::net::IpAddr>().ok()) .filter_map(|s| s.parse::<std::net::IpAddr>().ok())
.collect(), .collect(),
keep_alive: options.keep_alive.unwrap_or(true),
keep_alive_initial_delay_ms: options.keep_alive_initial_delay.unwrap_or(60_000),
max_connections: options.max_connections.unwrap_or(100_000),
} }
} }

View File

@@ -37,7 +37,7 @@ tap.test('should correctly handle HTTP-01 challenge requests with initial data c
routes: [{ routes: [{
name: 'acme-challenge-route', name: 'acme-challenge-route',
match: { match: {
ports: 8080, ports: 47700,
path: '/.well-known/acme-challenge/*' path: '/.well-known/acme-challenge/*'
}, },
action: { action: {
@@ -60,7 +60,7 @@ tap.test('should correctly handle HTTP-01 challenge requests with initial data c
// Connect to the proxy and send the HTTP-01 challenge request // Connect to the proxy and send the HTTP-01 challenge request
await new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
testClient.connect(8080, 'localhost', () => { testClient.connect(47700, 'localhost', () => {
// Send HTTP request for the challenge token // Send HTTP request for the challenge token
testClient.write( testClient.write(
`GET ${challengePath} HTTP/1.1\r\n` + `GET ${challengePath} HTTP/1.1\r\n` +
@@ -113,7 +113,7 @@ tap.test('should return 404 for non-existent challenge tokens', async (tapTest)
routes: [{ routes: [{
name: 'acme-challenge-route', name: 'acme-challenge-route',
match: { match: {
ports: 8081, ports: 47701,
path: '/.well-known/acme-challenge/*' path: '/.well-known/acme-challenge/*'
}, },
action: { action: {
@@ -135,7 +135,7 @@ tap.test('should return 404 for non-existent challenge tokens', async (tapTest)
// Connect and send a request for a non-existent token // Connect and send a request for a non-existent token
await new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
testClient.connect(8081, 'localhost', () => { testClient.connect(47701, 'localhost', () => {
testClient.write( testClient.write(
'GET /.well-known/acme-challenge/invalid-token HTTP/1.1\r\n' + 'GET /.well-known/acme-challenge/invalid-token HTTP/1.1\r\n' +
'Host: test.example.com\r\n' + 'Host: test.example.com\r\n' +

View File

@@ -24,8 +24,8 @@ tap.test('setup test servers', async () => {
}); });
await new Promise<void>((resolve) => { await new Promise<void>((resolve) => {
testServer.listen(7001, '127.0.0.1', () => { testServer.listen(47712, '127.0.0.1', () => {
console.log('TCP test server listening on port 7001'); console.log('TCP test server listening on port 47712');
resolve(); resolve();
}); });
}); });
@@ -45,8 +45,8 @@ tap.test('setup test servers', async () => {
); );
await new Promise<void>((resolve) => { await new Promise<void>((resolve) => {
tlsTestServer.listen(7002, '127.0.0.1', () => { tlsTestServer.listen(47713, '127.0.0.1', () => {
console.log('TLS test server listening on port 7002'); console.log('TLS test server listening on port 47713');
resolve(); resolve();
}); });
}); });
@@ -60,13 +60,13 @@ tap.test('should forward TCP connections correctly', async () => {
{ {
name: 'tcp-forward', name: 'tcp-forward',
match: { match: {
ports: 8080, ports: 47710,
}, },
action: { action: {
type: 'forward', type: 'forward',
targets: [{ targets: [{
host: '127.0.0.1', host: '127.0.0.1',
port: 7001, port: 47712,
}], }],
}, },
}, },
@@ -77,7 +77,7 @@ tap.test('should forward TCP connections correctly', async () => {
// Test TCP forwarding // Test TCP forwarding
const client = await new Promise<net.Socket>((resolve, reject) => { const client = await new Promise<net.Socket>((resolve, reject) => {
const socket = net.connect(8080, '127.0.0.1', () => { const socket = net.connect(47710, '127.0.0.1', () => {
console.log('Connected to proxy'); console.log('Connected to proxy');
resolve(socket); resolve(socket);
}); });
@@ -106,7 +106,7 @@ tap.test('should handle TLS passthrough correctly', async () => {
{ {
name: 'tls-passthrough', name: 'tls-passthrough',
match: { match: {
ports: 8443, ports: 47711,
domains: 'test.example.com', domains: 'test.example.com',
}, },
action: { action: {
@@ -116,7 +116,7 @@ tap.test('should handle TLS passthrough correctly', async () => {
}, },
targets: [{ targets: [{
host: '127.0.0.1', host: '127.0.0.1',
port: 7002, port: 47713,
}], }],
}, },
}, },
@@ -129,7 +129,7 @@ tap.test('should handle TLS passthrough correctly', async () => {
const client = await new Promise<tls.TLSSocket>((resolve, reject) => { const client = await new Promise<tls.TLSSocket>((resolve, reject) => {
const socket = tls.connect( const socket = tls.connect(
{ {
port: 8443, port: 47711,
host: '127.0.0.1', host: '127.0.0.1',
servername: 'test.example.com', servername: 'test.example.com',
rejectUnauthorized: false, rejectUnauthorized: false,
@@ -164,7 +164,7 @@ tap.test('should handle SNI-based forwarding', async () => {
{ {
name: 'domain-a', name: 'domain-a',
match: { match: {
ports: 8443, ports: 47711,
domains: 'a.example.com', domains: 'a.example.com',
}, },
action: { action: {
@@ -174,14 +174,14 @@ tap.test('should handle SNI-based forwarding', async () => {
}, },
targets: [{ targets: [{
host: '127.0.0.1', host: '127.0.0.1',
port: 7002, port: 47713,
}], }],
}, },
}, },
{ {
name: 'domain-b', name: 'domain-b',
match: { match: {
ports: 8443, ports: 47711,
domains: 'b.example.com', domains: 'b.example.com',
}, },
action: { action: {
@@ -191,7 +191,7 @@ tap.test('should handle SNI-based forwarding', async () => {
}, },
targets: [{ targets: [{
host: '127.0.0.1', host: '127.0.0.1',
port: 7002, port: 47713,
}], }],
}, },
}, },
@@ -204,7 +204,7 @@ tap.test('should handle SNI-based forwarding', async () => {
const clientA = await new Promise<tls.TLSSocket>((resolve, reject) => { const clientA = await new Promise<tls.TLSSocket>((resolve, reject) => {
const socket = tls.connect( const socket = tls.connect(
{ {
port: 8443, port: 47711,
host: '127.0.0.1', host: '127.0.0.1',
servername: 'a.example.com', servername: 'a.example.com',
rejectUnauthorized: false, rejectUnauthorized: false,
@@ -231,7 +231,7 @@ tap.test('should handle SNI-based forwarding', async () => {
const clientB = await new Promise<tls.TLSSocket>((resolve, reject) => { const clientB = await new Promise<tls.TLSSocket>((resolve, reject) => {
const socket = tls.connect( const socket = tls.connect(
{ {
port: 8443, port: 47711,
host: '127.0.0.1', host: '127.0.0.1',
servername: 'b.example.com', servername: 'b.example.com',
rejectUnauthorized: false, rejectUnauthorized: false,

View File

@@ -21,8 +21,8 @@ tap.test('forward connections should not be immediately closed', async (t) => {
// Listen on a non-privileged port // Listen on a non-privileged port
await new Promise<void>((resolve) => { await new Promise<void>((resolve) => {
testServer.listen(9090, '127.0.0.1', () => { testServer.listen(47721, '127.0.0.1', () => {
console.log('Test server listening on port 9090'); console.log('Test server listening on port 47721');
resolve(); resolve();
}); });
}); });
@@ -34,13 +34,13 @@ tap.test('forward connections should not be immediately closed', async (t) => {
{ {
name: 'forward-test', name: 'forward-test',
match: { match: {
ports: 8080, ports: 47720,
}, },
action: { action: {
type: 'forward', type: 'forward',
targets: [{ targets: [{
host: '127.0.0.1', host: '127.0.0.1',
port: 9090, port: 47721,
}], }],
}, },
}, },
@@ -51,7 +51,7 @@ tap.test('forward connections should not be immediately closed', async (t) => {
// Create a client connection through the proxy // Create a client connection through the proxy
const client = net.createConnection({ const client = net.createConnection({
port: 8080, port: 47720,
host: '127.0.0.1', host: '127.0.0.1',
}); });

View File

@@ -4,7 +4,7 @@ import * as http from 'http';
tap.test('should forward HTTP connections on port 8080', async (tapTest) => { tap.test('should forward HTTP connections on port 8080', async (tapTest) => {
// Create a mock HTTP server to act as our target // Create a mock HTTP server to act as our target
const targetPort = 8181; const targetPort = 47732;
let receivedRequest = false; let receivedRequest = false;
let receivedPath = ''; let receivedPath = '';
@@ -36,7 +36,7 @@ tap.test('should forward HTTP connections on port 8080', async (tapTest) => {
routes: [{ routes: [{
name: 'test-route', name: 'test-route',
match: { match: {
ports: 8080 ports: 47730
// Remove domain restriction for HTTP connections // Remove domain restriction for HTTP connections
// Domain matching happens after HTTP headers are received // Domain matching happens after HTTP headers are received
}, },
@@ -55,7 +55,7 @@ tap.test('should forward HTTP connections on port 8080', async (tapTest) => {
// Make an HTTP request to port 8080 // Make an HTTP request to port 8080
const options = { const options = {
hostname: 'localhost', hostname: 'localhost',
port: 8080, port: 47730,
path: '/.well-known/acme-challenge/test-token', path: '/.well-known/acme-challenge/test-token',
method: 'GET', method: 'GET',
headers: { headers: {
@@ -104,7 +104,7 @@ tap.test('should forward HTTP connections on port 8080', async (tapTest) => {
tap.test('should handle basic HTTP request forwarding', async (tapTest) => { tap.test('should handle basic HTTP request forwarding', async (tapTest) => {
// Create a simple target server // Create a simple target server
const targetPort = 8182; const targetPort = 47733;
let receivedRequest = false; let receivedRequest = false;
const targetServer = http.createServer((req, res) => { const targetServer = http.createServer((req, res) => {
@@ -126,7 +126,7 @@ tap.test('should handle basic HTTP request forwarding', async (tapTest) => {
routes: [{ routes: [{
name: 'simple-forward', name: 'simple-forward',
match: { match: {
ports: 8081 ports: 47731
// Remove domain restriction for HTTP connections // Remove domain restriction for HTTP connections
}, },
action: { action: {
@@ -142,7 +142,7 @@ tap.test('should handle basic HTTP request forwarding', async (tapTest) => {
// Make request // Make request
const options = { const options = {
hostname: 'localhost', hostname: 'localhost',
port: 8081, port: 47731,
path: '/test', path: '/test',
method: 'GET', method: 'GET',
headers: { headers: {

View File

@@ -0,0 +1,472 @@
import { tap, expect } from '@git.zone/tstest/tapbundle';
import { SmartProxy } from '../ts/index.js';
import * as http from 'http';
import * as https from 'https';
import * as http2 from 'http2';
import * as net from 'net';
import * as tls from 'tls';
import * as fs from 'fs';
import * as path from 'path';
// ---------------------------------------------------------------------------
// Port assignments (4760047620 range to avoid conflicts)
// ---------------------------------------------------------------------------
const HTTP_ECHO_PORT = 47600; // backend HTTP echo server
const PROXY_HTTP_PORT = 47601; // SmartProxy plain HTTP forwarding
const PROXY_HTTPS_PORT = 47602; // SmartProxy TLS-terminate HTTPS forwarding
const TCP_ECHO_PORT = 47603; // backend TCP echo server
const PROXY_TCP_PORT = 47604; // SmartProxy plain TCP forwarding
// ---------------------------------------------------------------------------
// Shared state
// ---------------------------------------------------------------------------
let httpEchoServer: http.Server;
let tcpEchoServer: net.Server;
let proxy: SmartProxy;
const certPem = fs.readFileSync(path.join(import.meta.dirname, '..', 'assets', 'certs', 'cert.pem'), 'utf8');
const keyPem = fs.readFileSync(path.join(import.meta.dirname, '..', 'assets', 'certs', 'key.pem'), 'utf8');
// ---------------------------------------------------------------------------
// Helper: make an HTTP request and return { status, body }
// ---------------------------------------------------------------------------
function httpRequest(
options: http.RequestOptions,
body?: string,
): Promise<{ status: number; body: string }> {
return new Promise((resolve, reject) => {
const req = http.request(options, (res) => {
let data = '';
res.on('data', (chunk: string) => (data += chunk));
res.on('end', () => resolve({ status: res.statusCode!, body: data }));
});
req.on('error', reject);
req.setTimeout(5000, () => {
req.destroy(new Error('timeout'));
});
if (body) req.end(body);
else req.end();
});
}
// Same but for HTTPS
function httpsRequest(
options: https.RequestOptions,
body?: string,
): Promise<{ status: number; body: string }> {
return new Promise((resolve, reject) => {
const req = https.request(options, (res) => {
let data = '';
res.on('data', (chunk: string) => (data += chunk));
res.on('end', () => resolve({ status: res.statusCode!, body: data }));
});
req.on('error', reject);
req.setTimeout(5000, () => {
req.destroy(new Error('timeout'));
});
if (body) req.end(body);
else req.end();
});
}
// Helper: wait for metrics to settle on a condition
async function waitForMetrics(
metrics: ReturnType<SmartProxy['getMetrics']>,
condition: () => boolean,
maxWaitMs = 3000,
): Promise<void> {
const start = Date.now();
while (Date.now() - start < maxWaitMs) {
// Force a fresh poll
await (proxy as any).metricsAdapter.poll();
if (condition()) return;
await new Promise((r) => setTimeout(r, 100));
}
}
// ===========================================================================
// 1. Setup backend servers
// ===========================================================================
tap.test('setup - backend servers', async () => {
// HTTP echo server: POST → echo:<body>, GET → ok
httpEchoServer = http.createServer((req, res) => {
if (req.method === 'POST') {
let body = '';
req.on('data', (chunk: string) => (body += chunk));
req.on('end', () => {
res.writeHead(200, { 'Content-Type': 'text/plain' });
res.end(`echo:${body}`);
});
} else {
res.writeHead(200, { 'Content-Type': 'text/plain' });
res.end('ok');
}
});
await new Promise<void>((resolve, reject) => {
httpEchoServer.on('error', reject);
httpEchoServer.listen(HTTP_ECHO_PORT, () => {
console.log(`HTTP echo server on port ${HTTP_ECHO_PORT}`);
resolve();
});
});
// TCP echo server
tcpEchoServer = net.createServer((socket) => {
socket.on('data', (data) => socket.write(data));
});
await new Promise<void>((resolve, reject) => {
tcpEchoServer.on('error', reject);
tcpEchoServer.listen(TCP_ECHO_PORT, () => {
console.log(`TCP echo server on port ${TCP_ECHO_PORT}`);
resolve();
});
});
});
// ===========================================================================
// 2. Setup SmartProxy
// ===========================================================================
tap.test('setup - SmartProxy with 3 routes', async () => {
proxy = new SmartProxy({
routes: [
// Plain HTTP forward: 47601 → 47600
{
name: 'http-forward',
match: { ports: PROXY_HTTP_PORT },
action: {
type: 'forward',
targets: [{ host: 'localhost', port: HTTP_ECHO_PORT }],
},
},
// TLS-terminate HTTPS: 47602 → 47600
{
name: 'https-terminate',
match: { ports: PROXY_HTTPS_PORT, domains: 'localhost' },
action: {
type: 'forward',
targets: [{ host: 'localhost', port: HTTP_ECHO_PORT }],
tls: {
mode: 'terminate',
certificate: {
key: keyPem,
cert: certPem,
},
},
},
},
// Plain TCP forward: 47604 → 47603
{
name: 'tcp-forward',
match: { ports: PROXY_TCP_PORT },
action: {
type: 'forward',
targets: [{ host: 'localhost', port: TCP_ECHO_PORT }],
},
},
],
metrics: {
enabled: true,
sampleIntervalMs: 100,
},
enableDetailedLogging: false,
});
await proxy.start();
// Give the proxy a moment to fully bind
await new Promise((r) => setTimeout(r, 500));
});
// ===========================================================================
// 3. HTTP/1.1 connection pooling: sequential requests reuse connections
// ===========================================================================
tap.test('HTTP/1.1 connection pooling: sequential requests reuse connections', async (tools) => {
tools.timeout(30000);
const metrics = proxy.getMetrics();
const REQUEST_COUNT = 20;
// Use a non-keepalive agent so each request closes the client→proxy socket
// (Rust's backend connection pool still reuses proxy→backend connections)
const agent = new http.Agent({ keepAlive: false });
for (let i = 0; i < REQUEST_COUNT; i++) {
const result = await httpRequest(
{
hostname: 'localhost',
port: PROXY_HTTP_PORT,
path: '/echo',
method: 'POST',
headers: { 'Content-Type': 'text/plain' },
agent,
},
`msg-${i}`,
);
expect(result.status).toEqual(200);
expect(result.body).toEqual(`echo:msg-${i}`);
}
agent.destroy();
// Wait for all connections to settle and metrics to update
await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
expect(metrics.connections.active()).toEqual(0);
// Bytes should have been transferred
await waitForMetrics(metrics, () => metrics.totals.bytesIn() > 0);
expect(metrics.totals.bytesIn()).toBeGreaterThan(0);
expect(metrics.totals.bytesOut()).toBeGreaterThan(0);
console.log(`HTTP pooling test: ${REQUEST_COUNT} requests completed. bytesIn=${metrics.totals.bytesIn()}, bytesOut=${metrics.totals.bytesOut()}`);
});
// ===========================================================================
// 4. HTTPS with TLS termination: multiple requests through TLS
// ===========================================================================
tap.test('HTTPS with TLS termination: multiple requests through TLS', async (tools) => {
tools.timeout(30000);
const REQUEST_COUNT = 10;
const agent = new https.Agent({ keepAlive: false, rejectUnauthorized: false });
for (let i = 0; i < REQUEST_COUNT; i++) {
const result = await httpsRequest(
{
hostname: 'localhost',
port: PROXY_HTTPS_PORT,
path: '/echo',
method: 'POST',
headers: { 'Content-Type': 'text/plain' },
rejectUnauthorized: false,
servername: 'localhost',
agent,
},
`tls-${i}`,
);
expect(result.status).toEqual(200);
expect(result.body).toEqual(`echo:tls-${i}`);
}
agent.destroy();
console.log(`HTTPS termination test: ${REQUEST_COUNT} requests completed successfully`);
});
// ===========================================================================
// 5. TLS ALPN negotiation verification
// ===========================================================================
tap.test('HTTP/2 end-to-end: ALPN h2 with multiplexed requests', async (tools) => {
tools.timeout(15000);
// Connect an HTTP/2 session over TLS
const session = http2.connect(`https://localhost:${PROXY_HTTPS_PORT}`, {
rejectUnauthorized: false,
});
await new Promise<void>((resolve, reject) => {
session.on('connect', () => resolve());
session.on('error', reject);
setTimeout(() => reject(new Error('h2 connect timeout')), 5000);
});
// Verify ALPN negotiated h2
const alpnProtocol = (session.socket as tls.TLSSocket).alpnProtocol;
console.log(`TLS ALPN negotiated protocol: ${alpnProtocol}`);
expect(alpnProtocol).toEqual('h2');
// Send 5 multiplexed POST requests on the same h2 session
const REQUEST_COUNT = 5;
const promises: Promise<{ status: number; body: string }>[] = [];
for (let i = 0; i < REQUEST_COUNT; i++) {
promises.push(
new Promise<{ status: number; body: string }>((resolve, reject) => {
const reqStream = session.request({
':method': 'POST',
':path': '/echo',
'content-type': 'text/plain',
});
let data = '';
let status = 0;
reqStream.on('response', (headers) => {
status = headers[':status'] as number;
});
reqStream.on('data', (chunk: Buffer) => {
data += chunk.toString();
});
reqStream.on('end', () => resolve({ status, body: data }));
reqStream.on('error', reject);
reqStream.end(`h2-msg-${i}`);
}),
);
}
const results = await Promise.all(promises);
for (let i = 0; i < REQUEST_COUNT; i++) {
expect(results[i].status).toEqual(200);
expect(results[i].body).toEqual(`echo:h2-msg-${i}`);
}
await new Promise<void>((resolve) => session.close(() => resolve()));
console.log(`HTTP/2 end-to-end: ${REQUEST_COUNT} multiplexed requests completed successfully`);
});
// ===========================================================================
// 6. Connection stability: no leaked connections after repeated open/close
// ===========================================================================
tap.test('connection stability: no leaked connections after repeated open/close', async (tools) => {
tools.timeout(60000);
const metrics = proxy.getMetrics();
const BATCH_SIZE = 50;
// Ensure we start clean
await waitForMetrics(metrics, () => metrics.connections.active() === 0);
// Record total connections before
await (proxy as any).metricsAdapter.poll();
const totalBefore = metrics.connections.total();
// --- Batch 1: 50 sequential TCP connections ---
for (let i = 0; i < BATCH_SIZE; i++) {
await new Promise<void>((resolve, reject) => {
const client = new net.Socket();
client.connect(PROXY_TCP_PORT, 'localhost', () => {
const msg = `batch1-${i}`;
client.write(msg);
client.once('data', (data) => {
expect(data.toString()).toEqual(msg);
client.end();
});
});
client.on('close', () => resolve());
client.on('error', reject);
client.setTimeout(5000, () => {
client.destroy(new Error('timeout'));
});
});
}
// Wait for all connections to drain
await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
expect(metrics.connections.active()).toEqual(0);
console.log(`Batch 1 done: active=${metrics.connections.active()}, total=${metrics.connections.total()}`);
// --- Batch 2: another 50 ---
for (let i = 0; i < BATCH_SIZE; i++) {
await new Promise<void>((resolve, reject) => {
const client = new net.Socket();
client.connect(PROXY_TCP_PORT, 'localhost', () => {
const msg = `batch2-${i}`;
client.write(msg);
client.once('data', (data) => {
expect(data.toString()).toEqual(msg);
client.end();
});
});
client.on('close', () => resolve());
client.on('error', reject);
client.setTimeout(5000, () => {
client.destroy(new Error('timeout'));
});
});
}
// Wait for all connections to drain again
await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
expect(metrics.connections.active()).toEqual(0);
// Total should reflect ~100 new connections
await (proxy as any).metricsAdapter.poll();
const totalAfter = metrics.connections.total();
const newConnections = totalAfter - totalBefore;
console.log(`Batch 2 done: active=${metrics.connections.active()}, total=${totalAfter}, new=${newConnections}`);
expect(newConnections).toBeGreaterThanOrEqual(BATCH_SIZE * 2);
});
// ===========================================================================
// 7. Concurrent connections: burst and drain
// ===========================================================================
tap.test('concurrent connections: burst and drain', async (tools) => {
tools.timeout(30000);
const metrics = proxy.getMetrics();
const CONCURRENT = 20;
// Ensure we start clean
await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
// Open 20 TCP connections simultaneously
const clients: net.Socket[] = [];
const connectPromises: Promise<void>[] = [];
for (let i = 0; i < CONCURRENT; i++) {
const client = new net.Socket();
clients.push(client);
connectPromises.push(
new Promise<void>((resolve, reject) => {
client.connect(PROXY_TCP_PORT, 'localhost', () => resolve());
client.on('error', reject);
client.setTimeout(5000, () => {
client.destroy(new Error('timeout'));
});
}),
);
}
await Promise.all(connectPromises);
// Send data on all connections and wait for echo
const echoPromises = clients.map((client, i) => {
return new Promise<void>((resolve, reject) => {
const msg = `concurrent-${i}`;
client.once('data', (data) => {
expect(data.toString()).toEqual(msg);
resolve();
});
client.write(msg);
client.on('error', reject);
});
});
await Promise.all(echoPromises);
// Poll metrics — active connections should be CONCURRENT
await waitForMetrics(metrics, () => metrics.connections.active() >= CONCURRENT, 3000);
const activeWhileOpen = metrics.connections.active();
console.log(`Burst: active connections while open = ${activeWhileOpen}`);
expect(activeWhileOpen).toBeGreaterThanOrEqual(CONCURRENT);
// Close all connections
for (const client of clients) {
client.end();
}
// Wait for drain
await waitForMetrics(metrics, () => metrics.connections.active() === 0, 5000);
expect(metrics.connections.active()).toEqual(0);
console.log('Drain: all connections closed, active=0');
});
// ===========================================================================
// 8. Cleanup
// ===========================================================================
tap.test('cleanup', async () => {
await proxy.stop();
await new Promise<void>((resolve) => {
httpEchoServer.close(() => {
console.log('HTTP echo server closed');
resolve();
});
});
await new Promise<void>((resolve) => {
tcpEchoServer.close(() => {
console.log('TCP echo server closed');
resolve();
});
});
});
export default tap.start();

View File

@@ -14,8 +14,8 @@ import type { IRouteConfig, IRouteContext } from '../ts/proxies/smart-proxy/mode
let testServers: Array<{ server: net.Server; port: number }> = []; let testServers: Array<{ server: net.Server; port: number }> = [];
let smartProxy: SmartProxy; let smartProxy: SmartProxy;
const TEST_PORT_START = 4000; const TEST_PORT_START = 47750;
const PROXY_PORT_START = 5000; const PROXY_PORT_START = 48750;
const TEST_DATA = 'Hello through dynamic port mapper!'; const TEST_DATA = 'Hello through dynamic port mapper!';
// Cleanup function to close all servers and proxies // Cleanup function to close all servers and proxies
@@ -103,9 +103,9 @@ function createTestClient(port: number, data: string): Promise<string> {
tap.test('setup port mapping test environment', async () => { tap.test('setup port mapping test environment', async () => {
// Create multiple test servers on different ports // Create multiple test servers on different ports
await Promise.all([ await Promise.all([
createTestServer(TEST_PORT_START), // Server on port 4000 createTestServer(TEST_PORT_START), // Server on port 47750
createTestServer(TEST_PORT_START + 1), // Server on port 4001 createTestServer(TEST_PORT_START + 1), // Server on port 47751
createTestServer(TEST_PORT_START + 2), // Server on port 4002 createTestServer(TEST_PORT_START + 2), // Server on port 47752
]); ]);
// Create a SmartProxy with dynamic port mapping routes // Create a SmartProxy with dynamic port mapping routes
@@ -119,7 +119,7 @@ tap.test('setup port mapping test environment', async () => {
name: 'Identity Port Mapping' name: 'Identity Port Mapping'
}), }),
// Offset port mapping from 5001 to 4001 (offset -1000) // Offset port mapping from 48751 to 47751 (offset -1000)
createOffsetPortMappingRoute({ createOffsetPortMappingRoute({
ports: PROXY_PORT_START + 1, ports: PROXY_PORT_START + 1,
targetHost: 'localhost', targetHost: 'localhost',
@@ -170,13 +170,13 @@ tap.test('setup port mapping test environment', async () => {
await smartProxy.start(); await smartProxy.start();
}); });
// Test 1: Simple identity port mapping (5000 -> 4000) // Test 1: Simple identity port mapping (48750 -> 47750)
tap.test('should map port using identity function', async () => { tap.test('should map port using identity function', async () => {
const response = await createTestClient(PROXY_PORT_START, TEST_DATA); const response = await createTestClient(PROXY_PORT_START, TEST_DATA);
expect(response).toEqual(`Server ${TEST_PORT_START} says: ${TEST_DATA}`); expect(response).toEqual(`Server ${TEST_PORT_START} says: ${TEST_DATA}`);
}); });
// Test 2: Offset port mapping (5001 -> 4001) // Test 2: Offset port mapping (48751 -> 47751)
tap.test('should map port using offset function', async () => { tap.test('should map port using offset function', async () => {
const response = await createTestClient(PROXY_PORT_START + 1, TEST_DATA); const response = await createTestClient(PROXY_PORT_START + 1, TEST_DATA);
expect(response).toEqual(`Server ${TEST_PORT_START + 1} says: ${TEST_DATA}`); expect(response).toEqual(`Server ${TEST_PORT_START + 1} says: ${TEST_DATA}`);

View File

@@ -4,8 +4,8 @@ import { SmartProxy } from '../ts/proxies/smart-proxy/index.js';
let testServer: net.Server; let testServer: net.Server;
let smartProxy: SmartProxy; let smartProxy: SmartProxy;
const TEST_SERVER_PORT = 4000; const TEST_SERVER_PORT = 47770;
const PROXY_PORT = 4001; const PROXY_PORT = 47771;
const TEST_DATA = 'Hello through port proxy!'; const TEST_DATA = 'Hello through port proxy!';
// Track all created servers and proxies for proper cleanup // Track all created servers and proxies for proper cleanup

View File

@@ -10,7 +10,7 @@ tap.test('setup socket handler test', async () => {
const routes: IRouteConfig[] = [{ const routes: IRouteConfig[] = [{
name: 'echo-handler', name: 'echo-handler',
match: { match: {
ports: 9999 ports: 47780
// No domains restriction - matches all connections // No domains restriction - matches all connections
}, },
action: { action: {
@@ -43,7 +43,7 @@ tap.test('should handle socket with custom function', async () => {
let response = ''; let response = '';
await new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
client.connect(9999, 'localhost', () => { client.connect(47780, 'localhost', () => {
console.log('Client connected to proxy'); console.log('Client connected to proxy');
resolve(); resolve();
}); });
@@ -78,7 +78,7 @@ tap.test('should handle async socket handler', async () => {
// Update route with async handler // Update route with async handler
await proxy.updateRoutes([{ await proxy.updateRoutes([{
name: 'async-handler', name: 'async-handler',
match: { ports: 9999 }, match: { ports: 47780 },
action: { action: {
type: 'socket-handler', type: 'socket-handler',
socketHandler: async (socket, context) => { socketHandler: async (socket, context) => {
@@ -108,7 +108,7 @@ tap.test('should handle async socket handler', async () => {
}); });
await new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
client.connect(9999, 'localhost', () => { client.connect(47780, 'localhost', () => {
// Send initial data to trigger the handler // Send initial data to trigger the handler
client.write('test data\n'); client.write('test data\n');
resolve(); resolve();
@@ -131,7 +131,7 @@ tap.test('should handle errors in socket handler', async () => {
// Update route with error-throwing handler // Update route with error-throwing handler
await proxy.updateRoutes([{ await proxy.updateRoutes([{
name: 'error-handler', name: 'error-handler',
match: { ports: 9999 }, match: { ports: 47780 },
action: { action: {
type: 'socket-handler', type: 'socket-handler',
socketHandler: (socket, context) => { socketHandler: (socket, context) => {
@@ -148,7 +148,7 @@ tap.test('should handle errors in socket handler', async () => {
}); });
await new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
client.connect(9999, 'localhost', () => { client.connect(47780, 'localhost', () => {
// Connection established - send data to trigger handler // Connection established - send data to trigger handler
client.write('trigger\n'); client.write('trigger\n');
resolve(); resolve();

View File

@@ -3,6 +3,6 @@
*/ */
export const commitinfo = { export const commitinfo = {
name: '@push.rocks/smartproxy', name: '@push.rocks/smartproxy',
version: '25.7.8', version: '25.7.9',
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.' 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.'
} }