quinn_proto/
endpoint.rs

1use std::{
2    collections::{HashMap, hash_map},
3    convert::TryFrom,
4    fmt, mem,
5    net::{IpAddr, SocketAddr},
6    ops::{Index, IndexMut},
7    sync::Arc,
8};
9
10use bytes::{BufMut, Bytes, BytesMut};
11use rand::{Rng, RngCore, SeedableRng, rngs::StdRng};
12use rustc_hash::FxHashMap;
13use slab::Slab;
14use thiserror::Error;
15use tracing::{debug, error, trace, warn};
16
17use crate::{
18    Duration, INITIAL_MTU, Instant, MAX_CID_SIZE, MIN_INITIAL_SIZE, RESET_TOKEN_SIZE, ResetToken,
19    Side, Transmit, TransportConfig, TransportError,
20    cid_generator::ConnectionIdGenerator,
21    coding::BufMutExt,
22    config::{ClientConfig, EndpointConfig, ServerConfig},
23    connection::{Connection, ConnectionError, SideArgs},
24    crypto::{self, Keys, UnsupportedVersion},
25    frame,
26    packet::{
27        FixedLengthConnectionIdParser, Header, InitialHeader, InitialPacket, PacketDecodeError,
28        PacketNumber, PartialDecode, ProtectedInitialHeader,
29    },
30    shared::{
31        ConnectionEvent, ConnectionEventInner, ConnectionId, DatagramConnectionEvent, EcnCodepoint,
32        EndpointEvent, EndpointEventInner, IssuedCid,
33    },
34    token::{IncomingToken, InvalidRetryTokenError, Token, TokenPayload},
35    transport_parameters::{PreferredAddress, TransportParameters},
36};
37
38/// The main entry point to the library
39///
40/// This object performs no I/O whatsoever. Instead, it consumes incoming packets and
41/// connection-generated events via `handle` and `handle_event`.
42pub struct Endpoint {
43    rng: StdRng,
44    index: ConnectionIndex,
45    connections: Slab<ConnectionMeta>,
46    local_cid_generator: Box<dyn ConnectionIdGenerator>,
47    config: Arc<EndpointConfig>,
48    server_config: Option<Arc<ServerConfig>>,
49    /// Whether the underlying UDP socket promises not to fragment packets
50    allow_mtud: bool,
51    /// Time at which a stateless reset was most recently sent
52    last_stateless_reset: Option<Instant>,
53    /// Buffered Initial and 0-RTT messages for pending incoming connections
54    incoming_buffers: Slab<IncomingBuffer>,
55    all_incoming_buffers_total_bytes: u64,
56}
57
58impl Endpoint {
59    /// Create a new endpoint
60    ///
61    /// `allow_mtud` enables path MTU detection when requested by `Connection` configuration for
62    /// better performance. This requires that outgoing packets are never fragmented, which can be
63    /// achieved via e.g. the `IPV6_DONTFRAG` socket option.
64    ///
65    /// If `rng_seed` is provided, it will be used to initialize the endpoint's rng (having priority
66    /// over the rng seed configured in [`EndpointConfig`]). Note that the `rng_seed` parameter will
67    /// be removed in a future release, so prefer setting it to `None` and configuring rng seeds
68    /// using [`EndpointConfig::rng_seed`].
69    pub fn new(
70        config: Arc<EndpointConfig>,
71        server_config: Option<Arc<ServerConfig>>,
72        allow_mtud: bool,
73        rng_seed: Option<[u8; 32]>,
74    ) -> Self {
75        let rng_seed = rng_seed.or(config.rng_seed);
76        Self {
77            rng: rng_seed.map_or(StdRng::from_os_rng(), StdRng::from_seed),
78            index: ConnectionIndex::default(),
79            connections: Slab::new(),
80            local_cid_generator: (config.connection_id_generator_factory.as_ref())(),
81            config,
82            server_config,
83            allow_mtud,
84            last_stateless_reset: None,
85            incoming_buffers: Slab::new(),
86            all_incoming_buffers_total_bytes: 0,
87        }
88    }
89
90    /// Replace the server configuration, affecting new incoming connections only
91    pub fn set_server_config(&mut self, server_config: Option<Arc<ServerConfig>>) {
92        self.server_config = server_config;
93    }
94
95    /// Process `EndpointEvent`s emitted from related `Connection`s
96    ///
97    /// In turn, processing this event may return a `ConnectionEvent` for the same `Connection`.
98    pub fn handle_event(
99        &mut self,
100        ch: ConnectionHandle,
101        event: EndpointEvent,
102    ) -> Option<ConnectionEvent> {
103        use EndpointEventInner::*;
104        match event.0 {
105            NeedIdentifiers(now, n) => {
106                return Some(self.send_new_identifiers(now, ch, n));
107            }
108            ResetToken(remote, token) => {
109                if let Some(old) = self.connections[ch].reset_token.replace((remote, token)) {
110                    self.index.connection_reset_tokens.remove(old.0, old.1);
111                }
112                if self.index.connection_reset_tokens.insert(remote, token, ch) {
113                    warn!("duplicate reset token");
114                }
115            }
116            RetireConnectionId(now, seq, allow_more_cids) => {
117                if let Some(cid) = self.connections[ch].loc_cids.remove(&seq) {
118                    trace!("peer retired CID {}: {}", seq, cid);
119                    self.index.retire(cid);
120                    if allow_more_cids {
121                        return Some(self.send_new_identifiers(now, ch, 1));
122                    }
123                }
124            }
125            Drained => {
126                if let Some(conn) = self.connections.try_remove(ch.0) {
127                    self.index.remove(&conn);
128                } else {
129                    // This indicates a bug in downstream code, which could cause spurious
130                    // connection loss instead of this error if the CID was (re)allocated prior to
131                    // the illegal call.
132                    error!(id = ch.0, "unknown connection drained");
133                }
134            }
135        }
136        None
137    }
138
139    /// Process an incoming UDP datagram
140    pub fn handle(
141        &mut self,
142        now: Instant,
143        remote: SocketAddr,
144        local_ip: Option<IpAddr>,
145        ecn: Option<EcnCodepoint>,
146        data: BytesMut,
147        buf: &mut Vec<u8>,
148    ) -> Option<DatagramEvent> {
149        // Partially decode packet or short-circuit if unable
150        let datagram_len = data.len();
151        let event = match PartialDecode::new(
152            data,
153            &FixedLengthConnectionIdParser::new(self.local_cid_generator.cid_len()),
154            &self.config.supported_versions,
155            self.config.grease_quic_bit,
156        ) {
157            Ok((first_decode, remaining)) => DatagramConnectionEvent {
158                now,
159                remote,
160                ecn,
161                first_decode,
162                remaining,
163            },
164            Err(PacketDecodeError::UnsupportedVersion {
165                src_cid,
166                dst_cid,
167                version,
168            }) => {
169                if self.server_config.is_none() {
170                    debug!("dropping packet with unsupported version");
171                    return None;
172                }
173                trace!("sending version negotiation");
174                // Negotiate versions
175                Header::VersionNegotiate {
176                    random: self.rng.random::<u8>() | 0x40,
177                    src_cid: dst_cid,
178                    dst_cid: src_cid,
179                }
180                .encode(buf);
181                // Grease with a reserved version
182                buf.write::<u32>(match version {
183                    0x0a1a_2a3a => 0x0a1a_2a4a,
184                    _ => 0x0a1a_2a3a,
185                });
186                for &version in &self.config.supported_versions {
187                    buf.write(version);
188                }
189                return Some(DatagramEvent::Response(Transmit {
190                    destination: remote,
191                    ecn: None,
192                    size: buf.len(),
193                    segment_size: None,
194                    src_ip: local_ip,
195                }));
196            }
197            Err(e) => {
198                trace!("malformed header: {}", e);
199                return None;
200            }
201        };
202
203        let addresses = FourTuple { remote, local_ip };
204        let dst_cid = event.first_decode.dst_cid();
205
206        if let Some(route_to) = self.index.get(&addresses, &event.first_decode) {
207            // Handle packet on existing connection
208            match route_to {
209                RouteDatagramTo::Incoming(incoming_idx) => {
210                    let incoming_buffer = &mut self.incoming_buffers[incoming_idx];
211                    let config = &self.server_config.as_ref().unwrap();
212
213                    if incoming_buffer
214                        .total_bytes
215                        .checked_add(datagram_len as u64)
216                        .is_some_and(|n| n <= config.incoming_buffer_size)
217                        && self
218                            .all_incoming_buffers_total_bytes
219                            .checked_add(datagram_len as u64)
220                            .is_some_and(|n| n <= config.incoming_buffer_size_total)
221                    {
222                        incoming_buffer.datagrams.push(event);
223                        incoming_buffer.total_bytes += datagram_len as u64;
224                        self.all_incoming_buffers_total_bytes += datagram_len as u64;
225                    }
226
227                    None
228                }
229                RouteDatagramTo::Connection(ch) => Some(DatagramEvent::ConnectionEvent(
230                    ch,
231                    ConnectionEvent(ConnectionEventInner::Datagram(event)),
232                )),
233            }
234        } else if event.first_decode.initial_header().is_some() {
235            // Potentially create a new connection
236
237            self.handle_first_packet(datagram_len, event, addresses, buf)
238        } else if event.first_decode.has_long_header() {
239            debug!(
240                "ignoring non-initial packet for unknown connection {}",
241                dst_cid
242            );
243            None
244        } else if !event.first_decode.is_initial()
245            && self.local_cid_generator.validate(dst_cid).is_err()
246        {
247            debug!("dropping packet with invalid CID");
248            None
249        } else if dst_cid.is_empty() {
250            trace!("dropping unrecognized short packet without ID");
251            None
252        } else {
253            // If we got this far, we're receiving a seemingly valid packet for an unknown
254            // connection. Send a stateless reset if possible.
255            self.stateless_reset(now, datagram_len, addresses, *dst_cid, buf)
256                .map(DatagramEvent::Response)
257        }
258    }
259
260    fn stateless_reset(
261        &mut self,
262        now: Instant,
263        inciting_dgram_len: usize,
264        addresses: FourTuple,
265        dst_cid: ConnectionId,
266        buf: &mut Vec<u8>,
267    ) -> Option<Transmit> {
268        if self
269            .last_stateless_reset
270            .is_some_and(|last| last + self.config.min_reset_interval > now)
271        {
272            debug!("ignoring unexpected packet within minimum stateless reset interval");
273            return None;
274        }
275
276        /// Minimum amount of padding for the stateless reset to look like a short-header packet
277        const MIN_PADDING_LEN: usize = 5;
278
279        // Prevent amplification attacks and reset loops by ensuring we pad to at most 1 byte
280        // smaller than the inciting packet.
281        let max_padding_len = match inciting_dgram_len.checked_sub(RESET_TOKEN_SIZE) {
282            Some(headroom) if headroom > MIN_PADDING_LEN => headroom - 1,
283            _ => {
284                debug!(
285                    "ignoring unexpected {} byte packet: not larger than minimum stateless reset size",
286                    inciting_dgram_len
287                );
288                return None;
289            }
290        };
291
292        debug!(
293            "sending stateless reset for {} to {}",
294            dst_cid, addresses.remote
295        );
296        self.last_stateless_reset = Some(now);
297        // Resets with at least this much padding can't possibly be distinguished from real packets
298        const IDEAL_MIN_PADDING_LEN: usize = MIN_PADDING_LEN + MAX_CID_SIZE;
299        let padding_len = if max_padding_len <= IDEAL_MIN_PADDING_LEN {
300            max_padding_len
301        } else {
302            self.rng
303                .random_range(IDEAL_MIN_PADDING_LEN..max_padding_len)
304        };
305        buf.reserve(padding_len + RESET_TOKEN_SIZE);
306        buf.resize(padding_len, 0);
307        self.rng.fill_bytes(&mut buf[0..padding_len]);
308        buf[0] = 0b0100_0000 | (buf[0] >> 2);
309        buf.extend_from_slice(&ResetToken::new(&*self.config.reset_key, dst_cid));
310
311        debug_assert!(buf.len() < inciting_dgram_len);
312
313        Some(Transmit {
314            destination: addresses.remote,
315            ecn: None,
316            size: buf.len(),
317            segment_size: None,
318            src_ip: addresses.local_ip,
319        })
320    }
321
322    /// Initiate a connection
323    pub fn connect(
324        &mut self,
325        now: Instant,
326        config: ClientConfig,
327        remote: SocketAddr,
328        server_name: &str,
329    ) -> Result<(ConnectionHandle, Connection), ConnectError> {
330        if self.cids_exhausted() {
331            return Err(ConnectError::CidsExhausted);
332        }
333        if remote.port() == 0 || remote.ip().is_unspecified() {
334            return Err(ConnectError::InvalidRemoteAddress(remote));
335        }
336        if !self.config.supported_versions.contains(&config.version) {
337            return Err(ConnectError::UnsupportedVersion);
338        }
339
340        let remote_id = (config.initial_dst_cid_provider)();
341        trace!(initial_dcid = %remote_id);
342
343        let ch = ConnectionHandle(self.connections.vacant_key());
344        let loc_cid = self.new_cid(ch);
345        let params = TransportParameters::new(
346            &config.transport,
347            &self.config,
348            self.local_cid_generator.as_ref(),
349            loc_cid,
350            None,
351            &mut self.rng,
352        );
353        let tls = config
354            .crypto
355            .start_session(config.version, server_name, &params)?;
356
357        let conn = self.add_connection(
358            ch,
359            config.version,
360            remote_id,
361            loc_cid,
362            remote_id,
363            FourTuple {
364                remote,
365                local_ip: None,
366            },
367            now,
368            tls,
369            config.transport,
370            SideArgs::Client {
371                token_store: config.token_store,
372                server_name: server_name.into(),
373            },
374        );
375        Ok((ch, conn))
376    }
377
378    fn send_new_identifiers(
379        &mut self,
380        now: Instant,
381        ch: ConnectionHandle,
382        num: u64,
383    ) -> ConnectionEvent {
384        let mut ids = vec![];
385        for _ in 0..num {
386            let id = self.new_cid(ch);
387            let meta = &mut self.connections[ch];
388            let sequence = meta.cids_issued;
389            meta.cids_issued += 1;
390            meta.loc_cids.insert(sequence, id);
391            ids.push(IssuedCid {
392                sequence,
393                id,
394                reset_token: ResetToken::new(&*self.config.reset_key, id),
395            });
396        }
397        ConnectionEvent(ConnectionEventInner::NewIdentifiers(ids, now))
398    }
399
400    /// Generate a connection ID for `ch`
401    fn new_cid(&mut self, ch: ConnectionHandle) -> ConnectionId {
402        loop {
403            let cid = self.local_cid_generator.generate_cid();
404            if cid.is_empty() {
405                // Zero-length CID; nothing to track
406                debug_assert_eq!(self.local_cid_generator.cid_len(), 0);
407                return cid;
408            }
409            if let hash_map::Entry::Vacant(e) = self.index.connection_ids.entry(cid) {
410                e.insert(ch);
411                break cid;
412            }
413        }
414    }
415
416    fn handle_first_packet(
417        &mut self,
418        datagram_len: usize,
419        event: DatagramConnectionEvent,
420        addresses: FourTuple,
421        buf: &mut Vec<u8>,
422    ) -> Option<DatagramEvent> {
423        let dst_cid = event.first_decode.dst_cid();
424        let header = event.first_decode.initial_header().unwrap();
425
426        let Some(server_config) = &self.server_config else {
427            debug!("packet for unrecognized connection {}", dst_cid);
428            return self
429                .stateless_reset(event.now, datagram_len, addresses, *dst_cid, buf)
430                .map(DatagramEvent::Response);
431        };
432
433        if datagram_len < MIN_INITIAL_SIZE as usize {
434            debug!("ignoring short initial for connection {}", dst_cid);
435            return None;
436        }
437
438        // Saturation only happens under heavy load, where deriving initial keys per Initial just to
439        // reply with CONNECTION_REFUSED would starve packet processing for existing connections.
440        if self.cids_exhausted() || self.incoming_buffers.len() >= server_config.max_incoming {
441            debug!(
442                "ignoring initial for connection {} due to saturation",
443                dst_cid
444            );
445            return None;
446        }
447
448        let crypto = match server_config.crypto.initial_keys(header.version, dst_cid) {
449            Ok(keys) => keys,
450            Err(UnsupportedVersion) => {
451                // This probably indicates that the user set supported_versions incorrectly in
452                // `EndpointConfig`.
453                debug!(
454                    "ignoring initial packet version {:#x} unsupported by cryptographic layer",
455                    header.version
456                );
457                return None;
458            }
459        };
460
461        if let Err(reason) = self.early_validate_first_packet(header) {
462            return Some(DatagramEvent::Response(self.initial_close(
463                header.version,
464                addresses,
465                &crypto,
466                &header.src_cid,
467                reason,
468                buf,
469            )));
470        }
471
472        let packet = match event.first_decode.finish(Some(&*crypto.header.remote)) {
473            Ok(packet) => packet,
474            Err(e) => {
475                trace!("unable to decode initial packet: {}", e);
476                return None;
477            }
478        };
479
480        if !packet.reserved_bits_valid() {
481            debug!("dropping connection attempt with invalid reserved bits");
482            return None;
483        }
484
485        let Header::Initial(header) = packet.header else {
486            panic!("non-initial packet in handle_first_packet()");
487        };
488
489        let server_config = self.server_config.as_ref().unwrap().clone();
490
491        let token = match IncomingToken::from_header(&header, &server_config, addresses.remote) {
492            Ok(token) => token,
493            Err(InvalidRetryTokenError) => {
494                debug!("rejecting invalid retry token");
495                return Some(DatagramEvent::Response(self.initial_close(
496                    header.version,
497                    addresses,
498                    &crypto,
499                    &header.src_cid,
500                    TransportError::INVALID_TOKEN(""),
501                    buf,
502                )));
503            }
504        };
505
506        let incoming_idx = self.incoming_buffers.insert(IncomingBuffer::default());
507        self.index
508            .insert_initial_incoming(header.dst_cid, incoming_idx);
509
510        Some(DatagramEvent::NewConnection(Incoming {
511            received_at: event.now,
512            addresses,
513            ecn: event.ecn,
514            packet: InitialPacket {
515                header,
516                header_data: packet.header_data,
517                payload: packet.payload,
518            },
519            rest: event.remaining,
520            crypto,
521            token,
522            incoming_idx,
523            improper_drop_warner: IncomingImproperDropWarner,
524        }))
525    }
526
527    /// Attempt to accept this incoming connection (an error may still occur)
528    // AcceptError cannot be made smaller without semver breakage
529    #[allow(clippy::result_large_err)]
530    pub fn accept(
531        &mut self,
532        mut incoming: Incoming,
533        now: Instant,
534        buf: &mut Vec<u8>,
535        server_config: Option<Arc<ServerConfig>>,
536    ) -> Result<(ConnectionHandle, Connection), AcceptError> {
537        let remote_address_validated = incoming.remote_address_validated();
538        incoming.improper_drop_warner.dismiss();
539        let incoming_buffer = self.incoming_buffers.remove(incoming.incoming_idx);
540        self.all_incoming_buffers_total_bytes -= incoming_buffer.total_bytes;
541
542        let packet_number = incoming.packet.header.number.expand(0);
543        let InitialHeader {
544            src_cid,
545            dst_cid,
546            version,
547            ..
548        } = incoming.packet.header;
549        let server_config =
550            server_config.unwrap_or_else(|| self.server_config.as_ref().unwrap().clone());
551
552        if server_config
553            .transport
554            .max_idle_timeout
555            .is_some_and(|timeout| {
556                incoming.received_at + Duration::from_millis(timeout.into()) <= now
557            })
558        {
559            debug!("abandoning accept of stale initial");
560            self.index.remove_initial(dst_cid);
561            return Err(AcceptError {
562                cause: ConnectionError::TimedOut,
563                response: None,
564            });
565        }
566
567        if self.cids_exhausted() {
568            debug!("refusing connection");
569            self.index.remove_initial(dst_cid);
570            return Err(AcceptError {
571                cause: ConnectionError::CidsExhausted,
572                response: Some(self.initial_close(
573                    version,
574                    incoming.addresses,
575                    &incoming.crypto,
576                    &src_cid,
577                    TransportError::CONNECTION_REFUSED(""),
578                    buf,
579                )),
580            });
581        }
582
583        if incoming
584            .crypto
585            .packet
586            .remote
587            .decrypt(
588                packet_number,
589                &incoming.packet.header_data,
590                &mut incoming.packet.payload,
591            )
592            .is_err()
593        {
594            debug!(packet_number, "failed to authenticate initial packet");
595            self.index.remove_initial(dst_cid);
596            return Err(AcceptError {
597                cause: TransportError::PROTOCOL_VIOLATION("authentication failed").into(),
598                response: None,
599            });
600        };
601
602        let ch = ConnectionHandle(self.connections.vacant_key());
603        let loc_cid = self.new_cid(ch);
604        let mut params = TransportParameters::new(
605            &server_config.transport,
606            &self.config,
607            self.local_cid_generator.as_ref(),
608            loc_cid,
609            Some(&server_config),
610            &mut self.rng,
611        );
612        params.stateless_reset_token = Some(ResetToken::new(&*self.config.reset_key, loc_cid));
613        params.original_dst_cid = Some(incoming.token.orig_dst_cid);
614        params.retry_src_cid = incoming.token.retry_src_cid;
615        let mut pref_addr_cid = None;
616        if server_config.has_preferred_address() {
617            let cid = self.new_cid(ch);
618            pref_addr_cid = Some(cid);
619            params.preferred_address = Some(PreferredAddress {
620                address_v4: server_config.preferred_address_v4,
621                address_v6: server_config.preferred_address_v6,
622                connection_id: cid,
623                stateless_reset_token: ResetToken::new(&*self.config.reset_key, cid),
624            });
625        }
626
627        let tls = server_config.crypto.clone().start_session(version, &params);
628        let transport_config = server_config.transport.clone();
629        let mut conn = self.add_connection(
630            ch,
631            version,
632            dst_cid,
633            loc_cid,
634            src_cid,
635            incoming.addresses,
636            incoming.received_at,
637            tls,
638            transport_config,
639            SideArgs::Server {
640                server_config,
641                pref_addr_cid,
642                path_validated: remote_address_validated,
643            },
644        );
645        self.index.insert_initial(dst_cid, ch);
646
647        match conn.handle_first_packet(
648            incoming.received_at,
649            incoming.addresses.remote,
650            incoming.ecn,
651            packet_number,
652            incoming.packet,
653            incoming.rest,
654        ) {
655            Ok(()) => {
656                trace!(id = ch.0, icid = %dst_cid, "new connection");
657
658                for event in incoming_buffer.datagrams {
659                    conn.handle_event(ConnectionEvent(ConnectionEventInner::Datagram(event)))
660                }
661
662                Ok((ch, conn))
663            }
664            Err(e) => {
665                debug!("handshake failed: {}", e);
666                self.handle_event(ch, EndpointEvent(EndpointEventInner::Drained));
667                let response = match e {
668                    ConnectionError::TransportError(ref e) => Some(self.initial_close(
669                        version,
670                        incoming.addresses,
671                        &incoming.crypto,
672                        &src_cid,
673                        e.clone(),
674                        buf,
675                    )),
676                    _ => None,
677                };
678                Err(AcceptError { cause: e, response })
679            }
680        }
681    }
682
683    /// Check if we should refuse a connection attempt regardless of the packet's contents
684    fn early_validate_first_packet(
685        &mut self,
686        header: &ProtectedInitialHeader,
687    ) -> Result<(), TransportError> {
688        // RFC9000 §7.2 dictates that initial (client-chosen) destination CIDs must be at least 8
689        // bytes. If this is a Retry packet, then the length must instead match our usual CID
690        // length. If we ever issue non-Retry address validation tokens via `NEW_TOKEN`, then we'll
691        // also need to validate CID length for those after decoding the token.
692        if header.dst_cid.len() < 8
693            && (header.token_pos.is_empty()
694                || header.dst_cid.len() != self.local_cid_generator.cid_len())
695        {
696            debug!(
697                "rejecting connection due to invalid DCID length {}",
698                header.dst_cid.len()
699            );
700            return Err(TransportError::PROTOCOL_VIOLATION(
701                "invalid destination CID length",
702            ));
703        }
704
705        Ok(())
706    }
707
708    /// Reject this incoming connection attempt
709    pub fn refuse(&mut self, incoming: Incoming, buf: &mut Vec<u8>) -> Transmit {
710        self.clean_up_incoming(&incoming);
711        incoming.improper_drop_warner.dismiss();
712
713        self.initial_close(
714            incoming.packet.header.version,
715            incoming.addresses,
716            &incoming.crypto,
717            &incoming.packet.header.src_cid,
718            TransportError::CONNECTION_REFUSED(""),
719            buf,
720        )
721    }
722
723    /// Respond with a retry packet, requiring the client to retry with address validation
724    ///
725    /// Errors if `incoming.may_retry()` is false.
726    pub fn retry(&mut self, incoming: Incoming, buf: &mut Vec<u8>) -> Result<Transmit, RetryError> {
727        if !incoming.may_retry() {
728            return Err(RetryError(Box::new(incoming)));
729        }
730
731        self.clean_up_incoming(&incoming);
732        incoming.improper_drop_warner.dismiss();
733
734        let server_config = self.server_config.as_ref().unwrap();
735
736        // First Initial
737        // The peer will use this as the DCID of its following Initials. Initial DCIDs are
738        // looked up separately from Handshake/Data DCIDs, so there is no risk of collision
739        // with established connections. In the unlikely event that a collision occurs
740        // between two connections in the initial phase, both will fail fast and may be
741        // retried by the application layer.
742        let loc_cid = self.local_cid_generator.generate_cid();
743
744        let payload = TokenPayload::Retry {
745            address: incoming.addresses.remote,
746            orig_dst_cid: incoming.packet.header.dst_cid,
747            issued: server_config.time_source.now(),
748        };
749        let token = Token::new(payload, &mut self.rng).encode(&*server_config.token_key);
750
751        let header = Header::Retry {
752            src_cid: loc_cid,
753            dst_cid: incoming.packet.header.src_cid,
754            version: incoming.packet.header.version,
755        };
756
757        let encode = header.encode(buf);
758        buf.put_slice(&token);
759        buf.extend_from_slice(&server_config.crypto.retry_tag(
760            incoming.packet.header.version,
761            &incoming.packet.header.dst_cid,
762            buf,
763        ));
764        encode.finish(buf, &*incoming.crypto.header.local, None);
765
766        Ok(Transmit {
767            destination: incoming.addresses.remote,
768            ecn: None,
769            size: buf.len(),
770            segment_size: None,
771            src_ip: incoming.addresses.local_ip,
772        })
773    }
774
775    /// Ignore this incoming connection attempt, not sending any packet in response
776    ///
777    /// Doing this actively, rather than merely dropping the [`Incoming`], is necessary to prevent
778    /// memory leaks due to state within [`Endpoint`] tracking the incoming connection.
779    pub fn ignore(&mut self, incoming: Incoming) {
780        self.clean_up_incoming(&incoming);
781        incoming.improper_drop_warner.dismiss();
782    }
783
784    /// Clean up endpoint data structures associated with an `Incoming`.
785    fn clean_up_incoming(&mut self, incoming: &Incoming) {
786        self.index.remove_initial(incoming.packet.header.dst_cid);
787        let incoming_buffer = self.incoming_buffers.remove(incoming.incoming_idx);
788        self.all_incoming_buffers_total_bytes -= incoming_buffer.total_bytes;
789    }
790
791    fn add_connection(
792        &mut self,
793        ch: ConnectionHandle,
794        version: u32,
795        init_cid: ConnectionId,
796        loc_cid: ConnectionId,
797        rem_cid: ConnectionId,
798        addresses: FourTuple,
799        now: Instant,
800        tls: Box<dyn crypto::Session>,
801        transport_config: Arc<TransportConfig>,
802        side_args: SideArgs,
803    ) -> Connection {
804        let mut rng_seed = [0; 32];
805        self.rng.fill_bytes(&mut rng_seed);
806        let side = side_args.side();
807        let pref_addr_cid = side_args.pref_addr_cid();
808        let conn = Connection::new(
809            self.config.clone(),
810            transport_config,
811            init_cid,
812            loc_cid,
813            rem_cid,
814            addresses.remote,
815            addresses.local_ip,
816            tls,
817            self.local_cid_generator.as_ref(),
818            now,
819            version,
820            self.allow_mtud,
821            rng_seed,
822            side_args,
823        );
824
825        let mut cids_issued = 0;
826        let mut loc_cids = FxHashMap::default();
827
828        loc_cids.insert(cids_issued, loc_cid);
829        cids_issued += 1;
830
831        if let Some(cid) = pref_addr_cid {
832            debug_assert_eq!(cids_issued, 1, "preferred address cid seq must be 1");
833            loc_cids.insert(cids_issued, cid);
834            cids_issued += 1;
835        }
836
837        let id = self.connections.insert(ConnectionMeta {
838            init_cid,
839            cids_issued,
840            loc_cids,
841            addresses,
842            side,
843            reset_token: None,
844        });
845        debug_assert_eq!(id, ch.0, "connection handle allocation out of sync");
846
847        self.index.insert_conn(addresses, loc_cid, ch, side);
848
849        conn
850    }
851
852    fn initial_close(
853        &mut self,
854        version: u32,
855        addresses: FourTuple,
856        crypto: &Keys,
857        remote_id: &ConnectionId,
858        reason: TransportError,
859        buf: &mut Vec<u8>,
860    ) -> Transmit {
861        // We don't need to worry about CID collisions in initial closes because the peer
862        // shouldn't respond, and if it does, and the CID collides, we'll just drop the
863        // unexpected response.
864        let local_id = self.local_cid_generator.generate_cid();
865        let number = PacketNumber::U8(0);
866        let header = Header::Initial(InitialHeader {
867            dst_cid: *remote_id,
868            src_cid: local_id,
869            number,
870            token: Bytes::new(),
871            version,
872        });
873
874        let partial_encode = header.encode(buf);
875        let max_len =
876            INITIAL_MTU as usize - partial_encode.header_len - crypto.packet.local.tag_len();
877        frame::Close::from(reason).encode(buf, max_len);
878        buf.resize(buf.len() + crypto.packet.local.tag_len(), 0);
879        partial_encode.finish(buf, &*crypto.header.local, Some((0, &*crypto.packet.local)));
880        Transmit {
881            destination: addresses.remote,
882            ecn: None,
883            size: buf.len(),
884            segment_size: None,
885            src_ip: addresses.local_ip,
886        }
887    }
888
889    /// Access the configuration used by this endpoint
890    pub fn config(&self) -> &EndpointConfig {
891        &self.config
892    }
893
894    /// Number of connections that are currently open
895    pub fn open_connections(&self) -> usize {
896        self.connections.len()
897    }
898
899    /// Counter for the number of bytes currently used
900    /// in the buffers for Initial and 0-RTT messages for pending incoming connections
901    pub fn incoming_buffer_bytes(&self) -> u64 {
902        self.all_incoming_buffers_total_bytes
903    }
904
905    #[cfg(test)]
906    pub(crate) fn known_connections(&self) -> usize {
907        let x = self.connections.len();
908        debug_assert_eq!(x, self.index.connection_ids_initial.len());
909        // Not all connections have known reset tokens
910        debug_assert!(x >= self.index.connection_reset_tokens.0.len());
911        // Not all connections have unique remotes, and 0-length CIDs might not be in use.
912        debug_assert!(x >= self.index.incoming_connection_remotes.len());
913        debug_assert!(x >= self.index.outgoing_connection_remotes.len());
914        x
915    }
916
917    #[cfg(test)]
918    pub(crate) fn known_cids(&self) -> usize {
919        self.index.connection_ids.len()
920    }
921
922    /// Whether we've used up 3/4 of the available CID space
923    ///
924    /// We leave some space unused so that `new_cid` can be relied upon to finish quickly. We don't
925    /// bother to check when CID longer than 4 bytes are used because 2^40 connections is a lot.
926    fn cids_exhausted(&self) -> bool {
927        self.local_cid_generator.cid_len() <= 4
928            && self.local_cid_generator.cid_len() != 0
929            && (2usize.pow(self.local_cid_generator.cid_len() as u32 * 8)
930                - self.index.connection_ids.len())
931                < 2usize.pow(self.local_cid_generator.cid_len() as u32 * 8 - 2)
932    }
933}
934
935impl fmt::Debug for Endpoint {
936    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
937        fmt.debug_struct("Endpoint")
938            .field("rng", &self.rng)
939            .field("index", &self.index)
940            .field("connections", &self.connections)
941            .field("config", &self.config)
942            .field("server_config", &self.server_config)
943            // incoming_buffers too large
944            .field("incoming_buffers.len", &self.incoming_buffers.len())
945            .field(
946                "all_incoming_buffers_total_bytes",
947                &self.all_incoming_buffers_total_bytes,
948            )
949            .finish()
950    }
951}
952
953/// Buffered Initial and 0-RTT messages for a pending incoming connection
954#[derive(Default)]
955struct IncomingBuffer {
956    datagrams: Vec<DatagramConnectionEvent>,
957    total_bytes: u64,
958}
959
960/// Part of protocol state incoming datagrams can be routed to
961#[derive(Copy, Clone, Debug)]
962enum RouteDatagramTo {
963    Incoming(usize),
964    Connection(ConnectionHandle),
965}
966
967/// Maps packets to existing connections
968#[derive(Default, Debug)]
969struct ConnectionIndex {
970    /// Identifies connections based on the initial DCID the peer utilized
971    ///
972    /// Uses a standard `HashMap` to protect against hash collision attacks.
973    ///
974    /// Used by the server, not the client.
975    connection_ids_initial: HashMap<ConnectionId, RouteDatagramTo>,
976    /// Identifies connections based on locally created CIDs
977    ///
978    /// Uses a cheaper hash function since keys are locally created
979    connection_ids: FxHashMap<ConnectionId, ConnectionHandle>,
980    /// Identifies incoming connections with zero-length CIDs
981    ///
982    /// Uses a standard `HashMap` to protect against hash collision attacks.
983    incoming_connection_remotes: HashMap<FourTuple, ConnectionHandle>,
984    /// Identifies outgoing connections with zero-length CIDs
985    ///
986    /// We don't yet support explicit source addresses for client connections, and zero-length CIDs
987    /// require a unique four-tuple, so at most one client connection with zero-length local CIDs
988    /// may be established per remote. We must omit the local address from the key because we don't
989    /// necessarily know what address we're sending from, and hence receiving at.
990    ///
991    /// Uses a standard `HashMap` to protect against hash collision attacks.
992    outgoing_connection_remotes: HashMap<SocketAddr, ConnectionHandle>,
993    /// Reset tokens provided by the peer for the CID each connection is currently sending to
994    ///
995    /// Incoming stateless resets do not have correct CIDs, so we need this to identify the correct
996    /// recipient, if any.
997    connection_reset_tokens: ResetTokenTable,
998}
999
1000impl ConnectionIndex {
1001    /// Associate an incoming connection with its initial destination CID
1002    fn insert_initial_incoming(&mut self, dst_cid: ConnectionId, incoming_key: usize) {
1003        if dst_cid.is_empty() {
1004            return;
1005        }
1006        self.connection_ids_initial
1007            .insert(dst_cid, RouteDatagramTo::Incoming(incoming_key));
1008    }
1009
1010    /// Remove an association with an initial destination CID
1011    fn remove_initial(&mut self, dst_cid: ConnectionId) {
1012        if dst_cid.is_empty() {
1013            return;
1014        }
1015        let removed = self.connection_ids_initial.remove(&dst_cid);
1016        debug_assert!(removed.is_some());
1017    }
1018
1019    /// Associate a connection with its initial destination CID
1020    fn insert_initial(&mut self, dst_cid: ConnectionId, connection: ConnectionHandle) {
1021        if dst_cid.is_empty() {
1022            return;
1023        }
1024        self.connection_ids_initial
1025            .insert(dst_cid, RouteDatagramTo::Connection(connection));
1026    }
1027
1028    /// Associate a connection with its first locally-chosen destination CID if used, or otherwise
1029    /// its current 4-tuple
1030    fn insert_conn(
1031        &mut self,
1032        addresses: FourTuple,
1033        dst_cid: ConnectionId,
1034        connection: ConnectionHandle,
1035        side: Side,
1036    ) {
1037        match dst_cid.len() {
1038            0 => match side {
1039                Side::Server => {
1040                    self.incoming_connection_remotes
1041                        .insert(addresses, connection);
1042                }
1043                Side::Client => {
1044                    self.outgoing_connection_remotes
1045                        .insert(addresses.remote, connection);
1046                }
1047            },
1048            _ => {
1049                self.connection_ids.insert(dst_cid, connection);
1050            }
1051        }
1052    }
1053
1054    /// Discard a connection ID
1055    fn retire(&mut self, dst_cid: ConnectionId) {
1056        self.connection_ids.remove(&dst_cid);
1057    }
1058
1059    /// Remove all references to a connection
1060    fn remove(&mut self, conn: &ConnectionMeta) {
1061        if conn.side.is_server() {
1062            self.remove_initial(conn.init_cid);
1063        }
1064        for cid in conn.loc_cids.values() {
1065            self.connection_ids.remove(cid);
1066        }
1067        self.incoming_connection_remotes.remove(&conn.addresses);
1068        self.outgoing_connection_remotes
1069            .remove(&conn.addresses.remote);
1070        if let Some((remote, token)) = conn.reset_token {
1071            self.connection_reset_tokens.remove(remote, token);
1072        }
1073    }
1074
1075    /// Find the existing connection that `datagram` should be routed to, if any
1076    fn get(&self, addresses: &FourTuple, datagram: &PartialDecode) -> Option<RouteDatagramTo> {
1077        if !datagram.dst_cid().is_empty() {
1078            if let Some(&ch) = self.connection_ids.get(datagram.dst_cid()) {
1079                return Some(RouteDatagramTo::Connection(ch));
1080            }
1081        }
1082        if datagram.is_initial() || datagram.is_0rtt() {
1083            if let Some(&ch) = self.connection_ids_initial.get(datagram.dst_cid()) {
1084                return Some(ch);
1085            }
1086        }
1087        if datagram.dst_cid().is_empty() {
1088            if let Some(&ch) = self.incoming_connection_remotes.get(addresses) {
1089                return Some(RouteDatagramTo::Connection(ch));
1090            }
1091            if let Some(&ch) = self.outgoing_connection_remotes.get(&addresses.remote) {
1092                return Some(RouteDatagramTo::Connection(ch));
1093            }
1094        }
1095        let data = datagram.data();
1096        if data.len() < RESET_TOKEN_SIZE {
1097            return None;
1098        }
1099        self.connection_reset_tokens
1100            .get(addresses.remote, &data[data.len() - RESET_TOKEN_SIZE..])
1101            .cloned()
1102            .map(RouteDatagramTo::Connection)
1103    }
1104}
1105
1106#[derive(Debug)]
1107pub(crate) struct ConnectionMeta {
1108    init_cid: ConnectionId,
1109    /// Number of local connection IDs that have been issued in NEW_CONNECTION_ID frames.
1110    cids_issued: u64,
1111    loc_cids: FxHashMap<u64, ConnectionId>,
1112    /// Remote/local addresses the connection began with
1113    ///
1114    /// Only needed to support connections with zero-length CIDs, which cannot migrate, so we don't
1115    /// bother keeping it up to date.
1116    addresses: FourTuple,
1117    side: Side,
1118    /// Reset token provided by the peer for the CID we're currently sending to, and the address
1119    /// being sent to
1120    reset_token: Option<(SocketAddr, ResetToken)>,
1121}
1122
1123/// Internal identifier for a `Connection` currently associated with an endpoint
1124#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
1125pub struct ConnectionHandle(pub usize);
1126
1127impl From<ConnectionHandle> for usize {
1128    fn from(x: ConnectionHandle) -> Self {
1129        x.0
1130    }
1131}
1132
1133impl Index<ConnectionHandle> for Slab<ConnectionMeta> {
1134    type Output = ConnectionMeta;
1135    fn index(&self, ch: ConnectionHandle) -> &ConnectionMeta {
1136        &self[ch.0]
1137    }
1138}
1139
1140impl IndexMut<ConnectionHandle> for Slab<ConnectionMeta> {
1141    fn index_mut(&mut self, ch: ConnectionHandle) -> &mut ConnectionMeta {
1142        &mut self[ch.0]
1143    }
1144}
1145
1146/// Event resulting from processing a single datagram
1147pub enum DatagramEvent {
1148    /// The datagram is redirected to its `Connection`
1149    ConnectionEvent(ConnectionHandle, ConnectionEvent),
1150    /// The datagram may result in starting a new `Connection`
1151    NewConnection(Incoming),
1152    /// Response generated directly by the endpoint
1153    Response(Transmit),
1154}
1155
1156/// An incoming connection for which the server has not yet begun its part of the handshake.
1157pub struct Incoming {
1158    received_at: Instant,
1159    addresses: FourTuple,
1160    ecn: Option<EcnCodepoint>,
1161    packet: InitialPacket,
1162    rest: Option<BytesMut>,
1163    crypto: Keys,
1164    token: IncomingToken,
1165    incoming_idx: usize,
1166    improper_drop_warner: IncomingImproperDropWarner,
1167}
1168
1169impl Incoming {
1170    /// The local IP address which was used when the peer established the connection
1171    ///
1172    /// This has the same behavior as [`Connection::local_ip`].
1173    pub fn local_ip(&self) -> Option<IpAddr> {
1174        self.addresses.local_ip
1175    }
1176
1177    /// The peer's UDP address
1178    pub fn remote_address(&self) -> SocketAddr {
1179        self.addresses.remote
1180    }
1181
1182    /// Whether the socket address that is initiating this connection has been validated
1183    ///
1184    /// This means that the sender of the initial packet has proved that they can receive traffic
1185    /// sent to `self.remote_address()`.
1186    ///
1187    /// If `self.remote_address_validated()` is false, `self.may_retry()` is guaranteed to be true.
1188    /// The inverse is not guaranteed.
1189    pub fn remote_address_validated(&self) -> bool {
1190        self.token.validated
1191    }
1192
1193    /// Whether it is legal to respond with a retry packet
1194    ///
1195    /// If `self.remote_address_validated()` is false, `self.may_retry()` is guaranteed to be true.
1196    /// The inverse is not guaranteed.
1197    pub fn may_retry(&self) -> bool {
1198        self.token.retry_src_cid.is_none()
1199    }
1200
1201    /// The original destination connection ID sent by the client
1202    pub fn orig_dst_cid(&self) -> &ConnectionId {
1203        &self.token.orig_dst_cid
1204    }
1205}
1206
1207impl fmt::Debug for Incoming {
1208    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
1209        f.debug_struct("Incoming")
1210            .field("addresses", &self.addresses)
1211            .field("ecn", &self.ecn)
1212            // packet doesn't implement debug
1213            // rest is too big and not meaningful enough
1214            .field("token", &self.token)
1215            .field("incoming_idx", &self.incoming_idx)
1216            // improper drop warner contains no information
1217            .finish_non_exhaustive()
1218    }
1219}
1220
1221struct IncomingImproperDropWarner;
1222
1223impl IncomingImproperDropWarner {
1224    fn dismiss(self) {
1225        mem::forget(self);
1226    }
1227}
1228
1229impl Drop for IncomingImproperDropWarner {
1230    fn drop(&mut self) {
1231        warn!(
1232            "quinn_proto::Incoming dropped without passing to Endpoint::accept/refuse/retry/ignore \
1233               (may cause memory leak and eventual inability to accept new connections)"
1234        );
1235    }
1236}
1237
1238/// Errors in the parameters being used to create a new connection
1239///
1240/// These arise before any I/O has been performed.
1241#[derive(Debug, Error, Clone, PartialEq, Eq)]
1242pub enum ConnectError {
1243    /// The endpoint can no longer create new connections
1244    ///
1245    /// Indicates that a necessary component of the endpoint has been dropped or otherwise disabled.
1246    #[error("endpoint stopping")]
1247    EndpointStopping,
1248    /// The connection could not be created because not enough of the CID space is available
1249    ///
1250    /// Try using longer connection IDs
1251    #[error("CIDs exhausted")]
1252    CidsExhausted,
1253    /// The given server name was malformed
1254    #[error("invalid server name: {0}")]
1255    InvalidServerName(String),
1256    /// The remote [`SocketAddr`] supplied was malformed
1257    ///
1258    /// Examples include attempting to connect to port 0, or using an inappropriate address family.
1259    #[error("invalid remote address: {0}")]
1260    InvalidRemoteAddress(SocketAddr),
1261    /// No default client configuration was set up
1262    ///
1263    /// Use `Endpoint::connect_with` to specify a client configuration.
1264    #[error("no default client config")]
1265    NoDefaultClientConfig,
1266    /// The local endpoint does not support the QUIC version specified in the client configuration
1267    #[error("unsupported QUIC version")]
1268    UnsupportedVersion,
1269}
1270
1271/// Error type for attempting to accept an [`Incoming`]
1272#[derive(Debug)]
1273pub struct AcceptError {
1274    /// Underlying error describing reason for failure
1275    pub cause: ConnectionError,
1276    /// Optional response to transmit back
1277    pub response: Option<Transmit>,
1278}
1279
1280/// Error for attempting to retry an [`Incoming`] which already bears a token from a previous retry
1281#[derive(Debug, Error)]
1282#[error("retry() with validated Incoming")]
1283pub struct RetryError(Box<Incoming>);
1284
1285impl RetryError {
1286    /// Get the [`Incoming`]
1287    pub fn into_incoming(self) -> Incoming {
1288        *self.0
1289    }
1290}
1291
1292/// Reset Tokens which are associated with peer socket addresses
1293///
1294/// The standard `HashMap` is used since both `SocketAddr` and `ResetToken` are
1295/// peer generated and might be usable for hash collision attacks.
1296#[derive(Default, Debug)]
1297struct ResetTokenTable(HashMap<SocketAddr, HashMap<ResetToken, ConnectionHandle>>);
1298
1299impl ResetTokenTable {
1300    fn insert(&mut self, remote: SocketAddr, token: ResetToken, ch: ConnectionHandle) -> bool {
1301        self.0
1302            .entry(remote)
1303            .or_default()
1304            .insert(token, ch)
1305            .is_some()
1306    }
1307
1308    fn remove(&mut self, remote: SocketAddr, token: ResetToken) {
1309        use std::collections::hash_map::Entry;
1310        match self.0.entry(remote) {
1311            Entry::Vacant(_) => {}
1312            Entry::Occupied(mut e) => {
1313                e.get_mut().remove(&token);
1314                if e.get().is_empty() {
1315                    e.remove_entry();
1316                }
1317            }
1318        }
1319    }
1320
1321    fn get(&self, remote: SocketAddr, token: &[u8]) -> Option<&ConnectionHandle> {
1322        let token = ResetToken::from(<[u8; RESET_TOKEN_SIZE]>::try_from(token).ok()?);
1323        self.0.get(&remote)?.get(&token)
1324    }
1325}
1326
1327/// Identifies a connection by the combination of remote and local addresses
1328///
1329/// Including the local ensures good behavior when the host has multiple IP addresses on the same
1330/// subnet and zero-length connection IDs are in use.
1331#[derive(Hash, Eq, PartialEq, Debug, Copy, Clone)]
1332struct FourTuple {
1333    remote: SocketAddr,
1334    // A single socket can only listen on a single port, so no need to store it explicitly
1335    local_ip: Option<IpAddr>,
1336}
quinn_proto/endpoint.rs

quinn_proto/
endpoint.rs
