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

v0.26.0

Browse Source
This commit is contained in:
Mateusz Gruszczyński committed 2026-10-05 17:05:06 +02:00
1 parent d168ee2afa
commit ddc5c6593f
70 files changed
+2293 -594
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/http3.rs
+63 -122
View File
@@ -203,10 +203,15 @@ pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duratio
_=tasks.join_next(),if !tasks.is_empty()=>{},
incoming=bound.endpoint.accept()=>{
let Some(incoming)=incoming else {break;};
let peer=incoming.remote_address();
if !incoming.remote_address_validated() {let _=incoming.retry();continue;}
let Ok(permit)=slots.clone().try_acquire_owned() else {
state.metrics.http3_rejected_connections.fetch_add(1,Ordering::Relaxed);
state.metrics.rejected_connections.fetch_add(1,Ordering::Relaxed);
state.logs.push_protocol_error(
"http3_connection_limit_rejected", "HTTP/3 connection concurrency limit reached",
peer, bound.address, "quic", "3", 0, None,
);
incoming.refuse();continue;
};
let mut config=bound.server_config.clone();
@@ -214,7 +219,15 @@ pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duratio
// Certificate changes apply to new handshakes without changing
// active transport limits or regenerating the Retry token key.
config.crypto=crypto;
let connecting=match incoming.accept_with(Arc::new(config)) {Ok(c)=>c,Err(_)=>continue};
let connecting=match incoming.accept_with(Arc::new(config)) {
Ok(c)=>c,
Err(error)=>{
state.logs.push_protocol_error(
"http3_accept_error", error, peer, bound.address, "quic", "3", 0, None,
);
continue;
}
};
let router=router.clone();let state=state.clone();let address=bound.address;
let limit=bound.settings.max_concurrent_bidistream as usize;
tasks.spawn(async move {
@@ -222,7 +235,21 @@ pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duratio
_=state.shutdown.cancelled()=>return,
result=tokio::time::timeout(Duration::from_secs(10),connecting)=>match result {
Ok(Ok(connection))=>connection,
_=>{state.metrics.tls_handshake_errors.fetch_add(1,Ordering::Relaxed);return;},
Ok(Err(error))=>{
state.metrics.tls_handshake_errors.fetch_add(1,Ordering::Relaxed);
state.logs.push_protocol_error(
"http3_handshake_error", error, peer, address, "quic", "3", 0, None,
);
return;
},
Err(_)=>{
state.metrics.tls_handshake_errors.fetch_add(1,Ordering::Relaxed);
state.logs.push_protocol_error(
"http3_handshake_timeout", "HTTP/3 QUIC/TLS handshake timeout",
peer, address, "quic", "3", 0, None,
);
return;
},
}
};
// A wildcard without destination metadata must fail closed.
@@ -237,7 +264,10 @@ pub async fn serve(bound: Bound, router: Router, state: Arc<App>, grace: Duratio
state.metrics.http3_active_connections.fetch_add(1,Ordering::Relaxed);
state.metrics.tls_connections.fetch_add(1,Ordering::Relaxed);
let _lease=Lease {_permit:permit,metrics:state.metrics.clone()};
if let Err(error)=connection_loop(connection,local,router,state,limit,grace).await {
if let Err(error)=connection_loop(connection,local,router,state.clone(),limit,grace).await {
state.logs.push_protocol_error(
"http3_connection_error", &error, peer, local, "quic", "3", 0, None,
);
tracing::debug!(%error,"HTTP/3 connection ended");
}
});
@@ -285,16 +315,42 @@ async fn connection_loop(
next=h3.accept()=>match next {
Ok(Some(resolver))=>{
// Completed JoinSet entries are drained before accepting more.
if requests.len()>=limit {drop(resolver);continue;}
if requests.len()>=limit {
state.logs.push_protocol_error(
"http3_stream_limit_rejected", "HTTP/3 concurrent request stream limit reached",
connection.remote_address(), local, "quic", "3", 0, None,
);
drop(resolver);continue;
}
let router=router.clone();let state=state.clone();let peer=connection.remote_address();
requests.spawn(async move {
let Ok(Ok((mut request,stream)))=tokio::time::timeout(state.header_timeout,resolver.resolve_request()).await else {return;};
let resolved=tokio::time::timeout(state.header_timeout,resolver.resolve_request()).await;
let (mut request,stream)=match resolved {
Ok(Ok(request))=>request,
Ok(Err(error))=>{
state.logs.push_protocol_error(
"http3_request_error", error, peer, local, "quic", "3", 400, None,
);
return;
},
Err(_)=>{
state.logs.push_protocol_error(
"http3_header_timeout", "HTTP/3 request header timeout",
peer, local, "quic", "3", 408, None,
);
return;
},
};
*request.version_mut()=Version::HTTP_3;
let protocol=request.extensions().get::<h3::ext::Protocol>().map(|protocol|protocol.as_str().to_owned());
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();
if invalid_headers(request.headers()) {
state.logs.push_protocol_error(
"http3_invalid_headers", "HTTP/3 request contains forbidden connection-specific headers",
peer, local, "quic", "3", 400, None,
);
send.stop_stream(Code::H3_MESSAGE_ERROR);return;
}
let expected=request.headers().get("content-length").and_then(|v|v.to_str().ok()).and_then(|v|v.parse::<u64>().ok());
@@ -377,122 +433,7 @@ async fn connection_loop(
}
Ok(())
}
fn invalid_headers(headers: &HeaderMap) -> bool {
[
"connection",
"keep-alive",
"proxy-connection",
"transfer-encoding",
"upgrade",
]
.iter()
.any(|name| headers.contains_key(*name))
|| headers
.get_all("te")
.iter()
.any(|value| value.as_bytes() != b"trailers")
|| headers.get_all("content-length").iter().count() > 1
|| headers.get("content-length").is_some_and(|value| {
value
.to_str()
.ok()
.and_then(|s| s.parse::<u64>().ok())
.is_none()
})
}
struct ReceiveBody {
stream: h3::server::RequestStream<h3::quinn::RecvStream, Bytes>,
expected: Option<u64>,
received: u64,
data_done: bool,
done: bool,
}
impl HttpBody for ReceiveBody {
type Data = Bytes;
type Error = io::Error;
fn poll_frame(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Option<std::result::Result<Frame<Bytes>, io::Error>>> {
let this = self.get_mut();
if this.done {
return Poll::Ready(None);
}
if !this.data_done {
match this.stream.poll_recv_data(cx) {
Poll::Pending => return Poll::Pending,
Poll::Ready(Err(error)) => {
this.done = true;
return Poll::Ready(Some(Err(io::Error::other(error.to_string()))));
}
Poll::Ready(Ok(Some(mut data))) => {
let length = data.remaining();
this.received = this.received.saturating_add(length as u64);
if this
.expected
.is_some_and(|expected| this.received > expected)
{
drop(data);
this.done = true;
this.stream.stop_sending(Code::H3_MESSAGE_ERROR);
return Poll::Ready(Some(Err(io::Error::other(
"HTTP/3 Content-Length exceeded",
))));
}
return Poll::Ready(Some(Ok(Frame::data(data.copy_to_bytes(length)))));
}
Poll::Ready(Ok(None)) => this.data_done = true,
}
}
match this.stream.poll_recv_trailers(cx) {
Poll::Pending => Poll::Pending,
Poll::Ready(Err(error)) => {
this.done = true;
Poll::Ready(Some(Err(io::Error::other(error.to_string()))))
}
Poll::Ready(Ok(trailers)) => {
this.done = true;
if this
.expected
.is_some_and(|expected| expected != this.received)
{
this.stream.stop_sending(Code::H3_MESSAGE_ERROR);
return Poll::Ready(Some(Err(io::Error::other(
"HTTP/3 Content-Length mismatch",
))));
}
Poll::Ready(trailers.map(|mut headers| {
crate::proxy::headers::strip_hop_by_hop(&mut headers);
Ok(Frame::trailers(headers))
}))
}
}
}
fn is_end_stream(&self) -> bool {
self.done
}
fn size_hint(&self) -> SizeHint {
// A declared zero is not proof that the peer ended its stream. Unknown
// lengths are verified by the cache's bounded empty-body probe.
let mut hint = SizeHint::new();
if self.done {
hint.set_exact(0);
} else if let Some(remaining) = self
.expected
.map(|n| n.saturating_sub(self.received))
.filter(|n| *n > 0)
{
hint.set_upper(remaining);
}
hint
}
}
impl Drop for ReceiveBody {
fn drop(&mut self) {
// Dropping the request stream cancels the receive side at QUIC level.
// Explicit STOP_SENDING is used only from poll_frame paths where the
// stream handle is available and a protocol error was detected.
}
}
include!("http3/body.rs");
#[cfg(test)]
include!("http3/tests.rs");