RustProxy/src/http3.rs

use crate::{
    metrics::Metrics,
    runtime::App,
    server::{LocalAddr, PeerAddr, ServerShutdown, Transport},
    vhost::Listener,
};
use anyhow::{ensure, Context as _, Result};
use axum::{body::Body, Router};
use bytes::{Buf, Bytes};
use h3::error::Code;
use http::{HeaderMap, HeaderValue, Method, Version};
use http_body::{Body as HttpBody, Frame, SizeHint};
use http_body_util::BodyExt;
use serde::{Deserialize, Serialize};
use std::{
    io,
    net::{SocketAddr, UdpSocket},
    pin::Pin,
    sync::{atomic::Ordering, Arc},
    task::{Context, Poll},
    time::Duration,
};
use tokio::{
    sync::{OwnedSemaphorePermit, Semaphore},
    task::JoinSet,
};
use tower::ServiceExt;
include!("http3/settings.rs");
pub fn crypto(
    tls: &Arc<rustls::ServerConfig>,
    policy: &crate::tls::ProtocolPolicy,
) -> Result<Arc<quinn::crypto::rustls::QuicServerConfig>> {
    ensure!(
        policy.allows_http3(),
        "HTTP/3 requires TLS 1.3 and a TLS 1.3 cipher suite"
    );
    let provider = crate::tls::provider_for(policy)?;
    let mut quic = rustls::ServerConfig::builder_with_provider(Arc::new(provider))
        .with_protocol_versions(&[&rustls::version::TLS13])?
        .with_no_client_auth()
        .with_cert_resolver(tls.cert_resolver.clone());
    quic.alpn_protocols = vec![b"h3".to_vec()];
    quic.max_early_data_size = 0;
    Ok(Arc::new(quinn::crypto::rustls::QuicServerConfig::try_from(
        quic,
    )?))
}
pub struct Bound {
    endpoint: quinn::Endpoint,
    server_config: quinn::ServerConfig,
    address: SocketAddr,
    settings: Settings,
}
fn requested_socket_buffer_size(effective_bytes: u32) -> usize {
    #[cfg(target_os = "linux")]
    {
        (effective_bytes as usize / 2).max(1)
    }
    #[cfg(not(target_os = "linux"))]
    {
        effective_bytes as usize
    }
}

fn configure_udp_socket(socket: &UdpSocket, address: SocketAddr, settings: Settings) {
    let socket = socket2::SockRef::from(socket);
    for (direction, requested, result) in [
        (
            "receive",
            settings.udp_receive_buffer_bytes,
            if settings.udp_receive_buffer_bytes == 0 {
                Ok(())
            } else {
                socket.set_recv_buffer_size(requested_socket_buffer_size(
                    settings.udp_receive_buffer_bytes,
                ))
            },
        ),
        (
            "send",
            settings.udp_send_buffer_bytes,
            if settings.udp_send_buffer_bytes == 0 {
                Ok(())
            } else {
                socket.set_send_buffer_size(requested_socket_buffer_size(
                    settings.udp_send_buffer_bytes,
                ))
            },
        ),
    ] {
        if let Err(error) = result {
            tracing::warn!(%address, direction, requested, %error,
                "HTTP/3 UDP buffer request failed; continuing with OS socket default");
        }
    }
    let receive = socket.recv_buffer_size().ok();
    let send = socket.send_buffer_size().ok();
    tracing::info!(%address, ?receive, ?send,
        "HTTP/3 UDP effective socket buffers (bytes)");
    for (direction, requested, actual) in [
        ("receive", settings.udp_receive_buffer_bytes, receive),
        ("send", settings.udp_send_buffer_bytes, send),
    ] {
        if requested != 0 && actual.is_some_and(|size| size < requested as usize) {
            tracing::warn!(%address, direction, requested, ?actual,
                "HTTP/3 UDP socket buffer below configured target; check net.core.rmem_max and wmem_max on Linux");
        }
    }
}

pub fn bind(listener: Listener, state: &App) -> Result<Bound> {
    let settings = listener
        .http3
        .context("HTTP/3 is disabled on this listener")?;
    settings.validate()?;
    ensure!(listener.tls, "HTTP/3 listener requires TLS");
    let crypto = state
        .config
        .load()
        .quic_tls
        .clone()
        .context("HTTP/3 certificate unavailable")?;
    let mut server_config = quinn::ServerConfig::with_crypto(crypto);
    server_config
        .transport_config(Arc::new(settings.transport()?))
        .migration(false)
        .max_incoming(settings.max_concurrent_connections as usize)
        .incoming_buffer_size(64 * 1024)
        .incoming_buffer_size_total(1024 * 1024);
    let socket = UdpSocket::bind(listener.address)
        .with_context(|| format!("cannot bind HTTP/3 UDP {}", listener.address))?;
    configure_udp_socket(&socket, listener.address, settings);
    let endpoint = quinn::Endpoint::new(
        quinn::EndpointConfig::default(),
        Some(server_config.clone()),
        socket,
        Arc::new(quinn::TokioRuntime),
    )
    .with_context(|| format!("cannot initialize HTTP/3 UDP {}", listener.address))?;
    Ok(Bound {
        endpoint,
        server_config,
        address: listener.address,
        settings,
    })
}
fn desired_for_bound(address: SocketAddr, tls: bool, desired: &[Listener]) -> Option<Settings> {
    desired
        .iter()
        .find(|listener| listener.address == address && listener.tls == tls)
        .and_then(|listener| listener.http3)
}

pub fn listeners_require_restart(active: &[Listener], desired: &[Listener]) -> bool {
    if active.len() != desired.len() {
        return true;
    }
    active.iter().any(|bound| {
        let Some(next) = desired.iter().find(|next| next.address == bound.address) else {
            return true;
        };
        if bound.tls != next.tls
            || bound.http2 != next.http2
            || bound.http2_settings != next.http2_settings
            || bound.h2c != next.h2c
        {
            return true;
        }
        match (bound.http3, next.http3) {
            (None, None) => false,
            (Some(initial), Some(updated)) => {
                updated.max_concurrent_connections > initial.max_concurrent_connections
                    || updated.udp_receive_buffer_bytes != initial.udp_receive_buffer_bytes
                    || updated.udp_send_buffer_bytes != initial.udp_send_buffer_bytes
            }
            _ => true,
        }
    })
}

pub fn withdraw_advertisement(
    headers: &mut HeaderMap,
    local: Option<SocketAddr>,
    tls: bool,
    active: &[Listener],
) {
    headers.remove(::http::header::ALT_SVC);
    if tls
        && local.is_some_and(|local| {
            active
                .iter()
                .any(|binding| binding.matches(local, true) && binding.http3.is_some())
        })
    {
        headers.insert(::http::header::ALT_SVC, HeaderValue::from_static("clear"));
    }
}

pub fn advertise(
    headers: &mut HeaderMap,
    local: Option<SocketAddr>,
    tls: bool,
    active: &[Listener],
    desired: &[Listener],
    host_http3_enabled: bool,
) {
    headers.remove(::http::header::ALT_SVC);
    if !tls {
        return;
    }
    let Some(local) = local else {
        return;
    };
    let Some(bound) = active.iter().find(|binding| binding.matches(local, true)) else {
        return;
    };
    if bound.http3.is_none() {
        return;
    }
    let settings = desired_for_bound(bound.address, bound.tls, desired);
    if !host_http3_enabled || settings.is_none() {
        withdraw_advertisement(headers, Some(local), true, active);
        return;
    }
    if let Some(settings) = settings {
        let value = format!("h3=\":{}\"; ma={}", local.port(), settings.alt_svc_max_age);
        if let Ok(value) = HeaderValue::from_str(&value) {
            headers.insert(::http::header::ALT_SVC, value);
        }
    }
}
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,
    ));
    let mut tasks = JoinSet::new();
    loop {
        tokio::select! {
            biased;
            _=state.shutdown.cancelled()=>break,
            _=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 desired = {
                    let snapshot = state.config.load();
                    desired_for_bound(bound.address, true, &snapshot.listeners)
                };
                let Some(settings) = desired else { incoming.refuse(); continue; };
                let open = bound.settings.max_concurrent_connections as usize - slots.available_permits();
                let capacity = (settings.max_concurrent_connections as usize)
                    .min(bound.settings.max_concurrent_connections as usize);
                if open >= capacity {
                    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 configured connection cap reached",
                        peer, bound.address, "quic", "3", 0, None,
                    );
                    incoming.refuse();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();
                let transport = match settings.transport() {
                    Ok(transport) => transport,
                    Err(error) => {
                        state.logs.push_protocol_error(
                            "http3_transport_config_error", error,
                            peer, bound.address, "quic", "3", 0, None,
                        );
                        incoming.refuse();continue;
                    }
                };
                config.transport = Arc::new(transport);
                let Some(crypto)=state.config.load().quic_tls.clone() else {incoming.refuse();continue;};
                config.crypto=crypto;
                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=settings.max_concurrent_bidistream as usize;
                tasks.spawn(async move {
                    let connection=tokio::select! {
                        _=state.shutdown.cancelled()=>return,
                        result=tokio::time::timeout(Duration::from_secs(10),connecting)=>match result {
                            Ok(Ok(connection))=>connection,
                            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;
                            },
                        }
                    };
                    let local=match connection.local_ip() {
                        Some(ip)=>SocketAddr::new(ip,address.port()),
                        None if !address.ip().is_unspecified()=>address,
                        None=>{connection.close(0x102u32.into(),b"local destination unavailable");return;},
                    };
                    let tls_sni=connection_sni(&connection);
                    state.metrics.connections.fetch_add(1,Ordering::Relaxed);
                    state.metrics.active_connections.fetch_add(1,Ordering::Relaxed);
                    state.metrics.http3_connections.fetch_add(1,Ordering::Relaxed);
                    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.clone(),limit,grace,tls_sni.clone()).await {
                        log_protocol_error(
                            &state, "http3_connection_error", &error, peer, local, 0, None, None, tls_sni,
                        );
                        tracing::debug!(%error,"HTTP/3 connection ended");
                    }
                });
            }
        }
    }
    let _ = tokio::time::timeout(grace, async { while tasks.join_next().await.is_some() {} }).await;
    tasks.abort_all();
    while tasks.join_next().await.is_some() {}
    bound.endpoint.close(0x100u32.into(), b"shutdown");
    let _ = tokio::time::timeout(Duration::from_secs(1), bound.endpoint.wait_idle()).await;
    Ok(())
}
async fn connection_loop(
    connection: quinn::Connection,
    local: SocketAddr,
    router: Router,
    state: Arc<App>,
    limit: usize,
    grace: Duration,
    tls_sni: Option<String>,
) -> Result<()> {
    let mut builder = h3::server::builder();
    builder.max_field_section_size(32 * 1024);
    builder.enable_extended_connect(state.config.load().websocket_rfc9220_on(local));
    let mut h3 = builder
        .build(h3::quinn::Connection::new(connection.clone()))
        .await?;
    let mut requests = JoinSet::new();
    let mut draining = false;
    let deadline = tokio::time::sleep(Duration::from_secs(86400));
    tokio::pin!(deadline);
    loop {
        tokio::select! {
            biased;
            _=state.shutdown.cancelled(),if !draining=>{
                draining=true;deadline.as_mut().reset(tokio::time::Instant::now()+grace);
                h3.shutdown(0).await?;
            },
            _=&mut deadline,if draining=>break,
            _=requests.join_next(),if !requests.is_empty()=>{},
            next=h3.accept()=>match next {
                Ok(Some(resolver))=>{
                    if requests.len()>=limit {
                        log_protocol_error(
                            &state, "http3_stream_limit_rejected", "HTTP/3 concurrent request stream limit reached",
                            connection.remote_address(), local, 0, None, None, tls_sni.clone(),
                        );
                        drop(resolver);continue;
                    }
                    let router=router.clone();let state=state.clone();let peer=connection.remote_address();let tls_sni=tls_sni.clone();
                    let quic_diagnostics = connection.clone();
                    requests.spawn(async move {
                        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))=>{
                                if client_cancelled_request(&error) {
                                    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(_)=>{
                                log_protocol_error(
                                    &state, "http3_header_timeout", "HTTP/3 request header timeout",
                                    peer, local, 408, None, None, tls_sni.clone(),
                                );
                                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();
                        let stream_id=format!("{:?}",send.send_id());
                        let diagnostic_method=request.method().as_str().to_owned();
                        let diagnostic_path=request.uri().path().chars().take(160).collect::<String>();
                        let (diagnostic_authority, diagnostic_host)=request_identity(&request);
                        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(
                                &state, "http3_invalid_headers", "HTTP/3 request contains forbidden connection-specific headers",
                                peer, local, 400, authority, host, tls_sni.clone(),
                            );
                            send.stop_stream(Code::H3_MESSAGE_ERROR);return;
                        }
                        let expected=request.headers().get(::http::header::CONTENT_LENGTH).and_then(|v|v.to_str().ok()).and_then(|v|v.parse::<u64>().ok());
                        let mut tunnel_receive=None;
                        let mut request=if websocket {
                            tunnel_receive=Some(receive);
                            request.map(|_|Body::empty())
                        } else if extended_connect {
                            drop(receive);
                            request.map(|_|Body::empty())
                        } else {
                            request.map(|_|Body::new(ReceiveBody {stream:receive,expected,received:0,data_done:false,done:false}))
                        };
                        request.extensions_mut().insert(PeerAddr(peer));request.extensions_mut().insert(LocalAddr(local));
                        request.extensions_mut().insert(Transport {tls:true});
                        request.extensions_mut().insert(ServerShutdown(state.shutdown.clone()));
                        let response=match router.oneshot(request).await {Ok(response)=>response,Err(never)=>match never {}};
                        let (mut parts,mut body)=response.into_parts();
                        let tunnel=parts.extensions.remove::<crate::proxy::websocket::H3Tunnel>();
                        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);
                                    return Err(anyhow::anyhow!("HTTP/3 tunnel response without a WebSocket Extended CONNECT request"));
                                };
                                drop(body);
                                if let Err(error)=crate::proxy::websocket::run_h3(state.clone(),&mut send,&mut receive,tunnel).await {
                                    tracing::debug!(%error,"HTTP/3 WebSocket tunnel closed");
                                }
                                return Ok::<(),anyhow::Error>(());
                            }
                            if let Some(mut receive)=tunnel_receive.take() {
                                receive.stop_sending(Code::H3_REQUEST_CANCELLED);
                            }
                            while let Some(frame)=body.frame().await {
                                let frame=frame?;
                                match frame.into_data() {
                                    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 {
                            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 {
                                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);
                        }
                        if sent_body_bytes >= 8 * 1024 * 1024 || expected.is_some_and(|n| n >= 8 * 1024 * 1024) {
                            let stats = quic_diagnostics.stats();
                            tracing::debug!(
                                %peer, local=%local, stream_id=%stream_id,
                                response_bytes=sent_body_bytes,
                                request_content_length=?expected,
                                quic_udp_tx_datagrams=stats.udp_tx.datagrams,
                                quic_udp_tx_ios=stats.udp_tx.ios,
                                quic_udp_rx_datagrams=stats.udp_rx.datagrams,
                                quic_sent_packets=stats.path.sent_packets,
                                quic_lost_packets=stats.path.lost_packets,
                                quic_congestion_events=stats.path.congestion_events,
                                quic_rtt_ms=stats.path.rtt.as_secs_f64() * 1000.0,
                                quic_cwnd_bytes=stats.path.cwnd,
                                quic_path_mtu=stats.path.current_mtu,
                                "HTTP/3 bulk transfer transport snapshot"
                            );
                        }
                    });
                },
                Ok(None)=>break,
                Err(error)=>{
                    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;
                },
            }
        }
    }
    let _ = tokio::time::timeout(grace, async {
        while requests.join_next().await.is_some() {}
    })
    .await;
    requests.abort_all();
    while requests.join_next().await.is_some() {}
    if draining {
        let _ = tokio::time::timeout(grace, connection.closed()).await;
    }
    Ok(())
}
include!("http3/identity.rs");
include!("http3/diagnostics.rs");
include!("http3/body.rs");

#[cfg(test)]
include!("http3/tests.rs");