Compare commits
18 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 33fdf42a70 | |||
| fb1c59ac9a | |||
| ea8224c400 | |||
| da1cc58a3d | |||
| 606c620849 | |||
| 4ae09ac6ae | |||
| 2fce910795 | |||
| ff09cef350 | |||
| d0148b2ac3 | |||
| 7217e15649 | |||
| bfcf92a855 | |||
| 8e0804cd20 | |||
| c63f6fcd5f | |||
| f3cd4d193e | |||
| 81de611255 | |||
| 91598b3be9 | |||
| 4e3c548012 | |||
| 1a2d7529db |
58
changelog.md
58
changelog.md
@@ -1,5 +1,63 @@
|
||||
# Changelog
|
||||
|
||||
## 2026-03-20 - 25.17.10 - fix(rustproxy-http)
|
||||
reuse the shared HTTP proxy service for HTTP/3 request handling
|
||||
|
||||
- Refactors H3ProxyService to delegate requests to the shared HttpProxyService instead of maintaining separate routing and backend forwarding logic.
|
||||
- Aligns HTTP/3 with the TCP/HTTP path for route matching, connection pooling, and ALPN-based upstream protocol detection.
|
||||
- Generalizes request handling and filters to accept boxed/generic HTTP bodies so both HTTP/3 and existing HTTP paths share the same proxy pipeline.
|
||||
- Updates the HTTP/3 integration route matcher to allow transport matching across shared HTTP and QUIC handling.
|
||||
|
||||
## 2026-03-20 - 25.17.9 - fix(rustproxy-http)
|
||||
correct HTTP/3 host extraction and avoid protocol filtering during UDP route lookup
|
||||
|
||||
- Use the URI host or strip the port from the Host header so HTTP/3 requests match routes consistently with TCP/HTTP handling.
|
||||
- Remove protocol filtering from HTTP/3 route lookup because QUIC transport already constrains routing to UDP and protocol validation happens earlier.
|
||||
|
||||
## 2026-03-20 - 25.17.8 - fix(rustproxy)
|
||||
use SNI-based certificate resolution for QUIC TLS connections
|
||||
|
||||
- Replaces static first-certificate selection with the shared CertResolver used by the TCP/TLS path.
|
||||
- Ensures QUIC connections can present the correct certificate per requested domain.
|
||||
- Keeps HTTP/3 ALPN configuration while improving multi-domain TLS handling.
|
||||
|
||||
## 2026-03-20 - 25.17.7 - fix(readme)
|
||||
document QUIC and HTTP/3 compatibility caveats
|
||||
|
||||
- Add notes explaining that GREASE frames are disabled on both server and client HTTP/3 paths to avoid interoperability issues
|
||||
- Document that the current HTTP/3 stack depends on pre-1.0 h3 ecosystem components and may still have rough edges
|
||||
|
||||
## 2026-03-20 - 25.17.6 - fix(rustproxy-http)
|
||||
disable HTTP/3 GREASE for client and server connections
|
||||
|
||||
- Switch the HTTP/3 server connection setup to use the builder API with send_grease(false)
|
||||
- Switch the HTTP/3 client handshake to use the builder API with send_grease(false) to improve compatibility
|
||||
|
||||
## 2026-03-20 - 25.17.5 - fix(rustproxy)
|
||||
add HTTP/3 integration test for QUIC response stream FIN handling
|
||||
|
||||
- adds an integration test covering HTTP/3 proxying over QUIC with TLS termination
|
||||
- verifies response bodies fully arrive and the client receives stream termination instead of hanging
|
||||
- adds test-only dependencies for quinn, h3, h3-quinn, rustls, bytes, and http
|
||||
|
||||
## 2026-03-20 - 25.17.4 - fix(rustproxy-http)
|
||||
prevent HTTP/3 response body streaming from hanging on backend completion
|
||||
|
||||
- extract and track Content-Length before consuming the response body
|
||||
- stop the HTTP/3 body loop when the stream reports end-of-stream or the expected byte count has been sent
|
||||
- add a per-frame idle timeout to avoid indefinite waits on stalled or close-delimited backend bodies
|
||||
|
||||
## 2026-03-20 - 25.17.3 - fix(repository)
|
||||
no changes detected
|
||||
|
||||
|
||||
## 2026-03-20 - 25.17.2 - fix(rustproxy-http)
|
||||
enable TLS connections for HTTP/3 upstream requests when backend re-encryption or TLS is configured
|
||||
|
||||
- Pass backend TLS client configuration into the HTTP/3 request handler.
|
||||
- Detect TLS-required upstream targets using route and target TLS settings before connecting.
|
||||
- Build backend request URIs with the correct http or https scheme to match the upstream connection.
|
||||
|
||||
## 2026-03-20 - 25.17.1 - fix(rustproxy-routing)
|
||||
allow QUIC UDP TLS connections without SNI to match domain-restricted routes
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@push.rocks/smartproxy",
|
||||
"version": "25.17.1",
|
||||
"version": "25.17.10",
|
||||
"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",
|
||||
|
||||
@@ -1111,6 +1111,10 @@ SmartProxy searches for the Rust binary in this order:
|
||||
5. Local dev build (`./rust/target/release/rustproxy`)
|
||||
6. System PATH (`rustproxy`)
|
||||
|
||||
### QUIC / HTTP3 Caveats
|
||||
- **GREASE frames are disabled.** The underlying h3 crate sends [GREASE frames](https://www.rfc-editor.org/rfc/rfc9114.html#frame-reserved) by default to test protocol extensibility. However, some HTTP/3 clients and servers don't properly ignore unknown frame types, causing 400/500 errors or stream hangs ([h3#206](https://github.com/hyperium/h3/issues/206)). SmartProxy disables GREASE on both the server side (for incoming H3 requests) and the client side (for H3 backend connections) to maximize compatibility.
|
||||
- **HTTP/3 is pre-release.** The h3 ecosystem (h3 0.0.8, h3-quinn 0.0.10, quinn 0.11) is still pre-1.0. Expect rough edges.
|
||||
|
||||
### Performance Tuning
|
||||
- ✅ Use NFTables forwarding for high-traffic routes (Linux only)
|
||||
- ✅ Enable connection keep-alive where appropriate
|
||||
|
||||
4
rust/Cargo.lock
generated
4
rust/Cargo.lock
generated
@@ -1224,10 +1224,14 @@ dependencies = [
|
||||
"bytes",
|
||||
"clap",
|
||||
"dashmap",
|
||||
"h3",
|
||||
"h3-quinn",
|
||||
"http",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"mimalloc",
|
||||
"quinn",
|
||||
"rcgen",
|
||||
"rustls",
|
||||
"rustls-pemfile",
|
||||
|
||||
@@ -1,64 +1,36 @@
|
||||
//! HTTP/3 proxy service.
|
||||
//!
|
||||
//! Accepts QUIC connections via quinn, runs h3 server to handle HTTP/3 requests,
|
||||
//! and forwards them to backends using the same routing and pool infrastructure
|
||||
//! as the HTTP/1+2 proxy.
|
||||
//! and delegates backend forwarding to the shared `HttpProxyService` — same
|
||||
//! route matching, connection pool, and protocol auto-detection as TCP/HTTP.
|
||||
|
||||
use std::net::SocketAddr;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::task::{Context, Poll};
|
||||
use std::time::Duration;
|
||||
|
||||
use arc_swap::ArcSwap;
|
||||
use bytes::{Buf, Bytes};
|
||||
use http_body::Frame;
|
||||
use http_body_util::BodyExt;
|
||||
use http_body_util::combinators::BoxBody;
|
||||
use tracing::{debug, warn};
|
||||
|
||||
use rustproxy_config::{RouteConfig, TransportProtocol};
|
||||
use rustproxy_metrics::MetricsCollector;
|
||||
use rustproxy_routing::{MatchContext, RouteManager};
|
||||
use rustproxy_config::RouteConfig;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::connection_pool::ConnectionPool;
|
||||
use crate::protocol_cache::ProtocolCache;
|
||||
use crate::upstream_selector::UpstreamSelector;
|
||||
use crate::proxy_service::{ConnActivity, HttpProxyService};
|
||||
|
||||
/// HTTP/3 proxy service.
|
||||
///
|
||||
/// Handles QUIC connections with the h3 crate, parses HTTP/3 requests,
|
||||
/// and forwards them to backends using per-request route matching and
|
||||
/// shared connection pooling.
|
||||
/// Accepts QUIC connections, parses HTTP/3 requests, and delegates backend
|
||||
/// forwarding to the shared `HttpProxyService`.
|
||||
pub struct H3ProxyService {
|
||||
route_manager: Arc<ArcSwap<RouteManager>>,
|
||||
metrics: Arc<MetricsCollector>,
|
||||
connection_pool: Arc<ConnectionPool>,
|
||||
#[allow(dead_code)]
|
||||
protocol_cache: Arc<ProtocolCache>,
|
||||
#[allow(dead_code)]
|
||||
upstream_selector: UpstreamSelector,
|
||||
#[allow(dead_code)]
|
||||
backend_tls_config: Arc<rustls::ClientConfig>,
|
||||
connect_timeout: Duration,
|
||||
http_proxy: Arc<HttpProxyService>,
|
||||
}
|
||||
|
||||
impl H3ProxyService {
|
||||
pub fn new(
|
||||
route_manager: Arc<ArcSwap<RouteManager>>,
|
||||
metrics: Arc<MetricsCollector>,
|
||||
connection_pool: Arc<ConnectionPool>,
|
||||
protocol_cache: Arc<ProtocolCache>,
|
||||
backend_tls_config: Arc<rustls::ClientConfig>,
|
||||
connect_timeout: Duration,
|
||||
) -> Self {
|
||||
Self {
|
||||
route_manager: Arc::clone(&route_manager),
|
||||
metrics: Arc::clone(&metrics),
|
||||
connection_pool,
|
||||
protocol_cache,
|
||||
upstream_selector: UpstreamSelector::new(),
|
||||
backend_tls_config,
|
||||
connect_timeout,
|
||||
}
|
||||
pub fn new(http_proxy: Arc<HttpProxyService>) -> Self {
|
||||
Self { http_proxy }
|
||||
}
|
||||
|
||||
/// Handle an accepted QUIC connection as HTTP/3.
|
||||
@@ -76,12 +48,12 @@ impl H3ProxyService {
|
||||
debug!("HTTP/3 connection from {} on port {}", remote_addr, port);
|
||||
|
||||
let mut h3_conn: h3::server::Connection<h3_quinn::Connection, Bytes> =
|
||||
h3::server::Connection::new(h3_quinn::Connection::new(connection))
|
||||
h3::server::builder()
|
||||
.send_grease(false)
|
||||
.build(h3_quinn::Connection::new(connection))
|
||||
.await
|
||||
.map_err(|e| anyhow::anyhow!("H3 connection setup failed: {}", e))?;
|
||||
|
||||
let client_ip = remote_addr.ip().to_string();
|
||||
|
||||
loop {
|
||||
match h3_conn.accept().await {
|
||||
Ok(Some(resolver)) => {
|
||||
@@ -93,19 +65,13 @@ impl H3ProxyService {
|
||||
}
|
||||
};
|
||||
|
||||
self.metrics.record_http_request();
|
||||
|
||||
let rm = self.route_manager.load();
|
||||
let pool = Arc::clone(&self.connection_pool);
|
||||
let metrics = Arc::clone(&self.metrics);
|
||||
let connect_timeout = self.connect_timeout;
|
||||
let client_ip = client_ip.clone();
|
||||
let http_proxy = Arc::clone(&self.http_proxy);
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = handle_h3_request(
|
||||
request, stream, port, &client_ip, &rm, &pool, &metrics, connect_timeout,
|
||||
request, stream, port, remote_addr, &http_proxy,
|
||||
).await {
|
||||
debug!("HTTP/3 request error from {}: {}", client_ip, e);
|
||||
debug!("HTTP/3 request error from {}: {}", remote_addr, e);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -124,86 +90,27 @@ impl H3ProxyService {
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle a single HTTP/3 request with per-request route matching.
|
||||
/// Handle a single HTTP/3 request by delegating to HttpProxyService.
|
||||
///
|
||||
/// 1. Read the H3 request body via an mpsc channel (streaming, not buffered)
|
||||
/// 2. Build a `hyper::Request<BoxBody>` that HttpProxyService can handle
|
||||
/// 3. Call `HttpProxyService::handle_request` — same route matching, connection
|
||||
/// pool, ALPN protocol detection (H1/H2/H3) as the TCP/HTTP path
|
||||
/// 4. Stream the response back over the H3 stream
|
||||
async fn handle_h3_request(
|
||||
request: hyper::Request<()>,
|
||||
mut stream: h3::server::RequestStream<h3_quinn::BidiStream<Bytes>, Bytes>,
|
||||
port: u16,
|
||||
client_ip: &str,
|
||||
route_manager: &RouteManager,
|
||||
_connection_pool: &ConnectionPool,
|
||||
metrics: &MetricsCollector,
|
||||
connect_timeout: Duration,
|
||||
peer_addr: SocketAddr,
|
||||
http_proxy: &HttpProxyService,
|
||||
) -> anyhow::Result<()> {
|
||||
let method = request.method().clone();
|
||||
let uri = request.uri().clone();
|
||||
let path = uri.path().to_string();
|
||||
|
||||
// Extract host from :authority or Host header
|
||||
let host = request.uri().authority()
|
||||
.map(|a| a.as_str().to_string())
|
||||
.or_else(|| request.headers().get("host").and_then(|v| v.to_str().ok()).map(|s| s.to_string()))
|
||||
.unwrap_or_default();
|
||||
|
||||
debug!("HTTP/3 {} {} (host: {}, client: {})", method, path, host, client_ip);
|
||||
|
||||
// Per-request route matching
|
||||
let ctx = MatchContext {
|
||||
port,
|
||||
domain: if host.is_empty() { None } else { Some(&host) },
|
||||
path: Some(&path),
|
||||
client_ip: Some(client_ip),
|
||||
tls_version: Some("TLSv1.3"),
|
||||
headers: None,
|
||||
is_tls: true,
|
||||
protocol: Some("http"),
|
||||
transport: Some(TransportProtocol::Udp),
|
||||
};
|
||||
|
||||
let route_match = route_manager.find_route(&ctx)
|
||||
.ok_or_else(|| anyhow::anyhow!("No route matched for HTTP/3 request to {}{}", host, path))?;
|
||||
let route = route_match.route;
|
||||
|
||||
// Resolve backend target (use matched target or first target)
|
||||
let target = route_match.target
|
||||
.or_else(|| route.action.targets.as_ref().and_then(|t| t.first()))
|
||||
.ok_or_else(|| anyhow::anyhow!("No target for HTTP/3 route"))?;
|
||||
|
||||
let backend_host = target.host.first();
|
||||
let backend_port = target.port.resolve(port);
|
||||
let backend_addr = format!("{}:{}", backend_host, backend_port);
|
||||
|
||||
// Connect to backend via TCP HTTP/1.1 with timeout
|
||||
let tcp_stream = tokio::time::timeout(
|
||||
connect_timeout,
|
||||
tokio::net::TcpStream::connect(&backend_addr),
|
||||
).await
|
||||
.map_err(|_| anyhow::anyhow!("Backend connect timeout to {}", backend_addr))?
|
||||
.map_err(|e| anyhow::anyhow!("Backend connect to {} failed: {}", backend_addr, e))?;
|
||||
|
||||
let _ = tcp_stream.set_nodelay(true);
|
||||
|
||||
let io = hyper_util::rt::TokioIo::new(tcp_stream);
|
||||
let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await
|
||||
.map_err(|e| anyhow::anyhow!("Backend handshake failed: {}", e))?;
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = conn.await {
|
||||
debug!("Backend connection closed: {}", e);
|
||||
}
|
||||
});
|
||||
|
||||
// Stream request body from H3 client to backend via an mpsc channel.
|
||||
// This avoids buffering the entire request body in memory.
|
||||
// Stream request body from H3 client via an mpsc channel.
|
||||
let (body_tx, body_rx) = tokio::sync::mpsc::channel::<Bytes>(4);
|
||||
let total_bytes_in = Arc::new(std::sync::atomic::AtomicU64::new(0));
|
||||
let total_bytes_in_writer = Arc::clone(&total_bytes_in);
|
||||
|
||||
// Spawn the H3 body reader task
|
||||
let body_reader = tokio::spawn(async move {
|
||||
while let Ok(Some(mut chunk)) = stream.recv_data().await {
|
||||
let data = Bytes::copy_from_slice(chunk.chunk());
|
||||
total_bytes_in_writer.fetch_add(data.len() as u64, std::sync::atomic::Ordering::Relaxed);
|
||||
chunk.advance(chunk.remaining());
|
||||
if body_tx.send(data).await.is_err() {
|
||||
break;
|
||||
@@ -212,106 +119,64 @@ async fn handle_h3_request(
|
||||
stream
|
||||
});
|
||||
|
||||
// Create a body that polls from the mpsc receiver
|
||||
// Build a hyper::Request<BoxBody> from the H3 request + streaming body.
|
||||
// The URI already has scheme + authority + path set by the h3 crate.
|
||||
let body = H3RequestBody { receiver: body_rx };
|
||||
let backend_req = build_backend_request(&method, &backend_addr, &path, &host, &request, body)?;
|
||||
let (parts, _) = request.into_parts();
|
||||
let boxed_body: BoxBody<Bytes, hyper::Error> = BoxBody::new(body);
|
||||
let req = hyper::Request::from_parts(parts, boxed_body);
|
||||
|
||||
let response = sender.send_request(backend_req).await
|
||||
// Delegate to HttpProxyService — same backend path as TCP/HTTP:
|
||||
// route matching, ALPN protocol detection, connection pool, H1/H2/H3 auto.
|
||||
let cancel = CancellationToken::new();
|
||||
let conn_activity = ConnActivity::new_standalone();
|
||||
let response = http_proxy.handle_request(req, peer_addr, port, cancel, conn_activity).await
|
||||
.map_err(|e| anyhow::anyhow!("Backend request failed: {}", e))?;
|
||||
|
||||
// Await the body reader to get the stream back
|
||||
// Await the body reader to get the H3 stream back
|
||||
let mut stream = body_reader.await
|
||||
.map_err(|e| anyhow::anyhow!("Body reader task failed: {}", e))?;
|
||||
let total_bytes_in = total_bytes_in.load(std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
// Build H3 response
|
||||
let status = response.status();
|
||||
let mut h3_response = hyper::Response::builder().status(status);
|
||||
|
||||
// Copy response headers (skip hop-by-hop)
|
||||
for (name, value) in response.headers() {
|
||||
let n = name.as_str().to_lowercase();
|
||||
// Send response headers over H3 (skip hop-by-hop headers)
|
||||
let (resp_parts, resp_body) = response.into_parts();
|
||||
let mut h3_response = hyper::Response::builder().status(resp_parts.status);
|
||||
for (name, value) in &resp_parts.headers {
|
||||
let n = name.as_str();
|
||||
if n == "transfer-encoding" || n == "connection" || n == "keep-alive" || n == "upgrade" {
|
||||
continue;
|
||||
}
|
||||
h3_response = h3_response.header(name, value);
|
||||
}
|
||||
|
||||
// Add Alt-Svc for HTTP/3 advertisement
|
||||
let alt_svc = route.action.udp.as_ref()
|
||||
.and_then(|u| u.quic.as_ref())
|
||||
.map(|q| {
|
||||
let p = q.alt_svc_port.unwrap_or(port);
|
||||
let ma = q.alt_svc_max_age.unwrap_or(86400);
|
||||
format!("h3=\":{}\"; ma={}", p, ma)
|
||||
})
|
||||
.unwrap_or_else(|| format!("h3=\":{}\"; ma=86400", port));
|
||||
h3_response = h3_response.header("alt-svc", alt_svc);
|
||||
|
||||
let h3_response = h3_response.body(())
|
||||
.map_err(|e| anyhow::anyhow!("Failed to build H3 response: {}", e))?;
|
||||
|
||||
// Send response headers
|
||||
stream.send_response(h3_response).await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to send H3 response: {}", e))?;
|
||||
|
||||
// Stream response body back
|
||||
use http_body_util::BodyExt;
|
||||
let mut body = response.into_body();
|
||||
let mut total_bytes_out: u64 = 0;
|
||||
while let Some(frame) = body.frame().await {
|
||||
// Stream response body back over H3
|
||||
let mut resp_body = resp_body;
|
||||
while let Some(frame) = resp_body.frame().await {
|
||||
match frame {
|
||||
Ok(frame) => {
|
||||
if let Some(data) = frame.data_ref() {
|
||||
total_bytes_out += data.len() as u64;
|
||||
stream.send_data(Bytes::copy_from_slice(data)).await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to send H3 data: {}", e))?;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Backend body read error: {}", e);
|
||||
warn!("Response body read error: {}", e);
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Record metrics
|
||||
let route_id = route.name.as_deref().or(route.id.as_deref());
|
||||
metrics.record_bytes(total_bytes_in, total_bytes_out, route_id, Some(client_ip));
|
||||
|
||||
// Finish the stream
|
||||
// Finish the H3 stream (send QUIC FIN)
|
||||
stream.finish().await
|
||||
.map_err(|e| anyhow::anyhow!("Failed to finish H3 stream: {}", e))?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Build an HTTP/1.1 backend request from the H3 frontend request.
|
||||
fn build_backend_request<B>(
|
||||
method: &hyper::Method,
|
||||
backend_addr: &str,
|
||||
path: &str,
|
||||
host: &str,
|
||||
original_request: &hyper::Request<()>,
|
||||
body: B,
|
||||
) -> anyhow::Result<hyper::Request<B>> {
|
||||
let mut req = hyper::Request::builder()
|
||||
.method(method)
|
||||
.uri(format!("http://{}{}", backend_addr, path))
|
||||
.header("host", host);
|
||||
|
||||
// Forward non-pseudo headers
|
||||
for (name, value) in original_request.headers() {
|
||||
let n = name.as_str();
|
||||
if !n.starts_with(':') && n != "host" {
|
||||
req = req.header(name, value);
|
||||
}
|
||||
}
|
||||
|
||||
req.body(body)
|
||||
.map_err(|e| anyhow::anyhow!("Failed to build backend request: {}", e))
|
||||
}
|
||||
|
||||
/// A streaming request body backed by an mpsc channel receiver.
|
||||
///
|
||||
/// Implements `http_body::Body` so hyper can poll chunks as they arrive
|
||||
|
||||
@@ -36,7 +36,7 @@ use crate::upstream_selector::UpstreamSelector;
|
||||
/// Per-connection context for keeping the idle watchdog alive during body streaming.
|
||||
/// Passed through the forwarding chain so CountingBody can update the timestamp.
|
||||
#[derive(Clone)]
|
||||
struct ConnActivity {
|
||||
pub struct ConnActivity {
|
||||
last_activity: Arc<AtomicU64>,
|
||||
start: std::time::Instant,
|
||||
/// Active-request counter from handle_io's idle watchdog. When set, CountingBody
|
||||
@@ -49,6 +49,19 @@ struct ConnActivity {
|
||||
alt_svc_cache_key: Option<crate::protocol_cache::ProtocolCacheKey>,
|
||||
}
|
||||
|
||||
impl ConnActivity {
|
||||
/// Create a minimal ConnActivity (no idle watchdog, no Alt-Svc cache).
|
||||
/// Used by H3ProxyService where the TCP idle watchdog doesn't apply.
|
||||
pub fn new_standalone() -> Self {
|
||||
Self {
|
||||
last_activity: Arc::new(AtomicU64::new(0)),
|
||||
start: std::time::Instant::now(),
|
||||
active_requests: None,
|
||||
alt_svc_cache_key: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Default upstream connect timeout (30 seconds).
|
||||
const DEFAULT_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
|
||||
|
||||
@@ -347,6 +360,7 @@ impl HttpProxyService {
|
||||
let st = start;
|
||||
let ca = ConnActivity { last_activity: Arc::clone(&la_inner), start, active_requests: Some(Arc::clone(&ar_inner)), alt_svc_cache_key: None };
|
||||
async move {
|
||||
let req = req.map(|body| BoxBody::new(body));
|
||||
let result = svc.handle_request(req, peer, port, cn, ca).await;
|
||||
// Mark request end — update activity timestamp before guard drops
|
||||
la.store(st.elapsed().as_millis() as u64, Ordering::Relaxed);
|
||||
@@ -416,9 +430,13 @@ impl HttpProxyService {
|
||||
}
|
||||
|
||||
/// Handle a single HTTP request.
|
||||
async fn handle_request(
|
||||
///
|
||||
/// Accepts a generic body (`BoxBody`) so both the TCP/HTTP path (which boxes
|
||||
/// `Incoming`) and the H3 path (which boxes the H3 request body stream) can
|
||||
/// share the same backend forwarding logic.
|
||||
pub async fn handle_request(
|
||||
&self,
|
||||
req: Request<Incoming>,
|
||||
req: Request<BoxBody<Bytes, hyper::Error>>,
|
||||
peer_addr: std::net::SocketAddr,
|
||||
port: u16,
|
||||
cancel: CancellationToken,
|
||||
@@ -965,7 +983,7 @@ impl HttpProxyService {
|
||||
&self,
|
||||
io: TokioIo<BackendStream>,
|
||||
parts: hyper::http::request::Parts,
|
||||
body: Incoming,
|
||||
body: BoxBody<Bytes, hyper::Error>,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
_upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
@@ -1013,7 +1031,7 @@ impl HttpProxyService {
|
||||
&self,
|
||||
mut sender: hyper::client::conn::http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
|
||||
parts: hyper::http::request::Parts,
|
||||
body: Incoming,
|
||||
body: BoxBody<Bytes, hyper::Error>,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
route: &rustproxy_config::RouteConfig,
|
||||
@@ -1077,7 +1095,7 @@ impl HttpProxyService {
|
||||
&self,
|
||||
io: TokioIo<BackendStream>,
|
||||
parts: hyper::http::request::Parts,
|
||||
body: Incoming,
|
||||
body: BoxBody<Bytes, hyper::Error>,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
_upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
@@ -1151,7 +1169,7 @@ impl HttpProxyService {
|
||||
&self,
|
||||
sender: hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
|
||||
parts: hyper::http::request::Parts,
|
||||
body: Incoming,
|
||||
body: BoxBody<Bytes, hyper::Error>,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
route: &rustproxy_config::RouteConfig,
|
||||
@@ -1344,7 +1362,7 @@ impl HttpProxyService {
|
||||
&self,
|
||||
io: TokioIo<BackendStream>,
|
||||
parts: hyper::http::request::Parts,
|
||||
body: Incoming,
|
||||
body: BoxBody<Bytes, hyper::Error>,
|
||||
mut upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
@@ -1675,7 +1693,7 @@ impl HttpProxyService {
|
||||
&self,
|
||||
mut sender: hyper::client::conn::http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
|
||||
parts: hyper::http::request::Parts,
|
||||
body: Incoming,
|
||||
body: BoxBody<Bytes, hyper::Error>,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
route: &rustproxy_config::RouteConfig,
|
||||
@@ -1816,7 +1834,7 @@ impl HttpProxyService {
|
||||
/// Handle a WebSocket upgrade request (H1 Upgrade or H2 Extended CONNECT per RFC 8441).
|
||||
async fn handle_websocket_upgrade(
|
||||
&self,
|
||||
req: Request<Incoming>,
|
||||
req: Request<BoxBody<Bytes, hyper::Error>>,
|
||||
peer_addr: std::net::SocketAddr,
|
||||
upstream: &crate::upstream_selector::UpstreamSelection,
|
||||
route: &rustproxy_config::RouteConfig,
|
||||
@@ -2538,7 +2556,7 @@ impl HttpProxyService {
|
||||
&self,
|
||||
quic_conn: quinn::Connection,
|
||||
parts: hyper::http::request::Parts,
|
||||
body: Incoming,
|
||||
body: BoxBody<Bytes, hyper::Error>,
|
||||
upstream_headers: hyper::HeaderMap,
|
||||
upstream_path: &str,
|
||||
route: &rustproxy_config::RouteConfig,
|
||||
@@ -2550,7 +2568,11 @@ impl HttpProxyService {
|
||||
backend_key: &str,
|
||||
) -> Result<Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
|
||||
let h3_quinn_conn = h3_quinn::Connection::new(quic_conn.clone());
|
||||
let (mut driver, mut send_request) = match h3::client::new(h3_quinn_conn).await {
|
||||
let (mut driver, mut send_request) = match h3::client::builder()
|
||||
.send_grease(false)
|
||||
.build(h3_quinn_conn)
|
||||
.await
|
||||
{
|
||||
Ok(pair) => pair,
|
||||
Err(e) => {
|
||||
error!(backend = %backend_key, domain = %domain, error = %e, "H3 client handshake failed");
|
||||
|
||||
@@ -6,7 +6,6 @@ use std::sync::Arc;
|
||||
use bytes::Bytes;
|
||||
use http_body_util::Full;
|
||||
use http_body_util::BodyExt;
|
||||
use hyper::body::Incoming;
|
||||
use hyper::{Request, Response, StatusCode};
|
||||
use http_body_util::combinators::BoxBody;
|
||||
|
||||
@@ -19,7 +18,7 @@ impl RequestFilter {
|
||||
/// Apply security filters. Returns Some(response) if the request should be blocked.
|
||||
pub fn apply(
|
||||
security: &RouteSecurity,
|
||||
req: &Request<Incoming>,
|
||||
req: &Request<impl hyper::body::Body>,
|
||||
peer_addr: &SocketAddr,
|
||||
) -> Option<Response<BoxBody<Bytes, hyper::Error>>> {
|
||||
Self::apply_with_rate_limiter(security, req, peer_addr, None)
|
||||
@@ -29,7 +28,7 @@ impl RequestFilter {
|
||||
/// Returns Some(response) if the request should be blocked.
|
||||
pub fn apply_with_rate_limiter(
|
||||
security: &RouteSecurity,
|
||||
req: &Request<Incoming>,
|
||||
req: &Request<impl hyper::body::Body>,
|
||||
peer_addr: &SocketAddr,
|
||||
rate_limiter: Option<&Arc<RateLimiter>>,
|
||||
) -> Option<Response<BoxBody<Bytes, hyper::Error>>> {
|
||||
@@ -182,7 +181,7 @@ impl RequestFilter {
|
||||
/// Determine the rate limit key based on configuration.
|
||||
fn rate_limit_key(
|
||||
config: &rustproxy_config::RouteRateLimit,
|
||||
req: &Request<Incoming>,
|
||||
req: &Request<impl hyper::body::Body>,
|
||||
peer_addr: &SocketAddr,
|
||||
) -> String {
|
||||
use rustproxy_config::RateLimitKeyBy;
|
||||
@@ -220,7 +219,7 @@ impl RequestFilter {
|
||||
/// Handle CORS preflight (OPTIONS) requests.
|
||||
/// Returns Some(response) if this is a CORS preflight that should be handled.
|
||||
pub fn handle_cors_preflight(
|
||||
req: &Request<Incoming>,
|
||||
req: &Request<impl hyper::body::Body>,
|
||||
) -> Option<Response<BoxBody<Bytes, hyper::Error>>> {
|
||||
if req.method() != hyper::Method::OPTIONS {
|
||||
return None;
|
||||
|
||||
@@ -428,6 +428,11 @@ impl TcpListenerManager {
|
||||
self.http_proxy.prune_stale_routes(active_route_ids);
|
||||
}
|
||||
|
||||
/// Get a reference to the HTTP proxy service (shared with H3).
|
||||
pub fn http_proxy(&self) -> &Arc<HttpProxyService> {
|
||||
&self.http_proxy
|
||||
}
|
||||
|
||||
/// Get a reference to the connection tracker.
|
||||
pub fn conn_tracker(&self) -> &Arc<ConnectionTracker> {
|
||||
&self.conn_tracker
|
||||
|
||||
@@ -44,3 +44,9 @@ mimalloc = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
rcgen = { workspace = true }
|
||||
quinn = { workspace = true }
|
||||
h3 = { workspace = true }
|
||||
h3-quinn = { workspace = true }
|
||||
bytes = { workspace = true }
|
||||
rustls = { workspace = true }
|
||||
http = "1"
|
||||
|
||||
@@ -343,15 +343,10 @@ impl RustProxy {
|
||||
);
|
||||
udp_mgr.set_proxy_ips(udp_proxy_ips.clone());
|
||||
|
||||
// Construct H3ProxyService for HTTP/3 request handling
|
||||
let h3_svc = rustproxy_http::h3_service::H3ProxyService::new(
|
||||
Arc::new(ArcSwap::from(Arc::clone(&*self.route_table.load()))),
|
||||
Arc::clone(&self.metrics),
|
||||
Arc::new(rustproxy_http::connection_pool::ConnectionPool::new()),
|
||||
Arc::new(rustproxy_http::protocol_cache::ProtocolCache::new()),
|
||||
rustproxy_passthrough::tls_handler::shared_backend_tls_config(),
|
||||
std::time::Duration::from_secs(30),
|
||||
);
|
||||
// Share HttpProxyService with H3 — same route matching, connection
|
||||
// pool, and ALPN protocol detection as the TCP/HTTP path.
|
||||
let http_proxy = self.listener_manager.as_ref().unwrap().http_proxy().clone();
|
||||
let h3_svc = rustproxy_http::h3_service::H3ProxyService::new(http_proxy);
|
||||
udp_mgr.set_h3_service(Arc::new(h3_svc));
|
||||
|
||||
for port in &udp_ports {
|
||||
@@ -1003,44 +998,25 @@ impl RustProxy {
|
||||
fn build_quic_tls_config(
|
||||
tls_configs: &HashMap<String, TlsCertConfig>,
|
||||
) -> Option<Arc<rustls::ServerConfig>> {
|
||||
// Find the first available cert (prefer wildcard, then any)
|
||||
let cert_config = tls_configs.get("*")
|
||||
.or_else(|| tls_configs.values().next());
|
||||
|
||||
let cert_config = match cert_config {
|
||||
Some(c) => c,
|
||||
None => return None,
|
||||
};
|
||||
|
||||
// Parse cert chain from PEM
|
||||
let mut cert_reader = std::io::BufReader::new(cert_config.cert_pem.as_bytes());
|
||||
let certs: Vec<rustls::pki_types::CertificateDer<'static>> =
|
||||
rustls_pemfile::certs(&mut cert_reader)
|
||||
.filter_map(|r| r.ok())
|
||||
.collect();
|
||||
|
||||
if certs.is_empty() {
|
||||
if tls_configs.is_empty() {
|
||||
return None;
|
||||
}
|
||||
|
||||
// Parse private key from PEM
|
||||
let mut key_reader = std::io::BufReader::new(cert_config.key_pem.as_bytes());
|
||||
let key = match rustls_pemfile::private_key(&mut key_reader) {
|
||||
Ok(Some(key)) => key,
|
||||
_ => return None,
|
||||
};
|
||||
|
||||
let mut tls_config = match rustls::ServerConfig::builder()
|
||||
.with_no_client_auth()
|
||||
.with_single_cert(certs, key)
|
||||
{
|
||||
Ok(c) => c,
|
||||
// Reuse CertResolver for SNI-based cert selection (same as TCP/TLS path).
|
||||
// This ensures QUIC connections get the correct certificate for each domain
|
||||
// instead of a single static cert.
|
||||
let resolver = match rustproxy_passthrough::tls_handler::CertResolver::new(tls_configs) {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
warn!("Failed to build QUIC TLS config: {}", e);
|
||||
warn!("Failed to build QUIC cert resolver: {}", e);
|
||||
return None;
|
||||
}
|
||||
};
|
||||
|
||||
let mut tls_config = rustls::ServerConfig::builder()
|
||||
.with_no_client_auth()
|
||||
.with_cert_resolver(Arc::new(resolver));
|
||||
|
||||
// QUIC requires h3 ALPN
|
||||
tls_config.alpn_protocols = vec![b"h3".to_vec()];
|
||||
|
||||
|
||||
195
rust/crates/rustproxy/tests/integration_h3_proxy.rs
Normal file
195
rust/crates/rustproxy/tests/integration_h3_proxy.rs
Normal file
@@ -0,0 +1,195 @@
|
||||
mod common;
|
||||
|
||||
use common::*;
|
||||
use rustproxy::RustProxy;
|
||||
use rustproxy_config::{RustProxyOptions, TransportProtocol, RouteUdp, RouteQuic};
|
||||
use bytes::Buf;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Build a route that listens on UDP with HTTP/3 enabled and TLS terminate.
|
||||
fn make_h3_route(
|
||||
port: u16,
|
||||
target_host: &str,
|
||||
target_port: u16,
|
||||
cert_pem: &str,
|
||||
key_pem: &str,
|
||||
) -> rustproxy_config::RouteConfig {
|
||||
let mut route = make_tls_terminate_route(port, "localhost", target_host, target_port, cert_pem, key_pem);
|
||||
route.route_match.transport = Some(TransportProtocol::All);
|
||||
// Keep domain="localhost" from make_tls_terminate_route — needed for TLS cert extraction
|
||||
route.action.udp = Some(RouteUdp {
|
||||
session_timeout: None,
|
||||
max_sessions_per_ip: None,
|
||||
max_datagram_size: None,
|
||||
quic: Some(RouteQuic {
|
||||
max_idle_timeout: Some(30000),
|
||||
max_concurrent_bidi_streams: None,
|
||||
max_concurrent_uni_streams: None,
|
||||
enable_http3: Some(true),
|
||||
alt_svc_port: None,
|
||||
alt_svc_max_age: None,
|
||||
initial_congestion_window: None,
|
||||
}),
|
||||
});
|
||||
route
|
||||
}
|
||||
|
||||
/// Build a quinn client endpoint with insecure TLS for testing.
|
||||
fn make_h3_client_endpoint() -> quinn::Endpoint {
|
||||
let mut tls_config = rustls::ClientConfig::builder()
|
||||
.dangerous()
|
||||
.with_custom_certificate_verifier(Arc::new(InsecureVerifier))
|
||||
.with_no_client_auth();
|
||||
tls_config.alpn_protocols = vec![b"h3".to_vec()];
|
||||
|
||||
let quic_client_config = quinn::crypto::rustls::QuicClientConfig::try_from(tls_config)
|
||||
.expect("Failed to build QUIC client config");
|
||||
let client_config = quinn::ClientConfig::new(Arc::new(quic_client_config));
|
||||
|
||||
let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse().unwrap())
|
||||
.expect("Failed to create QUIC client endpoint");
|
||||
endpoint.set_default_client_config(client_config);
|
||||
endpoint
|
||||
}
|
||||
|
||||
/// Test that HTTP/3 response streams properly finish (FIN is received by client).
|
||||
///
|
||||
/// This is the critical regression test for the FIN bug: the proxy must send
|
||||
/// a QUIC stream FIN after the response body so the client's `recv_data()`
|
||||
/// returns `None` instead of hanging forever.
|
||||
#[tokio::test]
|
||||
async fn test_h3_response_stream_finishes() {
|
||||
let backend_port = next_port();
|
||||
let proxy_port = next_port();
|
||||
let body_text = "Hello from HTTP/3 backend! This body has a known length for testing.";
|
||||
|
||||
// 1. Start plain HTTP backend with known body + content-length
|
||||
let _backend = start_http_server(backend_port, 200, body_text).await;
|
||||
|
||||
// 2. Generate self-signed cert and configure H3 route
|
||||
let (cert_pem, key_pem) = generate_self_signed_cert("localhost");
|
||||
let route = make_h3_route(proxy_port, "127.0.0.1", backend_port, &cert_pem, &key_pem);
|
||||
|
||||
let options = RustProxyOptions {
|
||||
routes: vec![route],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// 3. Start proxy and wait for UDP bind
|
||||
let mut proxy = RustProxy::new(options).unwrap();
|
||||
proxy.start().await.unwrap();
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
|
||||
// 4. Connect QUIC/H3 client
|
||||
let endpoint = make_h3_client_endpoint();
|
||||
let addr: std::net::SocketAddr = format!("127.0.0.1:{}", proxy_port).parse().unwrap();
|
||||
let connection = endpoint
|
||||
.connect(addr, "localhost")
|
||||
.expect("Failed to initiate QUIC connection")
|
||||
.await
|
||||
.expect("QUIC handshake failed");
|
||||
|
||||
let (mut driver, mut send_request) = h3::client::new(
|
||||
h3_quinn::Connection::new(connection),
|
||||
)
|
||||
.await
|
||||
.expect("H3 connection setup failed");
|
||||
|
||||
// Drive the H3 connection in background
|
||||
tokio::spawn(async move {
|
||||
let _ = driver.wait_idle().await;
|
||||
});
|
||||
|
||||
// 5. Send GET request
|
||||
let req = http::Request::builder()
|
||||
.method("GET")
|
||||
.uri("https://localhost/")
|
||||
.header("host", "localhost")
|
||||
.body(())
|
||||
.unwrap();
|
||||
|
||||
let mut stream = send_request.send_request(req).await
|
||||
.expect("Failed to send H3 request");
|
||||
stream.finish().await
|
||||
.expect("Failed to finish sending H3 request body");
|
||||
|
||||
// 6. Read response headers
|
||||
let resp = stream.recv_response().await
|
||||
.expect("Failed to receive H3 response");
|
||||
assert_eq!(resp.status(), http::StatusCode::OK,
|
||||
"Expected 200 OK, got {}", resp.status());
|
||||
|
||||
// 7. Read body and verify stream ends (FIN received)
|
||||
// This is the critical assertion: recv_data() must return None (stream ended)
|
||||
// within the timeout, NOT hang forever waiting for a FIN that never arrives.
|
||||
let result = with_timeout(async {
|
||||
let mut total = 0usize;
|
||||
while let Some(chunk) = stream.recv_data().await.expect("H3 data receive error") {
|
||||
total += chunk.remaining();
|
||||
}
|
||||
// recv_data() returned None => stream ended (FIN received)
|
||||
total
|
||||
}, 10)
|
||||
.await;
|
||||
|
||||
let bytes_received = result.expect(
|
||||
"TIMEOUT: H3 stream never ended (FIN not received by client). \
|
||||
The proxy sent all response data but failed to send the QUIC stream FIN."
|
||||
);
|
||||
assert_eq!(
|
||||
bytes_received,
|
||||
body_text.len(),
|
||||
"Expected {} bytes, got {}",
|
||||
body_text.len(),
|
||||
bytes_received
|
||||
);
|
||||
|
||||
// 8. Cleanup
|
||||
endpoint.close(quinn::VarInt::from_u32(0), b"test done");
|
||||
proxy.stop().await.unwrap();
|
||||
}
|
||||
|
||||
/// Insecure TLS verifier that accepts any certificate (for tests only).
|
||||
#[derive(Debug)]
|
||||
struct InsecureVerifier;
|
||||
|
||||
impl rustls::client::danger::ServerCertVerifier for InsecureVerifier {
|
||||
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::ECDSA_NISTP256_SHA256,
|
||||
rustls::SignatureScheme::ECDSA_NISTP384_SHA384,
|
||||
rustls::SignatureScheme::ED25519,
|
||||
rustls::SignatureScheme::RSA_PSS_SHA256,
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@push.rocks/smartproxy',
|
||||
version: '25.17.1',
|
||||
version: '25.17.10',
|
||||
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.'
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user