retina_core/conntrack/
mod.rs

1//! Connection state management.
2//!
3//! Most of this module's functionality is maintained internally by Retina and is not meant to be
4//! directly managed by users. However, it publicly exposes some useful connection identifiers for
5//! convenience.
6
7pub(crate) mod conn;
8pub mod conn_id;
9pub(crate) mod pdu;
10mod timerwheel;
11
12use self::conn::conn_info::ConnState;
13use self::conn::{Conn, L4Conn};
14use self::conn_id::ConnId;
15use self::pdu::{L4Context, L4Pdu};
16use self::timerwheel::TimerWheel;
17use crate::config::ConnTrackConfig;
18use crate::memory::mbuf::Mbuf;
19use crate::protocols::packet::tcp::TCP_PROTOCOL;
20use crate::protocols::packet::udp::UDP_PROTOCOL;
21use crate::protocols::stream::ParserRegistry;
22use crate::subscription::{Subscription, Trackable};
23
24use std::cmp;
25use std::time::Instant;
26
27use anyhow::anyhow;
28use hashlink::linked_hash_map::{LinkedHashMap, RawEntryMut};
29
30/// Manages state for all TCP and UDP connections.
31///
32/// One `ConnTracker` is maintained per core. `ConnTracker` is not meant to be directly managed by
33/// users, but can be configured at runtime with a maximum capacity, out-of-order tolerance,
34/// different timeout values, and other options. See
35/// [ConnTrackConfig](crate::config::ConnTrackConfig) for details.
36pub struct ConnTracker<T>
37where
38    T: Trackable,
39{
40    /// Configuration
41    config: TrackerConfig,
42    /// Contains required protocol parsers for `T`.
43    registry: ParserRegistry,
44    /// Manages `ConnId` to `Conn<T>` mappings.
45    table: LinkedHashMap<ConnId, Conn<T>>,
46    /// Manages connection timeouts.
47    timerwheel: TimerWheel,
48}
49
50impl<T> ConnTracker<T>
51where
52    T: Trackable,
53{
54    /// Creates a new `ConnTracker`.
55    pub(crate) fn new(config: TrackerConfig, registry: ParserRegistry) -> Self {
56        let table = LinkedHashMap::with_capacity(config.max_connections);
57        let timerwheel = TimerWheel::new(
58            cmp::max(config.tcp_inactivity_timeout, config.udp_inactivity_timeout),
59            config.timeout_resolution,
60        );
61        ConnTracker {
62            config,
63            registry,
64            table,
65            timerwheel,
66        }
67    }
68
69    /// Returns the number of entries in the table.
70    #[inline]
71    pub(crate) fn size(&self) -> usize {
72        self.table.len()
73    }
74
75    /// Process a single incoming packet `mbuf` with layer-4 context `ctxt`.
76    pub(crate) fn process(
77        &mut self,
78        mbuf: Mbuf,
79        ctxt: L4Context,
80        subscription: &Subscription<T::Subscribed>,
81    ) {
82        let conn_id = ConnId::new(ctxt.src, ctxt.dst, ctxt.proto);
83        match self.table.raw_entry_mut().from_key(&conn_id) {
84            RawEntryMut::Occupied(mut occupied) => {
85                let conn = occupied.get_mut();
86                conn.last_seen_ts = Instant::now();
87                if conn.state() == ConnState::Dropped {
88                    // Allow connection to age out.
89                    // last_seen_ts is updated to avoid aging out long-lived UDP
90                    // connections prematurely
91                    return;
92                }
93                let dir = conn.packet_dir(&ctxt);
94                conn.inactivity_window = match &conn.l4conn {
95                    L4Conn::Tcp(_) => self.config.tcp_inactivity_timeout,
96                    L4Conn::Udp(_) => self.config.udp_inactivity_timeout,
97                };
98                if conn.state() == ConnState::Remove {
99                    log::error!("Conn in Remove state when occupied in table");
100                }
101                let pdu = L4Pdu::new(mbuf, ctxt, dir);
102                conn.update(pdu, subscription, &self.registry);
103                if conn.state() == ConnState::Remove {
104                    occupied.remove();
105                    return;
106                }
107
108                if conn.terminated() {
109                    conn.terminate(subscription);
110                    occupied.remove();
111                }
112            }
113            RawEntryMut::Vacant(_) => {
114                if self.size() < self.config.max_connections {
115                    let conn = match ctxt.proto {
116                        TCP_PROTOCOL => Conn::new_tcp(
117                            ctxt,
118                            self.config.tcp_establish_timeout,
119                            self.config.max_out_of_order,
120                        ),
121                        UDP_PROTOCOL => Conn::new_udp(ctxt, self.config.udp_inactivity_timeout),
122                        _ => Err(anyhow!("Invalid L4 Protocol")),
123                    };
124                    if let Ok(mut conn) = conn {
125                        let pdu = L4Pdu::new(mbuf, ctxt, true);
126                        conn.info.consume_pdu(pdu, subscription, &self.registry);
127                        if conn.state() != ConnState::Remove {
128                            self.timerwheel.insert(
129                                &conn_id,
130                                conn.last_seen_ts,
131                                conn.inactivity_window,
132                            );
133                            self.table.insert(conn_id, conn);
134                        }
135                    }
136                } else {
137                    log::error!("Table full. Dropping packet.");
138                }
139            }
140        }
141    }
142
143    /// Drains any remaining connections that satisfy the filter on runtime termination.
144    pub(crate) fn drain(&mut self, subscription: &Subscription<T::Subscribed>) {
145        log::info!("Draining Connection table");
146        for (_, mut conn) in self.table.drain() {
147            conn.terminate(subscription);
148        }
149    }
150
151    /// Checks for and removes inactive connections.
152    pub(crate) fn check_inactive(&mut self, subscription: &Subscription<T::Subscribed>) {
153        self.timerwheel
154            .check_inactive(&mut self.table, subscription);
155    }
156}
157
158/// Configurable options for a `ConnTracker`.
159#[derive(Debug)]
160pub(crate) struct TrackerConfig {
161    /// Maximum number of connections that can be tracked per-core.
162    pub(super) max_connections: usize,
163    /// Maximum number of out-of-order packets allowed per TCP connection.
164    pub(super) max_out_of_order: usize,
165    /// Time to expire inactive UDP connections (in milliseconds).
166    pub(super) udp_inactivity_timeout: usize,
167    /// Time to expire inactive TCP connections (in milliseconds).
168    pub(super) tcp_inactivity_timeout: usize,
169    /// Time to expire unestablished TCP connections (in milliseconds).
170    pub(super) tcp_establish_timeout: usize,
171    /// Frequency to check for inactive streams (in milliseconds).
172    pub(super) timeout_resolution: usize,
173}
174
175impl From<&ConnTrackConfig> for TrackerConfig {
176    fn from(config: &ConnTrackConfig) -> Self {
177        TrackerConfig {
178            max_connections: config.max_connections,
179            max_out_of_order: config.max_out_of_order,
180            udp_inactivity_timeout: config.udp_inactivity_timeout,
181            tcp_inactivity_timeout: config.tcp_inactivity_timeout,
182            tcp_establish_timeout: config.tcp_establish_timeout,
183            timeout_resolution: config.timeout_resolution,
184        }
185    }
186}
retina_core/conntrack/mod.rs

retina_core/conntrack/
mod.rs
