iris_core/conntrack/conn/
conn_info.rs

1#![doc(hidden)]
2/// Per-connection data structure for subscription management.
3use super::conn_actions::TrackedActions;
4use crate::lcore::CoreId;
5use crate::protocols::packet::tcp::TCP_PROTOCOL;
6use crate::protocols::stream::{ConnData, ParserRegistry};
7use crate::subscription::{Subscription, Trackable};
8use crate::FiveTuple;
9use crate::L4Pdu;
10
11use super::{conn_layers::*, conn_state::*};
12
13/// Per-connection. Tracks all subscription-requested
14/// datatypes (`tracked` data). Maintains the State of the connection
15/// at each layer, including the `Actions` to execute when new
16/// packets are received.
17/// This must be public in order to be accessible by generated filter
18/// and update code.
19#[derive(Debug)]
20pub struct ConnInfo<T>
21where
22    T: Trackable,
23{
24    /// Actions and state from the perspective of L4 (TCP or UDP).
25    /// Valid states are Payload or None.
26    pub linfo: LayerInfo,
27    /// Connection five-tuple for filtering and determining directionality
28    /// of future packets.
29    pub cdata: ConnData,
30    /// Additional Layers that the L4 conn. should pass
31    /// data to.
32    pub layers: [Layer; NUM_LAYERS],
33    /// Subscription data (for delivering)
34    pub tracked: T,
35}
36
37impl<T> ConnInfo<T>
38where
39    T: Trackable,
40{
41    pub(super) fn new(pdu: &L4Pdu, core_id: CoreId) -> Self {
42        let five_tuple = FiveTuple::from_ctxt(&pdu.ctxt);
43        ConnInfo {
44            linfo: LayerInfo {
45                state: if pdu.ctxt.proto == TCP_PROTOCOL {
46                    LayerState::Headers // Pre-TCP handshake
47                } else {
48                    LayerState::Payload
49                },
50                actions: TrackedActions::new(),
51            },
52            cdata: ConnData::new(five_tuple),
53            layers: [Layer::L7(L7Session::new())],
54            tracked: T::new(pdu, core_id),
55        }
56    }
57
58    /// Initializes actions at all layers when first packet
59    /// in L4 connection is observed.
60    pub(crate) fn filter_first_packet(
61        &mut self,
62        subscription: &Subscription<T::Subscribed>,
63        pdu: &L4Pdu,
64    ) {
65        subscription.state_tx::<T>(self, &StateTransition::L4FirstPacket, Some(pdu));
66    }
67
68    /// Update tracked data when new packet is observed.
69    /// This is invoked up to twice:
70    /// - InL4Conn (pre-reassembly, if applicable) by `conn`
71    /// - InL4Stream (post-reassembly, if applicable) by `consume_stream`
72    pub(crate) fn new_packet(&mut self, pdu: &L4Pdu, subscription: &Subscription<T::Subscribed>) {
73        #[cfg(debug_assertions)]
74        {
75            log::debug!(
76                "New packet for conn {:?}, state: {:?}, L4 actions: {:?}",
77                self.cdata.five_tuple,
78                self.linfo.state,
79                self.linfo.actions.active
80            );
81        }
82
83        let mut needs_update = self.linfo.actions.needs_update();
84        let tx = if pdu.ctxt.reassembled {
85            needs_update = self.linfo.actions.needs_parse();
86            StateTransition::InL4Stream
87        } else {
88            StateTransition::InL4Conn
89        };
90        if needs_update && subscription.update(self, pdu, tx) {
91            self.exec_state_tx(tx, subscription);
92        }
93    }
94
95    /// Invoked by reassembly infrastructure when the TCP handshake is completed.
96    pub(super) fn handshake_done(&mut self, subscription: &Subscription<T::Subscribed>) {
97        self.linfo.state = LayerState::Payload;
98        self.exec_state_tx(StateTransition::L4EndHshk, subscription);
99    }
100
101    /// Invoked by transport layer to update data for encapsulated layers.
102    /// This is invoked in reassembled order for TCP and received order for UDP.
103    pub(crate) fn consume_stream(
104        &mut self,
105        pdu: &mut L4Pdu,
106        subscription: &Subscription<T::Subscribed>,
107        registry: &ParserRegistry,
108    ) {
109        // Pass to next layer(s) if applicable for parsing
110        if self.layers[0].needs_stream() {
111            let tx = self.layers[0].process_stream(pdu, registry);
112            self.exec_state_tx(tx, subscription);
113            if self.layers[0].needs_process(tx, pdu) {
114                let tx = self.layers[0].process_stream(pdu, registry);
115                self.exec_state_tx(tx, subscription);
116            }
117        }
118
119        // Update tracked data post-reassembly if needed
120        self.new_packet(pdu, subscription);
121    }
122
123    /// Drop the connection, e.g. due to timeout
124    pub(crate) fn exec_drop(&mut self) {
125        self.linfo.state = LayerState::None
126    }
127
128    /// Returns true if the connection should be dropped
129    pub(crate) fn drop(&self) -> bool {
130        self.linfo.state == LayerState::None
131    }
132
133    /// Invoked when the connection has terminated (by timeout or TCP FIN/ACK sequence)
134    /// Delivers any "end of connection" data.
135    pub(crate) fn handle_terminate(&mut self, subscription: &Subscription<T::Subscribed>) {
136        while let Some(tx) = self.layers[0].handle_terminate() {
137            self.exec_state_tx(tx, subscription);
138            if self.drop() {
139                break;
140            }
141        }
142        if !self.drop() {
143            self.exec_state_tx(StateTransition::L4Terminated, subscription);
144        }
145    }
146
147    /// Update subscription data and current state, including actions,
148    /// upon state transition.
149    fn exec_state_tx(&mut self, tx: StateTransition, subscription: &Subscription<T::Subscribed>) {
150        #[cfg(debug_assertions)]
151        {
152            log::debug!("State transition {:?} for conn {:?}, state: {:?}, L4 actions: {:?}, L7 actions: {:?}",
153                         tx, self.cdata.five_tuple, self.linfo.state, self.linfo.actions.active, self.layers[0].layer_info().actions.active);
154        }
155        // Packet is "no-op"; FirstPacket is handled separately
156        if matches!(tx, StateTransition::Packet | StateTransition::L4FirstPacket) {
157            return;
158        }
159
160        // Nothing to do at all layers
161        if self.linfo.actions.skip_tx(&tx)
162            && self
163                .layers
164                .iter()
165                .all(|l| l.layer_info().actions.skip_tx(&tx))
166        {
167            return;
168        }
169        self.linfo.actions.start_state_tx(tx);
170        for layer in self.layers.iter_mut() {
171            layer.layer_info_mut().actions.start_state_tx(tx);
172        }
173        subscription.state_tx::<T>(self, &tx, None);
174        for layer in &mut self.layers {
175            layer.end_state_tx();
176        }
177        if self.linfo.drop() && self.layers.iter().all(|l| l.drop()) {
178            self.exec_drop();
179        } else {
180            if self.layers.iter().any(|l| !l.drop()) {
181                self.linfo.actions.set_next_layer();
182            }
183        }
184    }
185
186    pub(crate) fn clear(&mut self) {
187        self.tracked.clear();
188    }
189
190    pub(crate) fn needs_reassembly(&self) -> bool {
191        self.linfo.actions.needs_parse() || self.layers.iter().any(|l| l.needs_stream())
192    }
193
194    pub(crate) fn needs_update(&self) -> bool {
195        self.linfo.actions.needs_update()
196    }
197}
iris_core/conntrack/conn/conn_info.rs

iris_core/conntrack/conn/
conn_info.rs
