Collect timestamps

Author: joerivanruth

src/event.rs

@@ -1,4 +1,8 @@
-use std::{error::Error, fmt, io};
+use std::{
+    error::Error,
+    fmt, io,
+    time::{Duration, SystemTime, UNIX_EPOCH},
+};
 
 use smallvec::SmallVec;
 
@@ -261,3 +265,14 @@ impl<'a> ConnectionSink<'a> {
             .emit_event(MapiEvent::Oob(self.id(), direction, byte))
     }
 }
+
+/// A timestamp represented as a [Duration] since the
+/// [UNIX_EPOCH][std::time::UNIX_EPOCH].
+pub struct Timestamp(pub Duration);
+
+impl From<SystemTime> for Timestamp {
+    fn from(t: SystemTime) -> Self {
+        let duration = t.duration_since(UNIX_EPOCH).unwrap();
+        Timestamp(duration)
+    }
+}

src/main.rs

@@ -11,12 +11,13 @@ use std::fs::File;
 use std::panic::PanicInfo;
 use std::path::{Path, PathBuf};
 use std::process::ExitCode;
+use std::time::SystemTime;
 use std::{io, panic, process, thread};
 
 use addr::MonetAddr;
 use anyhow::{bail, Context, Result as AResult};
 use argsplitter::{ArgError, ArgSplitter};
-use event::MapiEvent;
+use event::{MapiEvent, Timestamp};
 use pcap::Tracker;
 
 use crate::{proxy::Proxy, render::Renderer};
@@ -134,14 +135,15 @@ fn run_proxy(
 ) -> AResult<()> {
     let (send_events, receive_events) = std::sync::mpsc::sync_channel(500);
     let handler = move |event| {
-        let _ = send_events.send(event);
+        let timestamp = SystemTime::now().into();
+        let _ = send_events.send((timestamp, event));
     };
     let mut proxy = Proxy::new(listen_addr, forward_addr, handler)?;
     install_ctrl_c_handler(proxy.get_shutdown_trigger())?;
     thread::spawn(move || proxy.run().unwrap());
 
-    while let Ok(ev) = receive_events.recv() {
-        mapi_state.handle(&ev, renderer)?;
+    while let Ok((ts, ev)) = receive_events.recv() {
+        mapi_state.handle(&ts, &ev, renderer)?;
     }
     Ok(())
 }
@@ -160,7 +162,7 @@ fn run_pcap(path: &Path, mut mapi_state: mapi::State, renderer: &mut Renderer) -
         owned_file.as_mut().unwrap()
     };
 
-    let handler = |ev: MapiEvent| mapi_state.handle(&ev, renderer);
+    let handler = |ts: &Timestamp, ev: MapiEvent| mapi_state.handle(ts, &ev, renderer);
     let mut tracker = Tracker::new(handler);
     pcap::parse_pcap_file(reader, &mut tracker)
 }

src/mapi/mod.rs

@@ -6,7 +6,7 @@ use std::{
 };
 
 use crate::{
-    event::{ConnectionId, Direction, MapiEvent},
+    event::{ConnectionId, Direction, MapiEvent, Timestamp},
     render::{Renderer, Style},
     Level,
 };
@@ -29,7 +29,12 @@ impl State {
         }
     }
 
-    pub fn handle(&mut self, event: &MapiEvent, renderer: &mut Renderer) -> io::Result<()> {
+    pub fn handle(
+        &mut self,
+        _timestamp: &Timestamp,
+        event: &MapiEvent,
+        renderer: &mut Renderer,
+    ) -> io::Result<()> {
         match event {
             MapiEvent::BoundPort(port) => {
                 renderer.message(None, None, format_args!("LISTEN on port {port}"))?;

src/pcap/mod.rs

@@ -2,21 +2,28 @@ mod mybufread;
 mod tcp;
 mod tracker;
 
-use std::io;
+use std::{
+    io,
+    time::{Duration, SystemTime},
+};
 
 use anyhow::{bail, Result as AResult};
 
 use pcap_file::{
     pcap::PcapReader,
-    pcapng::{Block, PcapNgReader},
+    pcapng::{blocks::interface_description::InterfaceDescriptionOption, Block, PcapNgReader},
     DataLink,
 };
 
+use crate::event::Timestamp;
+
 use self::mybufread::MyBufReader;
 pub use self::tracker::Tracker;
 
 /// Parse PCAP records from the reader and hand the packets to the Tracker. This
 /// function works with both the old-style PCAP and with PCAP-NG file formats.
+///
+/// See also https://www.ietf.org/archive/id/draft-tuexen-opsawg-pcapng-04.html
 pub fn parse_pcap_file(mut rd: impl io::Read, tracker: &mut Tracker) -> AResult<()> {
     // read ahead to inspect the file header
     let mut signature = [0u8; 4];
@@ -52,11 +59,12 @@ fn parse_legacy_pcap(rd: MyBufReader, tracker: &mut Tracker) -> AResult<()> {
 
     while let Some(pkt) = pcap_reader.next_packet() {
         let pkt = pkt?;
+        let timestamp = Timestamp(pkt.timestamp);
         if pkt.data.len() == header.snaplen as usize {
             bail!("truncated packet");
         }
 
-        process_packet(header.datalink, &pkt.data, tracker)?;
+        process_packet(&timestamp, header.datalink, &pkt.data, tracker)?;
     }
 
     Ok(())
@@ -71,22 +79,51 @@ fn parse_pcap_ng(rd: MyBufReader, tracker: &mut Tracker) -> AResult<()> {
     // This mutable holds the latest value we have seen.
     let mut linktype = None;
 
+    // Only used for legacy Block::Packet, completely untested
+    let mut timestamp_resolution = Duration::from_micros(1);
+
+    let mut timestamp: Timestamp = SystemTime::now().into();
     while let Some(block) = pcapng_reader.next_block() {
         let data = match block? {
             Block::InterfaceDescription(iface) => {
                 linktype = Some(iface.linktype);
+                for opt in iface.options {
+                    // This is all completely untested
+                    if let InterfaceDescriptionOption::IfTsResol(reso) = opt {
+                        let base = if reso & 0x80 == 0 { 10u32 } else { 2 };
+                        let divisor = base.pow(reso as u32 & 0x7F);
+                        timestamp_resolution = Duration::from_secs(1) / divisor;
+                    }
+                }
                 continue;
             }
-            Block::Packet(packet) => packet.data,
-            Block::SimplePacket(packet) => packet.data,
-            Block::EnhancedPacket(packet) => packet.data,
+            Block::Packet(packet) => {
+                // This is all completely untested
+                let units = packet.timestamp;
+                // Duration can be multiplied by u32, not by u64.
+                let units_lo = (units & 0xFFFF_FFFF) as u32;
+                let units_hi = (units >> 32) as u32;
+                let duration_lo = timestamp_resolution * units_lo;
+                let duration_hi = timestamp_resolution * units_hi;
+                let duration = duration_hi * 0x1_0000 * 0x1_0000 + duration_lo;
+                timestamp = Timestamp(duration);
+                packet.data
+            }
+            Block::SimplePacket(packet) => {
+                // has no timestamp, keep existing
+                packet.data
+            }
+            Block::EnhancedPacket(packet) => {
+                timestamp = Timestamp(packet.timestamp);
+                packet.data
+            }
             _ => continue,
         };
 
         // Broken files might contain packets before the first interface description block.
         // Ignore them.
         if let Some(lt) = linktype {
-            process_packet(lt, &data, tracker)?;
+            process_packet(&timestamp, lt, &data, tracker)?;
         }
     }
 
@@ -95,11 +132,16 @@ fn parse_pcap_ng(rd: MyBufReader, tracker: &mut Tracker) -> AResult<()> {
 
 /// This function is called from both [parse_legacy_pcap] and [parse_pcap_ng]
 /// for each packet in the file.
-fn process_packet(linktype: DataLink, data: &[u8], tracker: &mut Tracker) -> AResult<()> {
+fn process_packet(
+    timestamp: &Timestamp,
+    linktype: DataLink,
+    data: &[u8],
+    tracker: &mut Tracker,
+) -> AResult<()> {
     // We expect to read ethernet frames but it's also possible for pcap files to
     // capture at the IP level. Right now we only support Ethernet.
     match linktype {
-        DataLink::ETHERNET => tracker.process_ethernet(data),
+        DataLink::ETHERNET => tracker.process_ethernet(timestamp, data),
         _ => bail!("pcap file contains packet of type {linktype:?}, this is not supported"),
     }
 }

src/pcap/tcp.rs

@@ -7,9 +7,9 @@ use std::{
 
 use etherparse::TcpSlice;
 
-use crate::event::{ConnectionId, Direction, MapiEvent};
+use crate::event::{ConnectionId, Direction, MapiEvent, Timestamp};
 
-type Handler<'a> = dyn FnMut(MapiEvent) -> io::Result<()> + 'a;
+type Handler<'a> = dyn for<'t> FnMut(&'t Timestamp, MapiEvent) -> io::Result<()> + 'a;
 
 /// TCP connection state is identified by (src_ip,src_port, dest_ip,dest_port) tuples.
 /// This struct represents those.
@@ -52,6 +52,7 @@ impl TcpTracker {
     /// Handle a TCP packet.
     pub fn handle(
         &mut self,
+        timestamp: &Timestamp,
         src_addr: IpAddr,
         dest_addr: IpAddr,
         tcp: &TcpSlice,
@@ -63,13 +64,19 @@ impl TcpTracker {
         };
 
         match (tcp.syn(), tcp.ack()) {
-            (true, false) => self.handle_syn(key, tcp, handler),
-            (true, true) => self.handle_syn_ack(key, tcp, handler),
-            _ => self.handle_existing(key, tcp, handler),
+            (true, false) => self.handle_syn(timestamp, key, tcp, handler),
+            (true, true) => self.handle_syn_ack(timestamp, key, tcp, handler),
+            _ => self.handle_existing(timestamp, key, tcp, handler),
         }
     }
 
-    fn handle_syn(&mut self, key: Key, tcp: &TcpSlice, handler: &mut Handler) -> io::Result<()> {
+    fn handle_syn(
+        &mut self,
+        timestamp: &Timestamp,
+        key: Key,
+        tcp: &TcpSlice,
+        handler: &mut Handler,
+    ) -> io::Result<()> {
         let flipped = key.flip();
         if self.streams.contains_key(&key) || self.streams.contains_key(&flipped) {
             return Ok(());
@@ -85,14 +92,15 @@ impl TcpTracker {
             local: key.dest.into(),
             peer: key.src.into(),
         };
-        handler(ev)?;
+        handler(timestamp, ev)?;
 
         self.streams.insert(key, upstream);
         Ok(())
     }
 
     fn handle_syn_ack(
         &mut self,
+        timestamp: &Timestamp,
         key: Key,
         tcp: &TcpSlice,
         handler: &mut Handler,
@@ -111,14 +119,15 @@ impl TcpTracker {
             id,
             peer: key.src.into(),
         };
-        handler(ev)?;
+        handler(timestamp, ev)?;
 
         self.streams.insert(key, downstream);
         Ok(())
     }
 
     fn handle_existing(
         &mut self,
+        timestamp: &Timestamp,
         key: Key,
         tcp: &TcpSlice,
         handler: &mut Handler,
@@ -139,13 +148,13 @@ impl TcpTracker {
         let Some(payload) = stream.reorder(seqno, tcp.fin(), payload) else {
             return Ok(());
         };
-        Self::emit_data(id, direction, payload, handler)?;
+        Self::emit_data(timestamp, id, direction, payload, handler)?;
 
         // If stream.reorder above returned this packet, it means it was exactly
         // the packet we needed right now. Packets do not always arrive in-order
         // so it's possible that the next packet is already in our cache.
         while let Some(payload) = stream.next_ready() {
-            Self::emit_data(id, direction, &payload, handler)?;
+            Self::emit_data(timestamp, id, direction, &payload, handler)?;
         }
 
         // Stream.finished is set by stream.reorder and stream.next_ready.
@@ -157,20 +166,21 @@ impl TcpTracker {
         // Report this and drop all state if the other direction has also finished.
 
         let ev = MapiEvent::ShutdownRead { id, direction };
-        handler(ev)?;
+        handler(timestamp, ev)?;
 
         let flipped = key.flip();
         if let Some(StreamState { finished: true, .. }) = self.streams.get(&flipped) {
             self.streams.remove(&key);
             self.streams.remove(&flipped);
             let ev = MapiEvent::End { id };
-            handler(ev)?;
+            handler(timestamp, ev)?;
         }
 
         Ok(())
     }
 
     fn emit_data(
+        timestamp: &Timestamp,
         id: ConnectionId,
         direction: Direction,
         payload: &[u8],
@@ -182,7 +192,7 @@ impl TcpTracker {
                 direction,
                 data: payload.into(),
             };
-            handler(ev)?;
+            handler(timestamp, ev)?;
         }
         Ok(())
     }