ipfix_parser/parsers/
decoder.rs

1//! Message, Set and Template parsing, and the stateful Template cache
2//!
3//! # Template management (RFC 7011 Section 8)
4//!
5//! Templates are scoped to a Transport Session and an Observation Domain. A
6//! [`Decoder`] represents one Transport Session: create one per Exporter (TCP
7//! connection, SCTP association, or UDP source) and keep it alive while the
8//! session lasts. Within a [`Decoder`], Templates are keyed by
9//! (Observation Domain ID, Template ID).
10//!
11//! Templates and Template Withdrawals take effect in the order in which they
12//! appear in a Message (RFC 7011 Section 8.1), so a Data Set can be decoded
13//! with a Template defined earlier in the same Message.
14//!
15//! # Transport (RFC 7011 Section 8.4)
16//!
17//! Template management depends on the transport, set with
18//! [`DecoderConfig::transport`] ([`Transport::Udp`] by default):
19//!
20//! - Over UDP, Template Withdrawals MUST be ignored by the Collecting Process.
21//!   They are still returned in the parsed [`Set`], but do not change the
22//!   cache. This also prevents a spoofed UDP packet from wiping the cache.
23//! - Over UDP, Templates not refreshed within
24//!   [`DecoderConfig::template_lifetime`] are discarded. Time is the Export
25//!   Time of the Message Headers, so decoding stays deterministic and does not
26//!   depend on the system clock: a Template is expired when the Export Time of
27//!   the current Message is more than the lifetime after the Export Time of
28//!   the Message that last carried it. The RFC suggests a lifetime of at least
29//!   3 times the Exporter's Template retransmission interval.
30//! - Over TCP and SCTP, Templates live until withdrawn or until the Transport
31//!   Session ends ([`Decoder::clear`]).
32//! - IPFIX Files (RFC 5655) are a serialized Message stream read as one
33//!   Transport Session. A File Reader must support the union of the Template
34//!   management features of all transports (RFC 5655 Section 7.1): Templates
35//!   can be redefined, Template Withdrawals are applied, and Templates never
36//!   expire.
37//!
38//! # Known Specification Errata
39//!
40//! - **RFC 7011 Errata 7413 (Section 3.4.1):** a Template whose Data Records
41//!   would be zero octets long is rejected. Otherwise every octet of a Data Set
42//!   would hold infinitely many zero-length records.
43
44use super::error::{ParseError, Result};
45use super::value::{decode_record, DecodeContext};
46use super::Cursor;
47use crate::models::*;
48use crate::registry::{MIN_DATA_SET_ID, OPTIONS_TEMPLATE_SET_ID, TEMPLATE_SET_ID};
49use std::collections::HashMap;
50
51/// Length of the IPFIX Message Header, in octets
52const MESSAGE_HEADER_LENGTH: usize = 16;
53
54/// Length of the Set Header, in octets
55const SET_HEADER_LENGTH: usize = 4;
56
57/// Transport protocol of the Transport Session (RFC 7011 Section 10)
58#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
59pub enum Transport {
60    /// UDP: Template Withdrawals are ignored, and Templates can expire (see
61    /// [`DecoderConfig::template_lifetime`])
62    #[default]
63    Udp,
64    /// TCP: Template Withdrawals are applied, Templates never expire
65    Tcp,
66    /// SCTP: Template Withdrawals are applied, Templates never expire
67    Sctp,
68    /// IPFIX File (RFC 5655): Template Withdrawals are applied, Templates never
69    /// expire. One file is one Transport Session.
70    File,
71}
72
73/// Decoder configuration
74#[derive(Debug, Clone, Copy, PartialEq, Eq)]
75pub struct DecoderConfig {
76    /// Transport protocol of the Transport Session, which drives Template
77    /// management (RFC 7011 Section 8.4). Defaults to UDP, the transport of
78    /// most deployments.
79    pub transport: Transport,
80    /// Template lifetime in seconds, for UDP only (RFC 7011 Section 8.4):
81    /// Templates not refreshed for longer than this are discarded. `None`
82    /// (the default) keeps Templates until they are redefined.
83    pub template_lifetime: Option<u32>,
84    /// Maximum number of Templates and Options Templates kept in the cache,
85    /// across all Observation Domains. Defining a new Template beyond this
86    /// limit returns [`ParseError::TemplateLimitExceeded`]; redefining an
87    /// existing Template is always allowed.
88    pub max_templates: usize,
89    /// Maximum nesting depth of structured data (RFC 6313 lists). Deeper lists
90    /// are kept as [`FieldValue::Raw`].
91    pub max_list_depth: usize,
92}
93
94impl Default for DecoderConfig {
95    fn default() -> Self {
96        Self {
97            transport: Transport::Udp,
98            template_lifetime: None,
99            max_templates: 4096,
100            max_list_depth: 8,
101        }
102    }
103}
104
105/// A Template or Options Template stored in the cache
106#[derive(Debug, Clone, PartialEq, Eq)]
107pub(crate) struct Template {
108    pub(crate) field_specifiers: Vec<FieldSpecifier>,
109    /// 0 for (non-options) Templates
110    pub(crate) scope_field_count: u16,
111    /// Smallest possible encoded Data Record length (variable-length fields
112    /// count as one octet)
113    pub(crate) min_record_length: usize,
114    /// Export Time of the Message that last carried this Template
115    pub(crate) last_received: u32,
116}
117
118impl Template {
119    fn new(field_specifiers: Vec<FieldSpecifier>, scope_field_count: u16) -> Result<Self> {
120        let min_record_length = field_specifiers
121            .iter()
122            .map(|f| {
123                if f.is_variable_length() {
124                    1
125                } else {
126                    f.field_length as usize
127                }
128            })
129            .sum();
130        if min_record_length == 0 {
131            return Err(ParseError::InvalidData(
132                "Template describes zero-length Data Records".to_string(),
133            ));
134        }
135        Ok(Self {
136            field_specifiers,
137            scope_field_count,
138            min_record_length,
139            last_received: 0,
140        })
141    }
142
143    fn is_options(&self) -> bool {
144        self.scope_field_count > 0
145    }
146}
147
148pub(crate) type TemplateCache = HashMap<(u32, u16), Template>;
149
150/// Stateful IPFIX decoder holding the Template cache of one Transport Session
151#[derive(Debug, Clone, Default)]
152pub struct Decoder {
153    config: DecoderConfig,
154    templates: TemplateCache,
155    /// Export Time of the last expiry sweep (sweeps run at most once per
156    /// second of Export Time)
157    last_sweep: Option<u32>,
158}
159
160impl Decoder {
161    /// Create a decoder with the default configuration
162    pub fn new() -> Self {
163        Self::default()
164    }
165
166    /// Create a decoder with a custom configuration
167    pub fn with_config(config: DecoderConfig) -> Self {
168        Self {
169            config,
170            templates: HashMap::new(),
171            last_sweep: None,
172        }
173    }
174
175    /// Number of Templates and Options Templates in the cache
176    pub fn template_count(&self) -> usize {
177        self.templates.len()
178    }
179
180    /// Whether a Template (or Options Template) is known for this Observation
181    /// Domain and Template ID
182    pub fn has_template(&self, observation_domain_id: u32, template_id: u16) -> bool {
183        self.templates
184            .contains_key(&(observation_domain_id, template_id))
185    }
186
187    /// Export Time of the Message that last carried this Template (or Options
188    /// Template), i.e. the "Last Received" of RFC 7011 Section 8.4
189    pub fn template_last_received(
190        &self,
191        observation_domain_id: u32,
192        template_id: u16,
193    ) -> Option<u32> {
194        self.templates
195            .get(&(observation_domain_id, template_id))
196            .map(|t| t.last_received)
197    }
198
199    /// Discard the Templates not refreshed within
200    /// [`DecoderConfig::template_lifetime`] at time `now` (in seconds since the
201    /// UNIX epoch, like the Export Time). This runs automatically for each
202    /// decoded Message over UDP; call it to expire Templates with your own
203    /// clock, e.g. when an Exporter stopped sending. Does nothing without a
204    /// lifetime.
205    pub fn expire_templates(&mut self, now: u32) {
206        if let Some(lifetime) = self.config.template_lifetime {
207            self.templates
208                .retain(|_, t| now.saturating_sub(t.last_received) <= lifetime);
209        }
210    }
211
212    /// Remove all Templates (e.g. at the end of the Transport Session)
213    pub fn clear(&mut self) {
214        self.templates.clear();
215    }
216
217    /// Remove all Templates of an Observation Domain
218    pub fn remove_observation_domain(&mut self, observation_domain_id: u32) {
219        self.templates
220            .retain(|(domain, _), _| *domain != observation_domain_id);
221    }
222
223    /// Decode one IPFIX Message, updating the Template cache.
224    ///
225    /// Octets after the Message Length are ignored; use
226    /// [`decode_all`](Self::decode_all) for a buffer holding several Messages.
227    pub fn decode(&mut self, data: &[u8]) -> Result<Message> {
228        let mut cursor = Cursor::new(data);
229        let header = parse_message_header(&mut cursor)?;
230        let body = cursor.read_bytes(
231            header.length as usize - MESSAGE_HEADER_LENGTH,
232            "message body",
233        )?;
234
235        let now = header.export_time;
236        if self.config.transport == Transport::Udp && self.last_sweep != Some(now) {
237            self.expire_templates(now);
238            self.last_sweep = Some(now);
239        }
240
241        let mut sets = Vec::new();
242        let mut cursor = Cursor::new(body);
243        while !cursor.is_empty() {
244            sets.push(self.parse_set(&mut cursor, header.observation_domain_id, now)?);
245        }
246
247        Ok(Message { header, sets })
248    }
249
250    /// Decode a buffer of consecutive IPFIX Messages (e.g. a TCP stream or an
251    /// IPFIX File), updating the Template cache.
252    pub fn decode_all(&mut self, data: &[u8]) -> Result<Vec<Message>> {
253        let mut messages = Vec::new();
254        let mut offset = 0;
255        while offset < data.len() {
256            let message = self.decode(&data[offset..])?;
257            offset += message.header.length as usize;
258            messages.push(message);
259        }
260        Ok(messages)
261    }
262
263    fn parse_set(&mut self, cursor: &mut Cursor, domain: u32, now: u32) -> Result<Set> {
264        let header = SetHeader {
265            set_id: cursor.read_u16("set header")?,
266            length: cursor.read_u16("set header")?,
267        };
268        if (header.length as usize) < SET_HEADER_LENGTH {
269            return Err(ParseError::InvalidData(format!(
270                "Set length {} is smaller than the Set Header",
271                header.length
272            )));
273        }
274        let content = cursor.read_bytes(header.length as usize - SET_HEADER_LENGTH, "set")?;
275        let mut content = Cursor::new(content);
276
277        match header.set_id {
278            TEMPLATE_SET_ID => Ok(Set::Template {
279                header,
280                records: self.parse_template_set(&mut content, domain, now)?,
281            }),
282            OPTIONS_TEMPLATE_SET_ID => Ok(Set::OptionsTemplate {
283                header,
284                records: self.parse_options_template_set(&mut content, domain, now)?,
285            }),
286            id if id >= MIN_DATA_SET_ID => match self.templates.get(&(domain, id)) {
287                Some(template) => {
288                    let ctx = DecodeContext {
289                        templates: &self.templates,
290                        observation_domain_id: domain,
291                        max_depth: self.config.max_list_depth,
292                    };
293                    let mut records = Vec::new();
294                    // Records are read until the remaining octets are fewer than the
295                    // smallest possible record: those are padding (RFC 7011 Section 3.3.1)
296                    while content.remaining() >= template.min_record_length {
297                        records.push(decode_record(&ctx, id, template, &mut content, 0)?);
298                    }
299                    Ok(Set::Data { header, records })
300                }
301                None => Ok(Set::UndecodedData {
302                    header,
303                    payload: content.read_rest().to_vec(),
304                }),
305            },
306            _ => Ok(Set::Reserved {
307                header,
308                payload: content.read_rest().to_vec(),
309            }),
310        }
311    }
312
313    /// Parse a Template Set (RFC 7011 Section 3.4.1) and apply it to the cache
314    fn parse_template_set(
315        &mut self,
316        cursor: &mut Cursor,
317        domain: u32,
318        now: u32,
319    ) -> Result<Vec<TemplateRecord>> {
320        let mut records = Vec::new();
321        // A Template Record is at least 4 octets: fewer remaining octets are padding
322        while cursor.remaining() >= 4 {
323            let template_id = cursor.read_u16("template record header")?;
324            let field_count = cursor.read_u16("template record header")?;
325
326            if field_count == 0 {
327                self.withdraw(domain, template_id, false)?;
328                records.push(TemplateRecord {
329                    template_id,
330                    field_count,
331                    field_specifiers: Vec::new(),
332                });
333                continue;
334            }
335
336            check_template_id(template_id)?;
337            let field_specifiers = parse_field_specifiers(cursor, field_count)?;
338            self.insert(
339                domain,
340                template_id,
341                Template::new(field_specifiers.clone(), 0)?,
342                now,
343            )?;
344            records.push(TemplateRecord {
345                template_id,
346                field_count,
347                field_specifiers,
348            });
349        }
350        Ok(records)
351    }
352
353    /// Parse an Options Template Set (RFC 7011 Section 3.4.2.2) and apply it to
354    /// the cache
355    fn parse_options_template_set(
356        &mut self,
357        cursor: &mut Cursor,
358        domain: u32,
359        now: u32,
360    ) -> Result<Vec<OptionsTemplateRecord>> {
361        let mut records = Vec::new();
362        // An Options Template Withdrawal is 4 octets: fewer remaining octets are padding
363        while cursor.remaining() >= 4 {
364            let template_id = cursor.read_u16("options template record header")?;
365            let field_count = cursor.read_u16("options template record header")?;
366
367            if field_count == 0 {
368                self.withdraw(domain, template_id, true)?;
369                records.push(OptionsTemplateRecord {
370                    template_id,
371                    field_count,
372                    scope_field_count: 0,
373                    field_specifiers: Vec::new(),
374                });
375                continue;
376            }
377
378            check_template_id(template_id)?;
379            let scope_field_count = cursor.read_u16("options template record header")?;
380            if scope_field_count == 0 || scope_field_count > field_count {
381                return Err(ParseError::InvalidData(format!(
382                    "Invalid Scope Field Count {} for Field Count {}",
383                    scope_field_count, field_count
384                )));
385            }
386            let field_specifiers = parse_field_specifiers(cursor, field_count)?;
387            self.insert(
388                domain,
389                template_id,
390                Template::new(field_specifiers.clone(), scope_field_count)?,
391                now,
392            )?;
393            records.push(OptionsTemplateRecord {
394                template_id,
395                field_count,
396                scope_field_count,
397                field_specifiers,
398            });
399        }
400        Ok(records)
401    }
402
403    fn insert(
404        &mut self,
405        domain: u32,
406        template_id: u16,
407        mut template: Template,
408        now: u32,
409    ) -> Result<()> {
410        template.last_received = now;
411        let key = (domain, template_id);
412        if !self.templates.contains_key(&key) && self.templates.len() >= self.config.max_templates {
413            return Err(ParseError::TemplateLimitExceeded(self.config.max_templates));
414        }
415        self.templates.insert(key, template);
416        Ok(())
417    }
418
419    /// Apply a Template Withdrawal (RFC 7011 Section 8.1). Over UDP,
420    /// withdrawals MUST be ignored (RFC 7011 Section 8.4).
421    fn withdraw(&mut self, domain: u32, template_id: u16, options: bool) -> Result<()> {
422        if self.config.transport == Transport::Udp {
423            return Ok(());
424        }
425        let withdraw_all_id = if options {
426            OPTIONS_TEMPLATE_SET_ID
427        } else {
428            TEMPLATE_SET_ID
429        };
430        if template_id == withdraw_all_id {
431            // All (Options) Templates Withdrawal
432            self.templates
433                .retain(|(d, _), t| *d != domain || t.is_options() != options);
434            Ok(())
435        } else if template_id >= MIN_DATA_SET_ID {
436            self.templates.remove(&(domain, template_id));
437            Ok(())
438        } else {
439            Err(ParseError::InvalidData(format!(
440                "Invalid Template ID {} in Template Withdrawal",
441                template_id
442            )))
443        }
444    }
445}
446
447/// Parse an IPFIX Message Header (RFC 7011 Section 3.1)
448fn parse_message_header(cursor: &mut Cursor) -> Result<MessageHeader> {
449    let available = cursor.remaining();
450    let header = MessageHeader {
451        version: cursor.read_u16("message header")?,
452        length: cursor.read_u16("message header")?,
453        export_time: cursor.read_u32("message header")?,
454        sequence_number: cursor.read_u32("message header")?,
455        observation_domain_id: cursor.read_u32("message header")?,
456    };
457    if header.version != crate::registry::IPFIX_VERSION {
458        return Err(ParseError::InvalidVersion(header.version));
459    }
460    if (header.length as usize) < MESSAGE_HEADER_LENGTH {
461        return Err(ParseError::InvalidData(format!(
462            "Message length {} is smaller than the Message Header",
463            header.length
464        )));
465    }
466    if header.length as usize > available {
467        return Err(ParseError::Truncated {
468            context: "message",
469            needed: header.length as usize,
470            available,
471        });
472    }
473    Ok(header)
474}
475
476fn check_template_id(template_id: u16) -> Result<()> {
477    if template_id < MIN_DATA_SET_ID {
478        return Err(ParseError::InvalidData(format!(
479            "Invalid Template ID {} (must be > 255)",
480            template_id
481        )));
482    }
483    Ok(())
484}
485
486/// Parse `field_count` Field Specifiers (RFC 7011 Section 3.2)
487fn parse_field_specifiers(cursor: &mut Cursor, field_count: u16) -> Result<Vec<FieldSpecifier>> {
488    // Each Field Specifier is at least 4 octets: bound the allocation by the input
489    let mut field_specifiers =
490        Vec::with_capacity((field_count as usize).min(cursor.remaining() / 4));
491    for _ in 0..field_count {
492        let (information_element_identifier, enterprise) =
493            cursor.read_field_id("field specifier")?;
494        let field_length = cursor.read_u16("field specifier")?;
495        let enterprise_number = if enterprise {
496            Some(cursor.read_u32("field specifier")?)
497        } else {
498            None
499        };
500        field_specifiers.push(FieldSpecifier {
501            information_element_identifier,
502            field_length,
503            enterprise_number,
504        });
505    }
506    Ok(field_specifiers)
507}
508
509/// Parse a single self-contained IPFIX Message.
510///
511/// Only Templates defined in the Message itself are used to decode its Data
512/// Sets; Data Sets with unknown Templates are returned as
513/// [`Set::UndecodedData`]. Use a [`Decoder`] to keep Templates across Messages.
514///
515/// This uses the default [`DecoderConfig`] ([`Transport::Udp`], the transport
516/// of a single datagram): Template Withdrawals are ignored.
517pub fn parse_message(data: &[u8]) -> Result<Message> {
518    Decoder::new().decode(data)
519}
520
521/// Parse a buffer of consecutive IPFIX Messages, sharing Templates across
522/// them (as if they belonged to the same Transport Session).
523///
524/// A buffer of consecutive Messages is read as an IPFIX File (RFC 5655), with
525/// [`Transport::File`] semantics: Template Withdrawals are applied and
526/// Templates never expire. Compressed files must be decompressed by the
527/// caller.
528pub fn parse_messages(data: &[u8]) -> Result<Vec<Message>> {
529    Decoder::with_config(DecoderConfig {
530        transport: Transport::File,
531        ..DecoderConfig::default()
532    })
533    .decode_all(data)
534}
ipfix_parser/parsers/decoder.rs

ipfix_parser/parsers/
decoder.rs
