tokio_quiche/quic/router/
acceptor.rs

1// Copyright (C) 2025, Cloudflare, Inc.
2// All rights reserved.
3//
4// Redistribution and use in source and binary forms, with or without
5// modification, are permitted provided that the following conditions are
6// met:
7//
8//     * Redistributions of source code must retain the above copyright notice,
9//       this list of conditions and the following disclaimer.
10//
11//     * Redistributions in binary form must reproduce the above copyright
12//       notice, this list of conditions and the following disclaimer in the
13//       documentation and/or other materials provided with the distribution.
14//
15// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
16// IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
17// THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
18// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
19// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
20// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
21// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
22// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
23// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
24// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
25// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
26
27use std::fs::File;
28use std::io;
29use std::sync::Arc;
30use std::time::Instant;
31
32use datagram_socket::DatagramSocketSend;
33use datagram_socket::DatagramSocketSendExt;
34use datagram_socket::MAX_DATAGRAM_SIZE;
35use qlog::writer::make_qlog_writer_from_path;
36use qlog::writer::qlog_file_name;
37use qlog::writer::QlogCompression;
38use quiche::ConnectionId;
39use quiche::Header;
40use quiche::RetryConnectionIds;
41use quiche::Type as PacketType;
42use task_killswitch::spawn_with_killswitch;
43
44use crate::metrics::labels;
45use crate::metrics::Metrics;
46use crate::quic::addr_validation_token::AddrValidationTokenManager;
47use crate::quic::connection::SharedConnectionIdGenerator;
48use crate::quic::router::NewConnection;
49use crate::quic::Incoming;
50use crate::QuicResultExt;
51
52use super::InitialPacketHandler;
53
54/// A [`ConnectionAcceptor`] is an [`InitialPacketHandler`] that acts as a
55/// server and accepts quic connections.
56pub(crate) struct ConnectionAcceptor<S, M> {
57    config: ConnectionAcceptorConfig,
58    socket: Arc<S>,
59    token_manager: AddrValidationTokenManager,
60    cid_generator: SharedConnectionIdGenerator,
61    metrics: M,
62}
63
64pub(crate) struct ConnectionAcceptorConfig {
65    pub(crate) disable_client_ip_validation: bool,
66    pub(crate) qlog_dir: Option<String>,
67    pub(crate) qlog_compression: QlogCompression,
68    pub(crate) keylog_file: Option<File>,
69    #[cfg(target_os = "linux")]
70    pub(crate) with_pktinfo: bool,
71}
72
73impl<S, M> ConnectionAcceptor<S, M>
74where
75    S: DatagramSocketSend + Send + 'static,
76    M: Metrics,
77{
78    pub(crate) fn new(
79        config: ConnectionAcceptorConfig, socket: Arc<S>,
80        token_manager: AddrValidationTokenManager,
81        cid_generator: SharedConnectionIdGenerator, metrics: M,
82    ) -> Self {
83        Self {
84            config,
85            socket,
86            token_manager,
87            cid_generator,
88            metrics,
89        }
90    }
91
92    fn accept_conn(
93        &mut self, incoming: Incoming, retry_cids: Option<RetryConnectionIds>,
94        pending_cid: ConnectionId<'static>, quiche_config: &mut quiche::Config,
95    ) -> io::Result<Option<NewConnection>> {
96        let handshake_start_time = Instant::now();
97        let scid = self.cid_generator.new_connection_id();
98
99        let mut conn = if let Some(retry_cids) = retry_cids {
100            quiche::accept_with_retry(
101                &scid,
102                retry_cids,
103                incoming.local_addr,
104                incoming.peer_addr,
105                quiche_config,
106            )
107        } else {
108            quiche::accept_with_buf_factory(
109                &scid,
110                None,
111                incoming.local_addr,
112                incoming.peer_addr,
113                quiche_config,
114            )
115        }
116        .into_io()?;
117
118        if let Some(qlog_dir) = &self.config.qlog_dir {
119            let id = format!("{:?}", scid);
120            let path = std::path::Path::new(qlog_dir)
121                .join(qlog_file_name(&id, self.config.qlog_compression));
122            if let Ok(writer) =
123                make_qlog_writer_from_path(&path, self.config.qlog_compression)
124            {
125                conn.set_qlog(
126                    writer,
127                    "tokio-quiche qlog".to_string(),
128                    format!("tokio-quiche qlog id={id}"),
129                );
130            }
131        }
132
133        if let Some(keylog_file) = &self.config.keylog_file {
134            if let Ok(keylog_clone) = keylog_file.try_clone() {
135                conn.set_keylog(Box::new(keylog_clone));
136            }
137        }
138
139        Ok(Some(NewConnection {
140            conn: Box::new(conn),
141            handshake_start_time,
142            pending_cid: Some(pending_cid),
143            cid_generator: Some(Arc::clone(&self.cid_generator)),
144            initial_pkt: Some(incoming),
145        }))
146    }
147
148    fn handshake_reply(
149        &self, incoming: Incoming,
150        writer: impl FnOnce(&mut [u8]) -> io::Result<usize>,
151    ) -> io::Result<Option<NewConnection>> {
152        let mut send_buf = [0u8; MAX_DATAGRAM_SIZE];
153        let written = writer(&mut send_buf)?;
154        let socket = Arc::clone(&self.socket);
155        #[cfg(target_os = "linux")]
156        let with_pktinfo = self.config.with_pktinfo;
157        #[cfg(target_os = "linux")]
158        let would_block_metric = self
159            .metrics
160            .write_errors(labels::QuicWriteError::WouldBlock);
161        #[cfg(target_os = "linux")]
162        let send_to_wouldblock_duration_s =
163            self.metrics.send_to_wouldblock_duration_s();
164
165        spawn_with_killswitch(async move {
166            let send_buf = &send_buf[..written];
167            let to = incoming.peer_addr;
168
169            #[allow(unused_variables)]
170            let Some(udp) = socket.as_udp_socket() else {
171                let _ = socket.send_to(send_buf, to).await;
172                return;
173            };
174
175            #[cfg(target_os = "linux")]
176            {
177                let from = with_pktinfo.then_some(incoming.local_addr);
178                let _ = crate::quic::io::gso::send_to(
179                    udp,
180                    to,
181                    from,
182                    send_buf,
183                    send_buf.len(),
184                    None,
185                    would_block_metric,
186                    send_to_wouldblock_duration_s,
187                )
188                .await;
189            }
190
191            #[cfg(not(target_os = "linux"))]
192            let _ = socket.send_to(send_buf, to).await;
193        });
194
195        Ok(None)
196    }
197
198    fn stateless_retry(
199        &mut self, incoming: Incoming, hdr: Header,
200    ) -> io::Result<Option<NewConnection>> {
201        let scid = self.cid_generator.new_connection_id();
202
203        let token = self.token_manager.gen(&hdr.dcid, incoming.peer_addr);
204
205        self.handshake_reply(incoming, move |buf| {
206            quiche::retry(&hdr.scid, &hdr.dcid, &scid, &token, hdr.version, buf)
207                .into_io()
208        })
209    }
210}
211
212impl<S, M> InitialPacketHandler for ConnectionAcceptor<S, M>
213where
214    S: DatagramSocketSend + Send + 'static,
215    M: Metrics,
216{
217    fn handle_initials(
218        &mut self, incoming: Incoming, hdr: quiche::Header<'static>,
219        quiche_config: &mut quiche::Config,
220    ) -> io::Result<Option<NewConnection>> {
221        if hdr.ty != PacketType::Initial {
222            // Non-initial packets should have a valid CID, but we want to have
223            // some telemetry if this isn't the case.
224            if let Err(e) = self.cid_generator.verify_connection_id(&hdr.dcid) {
225                self.metrics.invalid_cid_packet_count(e).inc();
226            }
227
228            Err(labels::QuicInvalidInitialPacketError::WrongType(hdr.ty))?;
229        }
230
231        if !quiche::version_is_supported(hdr.version) {
232            return self.handshake_reply(incoming, |buf| {
233                quiche::negotiate_version(&hdr.scid, &hdr.dcid, buf).into_io()
234            });
235        }
236
237        if self.config.disable_client_ip_validation {
238            return self.accept_conn(incoming, None, hdr.dcid, quiche_config);
239        }
240
241        // NOTE: token is always present in Initial packets
242        let token = hdr.token.as_ref().unwrap();
243        if token.is_empty() {
244            return self.stateless_retry(incoming, hdr);
245        }
246
247        let original_dcid = self
248            .token_manager
249            .validate_and_extract_original_dcid(token, incoming.peer_addr)
250            .or(Err(
251                labels::QuicInvalidInitialPacketError::TokenValidationFail,
252            ))?;
253
254        let retry_cids = Some(RetryConnectionIds {
255            original_destination_cid: &original_dcid,
256            retry_source_cid: &hdr.dcid,
257        });
258
259        self.accept_conn(incoming, retry_cids, hdr.dcid.clone(), quiche_config)
260    }
261}
tokio_quiche/quic/router/acceptor.rs

tokio_quiche/quic/router/
acceptor.rs
