mod idle; pubmod params; mod state; #[cfg(any(test, feature = "build-fuzzing-corpus"))] #[cfg_attr(coverage_nightly, coverage(off))] pubmod test_internal;
use idle::IdleTimeout; pubuse params::ConnectionParameters; use params::PreferredAddressConfig; use state::StateSignaling; pubuse state::{ClosingFrame, State};
#[derive(Clone, Debug, PartialEq, Eq)] /// Type returned from `process()` and `process_output()`. Users are required to /// call these repeatedly until `Callback` or `None` is returned. pubenum Output { /// Connection requires no action.
None, /// Connection requires the datagram be sent.
Datagram(Datagram), /// Connection requires `process_input()` be called when the `Duration` /// elapses.
Callback(Duration),
}
impl TryFrom<OutputBatch> for Output { type Error = ();
#[derive(Clone, Debug, PartialEq, Eq)] pubenum OutputBatch { /// Connection requires no action.
None, /// Connection requires the datagram batch be sent.
DatagramBatch(datagram::Batch), /// Connection requires `process_input()` be called when the `Duration` /// elapses.
Callback(Duration),
}
impl From<Output> for OutputBatch { fn from(value: Output) -> Self { match value {
Output::None => Self::None,
Output::Datagram(dg) => Self::DatagramBatch(datagram::Batch::from(dg)),
Output::Callback(t) => Self::Callback(t),
}
}
}
/// Get a reference to the Datagram, if any. #[must_use] pubconstfn as_dgram_ref(&self) -> Option<&Datagram> { matchself { Self::Datagram(dg) => Some(dg),
_ => None,
}
}
/// Ask how long the caller should wait before calling back. #[must_use] pubconstfn callback(&self) -> Duration { matchself { Self::Callback(t) => *t,
_ => Duration::new(0, 0),
}
}
}
/// Used by inner functions like `Connection::output`. enum SendOptionBatch { /// Yes, please send this datagram.
Yes(datagram::Batch), /// Don't send. If this was blocked on the pacer (the arg is true).
No(bool),
}
/// Used by inner functions like `Connection::output`. enum SendOption { /// Yes, please send this datagram.
Yes, /// Don't send.
No( /// Whether this was blocked on the pacer.
bool,
),
}
/// Used by `Connection::preprocess` to determine what to do /// with an packet before attempting to remove protection. #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum PreprocessResult { /// End processing and return successfully.
End, /// Stop processing this datagram and move on to the next.
Next, /// Continue and process this packet. Continue,
}
/// `AddressValidationInfo` holds information relevant to either /// responding to address validation (`NewToken`, `Retry`) or generating /// tokens for address validation (`Server`). enum AddressValidationInfo {
None, // We are a client and have information from `NEW_TOKEN`.
NewToken(Vec<u8>), // We are a client and have received a `Retry` packet.
Retry {
token: Vec<u8>,
retry_source_cid: ConnectionId,
}, // We are a server and can generate tokens.
Server(Weak<RefCell<AddressValidation>>),
}
pubfn generate_new_token(&self, peer_address: SocketAddr, now: Instant) -> Option<Vec<u8>> { matchself { Self::Server(w) => w
.upgrade()?
.borrow()
.generate_new_token(peer_address, now)
.ok(), Self::None => None,
_ => unreachable!("called a server function on a client"),
}
}
}
/// A QUIC Connection /// /// First, create a new connection using `new_client()` or `new_server()`. /// /// For the life of the connection, handle activity in the following manner: /// 1. Perform operations using the `stream_*()` methods. /// 1. Call `process_input()` when a datagram is received or the timer expires. Obtain information /// on connection state changes by checking `events()`. /// 1. Having completed handling current activity, repeatedly call `process_output()` for packets to /// send, until it returns `Output::Callback` or `Output::None`. /// /// After the connection is closed (either by calling `close()` or by the /// remote) continue processing until `state()` returns `Closed`. pubstruct Connection {
role: Role,
version: Version,
state: State,
tps: Rc<RefCell<TransportParametersHandler>>, /// What we are doing with 0-RTT.
zero_rtt_state: ZeroRttState, /// All of the network paths that we are aware of.
paths: Paths, /// This object will generate connection IDs for the connection.
cid_manager: ConnectionIdManager,
address_validation: AddressValidationInfo, /// The connection IDs that were provided by the peer.
cids: ConnectionIdStore<Srt>,
/// The source connection ID that this endpoint uses for the handshake. /// Since we need to communicate this to our peer in tparams, setting this /// value is part of constructing the struct.
local_initial_source_cid: ConnectionId, /// The source connection ID from the first packet from the other end. /// This is checked against the peer's transport parameters.
remote_initial_source_cid: Option<ConnectionId>, /// The destination connection ID from the first packet from the client. /// This is checked by the client against the server's transport parameters.
original_destination_cid: Option<ConnectionId>,
/// We sometimes save a datagram against the possibility that keys will later /// become available. This avoids reporting packets as dropped during the handshake /// when they are either just reordered or we haven't been able to install keys yet. /// In particular, this occurs when asynchronous certificate validation happens.
saved_datagrams: SavedDatagrams, /// Some packets were received, but not tracked.
received_untracked: bool,
crypto: Crypto,
acks: AckTracker,
idle_timeout: IdleTimeout,
streams: Streams,
state_signaling: StateSignaling,
loss_recovery: recovery::Loss,
events: ConnectionEvents,
new_token: NewTokenState,
stats: StatsCell,
qlog: Qlog, /// A session ticket was received without `NEW_TOKEN`, /// this is when that turns into an event without `NEW_TOKEN`.
release_resumption_token_timer: Option<Instant>,
conn_params: ConnectionParameters,
hrtime: hrtime::Handle,
/// For testing purposes it is sometimes necessary to inject frames that wouldn't /// otherwise be sent, just to see how a connection handles them. Inserting them /// into packets proper mean that the frames follow the entire processing path. #[cfg(any(test, feature = "build-fuzzing-corpus"))]
test_frame_writer: Option<Box<dyn test_internal::FrameWriter>>,
}
impl Connection { /// A long default for timer resolution, so that we don't tax the /// system too hard when we don't need to. const LOOSE_TIMER_RESOLUTION: Duration = Duration::from_millis(50); /// The SCONE indicator. const SCONE_INDICATION: &[u8] = &[0xc8, 0x13];
/// Create a new QUIC connection with Client role. /// # Errors /// When NSS fails and an agent cannot be created. pubfn new_client<I: Into<String>, A: AsRef<str>>(
server_name: I,
protocols: &[A],
cid_generator: Rc<RefCell<dyn ConnectionIdGenerator>>,
local_addr: SocketAddr,
remote_addr: SocketAddr,
conn_params: ConnectionParameters,
now: Instant,
) -> Res<Self> { let dcid = ConnectionId::generate_initial(); letmut c = Self::new(
Role::Client,
Agent::from(Client::new(server_name.into(), conn_params.is_greasing())?),
cid_generator,
protocols,
conn_params,
)?;
c.crypto.states_mut().init(
c.conn_params.get_versions().compatible(),
Role::Client,
&dcid,
c.conn_params.randomize_first_pn_enabled(),
)?;
c.original_destination_cid = Some(dcid); let path = Path::temporary(
local_addr,
remote_addr,
&c.conn_params,
Qlog::default(),
now,
&mut c.stats.borrow_mut(),
);
c.setup_handshake_path(&Rc::new(RefCell::new(path)), now);
Ok(c)
}
/// Create a new QUIC connection with Server role. /// # Errors /// When NSS fails and an agent cannot be created. pubfn new_server<A1: AsRef<str>, A2: AsRef<str>>(
certs: &[A1],
protocols: &[A2],
cid_generator: Rc<RefCell<dyn ConnectionIdGenerator>>,
conn_params: ConnectionParameters,
) -> Res<Self> { Self::new(
Role::Server,
Agent::from(Server::new(certs)?),
cid_generator,
protocols,
conn_params,
)
}
/// Get the active ECH configuration, which is empty if ECH is disabled. #[must_use] pubfn ech_config(&self) -> &[u8] { self.crypto.ech_config()
}
/// # Errors /// When the operation fails. pubfn client_enable_ech<A: AsRef<[u8]>>(&mutself, ech_config_list: A) -> Res<()> { self.crypto.client_enable_ech(ech_config_list)
}
/// Set or clear the qlog for this connection. pubfn set_qlog(&mutself, qlog: Qlog) { self.loss_recovery.set_qlog(qlog.clone()); self.paths.set_qlog(qlog.clone()); self.qlog = qlog;
}
/// Get the qlog (if any) for this connection. pubconstfn qlog_mut(&mutself) -> &mut Qlog {
&mutself.qlog
}
/// Get the original destination connection id for this connection. This /// will always be present for `Role::Client` but not if `Role::Server` is in /// `State::Init`. #[must_use] pubconstfn odcid(&self) -> Option<&ConnectionId> { self.original_destination_cid.as_ref()
}
/// Set a local transport parameter, possibly overriding a default value. /// This only sets transport parameters without dealing with other aspects of /// setting the value. /// /// # Errors /// When the transport parameter is invalid. /// # Panics /// This panics if the transport parameter is known to this crate. #[cfg(test)] pubfn set_local_tparam(
&self,
tp: TransportParameterId,
value: tparams::TransportParameter,
) -> Res<()> { if *self.state() == State::Init { self.tps.borrow_mut().local_mut().set(tp, value);
Ok(())
} else {
qerror!("Current state: {:?}", self.state());
qerror!("Cannot set local tparam when not in an initial connection state");
Err(Error::ConnectionState)
}
}
/// `odcid` is their original choice for our CID, which we get from the Retry token. /// `remote_cid` is the value from the Source Connection ID field of an incoming packet: what /// the peer wants us to use now. `retry_cid` is what we asked them to use when we sent the /// Retry. pub(crate) fn set_retry_cids(
&mutself,
odcid: &ConnectionId,
remote_cid: ConnectionId,
retry_cid: &ConnectionId,
) {
debug_assert_eq!(self.role, Role::Server);
qtrace!("[{self}] Retry CIDs: odcid={odcid} remote={remote_cid} retry={retry_cid}"); // We advertise "our" choices in transport parameters. self.tps
.borrow_mut()
.local_mut()
.set_bytes(OriginalDestinationConnectionId, odcid.to_vec()); self.tps
.borrow_mut()
.local_mut()
.set_bytes(RetrySourceConnectionId, retry_cid.to_vec());
// ...and save their choices for later validation. self.remote_initial_source_cid = Some(remote_cid);
}
/// Set ALPN preferences. Strings that appear earlier in the list are given /// higher preference. /// # Errors /// When the operation fails, which is usually due to bad inputs or bad connection state. pubfn set_alpn<A: AsRef<[u8]>>(&mutself, protocols: &[A]) -> Res<()> { self.crypto.tls_mut().set_alpn(protocols)?;
Ok(())
}
/// Enable a set of ciphers. /// # Errors /// When the operation fails, which is usually due to bad inputs or bad connection state. pubfn set_ciphers(&mutself, ciphers: &[Cipher]) -> Res<()> { ifself.state != State::Init {
qerror!("[{self}] Cannot enable ciphers in state {:?}", self.state); return Err(Error::ConnectionState);
} self.crypto.tls_mut().set_ciphers(ciphers)?;
Ok(())
}
/// Enable a set of key exchange groups. /// # Errors /// When the operation fails, which is usually due to bad inputs or bad connection state. pubfn set_groups(&mutself, groups: &[Group]) -> Res<()> { ifself.state != State::Init {
qerror!("[{self}] Cannot enable groups in state {:?}", self.state); return Err(Error::ConnectionState);
} self.crypto.tls_mut().set_groups(groups)?;
Ok(())
}
/// Set the number of additional key shares to send in the client hello. /// # Errors /// When the operation fails, which is usually due to bad inputs or bad connection state. pubfn send_additional_key_shares(&mutself, count: usize) -> Res<()> { ifself.state != State::Init {
qerror!("[{self}] Cannot enable groups in state {:?}", self.state); return Err(Error::ConnectionState);
} self.crypto.tls_mut().send_additional_key_shares(count)?;
Ok(())
}
fn make_resumption_token(&mutself) -> ResumptionToken {
debug_assert_eq!(self.role, Role::Client);
debug_assert!(self.crypto.has_resumption_token()); // Values less than GRANULARITY are ignored when using the token, so use 0 where needed. let rtt = self.paths.primary().map_or_else( // If we don't have a path, we don't have an RTT.
|| Duration::from_millis(0),
|p| { let rtt = p.borrow().rtt().estimate(); if p.borrow().rtt().is_guesstimate() { // When we have no actual RTT sample, do not encode a guestimated RTT larger // than the default initial RTT. (The guess can be very large under lossy // conditions.) if rtt < self.conn_params.get_initial_rtt() {
rtt
} else {
Duration::from_millis(0)
}
} else {
rtt
}
},
);
self.crypto
.create_resumption_token( self.new_token.take_token(), self.tps
.borrow()
.remote_handshake()
.as_ref()
.expect("should have transport parameters"), self.version,
u64::try_from(rtt.as_millis()).unwrap_or(0),
)
.expect("caller checked if a resumption token existed")
}
/// Get the simplest PTO calculation for all those cases where we need /// a value of this approximate order. Don't use this for loss recovery, /// only use it where a more precise value is not important. fn pto(&self) -> Duration { self.paths.primary().map_or_else(
|| RttEstimate::new(self.conn_params.get_initial_rtt()).pto(self.confirmed()),
|p| p.borrow().rtt().pto(self.confirmed()),
)
}
whileself.crypto.has_resumption_token() && self.new_token.has_token() { let token = self.make_resumption_token(); self.events.client_resumption_token(token);
}
// If we have a resumption ticket check or set a timer. ifself.crypto.has_resumption_token() { let arm = iflet Some(expiration_time) = self.release_resumption_token_timer { if expiration_time <= now { let token = self.make_resumption_token(); self.events.client_resumption_token(token); self.release_resumption_token_timer = None;
// This means that we release one session ticket every 3 PTOs // if no NEW_TOKEN frame is received. self.crypto.has_resumption_token()
} else { false
}
} else { true
};
if arm { self.release_resumption_token_timer = Some(now + 3 * self.pto());
}
}
}
/// The correct way to obtain a resumption token is to wait for the /// `ConnectionEvent::ResumptionToken` event. To emit the event we are waiting for a /// resumption token and a `NEW_TOKEN` frame to arrive. Some servers don't send `NEW_TOKEN` /// frames and in this case, we wait for 3xPTO before emitting an event. This is especially a /// problem for short-lived connections, where the connection is closed before any events are /// released. This function retrieves the token, without waiting for a `NEW_TOKEN` frame to /// arrive. /// /// # Panics /// /// If this is called on a server. pubfn take_resumption_token(&mutself, now: Instant) -> Option<ResumptionToken> {
assert_eq!(self.role, Role::Client);
/// Enable resumption, using a token previously provided. /// This can only be called once and only on the client. /// After calling the function, it should be possible to attempt 0-RTT /// if the token supports that. /// /// This function starts the TLS stack, which means that any configuration change /// to that stack needs to occur prior to calling this. /// /// # Errors /// When the operation fails, which is usually due to bad inputs or bad connection state. pubfn enable_resumption<A: AsRef<[u8]>>(&mutself, now: Instant, token: A) -> Res<()> { ifself.state != State::Init {
qerror!("[{self}] set token in state {:?}", self.state); return Err(Error::ConnectionState);
} ifself.role == Role::Server { return Err(Error::ConnectionState);
}
let version = Version::try_from(
dec.decode_uint::<version::Wire>()
.ok_or(Error::InvalidResumptionToken)?,
)?;
qtrace!("[{self}] version {version:?}"); if !self.conn_params.get_versions().all().contains(&version) { return Err(Error::DisabledVersion);
}
let rtt = Duration::from_millis(dec.decode_varint().ok_or(Error::InvalidResumptionToken)?);
qtrace!("[{self}] RTT {rtt:?}");
let tp_slice = dec.decode_vvec().ok_or(Error::InvalidResumptionToken)?;
qtrace!("[{self}] transport parameters {}", hex(tp_slice)); letmut dec_tp = Decoder::from(tp_slice); let tp =
TransportParameters::decode(&mut dec_tp).map_err(|_| Error::InvalidResumptionToken)?;
let init_token = dec.decode_vvec().ok_or(Error::InvalidResumptionToken)?;
qtrace!("[{self}] Initial token {}", hex(init_token));
let tok = dec.decode_remainder();
qtrace!("[{self}] TLS token {}", hex(tok));
matchself.crypto.tls_mut() {
Agent::Client(c) => { let res = c.enable_resumption(tok); iflet Err(e) = res { self.absorb_error::<Error>(now, Err(Error::from(e))); return Ok(());
}
}
Agent::Server(_) => return Err(Error::WrongRole),
}
self.version = version; self.conn_params.get_versions_mut().set_initial(version); self.tps.borrow_mut().set_version(version); self.tps.borrow_mut().set_remote_0rtt(Some(tp)); if !init_token.is_empty() { self.address_validation = AddressValidationInfo::NewToken(init_token.to_vec());
} self.paths
.primary()
.ok_or(Error::Internal)?
.borrow_mut()
.rtt_mut()
.set_initial(rtt); self.set_initial_limits(); // Start up TLS, which has the effect of setting up all the necessary // state for 0-RTT. This only stages the CRYPTO frames. let res = self.client_start(now); self.absorb_error(now, res);
Ok(())
}
/// Send a TLS session ticket AND a `NEW_TOKEN` frame (if possible). /// # Errors /// When the operation fails, which is usually due to bad inputs or bad connection state. pubfn send_ticket(&mutself, now: Instant, extra: &[u8]) -> Res<()> { ifself.role == Role::Client { return Err(Error::WrongRole);
}
// If we are able, also send a NEW_TOKEN frame. // This should be recording all remote addresses that are valid, // but there are just 0 or 1 in the current implementation. matchself.paths.primary() {
Some(path) => { iflet Some(token) = self
.address_validation
.generate_new_token(path.borrow().remote_address(), now)
{ self.new_token.send_new_token(token);
}
Ok(())
}
None => Err(Error::NotConnected),
}
}
/// # Errors /// When there is no information to obtain. pubfn tls_preinfo(&self) -> Res<SecretAgentPreInfo> {
Ok(self.crypto.tls().preinfo()?)
}
/// Get the peer's certificate chain and other info. #[must_use] pubfn peer_certificate(&self) -> Option<CertificateInfo> { self.crypto.tls().peer_certificate()
}
/// Call by application when the peer cert has been verified. /// /// This panics if there is no active peer. It's OK to call this /// when authentication isn't needed, that will likely only cause /// the connection to fail. However, if no packets have been /// exchanged, it's not OK. pubfn authenticated(&mutself, status: AuthenticationStatus, now: Instant) {
qdebug!("[{self}] Authenticated {status:?}"); self.crypto.tls_mut().authenticated(status); let res = self.handshake(now, self.version, PacketNumberSpace::Handshake, None); self.absorb_error(now, res); self.process_saved(now);
}
/// Get the role of the connection. #[must_use] pubconstfn role(&self) -> Role { self.role
}
/// Get the state of the connection. #[must_use] pubconstfn state(&self) -> &State {
&self.state
}
/// The QUIC version in use. #[must_use] pubconstfn version(&self) -> Version { self.version
}
/// Get the 0-RTT state of the connection. #[must_use] pubconstfn zero_rtt_state(&self) -> ZeroRttState { self.zero_rtt_state
}
/// Get a snapshot of collected statistics. #[must_use] pubfn stats(&self) -> Stats { letmut v = self.stats.borrow().clone();
v.version = self.version; iflet Some(p) = self.paths.primary() { let p = p.borrow();
v.rtt = p.rtt().estimate();
v.rttvar = p.rtt().rttvar();
v.min_rtt = p.rtt().minimum();
}
v
}
// This function wraps a call to another function and sets the connection state // properly if that call fails. fn capture_error<T>(
&mutself,
path: Option<PathRef>,
now: Instant,
frame_type: FrameType,
res: Res<T>,
) -> Res<T> { iflet Err(v) = &res { #[cfg(debug_assertions)] let msg = format!("{v:?}"); #[cfg(not(debug_assertions))] let msg = ""; let error = CloseReason::Transport(v.clone()); match &self.state {
State::Closing { error: err, .. }
| State::Draining { error: err, .. }
| State::Closed(err) => {
qwarn!("[{self}] Closing again after error {err:?}");
}
State::Init => { // We have not even sent anything just close the connection without sending any // error. This may happen when client_start fails. self.set_state(State::Closed(error), now);
}
State::WaitInitial | State::WaitVersion => { // We don't have any state yet, so don't bother with // the closing state, just send one CONNECTION_CLOSE. iflet Some(path) = path.or_else(|| self.paths.primary()) { self.state_signaling
.close(path, error.clone(), frame_type, msg);
} self.set_state(State::Closed(error), now);
}
_ => match path.or_else(|| self.paths.primary()) {
Some(path) => { self.state_signaling
.close(path, error.clone(), frame_type, msg); if matches!(v, Error::KeysExhausted) { self.set_state(State::Closed(error), now);
} else { self.set_state(
State::Closing {
error,
timeout: self.get_closing_period_time(now),
},
now,
);
}
}
None => { self.set_state(State::Closed(error), now);
}
},
}
}
res
}
/// For use with `process_input()`. Errors there can be ignored, but this /// needs to ensure that the state is updated. fn absorb_error<T>(&mutself, now: Instant, res: Res<T>) -> Option<T> { self.capture_error(None, now, FrameType::Padding, res).ok()
}
fn process_timer(&mutself, now: Instant) { match &self.state { // Only the client runs timers while waiting for Initial packets.
State::WaitInitial => debug_assert_eq!(self.role, Role::Client), // If Closing or Draining, check if it is time to move to Closed.
State::Closing { error, timeout } | State::Draining { error, timeout } if *timeout <= now =>
{ let st = State::Closed(error.clone()); self.set_state(st, now);
qinfo!("Closing timer expired"); return;
}
State::Closed(_) => {
qdebug!("Timer fired while closed"); return;
}
_ => (),
}
if !self
.paths
.process_timeout(now, pto, &mutself.stats.borrow_mut())
{
qinfo!("[{self}] last available path failed"); self.absorb_error::<Error>(now, Err(Error::NoAvailablePath));
}
}
/// Whether the given [`ConnectionIdRef`] is a valid local [`ConnectionId`]. #[must_use] pubfn is_valid_local_cid(&self, cid: ConnectionIdRef) -> bool { self.cid_manager.is_valid(cid)
}
/// Process a new input datagram on the connection. pubfn process_input<A: AsRef<[u8]> + AsMut<[u8]>>(&mutself, d: Datagram<A>, now: Instant) { self.process_multiple_input(iter::once(d), now);
}
/// Process new input datagrams on the connection. pubfn process_multiple_input<
A: AsRef<[u8]> + AsMut<[u8]>,
I: IntoIterator<Item = Datagram<A>>,
>(
&mutself,
dgrams: I,
now: Instant,
) { letmut dgrams = dgrams.into_iter().peekable(); if dgrams.peek().is_none() { return;
}
// Snapshot timer type before ACKs can alter loss state. iflet Some(path) = self.paths.primary() { self.loss_recovery.note_timeout_type(&path.borrow(), now);
} for d in dgrams { self.input(d, now, now);
} self.process_saved(now); self.streams.cleanup_closed_streams();
}
/// Get the time that we next need to be called back, relative to `now`. fn next_delay(&mutself, now: Instant, paced: bool) -> Duration {
qtrace!("[{self}] Get callback delay {now:?}");
// Only one timer matters when closing... iflet State::Closing { timeout, .. } | State::Draining { timeout, .. } = self.state { self.hrtime.update(Self::LOOSE_TIMER_RESOLUTION); return timeout.duration_since(now);
}
// `release_resumption_token_timer` is not considered here, because // it is not important enough to force the application to set a // timeout for it It is expected that other activities will // drive it.
let earliest = delays.into_iter().min().expect("at least one delay"); // TODO(agrover, mt) - need to analyze and fix #47 // rather than just clamping to zero here.
debug_assert!(earliest > now); let delay = earliest.saturating_duration_since(now);
qdebug!("[{self}] delay duration {delay:?}"); self.hrtime.update(delay / 4);
delay
}
/// Wrapper around [`Connection::process_multiple_output`] that processes a /// single output datagram only. #[expect(clippy::missing_panics_doc, reason = "see expect()")] #[must_use = "Output of the process_output function must be handled"] pubfn process_output(&mutself, now: Instant) -> Output { self.process_multiple_output(now, 1.try_into().expect(">0"))
.try_into()
.expect("max_datagrams is 1")
}
/// Get output packets, as a result of receiving packets, or actions taken /// by the application. /// Returns datagrams to send, and how long to wait before calling again /// even if no incoming packets. #[must_use = "OutputBatch of the process_multiple_output function must be handled"] pubfn process_multiple_output(
&mutself,
now: Instant,
max_datagrams: NonZeroUsize,
) -> OutputBatch {
qtrace!("[{self}] process_output {:?} {now:?}", self.state);
match (&self.state, self.role) {
(State::Init, Role::Client) => { let res = self.client_start(now); self.absorb_error(now, res);
}
(State::Init | State::WaitInitial, Role::Server) => { return OutputBatch::None;
}
_ => { self.process_timer(now);
}
}
/// A test-only output function that uses the provided writer to /// pack something extra into the output. #[cfg(any(test, feature = "build-fuzzing-corpus"))] pubfn test_write_frames<W>(&mutself, writer: W, now: Instant) -> Output where
W: test_internal::FrameWriter + 'static,
{ self.test_frame_writer = Some(Box::new(writer)); let res = self.process_output(now); self.test_frame_writer = None;
res
}
/// Wrapper around [`Connection::process_multiple`], processing a single /// input and single output datagram only. #[expect(clippy::missing_panics_doc, reason = "see expect()")] #[must_use = "Output of the process function must be handled"] pubfn process<A: AsRef<[u8]> + AsMut<[u8]>>(
&mutself,
dgram: Option<Datagram<A>>,
now: Instant,
) -> Output { self.process_multiple(dgram, now, 1.try_into().expect(">0"))
.try_into()
.expect("max_datagrams is 1")
}
/// Process input and generate output. #[must_use = "OutputBatch of the process_multiple function must be handled"] pubfn process_multiple<A: AsRef<[u8]> + AsMut<[u8]>>(
&mutself,
dgram: Option<Datagram<A>>,
now: Instant,
max_datagrams: NonZeroUsize,
) -> OutputBatch { iflet Some(d) = dgram { // Snapshot timer type before ACKs can alter loss state. iflet Some(path) = self.paths.primary() { self.loss_recovery.note_timeout_type(&path.borrow(), now);
} self.input(d, now, now); self.process_saved(now);
} let output = self.process_multiple_output(now, max_datagrams); #[cfg(feature = "build-fuzzing-corpus")] ifself.test_frame_writer.is_none()
&& let OutputBatch::DatagramBatch(batch) = &output
{ for dgram in batch.iter() {
neqo_common::write_item_to_fuzzing_corpus("packet", &dgram);
}
}
output
}
fn handle_retry(&mutself, packet: &packet::Public, now: Instant) -> Res<()> {
qinfo!("[{self}] received Retry"); if matches!(self.address_validation, AddressValidationInfo::Retry { .. }) { self.stats.borrow_mut().pkt_dropped("Extra Retry"); return Ok(());
} if packet.token().is_empty() { self.stats.borrow_mut().pkt_dropped("Retry without a token"); return Ok(());
} if !packet.is_valid_retry( self.original_destination_cid
.as_ref()
.ok_or(Error::InvalidRetry)?,
) { self.stats
.borrow_mut()
.pkt_dropped("Retry with bad integrity tag"); return Ok(());
} // At this point, we should only have the connection ID that we generated. // Update to the one that the server prefers. let Some(path) = self.paths.primary() else { self.stats
.borrow_mut()
.pkt_dropped("Retry without an existing path"); return Ok(());
};
fn discard_keys(&mutself, space: PacketNumberSpace, now: Instant) { ifself.crypto.discard(space) {
qdebug!("[{self}] Drop packet number space {space}"); iflet Some(path) = self.paths.primary() { self.loss_recovery.discard(&path, space, now);
} self.acks.drop_space(space);
}
}
fn is_stateless_reset(&self, path: &PathRef, d: &[u8]) -> bool { // If the datagram is too small, don't try. // If the connection is connected, then the reset token will be invalid. if d.len() < Srt::LEN || !self.state.connected() { returnfalse;
}
Srt::try_from(&d[d.len() - Srt::LEN..])
.is_ok_and(|token| path.borrow().is_stateless_reset(&token))
}
fn check_stateless_reset(
&mutself,
path: &PathRef,
d: &[u8],
first: bool,
now: Instant,
) -> Res<()> { if first && self.is_stateless_reset(path, d) { // Failing to process a packet in a datagram might // indicate that there is a stateless reset present.
qdebug!( "[{self}] Stateless reset: {}",
hex(&d[d.len() - Srt::LEN..])
); self.state_signaling.reset(); self.set_state(
State::Draining {
error: CloseReason::Transport(Error::StatelessReset),
timeout: self.get_closing_period_time(now),
},
now,
);
Err(Error::StatelessReset)
} else {
Ok(())
}
}
/// Process any saved datagrams that might be available for processing. fn process_saved(&mutself, now: Instant) { whilelet Some(epoch) = self.saved_datagrams.available() {
qdebug!("[{self}] process saved for epoch {epoch:?}");
debug_assert!( self.crypto
.states_mut()
.rx_hp(self.version, epoch)
.is_some()
); for saved inself.saved_datagrams.take_saved() {
qtrace!("[{self}] input saved @{:?}: {:?}", saved.t, saved.d); self.input(saved.d, saved.t, now);
}
}
}
/// In case a datagram arrives that we can only partially process, save any /// part that we don't have keys for. #[expect(
clippy::needless_pass_by_value,
reason = "To consume an owned datagram below."
)] fn save_datagram(
&mutself,
epoch: Epoch,
d: Datagram<impl AsRef<[u8]>>,
remaining: usize,
now: Instant,
) { let d = Datagram::new(
d.source(),
d.destination(),
d.tos(),
d[d.len() - remaining..].to_vec(),
); self.saved_datagrams.save(epoch, d, now); self.stats.borrow_mut().saved_datagrams += 1; // We already counted the datagram as received in [`input_path`]. We // will do so again when we (re-)process it, so reduce the count now. self.stats.borrow_mut().packets_rx -= 1;
}
qinfo!("[{self}] Version negotiation: trying {version:?}"); let path = self.paths.primary().ok_or(Error::NoAvailablePath)?; let local_addr = path.borrow().local_address(); let remote_addr = path.borrow().remote_address(); let conn_params = self
.conn_params
.clone()
.versions(version, self.conn_params.get_versions().all().to_vec()); letmut c = Self::new_client( self.crypto.server_name().ok_or(Error::VersionNegotiation)?, self.crypto.protocols(), self.cid_manager.generator(),
local_addr,
remote_addr,
conn_params,
now,
)?;
c.conn_params
.get_versions_mut()
.set_initial(self.conn_params.get_versions().initial());
mem::swap(self, &mut c);
qlog::client_version_information_negotiated(
&mutself.qlog, self.conn_params.get_versions().all(),
supported,
version,
now,
);
Ok(())
} else {
qinfo!("[{self}] Version negotiation: failed with {supported:?}"); // This error goes straight to closed. self.set_state(
State::Closed(CloseReason::Transport(Error::VersionNegotiation)),
now,
);
Err(Error::VersionNegotiation)
}
}
/// Perform any processing that we might have to do on packets prior to /// attempting to remove protection. #[expect(clippy::too_many_lines, reason = "Yeah, it's a work in progress.")] fn preprocess_packet(
&mutself,
packet: &packet::Public,
path: &PathRef,
dcid: Option<&ConnectionId>,
now: Instant,
) -> Res<PreprocessResult> { if dcid.is_some_and(|d| d != &packet.dcid()) { self.stats
.borrow_mut()
.pkt_dropped("Coalesced packet has different DCID"); return Ok(PreprocessResult::Next);
}
if (packet.packet_type() == packet::Type::Initial
|| packet.packet_type() == packet::Type::Handshake)
&& self.role == Role::Client
&& !path.borrow().is_primary()
{ // If we have received a packet from a different address than we have sent to // we should ignore the packet. In such a case a path will be a newly created // temporary path, not the primary path. return Ok(PreprocessResult::Next);
}
match (packet.packet_type(), &self.state, &self.role) {
(packet::Type::Initial, State::Init, Role::Server) => { let version = packet.version().ok_or(Error::ProtocolViolation)?; if !packet.is_valid_initial()
|| !self.conn_params.get_versions().all().contains(&version)
{ self.stats.borrow_mut().pkt_dropped("Invalid Initial"); return Ok(PreprocessResult::Next);
}
qinfo!( "[{self}] Received valid Initial packet with scid {:?} dcid {:?}",
packet.scid(),
packet.dcid()
); // Record the client's selected CID so that it can be accepted until // the client starts using a real connection ID. let dcid = ConnectionId::from(packet.dcid()); self.crypto.states_mut().init_server(
version,
&dcid, self.conn_params.randomize_first_pn_enabled(),
)?; self.original_destination_cid = Some(dcid); self.set_state(State::WaitInitial, now);
// We need to make sure that we set this transport parameter. // This has to happen prior to processing the packet so that // the TLS handshake has all it needs. if !self.retry_sent() { self.tps
.borrow_mut()
.local_mut()
.set_bytes(OriginalDestinationConnectionId, packet.dcid().to_vec());
}
}
(packet::Type::VersionNegotiation, State::WaitInitial, Role::Client) => { iflet Ok(versions) = packet.supported_versions() { if versions.is_empty()
|| versions.contains(&self.version().wire_version())
|| versions.contains(&0)
|| &packet.scid() != self.odcid().ok_or(Error::Internal)?
|| matches!(self.address_validation, AddressValidationInfo::Retry { .. })
{ // Ignore VersionNegotiation packets that contain the current version. // Or don't have the right connection ID. // Or are received after a Retry. self.stats.borrow_mut().pkt_dropped("Invalid VN");
} else { self.version_negotiation(&versions, now)?;
}
} else { self.stats.borrow_mut().pkt_dropped("VN with no versions");
} return Ok(PreprocessResult::End);
}
(packet::Type::Retry, State::WaitInitial, Role::Client) => { self.handle_retry(packet, now)?; return Ok(PreprocessResult::Next);
}
(packet::Type::Handshake | packet::Type::Short, State::WaitInitial, Role::Client) // This packet can't be processed now, but it could be a sign // that Initial packets were lost. // Resend Initial CRYPTO frames immediately a few times just // in case. As we don't have an RTT estimate yet, this helps // when there is a short RTT and losses. Also mark all 0-RTT // data as lost. if dcid.is_none()
&& self.cid_manager.is_valid(packet.dcid())
&& !self.saved_datagrams.is_either_full()
=> {
qtrace!("Resending Initial in response to an undecryptable packet"); self.crypto.resend_unacked(PacketNumberSpace::Initial); self.resend_0rtt(now);
}
(
packet::Type::VersionNegotiation | packet::Type::Retry | packet::Type::OtherVersion,
..,
) => { self.stats
.borrow_mut()
.pkt_dropped(format!("{:?}", packet.packet_type())); return Ok(PreprocessResult::Next);
}
_ => {}
}
let res = matchself.state {
State::Init => { self.stats
.borrow_mut()
.pkt_dropped("Received while in Init state");
PreprocessResult::Next
}
State::WaitInitial => PreprocessResult::Continue,
State::WaitVersion | State::Handshaking | State::Connected | State::Confirmed => { ifself.cid_manager.is_valid(packet.dcid()) { ifself.role == Role::Server && packet.packet_type() == packet::Type::Handshake
{ // Server has received a Handshake packet -> discard Initial keys and states self.discard_keys(PacketNumberSpace::Initial, now);
}
PreprocessResult::Continue
} else { self.stats
.borrow_mut()
.pkt_dropped(format!("Invalid DCID {:?}", packet.dcid()));
PreprocessResult::Next
}
}
State::Closing { .. } => { // Don't bother processing the packet. Instead ask to get a // new close frame. // // > In the closing state, an endpoint retains only enough // > information to generate a packet containing a // > CONNECTION_CLOSE frame and to identify packets as belonging // > to the connection. An endpoint in the closing state sends a // > packet containing a CONNECTION_CLOSE frame in response to any // > incoming packet that it attributes to the connection. // // <https://www.rfc-editor.org/rfc/rfc9000.html#section-10.2.1-2> self.state_signaling.send_close();
PreprocessResult::Next
}
State::Draining { .. } | State::Closed(..) => { // Do nothing. self.stats
.borrow_mut()
.pkt_dropped(format!("State {:?}", self.state));
PreprocessResult::Next
}
};
Ok(res)
}
/// After a Initial, Handshake, `ZeroRtt`, or Short packet is successfully processed. #[expect(clippy::too_many_arguments, reason = "Yes, but they're needed.")] fn postprocess_packet(
&mutself,
path: &PathRef,
tos: Tos,
remote: SocketAddr,
packet: &packet::Decrypted,
packet_number: packet::Number,
migrate: bool,
now: Instant,
) { let ecn_mark = Ecn::from(tos); letmut stats = self.stats.borrow_mut();
stats.ecn_rx[packet.packet_type()] += ecn_mark; iflet Some(last_ecn_mark) = stats.ecn_last_mark.filter(|&last_ecn_mark| {
last_ecn_mark != ecn_mark && stats.ecn_rx_transition[last_ecn_mark][ecn_mark].is_none()
}) {
stats.ecn_rx_transition[last_ecn_mark][ecn_mark] =
Some((packet.packet_type(), packet_number));
}
stats.ecn_last_mark = Some(ecn_mark);
drop(stats); let space = PacketNumberSpace::from(packet.packet_type()); iflet Some(space) = self.acks.get_mut(space) {
*space.ecn_marks() += ecn_mark;
} else {
qtrace!("Not tracking ECN for dropped packet number space");
}
/// Take a datagram as input. This reports an error if the packet was bad. /// This takes two times: when the datagram was received, and the current time. fn input(
&mutself,
d: Datagram<impl AsRef<[u8]> + AsMut<[u8]>>,
received: Instant,
now: Instant,
) { // First determine the path. let path = self.paths.find_path(
d.destination(),
d.source(),
&self.conn_params,
now,
&mutself.stats.borrow_mut(),
);
path.borrow_mut().add_received(d.len()); let res = self.input_path(&path, d, received);
_ = self.capture_error(Some(path), now, FrameType::Padding, res);
}
// Handle each packet in the datagram. while !slc.is_empty() { self.stats.borrow_mut().packets_rx += 1; self.stats.borrow_mut().dscp_rx[tos.into()] += 1; let slc_len = slc.len(); let (packet, remainder) = match packet::Public::decode(slc, self.cid_manager.decoder().as_ref()) {
Ok((packet, remainder)) => { #[cfg(feature = "build-fuzzing-corpus")]
neqo_common::write_item_to_fuzzing_corpus("packet", packet.data());
(packet, remainder)
}
Err(e) => {
qinfo!("[{self}] Garbage packet: {e}"); self.stats.borrow_mut().pkt_dropped("Garbage packet"); break;
}
}; matchself.preprocess_packet(&packet, path, dcid.as_ref(), now)? {
PreprocessResult::Continue => (),
PreprocessResult::Next => break,
PreprocessResult::End => return Ok(()),
}
qtrace!("[{self}] Received unverified packet {packet:?}");
let packet_len = packet.len(); match packet.decrypt(self.crypto.states_mut(), now + pto) {
Ok(payload) => { // OK, we have a valid packet. let pn = payload.pn(); self.idle_timeout.on_packet_received(now); self.log_packet(
packet::MetaData::new_in(path, tos, packet_len, &payload, self.version),
now,
);
let space = PacketNumberSpace::from(payload.packet_type()); iflet Some(space) = self.acks.get_mut(space) { if space.is_duplicate(pn) {
qdebug!("Duplicate packet {space}-{pn}"); self.stats.borrow_mut().dups_rx += 1;
} else { matchself.process_packet(path, &payload, now) {
Ok(migrate) => { self.postprocess_packet(
path, tos, remote, &payload, pn, migrate, now,
);
}
Err(e) => { self.ensure_error_path(path, &payload, now); return Err(e);
}
}
}
} else {
qdebug!( "[{self}] Received packet {space} for untracked space {}",
payload.pn()
); return Err(Error::ProtocolViolation);
}
dcid = Some(ConnectionId::from(payload.dcid()));
}
Err(e) => { match e.error {
Error::KeysPending(epoch) => { // This packet can't be decrypted because we don't have the keys yet. // Don't check this packet for a stateless reset, just return. let remaining = slc_len; self.save_datagram(epoch, d, remaining, now); return Ok(());
}
Error::KeysExhausted => { // Exhausting read keys is fatal. return Err(e.error);
}
Error::KeysDiscarded(epoch) => self.handle_keys_discarded(epoch),
_ => (),
} // Decryption failure, or not having keys is not fatal. // If the state isn't available, or we can't decrypt the packet, drop // the rest of the datagram on the floor, but don't generate an error. self.check_stateless_reset(path, e.data, dcid.is_none(), now)?; self.stats.borrow_mut().pkt_dropped("Decryption failure");
qlog::packet_dropped(&mutself.qlog, &e, now);
dcid = Some(e.dcid);
}
}
slc = remainder;
} self.check_stateless_reset(path, &d, dcid.is_none(), now)?;
Ok(())
}
/// Handle receiving a packet for which keys have been discarded. fn handle_keys_discarded(&mutself, epoch: Epoch) { // Client: receiving undecryptable Initial packets while waiting // indicates server's Initial was lost. Probe with Handshake. self.received_untracked |= self.role == Role::Client && epoch == Epoch::Initial;
// Server: receiving undecryptable Handshake packets while Confirmed // indicates the client hasn't received HANDSHAKE_DONE. Resend it. ifself.role == Role::Server && epoch == Epoch::Handshake && self.state == State::Confirmed
{ self.state_signaling.handshake_done();
}
}
/// Process a packet. Returns true if the packet might initiate migration. fn process_packet(
&mutself,
path: &PathRef,
packet: &packet::Decrypted,
now: Instant,
) -> Res<bool> {
(!packet.is_empty())
.then_some(())
.ok_or(Error::ProtocolViolation)?;
// TODO(ekr@rtfm.com): Have the server blow away the initial // crypto state if this fails? Otherwise, we will get a panic // on the assert for doesn't exist. // OK, we have a valid packet.
// Get the next packet number we'll send, for ACK verification. // This is used by `input_frame` to verify that ACKs don't acknowledge unsent packets. let next_pn = self
.crypto
.states()
.select_tx(self.version, PacketNumberSpace::from(packet.packet_type()))
.map_or(0, |(_, tx)| tx.next_pn());
letmut ack_eliciting = false; letmut probing = true; letmut d = Decoder::from(&packet[..]); while d.remaining() > 0 { #[cfg(feature = "build-fuzzing-corpus")] let pos = d.offset(); let f = Frame::decode(&mut d)?; #[cfg(feature = "build-fuzzing-corpus")]
neqo_common::write_item_to_fuzzing_corpus("frame", &packet[pos..d.offset()]);
ack_eliciting |= f.ack_eliciting();
probing &= f.path_probing(); let t = f.get_type(); iflet Err(e) = self.input_frame(
path,
packet.version(),
packet.packet_type(),
f,
next_pn,
now,
) { self.capture_error(Some(Rc::clone(path)), now, t, Err(e))?;
}
}
let largest_received = iflet Some(space) = self
.acks
.get_mut(PacketNumberSpace::from(packet.packet_type()))
{
space.set_received(
now,
packet.pn(),
ack_eliciting,
&mutself.stats.borrow_mut(),
)?
} else {
qdebug!( "[{self}] processed a {:?} packet without tracking it",
packet.packet_type(),
); // This was a valid packet that caused the same packet number to be // discarded. This happens when the client discards the Initial packet // number space after receiving the ServerHello. Remember this so // that we guarantee that we send a Handshake packet. self.received_untracked = true; // We don't migrate during the handshake, so return false. false
};
Ok(largest_received && !probing)
}
/// During connection setup, the first path needs to be setup. /// This uses the connection IDs that were provided during the handshake /// to setup that path. fn setup_handshake_path(&mutself, path: &PathRef, now: Instant) { self.paths.make_permanent(
path,
Some(self.local_initial_source_cid.clone()), // Ideally we know what the peer wants us to use for the remote CID. // But we will use our own guess if necessary.
ConnectionIdEntry::initial_remote( self.remote_initial_source_cid
.as_ref()
.or(self.original_destination_cid.as_ref())
.expect("have either remote_initial_source_cid or original_destination_cid")
.clone(),
),
now,
); ifself.role == Role::Client {
path.borrow_mut().set_valid(now);
}
}
/// If the path isn't permanent, assign it a connection ID to make it so. fn ensure_permanent(&mutself, path: &PathRef, now: Instant) -> Res<()> { ifself.paths.is_temporary(path) { // If there isn't a connection ID to use for this path, the packet // will be processed, but it won't be attributed to a path. That means // no path probes or PATH_RESPONSE. But it's not fatal. matchself.cids.next() {
Some(cid) => { self.paths.make_permanent(path, None, cid, now);
Ok(())
}
None => { iflet Some(primary) = self.paths.primary() { if primary.borrow().remote_cid().is_none_or(|id| id.is_empty()) { self.paths.make_permanent(
path,
None,
ConnectionIdEntry::empty_remote(),
now,
);
Ok(())
} else {
qtrace!("[{self}] Unable to make path permanent: {}", path.borrow());
Err(Error::InvalidMigration)
}
} else {
qtrace!("[{self}] Unable to make path permanent: {}", path.borrow());
Err(Error::InvalidMigration)
}
}
}
} else {
Ok(())
}
}
/// After an error, a permanent path is needed to send the `CONNECTION_CLOSE`. /// This attempts to ensure that this exists. As the connection is now /// temporary, there is no reason to do anything special here. fn ensure_error_path(&mutself, path: &PathRef, packet: &packet::Decrypted, now: Instant) {
path.borrow_mut().set_valid(now); ifself.paths.is_temporary(path) { // First try to fill in handshake details. if packet.packet_type() == packet::Type::Initial { self.remote_initial_source_cid = Some(ConnectionId::from(packet.scid())); self.setup_handshake_path(path, now);
} else { // Otherwise try to get a usable connection ID.
drop(self.ensure_permanent(path, now));
}
}
}
ifself.role == Role::Server { let Some(original_destination_cid) = self.original_destination_cid.as_ref() else {
qdebug!("[{self}] No original destination DCID"); return;
}; self.cid_manager.add_odcid(original_destination_cid.clone()); // Make a path on which to run the handshake. self.setup_handshake_path(path, now);
} else {
qdebug!("[{self}] Changing to use Server CID={}", packet.scid());
debug_assert!(path.borrow().is_primary());
path.borrow_mut().set_remote_cid(packet.scid());
}
}
/// Migrate to the provided path. /// Either local or remote address (but not both) may be provided as `None` to have /// the address from the current primary path used. /// If `force` is true, then migration is immediate. /// Otherwise, migration occurs after the path is probed successfully. /// Either way, the path is probed and will be abandoned if the probe fails. /// /// # Errors /// /// Fails if this is not a client, not confirmed, the peer disabled connection migration, or /// there are not enough connection IDs available to use. pubfn migrate(
&mutself,
local: Option<SocketAddr>,
remote: Option<SocketAddr>,
force: bool,
now: Instant,
) -> Res<()> { ifself.role != Role::Client { return Err(Error::InvalidMigration);
} if !matches!(self.state(), State::Confirmed) { return Err(Error::InvalidMigration);
} ifself.tps.borrow().remote().get_empty(DisableMigration) { return Err(Error::InvalidMigration);
}
// Fill in the blanks, using the current primary path. if local.is_none() && remote.is_none() { // Pointless migration is pointless. return Err(Error::InvalidMigration);
}
let path = self.paths.primary().ok_or(Error::InvalidMigration)?; let local = local.unwrap_or_else(|| path.borrow().local_address()); let remote = remote.unwrap_or_else(|| path.borrow().remote_address());
if mem::discriminant(&local.ip()) != mem::discriminant(&remote.ip()) { // Can't mix address families. return Err(Error::InvalidMigration);
} if local.port() == 0 || remote.ip().is_unspecified() || remote.port() == 0 { // All but the local address need to be specified. return Err(Error::InvalidMigration);
} if (local.ip().is_loopback() ^ remote.ip().is_loopback()) && !local.ip().is_unspecified() { // Block attempts to migrate to a path with loopback on only one end, unless the local // address is unspecified. return Err(Error::InvalidMigration);
}
fn path_migrated(&self, path: &PathRef) { let p = path.borrow(); self.events
.path_migrated(p.local_address(), p.remote_address());
}
fn migrate_to_preferred_address(&mutself, now: Instant) -> Res<()> { let spa: Option<(tparams::PreferredAddress, ConnectionIdEntry<Srt>)> = if matches!( self.conn_params.get_preferred_address(),
PreferredAddressConfig::Disabled
) {
qdebug!("[{self}] Preferred address is disabled");
None
} else { self.tps.borrow_mut().remote().get_preferred_address()
}; iflet Some((addr, cid)) = spa { // The connection ID isn't special, so just save it. self.cids.add_remote(cid)?;
// The preferred address doesn't dictate what the local address is, so this // has to use the existing address. So only pay attention to a preferred // address from the same family as is currently in use. More thought will // be needed to work out how to get addresses from a different family. let prev = self
.paths
.primary()
.ok_or(Error::NoAvailablePath)?
.borrow()
.remote_address(); let remote = match prev.ip() {
IpAddr::V4(_) => addr.ipv4().map(SocketAddr::V4),
IpAddr::V6(_) => addr.ipv6().map(SocketAddr::V6),
};
iflet Some(remote) = remote { // Ignore preferred address that move to loopback from non-loopback. // `migrate` doesn't enforce this rule. if !prev.ip().is_loopback() && remote.ip().is_loopback() {
qwarn!("[{self}] Ignoring a move to a loopback address: {remote}"); return Ok(());
}
ifself.migrate(None, Some(remote), false, now).is_err() {
qwarn!("[{self}] Ignoring bad preferred address: {remote}");
}
} else {
qwarn!("[{self}] Unable to migrate to a different address family");
}
} else {
qdebug!("[{self}] No preferred address to migrate to");
}
Ok(())
}
ifself.ensure_permanent(path, now).is_ok() { let was_primary = path.borrow().is_primary(); self.paths
.handle_migration(path, remote, now, &mutself.stats.borrow_mut()); if !was_primary { self.path_migrated(path);
}
} else {
qinfo!( "[{self}] {} Peer migrated, but no connection ID available",
path.borrow()
);
}
}
fn output(&mutself, now: Instant, max_datagrams: NonZeroUsize) -> SendOptionBatch {
qtrace!("[{self}] output {now:?}"); let res = match &self.state {
State::Init
| State::WaitInitial
| State::WaitVersion
| State::Handshaking
| State::Connected
| State::Confirmed => self.paths.select_path().map_or_else(
|| Ok(SendOptionBatch::default()),
|path| { let res = self.output_dgram_batch_on_path(&path, now, None, max_datagrams); self.capture_error(Some(path), now, FrameType::Padding, res)
},
),
State::Closing { .. } | State::Draining { .. } | State::Closed(_) => { self.state_signaling.close_frame().map_or_else(
|| Ok(SendOptionBatch::default()),
|details| { let path = Rc::clone(details.path()); // In some error cases, we will not be able to make a new, permanent path. // For example, if we run out of connection IDs and the error results from // a packet on a new path, we avoid sending (and the privacy risk) rather // than reuse a connection ID. let res = if path.borrow().is_temporary() {
qerror!("[{self}] Attempting to close with a temporary path");
Err(Error::Internal)
} else { self.output_dgram_batch_on_path(
&path,
now,
Some(&details),
max_datagrams,
)
}; self.capture_error(Some(path), now, FrameType::Padding, res)
},
)
}
};
res.unwrap_or_default()
}
let pn = tx.next_pn(); let unacked_range = largest_acknowledged.map_or_else(|| pn + 1, |la| (pn - la) << 1); // Count how many bytes in this range are non-zero. let pn_len = size_of::<packet::Number>()
- usize::try_from(unacked_range.leading_zeros() / 8).expect("u32 fits in usize");
assert!(
pn_len > 0, "pn_len can't be zero as unacked_range should be > 0, pn {pn}, largest_acknowledged {largest_acknowledged:?}, tx {tx}"
); // TODO(mt) also use `4*path CWND/path MTU` to set a minimum length.
builder.pn(pn, pn_len);
/// Write the frames that are exchanged in the application data space. /// The order of calls here determines the relative priority of frames. fn write_appdata_frames(
&mutself,
builder: &mut packet::Builder<&mut Vec<u8>>,
tokens: &mut recovery::Tokens,
now: Instant,
) { let rtt = self.paths.primary().map_or_else(
|| RttEstimate::new(self.conn_params.get_initial_rtt()).estimate(),
|p| p.borrow().rtt().estimate(),
);
let stats = &mutself.stats.borrow_mut(); let frame_stats = &mut stats.frame_tx; ifself.role == Role::Server
&& let Some(t) = self.state_signaling.write_done(builder)
{
tokens.push(t);
frame_stats.handshake_done += 1;
}
self.streams
.write_frames(TransmissionPriority::Critical, builder, tokens, frame_stats); if builder.is_full() { return;
}
// NEW_CONNECTION_ID, RETIRE_CONNECTION_ID, and ACK_FREQUENCY. self.cid_manager.write_frames(builder, tokens, frame_stats); if builder.is_full() { return;
}
self.paths.write_frames(builder, tokens, frame_stats); if builder.is_full() { return;
}
for prio in [TransmissionPriority::High, TransmissionPriority::Normal] { self.streams
.write_frames(prio, builder, tokens, &mut stats.frame_tx); if builder.is_full() { return;
}
}
// Datagrams are best-effort and unreliable. Let streams starve them for now. self.quic_datagrams.write_frames(builder, tokens, stats); if builder.is_full() { return;
}
// CRYPTO here only includes NewSessionTicket, plus NEW_TOKEN. // Both of these are only used for resumption and so can be relatively low priority. let frame_stats = &mut stats.frame_tx; self.crypto.write_frame(
PacketNumberSpace::ApplicationData, self.conn_params.sni_slicing_enabled(),
builder,
tokens,
frame_stats,
); if builder.is_full() { return;
}
self.new_token.write_frames(builder, tokens, frame_stats); if builder.is_full() { return;
}
// Maybe send a probe. Return true if the packet was ack-eliciting. fn maybe_probe<B: Buffer>(
&mutself,
path: &PathRef,
force_probe: bool,
builder: &mut packet::Builder<B>,
ack_end: usize,
tokens: &mut recovery::Tokens,
now: Instant,
) -> bool { let untracked = self.received_untracked && !self.state.connected(); self.received_untracked = false;
// Anything written after an ACK already elicits acknowledgment. // If we need to probe and nothing has been written, send a PING. if builder.len() > ack_end { returntrue;
}
let pto = path.borrow().rtt().pto(self.confirmed()); letmut probe = if untracked && builder.packet_empty() || force_probe { // If we received an untracked packet and we aren't probing already // or the PTO timer fired: probe. true
} elseif !builder.packet_empty() { // The packet only contains an ACK. Check whether we want to // force an ACK with a PING so we can stop tracking packets. self.loss_recovery.should_probe(pto, now)
} else { false
};
ifself.streams.need_keep_alive() { // We need to keep the connection alive, including sending a PING // again. If a PING is already scheduled (i.e. `probe` is `true`) // piggy back on it. If not, schedule one.
probe |= self.idle_timeout.send_keep_alive(now, pto, tokens);
}
if probe { // Nothing ack-eliciting and we need to probe; send PING.
debug_assert_ne!(builder.remaining(), 0);
builder.encode_frame(FrameType::Ping, |_| {}); let stats = &mutself.stats.borrow_mut().frame_tx;
stats.ping += 1;
}
probe
}
/// Write frames to the provided builder. Returns a list of tokens used for /// tracking loss or acknowledgment, whether any frame was ACK eliciting, and /// whether the packet was padded. fn write_frames(
&mutself,
path: &PathRef,
space: PacketNumberSpace,
profile: &SendProfile,
builder: &mut packet::Builder<&mut Vec<u8>>,
coalesced: bool, // Whether this packet is coalesced behind another one.
now: Instant,
) -> (recovery::Tokens, bool, bool) { letmut tokens = recovery::Tokens::new(); let primary = path.borrow().is_primary(); letmut ack_eliciting = false;
if primary { let stats = &mutself.stats.borrow_mut().frame_tx; self.acks.write_frame(
space,
now,
path.borrow().rtt().estimate(),
builder,
&mut tokens,
stats,
);
} let ack_end = builder.len();
// Avoid sending path validation probes until the handshake completes, // but send them even when we don't have space. let full_mtu = profile.limit() == path.borrow().plpmtu(); if space == PacketNumberSpace::ApplicationData && self.state.connected() { // Path validation probes should only be padded if the full MTU is available. // The probing code needs to know so it can track that. if path.borrow_mut().write_frames(
builder,
&mutself.stats.borrow_mut().frame_tx,
full_mtu,
now,
) {
builder.enable_padding(true);
}
}
if profile.ack_only() { // If we are CC limited we can only send ACKs! return (tokens, false, false);
}
if primary { if space == PacketNumberSpace::ApplicationData { ifself.state.connected()
&& path.borrow().pmtud().needs_probe()
&& !coalesced // Only send PMTUD probes using non-coalesced packets.
&& full_mtu
{
path.borrow_mut().pmtud_mut().send_probe(
builder,
&mut tokens,
&mutself.stats.borrow_mut(),
);
ack_eliciting = true;
} self.write_appdata_frames(builder, &mut tokens, now);
} else { let stats = &mutself.stats.borrow_mut().frame_tx; self.crypto.write_frame(
space, self.conn_params.sni_slicing_enabled(),
builder,
&mut tokens,
stats,
);
}
#[cfg(test)] iflet Some(w) = &mutself.test_frame_writer {
assert!(!builder.is_full(), "test_frame_writer set on full packet");
w.write_frames(builder);
}
}
// Maybe send a probe now, either to probe for losses or to keep the connection live. let force_probe = profile.should_probe(space);
ack_eliciting |= self.maybe_probe(path, force_probe, builder, ack_end, &mut tokens, now); // If this is not the primary path, this should be ack-eliciting.
debug_assert!(primary || ack_eliciting);
// Add padding. Only pad 1-RTT packets so that we don't prevent coalescing. // And avoid padding packets that otherwise only contain ACK because adding PADDING // causes those packets to consume congestion window, which is not tracked (yet). // And avoid padding if we don't have a full MTU available. let stats = &mutself.stats.borrow_mut().frame_tx; let padded = if ack_eliciting && full_mtu && builder.pad() {
stats.padding += 1; true
} else { false
};
(tokens, ack_eliciting, padded)
}
fn write_closing_frames<B: Buffer>(
&mutself,
close: &ClosingFrame,
builder: &mut packet::Builder<B>,
space: PacketNumberSpace,
now: Instant,
path: &PathRef,
tokens: &mut recovery::Tokens,
) { if builder.remaining() > ClosingFrame::MIN_LENGTH + RecvdPackets::USEFUL_ACK_LEN { // Include an ACK frame with the CONNECTION_CLOSE. let limit = builder.limit();
builder.set_limit(limit - ClosingFrame::MIN_LENGTH); self.acks.immediate_ack(space, now); self.acks.write_frame(
space,
now,
path.borrow().rtt().estimate(),
builder,
tokens,
&mutself.stats.borrow_mut().frame_tx,
);
builder.set_limit(limit);
} // CloseReason::Application is only allowed at 1RTT. let sanitized = if space == PacketNumberSpace::ApplicationData {
None
} else {
close.sanitize()
};
sanitized.as_ref().unwrap_or(close).write_frame(builder); self.stats.borrow_mut().frame_tx.connection_close += 1;
}
/// Build batch of datagrams to be sent on the provided path. fn output_dgram_batch_on_path(
&mutself,
path: &PathRef,
now: Instant, mut closing_frame: Option<&ClosingFrame>,
max_datagrams: NonZeroUsize,
) -> Res<SendOptionBatch> { let packet_tos = path.borrow().tos(); letmut send_buffer = Vec::new(); letmut max_datagram_size = None; letmut num_datagrams = 0; let mtu = path.borrow().plpmtu(); let address_family_max_mtu = path.borrow().pmtud().address_family_max_mtu();
loop { if max_datagrams.get() <= num_datagrams { break;
} if path.borrow().pmtud().needs_probe() && num_datagrams != 0 { // Next datagram will be larger due to PMTUD probing. GSO // requires that all datagrams in a batch are of equal size. // Only the last datagram can be smaller. Given that this would // not be the first datagram, close the batch early to uphold // the above GSO requirement. break;
}
let send_buffer_len_before = send_buffer.len();
// Check if we can fit another PMTUD sized datagram into the batch. if max_datagram_size.is_some_and(|datagram_size| { // GSO requires that all datagrams in a batch are of equal size. // The last datagram can be smaller. The datagrams already in // the batch are each `datagram_size` large. The next datagram // can be up to `mtu` large. Break in case the next could be // larger than the ones already in the batch.
datagram_size < mtu // GSO allows total datagram batch size up to the address family // max MTU. If the next datagram could exceed that limit, break. // // See for example Linux kernel: // https://github.com/torvalds/linux/blob/fb4d33ab452ea254e2c319bac5703d1b56d895bf/include/linux/netdevice.h#L2402
|| address_family_max_mtu - send_buffer.len() < mtu
}) { break;
}
matchself.output_dgram_on_path(
path,
now,
closing_frame.take(),
Encoder::new_borrowed_vec(&mut send_buffer),
packet_tos,
)? {
SendOption::Yes => {
debug_assert_eq!(
mtu,
path.borrow().plpmtu(), "MTU does not change within batch"
);
num_datagrams += 1; let datagram_size = send_buffer.len() - send_buffer_len_before; let max_datagram_size = *max_datagram_size.get_or_insert(datagram_size);
// GSO requires that all datagrams in a batch are of equal // size. Only the last datagram can be smaller.
debug_assert!(datagram_size <= max_datagram_size); if datagram_size < max_datagram_size { // This packet was smaller. Make sure it is the last by // breaking the loop. break;
}
}
SendOption::No(paced) => { if num_datagrams == 0 {
debug_assert!(send_buffer.is_empty()); return Ok(SendOptionBatch::No(paced));
} break;
}
}
}
/// Build a datagram, possibly from multiple packets (for different PN /// spaces) and each containing 1+ frames. #[expect(clippy::too_many_lines, reason = "Yeah, that's just the way it is.")] fn output_dgram_on_path(
&mutself,
path: &PathRef,
now: Instant,
closing_frame: Option<&ClosingFrame>, mut encoder: Encoder<&mut Vec<u8>>,
packet_tos: Tos,
) -> Res<SendOption> { letmut initial_sent = None; letmut needs_padding = false; let grease_quic_bit = self.can_grease_quic_bit(); let version = self.version();
// Determine how we are sending packets (PTO, etc..). let profile = self.loss_recovery.send_profile(&path.borrow(), now);
qdebug!("[{self}] output_dgram_on_path send_profile {profile:?}");
// Frames for different epochs must go in different packets, but then these // packets can go in a single datagram for space in PacketNumberSpace::iter() { // Ensure we have tx crypto state for this epoch, or skip it. let Some((epoch, tx)) = self.crypto.states_mut().select_tx_mut(self.version, space) else { continue;
}; let aead_expansion = tx.expansion();
let header_start = encoder.len();
// Configure the limits and padding for this packet. let limit = if path.borrow().pmtud().needs_probe() {
needs_padding = true;
debug_assert!(path.borrow().pmtud().probe_size() >= profile.limit());
path.borrow().pmtud().probe_size()
} else {
profile.limit()
- if space == PacketNumberSpace::Initial && self.conn_params.scone_enabled() { // Reserve some space for the SCONE indication in an Initial. // This reduces the amount available for building the packet, // but we'll pad to `profile.limit()` when padding. // This will not reserve space for the indication if packets // are coalesced (with Handshake or 0-RTT). That's too bad. Self::SCONE_INDICATION.len()
} else { 0
}
} - aead_expansion;
let (pt, mut builder, pn) = Self::build_packet_header(
&path.borrow(),
epoch,
encoder,
tx,
&self.address_validation,
version,
grease_quic_bit,
limit, self.loss_recovery.largest_acknowledged_pn(space),
); // The builder will set the limit to 0 if there isn't enough space for the header. if builder.is_full() {
encoder = builder.abort(); break;
}
builder.enable_padding(needs_padding); if builder.is_full() {
encoder = builder.abort(); break;
}
// Add frames to the packet. let payload_start = builder.len(); let (mut tokens, mut ack_eliciting, mut padded) =
(recovery::Tokens::new(), false, false); iflet Some(close) = closing_frame { self.write_closing_frames(close, &mut builder, space, now, path, &mut tokens);
} else {
(tokens, ack_eliciting, padded) = self.write_frames(path, space, &profile, &mut builder, header_start != 0, now);
} if builder.packet_empty() { // Nothing to include in this packet.
encoder = builder.abort();
continue;
}
if packet_tos.is_ecn_marked() {
tokens.push(recovery::Token::EcnEct0);
}
self.stats.borrow_mut().packets_tx += 1; // Track which packet types are sent with which ECN codepoints. For // coalesced packets, this increases the counts for each packet type // contained in the coalesced packet. This is per Section 13.4.1 of // RFC 9000. self.stats.borrow_mut().ecn_tx[pt] += Ecn::from(packet_tos); let tx = self
.crypto
.states_mut()
.tx_mut(self.version, epoch)
.ok_or(Error::Internal)?;
encoder = builder.build(tx)?; self.crypto.states_mut().auto_update()?;
if ack_eliciting { self.idle_timeout.on_packet_sent(now);
} let sent = sent::Packet::new(
pt,
pn,
now,
ack_eliciting,
tokens,
encoder.len() - header_start,
); if padded {
needs_padding = false; self.loss_recovery.on_packet_sent(path, sent, now);
} elseif pt == packet::Type::Initial && (self.role == Role::Client || ack_eliciting) { // Packets containing Initial packets might need padding, and we want to // track that padding along with the Initial packet. So defer tracking.
initial_sent = Some(sent);
needs_padding = true;
} else { if pt.is_long() && self.role == Role::Client && initial_sent.is_none() { // Disable padding for any long header packet if the UDP packet doesn't include // an Initial packet.
needs_padding = false;
} self.loss_recovery.on_packet_sent(path, sent, now);
}
if space == PacketNumberSpace::Handshake { ifself.role == Role::Client { // We're sending a Handshake packet, so we can discard Initial keys. self.discard_keys(PacketNumberSpace::Initial, now);
} elseifself.role == Role::Server && self.state == State::Confirmed { // We could discard handshake keys in set_state, // but wait until after sending an ACK. self.discard_keys(PacketNumberSpace::Handshake, now);
}
}
// If the client has more CRYPTO data queued up, do not coalesce if // this packet is an Initial. Without this, 0-RTT packets could be // coalesced with the first Initial, which some server (e.g., ours) // do not support, because they may not save packets they can't // decrypt yet. ifself.role == Role::Client
&& space == PacketNumberSpace::Initial
&& !self.crypto.streams_mut().is_empty(space)
{ break;
}
}
qdebug!( "[{self}] pad Initial from {} to {}",
encoder.len(),
profile.limit()
); let pad_amount = profile.limit() - encoder.len();
initial.track_padding(pad_amount); ifself.conn_params.scone_enabled() { // This ensures that the last bytes are a SCONE indication, if there is enough space. // This is not tracked, other than for congestion control (above) if pad_amount >= Self::SCONE_INDICATION.len() {
encoder.pad_to(
profile.limit() - Self::SCONE_INDICATION.len() + 1, Self::SCONE_INDICATION[0],
);
encoder.encode(&Self::SCONE_INDICATION[1..]);
} else {
encoder.pad_to(profile.limit(), Self::SCONE_INDICATION[0]);
}
} else {
encoder.pad_to(profile.limit(), 0);
}
}
/// # Errors /// When connection state is not valid. pubfn initiate_key_update(&mutself) -> Res<()> { ifself.state == State::Confirmed { let la = self
.loss_recovery
.largest_acknowledged_pn(PacketNumberSpace::ApplicationData);
qinfo!("[{self}] Initiating key update"); self.crypto.states_mut().initiate_key_update(la)
} else {
Err(Error::KeyUpdateBlocked)
}
}
/// Process the final set of transport parameters. fn process_tps(&mutself, now: Instant) -> Res<()> { self.validate_cids()?; self.validate_versions()?;
{ let tps = self.tps.borrow(); let remote = tps.remote_handshake().ok_or(Error::TransportParameter)?;
// If the peer provided a preferred address, then we have to be a client // and they have to be using a non-empty connection ID. if remote.get_preferred_address().is_some()
&& (self.role == Role::Server
|| self
.remote_initial_source_cid
.as_ref()
.ok_or(Error::UnknownConnectionId)?
.is_empty())
{ return Err(Error::TransportParameter);
}
let reset_token = remote.get_bytes(StatelessResetToken).map_or_else(
|| Ok(Srt::random()),
|token| Srt::try_from(token).map_err(|_| Error::TransportParameter),
)?; let path = self.paths.primary().ok_or(Error::NoAvailablePath)?;
path.borrow_mut().set_reset_token(reset_token);
/// Validate the `version_negotiation` transport parameter from the peer. fn validate_versions(&self) -> Res<()> { let tph = self.tps.borrow(); let remote_tps = tph.remote_handshake().ok_or(Error::TransportParameter)?; // `current` and `other` are the value from the peer's transport parameters. // We're checking that these match our expectations. iflet Some((current, other)) = remote_tps.get_versions() {
qtrace!( "[{self}] validate_versions: current={:x} chosen={current:x} other={other:x?}", self.version.wire_version(),
); ifself.role == Role::Server { // 1. A server acts on transport parameters, with validation // of `current` happening in the transport parameter handler. // All we need to do is confirm that the transport parameter // was provided.
Ok(())
} elseifself.version().wire_version() != current {
qinfo!("[{self}] validate_versions: current version mismatch");
Err(Error::VersionNegotiation)
} elseifself
.conn_params
.get_versions()
.initial()
.is_compatible(self.version)
{ // 2. The current version is compatible with what we attempted. // That's a compatible upgrade and that's OK.
Ok(())
} else { // 3. The initial version we attempted isn't compatible. Check that // the one we would have chosen is compatible with this one. letmut all_versions = other.to_owned();
all_versions.push(current); ifself
.conn_params
.get_versions()
.preferred(&all_versions)
.ok_or(Error::VersionNegotiation)?
.is_compatible(self.version)
{
Ok(())
} else {
qinfo!("[{self}] validate_versions: failed");
Err(Error::VersionNegotiation)
}
}
} elseifself.version != Version::Version1 && !self.version.is_draft() {
qinfo!("[{self}] validate_versions: missing extension");
Err(Error::VersionNegotiation)
} else {
Ok(())
}
}
/// Commit to a particular version. fn compatible_upgrade(&mutself, packet_version: Version) -> Res<()> { if !matches!(self.state, State::WaitInitial | State::WaitVersion) { return Ok(());
}
let v = ifself.role == Role::Client {
packet_version
} else { let version = self.tps.borrow().version(); let dcid = self
.original_destination_cid
.as_ref()
.ok_or(Error::ProtocolViolation)?; // No need to randomize the starting packet number; that's already taken care of. self.crypto.states_mut().init_server(version, dcid, false)?;
version
};
let was_authentication_pending =
*self.crypto.tls().state() == HandshakeState::AuthenticationPending; let try_update = data.is_some(); matchself.crypto.handshake(now, space, data)? {
HandshakeState::Authenticated(_) | HandshakeState::InProgress => (),
HandshakeState::AuthenticationPending => { if !was_authentication_pending { self.events.authentication_needed();
}
}
HandshakeState::EchFallbackAuthenticationPending(public_name) => self
.events
.ech_fallback_authentication_needed(public_name.clone()),
HandshakeState::Complete(_) => { if !self.state.connected() { self.set_connected(now)?;
}
}
_ => {
qerror!("Crypto state should not be new or failed after successful handshake"); return Err(Error::Crypto(nss::Error::Internal));
}
}
// There is a chance that this could be called less often, but getting the // conditions right is a little tricky, so call whenever CRYPTO data is used. if try_update { // We have transport parameters, it's go time. ifself.tps.borrow().remote_handshake().is_some() { self.set_initial_limits();
} ifself.crypto.tls().has_secret(Epoch::Handshake) { self.compatible_upgrade(packet_version)?;
} ifself.crypto.install_keys(self.role)? { self.saved_datagrams.make_available(Epoch::Handshake);
}
}
#[expect(clippy::too_many_lines, reason = "Yep, but it's a nice big match.")] fn input_frame(
&mutself,
path: &PathRef,
packet_version: Version,
packet_type: packet::Type,
frame: Frame,
next_pn: packet::Number,
now: Instant,
) -> Res<()> { if !frame.is_allowed(packet_type) {
qinfo!("frame not allowed: {frame:?} {packet_type:?}"); return Err(Error::ProtocolViolation);
} let space = PacketNumberSpace::from(packet_type); if frame.is_stream() { returnself
.streams
.input_frame(&frame, &mutself.stats.borrow_mut().frame_rx);
} match frame {
Frame::Padding(length) => { self.stats.borrow_mut().frame_rx.padding += usize::from(length);
}
Frame::Ping => { // If we get a PING and there are outstanding CRYPTO frames, // prepare to resend them. self.stats.borrow_mut().frame_rx.ping += 1; self.crypto.resend_unacked(space); // Send an ACK immediately if we might not otherwise do so. self.acks.immediate_ack(space, now);
}
Frame::Ack {
largest_acknowledged,
ack_delay,
first_ack_range,
ack_ranges,
ecn_count,
} => { // Ensure that the largest acknowledged packet number was actually sent. // (If we ever start using non-contiguous packet numbers, we need to check all the // packet numbers in the ACKed ranges.) if largest_acknowledged >= next_pn {
qwarn!("Largest ACKed {largest_acknowledged} was never sent"); return Err(Error::AckedUnsentPacket);
}
let ranges =
Frame::decode_ack_frame(largest_acknowledged, first_ack_range, &ack_ranges)?; self.handle_ack(space, ranges, ecn_count.as_ref(), ack_delay, now)?;
}
Frame::Crypto { offset, data } => {
qtrace!( "[{self}] Crypto frame on space={space} offset={offset}: {d}",
d = hex_snip_middle(data),
); self.stats.borrow_mut().frame_rx.crypto += 1; self.crypto
.streams_mut()
.inbound_frame(space, offset, data)?;
ifself.role == Role::Client
&& space == PacketNumberSpace::Initial
&& packet_version != self.version
{ // If the server has switched versions, switch to that version. // This is an assumption, but very often a good one. // This function does nothing if we already have a version. self.compatible_upgrade(packet_version)?;
}
letmut buf = Vec::new(); ifself.crypto.streams().data_ready(space)
&& self.crypto.streams_mut().read_to_end(space, &mut buf)? > 0
{ self.handshake(now, packet_version, space, Some(&buf))?; self.create_resumption_token(now);
} else { // If we get a useless CRYPTO frame send outstanding CRYPTO frames and 0-RTT // data again. self.crypto.resend_unacked(space); if space == PacketNumberSpace::Initial { self.crypto.resend_unacked(PacketNumberSpace::Handshake); self.resend_0rtt(now);
}
}
}
Frame::NewToken { token } => { ifself.role == Role::Server || !self.state.connected() { // > Clients MUST NOT send NEW_TOKEN frames. A server MUST // > treat receipt of a NEW_TOKEN frame as a connection error of // > type PROTOCOL_VIOLATION. // // <https://www.rfc-editor.org/rfc/rfc9000.html#name-new_token-frames> return Err(Error::ProtocolViolation);
} self.stats.borrow_mut().frame_rx.new_token += 1; self.new_token.save_token(token.to_vec()); self.create_resumption_token(now);
}
Frame::NewConnectionId {
sequence_number,
connection_id,
stateless_reset_token,
retire_prior,
} => { self.stats.borrow_mut().frame_rx.new_connection_id += 1; self.cids.add_remote(ConnectionIdEntry::new(
sequence_number,
ConnectionId::from(connection_id),
stateless_reset_token,
))?; self.paths.retire_cids(retire_prior, &mutself.cids); ifself.cids.len() >= ConnectionIdManager::ACTIVE_LIMIT {
qinfo!("[{self}] received too many connection IDs"); return Err(Error::ConnectionIdLimitExceeded);
}
}
Frame::RetireConnectionId { sequence_number } => { self.stats.borrow_mut().frame_rx.retire_connection_id += 1; self.cid_manager.retire(sequence_number);
}
Frame::PathChallenge { data } => { self.stats.borrow_mut().frame_rx.path_challenge += 1; // If we were challenged, try to make the path permanent. // Report an error if we don't have enough connection IDs. self.ensure_permanent(path, now)?;
path.borrow_mut().challenged(data); // A PATH_CHALLENGE indicates the peer sees a different path, // so start PMTUD to discover any MTU changes. ifself.conn_params.pmtud_enabled() {
path.borrow_mut()
.pmtud_mut()
.start(now, &mutself.stats.borrow_mut());
}
}
Frame::PathResponse { data } => { self.stats.borrow_mut().frame_rx.path_response += 1; iflet Some(primary) = self.paths
.path_response(data, now, &mutself.stats.borrow_mut())
{ self.path_migrated(&primary); self.loss_recovery.migrate();
}
}
Frame::ConnectionClose {
error_code,
frame_type,
reason_phrase,
} => { self.stats.borrow_mut().frame_rx.connection_close += 1;
qinfo!( "[{self}] ConnectionClose received. Error code: {error_code:?} frame type {frame_type:x} reason {reason_phrase}"
); let (detail, frame_type) = iflet CloseError::Application(_) = error_code { // Use a transport error here because we want to send // NO_ERROR in this case.
(
Error::PeerApplication(error_code.code()),
FrameType::ConnectionCloseApplication,
)
} else {
(
Error::Peer(error_code.code()),
FrameType::ConnectionCloseTransport,
)
}; let error = CloseReason::Transport(detail); self.state_signaling
.drain(Rc::clone(path), error.clone(), frame_type, ""); self.set_state(
State::Draining {
error,
timeout: self.get_closing_period_time(now),
},
now,
);
}
Frame::HandshakeDone => { self.stats.borrow_mut().frame_rx.handshake_done += 1; ifself.role == Role::Server || !self.state.connected() { return Err(Error::ProtocolViolation);
} self.set_confirmed(now)?; self.discard_keys(PacketNumberSpace::Handshake, now); self.migrate_to_preferred_address(now)?;
}
Frame::AckFrequency {
seqno,
tolerance,
delay,
ignore_order,
} => { self.stats.borrow_mut().frame_rx.ack_frequency += 1; let delay = Duration::from_micros(delay); if delay < GRANULARITY { return Err(Error::ProtocolViolation);
} self.acks
.ack_freq(seqno, tolerance - 1, delay, ignore_order);
}
Frame::Datagram { data, .. } => { self.stats.borrow_mut().frame_rx.datagram += 1; self.quic_datagrams
.handle_datagram(data, &mutself.stats.borrow_mut())?;
}
_ => unreachable!("All other frames are for streams"),
}
Ok(())
}
/// Given a set of `sent::Packet` instances, ensure that the source of the packet /// is told that they are lost. This gives the frame generation code a chance /// to retransmit the frame as needed. fn handle_lost_packets(&mutself, lost_packets: &[sent::Packet]) { for lost in lost_packets { for token in lost.tokens() {
qdebug!("[{self}] Lost: {token:?}"); match token {
recovery::Token::Ack(ack_token) => { // If we lost an ACK frame during the handshake, send another one. if ack_token.space() != PacketNumberSpace::ApplicationData { self.acks.immediate_ack(ack_token.space(), lost.time_sent());
}
}
recovery::Token::Crypto(ct) => self.crypto.lost(ct),
recovery::Token::HandshakeDone => self.state_signaling.handshake_done(),
recovery::Token::NewToken(seqno) => self.new_token.lost(*seqno),
recovery::Token::NewConnectionId(ncid) => self.cid_manager.lost(ncid),
recovery::Token::RetireConnectionId(seqno) => { self.paths.lost_retire_cid(*seqno);
}
recovery::Token::AckFrequency(rate) => self.paths.lost_ack_frequency(rate),
recovery::Token::KeepAlive => self.idle_timeout.lost_keep_alive(),
recovery::Token::Stream(stream_token) => self.streams.lost(stream_token),
recovery::Token::Datagram(dgram_tracker) => { self.events
.datagram_outcome(dgram_tracker, OutgoingDatagramOutcome::Lost); self.stats.borrow_mut().datagram_tx.lost += 1;
}
recovery::Token::EcnEct0 => self.paths.lost_ecn(&mutself.stats.borrow_mut()), // PMTUD probe loss is handled by the PMTUD state machine.
recovery::Token::PmtudProbe => (),
}
}
}
}
fn decode_ack_delay(&self, v: u64) -> Res<Duration> { // If we have remote transport parameters, use them. // Otherwise, ack delay should be zero (because it's the handshake). self.tps.borrow().remote_handshake().map_or_else(
|| Ok(Duration::default()),
|r| { let exponent = u32::try_from(r.get_integer(AckDelayExponent))?; // ACK_DELAY_EXPONENT > 20 is invalid per RFC9000. We already checked that in // TransportParameter::decode. let corrected = if v.leading_zeros() >= exponent {
v << exponent
} else {
u64::MAX
};
Ok(Duration::from_micros(corrected))
},
)
}
let Some(path) = self.paths.primary() else { return Ok(());
}; let (acked_packets, lost_packets) = self.loss_recovery.on_ack_received(
&path,
space,
ack_ranges,
ack_ecn, self.decode_ack_delay(ack_delay)?,
now,
); let largest_acknowledged = acked_packets.first().map(sent::Packet::pn);
qlog::packets_acked(&mutself.qlog, space, &acked_packets, now); for acked in acked_packets { for token in acked.tokens() { match token {
recovery::Token::Stream(stream_token) => self.streams.acked(stream_token),
recovery::Token::Ack(at) => self.acks.acked(at),
recovery::Token::Crypto(ct) => self.crypto.acked(ct),
recovery::Token::NewToken(seqno) => self.new_token.acked(*seqno),
recovery::Token::NewConnectionId(entry) => self.cid_manager.acked(entry),
recovery::Token::RetireConnectionId(seqno) => { self.paths.acked_retire_cid(*seqno);
}
recovery::Token::AckFrequency(rate) => self.paths.acked_ack_frequency(rate),
recovery::Token::KeepAlive => self.idle_timeout.ack_keep_alive(),
recovery::Token::Datagram(dgram_tracker) => self
.events
.datagram_outcome(dgram_tracker, OutgoingDatagramOutcome::Acked),
recovery::Token::EcnEct0 => self.paths.acked_ecn(), // We don't care about these being ACK'ed
recovery::Token::HandshakeDone | recovery::Token::PmtudProbe => (),
}
}
} self.handle_lost_packets(&lost_packets);
qlog::packets_lost(&mutself.qlog, &lost_packets, now); let stats = &mutself.stats.borrow_mut().frame_rx;
stats.ack += 1; iflet Some(largest_acknowledged) = largest_acknowledged {
stats.largest_acknowledged = max(stats.largest_acknowledged, largest_acknowledged);
}
Ok(())
}
/// Tell 0-RTT packets that they were "lost". fn resend_0rtt(&mutself, now: Instant) { iflet Some(path) = self.paths.primary() { let dropped = self.loss_recovery.drop_0rtt(&path, now); self.handle_lost_packets(&dropped);
}
}
/// When the server rejects 0-RTT we need to drop a bunch of stuff. fn client_0rtt_rejected(&mutself, now: Instant) { if !matches!(self.zero_rtt_state, ZeroRttState::Sending) { return;
}
qdebug!("[{self}] 0-RTT rejected"); self.resend_0rtt(now); self.streams.zero_rtt_rejected(); self.crypto.states_mut().discard_0rtt_keys(); self.events.client_0rtt_rejected();
}
fn set_connected(&mutself, now: Instant) -> Res<()> {
qdebug!("[{self}] TLS connection complete"); ifself
.crypto
.tls()
.info()
.map(SecretAgentInfo::alpn)
.is_none()
{
qwarn!("[{self}] No ALPN, closing connection"); // 120 = no_application_protocol return Err(Error::CryptoAlert(120));
} ifself.role == Role::Server { // Remove the randomized client CID from the list of acceptable CIDs. self.cid_manager.remove_odcid(); // Mark the path as validated, if it isn't already. let path = self.paths.primary().ok_or(Error::NoAvailablePath)?;
path.borrow_mut().set_valid(now); // Generate a qlog event that the server connection started.
qlog::server_connection_started(&mutself.qlog, &path, now);
qlog::recovery_parameters_set(
&mutself.qlog,
path.borrow().plpmtu(), self.conn_params.get_congestion_control(),
now,
);
qlog::congestion_state_updated(
&mutself.qlog,
None,
Phase::SlowStart.into(),
None,
now,
);
} else { self.zero_rtt_state = ifself
.crypto
.tls()
.info()
.ok_or(Error::Internal)?
.early_data_accepted()
{
ZeroRttState::AcceptedClient
} else {
self.client_0rtt_rejected(now);
ZeroRttState::Rejected
};
}
// Setting application keys has to occur after 0-RTT rejection.
let pto = self.pto();
self.crypto
.install_application_keys(self.version, now + pto)?;
self.process_tps(now)?;
self.set_state(State::Connected, now);
self.create_resumption_token(now);
self.saved_datagrams.make_available(Epoch::ApplicationData);
self.stats.borrow_mut().resumed =
self.crypto.tls().info().ok_or(Error::Internal)?.resumed(); if self.role == Role::Server {
self.state_signaling.handshake_done();
self.set_confirmed(now)?;
}
qinfo!("[{self}] Connection established");
Ok(())
}
fn set_state(&mut self, state: State, now: Instant) { if state > self.state {
qdebug!("[{self}] State change from {:?} -> {state:?}", self.state);
let old_state = self.state.clone();
self.state = state.clone(); if self.state.closed() {
self.streams.clear_streams();
}
self.events.connection_state_change(state);
qlog::connection_state_updated(&mut self.qlog, &old_state, &self.state, now); if let State::Closed(reason) = &self.state {
qlog::connection_closed(&mut self.qlog, reason, now);
}
} elseif mem::discriminant(&state) != mem::discriminant(&self.state) { // Only tolerate a regression in state if the new state is closing // and the connection is already closed.
debug_assert!(matches!(
state,
State::Closing { .. } | State::Draining { .. }
));
debug_assert!(self.state.closed());
}
}
/// Create a stream. /// Returns new stream id /// /// # Errors /// /// `ConnectionState` if the connection stat does not allow to create streams. /// `StreamLimitError` if we are limited by server's stream concurrence.
pub fn stream_create(&mut self, st: StreamType) -> Res<StreamId> { // Can't make streams while closing, otherwise rely on the stream limits.
match self.state {
State::Closing { .. } | State::Draining { .. } | State::Closed { .. } => { return Err(Error::ConnectionState);
}
State::WaitInitial | State::Handshaking if self.role == Role::Client && self.zero_rtt_state != ZeroRttState::Sending =>
{ return Err(Error::ConnectionState);
} // In all other states, trust that the stream limits are correct.
_ => (),
}
self.streams.stream_create(st)
}
/// Set the priority of a stream. /// /// # Errors /// /// `InvalidStreamId` the stream does not exist.
pub fn stream_priority(
&mut self,
stream_id: StreamId,
transmission: TransmissionPriority,
retransmission: RetransmissionPriority,
) -> Res<()> {
self.streams
.get_send_stream_mut(stream_id)?
.set_priority(transmission, retransmission);
Ok(())
}
/// Set the `SendOrder` of a stream. Re-enqueues to keep the ordering correct /// /// # Errors /// When the stream does not exist.
pub fn stream_sendorder(
&mut self,
stream_id: StreamId,
sendorder: Option<SendOrder>,
) -> Res<()> {
self.streams.set_sendorder(stream_id, sendorder)
}
/// Set the Fairness of a stream /// /// # Errors /// When the stream does not exist.
pub fn stream_fairness(&mut self, stream_id: StreamId, fairness: bool) -> Res<()> {
self.streams.set_fairness(stream_id, fairness)
}
/// # Errors /// When the stream does not exist.
pub fn send_stream_stats(&self, stream_id: StreamId) -> Res<send_stream::Stats> {
self.streams
.get_send_stream(stream_id)
.map(SendStream::stats)
}
/// # Errors /// When the stream does not exist.
pub fn recv_stream_stats(&mut self, stream_id: StreamId) -> Res<recv_stream::Stats> {
let stream = self.streams.get_recv_stream_mut(stream_id)?;
Ok(stream.stats())
}
/// Send data on a stream. /// Returns how many bytes were successfully sent. Could be less /// than total, based on receiver credit space available, etc. /// /// # Errors /// /// `InvalidStreamId` the stream does not exist, /// `InvalidInput` if length of `data` is zero, /// `FinalSizeError` if the stream has already been closed.
pub fn stream_send(&mut self, stream_id: StreamId, data: &[u8]) -> Res<usize> {
self.streams.get_send_stream_mut(stream_id)?.send(data)
}
/// Send all data or nothing on a stream. May cause `DATA_BLOCKED` or /// `STREAM_DATA_BLOCKED` frames to be sent. /// Returns true if data was successfully sent, otherwise false. /// /// # Errors /// /// `InvalidStreamId` the stream does not exist, /// `InvalidInput` if length of `data` is zero, /// `FinalSizeError` if the stream has already been closed.
pub fn stream_send_atomic(&mut self, stream_id: StreamId, data: &[u8]) -> Res<bool> {
let val = self
.streams
.get_send_stream_mut(stream_id)?
.send_atomic(data); if let Ok(val) = val {
debug_assert!(
val == 0 || val == data.len(), "Unexpected value {val} when trying to send {} bytes atomically",
data.len()
);
}
val.map(|v| v == data.len())
}
/// Bytes that `stream_send()` is guaranteed to accept for sending. /// i.e. that will not be blocked by flow credits or send buffer max /// capacity. /// # Errors /// When the stream ID is invalid.
pub fn stream_avail_send_space(&self, stream_id: StreamId) -> Res<usize> {
Ok(self.streams.get_send_stream(stream_id)?.avail())
}
/// Set low watermark for [`ConnectionEvent::SendStreamWritable`] event. /// /// Stream emits a [`crate::ConnectionEvent::SendStreamWritable`] event /// when: /// - the available sendable bytes increased to or above the watermark /// - and was previously below the watermark. /// /// Default value is `1`. In other words /// [`crate::ConnectionEvent::SendStreamWritable`] is emitted whenever the /// available sendable bytes was previously at `0` and now increased to `1` /// or more. /// /// Use this when your protocol needs at least `watermark` amount of available /// sendable bytes to make progress. /// /// # Errors /// When the stream ID is invalid.
pub fn stream_set_writable_event_low_watermark(
&mut self,
stream_id: StreamId,
watermark: NonZeroUsize,
) -> Res<()> {
self.streams
.get_send_stream_mut(stream_id)?
.set_writable_event_low_watermark(watermark);
Ok(())
}
/// Close the stream. Enqueued data will be sent. /// # Errors /// When the stream ID is invalid.
pub fn stream_close_send(&mut self, stream_id: StreamId) -> Res<()> {
self.streams.get_send_stream_mut(stream_id)?.close();
Ok(())
}
/// Abandon transmission of in-flight and future stream data. /// # Errors /// When the stream ID is invalid.
pub fn stream_reset_send(&mut self, stream_id: StreamId, err: AppError) -> Res<()> {
self.streams.get_send_stream_mut(stream_id)?.reset(err);
Ok(())
}
/// Read buffered data from stream. bool says whether read bytes includes /// the final data on stream. /// /// # Errors /// /// `InvalidStreamId` if the stream does not exist. /// `NoMoreData` if data and fin bit were previously read by the application.
pub fn stream_recv(&mut self, stream_id: StreamId, data: &mut [u8]) -> Res<(usize, bool)> {
self.streams.recv(stream_id, data)
}
/// Application is no longer interested in this stream. /// # Errors /// When the stream ID is invalid.
pub fn stream_stop_sending(&mut self, stream_id: StreamId, err: AppError) -> Res<()> {
self.streams.stop_sending(stream_id, err)
}
/// Increases `max_stream_data` for a `stream_id`. /// /// # Errors /// /// Returns `InvalidStreamId` if a stream does not exist or the receiving /// side is closed.
pub fn set_stream_max_data(&mut self, stream_id: StreamId, max_data: u64) -> Res<()> {
let stream = self.streams.get_recv_stream_mut(stream_id)?;
stream.set_stream_max_data(max_data);
Ok(())
}
/// Mark a receive stream as being important enough to keep the connection alive /// (if `keep` is `true`) or no longer important (if `keep` is `false`). If any /// stream is marked this way, PING frames will be used to keep the connection /// alive, even when there is no activity. /// /// # Errors /// /// Returns `InvalidStreamId` if a stream does not exist or the receiving /// side is closed.
pub fn stream_keep_alive(&mut self, stream_id: StreamId, keep: bool) -> Res<()> {
self.streams.keep_alive(stream_id, keep)
}
/// Returns the current max size of a datagram that can fit into a packet. /// The value will change over time depending on the encoded size of the /// packet number, ack frames, etc. /// /// # Errors /// The function returns `NotAvailable` if datagrams are not enabled. /// # Panics /// Basically never, because that unwrap won't fail.
pub fn max_datagram_size(&self) -> Res<u64> {
let max_dgram_size = self.quic_datagrams.remote_datagram_size(); if max_dgram_size == 0 { return Err(Error::NotAvailable);
}
let version = self.version();
let Some((epoch, tx)) = self
.crypto
.states()
.select_tx(self.version, PacketNumberSpace::ApplicationData) else { return Err(Error::NotAvailable);
};
let path = self.paths.primary().ok_or(Error::NotAvailable)?;
let mtu = path.borrow().plpmtu();
let mut buffer = Vec::new();
let encoder = Encoder::new_borrowed_vec(&mut buffer);
/// Queue a datagram for sending. /// /// # Errors /// /// The function returns `TooMuchData` if the supply buffer is bigger than /// the allowed remote datagram size. The function does not check if the /// datagram can fit into a packet (i.e. MTU limit). This is checked during /// creation of an actual packet and the datagram will be dropped if it does /// not fit into the packet. The app is encourage to use `max_datagram_size` /// to check the estimated max datagram size and to use smaller datagrams. /// `max_datagram_size` is just a current estimate and will change over /// time depending on the encoded size of the packet number, ack frames, etc.
pub fn send_datagram<I: Into<DatagramTracking>>(&mut self, buf: Vec<u8>, id: I) -> Res<()> {
self.quic_datagrams
.add_datagram(buf, id.into(), &mut self.stats.borrow_mut())
}
/// Return the PLMTU of the primary path. /// /// # Panics /// /// The function panics if there is no primary path. (Should be fine for test usage.) #[cfg(test)] #[must_use]
pub fn plpmtu(&self) -> usize {
self.paths.primary().unwrap().borrow().plpmtu()
}
fn log_packet(&mut self, meta: packet::MetaData, now: Instant) { if log::log_enabled!(log::Level::Debug) {
let mut s = String::new();
let mut d = Decoder::from(meta.payload()); while d.remaining() > 0 {
let Ok(f) = Frame::decode(&mut d) else {
s.push_str(" [broken]..."); break;
};
let x = f.dump(); if !x.is_empty() {
_ = write!(&mut s, "\n {} {x}", meta.direction());
}
}
qdebug!("[{self}] {meta}{s}");
}
qlog::packet_io(&mut self.qlog, meta, now);
}
}
impl EventProvider for Connection {
type Event = ConnectionEvent;
/// Return true if there are outstanding events.
fn has_events(&self) -> bool {
self.events.has_events()
}
/// Get events that indicate state changes on the connection. This method /// correctly handles cases where handling one event can obsolete /// previously-queued events, or cause new events to be generated.
fn next_event(&mut self) -> Option<Self::Event> {
self.events.next_event()
}
}
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.
