rtc_sctp/endpoint/
mod.rs

1#[cfg(test)]
2mod endpoint_test;
3
4use std::{
5    collections::HashMap,
6    fmt, iter,
7    net::SocketAddr,
8    ops::{Index, IndexMut},
9    sync::Arc,
10    time::Instant,
11};
12
13use rustc_hash::FxHashMap;
14
15use crate::Payload;
16use crate::association::Association;
17use crate::chunk::chunk_type::CT_INIT;
18use crate::config::{ClientConfig, EndpointConfig, ServerConfig, TransportConfig};
19use crate::packet::PartialDecode;
20use crate::shared::{
21    AssociationEvent, AssociationEventInner, AssociationId, EndpointEvent, EndpointEventInner,
22};
23use crate::util::{AssociationIdGenerator, RandomAssociationIdGenerator};
24use shared::{EcnCodepoint, TransportContext, TransportMessage, TransportProtocol};
25
26use bytes::Bytes;
27use log::{debug, trace, warn};
28use slab::Slab;
29use thiserror::Error;
30
31/// The main entry point to the library
32///
33/// This object performs no I/O whatsoever. Instead, it generates a stream of packets to send via
34/// `poll_transmit`, and consumes incoming packets and association-generated events via `handle` and
35/// `handle_event`.
36pub struct Endpoint {
37    local_addr: SocketAddr,
38    transport_protocol: TransportProtocol,
39    /// Identifies associations based on the INIT Dst AID the peer utilized
40    ///
41    /// Uses a standard `HashMap` to protect against hash collision attacks:
42    /// keys are remote-chosen initiate-tags.
43    association_ids_init: HashMap<AssociationId, AssociationHandle>,
44    /// Identifies associations based on locally created CIDs
45    ///
46    /// Uses a cheaper hash function since keys are locally created
47    association_ids: FxHashMap<AssociationId, AssociationHandle>,
48
49    associations: Slab<AssociationMeta>,
50    local_cid_generator: Box<dyn AssociationIdGenerator>,
51    endpoint_config: Arc<EndpointConfig>,
52    server_config: Option<Arc<ServerConfig>>,
53    /// Whether incoming associations should be unconditionally rejected by a server
54    ///
55    /// Equivalent to a `ServerConfig.accept_buffer` of `0`, but can be changed after the endpoint is constructed.
56    reject_new_associations: bool,
57}
58
59impl fmt::Debug for Endpoint {
60    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
61        fmt.debug_struct("Endpoint<T>")
62            .field("association_ids_initial", &self.association_ids_init)
63            .field("association_ids", &self.association_ids)
64            .field("associations", &self.associations)
65            .field("config", &self.endpoint_config)
66            .field("server_config", &self.server_config)
67            .field("reject_new_associations", &self.reject_new_associations)
68            .finish()
69    }
70}
71
72impl Endpoint {
73    /// Create a new endpoint
74    ///
75    /// Returns `Err` if the configuration is invalid.
76    pub fn new(
77        local_addr: SocketAddr,
78        transport_protocol: TransportProtocol,
79        endpoint_config: Arc<EndpointConfig>,
80        server_config: Option<Arc<ServerConfig>>,
81    ) -> Self {
82        Self {
83            local_addr,
84            transport_protocol,
85            association_ids_init: HashMap::default(),
86            association_ids: FxHashMap::default(),
87            associations: Slab::new(),
88            local_cid_generator: (endpoint_config.aid_generator_factory.as_ref())(),
89            reject_new_associations: false,
90            endpoint_config,
91            server_config,
92        }
93    }
94
95    /// Replace the server configuration, affecting new incoming associations only
96    pub fn set_server_config(&mut self, server_config: Option<Arc<ServerConfig>>) {
97        self.server_config = server_config;
98    }
99
100    /// Process `EndpointEvent`s emitted from related `Association`s
101    pub fn handle_event(&mut self, ch: AssociationHandle, event: EndpointEvent) {
102        match event.0 {
103            EndpointEventInner::Drained => {
104                let conn = self.associations.remove(ch.0);
105                self.association_ids_init.remove(&conn.init_cid);
106                for cid in conn.loc_cids.values() {
107                    self.association_ids.remove(cid);
108                }
109            }
110        }
111    }
112
113    /// Process an incoming UDP datagram
114    pub fn handle(
115        &mut self,
116        now: Instant,
117        remote: SocketAddr,
118        ecn: Option<EcnCodepoint>,
119        data: Bytes,
120    ) -> Option<(AssociationHandle, DatagramEvent)> {
121        let partial_decode = match PartialDecode::unmarshal(&data) {
122            Ok(x) => x,
123            Err(err) => {
124                trace!("malformed header: {}", err);
125                return None;
126            }
127        };
128
129        //
130        // Handle packet on existing association, if any
131        //
132        let dst_cid = partial_decode.common_header.verification_tag;
133        let known_ch = if dst_cid > 0 {
134            self.association_ids.get(&dst_cid).cloned()
135        } else {
136            //TODO: improve INIT handling for DoS attack
137            if partial_decode.first_chunk_type == CT_INIT {
138                if let Some(dst_cid) = partial_decode.initiate_tag {
139                    self.association_ids.get(&dst_cid).cloned()
140                } else {
141                    None
142                }
143            } else {
144                None
145            }
146        };
147
148        if let Some(ch) = known_ch {
149            return Some((
150                ch,
151                DatagramEvent::AssociationEvent(AssociationEvent(AssociationEventInner::Datagram(
152                    TransportMessage {
153                        now,
154                        transport: TransportContext {
155                            local_addr: self.local_addr,
156                            peer_addr: remote,
157                            ecn,
158                            transport_protocol: self.transport_protocol,
159                        },
160                        message: Payload::PartialDecode(partial_decode),
161                    },
162                ))),
163            ));
164        }
165
166        //
167        // Potentially create a new association
168        //
169        self.handle_first_packet(now, remote, ecn, partial_decode)
170            .map(|(ch, a)| (ch, DatagramEvent::NewAssociation(a)))
171    }
172
173    /// Initiate an Association
174    pub fn connect(
175        &mut self,
176        config: ClientConfig,
177        remote: SocketAddr,
178    ) -> Result<(AssociationHandle, Association), ConnectError> {
179        if self.is_full() {
180            return Err(ConnectError::TooManyAssociations);
181        }
182        if remote.port() == 0 {
183            return Err(ConnectError::InvalidRemoteAddress(remote));
184        }
185
186        let remote_aid = RandomAssociationIdGenerator::new().generate_aid();
187        let local_aid = self.new_aid();
188
189        let (ch, conn) = self.add_association(
190            remote_aid,
191            local_aid,
192            remote,
193            Instant::now(),
194            None,
195            config.transport,
196        );
197        Ok((ch, conn))
198    }
199
200    fn new_aid(&mut self) -> AssociationId {
201        loop {
202            let aid = self.local_cid_generator.generate_aid();
203            if !self.association_ids.contains_key(&aid) {
204                break aid;
205            }
206        }
207    }
208
209    fn handle_first_packet(
210        &mut self,
211        now: Instant,
212        remote: SocketAddr,
213        ecn: Option<EcnCodepoint>,
214        partial_decode: PartialDecode,
215    ) -> Option<(AssociationHandle, Association)> {
216        if partial_decode.first_chunk_type != CT_INIT
217            || (partial_decode.first_chunk_type == CT_INIT && partial_decode.initiate_tag.is_none())
218        {
219            debug!("refusing first packet with Non-INIT or empty initial_tag INIT");
220            return None;
221        }
222
223        let server_config = if let Some(server_config) = self.server_config.as_ref() {
224            server_config
225        } else {
226            warn!("refusing first packet due to empty server_config");
227            return None;
228        };
229
230        if self.associations.len() >= server_config.concurrent_associations as usize
231            || self.reject_new_associations
232            || self.is_full()
233        {
234            debug!("refusing association");
235            //TODO: self.initial_close();
236            return None;
237        }
238
239        let server_config = server_config.clone();
240        let transport_config = server_config.transport.clone();
241
242        let remote_aid = *partial_decode.initiate_tag.as_ref().unwrap();
243        let local_aid = self.new_aid();
244
245        let (ch, mut conn) = self.add_association(
246            remote_aid,
247            local_aid,
248            remote,
249            now,
250            Some(server_config),
251            transport_config,
252        );
253
254        conn.handle_event(AssociationEvent(AssociationEventInner::Datagram(
255            TransportMessage {
256                now,
257                transport: TransportContext {
258                    local_addr: self.local_addr,
259                    peer_addr: remote,
260                    ecn,
261                    transport_protocol: self.transport_protocol,
262                },
263                message: Payload::PartialDecode(partial_decode),
264            },
265        )));
266
267        Some((ch, conn))
268    }
269
270    #[allow(clippy::too_many_arguments)]
271    fn add_association(
272        &mut self,
273        remote_aid: AssociationId,
274        local_aid: AssociationId,
275        remote_addr: SocketAddr,
276        now: Instant,
277        server_config: Option<Arc<ServerConfig>>,
278        transport_config: Arc<TransportConfig>,
279    ) -> (AssociationHandle, Association) {
280        let conn = Association::new(
281            server_config,
282            transport_config,
283            self.endpoint_config.get_max_payload_size(),
284            local_aid,
285            remote_addr,
286            self.local_addr,
287            self.transport_protocol,
288            now,
289        );
290
291        let id = self.associations.insert(AssociationMeta {
292            init_cid: remote_aid,
293            cids_issued: 0,
294            loc_cids: iter::once((0, local_aid)).collect(),
295            initial_remote: remote_addr,
296        });
297
298        let ch = AssociationHandle(id);
299        self.association_ids.insert(local_aid, ch);
300
301        (ch, conn)
302    }
303
304    /// Unconditionally reject future incoming associations
305    pub fn reject_new_associations(&mut self) {
306        self.reject_new_associations = true;
307    }
308
309    /// Access the configuration used by this endpoint
310    pub fn endpoint_config(&self) -> &EndpointConfig {
311        &self.endpoint_config
312    }
313
314    /// Whether we've used up 3/4 of the available AID space
315    fn is_full(&self) -> bool {
316        (((u32::MAX >> 1) + (u32::MAX >> 2)) as usize) < self.association_ids.len()
317    }
318}
319
320#[derive(Debug)]
321pub(crate) struct AssociationMeta {
322    init_cid: AssociationId,
323    /// Number of local association IDs.
324    cids_issued: u64,
325    loc_cids: HashMap<u64, AssociationId>,
326    /// Remote address the association began with
327    ///
328    /// Only needed to support associations with zero-length AIDs, which cannot migrate, so we don't
329    /// bother keeping it up to date.
330    initial_remote: SocketAddr,
331}
332
333/// Internal identifier for an `Association` currently associated with an endpoint
334#[derive(Default, Debug, Copy, Clone, Eq, PartialEq, Hash, Ord, PartialOrd)]
335pub struct AssociationHandle(pub usize);
336
337impl From<AssociationHandle> for usize {
338    fn from(x: AssociationHandle) -> usize {
339        x.0
340    }
341}
342
343impl Index<AssociationHandle> for Slab<AssociationMeta> {
344    type Output = AssociationMeta;
345    fn index(&self, ch: AssociationHandle) -> &AssociationMeta {
346        &self[ch.0]
347    }
348}
349
350impl IndexMut<AssociationHandle> for Slab<AssociationMeta> {
351    fn index_mut(&mut self, ch: AssociationHandle) -> &mut AssociationMeta {
352        &mut self[ch.0]
353    }
354}
355
356/// Event resulting from processing a single datagram
357#[allow(clippy::large_enum_variant)] // Not passed around extensively
358pub enum DatagramEvent {
359    /// The datagram is redirected to its `Association`
360    AssociationEvent(AssociationEvent),
361    /// The datagram has resulted in starting a new `Association`
362    NewAssociation(Association),
363}
364
365/// Errors in the parameters being used to create a new association
366///
367/// These arise before any I/O has been performed.
368#[derive(Debug, Error, Clone, PartialEq, Eq)]
369pub enum ConnectError {
370    /// The endpoint can no longer create new associations
371    ///
372    /// Indicates that a necessary component of the endpoint has been dropped or otherwise disabled.
373    #[error("endpoint stopping")]
374    EndpointStopping,
375    /// The number of active associations on the local endpoint is at the limit
376    ///
377    /// Try using longer association IDs.
378    #[error("too many associations")]
379    TooManyAssociations,
380    /// The domain name supplied was malformed
381    #[error("invalid DNS name: {0}")]
382    InvalidDnsName(String),
383    /// The remote [`SocketAddr`] supplied was malformed
384    ///
385    /// Examples include attempting to connect to port 0, or using an inappropriate address family.
386    #[error("invalid remote address: {0}")]
387    InvalidRemoteAddress(SocketAddr),
388    /// No default client configuration was set up
389    ///
390    /// Use `Endpoint::connect_with` to specify a client configuration.
391    #[error("no default client config")]
392    NoDefaultClientConfig,
393}
rtc_sctp/endpoint/mod.rs

rtc_sctp/endpoint/
mod.rs
