gru/RustProxy
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.31.5

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-11 22:12:27 +02:00
1 parent 75104f87ba
commit 4cc994b3bc
86 files changed
+942 -223
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/http3.rs
+80 -137
View File
@@ -28,66 +28,7 @@ use tokio::{
task::JoinSet,
};
use tower::ServiceExt;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct Settings {
pub alt_svc_max_age: u64,
pub max_concurrent_connections: u32,
pub max_concurrent_bidistream: u32,
pub max_concurrent_unistream: u32,
/// QUIC idle timeout in seconds; zero disables the transport idle timeout.
pub max_idle_timeout: u64,
}
impl Default for Settings {
fn default() -> Self {
Self {
alt_svc_max_age: 3600,
max_concurrent_connections: 512,
max_concurrent_bidistream: 100,
max_concurrent_unistream: 100,
max_idle_timeout: 10,
}
}
}
impl Settings {
pub fn validate(&self) -> Result<()> {
ensure!(
self.alt_svc_max_age <= 604800,
"HTTP/3 Alt-Svc lifetime must be at most seven days"
);
ensure!(
(1..=4096).contains(&self.max_concurrent_connections),
"HTTP/3 connections per listener must be 1..4096"
);
ensure!(
(1..=1024).contains(&self.max_concurrent_bidistream),
"HTTP/3 bidirectional streams must be 1..1024"
);
ensure!((3..=1024).contains(&self.max_concurrent_unistream),"HTTP/3 requires at least three peer unidirectional streams (control and QPACK), maximum 1024");
ensure!(
self.max_idle_timeout <= 86400,
"HTTP/3 idle timeout must be 0..86400 seconds"
);
Ok(())
}
pub fn transport(&self) -> Result<quinn::TransportConfig> {
self.validate()?;
let mut transport = quinn::TransportConfig::default();
transport
.max_concurrent_bidi_streams(self.max_concurrent_bidistream.into())
.max_concurrent_uni_streams(self.max_concurrent_unistream.into())
.max_idle_timeout(if self.max_idle_timeout == 0 {
None
} else {
Some(Duration::from_secs(self.max_idle_timeout).try_into()?)
})
.stream_receive_window((256 * 1024u32).into())
.receive_window((8 * 1024 * 1024u32).into())
.send_window(8 * 1024 * 1024)
.datagram_receive_buffer_size(None);
Ok(transport)
}
}
include!("http3/settings.rs");
pub fn crypto(
tls: &Arc<rustls::ServerConfig>,
policy: &crate::tls::ProtocolPolicy,
@@ -176,77 +117,6 @@ pub fn advertise(
}
}
}
struct Lease {
_permit: OwnedSemaphorePermit,
metrics: Arc<Metrics>,
}
impl Drop for Lease {
fn drop(&mut self) {
self.metrics
.http3_active_connections
.fetch_sub(1, Ordering::Relaxed);
self.metrics
.active_connections
.fetch_sub(1, Ordering::Relaxed);
}
}
#[allow(clippy::too_many_arguments)]
fn log_protocol_error(
state: &App,
kind: &'static str,
detail: impl std::fmt::Display,
peer: SocketAddr,
local: SocketAddr,
status: u16,
authority: Option<String>,
host: Option<String>,
sni: Option<String>,
) {
let resolved = state.config.load().resolve_logging_vhost(
local,
true,
None,
authority.as_deref(),
host.as_deref(),
sni.as_deref(),
);
let (vhost, matched_by) = resolved
.map(|resolved| (Some(resolved.vhost), resolved.matched_by))
.unwrap_or((None, "none"));
state.logs.push_protocol_error_attributed(
kind,
detail,
peer,
local,
"quic",
"3",
status,
sni,
authority.or(host),
vhost,
matched_by,
);
}
fn request_identity<B>(request: &http::Request<B>) -> (Option<String>, Option<String>) {
let authority = request
.uri()
.authority()
.map(|value| value.as_str().to_owned());
let host = request
.headers()
.get(http::header::HOST)
.and_then(|value| value.to_str().ok())
.map(str::to_owned);
(authority, host)
}
fn connection_sni(connection: &quinn::Connection) -> Option<String> {
connection
.handshake_data()?
.downcast::<quinn::crypto::rustls::HandshakeData>()
.ok()?
.server_name
.and_then(|name| crate::config::request_host(&name).ok())
}
pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duration) -> Result<()> {
let slots = Arc::new(Semaphore::new(
bound.settings.max_concurrent_connections as usize,
@@ -386,9 +256,19 @@ async fn connection_loop(
let (mut request,stream)=match resolved {
Ok(Ok(request))=>request,
Ok(Err(error))=>{
log_protocol_error(
&state, "http3_request_error", error, peer, local, 400, None, None, tls_sni.clone(),
);
if client_cancelled_request(&error) {
// Record the cancellation with status=0 (no HTTP response)
// for iOS navigation investigations, never synthetic 400.
log_protocol_error(
&state, "http3_peer_cancelled_before_headers",
format!("phase=request_headers stream_status=cancelled error={error}"),
peer, local, 0, None, None, tls_sni.clone(),
);
} else {
log_protocol_error(
&state, "http3_request_error", error, peer, local, 400, None, None, tls_sni.clone(),
);
}
return;
},
Err(_)=>{
@@ -404,6 +284,22 @@ async fn connection_loop(
let extended_connect=request.method()==Method::CONNECT && protocol.is_some();
let websocket=extended_connect && protocol.as_deref()==Some("websocket");
let (mut send,receive)=stream.split();
let stream_id=format!("{:?}",send.send_id());
let diagnostic_method=request.method().as_str().to_owned();
// Do not record query strings or cookies in transport diagnostics.
let diagnostic_path=request.uri().path().chars().take(160).collect::<String>();
let (diagnostic_authority, diagnostic_host)=request_identity(&request);
// GET / is a full document navigation (the original logo).
// Emit only query-free, bounded navigation metadata.
let root_navigation=diagnostic_method=="GET" && diagnostic_path=="/";
if root_navigation { log_root_navigation(
"started", peer, local, &tls_sni, &stream_id, "request_headers", 0, 0, false,
); }
tracing::debug!(
peer=%peer, stream_id=%stream_id,
method=%diagnostic_method, path=%diagnostic_path,
"HTTP/3 request headers accepted"
);
if invalid_headers(request.headers()) {
let (authority, host) = request_identity(&request);
log_protocol_error(
@@ -437,8 +333,12 @@ async fn connection_loop(
crate::proxy::headers::strip_hop_by_hop(&mut parts.headers);
parts.version=Version::HTTP_3;
let response=http::Response::from_parts(parts,());
let response_status=response.status().as_u16();
let mut response_phase="response_headers";
let mut sent_body_bytes=0u64;
let sent=async {
tokio::time::timeout(state.body_idle_timeout,send.send_response(response)).await??;
response_phase="response_body";
if let Some(tunnel)=tunnel {
let Some(mut receive)=tunnel_receive.take() else {
send.stop_stream(Code::H3_INTERNAL_ERROR);
@@ -462,21 +362,62 @@ async fn connection_loop(
while let Some(frame)=body.frame().await {
let frame=frame?;
match frame.into_data() {
Ok(data)=>{if !data.is_empty() {tokio::time::timeout(state.body_idle_timeout,send.send_data(data)).await??;}},
Ok(data)=>{if !data.is_empty() {
let data_len=data.len() as u64;
tokio::time::timeout(state.body_idle_timeout,send.send_data(data)).await??;
sent_body_bytes=sent_body_bytes.saturating_add(data_len);
}},
Err(frame)=>{if let Ok(mut trailers)=frame.into_trailers() {
crate::proxy::headers::strip_hop_by_hop(&mut trailers);
tokio::time::timeout(state.body_idle_timeout,send.send_trailers(trailers)).await??;
}},
}
}
response_phase="response_finish";
tokio::time::timeout(state.body_idle_timeout,send.finish()).await??;
Ok::<(),anyhow::Error>(())
}.await;
if let Err(error)=sent {tracing::debug!(%error,"HTTP/3 response interrupted");send.stop_stream(Code::H3_INTERNAL_ERROR);}
if let Err(error)=sent {
let cancelled=error.downcast_ref::<h3::error::StreamError>()
.is_some_and(client_cancelled_request);
let kind=if cancelled { "http3_peer_cancelled_after_headers" }
else { "http3_response_write_error" };
log_protocol_error(
&state, kind,
format!("phase={response_phase} stream_id={stream_id} method={diagnostic_method} path={diagnostic_path} response_status={response_status} body_bytes_sent={sent_body_bytes} error={error}"),
peer, local, 0,
diagnostic_authority, diagnostic_host, tls_sni.clone(),
);
if root_navigation { log_root_navigation(
"interrupted", peer, local, &tls_sni, &stream_id, response_phase,
response_status, sent_body_bytes, cancelled,
); }
if !cancelled {
// Never mark a truncated upstream response as success.
// Reset only this response stream, not sibling requests.
send.stop_stream(Code::H3_INTERNAL_ERROR);
}
} else if root_navigation {
log_root_navigation("complete", peer, local, &tls_sni, &stream_id,
"response_finish", response_status, sent_body_bytes, false);
} else {
log_h3_response_complete(peer, &stream_id, response_status, sent_body_bytes);
}
});
},
Ok(None)=>break,
Err(error)=>{requests.abort_all();return Err(error.into());},
Err(error)=>{
// The parser/QUIC transport can stop accepting after a peer
// refresh while unrelated response streams are still alive.
// Abort_all here reset those streams (including a new GET /).
// Stop accepting and allow the bounded drain below instead.
log_protocol_error(
&state, "http3_accept_error", &error,
connection.remote_address(), local, 0, None, None, tls_sni.clone(),
);
tracing::debug!(%error, "HTTP/3 accept stopped; draining in-flight streams");
break;
},
}
}
}
@@ -492,6 +433,8 @@ async fn connection_loop(
}
Ok(())
}
include!("http3/identity.rs");
include!("http3/diagnostics.rs");
include!("http3/body.rs");
#[cfg(test)]