Allow server to use NEW_TOKEN frames

Whenever a path becomes validated, the server sends the client NEW_TOKEN
frames. These may cause an Incoming to be validated.

- Adds dependency on `fastbloom`
- Converts TokenInner to enum with Retry and Validation variants
- Adds relevant configuration to ServerConfig
- Incoming now has `may_retry`
- Adds `TokenLog` object to server to mitigate token reuse

Cargo.toml

@@ -25,6 +25,7 @@ crc = "3"
 directories-next = "2"
 futures-io = "0.3.19"
 getrandom = { version = "0.2", default-features = false }
+fastbloom = "0.7"
 hdrhistogram = { version = "7.2", default-features = false }
 hex-literal = "0.4"
 lazy_static = "1"

quinn-proto/Cargo.toml

@@ -34,6 +34,7 @@ rustls-log = ["rustls?/logging"]
 arbitrary = { workspace = true, optional = true }
 aws-lc-rs = { workspace = true, optional = true }
 bytes = { workspace = true }
+fastbloom = { workspace = true }
 rustc-hash = { workspace = true }
 rand = { workspace = true }
 ring = { workspace = true, optional = true }

quinn-proto/src/config.rs

@@ -18,7 +18,7 @@ use crate::{
     congestion,
     crypto::{self, HandshakeTokenKey, HmacKey},
     shared::ConnectionId,
-    Duration, RandomConnectionIdGenerator, VarInt, VarIntBoundsExceeded,
+    BloomTokenLog, Duration, RandomConnectionIdGenerator, TokenLog, VarInt, VarIntBoundsExceeded,
     DEFAULT_SUPPORTED_VERSIONS, INITIAL_MTU, MAX_CID_SIZE, MAX_UDP_PAYLOAD,
 };
 
@@ -810,6 +810,16 @@ pub struct ServerConfig {
     /// Duration after a retry token was issued for which it's considered valid
     pub(crate) retry_token_lifetime: Duration,
 
+    /// Duration after an address validation token was issued for which it's considered valid
+    pub(crate) validation_token_lifetime: Duration,
+
+    /// Responsible for limiting clients' ability to reuse tokens from NEW_TOKEN frames
+    pub(crate) validation_token_log: Option<Arc<dyn TokenLog>>,
+
+    /// Number of address validation tokens sent to a client via NEW_TOKEN frames when its path is
+    /// validated
+    pub(crate) validation_tokens_sent: u32,
+
     /// Whether to allow clients to migrate to new addresses
     ///
     /// Improves behavior for clients that move between different internet connections or suffer NAT
@@ -825,19 +835,28 @@ pub struct ServerConfig {
 }
 
 const DEFAULT_RETRY_TOKEN_LIFETIME_SECS: u64 = 15;
+const DEFAULT_VALIDATION_TOKEN_LIFETIME_SECS: u64 = 2 * 7 * 24 * 60 * 60;
 
 impl ServerConfig {
     /// Create a default config with a particular handshake token key
+    ///
+    /// Setting `validation_token_log` to `None` makes the server ignore all address validation
+    /// tokens originating from NEW_TOKEN frames, although stateless retry tokens may still be
+    /// accepted.
     pub fn new(
         crypto: Arc<dyn crypto::ServerConfig>,
         token_key: Arc<dyn HandshakeTokenKey>,
+        validation_token_log: Option<Arc<dyn TokenLog>>,
     ) -> Self {
         Self {
             transport: Arc::new(TransportConfig::default()),
             crypto,
 
             token_key,
             retry_token_lifetime: Duration::from_secs(DEFAULT_RETRY_TOKEN_LIFETIME_SECS),
+            validation_token_lifetime: Duration::from_secs(DEFAULT_VALIDATION_TOKEN_LIFETIME_SECS),
+            validation_token_log,
+            validation_tokens_sent: 2,
 
             migration: true,
 
@@ -870,6 +889,27 @@ impl ServerConfig {
         self
     }
 
+    /// Duration after an address validation token was issued for which it's considered valid
+    ///
+    /// This refers only to tokens sent in NEW_TOKEN frames, in contrast to stateless retry tokens.
+    ///
+    /// Defaults to 2 weeks.
+    pub fn validation_token_lifetime(&mut self, value: Duration) -> &mut Self {
+        self.validation_token_lifetime = value;
+        self
+    }
+
+    /// Number of address validation tokens sent to a client via NEW_TOKEN frames when its path is
+    /// validated
+    ///
+    /// This refers only to tokens sent in NEW_TOKEN frames, in contrast to stateless retry tokens.
+    ///
+    /// Defaults to 2.
+    pub fn validation_tokens_sent(&mut self, value: u32) -> &mut Self {
+        self.validation_tokens_sent = value;
+        self
+    }
+
     /// Whether to allow clients to migrate to new addresses
     ///
     /// Improves behavior for clients that move between different internet connections or suffer NAT
@@ -963,7 +1003,7 @@ impl ServerConfig {
 impl ServerConfig {
     /// Create a server config with the given [`crypto::ServerConfig`]
     ///
-    /// Uses a randomized handshake token key.
+    /// Uses a randomized handshake token key and a default `BloomTokenLog`.
     pub fn with_crypto(crypto: Arc<dyn crypto::ServerConfig>) -> Self {
         #[cfg(all(feature = "aws-lc-rs", not(feature = "ring")))]
         use aws_lc_rs::hkdf;
@@ -976,7 +1016,11 @@ impl ServerConfig {
         rng.fill_bytes(&mut master_key);
         let master_key = hkdf::Salt::new(hkdf::HKDF_SHA256, &[]).extract(&master_key);
 
-        Self::new(crypto, Arc::new(master_key))
+        Self::new(
+            crypto,
+            Arc::new(master_key),
+            Some(Arc::new(BloomTokenLog::default())),
+        )
     }
 }
 
@@ -987,6 +1031,8 @@ impl fmt::Debug for ServerConfig {
             // crypto not debug
             // token not debug
             .field("retry_token_lifetime", &self.retry_token_lifetime)
+            .field("validation_token_lifetime", &self.validation_token_lifetime)
+            .field("validation_tokens_sent", &self.validation_tokens_sent)
             .field("migration", &self.migration)
             .field("preferred_address_v4", &self.preferred_address_v4)
             .field("preferred_address_v6", &self.preferred_address_v6)

quinn-proto/src/connection/mod.rs

@@ -5,6 +5,7 @@ use std::{
     fmt, io, mem,
     net::{IpAddr, SocketAddr},
     sync::Arc,
+    time::SystemTime,
 };
 
 use bytes::{Bytes, BytesMut};
@@ -19,7 +20,7 @@ use crate::{
     coding::BufMutExt,
     config::{ServerConfig, TransportConfig},
     crypto::{self, KeyPair, Keys, PacketKey},
-    frame::{self, Close, Datagram, FrameStruct},
+    frame::{self, Close, Datagram, FrameStruct, NewToken},
     packet::{
         FixedLengthConnectionIdParser, Header, InitialHeader, InitialPacket, LongType, Packet,
         PacketNumber, PartialDecode, SpaceId,
@@ -29,7 +30,7 @@ use crate::{
         ConnectionEvent, ConnectionEventInner, ConnectionId, DatagramConnectionEvent, EcnCodepoint,
         EndpointEvent, EndpointEventInner,
     },
-    token::ResetToken,
+    token::{ResetToken, Token, TokenInner},
     transport_parameters::TransportParameters,
     Dir, Duration, EndpointConfig, Frame, Instant, Side, StreamId, Transmit, TransportError,
     TransportErrorCode, VarInt, INITIAL_MTU, MAX_CID_SIZE, MAX_STREAM_COUNT, MIN_INITIAL_SIZE,
@@ -360,6 +361,9 @@ impl Connection {
             stats: ConnectionStats::default(),
             version,
         };
+        if path_validated {
+            this.on_path_validated();
+        }
         if side.is_client() {
             // Kick off the connection
             this.write_crypto();
@@ -2444,7 +2448,7 @@ impl Connection {
                     );
                     return Ok(());
                 }
-                self.path.validated = true;
+                self.on_path_validated();
 
                 self.process_early_payload(now, packet)?;
                 if self.state.is_closed() {
@@ -2858,7 +2862,7 @@ impl Connection {
                         self.update_rem_cid();
                     }
                 }
-                Frame::NewToken { token } => {
+                Frame::NewToken(NewToken { token }) => {
                     if self.side.is_server() {
                         return Err(TransportError::PROTOCOL_VIOLATION("client sent NEW_TOKEN"));
                     }
@@ -3251,6 +3255,27 @@ impl Connection {
             self.datagrams.send_blocked = false;
         }
 
+        // NEW_TOKEN
+        while let Some((remote_addr, new_token)) = space.pending.new_tokens.pop() {
+            debug_assert_eq!(space_id, SpaceId::Data);
+
+            if remote_addr != self.path.remote {
+                continue;
+            }
+
+            if buf.len() + new_token.size() >= max_size {
+                space.pending.new_tokens.push((remote_addr, new_token));
+                break;
+            }
+
+            new_token.encode(buf);
+            sent.retransmits
+                .get_or_create()
+                .new_tokens
+                .push((remote_addr, new_token));
+            self.stats.frame_tx.new_token += 1;
+        }
+
         // STREAM
         if space_id == SpaceId::Data {
             sent.stream_frames =
@@ -3612,6 +3637,33 @@ impl Connection {
         // but that would needlessly prevent sending datagrams during 0-RTT.
         key.map_or(16, |x| x.tag_len())
     }
+
+    /// Mark the path as validated, and enqueue NEW_TOKEN frames to be sent as appropriate
+    fn on_path_validated(&mut self) {
+        self.path.validated = true;
+        if let Some(server_config) = self.server_config.as_ref() {
+            self.spaces[SpaceId::Data as usize]
+                .pending
+                .new_tokens
+                .clear();
+            for _ in 0..server_config.validation_tokens_sent {
+                let token_inner = TokenInner::Validation {
+                    issued: SystemTime::now(),
+                };
+                let token = Token::new(&mut self.rng, token_inner)
+                    .encode(&*server_config.token_key, &self.path.remote);
+                self.spaces[SpaceId::Data as usize]
+                    .pending
+                    .new_tokens
+                    .push((
+                        self.path.remote,
+                        NewToken {
+                            token: token.into(),
+                        },
+                    ));
+            }
+        }
+    }
 }
 
 impl fmt::Debug for Connection {

quinn-proto/src/connection/spaces.rs

@@ -2,6 +2,7 @@ use std::{
     cmp,
     collections::{BTreeMap, VecDeque},
     mem,
+    net::SocketAddr,
     ops::{Bound, Index, IndexMut},
 };
 
@@ -309,6 +310,8 @@ pub struct Retransmits {
     pub(super) retire_cids: Vec<u64>,
     pub(super) ack_frequency: bool,
     pub(super) handshake_done: bool,
+    /// NEW_TOKEN frames excluded from retransmission if path has changed
+    pub(super) new_tokens: Vec<(SocketAddr, frame::NewToken)>,
 }
 
 impl Retransmits {
@@ -326,6 +329,7 @@ impl Retransmits {
             && self.retire_cids.is_empty()
             && !self.ack_frequency
             && !self.handshake_done
+            && self.new_tokens.is_empty()
     }
 }
 
@@ -347,6 +351,7 @@ impl ::std::ops::BitOrAssign for Retransmits {
         self.retire_cids.extend(rhs.retire_cids);
         self.ack_frequency |= rhs.ack_frequency;
         self.handshake_done |= rhs.handshake_done;
+        self.new_tokens.extend_from_slice(&rhs.new_tokens);
     }
 }
 

quinn-proto/src/endpoint.rs

@@ -494,27 +494,44 @@ impl Endpoint {
 
         let server_config = self.server_config.as_ref().unwrap().clone();
 
-        let (retry_src_cid, orig_dst_cid) = if header.token.is_empty() {
-            (None, header.dst_cid)
+        let (retry_src_cid, orig_dst_cid, validated) = if header.token.is_empty() {
+            (None, header.dst_cid, false)
         } else {
             let valid_token =
                 Token::decode(&*server_config.token_key, &addresses.remote, &header.token)
-                    .and_then(|token| {
-                        let TokenInner {
-                            orig_dst_cid,
-                            issued,
-                        } = token.inner;
-                        if issued + server_config.retry_token_lifetime > SystemTime::now() {
-                            Ok((Some(header.dst_cid), orig_dst_cid))
-                        } else {
-                            Err(TokenDecodeError::InvalidRetry)
+                    .and_then(|token| match token.inner {
+                        TokenInner::Retry { orig_dst_cid, issued } => {
+                            if issued + server_config.retry_token_lifetime > SystemTime::now() {
+                                Ok((Some(header.dst_cid), orig_dst_cid))
+                            } else {
+                                Err(TokenDecodeError::InvalidRetry)
+                            }
+                        }
+                        TokenInner::Validation { issued } => {
+                            if server_config
+                                .validation_token_log
+                                .as_ref()
+                                .map(|log| {
+                                    let reuse_ok = log.check_and_insert(token.rand, issued, server_config.validation_token_lifetime).is_ok();
+                                    if !reuse_ok {
+                                        debug!("rejecting token from NEW_TOKEN frame because detected as reuse");
+                                    }
+                                    issued + server_config.validation_token_lifetime > SystemTime::now() && reuse_ok
+                                })
+                                .unwrap_or(false)
+                            {
+                                trace!("accepting token from NEW_TOKEN frame");
+                                Ok((None, header.dst_cid))
+                            } else {
+                                Err(TokenDecodeError::UnknownToken)
+                            }
                         }
                     });
             match valid_token {
-                Ok((retry_src_cid, orig_dst_cid)) => (retry_src_cid, orig_dst_cid),
+                Ok((retry_src_cid, orig_dst_cid)) => (retry_src_cid, orig_dst_cid, true),
                 Err(TokenDecodeError::UnknownToken) => {
                     trace!("ignoring unknown token");
-                    (None, header.dst_cid)
+                    (None, header.dst_cid, false)
                 }
                 Err(TokenDecodeError::InvalidRetry) => {
                     debug!("rejecting invalid retry token");
@@ -547,6 +564,7 @@ impl Endpoint {
             retry_src_cid,
             orig_dst_cid,
             incoming_idx,
+            validated,
             improper_drop_warner: IncomingImproperDropWarner,
         }))
     }
@@ -737,7 +755,7 @@ impl Endpoint {
 
     /// Respond with a retry packet, requiring the client to retry with address validation
     ///
-    /// Errors if `incoming.remote_address_validated()` is true.
+    /// Errors if `incoming.may_retry()` is false.
     pub fn retry(&mut self, incoming: Incoming, buf: &mut Vec<u8>) -> Result<Transmit, RetryError> {
         if incoming.remote_address_validated() {
             return Err(RetryError(incoming));
@@ -756,7 +774,7 @@ impl Endpoint {
         // retried by the application layer.
         let loc_cid = self.local_cid_generator.generate_cid();
 
-        let token_inner = TokenInner {
+        let token_inner = TokenInner::Retry {
             orig_dst_cid: incoming.packet.header.dst_cid,
             issued: SystemTime::now(),
         };
@@ -1185,6 +1203,7 @@ pub struct Incoming {
     retry_src_cid: Option<ConnectionId>,
     orig_dst_cid: ConnectionId,
     incoming_idx: usize,
+    validated: bool,
     improper_drop_warner: IncomingImproperDropWarner,
 }
 
@@ -1205,8 +1224,19 @@ impl Incoming {
     ///
     /// This means that the sender of the initial packet has proved that they can receive traffic
     /// sent to `self.remote_address()`.
+    ///
+    /// If `self.remote_address_validated()` is false, `self.may_retry()` is guaranteed to be true.
+    /// The inverse is not guaranteed.
     pub fn remote_address_validated(&self) -> bool {
-        self.retry_src_cid.is_some()
+        self.validated
+    }
+
+    /// Whether it is legal to respond with a retry packet
+    ///
+    /// If `self.remote_address_validated()` is false, `self.may_retry()` is guaranteed to be true.
+    /// The inverse is not guaranteed.
+    pub fn may_retry(&self) -> bool {
+        self.retry_src_cid.is_none()
     }
 
     /// The original destination connection ID sent by the client
@@ -1225,6 +1255,7 @@ impl fmt::Debug for Incoming {
             .field("retry_src_cid", &self.retry_src_cid)
             .field("orig_dst_cid", &self.orig_dst_cid)
             .field("incoming_idx", &self.incoming_idx)
+            .field("validated", &self.validated)
             // improper drop warner contains no information
             .finish_non_exhaustive()
     }