From ca1b4ffbcf466d9c8096d49c21e38a00a442c4e7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philipp=20Kr=C3=BCger?= Date: Mon, 1 Dec 2025 18:36:30 +0100 Subject: [PATCH] fix: Don't allow sending path status frames on non-multipath connections --- quinn-proto/src/connection/mod.rs | 53 +++++++++++++++-------------- quinn-proto/src/connection/paths.rs | 19 ++++++++--- quinn-proto/src/iroh_hp.rs | 2 +- quinn-proto/src/lib.rs | 7 ++-- quinn-proto/src/tests/multipath.rs | 6 ++-- quinn/src/connection.rs | 11 +++--- quinn/src/path.rs | 27 ++++++++------- 7 files changed, 71 insertions(+), 54 deletions(-) diff --git a/quinn-proto/src/connection/mod.rs b/quinn-proto/src/connection/mod.rs index bcd7a4845..7048f0bb2 100644 --- a/quinn-proto/src/connection/mod.rs +++ b/quinn-proto/src/connection/mod.rs @@ -65,7 +65,7 @@ mod packet_crypto; use packet_crypto::{PrevCrypto, ZeroRttCrypto}; mod paths; -pub use paths::{ClosedPath, PathEvent, PathId, PathStatus, RttEstimator}; +pub use paths::{NotOpen, PathEvent, PathId, PathStatus, RttEstimator, SetPathStatusError}; use paths::{PathData, PathState}; pub(crate) mod qlog; @@ -547,7 +547,7 @@ impl Connection { remote: SocketAddr, initial_status: PathStatus, now: Instant, - ) -> Result<(PathId, bool), PathError> { + ) -> Result<(PathId, bool), OpenPathError> { match self .paths .iter() @@ -569,12 +569,12 @@ impl Connection { remote: SocketAddr, initial_status: PathStatus, now: Instant, - ) -> Result { + ) -> Result { if !self.is_multipath_negotiated() { - return Err(PathError::MultipathNotNegotiated); + return Err(OpenPathError::MultipathNotNegotiated); } if self.side().is_server() { - return Err(PathError::ServerSideNotAllowed); + return Err(OpenPathError::ServerSideNotAllowed); } let max_abandoned = self.abandoned_paths.iter().max().copied(); @@ -585,18 +585,18 @@ impl Connection { .saturating_add(1u8); if Some(path_id) > self.max_path_id() { - return Err(PathError::MaxPathIdReached); + return Err(OpenPathError::MaxPathIdReached); } if path_id > self.remote_max_path_id { self.spaces[SpaceId::Data].pending.paths_blocked = true; - return Err(PathError::MaxPathIdReached); + return Err(OpenPathError::MaxPathIdReached); } if self.rem_cids.get(&path_id).map(CidQueue::active).is_none() { self.spaces[SpaceId::Data] .pending .path_cids_blocked .push(path_id); - return Err(PathError::RemoteCidsExhausted); + return Err(OpenPathError::RemoteCidsExhausted); } let path = self.ensure_path(path_id, remote, now, None); @@ -687,17 +687,17 @@ impl Connection { } /// Gets the local [`PathStatus`] for a known [`PathId`] - pub fn path_status(&self, path_id: PathId) -> Result { + pub fn path_status(&self, path_id: PathId) -> Result { self.path(path_id) .map(PathData::local_status) - .ok_or(ClosedPath { _private: () }) + .ok_or(NotOpen { _private: () }) } /// Returns the path's remote socket address - pub fn path_remote_address(&self, path_id: PathId) -> Result { + pub fn path_remote_address(&self, path_id: PathId) -> Result { self.path(path_id) .map(|path| path.remote) - .ok_or(ClosedPath { _private: () }) + .ok_or(NotOpen { _private: () }) } /// Sets the [`PathStatus`] for a known [`PathId`] @@ -707,8 +707,11 @@ impl Connection { &mut self, path_id: PathId, status: PathStatus, - ) -> Result { - let path = self.path_mut(path_id).ok_or(ClosedPath { _private: () })?; + ) -> Result { + if !self.is_multipath_negotiated() { + return Err(SetPathStatusError::MultipathNotNegotiated); + } + let path = self.path_mut(path_id).ok_or(SetPathStatusError::NotOpen)?; let prev = match path.status.local_update(status) { Some(prev) => { self.spaces[SpaceId::Data] @@ -739,11 +742,11 @@ impl Connection { &mut self, path_id: PathId, timeout: Option, - ) -> Result, ClosedPath> { + ) -> Result, NotOpen> { let path = self .paths .get_mut(&path_id) - .ok_or(ClosedPath { _private: () })?; + .ok_or(NotOpen { _private: () })?; Ok(std::mem::replace(&mut path.data.idle_timeout, timeout)) } @@ -756,11 +759,11 @@ impl Connection { &mut self, path_id: PathId, interval: Option, - ) -> Result, ClosedPath> { + ) -> Result, NotOpen> { let path = self .paths .get_mut(&path_id) - .ok_or(ClosedPath { _private: () })?; + .ok_or(NotOpen { _private: () })?; Ok(std::mem::replace(&mut path.data.keep_alive, interval)) } @@ -959,7 +962,7 @@ impl Connection { loop { // check if there is at least one active CID to use for sending let Some(remote_cid) = self.rem_cids.get(&path_id).map(CidQueue::active) else { - let err = PathError::RemoteCidsExhausted; + let err = OpenPathError::RemoteCidsExhausted; if !self.abandoned_paths.contains(&path_id) { debug!(?err, %path_id, "no active CID for path"); self.events.push_back(Event::Path(PathEvent::LocallyClosed { @@ -1890,7 +1893,7 @@ impl Connection { self.events.push_back(Event::Path(PathEvent::LocallyClosed { id: path_id, - error: PathError::ValidationFailed, + error: OpenPathError::ValidationFailed, })); } PathTimer::Pacing => trace!("pacing timer expired"), @@ -1997,11 +2000,11 @@ impl Connection { /// Ping the remote endpoint over a specific path /// /// Causes an ACK-eliciting packet to be transmitted on the path. - pub fn ping_path(&mut self, path: PathId) -> Result<(), ClosedPath> { + pub fn ping_path(&mut self, path: PathId) -> Result<(), NotOpen> { let path_data = self.spaces[self.highest_space] .number_spaces .get_mut(&path) - .ok_or(ClosedPath { _private: () })?; + .ok_or(NotOpen { _private: () })?; path_data.ping_pending = true; Ok(()) } @@ -2085,7 +2088,7 @@ impl Connection { } /// Get the address observed by the remote over the given path - pub fn path_observed_address(&self, path_id: PathId) -> Result, ClosedPath> { + pub fn path_observed_address(&self, path_id: PathId) -> Result, NotOpen> { self.path(path_id) .map(|path_data| { path_data @@ -2093,7 +2096,7 @@ impl Connection { .as_ref() .map(|observed| observed.socket_addr()) }) - .ok_or(ClosedPath { _private: () }) + .ok_or(NotOpen { _private: () }) } /// The local IP address which was used when the peer established @@ -6251,7 +6254,7 @@ impl From for io::Error { /// Errors that might trigger a path being closed // TODO(@divma): maybe needs to be reworked based on what we want to do with the public API #[derive(Debug, Error, PartialEq, Eq, Clone, Copy)] -pub enum PathError { +pub enum OpenPathError { /// The extension was not negotiated with the peer #[error("multipath extention not negotiated")] MultipathNotNegotiated, diff --git a/quinn-proto/src/connection/paths.rs b/quinn-proto/src/connection/paths.rs index 76e66797f..f774774fe 100644 --- a/quinn-proto/src/connection/paths.rs +++ b/quinn-proto/src/connection/paths.rs @@ -5,7 +5,7 @@ use thiserror::Error; use tracing::{debug, trace}; use super::{ - PathError, PathStats, + OpenPathError, PathStats, mtud::MtuDiscovery, pacing::Pacer, spaces::{PacketNumberSpace, SentPacket}, @@ -797,7 +797,7 @@ pub enum PathEvent { /// Path for which the error occurred id: PathId, /// The error that occurred - error: PathError, + error: OpenPathError, }, /// The remote changed the status of the path /// @@ -820,10 +820,21 @@ pub enum PathEvent { }, } +/// Error from setting path status +#[derive(Debug, Error, Clone, PartialEq, Eq)] +pub enum SetPathStatusError { + /// Error indicating that a path has not been opened or has already been abandoned + #[error("path not open")] + NotOpen, + /// Error indicating that this operation requires multipath to be negotiated whereas it hasn't been + #[error("multipath not negotiated")] + MultipathNotNegotiated, +} + /// Error indicating that a path has not been opened or has already been abandoned #[derive(Debug, Default, Error, Clone, PartialEq, Eq)] -#[error("closed path")] -pub struct ClosedPath { +#[error("path not open")] +pub struct NotOpen { pub(super) _private: (), } diff --git a/quinn-proto/src/iroh_hp.rs b/quinn-proto/src/iroh_hp.rs index d9b2f37cf..d87173329 100644 --- a/quinn-proto/src/iroh_hp.rs +++ b/quinn-proto/src/iroh_hp.rs @@ -33,7 +33,7 @@ pub enum Error { NotEnoughAddresses, /// Nat traversal attempt failed due to a multipath error #[error("Failed to establish paths {0}")] - Multipath(super::PathError), + Multipath(super::OpenPathError), } pub(crate) struct NatTraversalRound { diff --git a/quinn-proto/src/lib.rs b/quinn-proto/src/lib.rs index 9f3c3711e..e05efda8c 100644 --- a/quinn-proto/src/lib.rs +++ b/quinn-proto/src/lib.rs @@ -43,10 +43,11 @@ pub use bloom_token_log::BloomTokenLog; mod connection; pub use crate::connection::{ - Chunk, Chunks, ClosePathError, ClosedPath, ClosedStream, Connection, ConnectionError, - ConnectionStats, Datagrams, Event, FinishError, FrameStats, PathError, PathEvent, PathId, + Chunk, Chunks, ClosePathError, ClosedStream, Connection, ConnectionError, ConnectionStats, + Datagrams, Event, FinishError, FrameStats, NotOpen, OpenPathError, PathEvent, PathId, PathStats, PathStatus, ReadError, ReadableError, RecvStream, RttEstimator, SendDatagramError, - SendStream, ShouldTransmit, StreamEvent, Streams, UdpStats, WriteError, Written, + SendStream, SetPathStatusError, ShouldTransmit, StreamEvent, Streams, UdpStats, WriteError, + Written, }; #[cfg(feature = "qlog")] pub use connection::qlog::QlogStream; diff --git a/quinn-proto/src/tests/multipath.rs b/quinn-proto/src/tests/multipath.rs index b7bb2b1d2..6400b4409 100644 --- a/quinn-proto/src/tests/multipath.rs +++ b/quinn-proto/src/tests/multipath.rs @@ -14,7 +14,7 @@ use crate::{ EndpointConfig, Instant, LOC_CID_COUNT, PathId, PathStatus, RandomConnectionIdGenerator, ServerConfig, TransportConfig, cid_queue::CidQueue, }; -use crate::{Event, PathError, PathEvent}; +use crate::{Event, OpenPathError, PathEvent}; use super::util::{min_opt, subscribe}; use super::{Pair, client_config, server_config}; @@ -509,7 +509,7 @@ fn open_path_validation_fails_server_side() { let client_conn = pair.client_conn_mut(client_ch); assert_matches!( client_conn.poll().unwrap(), - Event::Path(crate::PathEvent::LocallyClosed { id, error: PathError::ValidationFailed }) if id == path_id + Event::Path(crate::PathEvent::LocallyClosed { id, error: OpenPathError::ValidationFailed }) if id == path_id ); let server_conn = pair.server_conn_mut(client_ch); @@ -541,7 +541,7 @@ fn open_path_validation_fails_client_side() { let server_conn = pair.server_conn_mut(client_ch); assert_matches!(server_conn.poll().unwrap(), - Event::Path(crate::PathEvent::LocallyClosed { id, error: PathError::ValidationFailed }) if id == path_id + Event::Path(crate::PathEvent::LocallyClosed { id, error: OpenPathError::ValidationFailed }) if id == path_id ); } diff --git a/quinn/src/connection.rs b/quinn/src/connection.rs index a89665736..6b04248c9 100644 --- a/quinn/src/connection.rs +++ b/quinn/src/connection.rs @@ -27,8 +27,9 @@ use crate::{ udp_transmit, }; use proto::{ - ConnectionError, ConnectionHandle, ConnectionStats, Dir, EndpointEvent, PathError, PathEvent, - PathId, PathStats, PathStatus, Side, StreamEvent, StreamId, congestion::Controller, iroh_hp, + ConnectionError, ConnectionHandle, ConnectionStats, Dir, EndpointEvent, OpenPathError, + PathEvent, PathId, PathStats, PathStatus, Side, StreamEvent, StreamId, congestion::Controller, + iroh_hp, }; /// In-progress connection attempt future @@ -392,7 +393,7 @@ impl Connection { .next() .unwrap_or_default(); if addr.is_ipv6() && !ipv6 { - return OpenPath::rejected(PathError::InvalidRemoteAddress(addr)); + return OpenPath::rejected(OpenPathError::InvalidRemoteAddress(addr)); } let addr = if ipv6 { SocketAddr::V6(ensure_ipv6(addr)) @@ -455,7 +456,7 @@ impl Connection { .next() .unwrap_or_default(); if addr.is_ipv6() && !ipv6 { - return OpenPath::rejected(PathError::InvalidRemoteAddress(addr)); + return OpenPath::rejected(OpenPathError::InvalidRemoteAddress(addr)); } let addr = if ipv6 { SocketAddr::V6(ensure_ipv6(addr)) @@ -1322,7 +1323,7 @@ pub(crate) struct State { /// Always set to Some before the connection becomes drained pub(crate) error: Option, /// Tracks paths being opened - open_path: FxHashMap>>, + open_path: FxHashMap>>, /// Tracks paths being closed pub(crate) close_path: FxHashMap>, pub(crate) path_events: tokio::sync::broadcast::Sender, diff --git a/quinn/src/path.rs b/quinn/src/path.rs index 124385bc6..5330225a2 100644 --- a/quinn/src/path.rs +++ b/quinn/src/path.rs @@ -6,7 +6,8 @@ use std::task::{Context, Poll, ready}; use std::time::Duration; use proto::{ - ClosePathError, ClosedPath, ConnectionError, PathError, PathEvent, PathId, PathStatus, VarInt, + ClosePathError, NotOpen, ConnectionError, OpenPathError, PathEvent, PathId, PathStatus, + SetPathStatusError, VarInt, }; use tokio::sync::{oneshot, watch}; use tokio_stream::{Stream, wrappers::WatchStream}; @@ -22,14 +23,14 @@ enum OpenPathInner { /// /// This migth fail later on. Ongoing { - opened: WatchStream>, + opened: WatchStream>, path_id: PathId, conn: ConnectionRef, }, /// Opening a path failed immediately Rejected { /// The error that occurred - err: PathError, + err: OpenPathError, }, /// The path is already open Ready { @@ -41,7 +42,7 @@ enum OpenPathInner { impl OpenPath { pub(crate) fn new( path_id: PathId, - opened: watch::Receiver>, + opened: watch::Receiver>, conn: ConnectionRef, ) -> Self { Self(OpenPathInner::Ongoing { @@ -55,7 +56,7 @@ impl OpenPath { Self(OpenPathInner::Ready { path_id, conn }) } - pub(crate) fn rejected(err: PathError) -> Self { + pub(crate) fn rejected(err: OpenPathError) -> Self { Self(OpenPathInner::Rejected { err }) } @@ -75,7 +76,7 @@ impl OpenPath { } impl Future for OpenPath { - type Output = Result; + type Output = Result; fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll { match self.get_mut().0 { OpenPathInner::Ongoing { @@ -91,7 +92,7 @@ impl Future for OpenPath { // This only happens if receiving a notification change failed, this means the // sender was dropped. This generally should not happen so we use a transient // error - Poll::Ready(Err(PathError::ValidationFailed)) + Poll::Ready(Err(OpenPathError::ValidationFailed)) } }, OpenPathInner::Ready { @@ -120,7 +121,7 @@ impl Path { } /// The current local [`PathStatus`] of this path. - pub fn status(&self) -> Result { + pub fn status(&self) -> Result { self.conn .state .lock("path status") @@ -129,7 +130,7 @@ impl Path { } /// Sets the [`PathStatus`] of this path. - pub fn set_status(&self, status: PathStatus) -> Result<(), ClosedPath> { + pub fn set_status(&self, status: PathStatus) -> Result<(), SetPathStatusError> { self.conn .state .lock("set path status") @@ -171,7 +172,7 @@ impl Path { pub fn set_max_idle_timeout( &self, timeout: Option, - ) -> Result, ClosedPath> { + ) -> Result, NotOpen> { let mut state = self.conn.state.lock("path_set_max_idle_timeout"); state.inner.set_path_max_idle_timeout(self.id, timeout) } @@ -186,7 +187,7 @@ impl Path { pub fn set_keep_alive_interval( &self, interval: Option, - ) -> Result, ClosedPath> { + ) -> Result, NotOpen> { let mut state = self.conn.state.lock("path_set_keep_alive_interval"); state.inner.set_path_keep_alive_interval(self.id, interval) } @@ -194,7 +195,7 @@ impl Path { /// Track changes on our external address as reported by the peer. /// /// If the address-discovery extension is not negotiated, the stream will never return. - pub fn observed_external_addr(&self) -> Result { + pub fn observed_external_addr(&self) -> Result { let state = self.conn.state.lock("per_path_observed_address"); let path_events = state.path_events.subscribe(); let initial_value = state.inner.path_observed_address(self.id)?; @@ -207,7 +208,7 @@ impl Path { } /// The peer's UDP address for this path. - pub fn remote_address(&self) -> Result { + pub fn remote_address(&self) -> Result { let state = self.conn.state.lock("per_path_remote_address"); state.inner.path_remote_address(self.id) }