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, FourTuple, INITIAL_MTU, Instant, MAX_CID_SIZE, MIN_INITIAL_SIZE, PathId,
19 RESET_TOKEN_SIZE, ResetToken, Side, Transmit, TransportConfig, TransportError,
20 cid_generator::ConnectionIdGenerator,
21 coding::{BufMutExt, Encodable},
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
38pub 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 allow_mtud: bool,
51 last_stateless_reset: Option<Instant>,
53 incoming_buffers: Slab<IncomingBuffer>,
55 all_incoming_buffers_total_bytes: u64,
56}
57
58impl Endpoint {
59 pub fn new(
65 config: Arc<EndpointConfig>,
66 server_config: Option<Arc<ServerConfig>>,
67 allow_mtud: bool,
68 ) -> Self {
69 Self {
70 rng: config
71 .rng_seed
72 .map_or_else(StdRng::from_os_rng, StdRng::from_seed),
73 index: ConnectionIndex::default(),
74 connections: Slab::new(),
75 local_cid_generator: (config.connection_id_generator_factory.as_ref())(),
76 config,
77 server_config,
78 allow_mtud,
79 last_stateless_reset: None,
80 incoming_buffers: Slab::new(),
81 all_incoming_buffers_total_bytes: 0,
82 }
83 }
84
85 pub fn set_server_config(&mut self, server_config: Option<Arc<ServerConfig>>) {
87 self.server_config = server_config;
88 }
89
90 pub fn handle_event(
94 &mut self,
95 ch: ConnectionHandle,
96 event: EndpointEvent,
97 ) -> Option<ConnectionEvent> {
98 use EndpointEventInner::*;
99 match event.0 {
100 NeedIdentifiers(path_id, now, n) => {
101 return Some(self.send_new_identifiers(path_id, now, ch, n));
102 }
103 ResetToken(path_id, remote, token) => {
104 if let Some(old) = self.connections[ch]
105 .reset_token
106 .insert(path_id, (remote, token))
107 {
108 self.index.connection_reset_tokens.remove(old.0, old.1);
109 }
110 if self.index.connection_reset_tokens.insert(remote, token, ch) {
111 warn!("duplicate reset token");
112 }
113 }
114 RetireResetToken(path_id) => {
115 if let Some(old) = self.connections[ch].reset_token.remove(&path_id) {
116 self.index.connection_reset_tokens.remove(old.0, old.1);
117 }
118 }
119 RetireConnectionId(now, path_id, seq, allow_more_cids) => {
120 if let Some(cid) = self.connections[ch]
121 .loc_cids
122 .get_mut(&path_id)
123 .and_then(|pcid| pcid.cids.remove(&seq))
124 {
125 trace!(%path_id, "local CID retired {}: {}", seq, cid);
126 self.index.retire(cid);
127 if allow_more_cids {
128 return Some(self.send_new_identifiers(path_id, now, ch, 1));
129 }
130 }
131 }
132 Drained => {
133 if let Some(conn) = self.connections.try_remove(ch.0) {
134 self.index.remove(&conn);
135 } else {
136 error!(id = ch.0, "unknown connection drained");
140 }
141 }
142 }
143 None
144 }
145
146 pub fn handle(
148 &mut self,
149 now: Instant,
150 network_path: FourTuple,
151 ecn: Option<EcnCodepoint>,
152 data: BytesMut,
153 buf: &mut Vec<u8>,
154 ) -> Option<DatagramEvent> {
155 let datagram_len = data.len();
157 let mut event = match PartialDecode::new(
158 data,
159 &FixedLengthConnectionIdParser::new(self.local_cid_generator.cid_len()),
160 &self.config.supported_versions,
161 self.config.grease_quic_bit,
162 ) {
163 Ok((first_decode, remaining)) => DatagramConnectionEvent {
164 now,
165 network_path,
166 path_id: PathId::ZERO, ecn,
168 first_decode,
169 remaining,
170 },
171 Err(PacketDecodeError::UnsupportedVersion {
172 src_cid,
173 dst_cid,
174 version,
175 }) => {
176 if self.server_config.is_none() {
177 debug!("dropping packet with unsupported version");
178 return None;
179 }
180 trace!("sending version negotiation");
181 Header::VersionNegotiate {
183 random: self.rng.random::<u8>() | 0x40,
184 src_cid: dst_cid,
185 dst_cid: src_cid,
186 }
187 .encode(buf);
188 buf.write::<u32>(match version {
190 0x0a1a_2a3a => 0x0a1a_2a4a,
191 _ => 0x0a1a_2a3a,
192 });
193 for &version in &self.config.supported_versions {
194 buf.write(version);
195 }
196 return Some(DatagramEvent::Response(Transmit {
197 destination: network_path.remote,
198 ecn: None,
199 size: buf.len(),
200 segment_size: None,
201 src_ip: network_path.local_ip,
202 }));
203 }
204 Err(e) => {
205 trace!("malformed header: {}", e);
206 return None;
207 }
208 };
209
210 let dst_cid = event.first_decode.dst_cid();
211
212 if let Some(route_to) = self.index.get(&network_path, &event.first_decode) {
213 event.path_id = match route_to {
214 RouteDatagramTo::Incoming(_) => PathId::ZERO,
215 RouteDatagramTo::Connection(_, path_id) => path_id,
216 };
217 match route_to {
218 RouteDatagramTo::Incoming(incoming_idx) => {
219 let incoming_buffer = &mut self.incoming_buffers[incoming_idx];
220 let config = &self.server_config.as_ref().unwrap();
221
222 if incoming_buffer
223 .total_bytes
224 .checked_add(datagram_len as u64)
225 .is_some_and(|n| n <= config.incoming_buffer_size)
226 && self
227 .all_incoming_buffers_total_bytes
228 .checked_add(datagram_len as u64)
229 .is_some_and(|n| n <= config.incoming_buffer_size_total)
230 {
231 incoming_buffer.datagrams.push(event);
232 incoming_buffer.total_bytes += datagram_len as u64;
233 self.all_incoming_buffers_total_bytes += datagram_len as u64;
234 }
235
236 None
237 }
238 RouteDatagramTo::Connection(ch, _path_id) => Some(DatagramEvent::ConnectionEvent(
239 ch,
240 ConnectionEvent(ConnectionEventInner::Datagram(event)),
241 )),
242 }
243 } else if event.first_decode.initial_header().is_some() {
244 self.handle_first_packet(datagram_len, event, network_path, buf)
247 } else if event.first_decode.has_long_header() {
248 debug!(
249 "ignoring non-initial packet for unknown connection {}",
250 dst_cid
251 );
252 None
253 } else if !event.first_decode.is_initial()
254 && self.local_cid_generator.validate(dst_cid).is_err()
255 {
256 debug!("dropping packet with invalid CID");
257 None
258 } else if dst_cid.is_empty() {
259 trace!("dropping unrecognized short packet without ID");
260 None
261 } else {
262 self.stateless_reset(now, datagram_len, network_path, dst_cid, buf)
265 .map(DatagramEvent::Response)
266 }
267 }
268
269 fn stateless_reset(
271 &mut self,
272 now: Instant,
273 inciting_dgram_len: usize,
274 network_path: FourTuple,
275 dst_cid: ConnectionId,
276 buf: &mut Vec<u8>,
277 ) -> Option<Transmit> {
278 if self
279 .last_stateless_reset
280 .is_some_and(|last| last + self.config.min_reset_interval > now)
281 {
282 debug!("ignoring unexpected packet within minimum stateless reset interval");
283 return None;
284 }
285
286 const MIN_PADDING_LEN: usize = 5;
288
289 let max_padding_len = match inciting_dgram_len.checked_sub(RESET_TOKEN_SIZE) {
292 Some(headroom) if headroom > MIN_PADDING_LEN => headroom - 1,
293 _ => {
294 debug!(
295 "ignoring unexpected {} byte packet: not larger than minimum stateless reset size",
296 inciting_dgram_len
297 );
298 return None;
299 }
300 };
301
302 debug!(%dst_cid, %network_path.remote, "sending stateless reset");
303 self.last_stateless_reset = Some(now);
304 const IDEAL_MIN_PADDING_LEN: usize = MIN_PADDING_LEN + MAX_CID_SIZE;
306 let padding_len = if max_padding_len <= IDEAL_MIN_PADDING_LEN {
307 max_padding_len
308 } else {
309 self.rng
310 .random_range(IDEAL_MIN_PADDING_LEN..max_padding_len)
311 };
312 buf.reserve(padding_len + RESET_TOKEN_SIZE);
313 buf.resize(padding_len, 0);
314 self.rng.fill_bytes(&mut buf[0..padding_len]);
315 buf[0] = 0b0100_0000 | (buf[0] >> 2);
316 buf.extend_from_slice(&ResetToken::new(&*self.config.reset_key, dst_cid));
317
318 debug_assert!(buf.len() < inciting_dgram_len);
319
320 Some(Transmit {
321 destination: network_path.remote,
322 ecn: None,
323 size: buf.len(),
324 segment_size: None,
325 src_ip: network_path.local_ip,
326 })
327 }
328
329 pub fn connect(
331 &mut self,
332 now: Instant,
333 config: ClientConfig,
334 remote: SocketAddr,
335 server_name: &str,
336 ) -> Result<(ConnectionHandle, Connection), ConnectError> {
337 if self.cids_exhausted() {
338 return Err(ConnectError::CidsExhausted);
339 }
340 if remote.port() == 0 || remote.ip().is_unspecified() {
341 return Err(ConnectError::InvalidRemoteAddress(remote));
342 }
343 if !self.config.supported_versions.contains(&config.version) {
344 return Err(ConnectError::UnsupportedVersion);
345 }
346
347 let remote_id = (config.initial_dst_cid_provider)();
348 trace!(initial_dcid = %remote_id);
349
350 let ch = ConnectionHandle(self.connections.vacant_key());
351 let loc_cid = self.new_cid(ch, PathId::ZERO);
352 let params = TransportParameters::new(
353 &config.transport,
354 &self.config,
355 self.local_cid_generator.as_ref(),
356 loc_cid,
357 None,
358 &mut self.rng,
359 );
360 let tls = config
361 .crypto
362 .start_session(config.version, server_name, ¶ms)?;
363
364 let conn = self.add_connection(
365 ch,
366 config.version,
367 remote_id,
368 loc_cid,
369 remote_id,
370 FourTuple {
371 remote,
372 local_ip: None,
373 },
374 now,
375 tls,
376 config.transport,
377 SideArgs::Client {
378 token_store: config.token_store,
379 server_name: server_name.into(),
380 },
381 ¶ms,
382 );
383 Ok((ch, conn))
384 }
385
386 fn send_new_identifiers(
388 &mut self,
389 path_id: PathId,
390 now: Instant,
391 ch: ConnectionHandle,
392 num: u64,
393 ) -> ConnectionEvent {
394 let mut ids = vec![];
395 for _ in 0..num {
396 let id = self.new_cid(ch, path_id);
397 let cid_meta = self.connections[ch].loc_cids.entry(path_id).or_default();
398 let sequence = cid_meta.issued;
399 cid_meta.issued += 1;
400 cid_meta.cids.insert(sequence, id);
401 ids.push(IssuedCid {
402 path_id,
403 sequence,
404 id,
405 reset_token: ResetToken::new(&*self.config.reset_key, id),
406 });
407 }
408 ConnectionEvent(ConnectionEventInner::NewIdentifiers(
409 ids,
410 now,
411 self.local_cid_generator.cid_len(),
412 self.local_cid_generator.cid_lifetime(),
413 ))
414 }
415
416 fn new_cid(&mut self, ch: ConnectionHandle, path_id: PathId) -> ConnectionId {
418 loop {
419 let cid = self.local_cid_generator.generate_cid();
420 if cid.is_empty() {
421 debug_assert_eq!(self.local_cid_generator.cid_len(), 0);
423 return cid;
424 }
425 if let hash_map::Entry::Vacant(e) = self.index.connection_ids.entry(cid) {
426 e.insert((ch, path_id));
427 break cid;
428 }
429 }
430 }
431
432 fn handle_first_packet(
433 &mut self,
434 datagram_len: usize,
435 event: DatagramConnectionEvent,
436 network_path: FourTuple,
437 buf: &mut Vec<u8>,
438 ) -> Option<DatagramEvent> {
439 let dst_cid = event.first_decode.dst_cid();
440 let header = event.first_decode.initial_header().unwrap();
441
442 let Some(server_config) = &self.server_config else {
443 debug!("packet for unrecognized connection {}", dst_cid);
444 return self
445 .stateless_reset(event.now, datagram_len, network_path, dst_cid, buf)
446 .map(DatagramEvent::Response);
447 };
448
449 if datagram_len < MIN_INITIAL_SIZE as usize {
450 debug!("ignoring short initial for connection {}", dst_cid);
451 return None;
452 }
453
454 let crypto = match server_config.crypto.initial_keys(header.version, dst_cid) {
455 Ok(keys) => keys,
456 Err(UnsupportedVersion) => {
457 debug!(
460 "ignoring initial packet version {:#x} unsupported by cryptographic layer",
461 header.version
462 );
463 return None;
464 }
465 };
466
467 if let Err(reason) = self.early_validate_first_packet(header) {
468 return Some(DatagramEvent::Response(self.initial_close(
469 header.version,
470 network_path,
471 &crypto,
472 header.src_cid,
473 reason,
474 buf,
475 )));
476 }
477
478 let packet = match event.first_decode.finish(Some(&*crypto.header.remote)) {
479 Ok(packet) => packet,
480 Err(e) => {
481 trace!("unable to decode initial packet: {}", e);
482 return None;
483 }
484 };
485
486 if !packet.reserved_bits_valid() {
487 debug!("dropping connection attempt with invalid reserved bits");
488 return None;
489 }
490
491 let Header::Initial(header) = packet.header else {
492 panic!("non-initial packet in handle_first_packet()");
493 };
494
495 let server_config = self.server_config.as_ref().unwrap().clone();
496
497 let token = match IncomingToken::from_header(&header, &server_config, network_path.remote) {
498 Ok(token) => token,
499 Err(InvalidRetryTokenError) => {
500 debug!("rejecting invalid retry token");
501 return Some(DatagramEvent::Response(self.initial_close(
502 header.version,
503 network_path,
504 &crypto,
505 header.src_cid,
506 TransportError::INVALID_TOKEN(""),
507 buf,
508 )));
509 }
510 };
511
512 let incoming_idx = self.incoming_buffers.insert(IncomingBuffer::default());
513 self.index
514 .insert_initial_incoming(header.dst_cid, incoming_idx);
515
516 Some(DatagramEvent::NewConnection(Incoming {
517 received_at: event.now,
518 network_path,
519 ecn: event.ecn,
520 packet: InitialPacket {
521 header,
522 header_data: packet.header_data,
523 payload: packet.payload,
524 },
525 rest: event.remaining,
526 crypto,
527 token,
528 incoming_idx,
529 improper_drop_warner: IncomingImproperDropWarner,
530 }))
531 }
532
533 pub fn accept(
536 &mut self,
537 mut incoming: Incoming,
538 now: Instant,
539 buf: &mut Vec<u8>,
540 server_config: Option<Arc<ServerConfig>>,
541 ) -> Result<(ConnectionHandle, Connection), Box<AcceptError>> {
542 let remote_address_validated = incoming.remote_address_validated();
543 incoming.improper_drop_warner.dismiss();
544 let incoming_buffer = self.incoming_buffers.remove(incoming.incoming_idx);
545 self.all_incoming_buffers_total_bytes -= incoming_buffer.total_bytes;
546
547 let packet_number = incoming.packet.header.number.expand(0);
548 let InitialHeader {
549 src_cid,
550 dst_cid,
551 version,
552 ..
553 } = incoming.packet.header;
554 let server_config =
555 server_config.unwrap_or_else(|| self.server_config.as_ref().unwrap().clone());
556
557 if server_config
558 .transport
559 .max_idle_timeout
560 .is_some_and(|timeout| {
561 incoming.received_at + Duration::from_millis(timeout.into()) <= now
562 })
563 {
564 debug!("abandoning accept of stale initial");
565 self.index.remove_initial(dst_cid);
566 return Err(Box::new(AcceptError {
567 cause: ConnectionError::TimedOut,
568 response: None,
569 }));
570 }
571
572 if self.cids_exhausted() {
573 debug!("refusing connection");
574 self.index.remove_initial(dst_cid);
575 return Err(Box::new(AcceptError {
576 cause: ConnectionError::CidsExhausted,
577 response: Some(self.initial_close(
578 version,
579 incoming.network_path,
580 &incoming.crypto,
581 src_cid,
582 TransportError::CONNECTION_REFUSED(""),
583 buf,
584 )),
585 }));
586 }
587
588 if incoming
589 .crypto
590 .packet
591 .remote
592 .decrypt(
593 PathId::ZERO,
594 packet_number,
595 &incoming.packet.header_data,
596 &mut incoming.packet.payload,
597 )
598 .is_err()
599 {
600 debug!(packet_number, "failed to authenticate initial packet");
601 self.index.remove_initial(dst_cid);
602 return Err(Box::new(AcceptError {
603 cause: TransportError::PROTOCOL_VIOLATION("authentication failed").into(),
604 response: None,
605 }));
606 };
607
608 let ch = ConnectionHandle(self.connections.vacant_key());
609 let loc_cid = self.new_cid(ch, PathId::ZERO);
610 let mut params = TransportParameters::new(
611 &server_config.transport,
612 &self.config,
613 self.local_cid_generator.as_ref(),
614 loc_cid,
615 Some(&server_config),
616 &mut self.rng,
617 );
618 params.stateless_reset_token = Some(ResetToken::new(&*self.config.reset_key, loc_cid));
619 params.original_dst_cid = Some(incoming.token.orig_dst_cid);
620 params.retry_src_cid = incoming.token.retry_src_cid;
621 let mut pref_addr_cid = None;
622 if server_config.has_preferred_address() {
623 let cid = self.new_cid(ch, PathId::ZERO);
624 pref_addr_cid = Some(cid);
625 params.preferred_address = Some(PreferredAddress {
626 address_v4: server_config.preferred_address_v4,
627 address_v6: server_config.preferred_address_v6,
628 connection_id: cid,
629 stateless_reset_token: ResetToken::new(&*self.config.reset_key, cid),
630 });
631 }
632
633 let tls = server_config.crypto.clone().start_session(version, ¶ms);
634 let transport_config = server_config.transport.clone();
635 let mut conn = self.add_connection(
636 ch,
637 version,
638 dst_cid,
639 loc_cid,
640 src_cid,
641 incoming.network_path,
642 incoming.received_at,
643 tls,
644 transport_config,
645 SideArgs::Server {
646 server_config,
647 pref_addr_cid,
648 path_validated: remote_address_validated,
649 },
650 ¶ms,
651 );
652 self.index.insert_initial(dst_cid, ch);
653
654 match conn.handle_first_packet(
655 incoming.received_at,
656 incoming.network_path,
657 incoming.ecn,
658 packet_number,
659 incoming.packet,
660 incoming.rest,
661 ) {
662 Ok(()) => {
663 trace!(id = ch.0, icid = %dst_cid, "new connection");
664
665 for event in incoming_buffer.datagrams {
666 conn.handle_event(ConnectionEvent(ConnectionEventInner::Datagram(event)))
667 }
668
669 Ok((ch, conn))
670 }
671 Err(e) => {
672 debug!("handshake failed: {}", e);
673 self.handle_event(ch, EndpointEvent(EndpointEventInner::Drained));
674 let response = match e {
675 ConnectionError::TransportError(ref e) => Some(self.initial_close(
676 version,
677 incoming.network_path,
678 &incoming.crypto,
679 src_cid,
680 e.clone(),
681 buf,
682 )),
683 _ => None,
684 };
685 Err(Box::new(AcceptError { cause: e, response }))
686 }
687 }
688 }
689
690 fn early_validate_first_packet(
692 &mut self,
693 header: &ProtectedInitialHeader,
694 ) -> Result<(), TransportError> {
695 let config = &self.server_config.as_ref().unwrap();
696 if self.cids_exhausted() || self.incoming_buffers.len() >= config.max_incoming {
697 return Err(TransportError::CONNECTION_REFUSED(""));
698 }
699
700 if header.dst_cid.len() < 8
705 && (header.token_pos.is_empty()
706 || header.dst_cid.len() != self.local_cid_generator.cid_len())
707 {
708 debug!(
709 "rejecting connection due to invalid DCID length {}",
710 header.dst_cid.len()
711 );
712 return Err(TransportError::PROTOCOL_VIOLATION(
713 "invalid destination CID length",
714 ));
715 }
716
717 Ok(())
718 }
719
720 pub fn refuse(&mut self, incoming: Incoming, buf: &mut Vec<u8>) -> Transmit {
722 self.clean_up_incoming(&incoming);
723 incoming.improper_drop_warner.dismiss();
724
725 self.initial_close(
726 incoming.packet.header.version,
727 incoming.network_path,
728 &incoming.crypto,
729 incoming.packet.header.src_cid,
730 TransportError::CONNECTION_REFUSED(""),
731 buf,
732 )
733 }
734
735 pub fn retry(&mut self, incoming: Incoming, buf: &mut Vec<u8>) -> Result<Transmit, RetryError> {
739 if !incoming.may_retry() {
740 return Err(RetryError(Box::new(incoming)));
741 }
742
743 self.clean_up_incoming(&incoming);
744 incoming.improper_drop_warner.dismiss();
745
746 let server_config = self.server_config.as_ref().unwrap();
747
748 let loc_cid = self.local_cid_generator.generate_cid();
755
756 let payload = TokenPayload::Retry {
757 address: incoming.network_path.remote,
758 orig_dst_cid: incoming.packet.header.dst_cid,
759 issued: server_config.time_source.now(),
760 };
761 let token = Token::new(payload, &mut self.rng).encode(&*server_config.token_key);
762
763 let header = Header::Retry {
764 src_cid: loc_cid,
765 dst_cid: incoming.packet.header.src_cid,
766 version: incoming.packet.header.version,
767 };
768
769 let encode = header.encode(buf);
770 buf.put_slice(&token);
771 buf.extend_from_slice(&server_config.crypto.retry_tag(
772 incoming.packet.header.version,
773 incoming.packet.header.dst_cid,
774 buf,
775 ));
776 encode.finish(buf, &*incoming.crypto.header.local, None);
777
778 Ok(Transmit {
779 destination: incoming.network_path.remote,
780 ecn: None,
781 size: buf.len(),
782 segment_size: None,
783 src_ip: incoming.network_path.local_ip,
784 })
785 }
786
787 pub fn ignore(&mut self, incoming: Incoming) {
792 self.clean_up_incoming(&incoming);
793 incoming.improper_drop_warner.dismiss();
794 }
795
796 fn clean_up_incoming(&mut self, incoming: &Incoming) {
798 self.index.remove_initial(incoming.packet.header.dst_cid);
799 let incoming_buffer = self.incoming_buffers.remove(incoming.incoming_idx);
800 self.all_incoming_buffers_total_bytes -= incoming_buffer.total_bytes;
801 }
802
803 fn add_connection(
804 &mut self,
805 ch: ConnectionHandle,
806 version: u32,
807 init_cid: ConnectionId,
808 loc_cid: ConnectionId,
809 rem_cid: ConnectionId,
810 network_path: FourTuple,
811 now: Instant,
812 tls: Box<dyn crypto::Session>,
813 transport_config: Arc<TransportConfig>,
814 side_args: SideArgs,
815 params: &TransportParameters,
817 ) -> Connection {
818 let mut rng_seed = [0; 32];
819 self.rng.fill_bytes(&mut rng_seed);
820 let side = side_args.side();
821 let pref_addr_cid = side_args.pref_addr_cid();
822
823 let qlog =
824 transport_config.create_qlog_sink(side_args.side(), network_path.remote, init_cid, now);
825
826 qlog.emit_connection_started(
827 now,
828 loc_cid,
829 rem_cid,
830 network_path.remote,
831 network_path.local_ip,
832 params,
833 );
834
835 let conn = Connection::new(
836 self.config.clone(),
837 transport_config,
838 init_cid,
839 loc_cid,
840 rem_cid,
841 network_path,
842 tls,
843 self.local_cid_generator.as_ref(),
844 now,
845 version,
846 self.allow_mtud,
847 rng_seed,
848 side_args,
849 qlog,
850 );
851
852 let mut path_cids = PathLocalCids::default();
853 path_cids.cids.insert(path_cids.issued, loc_cid);
854 path_cids.issued += 1;
855
856 if let Some(cid) = pref_addr_cid {
857 debug_assert_eq!(path_cids.issued, 1, "preferred address cid seq must be 1");
858 path_cids.cids.insert(path_cids.issued, cid);
859 path_cids.issued += 1;
860 }
861
862 let id = self.connections.insert(ConnectionMeta {
863 init_cid,
864 loc_cids: FxHashMap::from_iter([(PathId::ZERO, path_cids)]),
865 network_path,
866 side,
867 reset_token: Default::default(),
868 });
869 debug_assert_eq!(id, ch.0, "connection handle allocation out of sync");
870
871 self.index.insert_conn(network_path, loc_cid, ch, side);
872
873 conn
874 }
875
876 fn initial_close(
877 &mut self,
878 version: u32,
879 network_path: FourTuple,
880 crypto: &Keys,
881 remote_id: ConnectionId,
882 reason: TransportError,
883 buf: &mut Vec<u8>,
884 ) -> Transmit {
885 let local_id = self.local_cid_generator.generate_cid();
889 let number = PacketNumber::U8(0);
890 let header = Header::Initial(InitialHeader {
891 dst_cid: remote_id,
892 src_cid: local_id,
893 number,
894 token: Bytes::new(),
895 version,
896 });
897
898 let partial_encode = header.encode(buf);
899 let max_len =
900 INITIAL_MTU as usize - partial_encode.header_len - crypto.packet.local.tag_len();
901 frame::Close::from(reason).encoder(max_len).encode(buf);
902 buf.resize(buf.len() + crypto.packet.local.tag_len(), 0);
903 partial_encode.finish(
904 buf,
905 &*crypto.header.local,
906 Some((0, Default::default(), &*crypto.packet.local)),
907 );
908 Transmit {
909 destination: network_path.remote,
910 ecn: None,
911 size: buf.len(),
912 segment_size: None,
913 src_ip: network_path.local_ip,
914 }
915 }
916
917 pub fn config(&self) -> &EndpointConfig {
919 &self.config
920 }
921
922 pub fn open_connections(&self) -> usize {
924 self.connections.len()
925 }
926
927 pub fn incoming_buffer_bytes(&self) -> u64 {
930 self.all_incoming_buffers_total_bytes
931 }
932
933 #[cfg(test)]
934 pub(crate) fn known_connections(&self) -> usize {
935 let x = self.connections.len();
936 debug_assert_eq!(x, self.index.connection_ids_initial.len());
937 debug_assert!(x >= self.index.connection_reset_tokens.0.len());
939 debug_assert!(x >= self.index.incoming_connection_remotes.len());
941 debug_assert!(x >= self.index.outgoing_connection_remotes.len());
942 x
943 }
944
945 #[cfg(test)]
946 pub(crate) fn known_cids(&self) -> usize {
947 self.index.connection_ids.len()
948 }
949
950 fn cids_exhausted(&self) -> bool {
955 self.local_cid_generator.cid_len() <= 4
956 && self.local_cid_generator.cid_len() != 0
957 && (2usize.pow(self.local_cid_generator.cid_len() as u32 * 8)
958 - self.index.connection_ids.len())
959 < 2usize.pow(self.local_cid_generator.cid_len() as u32 * 8 - 2)
960 }
961}
962
963impl fmt::Debug for Endpoint {
964 fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
965 fmt.debug_struct("Endpoint")
966 .field("rng", &self.rng)
967 .field("index", &self.index)
968 .field("connections", &self.connections)
969 .field("config", &self.config)
970 .field("server_config", &self.server_config)
971 .field("incoming_buffers.len", &self.incoming_buffers.len())
973 .field(
974 "all_incoming_buffers_total_bytes",
975 &self.all_incoming_buffers_total_bytes,
976 )
977 .finish()
978 }
979}
980
981#[derive(Default)]
983struct IncomingBuffer {
984 datagrams: Vec<DatagramConnectionEvent>,
985 total_bytes: u64,
986}
987
988#[derive(Copy, Clone, Debug)]
990enum RouteDatagramTo {
991 Incoming(usize),
992 Connection(ConnectionHandle, PathId),
993}
994
995#[derive(Default, Debug)]
997struct ConnectionIndex {
998 connection_ids_initial: HashMap<ConnectionId, RouteDatagramTo>,
1004 connection_ids: FxHashMap<ConnectionId, (ConnectionHandle, PathId)>,
1008 incoming_connection_remotes: HashMap<FourTuple, ConnectionHandle>,
1012 outgoing_connection_remotes: HashMap<SocketAddr, ConnectionHandle>,
1022 connection_reset_tokens: ResetTokenTable,
1027}
1028
1029impl ConnectionIndex {
1030 fn insert_initial_incoming(&mut self, dst_cid: ConnectionId, incoming_key: usize) {
1032 if dst_cid.is_empty() {
1033 return;
1034 }
1035 self.connection_ids_initial
1036 .insert(dst_cid, RouteDatagramTo::Incoming(incoming_key));
1037 }
1038
1039 fn remove_initial(&mut self, dst_cid: ConnectionId) {
1041 if dst_cid.is_empty() {
1042 return;
1043 }
1044 let removed = self.connection_ids_initial.remove(&dst_cid);
1045 debug_assert!(removed.is_some());
1046 }
1047
1048 fn insert_initial(&mut self, dst_cid: ConnectionId, connection: ConnectionHandle) {
1050 if dst_cid.is_empty() {
1051 return;
1052 }
1053 self.connection_ids_initial.insert(
1054 dst_cid,
1055 RouteDatagramTo::Connection(connection, PathId::ZERO),
1056 );
1057 }
1058
1059 fn insert_conn(
1062 &mut self,
1063 network_path: FourTuple,
1064 dst_cid: ConnectionId,
1065 connection: ConnectionHandle,
1066 side: Side,
1067 ) {
1068 match dst_cid.len() {
1069 0 => match side {
1070 Side::Server => {
1071 self.incoming_connection_remotes
1072 .insert(network_path, connection);
1073 }
1074 Side::Client => {
1075 self.outgoing_connection_remotes
1076 .insert(network_path.remote, connection);
1077 }
1078 },
1079 _ => {
1080 self.connection_ids
1081 .insert(dst_cid, (connection, PathId::ZERO));
1082 }
1083 }
1084 }
1085
1086 fn retire(&mut self, dst_cid: ConnectionId) {
1088 self.connection_ids.remove(&dst_cid);
1089 }
1090
1091 fn remove(&mut self, conn: &ConnectionMeta) {
1093 if conn.side.is_server() {
1094 self.remove_initial(conn.init_cid);
1095 }
1096 for cid in conn.loc_cids.values().flat_map(|pcids| pcids.cids.values()) {
1097 self.connection_ids.remove(cid);
1098 }
1099 self.incoming_connection_remotes.remove(&conn.network_path);
1100 self.outgoing_connection_remotes
1101 .remove(&conn.network_path.remote);
1102 for (remote, token) in conn.reset_token.values() {
1103 self.connection_reset_tokens.remove(*remote, *token);
1104 }
1105 }
1106
1107 fn get(&self, network_path: &FourTuple, datagram: &PartialDecode) -> Option<RouteDatagramTo> {
1109 if !datagram.dst_cid().is_empty() {
1110 if let Some(&(ch, path_id)) = self.connection_ids.get(&datagram.dst_cid()) {
1111 return Some(RouteDatagramTo::Connection(ch, path_id));
1112 }
1113 }
1114 if datagram.is_initial() || datagram.is_0rtt() {
1115 if let Some(&ch) = self.connection_ids_initial.get(&datagram.dst_cid()) {
1116 return Some(ch);
1117 }
1118 }
1119 if datagram.dst_cid().is_empty() {
1120 if let Some(&ch) = self.incoming_connection_remotes.get(network_path) {
1121 return Some(RouteDatagramTo::Connection(ch, PathId::ZERO));
1124 }
1125 if let Some(&ch) = self.outgoing_connection_remotes.get(&network_path.remote) {
1126 return Some(RouteDatagramTo::Connection(ch, PathId::ZERO));
1128 }
1129 }
1130 let data = datagram.data();
1131 if data.len() < RESET_TOKEN_SIZE {
1132 return None;
1133 }
1134 self.connection_reset_tokens
1137 .get(network_path.remote, &data[data.len() - RESET_TOKEN_SIZE..])
1138 .cloned()
1139 .map(|ch| RouteDatagramTo::Connection(ch, PathId::ZERO))
1140 }
1141}
1142
1143#[derive(Debug)]
1144pub(crate) struct ConnectionMeta {
1145 init_cid: ConnectionId,
1146 loc_cids: FxHashMap<PathId, PathLocalCids>,
1148 network_path: FourTuple,
1153 side: Side,
1154 reset_token: FxHashMap<PathId, (SocketAddr, ResetToken)>,
1164}
1165
1166#[derive(Debug, Default)]
1168struct PathLocalCids {
1169 issued: u64,
1173 cids: FxHashMap<u64, ConnectionId>,
1175}
1176
1177#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
1179pub struct ConnectionHandle(pub usize);
1180
1181impl From<ConnectionHandle> for usize {
1182 fn from(x: ConnectionHandle) -> Self {
1183 x.0
1184 }
1185}
1186
1187impl Index<ConnectionHandle> for Slab<ConnectionMeta> {
1188 type Output = ConnectionMeta;
1189 fn index(&self, ch: ConnectionHandle) -> &ConnectionMeta {
1190 &self[ch.0]
1191 }
1192}
1193
1194impl IndexMut<ConnectionHandle> for Slab<ConnectionMeta> {
1195 fn index_mut(&mut self, ch: ConnectionHandle) -> &mut ConnectionMeta {
1196 &mut self[ch.0]
1197 }
1198}
1199
1200pub enum DatagramEvent {
1202 ConnectionEvent(ConnectionHandle, ConnectionEvent),
1204 NewConnection(Incoming),
1206 Response(Transmit),
1208}
1209
1210#[derive(derive_more::Debug)]
1212pub struct Incoming {
1213 #[debug(skip)]
1214 received_at: Instant,
1215 network_path: FourTuple,
1216 ecn: Option<EcnCodepoint>,
1217 #[debug(skip)]
1218 packet: InitialPacket,
1219 #[debug(skip)]
1220 rest: Option<BytesMut>,
1221 #[debug(skip)]
1222 crypto: Keys,
1223 token: IncomingToken,
1224 incoming_idx: usize,
1225 #[debug(skip)]
1226 improper_drop_warner: IncomingImproperDropWarner,
1227}
1228
1229impl Incoming {
1230 pub fn local_ip(&self) -> Option<IpAddr> {
1234 self.network_path.local_ip
1235 }
1236
1237 pub fn remote_address(&self) -> SocketAddr {
1239 self.network_path.remote
1240 }
1241
1242 pub fn remote_address_validated(&self) -> bool {
1250 self.token.validated
1251 }
1252
1253 pub fn may_retry(&self) -> bool {
1258 self.token.retry_src_cid.is_none()
1259 }
1260
1261 pub fn orig_dst_cid(&self) -> ConnectionId {
1263 self.token.orig_dst_cid
1264 }
1265}
1266
1267struct IncomingImproperDropWarner;
1268
1269impl IncomingImproperDropWarner {
1270 fn dismiss(self) {
1271 mem::forget(self);
1272 }
1273}
1274
1275impl Drop for IncomingImproperDropWarner {
1276 fn drop(&mut self) {
1277 warn!(
1278 "quinn_proto::Incoming dropped without passing to Endpoint::accept/refuse/retry/ignore \
1279 (may cause memory leak and eventual inability to accept new connections)"
1280 );
1281 }
1282}
1283
1284#[derive(Debug, Error, Clone, PartialEq, Eq)]
1288pub enum ConnectError {
1289 #[error("endpoint stopping")]
1293 EndpointStopping,
1294 #[error("CIDs exhausted")]
1298 CidsExhausted,
1299 #[error("invalid server name: {0}")]
1301 InvalidServerName(String),
1302 #[error("invalid remote address: {0}")]
1306 InvalidRemoteAddress(SocketAddr),
1307 #[error("no default client config")]
1311 NoDefaultClientConfig,
1312 #[error("unsupported QUIC version")]
1314 UnsupportedVersion,
1315}
1316
1317#[derive(Debug)]
1319pub struct AcceptError {
1320 pub cause: ConnectionError,
1322 pub response: Option<Transmit>,
1324}
1325
1326#[derive(Debug, Error)]
1328#[error("retry() with validated Incoming")]
1329pub struct RetryError(Box<Incoming>);
1330
1331impl RetryError {
1332 pub fn into_incoming(self) -> Incoming {
1334 *self.0
1335 }
1336}
1337
1338#[derive(Default, Debug)]
1343struct ResetTokenTable(HashMap<SocketAddr, HashMap<ResetToken, ConnectionHandle>>);
1344
1345impl ResetTokenTable {
1346 fn insert(&mut self, remote: SocketAddr, token: ResetToken, ch: ConnectionHandle) -> bool {
1347 self.0
1348 .entry(remote)
1349 .or_default()
1350 .insert(token, ch)
1351 .is_some()
1352 }
1353
1354 fn remove(&mut self, remote: SocketAddr, token: ResetToken) {
1355 use std::collections::hash_map::Entry;
1356 match self.0.entry(remote) {
1357 Entry::Vacant(_) => {}
1358 Entry::Occupied(mut e) => {
1359 e.get_mut().remove(&token);
1360 if e.get().is_empty() {
1361 e.remove_entry();
1362 }
1363 }
1364 }
1365 }
1366
1367 fn get(&self, remote: SocketAddr, token: &[u8]) -> Option<&ConnectionHandle> {
1368 let token = ResetToken::from(<[u8; RESET_TOKEN_SIZE]>::try_from(token).ok()?);
1369 self.0.get(&remote)?.get(&token)
1370 }
1371}