Compare commits
12 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0ae882731a | |||
| 53d73c7dc6 | |||
| b4b8bd925d | |||
| 5ac44b898b | |||
| 9b4393b5ac | |||
| 02b4ed8018 | |||
| e4e4b4f1ec | |||
| d361a21543 | |||
| 106713a546 | |||
| 101675b5f8 | |||
| 9fac17bc39 | |||
| 2e3cf515a4 |
50
changelog.md
50
changelog.md
@@ -1,5 +1,55 @@
|
||||
# Changelog
|
||||
|
||||
## 2026-02-19 - 25.7.5 - fix(rustproxy)
|
||||
prune stale per-route metrics, add per-route rate limiter caching and regex cache, and improve connection tracking cleanup to prevent memory growth
|
||||
|
||||
- Prune per-route metrics for routes removed from configuration via MetricsCollector::retain_routes invoked during route table updates
|
||||
- Introduce per-route shared RateLimiter instances (DashMap) with a request-count-triggered periodic cleanup to avoid stale limiters
|
||||
- Cache compiled URL-rewrite regexes (regex_cache) to avoid recompiling patterns on every request and insert compiled regex on first use
|
||||
- Improve upstream connection tracking to remove zero-count entries and guard against underflow, preventing unbounded DashMap growth
|
||||
- Evict per-IP metrics and timestamps when the last connection for an IP closes so per-IP DashMap entries are fully freed
|
||||
- Add unit tests validating connection tracking cleanup, per-IP eviction, and route-metrics retention behavior
|
||||
|
||||
## 2026-02-19 - 25.7.4 - fix(smart-proxy)
|
||||
include proxy IPs in smart proxy configuration
|
||||
|
||||
- Add proxyIps: this.settings.proxyIPs to proxy options in ts/proxies/smart-proxy/smart-proxy.ts
|
||||
- Ensures proxy IPs from settings are passed into the proxy implementation (enables proxy IP filtering/whitelisting)
|
||||
|
||||
## 2026-02-16 - 25.7.3 - fix(metrics)
|
||||
centralize connection-closed reporting via ConnectionGuard and remove duplicate explicit metrics.connection_closed calls
|
||||
|
||||
- Removed numerous explicit metrics.connection_closed calls from rust/crates/rustproxy-http/src/proxy_service.rs so connection teardown and byte counting are handled by the connection guard / counting body instead of ad-hoc calls.
|
||||
- Simplified ConnectionGuard in rust/crates/rustproxy-passthrough/src/tcp_listener.rs: removed the disarm flag and disarm() method so Drop always reports connection_closed.
|
||||
- Stopped disarming the TCP-level guard when handing connections off to HTTP proxy paths (HTTP/WebSocket/streaming flows) to avoid missing or double-reporting metrics.
|
||||
- Fixes incorrect/duplicate connection-closed metric emission and ensures consistent byte/connection accounting during streaming and WebSocket upgrades.
|
||||
|
||||
## 2026-02-16 - 25.7.2 - fix(rustproxy-http)
|
||||
preserve original Host header when proxying and add X-Forwarded-* headers; add TLS WebSocket echo backend helper and integration test for terminate-and-reencrypt websocket
|
||||
|
||||
- Preserve the client's original Host header instead of replacing it with backend host:port when proxying requests.
|
||||
- Add standard reverse-proxy headers: X-Forwarded-For (appends client IP), X-Forwarded-Host, and X-Forwarded-Proto for upstream requests.
|
||||
- Ensure raw TCP/HTTP upstream requests copy original headers and skip X-Forwarded-* (which are added explicitly).
|
||||
- Add start_tls_ws_echo_backend test helper to start a TLS WebSocket echo backend for tests.
|
||||
- Add integration test test_terminate_and_reencrypt_websocket to verify WS upgrade through terminate-and-reencrypt TLS path.
|
||||
- Rename unused parameter upstream to _upstream in proxy_service functions to avoid warnings.
|
||||
|
||||
## 2026-02-16 - 25.7.1 - fix(proxy)
|
||||
use TLS to backends for terminate-and-reencrypt routes
|
||||
|
||||
- Set upstream.use_tls = true when a route's TLS mode is TerminateAndReencrypt so the proxy re-encrypts to backend servers.
|
||||
- Add start_tls_http_backend test helper and update integration tests to run TLS-enabled backend servers validating re-encryption behavior.
|
||||
- Make the selected upstream mutable to allow toggling the use_tls flag during request handling.
|
||||
|
||||
## 2026-02-16 - 25.7.0 - feat(routes)
|
||||
add protocol-based route matching and ensure terminate-and-reencrypt routes HTTP through the full HTTP proxy; update docs and tests
|
||||
|
||||
- Introduce a new 'protocol' match field for routes (supports 'http' and 'tcp') and preserve it through cloning/merging.
|
||||
- Add Rust integration test verifying terminate-and-reencrypt decrypts TLS and routes HTTP traffic via the HTTP proxy (per-request Host/path routing) instead of raw tunneling.
|
||||
- Add TypeScript unit tests covering protocol field validation, preservation, interaction with terminate-and-reencrypt, cloning, merging, and matching behavior.
|
||||
- Update README with a Protocol-Specific Routing section and clarify terminate-and-reencrypt behavior (HTTP routed via HTTP proxy; non-HTTP uses raw TLS-to-TLS tunnel).
|
||||
- Example config: include health check thresholds (unhealthyThreshold and healthyThreshold) in the sample healthCheck settings.
|
||||
|
||||
## 2026-02-16 - 25.6.0 - feat(rustproxy)
|
||||
add protocol-based routing and backend TLS re-encryption support
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@push.rocks/smartproxy",
|
||||
"version": "25.6.0",
|
||||
"version": "25.7.5",
|
||||
"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",
|
||||
|
||||
57
readme.md
57
readme.md
@@ -27,7 +27,7 @@ Whether you're building microservices, deploying edge infrastructure, or need a
|
||||
| 🦀 **Rust-Powered Engine** | All networking handled by a high-performance Rust binary via IPC |
|
||||
| 🔀 **Unified Route-Based Config** | Clean match/action patterns for intuitive traffic routing |
|
||||
| 🔒 **Automatic SSL/TLS** | Zero-config HTTPS with Let's Encrypt ACME integration |
|
||||
| 🎯 **Flexible Matching** | Route by port, domain, path, client IP, TLS version, headers, or custom logic |
|
||||
| 🎯 **Flexible Matching** | Route by port, domain, path, protocol, client IP, TLS version, headers, or custom logic |
|
||||
| 🚄 **High-Performance** | Choose between user-space or kernel-level (NFTables) forwarding |
|
||||
| ⚖️ **Load Balancing** | Round-robin, least-connections, IP-hash with health checks |
|
||||
| 🛡️ **Enterprise Security** | IP filtering, rate limiting, basic auth, JWT auth, connection limits |
|
||||
@@ -89,7 +89,7 @@ SmartProxy uses a powerful **match/action** pattern that makes routing predictab
|
||||
```
|
||||
|
||||
Every route consists of:
|
||||
- **Match** — What traffic to capture (ports, domains, paths, IPs, headers)
|
||||
- **Match** — What traffic to capture (ports, domains, paths, protocol, IPs, headers)
|
||||
- **Action** — What to do with it (`forward` or `socket-handler`)
|
||||
- **Security** (optional) — IP allow/block lists, rate limits, authentication
|
||||
- **Headers** (optional) — Request/response header manipulation with template variables
|
||||
@@ -103,7 +103,7 @@ SmartProxy supports three TLS handling modes:
|
||||
|------|-------------|----------|
|
||||
| `passthrough` | Forward encrypted traffic as-is (SNI-based routing) | Backend handles TLS |
|
||||
| `terminate` | Decrypt at proxy, forward plain HTTP to backend | Standard reverse proxy |
|
||||
| `terminate-and-reencrypt` | Decrypt, then re-encrypt to backend | Zero-trust environments |
|
||||
| `terminate-and-reencrypt` | Decrypt at proxy, re-encrypt to backend. HTTP traffic gets full per-request routing (Host header, path matching) via the HTTP proxy; non-HTTP traffic uses a raw TLS-to-TLS tunnel | Zero-trust / defense-in-depth environments |
|
||||
|
||||
## 💡 Common Use Cases
|
||||
|
||||
@@ -135,13 +135,13 @@ const proxy = new SmartProxy({
|
||||
],
|
||||
{
|
||||
tls: { mode: 'terminate', certificate: 'auto' },
|
||||
loadBalancing: {
|
||||
algorithm: 'round-robin',
|
||||
healthCheck: {
|
||||
path: '/health',
|
||||
interval: 30000,
|
||||
timeout: 5000
|
||||
}
|
||||
algorithm: 'round-robin',
|
||||
healthCheck: {
|
||||
path: '/health',
|
||||
interval: 30000,
|
||||
timeout: 5000,
|
||||
unhealthyThreshold: 3,
|
||||
healthyThreshold: 2
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -318,6 +318,42 @@ const proxy = new SmartProxy({
|
||||
|
||||
> **Note:** Routes with dynamic functions (host/port callbacks) are automatically relayed through the TypeScript socket handler server, since JavaScript functions can't be serialized to Rust.
|
||||
|
||||
### 🔀 Protocol-Specific Routing
|
||||
|
||||
Restrict routes to specific application-layer protocols. When `protocol` is set, the Rust engine detects the protocol after connection (or after TLS termination) and only matches routes that accept that protocol:
|
||||
|
||||
```typescript
|
||||
// HTTP-only route (rejects raw TCP connections)
|
||||
const httpOnlyRoute: IRouteConfig = {
|
||||
name: 'http-api',
|
||||
match: {
|
||||
ports: 443,
|
||||
domains: 'api.example.com',
|
||||
protocol: 'http', // Only match HTTP/1.1, HTTP/2, and WebSocket upgrades
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'api-backend', port: 8080 }],
|
||||
tls: { mode: 'terminate', certificate: 'auto' }
|
||||
}
|
||||
};
|
||||
|
||||
// Raw TCP route (rejects HTTP traffic)
|
||||
const tcpOnlyRoute: IRouteConfig = {
|
||||
name: 'database-proxy',
|
||||
match: {
|
||||
ports: 5432,
|
||||
protocol: 'tcp', // Only match non-HTTP TCP streams
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'db-server', port: 5432 }]
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
> **Note:** Omitting `protocol` (the default) matches any protocol. For TLS routes, protocol detection happens *after* TLS termination — during the initial SNI-based route match, `protocol` is not yet known and the route is allowed to match. The protocol restriction is enforced after the proxy peeks at the decrypted data.
|
||||
|
||||
### 🔒 Security Controls
|
||||
|
||||
Comprehensive per-route security options:
|
||||
@@ -549,6 +585,7 @@ interface IRouteMatch {
|
||||
clientIp?: string[]; // ['10.0.0.0/8', '192.168.*']
|
||||
tlsVersion?: string[]; // ['TLSv1.2', 'TLSv1.3']
|
||||
headers?: Record<string, string | RegExp>; // Match by HTTP headers
|
||||
protocol?: 'http' | 'tcp'; // Match specific protocol ('http' includes h2 + WebSocket upgrades)
|
||||
}
|
||||
```
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
|
||||
use bytes::Bytes;
|
||||
use dashmap::DashMap;
|
||||
use http_body_util::{BodyExt, Full, combinators::BoxBody};
|
||||
use hyper::body::Incoming;
|
||||
use hyper::{Request, Response, StatusCode};
|
||||
@@ -23,6 +24,7 @@ use std::task::{Context, Poll};
|
||||
|
||||
use rustproxy_routing::RouteManager;
|
||||
use rustproxy_metrics::MetricsCollector;
|
||||
use rustproxy_security::RateLimiter;
|
||||
|
||||
use crate::counting_body::{CountingBody, Direction};
|
||||
use crate::request_filter::RequestFilter;
|
||||
@@ -164,6 +166,12 @@ pub struct HttpProxyService {
|
||||
upstream_selector: UpstreamSelector,
|
||||
/// Timeout for connecting to upstream backends.
|
||||
connect_timeout: std::time::Duration,
|
||||
/// Per-route rate limiters (keyed by route ID).
|
||||
route_rate_limiters: Arc<DashMap<String, Arc<RateLimiter>>>,
|
||||
/// Request counter for periodic rate limiter cleanup.
|
||||
request_counter: AtomicU64,
|
||||
/// Cache of compiled URL rewrite regexes (keyed by pattern string).
|
||||
regex_cache: DashMap<String, Regex>,
|
||||
}
|
||||
|
||||
impl HttpProxyService {
|
||||
@@ -173,6 +181,9 @@ impl HttpProxyService {
|
||||
metrics,
|
||||
upstream_selector: UpstreamSelector::new(),
|
||||
connect_timeout: DEFAULT_CONNECT_TIMEOUT,
|
||||
route_rate_limiters: Arc::new(DashMap::new()),
|
||||
request_counter: AtomicU64::new(0),
|
||||
regex_cache: DashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -187,6 +198,9 @@ impl HttpProxyService {
|
||||
metrics,
|
||||
upstream_selector: UpstreamSelector::new(),
|
||||
connect_timeout,
|
||||
route_rate_limiters: Arc::new(DashMap::new()),
|
||||
request_counter: AtomicU64::new(0),
|
||||
regex_cache: DashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -309,20 +323,37 @@ impl HttpProxyService {
|
||||
let route_id = route_match.route.id.as_deref();
|
||||
let ip_str = peer_addr.ip().to_string();
|
||||
self.metrics.record_http_request();
|
||||
self.metrics.connection_opened(route_id, Some(&ip_str));
|
||||
|
||||
// Apply request filters (IP check, rate limiting, auth)
|
||||
if let Some(ref security) = route_match.route.security {
|
||||
if let Some(response) = RequestFilter::apply(security, &req, &peer_addr) {
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
// Look up or create a shared rate limiter for this route
|
||||
let rate_limiter = security.rate_limit.as_ref()
|
||||
.filter(|rl| rl.enabled)
|
||||
.map(|rl| {
|
||||
let route_key = route_id.unwrap_or("__default__").to_string();
|
||||
self.route_rate_limiters
|
||||
.entry(route_key)
|
||||
.or_insert_with(|| Arc::new(RateLimiter::new(rl.max_requests, rl.window)))
|
||||
.clone()
|
||||
});
|
||||
if let Some(response) = RequestFilter::apply_with_rate_limiter(
|
||||
security, &req, &peer_addr, rate_limiter.as_ref(),
|
||||
) {
|
||||
return Ok(response);
|
||||
}
|
||||
}
|
||||
|
||||
// Periodic rate limiter cleanup (every 1000 requests)
|
||||
let count = self.request_counter.fetch_add(1, Ordering::Relaxed);
|
||||
if count % 1000 == 0 {
|
||||
for entry in self.route_rate_limiters.iter() {
|
||||
entry.value().cleanup();
|
||||
}
|
||||
}
|
||||
|
||||
// Check for test response (returns immediately, no upstream needed)
|
||||
if let Some(ref advanced) = route_match.route.action.advanced {
|
||||
if let Some(ref test_response) = advanced.test_response {
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
return Ok(Self::build_test_response(test_response));
|
||||
}
|
||||
}
|
||||
@@ -330,7 +361,6 @@ impl HttpProxyService {
|
||||
// Check for static file serving
|
||||
if let Some(ref advanced) = route_match.route.action.advanced {
|
||||
if let Some(ref static_files) = advanced.static_files {
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
return Ok(Self::serve_static_file(&path, static_files));
|
||||
}
|
||||
}
|
||||
@@ -339,12 +369,19 @@ impl HttpProxyService {
|
||||
let target = match route_match.target {
|
||||
Some(t) => t,
|
||||
None => {
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "No target available"));
|
||||
}
|
||||
};
|
||||
|
||||
let upstream = self.upstream_selector.select(target, &peer_addr, port);
|
||||
let mut upstream = self.upstream_selector.select(target, &peer_addr, port);
|
||||
|
||||
// If the route uses terminate-and-reencrypt, always re-encrypt to backend
|
||||
if let Some(ref tls) = route_match.route.action.tls {
|
||||
if tls.mode == rustproxy_config::TlsMode::TerminateAndReencrypt {
|
||||
upstream.use_tls = true;
|
||||
}
|
||||
}
|
||||
|
||||
let upstream_key = format!("{}:{}", upstream.host, upstream.port);
|
||||
self.upstream_selector.connection_started(&upstream_key);
|
||||
|
||||
@@ -376,7 +413,7 @@ impl HttpProxyService {
|
||||
Some(q) => format!("{}?{}", path, q),
|
||||
None => path.clone(),
|
||||
};
|
||||
Self::apply_url_rewrite(&raw_path, &route_match.route)
|
||||
self.apply_url_rewrite(&raw_path, &route_match.route)
|
||||
};
|
||||
|
||||
// Build upstream request - stream body instead of buffering
|
||||
@@ -396,6 +433,51 @@ impl HttpProxyService {
|
||||
}
|
||||
}
|
||||
|
||||
// Add standard reverse-proxy headers (X-Forwarded-*)
|
||||
{
|
||||
let original_host = parts.headers.get("host")
|
||||
.and_then(|h| h.to_str().ok())
|
||||
.unwrap_or("");
|
||||
let forwarded_proto = if route_match.route.action.tls.as_ref()
|
||||
.map(|t| matches!(t.mode,
|
||||
rustproxy_config::TlsMode::Terminate
|
||||
| rustproxy_config::TlsMode::TerminateAndReencrypt))
|
||||
.unwrap_or(false)
|
||||
{
|
||||
"https"
|
||||
} else {
|
||||
"http"
|
||||
};
|
||||
|
||||
// X-Forwarded-For: append client IP to existing chain
|
||||
let client_ip = peer_addr.ip().to_string();
|
||||
let xff_value = if let Some(existing) = upstream_headers.get("x-forwarded-for") {
|
||||
format!("{}, {}", existing.to_str().unwrap_or(""), client_ip)
|
||||
} else {
|
||||
client_ip
|
||||
};
|
||||
if let Ok(val) = hyper::header::HeaderValue::from_str(&xff_value) {
|
||||
upstream_headers.insert(
|
||||
hyper::header::HeaderName::from_static("x-forwarded-for"),
|
||||
val,
|
||||
);
|
||||
}
|
||||
// X-Forwarded-Host: original Host header
|
||||
if let Ok(val) = hyper::header::HeaderValue::from_str(original_host) {
|
||||
upstream_headers.insert(
|
||||
hyper::header::HeaderName::from_static("x-forwarded-host"),
|
||||
val,
|
||||
);
|
||||
}
|
||||
// X-Forwarded-Proto: original client protocol
|
||||
if let Ok(val) = hyper::header::HeaderValue::from_str(forwarded_proto) {
|
||||
upstream_headers.insert(
|
||||
hyper::header::HeaderName::from_static("x-forwarded-proto"),
|
||||
val,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Connect to upstream with timeout (TLS if upstream.use_tls is set)
|
||||
let backend = if upstream.use_tls {
|
||||
match tokio::time::timeout(
|
||||
@@ -406,13 +488,11 @@ impl HttpProxyService {
|
||||
Ok(Err(e)) => {
|
||||
error!("Failed TLS connect to upstream {}:{}: {}", upstream.host, upstream.port, e);
|
||||
self.upstream_selector.connection_ended(&upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend TLS unavailable"));
|
||||
}
|
||||
Err(_) => {
|
||||
error!("Upstream TLS connect timeout for {}:{}", upstream.host, upstream.port);
|
||||
self.upstream_selector.connection_ended(&upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend TLS connect timeout"));
|
||||
}
|
||||
}
|
||||
@@ -428,13 +508,11 @@ impl HttpProxyService {
|
||||
Ok(Err(e)) => {
|
||||
error!("Failed to connect to upstream {}:{}: {}", upstream.host, upstream.port, e);
|
||||
self.upstream_selector.connection_ended(&upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable"));
|
||||
}
|
||||
Err(_) => {
|
||||
error!("Upstream connect timeout for {}:{}", upstream.host, upstream.port);
|
||||
self.upstream_selector.connection_ended(&upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(&ip_str));
|
||||
return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend connect timeout"));
|
||||
}
|
||||
}
|
||||
@@ -461,7 +539,7 @@ impl HttpProxyService {
|
||||
body: Incoming,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
_upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
route: &rustproxy_config::RouteConfig,
|
||||
route_id: Option<&str>,
|
||||
source_ip: &str,
|
||||
@@ -470,7 +548,6 @@ impl HttpProxyService {
|
||||
Ok(h) => h,
|
||||
Err(e) => {
|
||||
error!("Upstream handshake failed: {}", e);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend handshake failed"));
|
||||
}
|
||||
};
|
||||
@@ -488,11 +565,6 @@ impl HttpProxyService {
|
||||
|
||||
if let Some(headers) = upstream_req.headers_mut() {
|
||||
*headers = upstream_headers;
|
||||
if let Ok(host_val) = hyper::header::HeaderValue::from_str(
|
||||
&format!("{}:{}", upstream.host, upstream.port)
|
||||
) {
|
||||
headers.insert(hyper::header::HOST, host_val);
|
||||
}
|
||||
}
|
||||
|
||||
// Wrap the request body in CountingBody to track bytes_in
|
||||
@@ -511,7 +583,6 @@ impl HttpProxyService {
|
||||
Ok(resp) => resp,
|
||||
Err(e) => {
|
||||
error!("Upstream request failed: {}", e);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend request failed"));
|
||||
}
|
||||
};
|
||||
@@ -527,7 +598,7 @@ impl HttpProxyService {
|
||||
body: Incoming,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
_upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
route: &rustproxy_config::RouteConfig,
|
||||
route_id: Option<&str>,
|
||||
source_ip: &str,
|
||||
@@ -537,7 +608,6 @@ impl HttpProxyService {
|
||||
Ok(h) => h,
|
||||
Err(e) => {
|
||||
error!("HTTP/2 upstream handshake failed: {}", e);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 handshake failed"));
|
||||
}
|
||||
};
|
||||
@@ -554,11 +624,6 @@ impl HttpProxyService {
|
||||
|
||||
if let Some(headers) = upstream_req.headers_mut() {
|
||||
*headers = upstream_headers;
|
||||
if let Ok(host_val) = hyper::header::HeaderValue::from_str(
|
||||
&format!("{}:{}", upstream.host, upstream.port)
|
||||
) {
|
||||
headers.insert(hyper::header::HOST, host_val);
|
||||
}
|
||||
}
|
||||
|
||||
// Wrap the request body in CountingBody to track bytes_in
|
||||
@@ -577,7 +642,6 @@ impl HttpProxyService {
|
||||
Ok(resp) => resp,
|
||||
Err(e) => {
|
||||
error!("HTTP/2 upstream request failed: {}", e);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend H2 request failed"));
|
||||
}
|
||||
};
|
||||
@@ -588,8 +652,7 @@ impl HttpProxyService {
|
||||
/// Build the client-facing response from an upstream response, streaming the body.
|
||||
///
|
||||
/// The response body is wrapped in a `CountingBody` that counts bytes as they
|
||||
/// stream from upstream to client. When the body is fully consumed (or dropped),
|
||||
/// it reports byte counts to the metrics collector and calls `connection_closed`.
|
||||
/// stream from upstream to client.
|
||||
async fn build_streaming_response(
|
||||
&self,
|
||||
upstream_response: Response<Incoming>,
|
||||
@@ -618,11 +681,6 @@ impl HttpProxyService {
|
||||
Direction::Out,
|
||||
);
|
||||
|
||||
// Close the connection metric now — the HTTP request/response cycle is done
|
||||
// from the proxy's perspective once we hand the streaming body to hyper.
|
||||
// Bytes will still be counted as they flow.
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
|
||||
let body: BoxBody<Bytes, hyper::Error> = BoxBody::new(counting_body);
|
||||
|
||||
Ok(response.body(body).unwrap())
|
||||
@@ -654,7 +712,6 @@ impl HttpProxyService {
|
||||
.unwrap_or("");
|
||||
if !allowed_origins.is_empty() && !allowed_origins.iter().any(|o| o == "*" || o == origin) {
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::FORBIDDEN, "Origin not allowed"));
|
||||
}
|
||||
}
|
||||
@@ -672,13 +729,11 @@ impl HttpProxyService {
|
||||
Ok(Err(e)) => {
|
||||
error!("WebSocket: failed TLS connect upstream {}:{}: {}", upstream.host, upstream.port, e);
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend TLS unavailable"));
|
||||
}
|
||||
Err(_) => {
|
||||
error!("WebSocket: upstream TLS connect timeout for {}:{}", upstream.host, upstream.port);
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend TLS connect timeout"));
|
||||
}
|
||||
}
|
||||
@@ -694,13 +749,11 @@ impl HttpProxyService {
|
||||
Ok(Err(e)) => {
|
||||
error!("WebSocket: failed to connect upstream {}:{}: {}", upstream.host, upstream.port, e);
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend unavailable"));
|
||||
}
|
||||
Err(_) => {
|
||||
error!("WebSocket: upstream connect timeout for {}:{}", upstream.host, upstream.port);
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::GATEWAY_TIMEOUT, "Backend connect timeout"));
|
||||
}
|
||||
}
|
||||
@@ -731,13 +784,44 @@ impl HttpProxyService {
|
||||
parts.method, upstream_path
|
||||
);
|
||||
|
||||
let upstream_host = format!("{}:{}", upstream.host, upstream.port);
|
||||
// Copy all original headers (preserving the client's Host header).
|
||||
// Skip X-Forwarded-* since we set them ourselves below.
|
||||
for (name, value) in parts.headers.iter() {
|
||||
if name == hyper::header::HOST {
|
||||
raw_request.push_str(&format!("host: {}\r\n", upstream_host));
|
||||
} else {
|
||||
raw_request.push_str(&format!("{}: {}\r\n", name, value.to_str().unwrap_or("")));
|
||||
let name_str = name.as_str();
|
||||
if name_str == "x-forwarded-for"
|
||||
|| name_str == "x-forwarded-host"
|
||||
|| name_str == "x-forwarded-proto"
|
||||
{
|
||||
continue;
|
||||
}
|
||||
raw_request.push_str(&format!("{}: {}\r\n", name, value.to_str().unwrap_or("")));
|
||||
}
|
||||
|
||||
// Add standard reverse-proxy headers (X-Forwarded-*)
|
||||
{
|
||||
let original_host = parts.headers.get("host")
|
||||
.and_then(|h| h.to_str().ok())
|
||||
.unwrap_or("");
|
||||
let forwarded_proto = if route.action.tls.as_ref()
|
||||
.map(|t| matches!(t.mode,
|
||||
rustproxy_config::TlsMode::Terminate
|
||||
| rustproxy_config::TlsMode::TerminateAndReencrypt))
|
||||
.unwrap_or(false)
|
||||
{
|
||||
"https"
|
||||
} else {
|
||||
"http"
|
||||
};
|
||||
|
||||
let client_ip = peer_addr.ip().to_string();
|
||||
let xff_value = if let Some(existing) = parts.headers.get("x-forwarded-for") {
|
||||
format!("{}, {}", existing.to_str().unwrap_or(""), client_ip)
|
||||
} else {
|
||||
client_ip
|
||||
};
|
||||
raw_request.push_str(&format!("x-forwarded-for: {}\r\n", xff_value));
|
||||
raw_request.push_str(&format!("x-forwarded-host: {}\r\n", original_host));
|
||||
raw_request.push_str(&format!("x-forwarded-proto: {}\r\n", forwarded_proto));
|
||||
}
|
||||
|
||||
if let Some(ref route_headers) = route.headers {
|
||||
@@ -762,7 +846,6 @@ impl HttpProxyService {
|
||||
if let Err(e) = upstream_stream.write_all(raw_request.as_bytes()).await {
|
||||
error!("WebSocket: failed to send upgrade request to upstream: {}", e);
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend write failed"));
|
||||
}
|
||||
|
||||
@@ -773,7 +856,6 @@ impl HttpProxyService {
|
||||
Ok(0) => {
|
||||
error!("WebSocket: upstream closed before completing handshake");
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend closed"));
|
||||
}
|
||||
Ok(_) => {
|
||||
@@ -787,14 +869,12 @@ impl HttpProxyService {
|
||||
if response_buf.len() > 8192 {
|
||||
error!("WebSocket: upstream response headers too large");
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend response too large"));
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("WebSocket: failed to read upstream response: {}", e);
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(StatusCode::BAD_GATEWAY, "Backend read failed"));
|
||||
}
|
||||
}
|
||||
@@ -812,7 +892,6 @@ impl HttpProxyService {
|
||||
if status_code != 101 {
|
||||
debug!("WebSocket: upstream rejected upgrade with status {}", status_code);
|
||||
self.upstream_selector.connection_ended(upstream_key);
|
||||
self.metrics.connection_closed(route_id, Some(source_ip));
|
||||
return Ok(error_response(
|
||||
StatusCode::from_u16(status_code).unwrap_or(StatusCode::BAD_GATEWAY),
|
||||
"WebSocket upgrade rejected by backend",
|
||||
@@ -856,9 +935,6 @@ impl HttpProxyService {
|
||||
Err(e) => {
|
||||
debug!("WebSocket: client upgrade failed: {}", e);
|
||||
upstream_selector.connection_ended(&upstream_key_owned);
|
||||
if let Some(ref rid) = route_id_owned {
|
||||
metrics.connection_closed(Some(rid.as_str()), Some(&source_ip_owned));
|
||||
}
|
||||
return;
|
||||
}
|
||||
};
|
||||
@@ -963,7 +1039,6 @@ impl HttpProxyService {
|
||||
upstream_selector.connection_ended(&upstream_key_owned);
|
||||
if let Some(ref rid) = route_id_owned {
|
||||
metrics.record_bytes(bytes_in, bytes_out, Some(rid.as_str()), Some(&source_ip_owned));
|
||||
metrics.connection_closed(Some(rid.as_str()), Some(&source_ip_owned));
|
||||
}
|
||||
});
|
||||
|
||||
@@ -993,8 +1068,8 @@ impl HttpProxyService {
|
||||
response.body(BoxBody::new(body)).unwrap()
|
||||
}
|
||||
|
||||
/// Apply URL rewriting rules from route config.
|
||||
fn apply_url_rewrite(path: &str, route: &rustproxy_config::RouteConfig) -> String {
|
||||
/// Apply URL rewriting rules from route config, using the compiled regex cache.
|
||||
fn apply_url_rewrite(&self, path: &str, route: &rustproxy_config::RouteConfig) -> String {
|
||||
let rewrite = match route.action.advanced.as_ref()
|
||||
.and_then(|a| a.url_rewrite.as_ref())
|
||||
{
|
||||
@@ -1013,10 +1088,20 @@ impl HttpProxyService {
|
||||
(path.to_string(), String::new())
|
||||
};
|
||||
|
||||
// Look up or compile the regex, caching for future requests
|
||||
let cached = self.regex_cache.get(&rewrite.pattern);
|
||||
if let Some(re) = cached {
|
||||
let result = re.replace_all(&subject, rewrite.target.as_str());
|
||||
return format!("{}{}", result, suffix);
|
||||
}
|
||||
|
||||
// Not cached — compile and insert
|
||||
match Regex::new(&rewrite.pattern) {
|
||||
Ok(re) => {
|
||||
let result = re.replace_all(&subject, rewrite.target.as_str());
|
||||
format!("{}{}", result, suffix)
|
||||
let out = format!("{}{}", result, suffix);
|
||||
self.regex_cache.insert(rewrite.pattern.clone(), re);
|
||||
out
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Invalid URL rewrite pattern '{}': {}", rewrite.pattern, e);
|
||||
@@ -1143,6 +1228,9 @@ impl Default for HttpProxyService {
|
||||
metrics: Arc::new(MetricsCollector::new()),
|
||||
upstream_selector: UpstreamSelector::new(),
|
||||
connect_timeout: DEFAULT_CONNECT_TIMEOUT,
|
||||
route_rate_limiters: Arc::new(DashMap::new()),
|
||||
request_counter: AtomicU64::new(0),
|
||||
regex_cache: DashMap::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -115,10 +115,18 @@ impl UpstreamSelector {
|
||||
/// Record that a connection to the given host has ended.
|
||||
pub fn connection_ended(&self, host: &str) {
|
||||
if let Some(counter) = self.active_connections.get(host) {
|
||||
let prev = counter.value().fetch_sub(1, Ordering::Relaxed);
|
||||
// Guard against underflow (shouldn't happen, but be safe)
|
||||
let prev = counter.value().load(Ordering::Relaxed);
|
||||
if prev == 0 {
|
||||
counter.value().store(0, Ordering::Relaxed);
|
||||
// Already at zero — just clean up the entry
|
||||
drop(counter);
|
||||
self.active_connections.remove(host);
|
||||
return;
|
||||
}
|
||||
counter.value().fetch_sub(1, Ordering::Relaxed);
|
||||
// Clean up zero-count entries to prevent memory growth
|
||||
if prev <= 1 {
|
||||
drop(counter);
|
||||
self.active_connections.remove(host);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -204,6 +212,31 @@ mod tests {
|
||||
assert_eq!(r4.host, "a");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_connection_tracking_cleanup() {
|
||||
let selector = UpstreamSelector::new();
|
||||
|
||||
selector.connection_started("backend:8080");
|
||||
selector.connection_started("backend:8080");
|
||||
assert_eq!(
|
||||
selector.active_connections.get("backend:8080").unwrap().load(Ordering::Relaxed),
|
||||
2
|
||||
);
|
||||
|
||||
selector.connection_ended("backend:8080");
|
||||
assert_eq!(
|
||||
selector.active_connections.get("backend:8080").unwrap().load(Ordering::Relaxed),
|
||||
1
|
||||
);
|
||||
|
||||
// Last connection ends — entry should be removed entirely
|
||||
selector.connection_ended("backend:8080");
|
||||
assert!(selector.active_connections.get("backend:8080").is_none());
|
||||
|
||||
// Ending on a non-existent key should not panic
|
||||
selector.connection_ended("nonexistent:9999");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_ip_hash_consistent() {
|
||||
let selector = UpstreamSelector::new();
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
use dashmap::DashMap;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashSet;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::Mutex;
|
||||
|
||||
@@ -196,6 +197,12 @@ impl MetricsCollector {
|
||||
if val <= 1 {
|
||||
drop(counter);
|
||||
self.ip_connections.remove(ip);
|
||||
// Evict all per-IP tracking data for this IP
|
||||
self.ip_total_connections.remove(ip);
|
||||
self.ip_bytes_in.remove(ip);
|
||||
self.ip_bytes_out.remove(ip);
|
||||
self.ip_pending_tp.remove(ip);
|
||||
self.ip_throughput.remove(ip);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -342,6 +349,17 @@ impl MetricsCollector {
|
||||
}
|
||||
}
|
||||
|
||||
/// Remove per-route metrics for route IDs that are no longer active.
|
||||
/// Call this after `update_routes()` to prune stale entries.
|
||||
pub fn retain_routes(&self, active_route_ids: &HashSet<String>) {
|
||||
self.route_connections.retain(|k, _| active_route_ids.contains(k));
|
||||
self.route_total_connections.retain(|k, _| active_route_ids.contains(k));
|
||||
self.route_bytes_in.retain(|k, _| active_route_ids.contains(k));
|
||||
self.route_bytes_out.retain(|k, _| active_route_ids.contains(k));
|
||||
self.route_pending_tp.retain(|k, _| active_route_ids.contains(k));
|
||||
self.route_throughput.retain(|k, _| active_route_ids.contains(k));
|
||||
}
|
||||
|
||||
/// Get current active connection count.
|
||||
pub fn active_connections(&self) -> u64 {
|
||||
self.active_connections.load(Ordering::Relaxed)
|
||||
@@ -633,6 +651,42 @@ mod tests {
|
||||
assert!(collector.ip_connections.get("1.2.3.4").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_per_ip_full_eviction_on_last_close() {
|
||||
let collector = MetricsCollector::with_retention(60);
|
||||
|
||||
// Open connections from two IPs
|
||||
collector.connection_opened(Some("route-a"), Some("10.0.0.1"));
|
||||
collector.connection_opened(Some("route-a"), Some("10.0.0.1"));
|
||||
collector.connection_opened(Some("route-b"), Some("10.0.0.2"));
|
||||
|
||||
// Record bytes to populate per-IP DashMaps
|
||||
collector.record_bytes(100, 200, Some("route-a"), Some("10.0.0.1"));
|
||||
collector.record_bytes(300, 400, Some("route-b"), Some("10.0.0.2"));
|
||||
collector.sample_all();
|
||||
|
||||
// Verify per-IP data exists
|
||||
assert!(collector.ip_total_connections.get("10.0.0.1").is_some());
|
||||
assert!(collector.ip_bytes_in.get("10.0.0.1").is_some());
|
||||
assert!(collector.ip_throughput.get("10.0.0.1").is_some());
|
||||
|
||||
// Close all connections for 10.0.0.1
|
||||
collector.connection_closed(Some("route-a"), Some("10.0.0.1"));
|
||||
collector.connection_closed(Some("route-a"), Some("10.0.0.1"));
|
||||
|
||||
// All per-IP data for 10.0.0.1 should be evicted
|
||||
assert!(collector.ip_connections.get("10.0.0.1").is_none());
|
||||
assert!(collector.ip_total_connections.get("10.0.0.1").is_none());
|
||||
assert!(collector.ip_bytes_in.get("10.0.0.1").is_none());
|
||||
assert!(collector.ip_bytes_out.get("10.0.0.1").is_none());
|
||||
assert!(collector.ip_pending_tp.get("10.0.0.1").is_none());
|
||||
assert!(collector.ip_throughput.get("10.0.0.1").is_none());
|
||||
|
||||
// 10.0.0.2 should still have data
|
||||
assert!(collector.ip_connections.get("10.0.0.2").is_some());
|
||||
assert!(collector.ip_total_connections.get("10.0.0.2").is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_http_request_tracking() {
|
||||
let collector = MetricsCollector::with_retention(60);
|
||||
@@ -650,6 +704,35 @@ mod tests {
|
||||
assert_eq!(snapshot.http_requests_per_sec, 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_retain_routes_prunes_stale() {
|
||||
let collector = MetricsCollector::with_retention(60);
|
||||
|
||||
// Create metrics for 3 routes
|
||||
collector.connection_opened(Some("route-a"), None);
|
||||
collector.connection_opened(Some("route-b"), None);
|
||||
collector.connection_opened(Some("route-c"), None);
|
||||
collector.record_bytes(100, 200, Some("route-a"), None);
|
||||
collector.record_bytes(100, 200, Some("route-b"), None);
|
||||
collector.record_bytes(100, 200, Some("route-c"), None);
|
||||
collector.sample_all();
|
||||
|
||||
// Now "route-b" is removed from config
|
||||
let active = HashSet::from(["route-a".to_string(), "route-c".to_string()]);
|
||||
collector.retain_routes(&active);
|
||||
|
||||
// route-b entries should be gone
|
||||
assert!(collector.route_connections.get("route-b").is_none());
|
||||
assert!(collector.route_total_connections.get("route-b").is_none());
|
||||
assert!(collector.route_bytes_in.get("route-b").is_none());
|
||||
assert!(collector.route_bytes_out.get("route-b").is_none());
|
||||
assert!(collector.route_throughput.get("route-b").is_none());
|
||||
|
||||
// route-a and route-c should still exist
|
||||
assert!(collector.route_total_connections.get("route-a").is_some());
|
||||
assert!(collector.route_total_connections.get("route-c").is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_throughput_history_in_snapshot() {
|
||||
let collector = MetricsCollector::with_retention(60);
|
||||
|
||||
@@ -95,10 +95,11 @@ impl ConnectionTracker {
|
||||
pub fn connection_closed(&self, ip: &IpAddr) {
|
||||
if let Some(counter) = self.active.get(ip) {
|
||||
let prev = counter.value().fetch_sub(1, Ordering::Relaxed);
|
||||
// Clean up zero entries
|
||||
// Clean up zero entries to prevent memory growth
|
||||
if prev <= 1 {
|
||||
drop(counter);
|
||||
self.active.remove(ip);
|
||||
self.timestamps.remove(ip);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -205,10 +206,13 @@ impl ConnectionTracker {
|
||||
let zombies = tracker.scan_zombies();
|
||||
if !zombies.is_empty() {
|
||||
warn!(
|
||||
"Detected {} zombie connection(s): {:?}",
|
||||
"Cleaning up {} zombie connection(s): {:?}",
|
||||
zombies.len(),
|
||||
zombies
|
||||
);
|
||||
for id in &zombies {
|
||||
tracker.unregister_connection(*id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -304,6 +308,30 @@ mod tests {
|
||||
assert_eq!(tracker.tracked_ips(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_timestamps_cleaned_on_last_close() {
|
||||
let tracker = ConnectionTracker::new(None, Some(100));
|
||||
let ip: IpAddr = "10.0.0.1".parse().unwrap();
|
||||
|
||||
// try_accept populates the timestamps map (when rate limiting is enabled)
|
||||
assert!(tracker.try_accept(&ip));
|
||||
tracker.connection_opened(&ip);
|
||||
assert!(tracker.try_accept(&ip));
|
||||
tracker.connection_opened(&ip);
|
||||
|
||||
// Timestamps should exist
|
||||
assert!(tracker.timestamps.get(&ip).is_some());
|
||||
|
||||
// Close one connection — timestamps should still exist
|
||||
tracker.connection_closed(&ip);
|
||||
assert!(tracker.timestamps.get(&ip).is_some());
|
||||
|
||||
// Close last connection — timestamps should be cleaned up
|
||||
tracker.connection_closed(&ip);
|
||||
assert!(tracker.timestamps.get(&ip).is_none());
|
||||
assert!(tracker.active.get(&ip).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_register_unregister_connection() {
|
||||
let tracker = ConnectionTracker::new(None, None);
|
||||
|
||||
@@ -22,7 +22,6 @@ struct ConnectionGuard {
|
||||
metrics: Arc<MetricsCollector>,
|
||||
route_id: Option<String>,
|
||||
source_ip: Option<String>,
|
||||
disarmed: bool,
|
||||
}
|
||||
|
||||
impl ConnectionGuard {
|
||||
@@ -31,22 +30,13 @@ impl ConnectionGuard {
|
||||
metrics,
|
||||
route_id: route_id.map(|s| s.to_string()),
|
||||
source_ip: source_ip.map(|s| s.to_string()),
|
||||
disarmed: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Disarm the guard — prevents the Drop from running.
|
||||
/// Use when handing off to a path that manages its own cleanup (e.g., HTTP proxy).
|
||||
fn disarm(mut self) {
|
||||
self.disarmed = true;
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ConnectionGuard {
|
||||
fn drop(&mut self) {
|
||||
if !self.disarmed {
|
||||
self.metrics.connection_closed(self.route_id.as_deref(), self.source_ip.as_deref());
|
||||
}
|
||||
self.metrics.connection_closed(self.route_id.as_deref(), self.source_ip.as_deref());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -844,8 +834,6 @@ impl TcpListenerManager {
|
||||
"TLS Terminate + HTTP: {} -> {}:{} (domain: {:?})",
|
||||
peer_addr, target_host, target_port, domain
|
||||
);
|
||||
// HTTP proxy manages its own per-request metrics — disarm TCP-level guard
|
||||
_conn_guard.disarm();
|
||||
http_proxy.handle_io(buf_stream, peer_addr, port, cancel.clone()).await;
|
||||
} else {
|
||||
debug!(
|
||||
@@ -917,7 +905,6 @@ impl TcpListenerManager {
|
||||
"TLS Terminate+Reencrypt + HTTP: {} (domain: {:?})",
|
||||
peer_addr, domain
|
||||
);
|
||||
_conn_guard.disarm();
|
||||
http_proxy.handle_io(buf_stream, peer_addr, port, cancel.clone()).await;
|
||||
} else {
|
||||
// Non-HTTP: TLS-to-TLS tunnel (existing behavior for raw TCP protocols)
|
||||
@@ -937,8 +924,6 @@ impl TcpListenerManager {
|
||||
if is_http {
|
||||
// Plain HTTP - use HTTP proxy for request-level routing
|
||||
debug!("HTTP proxy: {} on port {}", peer_addr, port);
|
||||
// HTTP proxy manages its own per-request metrics — disarm TCP-level guard
|
||||
_conn_guard.disarm();
|
||||
http_proxy.handle_connection(stream, peer_addr, port, cancel.clone()).await;
|
||||
Ok(())
|
||||
} else {
|
||||
|
||||
@@ -27,7 +27,7 @@
|
||||
pub mod challenge_server;
|
||||
pub mod management;
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
@@ -565,6 +565,12 @@ impl RustProxy {
|
||||
vec![]
|
||||
};
|
||||
|
||||
// Prune per-route metrics for route IDs that no longer exist
|
||||
let active_route_ids: HashSet<String> = routes.iter()
|
||||
.filter_map(|r| r.id.clone())
|
||||
.collect();
|
||||
self.metrics.retain_routes(&active_route_ids);
|
||||
|
||||
// Atomically swap the route table
|
||||
let new_manager = Arc::new(new_manager);
|
||||
self.route_table.store(Arc::clone(&new_manager));
|
||||
|
||||
@@ -185,6 +185,76 @@ pub async fn wait_for_port(port: u16, timeout_ms: u64) -> bool {
|
||||
false
|
||||
}
|
||||
|
||||
/// Start a TLS HTTP echo backend: accepts TLS, then responds with HTTP JSON
|
||||
/// containing request details. Combines TLS acceptance with HTTP echo behavior.
|
||||
pub async fn start_tls_http_backend(
|
||||
port: u16,
|
||||
backend_name: &str,
|
||||
cert_pem: &str,
|
||||
key_pem: &str,
|
||||
) -> JoinHandle<()> {
|
||||
use std::sync::Arc;
|
||||
|
||||
let acceptor = rustproxy_passthrough::build_tls_acceptor(cert_pem, key_pem)
|
||||
.expect("Failed to build TLS acceptor");
|
||||
let acceptor = Arc::new(acceptor);
|
||||
let name = backend_name.to_string();
|
||||
|
||||
let listener = TcpListener::bind(format!("127.0.0.1:{}", port))
|
||||
.await
|
||||
.unwrap_or_else(|_| panic!("Failed to bind TLS HTTP backend on port {}", port));
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let (stream, _) = match listener.accept().await {
|
||||
Ok(conn) => conn,
|
||||
Err(_) => break,
|
||||
};
|
||||
let acc = acceptor.clone();
|
||||
let backend = name.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut tls_stream = match acc.accept(stream).await {
|
||||
Ok(s) => s,
|
||||
Err(_) => return,
|
||||
};
|
||||
|
||||
let mut buf = vec![0u8; 16384];
|
||||
let n = match tls_stream.read(&mut buf).await {
|
||||
Ok(0) | Err(_) => return,
|
||||
Ok(n) => n,
|
||||
};
|
||||
let req_str = String::from_utf8_lossy(&buf[..n]);
|
||||
|
||||
// Parse first line: METHOD PATH HTTP/x.x
|
||||
let first_line = req_str.lines().next().unwrap_or("");
|
||||
let parts: Vec<&str> = first_line.split_whitespace().collect();
|
||||
let method = parts.first().copied().unwrap_or("UNKNOWN");
|
||||
let path = parts.get(1).copied().unwrap_or("/");
|
||||
|
||||
// Extract Host header
|
||||
let host = req_str
|
||||
.lines()
|
||||
.find(|l| l.to_lowercase().starts_with("host:"))
|
||||
.map(|l| l[5..].trim())
|
||||
.unwrap_or("unknown");
|
||||
|
||||
let body = format!(
|
||||
r#"{{"method":"{}","path":"{}","host":"{}","backend":"{}"}}"#,
|
||||
method, path, host, backend
|
||||
);
|
||||
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
body.len(),
|
||||
body,
|
||||
);
|
||||
let _ = tls_stream.write_all(response.as_bytes()).await;
|
||||
let _ = tls_stream.shutdown().await;
|
||||
});
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Helper to create a minimal route config for testing.
|
||||
pub fn make_test_route(
|
||||
port: u16,
|
||||
@@ -382,6 +452,86 @@ pub fn make_tls_terminate_route(
|
||||
route
|
||||
}
|
||||
|
||||
/// Start a TLS WebSocket echo backend: accepts TLS, performs WS handshake, then echoes data.
|
||||
/// Combines TLS acceptance (like `start_tls_http_backend`) with WebSocket echo (like `start_ws_echo_backend`).
|
||||
pub async fn start_tls_ws_echo_backend(
|
||||
port: u16,
|
||||
cert_pem: &str,
|
||||
key_pem: &str,
|
||||
) -> JoinHandle<()> {
|
||||
use std::sync::Arc;
|
||||
|
||||
let acceptor = rustproxy_passthrough::build_tls_acceptor(cert_pem, key_pem)
|
||||
.expect("Failed to build TLS acceptor");
|
||||
let acceptor = Arc::new(acceptor);
|
||||
|
||||
let listener = TcpListener::bind(format!("127.0.0.1:{}", port))
|
||||
.await
|
||||
.unwrap_or_else(|_| panic!("Failed to bind TLS WS echo backend on port {}", port));
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
let (stream, _) = match listener.accept().await {
|
||||
Ok(conn) => conn,
|
||||
Err(_) => break,
|
||||
};
|
||||
let acc = acceptor.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut tls_stream = match acc.accept(stream).await {
|
||||
Ok(s) => s,
|
||||
Err(_) => return,
|
||||
};
|
||||
|
||||
// Read the HTTP upgrade request
|
||||
let mut buf = vec![0u8; 4096];
|
||||
let n = match tls_stream.read(&mut buf).await {
|
||||
Ok(0) | Err(_) => return,
|
||||
Ok(n) => n,
|
||||
};
|
||||
|
||||
let req_str = String::from_utf8_lossy(&buf[..n]);
|
||||
|
||||
// Extract Sec-WebSocket-Key for handshake
|
||||
let ws_key = req_str
|
||||
.lines()
|
||||
.find(|l| l.to_lowercase().starts_with("sec-websocket-key:"))
|
||||
.map(|l| l.split(':').nth(1).unwrap_or("").trim().to_string())
|
||||
.unwrap_or_default();
|
||||
|
||||
// Send 101 Switching Protocols
|
||||
let accept_response = format!(
|
||||
"HTTP/1.1 101 Switching Protocols\r\n\
|
||||
Upgrade: websocket\r\n\
|
||||
Connection: Upgrade\r\n\
|
||||
Sec-WebSocket-Accept: {}\r\n\
|
||||
\r\n",
|
||||
ws_key
|
||||
);
|
||||
|
||||
if tls_stream
|
||||
.write_all(accept_response.as_bytes())
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
// Echo all data back (raw TCP after upgrade)
|
||||
let mut echo_buf = vec![0u8; 65536];
|
||||
loop {
|
||||
let n = match tls_stream.read(&mut echo_buf).await {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(n) => n,
|
||||
};
|
||||
if tls_stream.write_all(&echo_buf[..n]).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Helper to create a TLS passthrough route for testing.
|
||||
pub fn make_tls_passthrough_route(
|
||||
port: u16,
|
||||
|
||||
@@ -407,6 +407,305 @@ async fn test_websocket_through_proxy() {
|
||||
proxy.stop().await.unwrap();
|
||||
}
|
||||
|
||||
/// Test that terminate-and-reencrypt mode routes HTTP traffic through the
|
||||
/// full HTTP proxy with per-request Host-based routing.
|
||||
///
|
||||
/// This verifies the new behavior: after TLS termination, HTTP data is detected
|
||||
/// and routed through HttpProxyService (like nginx) instead of being blindly tunneled.
|
||||
#[tokio::test]
|
||||
async fn test_terminate_and_reencrypt_http_routing() {
|
||||
let backend1_port = next_port();
|
||||
let backend2_port = next_port();
|
||||
let proxy_port = next_port();
|
||||
|
||||
let (cert1, key1) = generate_self_signed_cert("alpha.example.com");
|
||||
let (cert2, key2) = generate_self_signed_cert("beta.example.com");
|
||||
|
||||
// Generate separate backend certs (backends are independent TLS servers)
|
||||
let (backend_cert1, backend_key1) = generate_self_signed_cert("localhost");
|
||||
let (backend_cert2, backend_key2) = generate_self_signed_cert("localhost");
|
||||
|
||||
// Start TLS HTTP echo backends (proxy re-encrypts to these)
|
||||
let _b1 = start_tls_http_backend(backend1_port, "alpha", &backend_cert1, &backend_key1).await;
|
||||
let _b2 = start_tls_http_backend(backend2_port, "beta", &backend_cert2, &backend_key2).await;
|
||||
|
||||
// Create terminate-and-reencrypt routes
|
||||
let mut route1 = make_tls_terminate_route(
|
||||
proxy_port, "alpha.example.com", "127.0.0.1", backend1_port, &cert1, &key1,
|
||||
);
|
||||
route1.action.tls.as_mut().unwrap().mode = rustproxy_config::TlsMode::TerminateAndReencrypt;
|
||||
|
||||
let mut route2 = make_tls_terminate_route(
|
||||
proxy_port, "beta.example.com", "127.0.0.1", backend2_port, &cert2, &key2,
|
||||
);
|
||||
route2.action.tls.as_mut().unwrap().mode = rustproxy_config::TlsMode::TerminateAndReencrypt;
|
||||
|
||||
let options = RustProxyOptions {
|
||||
routes: vec![route1, route2],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut proxy = RustProxy::new(options).unwrap();
|
||||
proxy.start().await.unwrap();
|
||||
assert!(wait_for_port(proxy_port, 2000).await);
|
||||
|
||||
// Test alpha domain - HTTP request through TLS terminate-and-reencrypt
|
||||
let alpha_result = with_timeout(async {
|
||||
let _ = rustls::crypto::ring::default_provider().install_default();
|
||||
let tls_config = rustls::ClientConfig::builder()
|
||||
.dangerous()
|
||||
.with_custom_certificate_verifier(std::sync::Arc::new(InsecureVerifier))
|
||||
.with_no_client_auth();
|
||||
let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(tls_config));
|
||||
|
||||
let stream = tokio::net::TcpStream::connect(format!("127.0.0.1:{}", proxy_port))
|
||||
.await
|
||||
.unwrap();
|
||||
let server_name = rustls::pki_types::ServerName::try_from("alpha.example.com".to_string()).unwrap();
|
||||
let mut tls_stream = connector.connect(server_name, stream).await.unwrap();
|
||||
|
||||
let request = "GET /api/data HTTP/1.1\r\nHost: alpha.example.com\r\nConnection: close\r\n\r\n";
|
||||
tls_stream.write_all(request.as_bytes()).await.unwrap();
|
||||
|
||||
let mut response = Vec::new();
|
||||
tls_stream.read_to_end(&mut response).await.unwrap();
|
||||
String::from_utf8_lossy(&response).to_string()
|
||||
}, 10)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let alpha_body = extract_body(&alpha_result);
|
||||
assert!(
|
||||
alpha_body.contains(r#""backend":"alpha"#),
|
||||
"Expected alpha backend, got: {}",
|
||||
alpha_body
|
||||
);
|
||||
assert!(
|
||||
alpha_body.contains(r#""method":"GET"#),
|
||||
"Expected GET method, got: {}",
|
||||
alpha_body
|
||||
);
|
||||
assert!(
|
||||
alpha_body.contains(r#""path":"/api/data"#),
|
||||
"Expected /api/data path, got: {}",
|
||||
alpha_body
|
||||
);
|
||||
// Verify original Host header is preserved (not replaced with backend IP:port)
|
||||
assert!(
|
||||
alpha_body.contains(r#""host":"alpha.example.com"#),
|
||||
"Expected original Host header alpha.example.com, got: {}",
|
||||
alpha_body
|
||||
);
|
||||
|
||||
// Test beta domain - different host goes to different backend
|
||||
let beta_result = with_timeout(async {
|
||||
let _ = rustls::crypto::ring::default_provider().install_default();
|
||||
let tls_config = rustls::ClientConfig::builder()
|
||||
.dangerous()
|
||||
.with_custom_certificate_verifier(std::sync::Arc::new(InsecureVerifier))
|
||||
.with_no_client_auth();
|
||||
let connector = tokio_rustls::TlsConnector::from(std::sync::Arc::new(tls_config));
|
||||
|
||||
let stream = tokio::net::TcpStream::connect(format!("127.0.0.1:{}", proxy_port))
|
||||
.await
|
||||
.unwrap();
|
||||
let server_name = rustls::pki_types::ServerName::try_from("beta.example.com".to_string()).unwrap();
|
||||
let mut tls_stream = connector.connect(server_name, stream).await.unwrap();
|
||||
|
||||
let request = "GET /other HTTP/1.1\r\nHost: beta.example.com\r\nConnection: close\r\n\r\n";
|
||||
tls_stream.write_all(request.as_bytes()).await.unwrap();
|
||||
|
||||
let mut response = Vec::new();
|
||||
tls_stream.read_to_end(&mut response).await.unwrap();
|
||||
String::from_utf8_lossy(&response).to_string()
|
||||
}, 10)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let beta_body = extract_body(&beta_result);
|
||||
assert!(
|
||||
beta_body.contains(r#""backend":"beta"#),
|
||||
"Expected beta backend, got: {}",
|
||||
beta_body
|
||||
);
|
||||
assert!(
|
||||
beta_body.contains(r#""path":"/other"#),
|
||||
"Expected /other path, got: {}",
|
||||
beta_body
|
||||
);
|
||||
// Verify original Host header is preserved for beta too
|
||||
assert!(
|
||||
beta_body.contains(r#""host":"beta.example.com"#),
|
||||
"Expected original Host header beta.example.com, got: {}",
|
||||
beta_body
|
||||
);
|
||||
|
||||
proxy.stop().await.unwrap();
|
||||
}
|
||||
|
||||
/// Test that WebSocket upgrade works through terminate-and-reencrypt mode.
|
||||
///
|
||||
/// Verifies the full chain: client→TLS→proxy terminates→re-encrypts→TLS→backend WebSocket.
|
||||
/// The proxy's `handle_websocket_upgrade` checks `upstream.use_tls` and calls
|
||||
/// `connect_tls_backend()` when true. This test covers that path.
|
||||
#[tokio::test]
|
||||
async fn test_terminate_and_reencrypt_websocket() {
|
||||
let backend_port = next_port();
|
||||
let proxy_port = next_port();
|
||||
let domain = "ws.example.com";
|
||||
|
||||
// Frontend cert (client→proxy TLS)
|
||||
let (frontend_cert, frontend_key) = generate_self_signed_cert(domain);
|
||||
// Backend cert (proxy→backend TLS)
|
||||
let (backend_cert, backend_key) = generate_self_signed_cert("localhost");
|
||||
|
||||
// Start TLS WebSocket echo backend
|
||||
let _backend = start_tls_ws_echo_backend(backend_port, &backend_cert, &backend_key).await;
|
||||
|
||||
// Create terminate-and-reencrypt route
|
||||
let mut route = make_tls_terminate_route(
|
||||
proxy_port,
|
||||
domain,
|
||||
"127.0.0.1",
|
||||
backend_port,
|
||||
&frontend_cert,
|
||||
&frontend_key,
|
||||
);
|
||||
route.action.tls.as_mut().unwrap().mode = rustproxy_config::TlsMode::TerminateAndReencrypt;
|
||||
|
||||
let options = RustProxyOptions {
|
||||
routes: vec![route],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut proxy = RustProxy::new(options).unwrap();
|
||||
proxy.start().await.unwrap();
|
||||
assert!(wait_for_port(proxy_port, 2000).await);
|
||||
|
||||
let result = with_timeout(
|
||||
async {
|
||||
let _ = rustls::crypto::ring::default_provider().install_default();
|
||||
let tls_config = rustls::ClientConfig::builder()
|
||||
.dangerous()
|
||||
.with_custom_certificate_verifier(std::sync::Arc::new(InsecureVerifier))
|
||||
.with_no_client_auth();
|
||||
let connector =
|
||||
tokio_rustls::TlsConnector::from(std::sync::Arc::new(tls_config));
|
||||
|
||||
let stream = tokio::net::TcpStream::connect(format!("127.0.0.1:{}", proxy_port))
|
||||
.await
|
||||
.unwrap();
|
||||
let server_name =
|
||||
rustls::pki_types::ServerName::try_from(domain.to_string()).unwrap();
|
||||
let mut tls_stream = connector.connect(server_name, stream).await.unwrap();
|
||||
|
||||
// Send WebSocket upgrade request through TLS
|
||||
let request = format!(
|
||||
"GET /ws HTTP/1.1\r\n\
|
||||
Host: {}\r\n\
|
||||
Upgrade: websocket\r\n\
|
||||
Connection: Upgrade\r\n\
|
||||
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
|
||||
Sec-WebSocket-Version: 13\r\n\
|
||||
\r\n",
|
||||
domain
|
||||
);
|
||||
tls_stream.write_all(request.as_bytes()).await.unwrap();
|
||||
|
||||
// Read the 101 response (byte-by-byte until \r\n\r\n)
|
||||
let mut response_buf = Vec::with_capacity(4096);
|
||||
let mut temp = [0u8; 1];
|
||||
loop {
|
||||
let n = tls_stream.read(&mut temp).await.unwrap();
|
||||
if n == 0 {
|
||||
break;
|
||||
}
|
||||
response_buf.push(temp[0]);
|
||||
if response_buf.len() >= 4 {
|
||||
let len = response_buf.len();
|
||||
if response_buf[len - 4..] == *b"\r\n\r\n" {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let response_str = String::from_utf8_lossy(&response_buf).to_string();
|
||||
assert!(
|
||||
response_str.contains("101"),
|
||||
"Expected 101 Switching Protocols, got: {}",
|
||||
response_str
|
||||
);
|
||||
assert!(
|
||||
response_str.to_lowercase().contains("upgrade: websocket"),
|
||||
"Expected Upgrade header, got: {}",
|
||||
response_str
|
||||
);
|
||||
|
||||
// After upgrade, send data and verify echo
|
||||
let test_data = b"Hello TLS WebSocket!";
|
||||
tls_stream.write_all(test_data).await.unwrap();
|
||||
|
||||
// Read echoed data
|
||||
let mut echo_buf = vec![0u8; 256];
|
||||
let n = tls_stream.read(&mut echo_buf).await.unwrap();
|
||||
let echoed = &echo_buf[..n];
|
||||
|
||||
assert_eq!(echoed, test_data, "Expected echo of sent data");
|
||||
|
||||
"ok".to_string()
|
||||
},
|
||||
10,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(result, "ok");
|
||||
proxy.stop().await.unwrap();
|
||||
}
|
||||
|
||||
/// Test that the protocol field on route config is accepted and processed.
|
||||
#[tokio::test]
|
||||
async fn test_protocol_field_in_route_config() {
|
||||
let backend_port = next_port();
|
||||
let proxy_port = next_port();
|
||||
|
||||
let _backend = start_http_echo_backend(backend_port, "main").await;
|
||||
|
||||
// Create a route with protocol: "http" - should only match HTTP traffic
|
||||
let mut route = make_test_route(proxy_port, None, "127.0.0.1", backend_port);
|
||||
route.route_match.protocol = Some("http".to_string());
|
||||
|
||||
let options = RustProxyOptions {
|
||||
routes: vec![route],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut proxy = RustProxy::new(options).unwrap();
|
||||
proxy.start().await.unwrap();
|
||||
assert!(wait_for_port(proxy_port, 2000).await);
|
||||
|
||||
// HTTP request should match the route and get proxied
|
||||
let result = with_timeout(async {
|
||||
let response = send_http_request(proxy_port, "example.com", "GET", "/test").await;
|
||||
extract_body(&response).to_string()
|
||||
}, 10)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(
|
||||
result.contains(r#""backend":"main"#),
|
||||
"Expected main backend, got: {}",
|
||||
result
|
||||
);
|
||||
assert!(
|
||||
result.contains(r#""path":"/test"#),
|
||||
"Expected /test path, got: {}",
|
||||
result
|
||||
);
|
||||
|
||||
proxy.stop().await.unwrap();
|
||||
}
|
||||
|
||||
/// InsecureVerifier for test TLS client connections.
|
||||
#[derive(Debug)]
|
||||
struct InsecureVerifier;
|
||||
|
||||
@@ -562,4 +562,168 @@ tap.test('Route Integration - Combining Multiple Route Types', async () => {
|
||||
}
|
||||
});
|
||||
|
||||
// --------------------------------- Protocol Match Field Tests ---------------------------------
|
||||
|
||||
tap.test('Routes: Should accept protocol field on route match', async () => {
|
||||
// Create a route with protocol: 'http'
|
||||
const httpOnlyRoute: IRouteConfig = {
|
||||
match: {
|
||||
ports: 443,
|
||||
domains: 'api.example.com',
|
||||
protocol: 'http',
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'backend', port: 8080 }],
|
||||
tls: {
|
||||
mode: 'terminate',
|
||||
certificate: 'auto',
|
||||
},
|
||||
},
|
||||
name: 'HTTP-only Route',
|
||||
};
|
||||
|
||||
// Validate the route - protocol field should not cause errors
|
||||
const validation = validateRouteConfig(httpOnlyRoute);
|
||||
expect(validation.valid).toBeTrue();
|
||||
|
||||
// Verify the protocol field is preserved
|
||||
expect(httpOnlyRoute.match.protocol).toEqual('http');
|
||||
});
|
||||
|
||||
tap.test('Routes: Should accept protocol tcp on route match', async () => {
|
||||
// Create a route with protocol: 'tcp'
|
||||
const tcpOnlyRoute: IRouteConfig = {
|
||||
match: {
|
||||
ports: 443,
|
||||
domains: 'db.example.com',
|
||||
protocol: 'tcp',
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'db-server', port: 5432 }],
|
||||
tls: {
|
||||
mode: 'passthrough',
|
||||
},
|
||||
},
|
||||
name: 'TCP-only Route',
|
||||
};
|
||||
|
||||
const validation = validateRouteConfig(tcpOnlyRoute);
|
||||
expect(validation.valid).toBeTrue();
|
||||
|
||||
expect(tcpOnlyRoute.match.protocol).toEqual('tcp');
|
||||
});
|
||||
|
||||
tap.test('Routes: Protocol field should work with terminate-and-reencrypt', async () => {
|
||||
// Create a terminate-and-reencrypt route that only accepts HTTP
|
||||
const reencryptRoute = createHttpsTerminateRoute(
|
||||
'secure.example.com',
|
||||
{ host: 'backend', port: 443 },
|
||||
{ reencrypt: true, certificate: 'auto', name: 'Reencrypt HTTP Route' }
|
||||
);
|
||||
|
||||
// Set protocol restriction to http
|
||||
reencryptRoute.match.protocol = 'http';
|
||||
|
||||
// Validate the route
|
||||
const validation = validateRouteConfig(reencryptRoute);
|
||||
expect(validation.valid).toBeTrue();
|
||||
|
||||
// Verify TLS mode
|
||||
expect(reencryptRoute.action.tls?.mode).toEqual('terminate-and-reencrypt');
|
||||
// Verify protocol field is preserved
|
||||
expect(reencryptRoute.match.protocol).toEqual('http');
|
||||
});
|
||||
|
||||
tap.test('Routes: Protocol field should not affect domain/port matching', async () => {
|
||||
// Routes with and without protocol field should both match the same domain/port
|
||||
const routeWithProtocol: IRouteConfig = {
|
||||
match: {
|
||||
ports: 443,
|
||||
domains: 'example.com',
|
||||
protocol: 'http',
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'backend', port: 8080 }],
|
||||
tls: { mode: 'terminate', certificate: 'auto' },
|
||||
},
|
||||
name: 'With Protocol',
|
||||
priority: 10,
|
||||
};
|
||||
|
||||
const routeWithoutProtocol: IRouteConfig = {
|
||||
match: {
|
||||
ports: 443,
|
||||
domains: 'example.com',
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'fallback', port: 8081 }],
|
||||
tls: { mode: 'terminate', certificate: 'auto' },
|
||||
},
|
||||
name: 'Without Protocol',
|
||||
priority: 5,
|
||||
};
|
||||
|
||||
const routes = [routeWithProtocol, routeWithoutProtocol];
|
||||
|
||||
// Both routes should match the domain/port (protocol is a hint for Rust-side matching)
|
||||
const matches = findMatchingRoutes(routes, { domain: 'example.com', port: 443 });
|
||||
expect(matches.length).toEqual(2);
|
||||
|
||||
// The one with higher priority should be first
|
||||
const best = findBestMatchingRoute(routes, { domain: 'example.com', port: 443 });
|
||||
expect(best).not.toBeUndefined();
|
||||
expect(best!.name).toEqual('With Protocol');
|
||||
});
|
||||
|
||||
tap.test('Routes: Protocol field preserved through route cloning', async () => {
|
||||
const original: IRouteConfig = {
|
||||
match: {
|
||||
ports: 8443,
|
||||
domains: 'clone-test.example.com',
|
||||
protocol: 'http',
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'backend', port: 3000 }],
|
||||
tls: { mode: 'terminate-and-reencrypt', certificate: 'auto' },
|
||||
},
|
||||
name: 'Clone Test',
|
||||
};
|
||||
|
||||
const cloned = cloneRoute(original);
|
||||
|
||||
// Verify protocol is preserved in clone
|
||||
expect(cloned.match.protocol).toEqual('http');
|
||||
expect(cloned.action.tls?.mode).toEqual('terminate-and-reencrypt');
|
||||
|
||||
// Modify clone should not affect original
|
||||
cloned.match.protocol = 'tcp';
|
||||
expect(original.match.protocol).toEqual('http');
|
||||
});
|
||||
|
||||
tap.test('Routes: Protocol field preserved through route merging', async () => {
|
||||
const base: IRouteConfig = {
|
||||
match: {
|
||||
ports: 443,
|
||||
domains: 'merge-test.example.com',
|
||||
protocol: 'http',
|
||||
},
|
||||
action: {
|
||||
type: 'forward',
|
||||
targets: [{ host: 'backend', port: 3000 }],
|
||||
tls: { mode: 'terminate-and-reencrypt', certificate: 'auto' },
|
||||
},
|
||||
name: 'Merge Base',
|
||||
};
|
||||
|
||||
// Merge with override that changes name but not protocol
|
||||
const merged = mergeRouteConfigs(base, { name: 'Merged Route' });
|
||||
expect(merged.match.protocol).toEqual('http');
|
||||
expect(merged.name).toEqual('Merged Route');
|
||||
});
|
||||
|
||||
export default tap.start();
|
||||
@@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@push.rocks/smartproxy',
|
||||
version: '25.6.0',
|
||||
version: '25.7.5',
|
||||
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.'
|
||||
}
|
||||
|
||||
@@ -409,6 +409,7 @@ export class SmartProxy extends plugins.EventEmitter {
|
||||
keepAliveTreatment: this.settings.keepAliveTreatment,
|
||||
keepAliveInactivityMultiplier: this.settings.keepAliveInactivityMultiplier,
|
||||
extendedKeepAliveLifetime: this.settings.extendedKeepAliveLifetime,
|
||||
proxyIps: this.settings.proxyIPs,
|
||||
acceptProxyProtocol: this.settings.acceptProxyProtocol,
|
||||
sendProxyProtocol: this.settings.sendProxyProtocol,
|
||||
metrics: this.settings.metrics,
|
||||
|
||||
Reference in New Issue
Block a user