diff --git a/web-transport-moq/src/error.rs b/web-transport-moq/src/error.rs index 50476cc8a..654260fc4 100644 --- a/web-transport-moq/src/error.rs +++ b/web-transport-moq/src/error.rs @@ -69,6 +69,9 @@ impl From for SessionError { } /// Why a session closed, shared by the session and its streams. The first close wins. +/// +/// It also knows whether the session is raw QUIC, which decides how its close and stream +/// codes are encoded. #[derive(Debug)] pub(crate) struct CloseReason { // Raw QUIC carries the application code as is; HTTP/3 maps it into its own code space. @@ -84,6 +87,26 @@ impl CloseReason { } } + /// The QUIC code to reset or stop a stream with. + pub(crate) fn encode_stream_code(&self, code: u32) -> noq::VarInt { + if self.raw { + return code.into(); + } + noq::VarInt::try_from(web_transport_proto::error_to_http3(code)).unwrap() + } + + /// Decode the peer's QUIC code for a stream reset or stop, or None if it is not one. + pub(crate) fn decode_stream_code(&self, code: noq::VarInt) -> Option { + let code = code.into_inner(); + // An older raw peer maps its codes into the HTTP/3 range too. No u32 reaches that + // range, so the mapped form cannot be mistaken for a raw code. + match web_transport_proto::error_from_http3(code) { + Some(code) => Some(code), + None if self.raw => u32::try_from(code).ok(), + None => None, + } + } + /// Record the close reason, returning false if one was already recorded. pub(crate) fn set(&self, err: SessionError) -> bool { self.reason.set(err).is_ok() @@ -348,3 +371,35 @@ impl web_transport_trait::Error for ReadError { } } } + +#[cfg(test)] +mod tests { + use super::*; + + /// WebTransport keeps the HTTP/3 mapping in both directions, and a code outside its + /// range is not a WebTransport code. + #[test] + fn http3_stream_codes_stay_mapped() { + let close = CloseReason::new(false); + let mapped = close.encode_stream_code(5); + assert_eq!(mapped.into_inner(), web_transport_proto::error_to_http3(5)); + assert_eq!(close.decode_stream_code(mapped), Some(5)); + assert_eq!(close.decode_stream_code(5u32.into()), None); + } + + /// Raw QUIC sends the code as is and reads both the raw and the legacy mapped form. + #[test] + fn raw_stream_codes_pass_through() { + let close = CloseReason::new(true); + assert_eq!(close.encode_stream_code(5).into_inner(), 5); + assert_eq!(close.decode_stream_code(5u32.into()), Some(5)); + assert_eq!(close.decode_stream_code(u32::MAX.into()), Some(u32::MAX)); + + let legacy = noq::VarInt::try_from(web_transport_proto::error_to_http3(5)).unwrap(); + assert_eq!(close.decode_stream_code(legacy), Some(5)); + + // Past u32 and outside the HTTP/3 range: neither form. + let invalid = noq::VarInt::from_u64(1 << 40).unwrap(); + assert_eq!(close.decode_stream_code(invalid), None); + } +} diff --git a/web-transport-moq/src/recv.rs b/web-transport-moq/src/recv.rs index 25e1059e4..4c3d18817 100644 --- a/web-transport-moq/src/recv.rs +++ b/web-transport-moq/src/recv.rs @@ -24,20 +24,25 @@ impl RecvStream { } } - /// Report a session error as the session's close reason. - fn map_error(&self, e: impl Into) -> ReadError { - match e.into() { - ReadError::SessionError(e) => ReadError::SessionError(self.close.map(e)), - e => e, + /// Decode the peer's reset code, and report a session error as the session's close reason. + fn map_error(&self, e: noq::ReadError) -> ReadError { + match e { + noq::ReadError::Reset(code) => match self.close.decode_stream_code(code) { + Some(code) => ReadError::Reset(code), + None => ReadError::InvalidReset(code), + }, + e => match ReadError::from(e) { + ReadError::SessionError(e) => ReadError::SessionError(self.close.map(e)), + e => e, + }, } } /// Tell the other end to stop sending data with the given error code. See /// [`noq::RecvStream::stop`]. This is a u32 with WebTransport since it shares the error - /// space with HTTP/3. + /// space with HTTP/3. A raw QUIC session sends the code as is. pub fn stop(&mut self, code: u32) -> Result<(), noq::ClosedStream> { - let code = web_transport_proto::error_to_http3(code); - let code = noq::VarInt::try_from(code).unwrap(); + let code = self.close.encode_stream_code(code); self.inner.stop(code) } @@ -93,7 +98,7 @@ impl RecvStream { pub async fn received_reset(&mut self) -> Result, SessionError> { match self.inner.received_reset().await { Ok(None) => Ok(None), - Ok(Some(code)) => Ok(web_transport_proto::error_from_http3(code.into_inner())), + Ok(Some(code)) => Ok(self.close.decode_stream_code(code)), Err(noq::ResetError::ConnectionLost(conn_err)) => Err(self.close.map(conn_err.into())), Err(noq::ResetError::ZeroRttRejected) => unreachable!("0-RTT not supported"), } diff --git a/web-transport-moq/src/send.rs b/web-transport-moq/src/send.rs index 63f59aa3a..39af0d1fc 100644 --- a/web-transport-moq/src/send.rs +++ b/web-transport-moq/src/send.rs @@ -24,19 +24,25 @@ impl SendStream { Self { stream, close } } - /// Report a session error as the session's close reason. - fn map_error(&self, e: impl Into) -> WriteError { - match e.into() { - WriteError::SessionError(e) => WriteError::SessionError(self.close.map(e)), - e => e, + /// Decode the peer's stop code, and report a session error as the session's close reason. + fn map_error(&self, e: noq::WriteError) -> WriteError { + match e { + noq::WriteError::Stopped(code) => match self.close.decode_stream_code(code) { + Some(code) => WriteError::Stopped(code), + None => WriteError::InvalidStopped(code), + }, + e => match WriteError::from(e) { + WriteError::SessionError(e) => WriteError::SessionError(self.close.map(e)), + e => e, + }, } } /// Abruptly reset the stream with the provided error code. See [`noq::SendStream::reset`]. /// This is a u32 with WebTransport because we share the error space with HTTP/3. + /// A raw QUIC session sends the code as is. pub fn reset(&mut self, code: u32) -> Result<(), ClosedStream> { - let code = web_transport_proto::error_to_http3(code); - let code = noq::VarInt::try_from(code).unwrap(); + let code = self.close.encode_stream_code(code); self.stream.reset(code).map_err(Into::into) } @@ -48,7 +54,7 @@ impl SendStream { /// supported. pub async fn stopped(&self) -> Result, SessionError> { match self.stream.stopped().await { - Ok(Some(code)) => Ok(web_transport_proto::error_from_http3(code.into_inner())), + Ok(Some(code)) => Ok(self.close.decode_stream_code(code)), Ok(None) => Ok(None), Err(noq::StoppedError::ConnectionLost(conn_err)) => { Err(self.close.map(conn_err.into())) diff --git a/web-transport-moq/tests/raw_stream_codes.rs b/web-transport-moq/tests/raw_stream_codes.rs new file mode 100644 index 000000000..c12e46e82 --- /dev/null +++ b/web-transport-moq/tests/raw_stream_codes.rs @@ -0,0 +1,167 @@ +//! A raw QUIC session sends RESET_STREAM and STOP_SENDING codes as is, so a plain QUIC +//! peer agrees on them. It still reads the HTTP/3-mapped codes an older raw peer sends. + +use std::{net::Ipv4Addr, sync::Arc, time::Duration}; + +use anyhow::{Context as _, Result}; +use rcgen::{CertifiedKey, KeyPair}; +use rustls::pki_types::{PrivateKeyDer, PrivatePkcs8KeyDer}; +use tokio::time::timeout; +use web_transport_moq::{ReadError, Session, WriteError, noq, proto}; +use web_transport_trait::Error as _; + +const CODE: u32 = 5; + +/// With both crypto features enabled, as `--all-features` does, the crate cannot pick a +/// provider for us, so the test installs one. +fn install_crypto_provider() { + #[cfg(all(feature = "aws-lc-rs", feature = "ring"))] + { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + } +} + +/// A raw session on the client and the plain QUIC connection it talks to, plus the +/// endpoints that drive them. +async fn raw_and_plain() -> Result<(Session, noq::Connection, [noq::Endpoint; 2])> { + install_crypto_provider(); + + let CertifiedKey { cert, signing_key } = + rcgen::generate_simple_self_signed(vec!["localhost".into()])?; + let key = PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(KeyPair::serialize_der( + &signing_key, + ))); + + let server_config = noq::ServerConfig::with_single_cert(vec![cert.der().clone()], key)?; + let server = noq::Endpoint::server(server_config, (Ipv4Addr::LOCALHOST, 0).into())?; + + let mut roots = rustls::RootCertStore::empty(); + roots.add(cert.der().clone())?; + let client_config = noq::ClientConfig::with_root_certificates(Arc::new(roots))?; + let client = noq::Endpoint::client((Ipv4Addr::LOCALHOST, 0).into())?; + + let connecting = client.connect_with(client_config, server.local_addr()?, "localhost")?; + let accept = async { anyhow::Ok(server.accept().await.context("no connection")?.await?) }; + let (client_conn, server_conn) = + tokio::try_join!(async { anyhow::Ok(connecting.await?) }, accept)?; + + Ok((Session::raw(client_conn), server_conn, [client, server])) +} + +/// Open a uni stream from the plain peer and accept it on the session. +async fn plain_to_raw( + session: &Session, + plain: &noq::Connection, +) -> Result<(noq::SendStream, web_transport_moq::RecvStream)> { + let mut send = plain.open_uni().await?; + send.write_all(b"x").await?; + let mut recv = timeout(Duration::from_secs(5), session.accept_uni()).await??; + let mut buf = [0u8; 1]; + recv.read_exact(&mut buf).await?; + Ok((send, recv)) +} + +/// Open a uni stream from the session and accept it on the plain peer. +async fn raw_to_plain( + session: &Session, + plain: &noq::Connection, +) -> Result<(web_transport_moq::SendStream, noq::RecvStream)> { + let mut send = session.open_uni().await?; + send.write_all(b"x").await?; + let mut recv = timeout(Duration::from_secs(5), plain.accept_uni()).await??; + let mut buf = [0u8; 1]; + recv.read_exact(&mut buf).await?; + Ok((send, recv)) +} + +async fn read_err(recv: &mut web_transport_moq::RecvStream) -> ReadError { + let mut buf = [0u8; 1]; + timeout(Duration::from_secs(5), recv.read(&mut buf)) + .await + .expect("no reset") + .expect_err("read after a reset") +} + +async fn write_err(send: &mut web_transport_moq::SendStream) -> WriteError { + timeout(Duration::from_secs(5), async { + loop { + if let Err(err) = send.write(&[0u8; 1024]).await { + return err; + } + } + }) + .await + .expect("no stop") +} + +#[tokio::test] +async fn reset_is_sent_as_is() -> Result<()> { + let (session, plain, _endpoints) = raw_and_plain().await?; + let (mut send, mut recv) = raw_to_plain(&session, &plain).await?; + + send.reset(CODE)?; + + let mut buf = [0u8; 1]; + let err = timeout(Duration::from_secs(5), recv.read(&mut buf)).await?; + assert!( + matches!(err, Err(noq::ReadError::Reset(code)) if code == CODE.into()), + "{err:?}" + ); + Ok(()) +} + +#[tokio::test] +async fn stop_is_sent_as_is() -> Result<()> { + let (session, plain, _endpoints) = raw_and_plain().await?; + let (send, mut recv) = plain_to_raw(&session, &plain).await?; + + recv.stop(CODE)?; + + let stopped = timeout(Duration::from_secs(5), send.stopped()).await??; + assert_eq!(stopped, Some(CODE.into())); + Ok(()) +} + +#[tokio::test] +async fn peer_reset_is_read_as_is() -> Result<()> { + let (session, plain, _endpoints) = raw_and_plain().await?; + let (mut send, mut recv) = plain_to_raw(&session, &plain).await?; + + send.reset(CODE.into())?; + + let err = read_err(&mut recv).await; + assert_eq!(err.stream_error(), Some(CODE), "{err}"); + Ok(()) +} + +#[tokio::test] +async fn peer_stop_is_read_as_is() -> Result<()> { + let (session, plain, _endpoints) = raw_and_plain().await?; + let (mut send, mut recv) = raw_to_plain(&session, &plain).await?; + + recv.stop(CODE.into())?; + + let err = write_err(&mut send).await; + assert_eq!(err.stream_error(), Some(CODE), "{err}"); + assert_eq!(send.stopped().await?, Some(CODE)); + Ok(()) +} + +/// An older raw peer maps its codes into the HTTP/3 range, which still decodes. +#[tokio::test] +async fn legacy_peer_codes_are_unmapped() -> Result<()> { + let (session, plain, _endpoints) = raw_and_plain().await?; + let legacy = noq::VarInt::try_from(proto::error_to_http3(CODE))?; + + let (mut send, mut recv) = plain_to_raw(&session, &plain).await?; + send.reset(legacy)?; + let err = read_err(&mut recv).await; + assert_eq!(err.stream_error(), Some(CODE), "{err}"); + + let (mut send, mut recv) = raw_to_plain(&session, &plain).await?; + recv.stop(legacy)?; + let err = write_err(&mut send).await; + assert_eq!(err.stream_error(), Some(CODE), "{err}"); + assert_eq!(send.stopped().await?, Some(CODE)); + Ok(()) +}