add support for flow timestamps in IPFIX

src/networking/ipfix/collect.rs

@@ -252,6 +252,7 @@ fn ingest_flow_record(
         record.dst_mac.map(format_mac),
     );
 
+    let timestamps_hint = record.flow_start.zip(record.flow_end);
     let (traffic_direction, service) = modify_or_insert_in_map(
         info_traffic_msg,
         &key,
@@ -263,6 +264,7 @@ fn ingest_flow_record(
         exchanged_packets,
         ip_blacklist,
         record.direction,
+        timestamps_hint,
     );
 
     info_traffic_msg.tot_data_info.add_packets(

src/networking/ipfix/wire.rs

@@ -17,6 +17,7 @@ use nom::number::complete::{be_u8, be_u16, be_u32};
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
 
 use crate::networking::types::traffic_direction::TrafficDirection;
+use crate::utils::types::timestamp::Timestamp;
 
 pub const IPFIX_VERSION: u16 = 0x000A;
 pub const SET_ID_TEMPLATE: u16 = 2;
@@ -107,6 +108,8 @@ pub struct FlowRecord {
     pub src_mac: Option<[u8; 6]>,
     pub dst_mac: Option<[u8; 6]>,
     pub direction: Option<TrafficDirection>,
+    pub flow_start: Option<Timestamp>,
+    pub flow_end: Option<Timestamp>,
 }
 
 /// Parse a complete IPFIX message (header + sets).
@@ -316,10 +319,28 @@ fn apply_ie(ie_id: u16, raw: &[u8], record: &mut FlowRecord) {
                 _ => None,
             };
         }
+        ie::FLOW_START_MILLISECONDS => {
+            record.flow_start = read_timestamp_ms(raw);
+        }
+        ie::FLOW_END_MILLISECONDS => {
+            record.flow_end = read_timestamp_ms(raw);
+        }
         _ => {}
     }
 }
 
+/// IPFIX `dateTimeMilliseconds` is 8 bytes big-endian, ms since UNIX epoch.
+/// Converted to Sniffnet's `Timestamp(secs, usecs)` representation.
+fn read_timestamp_ms(raw: &[u8]) -> Option<Timestamp> {
+    if raw.len() != 8 {
+        return None;
+    }
+    let ms = u64::from_be_bytes(raw.try_into().ok()?);
+    let secs = (ms / 1_000) as i64;
+    let usecs = ((ms % 1_000) * 1_000) as i64;
+    Some(Timestamp::new(secs, usecs))
+}
+
 /// Read a big-endian unsigned integer of 1..=8 bytes into a `u128`.
 fn read_unsigned(raw: &[u8]) -> Option<u128> {
     if raw.is_empty() || raw.len() > 8 {

src/networking/manage_packets.rs

@@ -19,6 +19,7 @@ use crate::networking::types::service::Service;
 use crate::networking::types::service_query::ServiceQuery;
 use crate::networking::types::traffic_direction::TrafficDirection;
 use crate::networking::types::traffic_type::TrafficType;
+use crate::utils::types::timestamp::Timestamp;
 use std::fmt::Write;
 use std::time::Instant;
 
@@ -267,6 +268,7 @@ pub fn modify_or_insert_in_map(
     exchanged_packets: u128,
     ip_blacklist: &IpBlacklist,
     direction_hint: Option<TrafficDirection>,
+    timestamps_hint: Option<(Timestamp, Timestamp)>,
 ) -> (TrafficDirection, Service) {
     let mut traffic_direction = TrafficDirection::default();
     let mut service = Service::Unknown;
@@ -294,14 +296,20 @@ pub fn modify_or_insert_in_map(
         is_blacklisted = ip_blacklist.contains(&address_to_lookup);
     }
 
-    let timestamp = info_traffic_msg.last_packet_timestamp;
+    let receipt_ts = info_traffic_msg.last_packet_timestamp;
+    let (initial_ts, final_ts) = timestamps_hint.unwrap_or((receipt_ts, receipt_ts));
     let new_info = info_traffic_msg
         .map
         .entry(*key)
         .and_modify(|info| {
             info.transmitted_bytes += exchanged_bytes;
             info.transmitted_packets += exchanged_packets;
-            info.final_timestamp = timestamp;
+            if initial_ts < info.initial_timestamp {
+                info.initial_timestamp = initial_ts;
+            }
+            if final_ts > info.final_timestamp {
+                info.final_timestamp = final_ts;
+            }
             info.final_instant = Instant::now();
             if key.protocol.eq(&Protocol::ICMP) {
                 info.icmp_types
@@ -321,8 +329,8 @@ pub fn modify_or_insert_in_map(
             mac_address2: mac_addresses.1,
             transmitted_bytes: exchanged_bytes,
             transmitted_packets: exchanged_packets,
-            initial_timestamp: timestamp,
-            final_timestamp: timestamp,
+            initial_timestamp: initial_ts,
+            final_timestamp: final_ts,
             final_instant: Instant::now(),
             service,
             traffic_direction,

src/networking/parse_packets.rs

@@ -186,6 +186,7 @@ pub fn parse_packets(
                         1,
                         ip_blacklist,
                         None,
+                        None,
                     );
 
                     info_traffic_msg