daqbuf/daqbuffer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

h3 handshake error

Browse Source
This commit is contained in:
Dominik Werder
2025-02-19 15:22:49 +01:00
parent 0e20b2e630
commit df67384b77
1 changed files with 107 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
crates/httpret/src/http3.rs
View File

@@ -1,23 +1,49 @@
use bytes::Bytes;
use h3::server::RequestStream;
use h3_quinn::quinn::crypto::rustls::QuicServerConfig;
use h3_quinn::BidiStream;
use http::Request;
use http::StatusCode;
use quinn;
use quinn::crypto::rustls::QuicServerConfig;
use quinn::Endpoint;
use quinn::EndpointConfig;
use quinn::Incoming;
use rustls::pki_types::pem::PemObject;
use rustls::server::ProducesTickets;
use std::future::Future;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::task::Context;
use std::task::Poll;
use std::time::Duration;
use taskrun::tokio;
const EARLY_DATA_MAX: u32 = u32::MAX;
const EARLY_DATA_MAX: u32 = u32::MAX * 0;
macro_rules! info { ($($arg:expr),*) => ( if true { netpod::log::info!($($arg),*); } ); }
#[derive(Debug)]
struct TicketerCustom {}
impl ProducesTickets for TicketerCustom {
fn enabled(&self) -> bool {
false
}
fn lifetime(&self) -> u32 {
60 * 60 * 24
}
fn encrypt(&self, plain: &[u8]) -> Option<Vec<u8>> {
todo!()
}
fn decrypt(&self, cipher: &[u8]) -> Option<Vec<u8>> {
todo!()
}
}
autoerr::create_error_v1!(
name(Error, "Http3Support"),
enum variants {
@@ -29,6 +55,7 @@ autoerr::create_error_v1!(
Rustls(#[from] rustls::Error),
NoInitialCipherSuite(#[from] quinn::crypto::rustls::NoInitialCipherSuite),
QuinnConnection(#[from] quinn::ConnectionError),
RingInstallDefault,
},
);
@@ -51,20 +78,32 @@ impl Http3Support {
async fn new(bind_addr: SocketAddr) -> Result<Self, Error> {
let key = PemObject::from_pem_file("key.pem")?;
let cert = PemObject::from_pem_file("cert.pem")?;
let mut provider = rustls::crypto::ring::default_provider();
info!("provider default {:?}", provider);
provider.cipher_suites = rustls::crypto::ring::ALL_CIPHER_SUITES.to_vec();
provider.kx_groups = rustls::crypto::ring::ALL_KX_GROUPS.to_vec();
info!("provider custom {:?}", provider);
provider.install_default().map_err(|_| Error::RingInstallDefault)?;
let mut tls_conf = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(vec![cert], key)?;
tls_conf.alpn_protocols = vec![b"HTTP/3".to_vec(), b"h3".to_vec()];
tls_conf.alpn_protocols = vec![b"h3".to_vec()];
tls_conf.max_early_data_size = EARLY_DATA_MAX;
tls_conf.ticketer = Arc::new(TicketerCustom {});
let tls_conf = tls_conf;
let v = QuicServerConfig::try_from(tls_conf)?;
let quic_conf = Arc::new(v);
let conf_srv = quinn::ServerConfig::with_crypto(quic_conf);
let mut conf_srv = quinn::ServerConfig::with_crypto(quic_conf);
let mut transport = quinn::TransportConfig::default();
transport.max_idle_timeout(Some(quinn::VarInt::from_u32(1000 * 10).into()));
transport.keep_alive_interval(Some(Duration::from_millis(1000 * 2)));
transport.send_window(quinn::VarInt::from_u32(1024 * 1024 * 8).into());
transport.receive_window(quinn::VarInt::from_u32(1024 * 1024 * 8).into());
transport.max_concurrent_bidi_streams(quinn::VarInt::from_u32(100).into());
let transport = Arc::new(transport);
conf_srv.transport_config(transport);
let ep2 = Endpoint::server(conf_srv, bind_addr)?;
{
let ep = ep2.clone();
tokio::task::spawn(Self::accept(ep));
}
tokio::task::spawn(Self::accept(ep2.clone()));
let ret = Self { ep: Some(ep2) };
Ok(ret)
}
@@ -86,10 +125,7 @@ impl Http3Support {
info!("h3 sock {:?}", sock);
let rt = quinn::default_runtime().ok_or_else(|| Error::NoRuntime)?;
let ep1 = Endpoint::new(conf, Some(conf_srv.clone()), sock, rt)?;
{
let ep = ep1.clone();
tokio::task::spawn(Self::accept(ep));
}
tokio::task::spawn(Self::accept(ep1.clone()));
let ret = Self { ep: Some(ep1) };
Ok(ret)
}
@@ -104,23 +140,24 @@ impl Http3Support {
async fn accept(ep: Endpoint) {
info!("accepting h3");
while let Some(inc) = ep.accept().await {
tokio::spawn(Self::handle_incoming(inc));
let addr_remote = inc.remote_address();
tokio::spawn(Self::handle_incoming(inc, addr_remote));
}
}
async fn handle_incoming(inc: Incoming) -> Result<(), Error> {
match Self::handle_incoming_inner(inc).await {
async fn handle_incoming(inc: Incoming, addr_remote: SocketAddr) -> Result<(), Error> {
info!("handle_incoming {}", addr_remote);
match Self::handle_incoming_inner_2(inc, addr_remote).await {
Ok(x) => Ok(x),
Err(e) => {
info!("error handle_connection {}", e);
info!("handle_incoming_inner_2 returns error {}", e);
Err(e)
}
}
}
async fn handle_incoming_inner(inc: Incoming) -> Result<(), Error> {
let addr_remote = inc.remote_address();
info!("new incoming {:?}", addr_remote);
async fn handle_incoming_inner_1(inc: Incoming, addr_remote: SocketAddr) -> Result<(), Error> {
info!("handle_incoming_inner_1 new incoming {:?}", addr_remote);
let conn1 = inc.accept()?.await?;
let conn2 = h3_quinn::Connection::new(conn1);
let mut conn3 = h3::server::builder().build::<_, Bytes>(conn2).await?;
@@ -142,12 +179,55 @@ impl Http3Support {
}
Ok(())
}
async fn handle_incoming_inner_2(inc: Incoming, addr_remote: SocketAddr) -> Result<(), Error> {
let selfname = "handle_incoming_inner_2";
info!("{} new incoming {:?}", selfname, addr_remote);
let conn1 = inc.await?;
info!("{} connected {:?}", selfname, addr_remote);
let conn2 = h3_quinn::Connection::new(conn1);
let mut conn3 = h3::server::Connection::new(conn2).await?;
info!("{} h3 Connection awaited {:?}", selfname, addr_remote);
// let mut conn3 = h3::server::builder().build::<_, Bytes>(conn2).await?;
loop {
break match conn3.accept().await {
Ok(Some((req, stream))) => {
info!("in h3 loop req {:?}", req);
tokio::spawn(async move {
let x = Self::handle_req(req, stream, addr_remote).await;
info!("handle_req return {:?}", x);
});
continue;
}
Ok(None) => {
info!("in h3 loop None");
}
Err(e) => {
info!("in h3 loop error {}", e);
}
};
}
Ok(())
}
async fn handle_req(
req: Request<()>,
mut stream: RequestStream<BidiStream<Bytes>, Bytes>,
addr_remote: SocketAddr,
) -> Result<(), Error> {
let method = req.method();
let uri = req.uri();
let headers = req.headers();
info!("see request {} {:?} {:?} {:?}", addr_remote, method, uri, headers);
let res = http::Response::builder()
// .version(http::Version::HTTP_3)
.status(StatusCode::OK)
.header("x-daqbuf-tmp", "8e4b217")
.body(())?;
stream.send_response(res).await?;
// stream.send_data(Bytes::from_static(b"2025-02-05T16:37:12Z")).await?;
// stream.finish().await?;
info!("response sent {}", addr_remote);
Ok(())
}
}
// impl Future for Http3Support {
// type Output = Result<(), Error>;
// fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
// todo!()
// }
// }