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

quinn_proto/
endpoint.rs
