diff --git a/mtp-webserver/src/h3.rs b/mtp-webserver/src/h3.rs index c09379c..abb17aa 100644 --- a/mtp-webserver/src/h3.rs +++ b/mtp-webserver/src/h3.rs @@ -143,6 +143,11 @@ pub(crate) async fn run_driver( return; } }; + tracing::debug!( + remote = %remote_addr, + session_id = ?session.session_id(), + "accepted WebTransport MTP session" + ); tokio::spawn(run_session_requests( session.clone(), router.clone(), diff --git a/mtp-webserver/src/transport.rs b/mtp-webserver/src/transport.rs index a484f26..31d8376 100644 --- a/mtp-webserver/src/transport.rs +++ b/mtp-webserver/src/transport.rs @@ -32,6 +32,8 @@ pub struct H3TransportSender { pub struct H3TransportReceiver { stream: H3RecvStream, + quinn: quinn::Connection, + read_exact_calls: u64, } impl H3TransportConnection { @@ -76,18 +78,42 @@ impl TransportSendStream for H3TransportSender { #[async_trait::async_trait] impl TransportRecvStream for H3TransportReceiver { async fn read_exact(&mut self, buf: &mut [u8]) -> Result<(), CommunicationError> { + let first_read = self.read_exact_calls == 0; + self.read_exact_calls += 1; self.stream .read_exact(buf) .await - .map(|_| ()) + .map(|_| { + if first_read { + tracing::debug!( + remote = %self.quinn.remote_address(), + bytes = buf.len(), + "received first bytes from WebTransport MTP stream" + ); + } + }) .map_err(|error| { - if error.kind() == std::io::ErrorKind::UnexpectedEof { - // Browser control frames are sent on one-frame uni streams. - // Reaching FIN while looking for another frame is normal. + if error.kind() == std::io::ErrorKind::UnexpectedEof + || self.quinn.close_reason().is_some() + { + /* + * Reaching FIN, or losing the enclosing QUIC connection, + * is a normal stream-closure path. Do not turn it into a + * frame-header failure and close the connection again. + */ return CommunicationError::StreamClosed; } - error!("[mtp-webserver] receive stream read_exact failed ({} bytes): {error}", buf.len()); - tracing::warn!(len = buf.len(), %error, "WebTransport receive stream read_exact failed"); + error!( + "[mtp-webserver] receive stream read_exact failed ({} bytes): {error}", + buf.len() + ); + tracing::warn!( + remote = %self.quinn.remote_address(), + first_read, + len = buf.len(), + %error, + "WebTransport receive stream read_exact failed" + ); CommunicationError::StreamError }) } @@ -101,6 +127,9 @@ impl TransportRecvStream for H3TransportReceiver { Ok(Some(buf)) } Err(error) => { + if self.quinn.close_reason().is_some() { + return Err(CommunicationError::StreamClosed); + } error!( "[mtp-webserver] receive stream read failed (max {} bytes): {error}", max @@ -167,10 +196,27 @@ impl TransportConnection for H3TransportConnection { loop { match self.session.accept_uni().await { Ok(Some((id, stream))) if id == self.session.session_id() => { - return Ok(H3TransportReceiver { stream }); + let stream_id = h3::quic::RecvStream::recv_id(&stream); + tracing::debug!( + remote = %self.quinn.remote_address(), + session_id = ?self.session.session_id(), + stream_id = ?stream_id, + "accepted WebTransport MTP receive stream" + ); + return Ok(H3TransportReceiver { + stream, + quinn: self.quinn.clone(), + read_exact_calls: 0, + }); } - Ok(Some(_)) => { + Ok(Some((stream_session_id, _stream))) => { consecutive_errors = 0; + tracing::debug!( + remote = %self.quinn.remote_address(), + session_id = ?self.session.session_id(), + stream_session_id = ?stream_session_id, + "ignored WebTransport receive stream belonging to another session" + ); continue; } Ok(None) => return Err(CommunicationError::StreamClosed), @@ -306,7 +352,18 @@ async fn accept_web_connection_inner( connection_id, }, ) - .await?; + .await; + #[cfg(feature = "crypto")] + if let Err(error) = &result { + tracing::warn!( + remote = %remote_addr, + connection_id, + %error, + "WebTransport MTP handshake failed" + ); + } + #[cfg(feature = "crypto")] + let result = result?; #[cfg(not(feature = "crypto"))] let result = engine.accept(&sender, &receiver).await?; diff --git a/transport/src/generic_connection.rs b/transport/src/generic_connection.rs index 23f3fe9..4fa6433 100644 --- a/transport/src/generic_connection.rs +++ b/transport/src/generic_connection.rs @@ -333,6 +333,18 @@ impl GenericReceiver { break 'stream; } Err(_) => { + if frames == 0 { + tracing::warn!( + timeout = ?policy.read_timeout, + "MTP receive stream timed out before its first complete frame" + ); + } else { + tracing::debug!( + frames, + timeout = ?policy.read_timeout, + "MTP receive stream idle timeout" + ); + } break; } } @@ -376,6 +388,9 @@ impl GenericReceiver { ) .await; if !matches!(&body_read, Ok(Ok(()))) { + if matches!(&body_read, Ok(Err(CommunicationError::StreamClosed))) { + break 'stream; + } tracing::warn!( pipe_chunk_len = chunk_len, ?body_read, diff --git a/wasm/src/client/authentication.rs b/wasm/src/client/authentication.rs index 45cec8d..df7989e 100644 --- a/wasm/src/client/authentication.rs +++ b/wasm/src/client/authentication.rs @@ -8,6 +8,14 @@ use crate::config::ConnectionConfig; use crate::error::js_error; use crate::transport::WasmTransport; +fn server_rejection_message(outcome: &CommunicationValue) -> Option<&str> { + (outcome.get_data(DataType::Connected) == Some(&DataValue::BoolFalse)).then(|| { + outcome + .get_str(DataType::ErrorMessage) + .unwrap_or("host rejected the connection") + }) +} + #[wasm_bindgen] #[allow(deprecated)] impl WasmClient { @@ -79,6 +87,18 @@ impl WasmClient { .unwrap_or("host does not support this protocol version"), )); } + + // Generic host rejections are IdentificationResponse frames with + // Connected=false. They intentionally do not carry a negotiated + // Version because negotiation never completed. Check this before + // reading Version, otherwise a useful server error such as an + // authentication timeout is reported as the misleading + // "host omitted a valid negotiated protocol version". + if let Some(message) = server_rejection_message(&outcome) { + self.set_state_if_current(generation, ConnectionState::Disconnected); + return Err(js_error(message)); + } + let negotiated_version = match outcome.get_data(DataType::Version) { Some(DataValue::Str(version)) => mtp_codec::Version::parse(version) .ok_or_else(|| js_error("host omitted a valid negotiated protocol version"))?, @@ -628,3 +648,24 @@ impl WasmClient { Ok(server_challenge) } } + +#[cfg(test)] +mod tests { + use super::server_rejection_message; + use mtp_codec::{CommunicationType, CommunicationValue, DataType, DataValue}; + + #[test] + fn reports_rejection_reason_without_a_negotiated_version() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default(DataType::Connected, DataValue::BoolFalse) + .add_typed_default( + DataType::ErrorMessage, + DataValue::Str("authentication handshake timed out".into()), + ); + + assert_eq!( + server_rejection_message(&response), + Some("authentication handshake timed out") + ); + } +}