//! Implementation of the _Connection Rendezvous Protocol_. use std::collections::HashMap;
use duplicate::duplicate_item; use educe::Educe; use libthreema_macros::{DebugVariantNames, VariantNames}; use prost::Message as _; use rand::{self, Rng as _}; use tracing::{debug, trace, warn}; use zeroize::ZeroizeOnDrop;
/// An error occurred while running the Connection Rendezvous Protocol. /// /// Note: Errors can occur when using the API incorrectly or when the remote party behaves /// incorrectly. Since the Connection Rendezvous Protocol is short-lived and all involved parties /// are required to behave on all paths, none of these errors are considered recoverable. /// /// When encountering an error: /// /// 1. Let `error` be the provided [`RendezvousProtocolError`]. /// 2. Abort the protocol due to `error`. /// /// TODO(LIB-30): Align errors with other protocols #[derive(Debug, thiserror::Error)] #[cfg_attr(feature = "uniffi", derive(uniffi::Error), uniffi(flat_error))] pubenum RendezvousProtocolError { /// Invalid parameter provided by foreign code. #[cfg(feature = "uniffi")] #[error("Invalid parameter: {0}")]
InvalidParameter(&'static str),
/// Exhausted the available sequence numbers to use for sending/receiving frames. #[error("Sequence number would overflow")]
SequenceNumberOverflow,
/// Oversized frame received. #[error("Oversized frame of {0} bytes")]
OversizedFrame(usize),
/// Unable to decrypt a frame's payload. #[error("Decryption failed")]
DecryptionFailed,
/// Unable to encrypt a frame's payload. #[error("Encryption failed")]
EncryptionFailed,
/// Unable to decode a protobuf message. #[error("Decoding failed: {0}")]
ProtobufDecodeFailed(#[from] prost::DecodeError),
/// Unexpected frame received (e.g. during the nomination phase where only one role is allowed /// to send frames). #[error("Frame received unexpectedly")]
UnexpectedFrame,
/// Unable to find the referenced path. It was either never created or already dropped due to /// nomination of another path. #[error("Unknown or dropped path with PID {0}")]
UnknownOrDroppedPath(u32),
/// The referenced path has been closed (most likely due to a previous error encountered on the /// path). #[error("Path with PID {0} is closed")]
PathClosed(u32),
/// The local role does not allow for nomination. #[error("Nomination is not allowed for the local role")]
NominateNotAllowed,
/// Nomination already occurred for a path. #[error("Nomination already occurred for PID {0}")]
NominationAlreadyDone(u32),
/// Nomination is not allowed in the current state. #[error("Nomination is not allowed in state '{0}'")]
InvalidStateForNomination(&'static str),
/// Nomination is required before sending ULP data. #[error("Nomination is required before sending ULP data")]
NominationRequired,
}
/// Rendezvous Path Hash (RPH), derived from the Shared Transport Key (STK). pubstruct RendezvousPathHash(pub [u8; 32]);
/// A path state update. #[derive(DebugVariantNames, VariantNames)] pubenum PathStateUpdate { /// The handshake on this path was successful and is await nomination (or being dropped).
AwaitingNominate { /// RTT measured during the handshake, to be used by the nominator to select a path.
measured_rtt: Duration,
},
/// The path has been nominated, allowing for ULP frames to be exchanged.
Nominated { /// The Rendezvous Path Hash (RPH) of the nominated path.
rph: RendezvousPathHash,
},
}
/// Result returned when interacting with the protocol state machine. The result is associated to /// the path whose PID was used when calling a function that yielded this result. /// /// When handling this result, run the following steps: /// /// 1. Let `path` be the context of the path whose PID was used in the function call that yielded this result. /// 2. If the current phase is the _handshake and nominaton phase_: /// 1. If `incoming_ulp_data` is present, abort the protocol due to an error and abort these steps. /// 2. If `outgoing_frame` is present, enqueue it to be sent on `path`. /// 3. If `state_update` is [`PathStateUpdate::AwaitingNominate`] and the protocol took the role of the /// nominator, run the _Path Awaiting Nomination Steps_ with `path` and abort these steps. /// 4. If `state_update` is [`PathStateUpdate::Nominated`]: /// 1. Mark `path` as _nominated_. /// 2. If the protocol took the role of the nominee, mark all other paths except `path` as _disregarded_ /// and close them (for WebSocket, use close code `1000`). /// 3. If the protocol took the rule of the nominator, mark all other paths except `path` as /// _disregarded_ but keep them open with a timeout of 15s after which they should be closed (for /// WebSocket, use close code `1000`).[^close-race] /// 4. Enter the _ULP phase_. At this point, the ULP may start creating frames through /// [`RendezvousProtocol::create_ulp_frame`]. /// 5. (Unreachable) /// 3. If the current phase is the _ULP phase_: /// 1. If `path` is not marked as _nominated_, abort the protocol due to an error and abort these steps. /// 2. If `state_update` is present, abort the protocol due to an error and abort these steps. /// 3. If `outgoing_frame` is present, enqueue it to be sent on `path`. /// 4. If `incoming_ulp_data` is present, hand it off to the ULP. /// 4. (Unreachable) /// /// [^close-race]: This prevents a race condition between RID nominating a path and path close /// detection on RRD's side. #[derive(Debug)] pubstruct PathProcessResult { /// The path's state updated. pub state_update: Option<PathStateUpdate>, /// An outgoing frame is ready to be sent on the path. pub outgoing_frame: Option<OutgoingFrame>, /// An incoming frame has been reassembled and is ready to be handed off to the ULP. pub incoming_ulp_data: Option<Vec<u8>>,
}
/// Ephemeral transport key (ETK). struct EphemeralTransportKey(x25519::SharedSecretHSalsa20);
/// Protocol context passed around to the various roles and states. struct Context {
is_nominator: bool,
ak: AuthenticationKey,
} impl Context { fn new(is_nominator: bool, ak: AuthenticationKey) -> Self { Self { is_nominator, ak }
}
}
/// Path states of RID. #[derive(VariantNames, DebugVariantNames)] enum RidPathState { /// Briefly used internally when moving from one state to another.
Invalid,
/// Awaiting an `RrdToRid.Hello` to start the handshake.
AwaitingHello { authentication_keys: rxdak::ForRid },
/// Sent an `RidToRrd.AuthHello`, awaiting an `RrdToRid.Auth`.
AwaitingAuth {
authentication_keys: rxdak::ForRid,
sent_at: Instant,
local_challenge: Challenge,
shared_etk: EphemeralTransportKey,
},
/// Expecting a `Nominate` to be sent or received (depending on the configuration).
AwaitingNominate {
transport_keys: rxdtk::ForRid,
rph: RendezvousPathHash,
},
/// The connection path was `Nominate`d and can now be used by the ULP.
Nominated { transport_keys: rxdtk::ForRid },
RidPathState::AwaitingNominate { mut transport_keys,
rph,
} => { // Check if the remote side is allowed to `Nominate`. if ctx.is_nominator { return Err(RendezvousProtocolError::UnexpectedFrame);
}
// States that must have been covered by code above
RidPathState::Invalid | RidPathState::Nominated { .. } | RidPathState::Closed => {
unreachable!("State should have been handled")
},
}
.map(|(state, result)| { self.state = state;
debug!(state = ?self.state, "Changed state");
result
})
}
.map_err(|error| { self.state = RidPathState::Closed;
warn!(?error, state = ?self.state, "Closed due to error");
error
})
}
fn nominate(&mutself) -> Result<PathProcessResult, RendezvousProtocolError> { // Ensure we are in the correct state to nominate if !matches!(&self.state, RidPathState::AwaitingNominate { .. }) { return Err(RendezvousProtocolError::InvalidStateForNomination( self.state.variant_name(),
));
}
// Encode and encrypt `RidToRrd.AuthHello` let (local_challenge, shared_etk, outgoing_frame) = { // Generate a challenge let local_challenge = Challenge::random();
// Generate local part of ETK let local_etk = x25519::EphemeralSecret::random_from_rng(rand::thread_rng());
/// Path states of RID. #[derive(VariantNames, DebugVariantNames)] enum RrdPathState { /// Briefly used internally when moving from one state to another.
Invalid,
/// Sent an `RrdtoRid.Hello`, awaiting an `RidToRrd.AuthHello`.
AwaitingAuthHello {
authentication_keys: rxdak::ForRrd,
sent_at: Instant,
local_challenge: Challenge,
local_etk: x25519::EphemeralSecret,
},
/// Expecting a `Nominate` to be sent or received (depending on the configuration).
AwaitingNominate {
transport_keys: rxdtk::ForRrd,
rph: RendezvousPathHash,
},
/// The connection path was `Nominate`d and can now be used by the ULP.
Nominated { transport_keys: rxdtk::ForRrd },
fn process_frame(
&mutself,
ctx: &Context, mut incoming_frame: IncomingFrame,
) -> Result<PathProcessResult, RendezvousProtocolError> {
trace!(state = ?self.state, ?incoming_frame, "Processing frame"); iflet RrdPathState::Nominated { transport_keys } = &mutself.state { // Handle `Nominated` state where the transport can be used by the ULP. Self::handle_ulp_data(transport_keys, incoming_frame).map(|incoming_ulp_data| PathProcessResult {
state_update: None,
outgoing_frame: None,
incoming_ulp_data: Some(incoming_ulp_data),
})
} else { // Handle `Closed` state iflet RrdPathState::Closed = &self.state { return Err(RendezvousProtocolError::PathClosed(self.pid));
}
// Handle states that should immediately transition into another state // // IMPORTANT: All match arms must be infallible! match core::mem::replace(&mutself.state, RrdPathState::Invalid) {
RrdPathState::AwaitingAuthHello { mut authentication_keys,
sent_at,
local_challenge,
local_etk,
} => { // Calculate RTT let measured_rtt = Instant::now().duration_since(sent_at);
RrdPathState::AwaitingNominate { mut transport_keys,
rph,
} => { // Check if the remote side is allowed to `Nominate`. if ctx.is_nominator { return Err(RendezvousProtocolError::UnexpectedFrame);
}
// States that must have been covered by code above
RrdPathState::Invalid | RrdPathState::Nominated { .. } | RrdPathState::Closed => {
unreachable!("State should have been handled")
},
}
.map(|(state, result)| { self.state = state;
debug!(state = ?self.state, "Changed state");
result
})
}
.map_err(|error| { self.state = RrdPathState::Closed;
warn!(?error, state = ?self.state, "Closed due to error");
error
})
}
fn nominate(&mutself) -> Result<PathProcessResult, RendezvousProtocolError> { // Ensure we are in the correct state to nominate if !matches!(&self.state, RrdPathState::AwaitingNominate { .. }) { return Err(RendezvousProtocolError::InvalidStateForNomination( self.state.variant_name(),
));
}
// Check if the frame exceeds the maximum supported length of this state let max_length = if matches!(&self.state, path_state_type::Nominated { .. }) {
FrameDecoder::MAX_LENGTH_AFTER_NOMINATION
} else {
FrameDecoder::MAX_LENGTH_BEFORE_NOMINATION
}; if length > max_length { return Err(RendezvousProtocolError::OversizedFrame(length));
}
Ok(())
}
/// Internal protocol state. #[derive(DebugVariantNames, VariantNames)] enum ProtocolState { /// The paths are currently racing, meaning we are in the _handshake and nomination phase_.
RacingPaths(HashMap<u32, Box<dyn Path>>),
/// One path has been nominated, all other paths have been discarded, meaning we are in the /// _ULP phase_.
Nominated { pid: u32, path: Box<dyn Path> },
}
/// Connection Rendezvous Protocol state machine. /// /// The protocol state machine can be constructed from a formerly exchanged a `RendezvousInit` and /// the associated roles by using [`RendezvousProtocol::new_as_rid`] and /// [`RendezvousProtocol::new_as_rrd`]. /// /// Any interaction with the protocol state machine that changes the internal state will yield a /// [`PathProcessResult`] that must be handled according to its documentation. /// /// The protocol goes through exactly two phases: /// /// - The _handshake and nomination phase_ where all paths are racing the handshake simultaneously until one /// has been nominated by the nominator. /// - The _ULP phase_ where ULP frames can be exchanged on the nominated path. /// /// The following steps are defined as the _Path Awaiting Nomination Steps_: /// /// 1. Let `path` be the associated path. /// 2. If the protocol did not take the role of the nominator, abort the protocol due to an error and abort /// these steps. /// 3. If `path` is the only path, run [`RendezvousProtocol::nominate_path`] for `path` and abort these steps. /// 4. (Unreachable / TODO(LIB-10): As of today, only one path is expected to be used.) /// /// When a path closed, run the following steps: /// /// 1. Let `path` be the path that closed (initiated locally or remotely). /// 2. If `path` is marked as _nominated_, abort the protocol normally with any available close information /// and abort these steps. /// 3. If `path` is marked as _disregarded_, log a notice and abort these steps. /// 4. If `path` is the last path that closed (i.e. all other paths already closed or there are no other /// paths), log a warning that all paths closed before nomination, abort the protocol normally with any /// available close information and abort these steps. /// 5. Log a warning that `path` closed before nomination. /// /// When receiving data on a path: /// /// 1. Run [`RendezvousProtocol::add_chunks`] with the respective path's PID. /// 2. In a loop, run [`RendezvousProtocol::process_frame`] with the respective path's PID and handle the /// result until it no longer produces a [`PathProcessResult`]. /// /// When the protocol is being aborted: /// /// 1. Let `cause` be an error or any information associated to normal closure. /// 2. Log the protocol abort due to `cause` as a notice or an error respectively. /// 3. Tear down the protocol state machine. /// 4. Close all remaining paths exceptionally. /// 5. Hand off `cause` to the ULP. #[derive(Educe)] #[educe(Debug)] pubstruct RendezvousProtocol { #[educe(Debug(ignore))]
ctx: Context,
state: ProtocolState,
}
// TODO(LIB-11): Add construction of the `RendezvousInit` from the paths here. impl RendezvousProtocol { /// Create a new Connection Rendezvous Protocol as the Rendezvous Initiator Device (RID) from a /// formerly exchanged `RendezvousInit`. /// /// `pids` must contain the set of available pre-initiated paths with their associated Path IDs /// (PID). /// /// Returns the protocol state machine instance. #[tracing::instrument(skip_all, fields(?is_nominator, ?pids))] pubfn new_as_rid(is_nominator: bool, ak: AuthenticationKey, pids: &[u32]) -> Self {
debug!("Creating D2D rendezvous protocol"); let ctx = Context::new(is_nominator, ak);
/// Create a new Connection Rendezvous Protocol as the Rendezvous Responder Device (RRD) from a /// formerly exchanged `RendezvousInit`. /// /// `pids` must contain the set of available pre-initiated paths with their associated Path IDs /// (PID). /// /// Returns a tuple of the protocol state machine instance and a list of PIDs and outgoing /// frames to be enqueued on the respective paths immediately. #[tracing::instrument(skip_all, fields(?is_nominator, ?pids))] pubfn new_as_rrd(
is_nominator: bool,
ak: AuthenticationKey,
pids: &[u32],
) -> (Self, Vec<(u32, OutgoingFrame)>) {
debug!("Creating protocol"); let ctx = Context::new(is_nominator, ak); letmut outgoing_frames = vec![];
/// Add chunks received on the specified path. The chunks may or may not contain complete frames /// or even contain multiple complete frames. /// /// # Errors /// /// Returns [`RendezvousProtocolError::UnknownOrDroppedPath`] if the path associated to `pid` /// could not be found. #[tracing::instrument(skip_all, fields(
?self, ?pid,
chunks_byte_length = chunks.iter().map(|chunk| chunk.len()).sum::<usize>(),
))] pubfn add_chunks(&mutself, pid: u32, chunks: &[&[u8]]) -> Result<(), RendezvousProtocolError> { let path = Self::lookup_path(&mutself.state, pid)?;
path.add_chunks(chunks)
}
/// Process any available buffered complete frame for the specified path. /// /// # Errors /// /// Returns [`RendezvousProtocolError`] for a plethora of reasons, e.g. if the path associated /// to `pid` could not be found, an incoming frame could not be decoded or decrypted, or an /// unexpected message was received, or, as a response to it, another outgoing frame could not /// be encrypted. #[tracing::instrument(skip_all, fields(?self, ?pid))] pubfn process_frame(&mutself, pid: u32) -> Result<Option<PathProcessResult>, RendezvousProtocolError> { let path = Self::lookup_path(&mutself.state, pid)?;
// Decode and process the next frame, if any can be decoded let result = path.process_frame(&self.ctx)?;
trace!(?result, "Processed frame"); let Some(result) = result else { return Ok(result);
};
// Update state if the path was nominated and we are still racing paths iflet (ProtocolState::RacingPaths(racing_paths), Some(PathStateUpdate::Nominated { .. })) =
(&mutself.state, &result.state_update)
{ // Nominate the path let path = racing_paths
.remove(&pid)
.ok_or(RendezvousProtocolError::UnknownOrDroppedPath(pid))?;
debug!(
dropped_pids = ?racing_paths.keys(), "Remote nominated, dropping all other paths"
); self.state = ProtocolState::Nominated { pid, path };
}
// Return the result
Ok(Some(result))
}
/// Nominate a path. /// /// # Errors /// /// Returns [`RendezvousProtocolError`] if the protocol did not take the role of the nominator, /// the path associated to `pid` could not be found, the path is not ready to be nominated or /// nomination already happened. #[tracing::instrument(skip_all, fields(?self, ?pid))] pubfn nominate_path(&mutself, pid: u32) -> Result<PathProcessResult, RendezvousProtocolError> { // Ensure we're allowed to nominate if !self.ctx.is_nominator { return Err(RendezvousProtocolError::NominateNotAllowed);
}
// Nominate if the paths are still racing match &mutself.state {
ProtocolState::RacingPaths(racing_paths) => { // Attempt to nominate the path letmut path = racing_paths
.remove(&pid)
.ok_or(RendezvousProtocolError::UnknownOrDroppedPath(pid))?; let result = path.nominate()?;
debug!(
dropped_pids = ?racing_paths.keys(), "Local nominated, dropping all other paths"
); self.state = ProtocolState::Nominated { pid, path };
Ok(result)
},
/// Create a ULP frame to be encrypted and sent as an outgoing frame on the nominated path. /// /// # Errors /// /// Returns [`RendezvousProtocolError`] if nomination of a path is still pending or the ULP /// frame could not be encrypted or encoded. #[tracing::instrument(skip_all, fields(?self, outgoing_data_length = outgoing_data.len()))] pubfn create_ulp_frame(
&mutself,
outgoing_data: Vec<u8>,
) -> Result<PathProcessResult, RendezvousProtocolError> { match &mutself.state {
ProtocolState::RacingPaths(..) => Err(RendezvousProtocolError::NominationRequired),
ProtocolState::Nominated { path, .. } => path.create_ulp_frame(outgoing_data),
}
}
fn lookup_path(
state: &mut ProtocolState,
pid: u32,
) -> Result<&mutBox<dyn Path>, RendezvousProtocolError> { // Lookup path based on the current state. let path = match state { // The nomination race is still ongoing. Lookup the path by its PID.
ProtocolState::RacingPaths(racing_paths) => racing_paths
.get_mut(&pid)
.ok_or(RendezvousProtocolError::UnknownOrDroppedPath(pid))?,
// There's only one nominated path. Ensure it's the correct one.
ProtocolState::Nominated {
pid: nominated_pid,
path,
} => { if pid != *nominated_pid { return Err(RendezvousProtocolError::UnknownOrDroppedPath(pid));
}
path
},
};
Ok(path)
}
}
Messung V0.5 in Prozent
¤ Dauer der Verarbeitung: 0.24 Sekunden
(vorverarbeitet am 2026-06-26)
¤
Die Informationen auf dieser Webseite wurden
nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit,
noch Qualität der bereit gestellten Informationen zugesichert.
Bemerkung:
Die farbliche Syntaxdarstellung und die Messung sind noch experimentell.