gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-09-19 17:55:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: Don't allow sending path status frames on non-multipath connections

Browse Source
This commit is contained in:
Philipp Krüger
2025-12-01 18:36:30 +01:00
parent d5123326ce
commit ca1b4ffbcf
7 changed files with 71 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quinn-proto/src/connection/mod.rs
+28 -25
View File
@@ -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<PathId, PathError> {
) -> Result<PathId, OpenPathError> {
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<PathStatus, ClosedPath> {
pub fn path_status(&self, path_id: PathId) -> Result<PathStatus, NotOpen> {
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<SocketAddr, ClosedPath> {
pub fn path_remote_address(&self, path_id: PathId) -> Result<SocketAddr, NotOpen> {
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<PathStatus, ClosedPath> {
let path = self.path_mut(path_id).ok_or(ClosedPath { _private: () })?;
) -> Result<PathStatus, SetPathStatusError> {
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<Duration>,
) -> Result<Option<Duration>, ClosedPath> {
) -> Result<Option<Duration>, 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<Duration>,
) -> Result<Option<Duration>, ClosedPath> {
) -> Result<Option<Duration>, 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<Option<SocketAddr>, ClosedPath> {
pub fn path_observed_address(&self, path_id: PathId) -> Result<Option<SocketAddr>, 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<ConnectionError> 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,
quinn-proto/src/connection/paths.rs
+15 -4
View File
@@ -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: (),
}
quinn-proto/src/iroh_hp.rs
+1 -1
View File
@@ -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 {
quinn-proto/src/lib.rs
+4 -3
View File
@@ -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;
quinn-proto/src/tests/multipath.rs
+3 -3
View File
@@ -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
);
}
quinn/src/connection.rs
+6 -5
View File
@@ -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<ConnectionError>,
/// Tracks paths being opened
open_path: FxHashMap<PathId, watch::Sender<Result<(), PathError>>>,
open_path: FxHashMap<PathId, watch::Sender<Result<(), OpenPathError>>>,
/// Tracks paths being closed
pub(crate) close_path: FxHashMap<PathId, oneshot::Sender<VarInt>>,
pub(crate) path_events: tokio::sync::broadcast::Sender<PathEvent>,
quinn/src/path.rs
+14 -13
View File
@@ -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<Result<(), PathError>>,
opened: WatchStream<Result<(), OpenPathError>>,
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<Result<(), PathError>>,
opened: watch::Receiver<Result<(), OpenPathError>>,
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<Path, PathError>;
type Output = Result<Path, OpenPathError>;
fn poll(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<Self::Output> {
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<PathStatus, ClosedPath> {
pub fn status(&self) -> Result<PathStatus, NotOpen> {
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<Duration>,
) -> Result<Option<Duration>, ClosedPath> {
) -> Result<Option<Duration>, 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<Duration>,
) -> Result<Option<Duration>, ClosedPath> {
) -> Result<Option<Duration>, 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<AddressDiscovery, ClosedPath> {
pub fn observed_external_addr(&self) -> Result<AddressDiscovery, NotOpen> {
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<SocketAddr, ClosedPath> {
pub fn remote_address(&self) -> Result<SocketAddr, NotOpen> {
let state = self.conn.state.lock("per_path_remote_address");
state.inner.path_remote_address(self.id)
}