netring 0.29.0

High-performance zero-copy packet I/O for Linux (AF_PACKET TPACKET_V3 + AF_XDP)
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

//! Offline pcap → typed `SessionEvent` stream in one expression.
//!
//! Demonstrates [`AsyncPcapSource::sessions`] (new in netring
//! 0.14.0 / flowscope 0.4). The hand-rolled parser below counts
//! request/response byte volumes per side; the stream yields
//! lifecycle `Started` / `Closed` events plus one `Application`
//! event per packet drained from the reassembler.
//!
//! The end-of-input flush (`finish()` at `Timestamp::MAX`) is
//! folded in — every still-open flow at EOF emits a terminal
//! `SessionEvent::Closed`.
//!
//! Usage:
//!     cargo run -p netring --example async_pcap_sessions \
//!         --features tokio,flow,parse,pcap -- trace.pcap [speed]

use std::env;

use flowscope::{FlowSide, SessionParser, Timestamp};
use futures::StreamExt;
use netring::AsyncPcapSource;
use netring::flow::SessionEvent;
use netring::flow::extract::FiveTuple;

/// Minimal `SessionParser` that emits a `(side, byte_count)` tuple
/// every time it sees bytes. Real parsers do framing here.
#[derive(Default, Clone, Debug)]
struct ByteCounter;

#[derive(Debug)]
struct Counted {
    side: FlowSide,
    bytes: usize,
    ts: Timestamp,
}

impl SessionParser for ByteCounter {
    type Message = Counted;

    fn feed_initiator(&mut self, b: &[u8], ts: Timestamp, out: &mut Vec<Self::Message>) {
        out.push(Counted {
            side: FlowSide::Initiator,
            bytes: b.len(),
            ts,
        });
    }

    fn feed_responder(&mut self, b: &[u8], ts: Timestamp, out: &mut Vec<Self::Message>) {
        out.push(Counted {
            side: FlowSide::Responder,
            bytes: b.len(),
            ts,
        });
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let mut args = env::args().skip(1);
    let path = args
        .next()
        .ok_or("usage: async_pcap_sessions <pcap> [speed]")?;

    eprintln!("[sessions] {path}");

    let source = AsyncPcapSource::open(&path).await?;
    let mut sessions = source.sessions(FiveTuple::bidirectional(), ByteCounter);

    let mut total_init = 0u64;
    let mut total_resp = 0u64;
    let mut messages = 0u64;
    let mut closed = 0u64;

    while let Some(evt) = sessions.next().await {
        match evt? {
            SessionEvent::Started { key, .. } => {
                println!("+ {a} <-> {b}", a = key.a, b = key.b);
            }
            SessionEvent::Application { message, .. } => {
                messages += 1;
                match message.side {
                    FlowSide::Initiator => total_init += message.bytes as u64,
                    FlowSide::Responder => total_resp += message.bytes as u64,
                }
                let _ = message.ts; // available for time-driven correlation
            }
            SessionEvent::Closed { key, reason, .. } => {
                closed += 1;
                println!("- {a} <-> {b} ({reason:?})", a = key.a, b = key.b);
            }
            SessionEvent::FlowAnomaly { kind, .. } => {
                eprintln!("! flow anomaly: {kind:?}");
            }
            SessionEvent::TrackerAnomaly { kind, .. } => {
                eprintln!("! tracker anomaly: {kind:?}");
            }
            _ => {}
        }
    }

    eprintln!(
        "[done] {closed} flows closed, {messages} messages \
         (init={total_init} bytes, resp={total_resp} bytes), \
         packets_read={}",
        sessions.packets_read()
    );
    Ok(())
}