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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160

//! Plan 26 — `MultiStreamConfig` propagation through
//! `AsyncMultiCapture::*_stream_with` constructors.
//!
//! Requires `CAP_NET_RAW` (multi captures use AF_PACKET sockets).

#![cfg(all(
    feature = "integration-tests",
    feature = "tokio",
    feature = "flow",
    feature = "parse"
))]

mod helpers;

use std::time::Duration;

use flowscope::{FlowTrackerConfig, OverflowPolicy, Timestamp};
use netring::flow::extract::{FiveTuple, FiveTupleKey};
use netring::{AsyncMultiCapture, Dedup, MultiStreamConfig};

#[test]
fn empty_config_matches_default_constructor() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        // Two `lo` captures, no config → fall through to defaults.
        let multi = AsyncMultiCapture::open([helpers::LOOPBACK, helpers::LOOPBACK]).unwrap();
        let stream = multi.flow_stream_with(
            FiveTuple::bidirectional(),
            MultiStreamConfig::<FiveTupleKey>::default(),
        );
        // Stats accessors work; alive_sources reflects the two
        // inner captures.
        assert_eq!(stream.alive_sources(), 2);
        let stats = stream.per_source_tracker_stats();
        assert_eq!(stats.len(), 2);
    });
}

#[test]
fn tracker_config_propagates_per_source() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        let mut tc = FlowTrackerConfig::default();
        tc.idle_timeout_tcp = Duration::from_secs(42);
        tc.overflow_policy = OverflowPolicy::DropFlow;

        let multi = AsyncMultiCapture::open([helpers::LOOPBACK]).unwrap();
        let stream = multi.flow_stream_with(
            FiveTuple::bidirectional(),
            MultiStreamConfig::new().with_tracker_config(tc),
        );

        // Stream is alive after construction (we can drop without
        // running it). The integration confirms the chain compiles
        // + builds without panicking.
        assert_eq!(stream.alive_sources(), 1);
    });
}

#[test]
fn dedup_template_clones_per_source() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        let multi = AsyncMultiCapture::open([helpers::LOOPBACK, helpers::LOOPBACK]).unwrap();
        let stream = multi.flow_stream_with(
            FiveTuple::bidirectional(),
            MultiStreamConfig::new().with_dedup(Dedup::loopback()),
        );
        // Compile-only smoke: dedup propagated without panic, and
        // each source got an independent clone.
        assert_eq!(stream.alive_sources(), 2);
    });
}

#[test]
fn idle_timeout_fn_shared_across_sources() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        let multi = AsyncMultiCapture::open([helpers::LOOPBACK]).unwrap();
        let stream = multi.flow_stream_with(
            FiveTuple::bidirectional(),
            MultiStreamConfig::<FiveTupleKey>::new().with_idle_timeout_fn(|k, _l4| {
                if k.either_port(15987) {
                    Some(Duration::from_secs(600))
                } else {
                    None
                }
            }),
        );
        assert_eq!(stream.alive_sources(), 1);
    });
}

#[test]
fn monotonic_timestamps_toggle() {
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        let multi = AsyncMultiCapture::open([helpers::LOOPBACK]).unwrap();
        let stream = multi.flow_stream_with(
            FiveTuple::bidirectional(),
            MultiStreamConfig::<FiveTupleKey>::new().with_monotonic_timestamps(true),
        );
        assert_eq!(stream.alive_sources(), 1);
    });
}

#[test]
fn session_stream_with_compiles_with_full_config() {
    use flowscope::{SessionParser, SessionParserFactory};

    #[derive(Clone, Debug, Default)]
    struct StubFactory;
    #[derive(Clone, Debug)]
    struct StubParser;

    impl SessionParser for StubParser {
        type Message = ();

        fn feed_initiator(&mut self, _: &[u8], _: Timestamp, _out: &mut Vec<Self::Message>) {}
        fn feed_responder(&mut self, _: &[u8], _: Timestamp, _out: &mut Vec<Self::Message>) {}
    }

    impl SessionParserFactory<FiveTupleKey> for StubFactory {
        type Parser = StubParser;
        fn new_parser(&mut self, _key: &FiveTupleKey) -> Self::Parser {
            StubParser
        }
    }

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .unwrap();
    rt.block_on(async {
        let multi = AsyncMultiCapture::open([helpers::LOOPBACK]).unwrap();
        let stream = multi.session_stream_with(
            FiveTuple::bidirectional(),
            StubFactory,
            MultiStreamConfig::<FiveTupleKey>::new()
                .with_dedup(Dedup::loopback())
                .with_monotonic_timestamps(true),
        );
        assert_eq!(stream.alive_sources(), 1);
    });
}