Fix netflow v9 parsing (#42)

Author: maksimtor

src/lib.rs

@@ -344,7 +344,6 @@ mod tests {
             },
         ];
         let template = V9Template {
-            length: 16,
             field_count: 2,
             template_id: 258,
             fields,
@@ -425,7 +424,6 @@ mod tests {
             0, 9, 0, 26, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 0, 1, 2, 0, 10, 0, 8, 0, 0, 1, 1,
         ];
         let template = V9Template {
-            length: 10,
             field_count: 2,
             template_id: 258,
             fields: vec![],

src/snapshots/netflow_parser__tests__it_parses_v9.snap

@@ -12,20 +12,20 @@ expression: "NetflowParser::default().parse_bytes(&packet)"
       source_id: 1
     flowsets:
       - flow_set_id: 0
-        template:
-          length: 16
-          template_id: 258
-          field_count: 2
-          fields:
-            - field_type_number: 1
-              field_type: InBytes
-              field_length: 4
-            - field_type_number: 8
-              field_type: Ipv4SrcAddr
-              field_length: 4
+        length: 16
+        templates:
+          - template_id: 258
+            field_count: 2
+            fields:
+              - field_type_number: 1
+                field_type: InBytes
+                field_length: 4
+              - field_type_number: 8
+                field_type: Ipv4SrcAddr
+                field_length: 4
       - flow_set_id: 258
+        length: 12
         data:
-          length: 12
           data_fields:
             - InBytes:
                 DataNumber: 151126788

src/snapshots/netflow_parser__tests__it_parses_v9_data_cached_template.snap

@@ -12,8 +12,8 @@ expression: parser.parse_bytes(&packet)
       source_id: 1
     flowsets:
       - flow_set_id: 258
+        length: 12
         data:
-          length: 12
           data_fields:
             - InBytes:
                 DataNumber: 151126788

src/snapshots/netflow_parser__tests__it_parses_v9_many_flows.snap

@@ -12,20 +12,20 @@ expression: "NetflowParser::default().parse_bytes(&packet)"
       source_id: 1
     flowsets:
       - flow_set_id: 0
-        template:
-          length: 16
-          template_id: 258
-          field_count: 2
-          fields:
-            - field_type_number: 1
-              field_type: InBytes
-              field_length: 4
-            - field_type_number: 8
-              field_type: Ipv4SrcAddr
-              field_length: 4
+        length: 16
+        templates:
+          - template_id: 258
+            field_count: 2
+            fields:
+              - field_type_number: 1
+                field_type: InBytes
+                field_length: 4
+              - field_type_number: 8
+                field_type: Ipv4SrcAddr
+                field_length: 4
       - flow_set_id: 258
+        length: 12
         data:
-          length: 12
           data_fields:
             - InBytes:
                 DataNumber: 151126788
@@ -35,4 +35,3 @@ expression: "NetflowParser::default().parse_bytes(&packet)"
                 DataNumber: 151126788
               Ipv4SrcAddr:
                 Ip4Addr: 9.9.9.8
-

src/snapshots/netflow_parser__tests__it_parses_v9_options_template.snap

@@ -12,25 +12,25 @@ expression: "NetflowParser::default().parse_bytes(&packet)"
       source_id: 1
     flowsets:
       - flow_set_id: 1
-        options_template:
-          length: 22
-          template_id: 275
-          options_scope_length: 4
-          options_length: 8
-          scope_fields:
-            - field_type_number: 2
-              field_type: Interface
-              field_length: 2
-          option_fields:
-            - field_type_number: 34
-              field_type: SamplingInterval
-              field_length: 2
-            - field_type_number: 36
-              field_type: FlowActiveTimeout
-              field_length: 1
+        length: 22
+        options_templates:
+          - template_id: 275
+            options_scope_length: 4
+            options_length: 8
+            scope_fields:
+              - field_type_number: 2
+                field_type: Interface
+                field_length: 2
+            option_fields:
+              - field_type_number: 34
+                field_type: SamplingInterval
+                field_length: 2
+              - field_type_number: 36
+                field_type: FlowActiveTimeout
+                field_length: 1
       - flow_set_id: 275
+        length: 9
         options_data:
-          length: 9
           scope_fields:
             - interface:
                 - 0

src/snapshots/netflow_parser__tests__it_parses_v9_with_no_template_fields_raises_error.snap

@@ -2,33 +2,22 @@
 source: src/lib.rs
 expression: parser.parse_bytes(&packet)
 ---
-- Error:
-    error_message: Could not parse v9_packet
-    bytes:
-      - 0
-      - 9
-      - 0
-      - 26
-      - 0
-      - 0
-      - 0
-      - 1
-      - 0
-      - 0
-      - 0
-      - 1
-      - 0
-      - 0
-      - 0
-      - 0
-      - 1
-      - 2
-      - 0
-      - 10
-      - 0
-      - 8
-      - 0
-      - 0
-      - 1
-      - 1
+- V9:
+    header:
+      version: 9
+      count: 26
+      sys_up_time: 1
+      unix_secs: 1
+      sequence_number: 0
+      source_id: 16908298
+    flowsets:
+      - flow_set_id: 8
+        length: 0
+        unparsed_data:
+          - 0
+          - 8
+          - 0
+          - 0
+          - 1
+          - 1
 

src/variable_versions/v9.rs

@@ -49,18 +49,35 @@ fn parse_flowsets<'a>(
     let mut remaining = i;
 
     // Header.count represents total number of records in data + records in templates
-    while count > 0 {
-        let (i, flowset) = FlowSet::parse(remaining, parser)?;
-        remaining = i;
+    while count > 0 && !remaining.is_empty() {
+        let (i, mut flowset) = FlowSet::parse(remaining, parser)?;
 
-        if flowset.template.is_some() || flowset.options_template.is_some() {
-            count = count.saturating_sub(1);
-        } else if let Some(data) = flowset.data.as_ref() {
+        if let Some(data) = &flowset.templates {
+            count = count.saturating_sub(data.len());
+        }
+
+        if let Some(data) = &flowset.options_templates {
+            count = count.saturating_sub(data.len());
+        }
+
+        if let Some(data) = &flowset.data.as_ref() {
             count = count.saturating_sub(data.data_fields.len());
-        } else if flowset.options_data.as_ref().is_some() {
+        }
+
+        if flowset.options_data.as_ref().is_some() {
             count = count.saturating_sub(1);
         }
 
+        if flowset.is_empty() {
+            flowset.unparsed_data = Some(remaining.to_vec());
+            remaining = &[];
+        } else if flowset.is_unparsed() {
+            flowset.unparsed_data = Some(remaining[..flowset.length as usize].to_vec());
+            remaining = &remaining[flowset.length as usize..];
+        } else {
+            remaining = i;
+        }
+
         flowsets.push(flowset)
     }
 
@@ -104,26 +121,39 @@ pub struct FlowSet {
     /// the template record that describes option fields (described below) has a
     /// FlowSet ID of 1. A data record always has a nonzero FlowSet ID greater than 255.
     pub flow_set_id: u16,
+    /// This field gives the length of the data FlowSet. Length is expressed in TLV format,
+    /// meaning that the value includes the bytes used for the FlowSet ID and the length bytes
+    /// themselves, as well as the combined lengths of any included data records.
+    pub length: u16,
     /// Templates
     #[nom(
         Cond = "flow_set_id == TEMPLATE_ID",
         // Save our templates
-        PostExec = "if let Some(template) = template.clone() { parser.templates.insert(template.template_id, template); }"
+        PostExec = "if let Some(templates) = templates.clone() { 
+            for template in templates {
+                parser.templates.insert(template.template_id, template); 
+            }
+        }"
     )]
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub template: Option<Template>,
+    pub templates: Option<Vec<Template>>,
     // Options template
     #[nom(
         Cond = "flow_set_id == OPTIONS_TEMPLATE_ID",
+        Parse = "{ |i| parse_options_template_vec(i, length) }",
         // Save our options templates
-        PostExec = "if let Some(options_template) = options_template.clone() { parser.options_templates.insert(options_template.template_id, options_template); }"
+        PostExec = "if let Some(options_templates) = options_templates.clone() { 
+            for template in options_templates {
+                parser.options_templates.insert(template.template_id, template); 
+            } 
+        }"
     )]
     #[serde(skip_serializing_if = "Option::is_none")]
-    pub options_template: Option<OptionsTemplate>,
+    pub options_templates: Option<Vec<OptionsTemplate>>,
     // Options Data
     #[nom(
         Cond = "flow_set_id > FLOW_SET_MIN_RANGE && parser.options_templates.get(&flow_set_id).is_some()",
-        Parse = "{ |i| OptionsData::parse(i, parser, flow_set_id) }"
+        Parse = "{ |i| OptionsData::parse(i, parser, flow_set_id, length) }"
     )]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub options_data: Option<OptionsData>,
@@ -134,18 +164,27 @@ pub struct FlowSet {
     )]
     #[serde(skip_serializing_if = "Option::is_none")]
     pub data: Option<Data>,
+    // Unparsed data
+    #[nom(Ignore)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub unparsed_data: Option<Vec<u8>>,
+}
+
+impl FlowSet {
+    fn is_unparsed(&self) -> bool {
+        self.templates.is_none()
+            && self.options_templates.is_none()
+            && self.data.is_none()
+            && self.options_data.is_none()
+    }
+
+    fn is_empty(&self) -> bool {
+        self.length == 0
+    }
 }
 
 #[derive(Debug, PartialEq, Clone, Serialize, Nom)]
 pub struct Template {
-    /// Length refers to the total length of this FlowSet. Because an individual
-    /// template FlowSet may contain multiple template IDs (as illustrated above),
-    /// the length value should be used to determine the position of the next FlowSet
-    /// record, which could be either a template or a data FlowSet.
-    /// Length is expressed in Type/Length/Value (TLV) format, meaning that the value
-    /// includes the bytes used for the FlowSet ID and the length bytes themselves, as
-    /// well as the combined lengths of all template records included in this FlowSet.
-    pub length: u16,
     /// As a router generates different template FlowSets to match the type of NetFlow
     /// data it will be exporting, each template is given a unique ID. This uniqueness
     /// is local to the router that generated the template ID.
@@ -162,10 +201,8 @@ pub struct Template {
 }
 
 #[derive(Debug, PartialEq, Clone, Serialize, Nom)]
+#[nom(ExtraArgs(flowset_length: u16))]
 pub struct OptionsTemplate {
-    /// This field gives the total length of this FlowSet. Because an individual template FlowSet might contain multiple template IDs, the length value must be used to determine the position of the next FlowSet record, which might be either a template or a data FlowSet.
-    /// Length is expressed in TLV format, meaning that the value includes the bytes used for the FlowSet ID and the length bytes themselves, and the combined lengths of all template records included in this FlowSet.
-    pub length: u16,
     /// As a router generates different template FlowSets to match the type of NetFlow data it is exporting, each template is given a unique ID. This uniqueness is local to the router that generated the template ID. The Template ID is greater than 255. Template IDs inferior to 255 are reserved.
     pub template_id: u16,
     /// This field gives the length in bytes of any scope fields that are contained in this options template.
@@ -181,12 +218,25 @@ pub struct OptionsTemplate {
     /// Padding
     #[nom(
         Map = "|i: &[u8]| i.to_vec()",
-        Take = "(length.saturating_sub(options_scope_length).saturating_sub(options_length).saturating_sub(10)) as usize"
+        Take = "(flowset_length.saturating_sub(options_scope_length).saturating_sub(options_length).saturating_sub(10)) as usize"
     )]
     #[serde(skip_serializing)]
     padding: Vec<u8>,
 }
 
+fn parse_options_template_vec<'a>(
+    i: &'a [u8],
+    flowset_length: u16,
+) -> IResult<&'a [u8], Vec<OptionsTemplate>> {
+    let mut fields = vec![];
+    let mut remaining = i;
+    while let Ok((rem, data)) = OptionsTemplate::parse(remaining, flowset_length) {
+        fields.push(data);
+        remaining = rem;
+    }
+    Ok((remaining, fields))
+}
+
 /// Options Scope Fields
 #[derive(Debug, PartialEq, Clone, Serialize, Nom)]
 pub struct OptionsTemplateScopeField {
@@ -214,10 +264,8 @@ pub struct TemplateField {
 }
 
 #[derive(Debug, PartialEq, Clone, Serialize, Nom)]
-#[nom(ExtraArgs(parser: &mut V9Parser, flow_set_id: u16))]
+#[nom(ExtraArgs(parser: &mut V9Parser, flow_set_id: u16, flowset_length: u16))]
 pub struct OptionsData {
-    // Length
-    pub length: u16,
     // Scope Data
     #[nom(
         Parse = "{ |i| parse_scope_data_fields(i, flow_set_id, &parser.options_templates) }"
@@ -230,7 +278,7 @@ pub struct OptionsData {
     pub options_fields: Vec<OptionDataField>,
     #[nom(
         Map = "|i: &[u8]| i.to_vec()",
-        Take = "get_total_options_length(flow_set_id, length, parser)"
+        Take = "get_total_options_length(flow_set_id, flowset_length, parser)"
     )]
     #[serde(skip_serializing)]
     padding: Vec<u8>,
@@ -319,10 +367,6 @@ pub struct ScopeDataField {
 #[derive(Debug, PartialEq, Clone, Serialize, Nom)]
 #[nom(ExtraArgs(parser: &mut V9Parser, flow_set_id: u16))]
 pub struct Data {
-    /// This field gives the length of the data FlowSet.  Length is expressed in TLV format,
-    /// meaning that the value includes the bytes used for the FlowSet ID and the length bytes
-    /// themselves, as well as the combined lengths of any included data records.
-    pub length: u16,
     // Data Fields
     #[nom(Parse = "{ |i| parse_fields(i, parser.templates.get(&flow_set_id)) }")]
     pub data_fields: Vec<BTreeMap<V9Field, FieldValue>>,