flowscope 0.24.1

Passive flow & session tracking for packet capture (runtime-free, cross-platform)
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

//! Per-flow resources are released even when `Ended` is suppressed
//! (#185).
//!
//! Cleanup normally rides on `FlowEvent::Ended`. But that event is
//! gated on `EventMask::ENDED` while the tracker reaps the flow
//! either way, so a consumer shedding events under load used to keep
//! every reassembler and parser for the life of the driver — worst
//! precisely when shedding is switched on.

#![cfg(all(
    feature = "tracker",
    feature = "extractors",
    feature = "reassembler",
    feature = "session",
    feature = "test-helpers"
))]

use flowscope::extract::{FiveTuple, parse::test_frames::ipv4_tcp};
use flowscope::{
    BufferedReassemblerFactory, EventMask, FlowDriver, FlowTrackerConfig, PacketView, Timestamp,
};

fn view(frame: &[u8], sec: u32) -> PacketView<'_> {
    PacketView::new(frame, Timestamp::new(sec, 0))
}

/// One data packet, then a RST to end the flow.
fn short_flow(sport: u16, payload: &[u8]) -> Vec<Vec<u8>> {
    let data = ipv4_tcp(
        [0; 6],
        [0; 6],
        [10, 0, 0, 1],
        [10, 0, 0, 2],
        sport,
        80,
        1000,
        0,
        0x18,
        payload,
    );
    let rst = ipv4_tcp(
        [0; 6],
        [0; 6],
        [10, 0, 0, 2],
        [10, 0, 0, 1],
        80,
        sport,
        1,
        0,
        0x04,
        b"",
    );
    vec![data, rst]
}

fn shedding_config() -> FlowTrackerConfig {
    FlowTrackerConfig::default().with_event_filter(EventMask::ENDED)
}

#[test]
fn reassemblers_are_released_when_ended_is_suppressed() {
    let mut d = FlowDriver::with_config(
        FiveTuple::bidirectional(),
        BufferedReassemblerFactory::default(),
        shedding_config(),
    );

    for port in 0..300u16 {
        for f in &short_flow(40000 + port, b"payload-bytes") {
            let _ = d.track(view(f, 0));
        }
    }
    // The tracker reaped every flow despite emitting no `Ended`.
    assert_eq!(d.tracker().flow_count(), 0, "flows were reaped");

    // Before #185 the reassemblers were stranded here, one pair per
    // flow, with their bytes still counted against the memcap pool.
    let _ = d.sweep(Timestamp::new(10, 0));
    assert_eq!(
        d.reassembly_memcap_bytes(),
        0,
        "stranded reassemblers must be released and their bytes refunded"
    );
}

#[test]
fn ordinary_teardown_still_refunds_and_reports() {
    // The reconciliation must not pre-empt normal cleanup: with the
    // filter off, an ended flow is still reported and still refunded.
    let mut d = FlowDriver::new(
        FiveTuple::bidirectional(),
        BufferedReassemblerFactory::default(),
    );
    let mut evs = Vec::new();
    for f in &short_flow(42000, b"payload-bytes") {
        evs.extend(d.track(view(f, 0)));
    }
    evs.extend(d.finish());
    assert!(
        evs.iter()
            .any(|e| matches!(e, flowscope::FlowEvent::Ended { .. })),
        "the flow must still be reported Ended"
    );
    assert_eq!(d.reassembly_memcap_bytes(), 0, "and still refunded");
}