1use bytes::{Buf, Bytes};
2use sova_core::extend::{wait_shutdown, BoxFuture as SovaBoxFuture, StateMap};
3use sova_core::{
4 BackgroundService, Error, Request as SovaRequest, Result, Server as SovaServer, Shutdown,
5};
6use std::net::SocketAddr;
7use std::path::Path;
8use std::sync::Arc;
9use tokio::task::JoinSet;
10
11use crate::config::build_server_config;
12use crate::endpoint::{accept_handshake, bind_server, prepare_incoming, PrepareIncoming};
13
14pub struct Http3Service {
17 bind_addr: SocketAddr,
18 tls: Arc<sova_core::Tls>,
19 alpn: Vec<Vec<u8>>,
20 server: SovaServer,
21 name: String,
22}
23
24impl Http3Service {
25 pub fn from_tls(
27 bind_addr: SocketAddr,
28 tls: sova_core::Tls,
29 alpn: Vec<Vec<u8>>,
30 server: SovaServer,
31 ) -> Result<Self> {
32 if alpn.is_empty() {
33 return Err(Error::Internal("Http3Service: `alpn` must not be empty".into()));
34 }
35 Ok(Self {
36 bind_addr,
37 tls: Arc::new(tls),
38 alpn,
39 server,
40 name: format!("h3:{bind_addr}"),
41 })
42 }
43
44 pub fn from_pem(
45 bind_addr: SocketAddr,
46 cert_path: impl AsRef<Path>,
47 key_path: impl AsRef<Path>,
48 alpn: Vec<Vec<u8>>,
49 server: SovaServer,
50 ) -> Result<Self> {
51 let tls = sova_core::Tls::from_pem(cert_path, key_path)?;
52 Self::from_tls(bind_addr, tls, alpn, server)
53 }
54
55 #[cfg(feature = "dev-tls")]
56 pub fn self_signed(
57 bind_addr: SocketAddr,
58 hosts: &[&str],
59 alpn: Vec<Vec<u8>>,
60 server: SovaServer,
61 ) -> Result<Self> {
62 let tls = sova_core::Tls::self_signed(hosts)?;
63 Self::from_tls(bind_addr, tls, alpn, server)
64 }
65
66 pub fn install(self, app: &mut sova_core::App) {
68 app.use_tls((*self.tls).clone());
69 app.service(self);
70 }
71}
72
73impl BackgroundService for Http3Service {
74 fn name(&self) -> &str {
75 &self.name
76 }
77
78 fn run(
79 self: Box<Self>,
80 _state: Arc<StateMap>,
81 shutdown: Shutdown,
82 ) -> SovaBoxFuture<()> {
83 Box::pin(async move {
84 let server_config = match build_server_config(&self.tls, self.alpn.clone(), false) {
85 Ok(v) => v,
86 Err(e) => {
87 tracing::error!(error = %e, "h3 server TLS config build failed");
88 return;
89 }
90 };
91
92 let endpoint = match bind_server(server_config, self.bind_addr, "h3") {
93 Some(ep) => ep,
94 None => return,
95 };
96
97 tracing::info!(addr = %self.bind_addr, "h3 listening");
98
99 let mut conn_tasks: JoinSet<()> = JoinSet::new();
100 loop {
101 tokio::select! {
102 _ = wait_shutdown(shutdown.clone()) => break,
103 incoming = endpoint.accept() => {
104 let Some(incoming) = incoming else { break; };
105 let incoming = match prepare_incoming(incoming) {
106 PrepareIncoming::RetrySent => continue,
107 PrepareIncoming::Ready(i) => *i,
108 };
109
110 let conn = match accept_handshake(incoming, "h3 quic handshake failed").await {
111 Some(c) => c,
112 None => continue,
113 };
114
115 let app_server = self.server.clone();
116 let shutdown = shutdown.clone();
117 conn_tasks.spawn(async move {
118 handle_h3_connection(conn, app_server, shutdown).await;
119 });
120 }
121 }
122 }
123
124 endpoint.close(0u8.into(), b"shutdown");
125 while let Some(res) = conn_tasks.join_next().await {
126 if let Err(e) = res {
127 tracing::debug!(error = %e, "h3 conn task join error");
128 }
129 }
130 })
131 }
132}
133
134async fn handle_h3_connection(conn: quinn::Connection, app_server: SovaServer, shutdown: Shutdown) {
135 let mut h3_conn = match h3::server::builder()
136 .build(h3_quinn::Connection::new(conn))
137 .await
138 {
139 Ok(c) => c,
140 Err(e) => {
141 tracing::debug!(error = %e, "h3 connection build failed");
142 return;
143 }
144 };
145
146 loop {
147 tokio::select! {
148 _ = wait_shutdown(shutdown.clone()) => break,
149 accepted = h3_conn.accept() => {
150 match accepted {
151 Ok(Some(resolver)) => {
152 let app_server = app_server.clone();
153 tokio::spawn(async move {
154 match resolver.resolve_request().await {
155 Ok((req, mut stream)) => {
156 let mut body = Vec::new();
157 loop {
158 match stream.recv_data().await {
159 Ok(Some(mut chunk)) => {
160 while chunk.has_remaining() {
161 let bytes = chunk.chunk();
162 body.extend_from_slice(bytes);
163 let n = bytes.len();
164 chunk.advance(n);
165 }
166 }
167 Ok(None) => break,
168 Err(e) => {
169 tracing::debug!(error = %e, "h3 recv_data failed");
170 return;
171 }
172 }
173 }
174
175 let path = if let Some(q) = req.uri().query() {
176 format!("{}?{q}", req.uri().path())
177 } else {
178 req.uri().path().to_string()
179 };
180 let mut sova_req = SovaRequest::builder()
181 .method(req.method().clone())
182 .path(path)
183 .scheme("https")
184 .host(
185 req.headers()
186 .get(http::header::HOST)
187 .and_then(|v| v.to_str().ok())
188 .unwrap_or("localhost"),
189 )
190 .body(Bytes::from(body))
191 .build();
192 for (name, value) in req.headers() {
193 sova_req.headers.insert(name.clone(), value.clone());
194 }
195
196 let mut res = app_server.handle(sova_req).await;
197 let status = res.status_code();
198 let headers = res.headers().clone();
199 let body = match res.take_body().collect().await {
200 Ok(b) => b,
201 Err(e) => {
202 tracing::error!(error = %e, "h3 response body collect failed");
203 Bytes::from_static(b"internal error")
204 }
205 };
206
207 let mut resp_builder = http::Response::builder().status(status);
208 for (name, value) in &headers {
209 if name == http::header::CONNECTION
210 || name == http::header::TRANSFER_ENCODING
211 || name == http::header::UPGRADE
212 || name.as_str().eq_ignore_ascii_case("keep-alive")
213 {
214 continue;
215 }
216 resp_builder = resp_builder.header(name, value);
217 }
218 let response = match resp_builder.body(()) {
219 Ok(r) => r,
220 Err(e) => {
221 tracing::error!(error = %e, "h3 response build failed");
222 return;
223 }
224 };
225
226 if stream.send_response(response).await.is_err() {
227 return;
228 }
229 if !body.is_empty() && stream.send_data(body).await.is_err() {
230 return;
231 }
232 let _ = stream.finish().await;
233 }
234 Err(e) => tracing::debug!(error = %e, "h3 resolve request failed"),
235 }
236 });
237 }
238 Ok(None) => break,
239 Err(e) => {
240 tracing::debug!(error = %e, "h3 accept failed");
241 break;
242 }
243 }
244 }
245 }
246 }
247}
248
249#[cfg(test)]
250mod tests {
251 use super::*;
252 use crate::config::SkipServerVerification;
253 use h3::client;
254 use sova_core::extend::StateMap;
255 use sova_core::{App, Request, Response};
256 use rcgen::generate_simple_self_signed;
257 use std::fs;
258 use tempfile::TempDir;
259
260 fn unused_local_addr() -> std::net::SocketAddr {
261 let sock = std::net::UdpSocket::bind("127.0.0.1:0").expect("bind udp");
262 sock.local_addr().expect("local addr")
263 }
264
265 fn write_self_signed_pem(tmp: &TempDir, subject: &[&str]) -> (std::path::PathBuf, std::path::PathBuf) {
266 let subject_alt_names: Vec<String> = subject.iter().map(|s| (*s).into()).collect();
267 let cert = generate_simple_self_signed(subject_alt_names).expect("rcgen");
268 let cert_pem = cert.cert.pem();
269 let key_pem = cert.key_pair.serialize_pem();
270
271 let cert_path = tmp.path().join("cert.pem");
272 let key_path = tmp.path().join("key.pem");
273 fs::write(&cert_path, cert_pem).expect("write cert");
274 fs::write(&key_path, key_pem).expect("write key");
275
276 (cert_path, key_path)
277 }
278
279 #[tokio::test]
280 async fn h3_service_dispatches_sova_routes() {
281 let bind_addr = unused_local_addr();
282 let client_bind = unused_local_addr();
283 let h3_alpn = vec![b"h3".to_vec()];
284
285 let tmp = TempDir::new().unwrap();
286 let (cert_path, key_path) = write_self_signed_pem(&tmp, &["localhost"]);
287
288 let mut app = App::new();
289 app.get("/h3", |_r: Request| async { Response::text("h3-ok") });
290 let server = app.build().unwrap();
291
292 let svc = Http3Service::from_pem(
293 bind_addr,
294 &cert_path,
295 &key_path,
296 h3_alpn.clone(),
297 server,
298 )
299 .unwrap();
300
301 let (tx, shutdown) = sova_core::shutdown_channel();
302 let handle = tokio::spawn(Box::new(svc).run(Arc::new(StateMap::new()), shutdown));
303 tokio::time::sleep(std::time::Duration::from_millis(50)).await;
304
305 let mut client_crypto = rustls::ClientConfig::builder()
306 .dangerous()
307 .with_custom_certificate_verifier(Arc::new(SkipServerVerification))
308 .with_no_client_auth();
309 client_crypto.alpn_protocols = h3_alpn;
310 client_crypto.enable_early_data = false;
311 let client_cfg = quinn::ClientConfig::new(Arc::new(
312 quinn::crypto::rustls::QuicClientConfig::try_from(client_crypto).unwrap(),
313 ));
314
315 let mut endpoint = quinn::Endpoint::client(client_bind).unwrap();
316 endpoint.set_default_client_config(client_cfg);
317 let conn = endpoint.connect(bind_addr, "localhost").unwrap().await.unwrap();
318
319 let (mut driver, mut send_request) = client::new(h3_quinn::Connection::new(conn))
320 .await
321 .unwrap();
322 let drive = tokio::spawn(async move {
323 let _ = std::future::poll_fn(|cx| driver.poll_close(cx)).await;
324 });
325
326 let req = http::Request::builder()
327 .method(http::Method::GET)
328 .uri("https://localhost/h3")
329 .body(())
330 .unwrap();
331 let mut stream = send_request.send_request(req).await.unwrap();
332 stream.finish().await.unwrap();
333 let _resp = stream.recv_response().await.unwrap();
334 let mut body = Vec::new();
335 while let Some(mut chunk) = stream.recv_data().await.unwrap() {
336 while chunk.has_remaining() {
337 let bytes = chunk.chunk();
338 body.extend_from_slice(bytes);
339 let n = bytes.len();
340 chunk.advance(n);
341 }
342 }
343 assert_eq!(body, b"h3-ok");
344
345 drop(send_request);
346 let _ = drive.await;
347 let _ = tx.send(true);
348 let _ = handle.await;
349 }
350}