eggress_protocol_h3/
lib.rs

1//! HTTP/3 CONNECT protocol adapters over the optional QUIC transport.
2
3use std::sync::Arc;
4
5use base64::Engine;
6use bytes::{Buf, Bytes};
7use eggress_core::BoxStream;
8use eggress_transport_quic::{QuicClient, QuicConnection, QuicError};
9use http::{Request, Response, StatusCode};
10use subtle::ConstantTimeEq;
11use tokio::io::{AsyncReadExt, AsyncWriteExt};
12use tokio::sync::{Mutex, Semaphore};
13use tokio_util::sync::CancellationToken;
14
15/// Upper bound on concurrently active CONNECT requests per server connection.
16/// Each request holds one permit covering its handler plus its two relay
17/// helper tasks, so a client cannot multiply spawned tasks without limit.
18const MAX_ACTIVE_REQUESTS_PER_CONNECTION: usize = 256;
19
20/// HTTP/3 CONNECT errors.
21#[derive(Debug, thiserror::Error)]
22pub enum H3Error {
23    #[error("QUIC transport error: {0}")]
24    Quic(#[from] QuicError),
25    #[error("HTTP/3 connection error: {0}")]
26    Connection(String),
27    #[error("HTTP/3 stream error: {0}")]
28    Stream(String),
29    #[error("HTTP/3 CONNECT rejected with status {0}")]
30    Rejected(StatusCode),
31    #[error("HTTP/3 request authority is missing or invalid")]
32    InvalidAuthority,
33    #[error("HTTP/3 request is not CONNECT")]
34    InvalidMethod,
35}
36
37/// Details of an accepted H3 proxy request.
38#[derive(Debug, Clone)]
39pub struct H3Request {
40    pub authority: String,
41    pub headers: http::HeaderMap,
42}
43
44impl H3Request {
45    /// Parse the CONNECT authority into an Eggress target.
46    pub fn target(&self) -> Result<eggress_core::TargetAddr, H3Error> {
47        self.authority
48            .parse()
49            .map_err(|_| H3Error::InvalidAuthority)
50    }
51}
52
53/// An H3 client session sharing one QUIC connection across concurrent requests.
54pub struct H3Client {
55    quic: Arc<QuicClient>,
56    session: Mutex<Option<Arc<H3Session>>>,
57    authorization: Option<(String, String)>,
58}
59
60struct H3Session {
61    sender: h3::client::SendRequest<h3_quinn::OpenStreams, Bytes>,
62}
63
64impl H3Client {
65    pub fn new(quic: Arc<QuicClient>, authorization: Option<(String, String)>) -> Self {
66        Self {
67            quic,
68            session: Mutex::new(None),
69            authorization,
70        }
71    }
72
73    async fn session(&self) -> Result<Arc<H3Session>, H3Error> {
74        // Fast path under the lock; drop the guard before the QUIC dial +
75        // H3 handshake so slow handshakes do not serialize all request
76        // tasks (companion to transport-quic B-02).
77        if let Some(session) = self.session.lock().await.clone() {
78            return Ok(session);
79        }
80        let connection = self.quic.get_connection().await?;
81        let (mut driver, sender) = h3::client::new(connection.into_h3())
82            .await
83            .map_err(|e| H3Error::Connection(e.to_string()))?;
84        tokio::spawn(async move {
85            // `wait_idle` resolves to the terminal `ConnectionError` (including
86            // clean shutdown); log it so broken connections leave a trace.
87            let err = driver.wait_idle().await;
88            tracing::debug!(%err, "H3 connection driver terminated");
89        });
90        let session = Arc::new(H3Session { sender });
91        let mut guard = self.session.lock().await;
92        if let Some(existing) = &*guard {
93            return Ok(existing.clone());
94        }
95        *guard = Some(session.clone());
96        Ok(session)
97    }
98
99    /// Open a multiplexed HTTP/3 CONNECT stream.
100    ///
101    /// On transport-level failure the cached session and QUIC connection are
102    /// dropped and the request is retried once, mirroring the QUIC client's
103    /// reconnect-on-error recovery path. Without this, a session cached over
104    /// a connection killed by an idle timeout or GOAWAY would fail forever.
105    pub async fn connect(&self, target: &eggress_core::TargetAddr) -> Result<BoxStream, H3Error> {
106        match self.connect_once(target).await {
107            Err(error @ (H3Error::Quic(_) | H3Error::Connection(_) | H3Error::Stream(_))) => {
108                tracing::debug!(%error, "H3 connect failed; resetting cached session");
109                self.session.lock().await.take();
110                self.quic.reset_connection().await;
111                self.connect_once(target).await
112            }
113            result => result,
114        }
115    }
116
117    async fn connect_once(&self, target: &eggress_core::TargetAddr) -> Result<BoxStream, H3Error> {
118        let session = self.session().await?;
119        let authority = target.to_string();
120        let mut request = Request::builder()
121            .method(http::Method::CONNECT)
122            .uri(format!("https://{authority}/"));
123        if let Some((username, password)) = &self.authorization {
124            let encoded =
125                base64::engine::general_purpose::STANDARD.encode(format!("{username}:{password}"));
126            request = request.header(
127                http::header::PROXY_AUTHORIZATION,
128                format!("Basic {encoded}"),
129            );
130        }
131        let request = request
132            .body(())
133            .map_err(|e| H3Error::Connection(e.to_string()))?;
134        let mut stream = session
135            .sender
136            .clone()
137            .send_request(request)
138            .await
139            .map_err(|e| H3Error::Stream(e.to_string()))?;
140        stream
141            .finish()
142            .await
143            .map_err(|e| H3Error::Stream(e.to_string()))?;
144        let response = stream
145            .recv_response()
146            .await
147            .map_err(|e| H3Error::Stream(e.to_string()))?;
148        if response.status() != StatusCode::OK {
149            return Err(H3Error::Rejected(response.status()));
150        }
151        let (send, recv) = stream.split();
152        Ok(bridge_client_stream(send, recv))
153    }
154
155    pub async fn close(&self) {
156        self.session.lock().await.take();
157        self.quic.close();
158    }
159}
160
161fn h3_response(status: StatusCode) -> Response<()> {
162    let mut response = Response::new(());
163    *response.status_mut() = status;
164    response
165}
166
167fn h3_auth_required_response() -> Response<()> {
168    let mut response = h3_response(StatusCode::PROXY_AUTHENTICATION_REQUIRED);
169    response.headers_mut().insert(
170        http::header::PROXY_AUTHENTICATE,
171        http::HeaderValue::from_static("Basic realm=\"eggress\""),
172    );
173    response
174}
175
176/// Serve all H3 CONNECT requests on one established QUIC connection.
177pub async fn serve_connection<F, Fut>(
178    connection: QuicConnection,
179    cancel: CancellationToken,
180    authorization: Option<(String, String)>,
181    handler: F,
182) -> Result<(), H3Error>
183where
184    F: Fn(H3Request, BoxStream, std::net::SocketAddr) -> Fut + Send + Sync + Clone + 'static,
185    Fut: std::future::Future<Output = ()> + Send + 'static,
186{
187    let peer = connection.remote_address();
188    let request_permits = Arc::new(Semaphore::new(MAX_ACTIVE_REQUESTS_PER_CONNECTION));
189    let mut h3_connection = h3::server::builder()
190        .build(connection.into_h3())
191        .await
192        .map_err(|e| H3Error::Connection(e.to_string()))?;
193    loop {
194        let resolver = tokio::select! {
195            resolver = h3_connection.accept() => resolver,
196            _ = cancel.cancelled() => break,
197        }
198        .map_err(|e| H3Error::Connection(e.to_string()))?;
199        let Some(resolver) = resolver else { break };
200        let (request, mut stream) = resolver
201            .resolve_request()
202            .await
203            .map_err(|e| H3Error::Stream(e.to_string()))?;
204        if request.method() != http::Method::CONNECT {
205            let _ = stream
206                .send_response(h3_response(StatusCode::METHOD_NOT_ALLOWED))
207                .await;
208            let _ = stream.finish().await;
209            continue;
210        }
211        let authority = request
212            .uri()
213            .authority()
214            .map(|authority| authority.as_str().to_string())
215            .ok_or(H3Error::InvalidAuthority)?;
216        let request = H3Request {
217            authority,
218            headers: request.headers().clone(),
219        };
220        if let Some((username, password)) = &authorization {
221            let valid = request
222                .headers
223                .get(http::header::PROXY_AUTHORIZATION)
224                .and_then(|value| value.to_str().ok())
225                .and_then(parse_basic_authorization)
226                .is_some_and(|(user, pass)| {
227                    (user.as_bytes().ct_eq(username.as_bytes())
228                        & pass.as_bytes().ct_eq(password.as_bytes()))
229                    .unwrap_u8()
230                        == 1
231                });
232            if !valid {
233                let _ = stream.send_response(h3_auth_required_response()).await;
234                let _ = stream.finish().await;
235                continue;
236            }
237        }
238        // Bound active requests: acquire before responding so the handler
239        // task and its two relay helper tasks all live under one permit.
240        let permit = tokio::select! {
241            permit = request_permits.clone().acquire_owned() => permit,
242            _ = cancel.cancelled() => break,
243        };
244        stream
245            .send_response(h3_response(StatusCode::OK))
246            .await
247            .map_err(|e| H3Error::Stream(e.to_string()))?;
248        let (send, recv) = stream.split();
249        let local = bridge_server_stream(send, recv);
250        let handler = handler.clone();
251        tokio::spawn(async move {
252            let _permit = permit;
253            handler(request, local, peer).await;
254        });
255    }
256    Ok(())
257}
258
259fn parse_basic_authorization(value: &str) -> Option<(String, String)> {
260    let encoded = value.strip_prefix("Basic ")?;
261    let decoded = base64::engine::general_purpose::STANDARD
262        .decode(encoded)
263        .ok()
264        .and_then(|bytes| String::from_utf8(bytes).ok())?;
265    let (username, password) = decoded.split_once(':')?;
266    if username.contains(['\r', '\n', '\0', '\x7f'])
267        || password.contains(['\r', '\n', '\0', '\x7f'])
268    {
269        return None;
270    }
271    if username.len() > 4096 || password.len() > 4096 {
272        return None;
273    }
274    Some((username.to_string(), password.to_string()))
275}
276
277fn bridge_client_stream<S, R>(
278    mut send: h3::client::RequestStream<S, Bytes>,
279    mut recv: h3::client::RequestStream<R, Bytes>,
280) -> BoxStream
281where
282    S: h3::quic::SendStream<Bytes> + Send + 'static,
283    R: h3::quic::RecvStream + Send + 'static,
284{
285    let (application, peer) = tokio::io::duplex(64 * 1024);
286    let (mut application_reader, application_writer) = tokio::io::split(application);
287    let (peer_reader, mut peer_writer) = tokio::io::split(peer);
288    tokio::spawn(async move {
289        let mut buf = vec![0u8; 16 * 1024];
290        loop {
291            match application_reader.read(&mut buf).await {
292                Ok(0) => {
293                    let _ = send.finish().await;
294                    break;
295                }
296                Ok(n) => {
297                    if send
298                        .send_data(Bytes::copy_from_slice(&buf[..n]))
299                        .await
300                        .is_err()
301                    {
302                        break;
303                    }
304                }
305                Err(_) => break,
306            }
307        }
308    });
309    tokio::spawn(async move {
310        while let Ok(Some(mut data)) = recv.recv_data().await {
311            let bytes = data.copy_to_bytes(data.remaining());
312            if peer_writer.write_all(&bytes).await.is_err() {
313                return;
314            }
315        }
316        let _ = peer_writer.shutdown().await;
317    });
318    Box::new(tokio::io::join(peer_reader, application_writer))
319}
320
321fn bridge_server_stream<S, R>(
322    mut send: h3::server::RequestStream<S, Bytes>,
323    mut recv: h3::server::RequestStream<R, Bytes>,
324) -> BoxStream
325where
326    S: h3::quic::SendStream<Bytes> + Send + 'static,
327    R: h3::quic::RecvStream + Send + 'static,
328{
329    let (application, peer) = tokio::io::duplex(64 * 1024);
330    let (mut application_reader, application_writer) = tokio::io::split(application);
331    let (peer_reader, mut peer_writer) = tokio::io::split(peer);
332    tokio::spawn(async move {
333        let mut buf = vec![0u8; 16 * 1024];
334        loop {
335            match application_reader.read(&mut buf).await {
336                Ok(0) => {
337                    let _ = send.finish().await;
338                    break;
339                }
340                Ok(n) => {
341                    if send
342                        .send_data(Bytes::copy_from_slice(&buf[..n]))
343                        .await
344                        .is_err()
345                    {
346                        break;
347                    }
348                }
349                Err(_) => break,
350            }
351        }
352    });
353    tokio::spawn(async move {
354        while let Ok(Some(mut data)) = recv.recv_data().await {
355            let bytes = data.copy_to_bytes(data.remaining());
356            if peer_writer.write_all(&bytes).await.is_err() {
357                return;
358            }
359        }
360        let _ = peer_writer.shutdown().await;
361    });
362    Box::new(tokio::io::join(peer_reader, application_writer))
363}
364
365#[cfg(test)]
366mod tests {
367    use super::*;
368    #[cfg(feature = "insecure-quic")]
369    use eggress_transport_quic::{QuicClient, QuicClientConfig, QuicListener, QuicServerConfig};
370    #[cfg(feature = "insecure-quic")]
371    use rcgen::{CertificateParams, KeyPair};
372    #[cfg(feature = "insecure-quic")]
373    use std::time::Duration;
374    #[cfg(feature = "insecure-quic")]
375    use tokio::io::{AsyncReadExt, AsyncWriteExt};
376    #[cfg(feature = "insecure-quic")]
377    use tokio_util::sync::CancellationToken;
378
379    #[test]
380    fn basic_authorization_is_deterministic() {
381        assert_eq!(
382            base64::engine::general_purpose::STANDARD.encode("user:pass"),
383            "dXNlcjpwYXNz"
384        );
385    }
386
387    #[test]
388    fn h3_request_target_parses_authority() {
389        let request = H3Request {
390            authority: "example.com:443".to_string(),
391            headers: http::HeaderMap::new(),
392        };
393        let target = request.target().unwrap();
394        assert_eq!(target.port, 443);
395
396        let bad = H3Request {
397            authority: "not a target".to_string(),
398            headers: http::HeaderMap::new(),
399        };
400        assert!(matches!(bad.target(), Err(H3Error::InvalidAuthority)));
401    }
402
403    #[test]
404    fn h3_basic_authorization_parses_and_rejects_controls() {
405        let encoded = base64::engine::general_purpose::STANDARD.encode("alice:s3cret");
406        let (user, pass) = parse_basic_authorization(&format!("Basic {encoded}")).unwrap();
407        assert_eq!(user, "alice");
408        assert_eq!(pass, "s3cret");
409        assert!(parse_basic_authorization("Bearer token").is_none());
410        assert!(parse_basic_authorization("Basic !!!").is_none());
411    }
412
413    #[cfg(feature = "insecure-quic")]
414    #[tokio::test]
415    async fn h3_connect_stream_round_trips_over_quic() {
416        let params = CertificateParams::new(vec!["localhost".to_string()]).unwrap();
417        let key = KeyPair::generate().unwrap();
418        let certificate = params.self_signed(&key).unwrap();
419        let listener = QuicListener::bind(
420            "127.0.0.1:0".parse().unwrap(),
421            QuicServerConfig {
422                certificate_pem: certificate.pem().into_bytes(),
423                private_key_pem: key.serialize_pem().into_bytes(),
424                idle_timeout: Duration::from_secs(60),
425                max_concurrent_streams: 16,
426                alpn_protocols: vec![b"h3".to_vec()],
427            },
428        )
429        .await
430        .unwrap();
431        let cancel = CancellationToken::new();
432        let server = listener.clone();
433        let server_cancel = cancel.clone();
434        let server_task = tokio::spawn(async move {
435            let connection = server
436                .accept_connection(&server_cancel)
437                .await
438                .unwrap()
439                .unwrap();
440            serve_connection(
441                connection,
442                server_cancel,
443                None,
444                |_, mut stream, _| async move {
445                    let mut data = [0u8; 5];
446                    stream.read_exact(&mut data).await.unwrap();
447                    stream.write_all(&data).await.unwrap();
448                },
449            )
450            .await
451            .unwrap();
452        });
453
454        let address = listener.local_addr().unwrap();
455        let client = QuicClient::connect(
456            "127.0.0.1",
457            address.port(),
458            QuicClientConfig {
459                insecure: true,
460                alpn_protocols: vec![b"h3".to_vec()],
461                ..Default::default()
462            },
463        )
464        .await
465        .unwrap();
466        let mut stream = H3Client::new(client, None)
467            .connect(&"example.com:443".parse().unwrap())
468            .await
469            .unwrap();
470        stream.write_all(b"hello").await.unwrap();
471        let mut output = [0u8; 5];
472        stream.read_exact(&mut output).await.unwrap();
473        assert_eq!(&output, b"hello");
474
475        cancel.cancel();
476        server_task.await.unwrap();
477    }
478}
eggress_protocol_h3/lib.rs

eggress_protocol_h3/
lib.rs
