Cleanup, small performance fixes (#75)

* Cleanup, small performance fixes


---------

Co-authored-by: Michael Mileusnich <mmileusnich@gmail.com>

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "netflow_parser"
 description = "Parser for Netflow Cisco V5, V7, V9, IPFIX"
-version = "0.3.6"
+version = "0.4.0"
 edition = "2021"
 author = "michael.mileusnich@gmail.com"
 license = "MIT OR Apache-2.0"

README.md

@@ -11,12 +11,12 @@ See: <https://en.wikipedia.org/wiki/NetFlow>
 ## V5:
 
 ```rust
-use netflow_parser::{NetflowParser, NetflowPacketResult};
+use netflow_parser::{NetflowParser, NetflowPacket};
 
 let v5_packet = [0, 5, 2, 0, 3, 0, 4, 0, 5, 0, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,];
 match NetflowParser::default().parse_bytes(&v5_packet).first() {
-    Some(NetflowPacketResult::V5(v5)) => assert_eq!(v5.header.version, 5),
-    Some(NetflowPacketResult::Error(e)) => println!("{:?}", e),
+    Some(NetflowPacket::V5(v5)) => assert_eq!(v5.header.version, 5),
+    Some(NetflowPacket::Error(e)) => println!("{:?}", e),
     _ => (),
 }
 ```
@@ -40,12 +40,12 @@ println!("{}", json!(NetflowParser::default().parse_bytes(&v5_packet)).to_string
 ## Filtering for a specific version
 
 ```rust
-use netflow_parser::{NetflowParser, NetflowPacketResult};
+use netflow_parser::{NetflowParser, NetflowPacket};
 
 let v5_packet = [0, 5, 2, 0, 3, 0, 4, 0, 5, 0, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,];
 let parsed = NetflowParser::default().parse_bytes(&v5_packet);
 
-let v5_parsed: Vec<NetflowPacketResult> = parsed.iter().filter(|p| p.is_v5()).map(|p| p.clone()).collect();
+let v5_parsed: Vec<NetflowPacket> = parsed.into_iter().filter(|p| p.is_v5()).collect();
 ```
 
 ## Re-Exporting flows
@@ -57,7 +57,7 @@ let packet = [
     4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,
     2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,
 ];
-if let NetflowPacketResult::V5(v5) = NetflowParser::default()
+if let NetflowPacket::V5(v5) = NetflowParser::default()
     .parse_bytes(&packet)
     .first()
     .unwrap()

RELEASES.md

@@ -1,3 +1,8 @@
+# 0.4.0
+ * NetflowPacketResult now simply NetflowPacket.
+ * General parser cleanup and removal of uneeded code.
+ * Small performance optimization in lib parse_bytes.
+
 # 0.3.6
  * Added V9 Post NAT fields 225-228.
  * Added Tokio Async Example

src/lib.rs

@@ -11,12 +11,12 @@
 //! ## V5:
 //!
 //! ```rust
-//! use netflow_parser::{NetflowParser, NetflowPacketResult};
+//! use netflow_parser::{NetflowParser, NetflowPacket};
 //!
 //! let v5_packet = [0, 5, 2, 0, 3, 0, 4, 0, 5, 0, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,];
 //! match NetflowParser::default().parse_bytes(&v5_packet).first() {
-//!     Some(NetflowPacketResult::V5(v5)) => assert_eq!(v5.header.version, 5),
-//!     Some(NetflowPacketResult::Error(e)) => println!("{:?}", e),
+//!     Some(NetflowPacket::V5(v5)) => assert_eq!(v5.header.version, 5),
+//!     Some(NetflowPacket::Error(e)) => println!("{:?}", e),
 //!     _ => (),
 //! }
 //! ```
@@ -40,25 +40,25 @@
 //! ## Filtering for a specific version
 //!
 //! ```rust
-//! use netflow_parser::{NetflowParser, NetflowPacketResult};
+//! use netflow_parser::{NetflowParser, NetflowPacket};
 //!
 //! let v5_packet = [0, 5, 2, 0, 3, 0, 4, 0, 5, 0, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,];
 //! let parsed = NetflowParser::default().parse_bytes(&v5_packet);
 //!
-//! let v5_parsed: Vec<NetflowPacketResult> = parsed.iter().filter(|p| p.is_v5()).map(|p| p.clone()).collect();
+//! let v5_parsed: Vec<NetflowPacket> = parsed.into_iter().filter(|p| p.is_v5()).collect();
 //! ```
 //!
 //! ## Re-Exporting flows
 //! Netflow Parser now supports parsed V5, V7, V9, IPFix can be re-exported back into bytes.
 //! ```rust
-//! use netflow_parser::{NetflowParser, NetflowPacketResult};
+//! use netflow_parser::{NetflowParser, NetflowPacket};
 //!
 //! let packet = [
 //!     0, 5, 0, 1, 3, 0, 4, 0, 5, 0, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3,
 //!     4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1,
 //!     2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7,
 //! ];
-//! if let NetflowPacketResult::V5(v5) = NetflowParser::default()
+//! if let NetflowPacket::V5(v5) = NetflowParser::default()
 //!     .parse_bytes(&packet)
 //!     .first()
 //!     .unwrap()
@@ -97,22 +97,26 @@
 //!
 //! ```cargo run --example netflow_udp_listener_tokio```
 
-mod parser;
 pub mod protocol;
 pub mod static_versions;
 mod tests;
 pub mod variable_versions;
 
-use parser::NetflowParseError;
 use static_versions::{v5::V5, v7::V7};
 use variable_versions::ipfix::{IPFix, IPFixParser};
 use variable_versions::v9::{V9Parser, V9};
 
+use crate::static_versions::v5;
+use crate::static_versions::v7;
+use crate::variable_versions::ipfix;
+use crate::variable_versions::v9;
+
+use nom_derive::{Nom, Parse};
 use serde::Serialize;
 
 /// Enum of supported Netflow Versions
 #[derive(Debug, Clone, Serialize)]
-pub enum NetflowPacketResult {
+pub enum NetflowPacket {
     /// Version 5
     V5(V5),
     /// Version 7
@@ -125,7 +129,7 @@ pub enum NetflowPacketResult {
     Error(NetflowPacketError),
 }
 
-impl NetflowPacketResult {
+impl NetflowPacket {
     pub fn is_v5(&self) -> bool {
         matches!(self, Self::V5(_v))
     }
@@ -143,10 +147,10 @@ impl NetflowPacketResult {
     }
 }
 
-#[derive(Debug, Clone, Serialize)]
-pub struct NetflowPacketError {
-    pub error: NetflowParseError,
-    pub remaining: Vec<u8>,
+#[derive(Nom)]
+/// Generic Netflow Header for shared versions
+struct GenericNetflowHeader {
+    version: u16,
 }
 
 #[derive(Default, Debug)]
@@ -155,6 +159,38 @@ pub struct NetflowParser {
     pub ipfix_parser: IPFixParser,
 }
 
+#[derive(Debug, Clone)]
+pub(crate) struct ParsedNetflow {
+    pub(crate) remaining: Vec<u8>,
+    /// Parsed Netflow Packet
+    pub(crate) result: NetflowPacket,
+}
+
+impl ParsedNetflow {
+    fn new(remaining: &[u8], result: NetflowPacket) -> Self {
+        Self {
+            remaining: remaining.to_vec(),
+            result,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub struct NetflowPacketError {
+    pub error: NetflowParseError,
+    pub remaining: Vec<u8>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+pub enum NetflowParseError {
+    V5(String),
+    V7(String),
+    V9(String),
+    IPFix(String),
+    Incomplete(String),
+    UnknownVersion(Vec<u8>),
+}
+
 impl NetflowParser {
     /// Takes a Netflow packet slice and returns a vector of Parsed Netflows.
     /// If we reach some parse error we return what items be have.
@@ -176,21 +212,45 @@ impl NetflowParser {
     /// ```
     ///
     #[inline]
-    pub fn parse_bytes(&mut self, packet: &[u8]) -> Vec<NetflowPacketResult> {
+    pub fn parse_bytes(&mut self, packet: &[u8]) -> Vec<NetflowPacket> {
         if packet.is_empty() {
             return vec![];
         }
-        self.parse(packet)
+        self.parse_packet_by_version(packet)
             .map(|parsed_netflow| {
-                let mut parsed = vec![parsed_netflow.result];
-                parsed.append(&mut self.parse_bytes(parsed_netflow.remaining.as_slice()));
-                parsed
+                let parsed_result = vec![parsed_netflow.result];
+                if !parsed_netflow.remaining.is_empty() {
+                    let parsed_remaining = self.parse_bytes(&parsed_netflow.remaining);
+                    [parsed_result, parsed_remaining].concat()
+                } else {
+                    parsed_result
+                }
             })
             .unwrap_or_else(|e| {
-                vec![NetflowPacketResult::Error(NetflowPacketError {
+                vec![NetflowPacket::Error(NetflowPacketError {
                     error: e,
                     remaining: packet.to_vec(),
                 })]
             })
     }
+
+    /// Checks the first u16 of the packet to determine the version.  Parses the packet based on the version.
+    /// If the version is unknown it returns an error.  If the packet is incomplete it returns an error.
+    /// If the packet is parsed successfully it returns the parsed Netflow packet and the remaining bytes.
+    fn parse_packet_by_version<'a>(
+        &'a mut self,
+        packet: &'a [u8],
+    ) -> Result<ParsedNetflow, NetflowParseError> {
+        let (packet, version) = GenericNetflowHeader::parse(packet)
+            .map(|(remaining, header)| (remaining, header.version))
+            .map_err(|e| NetflowParseError::Incomplete(e.to_string()))?;
+
+        match version {
+            5 => v5::parse_netflow_v5(packet),
+            7 => v7::parse_netflow_v7(packet),
+            9 => v9::parse_netflow_v9(packet, &mut self.v9_parser),
+            10 => ipfix::parse_netflow_ipfix(packet, &mut self.ipfix_parser),
+            _ => Err(NetflowParseError::UnknownVersion(packet.to_vec())),
+        }
+    }
 }

src/snapshots/netflow_parser__tests__base_tests__it_creates_error.snap

@@ -5,8 +5,6 @@ expression: "NetflowParser::default().parse_bytes(&packet)"
 - Error:
     error:
       UnknownVersion:
-        - 12
-        - 13
         - 14
     remaining:
       - 12

src/snapshots/netflow_parser__tests__base_tests__it_parses_v5_incomplete.snap

@@ -0,0 +1,17 @@
+---
+source: src/tests.rs
+expression: "NetflowParser::default().parse_bytes(&packet)"
+---
+- Error:
+    error:
+      V5: Parsing requires 4 bytes/chars
+    remaining:
+      - 0
+      - 5
+      - 0
+      - 0
+      - 1
+      - 1
+      - 1
+      - 1
+