iris_core/conntrack/conn/
conn_layers.rs

1#![doc(hidden)]
2/// Additional traffic layers built on top of the L4 base transport layer.
3use super::conn_actions::TrackedActions;
4use super::conn_state::{LayerState, StateTransition};
5use crate::conntrack::Actions;
6use crate::protocols::stream::{
7    ConnParser, ParseResult, ParserRegistry, ParsingState, ProbeRegistryResult, SessionData,
8    SessionProto,
9};
10use crate::protocols::Session;
11use crate::L4Pdu;
12
13lazy_static! {
14    static ref DEFAULT_SESSION: Session = Session {
15        data: SessionData::Null,
16        id: 0,
17    };
18}
19
20/// "Layers" that can be built on top of the transport layer (L4).
21/// Each associated datatype must implement LayerInfo API (see below)
22#[derive(Debug)]
23pub enum Layer {
24    /// L6/L7 Session
25    L7(L7Session),
26}
27pub const NUM_LAYERS: usize = 1;
28
29/// Convenience enum to be used at compile-time.
30/// Should correspond to the transport layer plus `Layer` variants.
31#[derive(PartialEq, Eq, Debug, Copy, Clone, Ord, PartialOrd, Hash)]
32#[repr(usize)]
33pub enum SupportedLayer {
34    L4,
35    L7,
36}
37
38/// Trait implemented for each Layer variant
39pub(crate) trait TrackableLayer {
40    /// Ingest the next packet in the stream (reassembled, if TCP).
41    /// Returns State transition(s) triggered.
42    /// If multiple state transitions are triggered, the "Streaming" (InX)
43    /// should be returned first. This will invoke methods on `T` based
44    /// on the Layer and current State.
45    fn process_stream(&mut self, pdu: &mut L4Pdu, registry: &ParserRegistry) -> StateTransition;
46
47    /// Should be checked directly after a state transition to see
48    /// if process_stream needs to be called again.
49    /// For example, the packet used to "discover" a protocol will
50    /// also be part of its header.
51    fn needs_process(&self, tx: StateTransition, pdu: &L4Pdu) -> bool;
52
53    /// No actions are active and, if applicable, no sub-layers have
54    /// active actions.
55    fn drop(&self) -> bool;
56
57    /// "Consume_stream" must be called by the transport layer.
58    /// TCP reassemly is expected if applicable.
59    fn needs_stream(&self) -> bool;
60
61    /// Used to remove any actions that are invalid at this layer.
62    /// For example: an L4 Update may trigger an L7 "parse" action, which
63    /// would be invalid once in payload if another session is not expected.
64    fn end_state_tx(&mut self);
65
66    /// Indicate that the connection has terminated.
67    /// Should be invoked repeatedly until it returns None
68    fn handle_terminate(&mut self) -> Option<StateTransition>;
69}
70
71impl Layer {
72    /// Accessors for LayerInfo
73    pub fn layer_info_mut(&mut self) -> &mut LayerInfo {
74        match self {
75            Layer::L7(session) => &mut session.linfo,
76        }
77    }
78
79    pub fn layer_info(&self) -> &LayerInfo {
80        match self {
81            Layer::L7(session) => &session.linfo,
82        }
83    }
84
85    /// Push an action
86    pub fn extend_actions(&mut self, action: &TrackedActions) {
87        self.layer_info_mut().actions.extend(action)
88    }
89
90    /// Accessors
91    pub fn last_session(&self) -> &Session {
92        match self {
93            Layer::L7(session) => match session.sessions.last() {
94                Some(s) => s,
95                None => &DEFAULT_SESSION,
96            },
97        }
98    }
99
100    pub fn drain_sessions(&mut self) -> Vec<Session> {
101        match self {
102            Layer::L7(session) => session.parser.drain_sessions(),
103        }
104    }
105
106    pub fn first_session(&self) -> &Session {
107        match self {
108            Layer::L7(session) => match session.sessions.first() {
109                Some(s) => s,
110                None => &DEFAULT_SESSION,
111            },
112        }
113    }
114
115    pub fn sessions(&self) -> &Vec<Session> {
116        match self {
117            Layer::L7(session) => &session.sessions,
118        }
119    }
120
121    pub fn last_protocol(&self) -> SessionProto {
122        match self {
123            Layer::L7(session) => session.get_protocol(),
124        }
125    }
126}
127
128impl TrackableLayer for Layer {
129    fn process_stream(&mut self, pdu: &mut L4Pdu, registry: &ParserRegistry) -> StateTransition {
130        match self {
131            Layer::L7(session) => session.process_stream(pdu, registry),
132        }
133    }
134
135    fn needs_process(&self, tx: StateTransition, pdu: &L4Pdu) -> bool {
136        match self {
137            Layer::L7(session) => session.needs_process(tx, pdu),
138        }
139    }
140
141    fn drop(&self) -> bool {
142        match self {
143            Layer::L7(session) => session.drop(),
144        }
145    }
146
147    fn needs_stream(&self) -> bool {
148        match self {
149            Layer::L7(session) => session.needs_stream(),
150        }
151    }
152
153    fn end_state_tx(&mut self) {
154        match self {
155            Layer::L7(session) => session.end_state_tx(),
156        }
157    }
158
159    fn handle_terminate(&mut self) -> Option<StateTransition> {
160        match self {
161            Layer::L7(session) => session.handle_terminate(),
162        }
163    }
164}
165
166/// Stored for each Layer
167#[derive(Debug, Clone, PartialEq, Eq)]
168pub struct LayerInfo {
169    pub state: LayerState,
170    pub actions: TrackedActions,
171}
172
173impl Default for LayerInfo {
174    fn default() -> Self {
175        Self::new()
176    }
177}
178
179impl LayerInfo {
180    pub fn new() -> Self {
181        Self {
182            state: LayerState::Discovery,
183            actions: TrackedActions::new(),
184        }
185    }
186
187    pub(crate) fn drop(&self) -> bool {
188        self.state == LayerState::None || self.actions.drop()
189    }
190}
191
192/// L6/L7 parsing infrastructure
193#[derive(Debug)]
194pub struct L7Session {
195    /// Layer management
196    pub linfo: LayerInfo,
197    /// Stateful protocol parser (once identified, or None)
198    pub parser: ConnParser,
199    /// Parsed sessions, if applicable
200    pub sessions: Vec<Session>,
201    /// Sessions seen on terminate that are not fully parsed
202    pub pending_sessions: Vec<Session>,
203    // Further encapsulated layers could go here.
204}
205
206impl L7Session {
207    /// Initialize infrastructure for probing, parsing, and tracking
208    /// L6/L7 (application-layer) sessions.
209    pub fn new() -> Self {
210        Self {
211            linfo: LayerInfo::new(),
212            parser: ConnParser::Unknown,
213            sessions: Vec::new(),
214            pending_sessions: Vec::new(),
215        }
216    }
217
218    /// Accessor for Protocol
219    pub fn get_protocol(&self) -> SessionProto {
220        match self.linfo.state {
221            LayerState::Discovery => SessionProto::Probing,
222            _ => self.parser.protocol(),
223        }
224    }
225}
226
227// Clippy #new_without_default warning for pub types
228impl Default for L7Session {
229    fn default() -> Self {
230        Self::new()
231    }
232}
233
234impl TrackableLayer for L7Session {
235    fn end_state_tx(&mut self) {
236        #[cfg(debug_assertions)]
237        {
238            log::debug!(
239                "End state transition, state: {:?}, L7 actions: {:?}",
240                self.linfo.state,
241                self.linfo.actions.active
242            );
243        }
244        // Nothing to parse if in payload and no more sessions expected
245        if self.linfo.actions.needs_parse()
246            && matches!(self.linfo.state, LayerState::Payload)
247            && !matches!(
248                self.parser.session_parsed_state(),
249                ParsingState::Parsing | ParsingState::Probing
250            )
251        {
252            self.linfo.actions.clear(&Actions::Parse);
253        }
254    }
255
256    fn needs_process(&self, tx: StateTransition, pdu: &L4Pdu) -> bool {
257        if self.linfo.state == LayerState::None {
258            return false;
259        }
260        (tx == StateTransition::L7OnDisc && pdu.length() > 0)
261            || (tx == StateTransition::L7EndHdrs && pdu.ctxt.app_offset.is_some())
262    }
263
264    fn drop(&self) -> bool {
265        self.linfo.drop()
266    }
267
268    fn needs_stream(&self) -> bool {
269        self.linfo.actions.needs_parse()
270    }
271
272    /// If some subscription is waiting for sessions, drain
273    /// pending (not yet fully parsed) sessions from the parser.
274    /// Move these sessions one-by-one to `self.sessions` until
275    /// none are left. This should be invoked until it returns None.
276    fn handle_terminate(&mut self) -> Option<StateTransition> {
277        let state = self.linfo.state;
278        self.linfo.state = LayerState::None;
279        if !self.linfo.actions.needs_parse() {
280            return None;
281        }
282        if matches!(state, LayerState::None) {
283            return None;
284        }
285        // Discovery failed
286        if matches!(state, LayerState::Discovery) {
287            return Some(StateTransition::L7OnDisc);
288        }
289        self.pending_sessions.extend(self.parser.drain_sessions());
290        // Parsing failed
291        if self.pending_sessions.is_empty() {
292            return Some(StateTransition::L7EndHdrs);
293        }
294        // New session ready
295        self.sessions.push(self.pending_sessions.pop().unwrap());
296        if !self.sessions.is_empty() {
297            // Handle multiple sessions
298            self.linfo.state = state;
299        }
300        Some(StateTransition::L7EndHdrs)
301    }
302
303    fn process_stream(&mut self, pdu: &mut L4Pdu, registry: &ParserRegistry) -> StateTransition {
304        match self.linfo.state {
305            LayerState::Discovery => {
306                match registry.probe_all(pdu) {
307                    ProbeRegistryResult::Some(conn_parser) => {
308                        // Application-layer protocol known
309                        self.parser = conn_parser;
310                        self.linfo.state = LayerState::Headers;
311                        return StateTransition::L7OnDisc;
312                    }
313                    ProbeRegistryResult::None => {
314                        // All relevant parsers have failed to match
315                        self.linfo.state = LayerState::None;
316                        self.linfo.actions.clear(&Actions::Parse);
317                        return StateTransition::L7OnDisc;
318                    }
319                    ProbeRegistryResult::Unsure => { /* skip */ }
320                }
321            }
322            LayerState::Headers => {
323                match self.parser.parse(pdu) {
324                    ParseResult::HeadersDone(id) => {
325                        if let Some(session) = self.parser.remove_session(id) {
326                            self.sessions.push(session);
327                        }
328                        if let Some(offset) = self.parser.body_offset() {
329                            pdu.ctxt.app_offset = Some(offset);
330                        }
331                        self.linfo.state = LayerState::Payload;
332                        return StateTransition::L7EndHdrs;
333                    }
334                    ParseResult::None => {
335                        self.linfo.state = LayerState::None;
336                        return StateTransition::L7EndHdrs;
337                    }
338                    ParseResult::Done(id) => {
339                        if let Some(session) = self.parser.remove_session(id) {
340                            self.sessions.push(session);
341                        }
342                        self.linfo.state = LayerState::None;
343                        return StateTransition::L7EndHdrs;
344                    }
345                    _ => { /* continue */ }
346                }
347            }
348            LayerState::Payload => {
349                pdu.ctxt.app_offset = Some(0);
350                if self.linfo.actions.needs_parse() {
351                    match self.parser.session_parsed_state() {
352                        ParsingState::Probing => {
353                            // TODO unimplemented: nested sessions
354                        }
355                        ParsingState::Parsing => {
356                            // TODO unimplemented: pipelined sessions
357                        }
358                        _ => {}
359                    }
360                }
361                // TODO - add API for parser to consume payload
362                // if applicable and return when session is "done"
363            }
364            LayerState::None => {
365                // Do nothing
366            }
367        }
368        StateTransition::Packet
369    }
370}
iris_core/conntrack/conn/conn_layers.rs

iris_core/conntrack/conn/
conn_layers.rs
