#![cfg(all(feature = "extractors", feature = "tracker"))]
use flowscope::{
FlowEvent, FlowSide, FlowStats, FlowTracker, Orientation, PacketView, Timestamp,
extract::{FiveTuple, parse::test_frames},
};
const A_IP: [u8; 4] = [10, 0, 0, 1];
const B_IP: [u8; 4] = [10, 0, 0, 2];
const A_PORT: u16 = 40000;
const B_PORT: u16 = 53;
fn a_to_b(payload: &[u8]) -> Vec<u8> {
test_frames::ipv4_udp(A_IP, B_IP, A_PORT, B_PORT, payload)
}
fn b_to_a(payload: &[u8]) -> Vec<u8> {
test_frames::ipv4_udp(B_IP, A_IP, B_PORT, A_PORT, payload)
}
fn run(frames: &[Vec<u8>]) -> (Vec<(FlowSide, Orientation)>, Orientation) {
let mut tracker: FlowTracker<_, ()> = FlowTracker::new(FiveTuple::bidirectional());
let mut seen = Vec::new();
let mut init_orientation = None;
for (t, frame) in frames.iter().enumerate() {
let evts = tracker.track(PacketView::new(frame, Timestamp::new(t as u32, 0)));
for evt in evts {
match evt {
FlowEvent::Started {
side, orientation, ..
} => {
seen.push((side, orientation));
}
FlowEvent::Packet {
side, orientation, ..
} => {
seen.push((side, orientation));
}
_ => {}
}
}
if init_orientation.is_none() {
for (_k, entry) in tracker.flows() {
init_orientation = Some(entry.initiator_orientation());
}
}
}
(seen, init_orientation.expect("flow was created"))
}
#[test]
fn orientation_is_stable_across_arrival_order_while_side_flips() {
let (a_events, a_init) = run(&[a_to_b(b"q1"), b_to_a(b"r1"), a_to_b(b"q2")]);
let (b_events, b_init) = run(&[b_to_a(b"r1"), a_to_b(b"q1"), b_to_a(b"r2")]);
assert!(!a_events.is_empty() && !b_events.is_empty());
assert_eq!(
a_events[0].1,
Orientation::Forward,
"run A: first packet A→B must be Forward"
);
assert_eq!(
b_events[0].1,
Orientation::Reverse,
"run B: first packet B→A must be Reverse"
);
assert_eq!(a_events[0].0, FlowSide::Initiator);
let ab_in_run_b = b_events
.iter()
.find(|(_s, o)| *o == Orientation::Forward)
.expect("run B sees at least one A→B packet");
assert_eq!(
ab_in_run_b.0,
FlowSide::Responder,
"run B: the A→B direction is the RESPONDER because the B→A \
packet was seen first — this is exactly the side fragility \
that Orientation fixes"
);
assert_eq!(a_init, Orientation::Forward, "run A initiator was A→B");
assert_eq!(b_init, Orientation::Reverse, "run B initiator was B→A");
assert_ne!(
a_init, b_init,
"the two runs disagree on who the initiator is — but agree on \
every packet's Orientation"
);
}
#[test]
fn flow_stats_translate_between_axes() {
use flowscope::FlowStats;
let mut stats = FlowStats::default();
stats.initiator_orientation = Orientation::Forward;
assert_eq!(stats.side_for(Orientation::Forward), FlowSide::Initiator);
assert_eq!(stats.side_for(Orientation::Reverse), FlowSide::Responder);
assert_eq!(
stats.orientation_for(FlowSide::Initiator),
Orientation::Forward
);
assert_eq!(
stats.orientation_for(FlowSide::Responder),
Orientation::Reverse
);
let mut flipped = FlowStats::default();
flipped.initiator_orientation = Orientation::Reverse;
assert_eq!(flipped.side_for(Orientation::Reverse), FlowSide::Initiator);
assert_eq!(flipped.side_for(Orientation::Forward), FlowSide::Responder);
}
fn drive_with_legs(frames: &[(Vec<u8>, u32)]) -> FlowStats {
let mut tracker: FlowTracker<_, ()> = FlowTracker::new(FiveTuple::bidirectional());
for (t, (frame, src)) in frames.iter().enumerate() {
let view = PacketView::new(frame, Timestamp::new(t as u32, 0)).with_source_idx(*src);
let _ = tracker.track(view);
}
let mut stats = None;
for (_k, entry) in tracker.flows() {
assert!(stats.is_none(), "expected exactly one merged flow");
stats = Some(entry.stats.clone());
}
stats.expect("flow was created")
}
#[test]
fn capture_leg_binds_per_orientation_on_merged_flow() {
let stats = drive_with_legs(&[
(a_to_b(b"q1"), 1),
(b_to_a(b"r1"), 2),
(a_to_b(b"q2"), 1),
(b_to_a(b"r2"), 2),
]);
assert_eq!(stats.source_idx_for(Orientation::Forward), Some(1));
assert_eq!(stats.source_idx_for(Orientation::Reverse), Some(2));
assert_eq!(stats.source_idx_forward, Some(1));
assert_eq!(stats.source_idx_reverse, Some(2));
assert!(
!stats.capture_leg_inconsistent,
"one leg per direction — consistent"
);
assert_eq!(stats.total_packets(), 4);
}
#[test]
fn inconsistent_leg_flips_the_ioc() {
let stats = drive_with_legs(&[
(a_to_b(b"q1"), 1),
(b_to_a(b"r1"), 2),
(a_to_b(b"q2"), 3), ]);
assert_eq!(
stats.source_idx_for(Orientation::Forward),
Some(1),
"original binding is kept, not overwritten"
);
assert_eq!(stats.source_idx_for(Orientation::Reverse), Some(2));
assert!(
stats.capture_leg_inconsistent,
"a second, different leg on the Forward direction is the IOC"
);
}
#[test]
fn no_source_idx_leaves_legs_unbound() {
let stats = drive_with_legs(&[(a_to_b(b"q1"), 0), (b_to_a(b"r1"), 0)]);
assert_eq!(stats.source_idx_for(Orientation::Forward), None);
assert_eq!(stats.source_idx_for(Orientation::Reverse), None);
assert!(!stats.capture_leg_inconsistent);
}
fn tcp_a_to_b(flags: u8, payload: &[u8]) -> Vec<u8> {
test_frames::ipv4_tcp(
[0; 6], [0; 6], A_IP, B_IP, A_PORT, B_PORT, 0, 0, flags, payload,
)
}
fn tcp_b_to_a(flags: u8, payload: &[u8]) -> Vec<u8> {
test_frames::ipv4_tcp(
[0; 6], [0; 6], B_IP, A_IP, B_PORT, A_PORT, 0, 0, flags, payload,
)
}
const SYN: u8 = 0x02;
const SYN_ACK: u8 = 0x12;
fn drive_tcp(infer: bool, frames: &[Vec<u8>]) -> (FlowStats, FlowSide) {
use flowscope::FlowTrackerConfig;
let mut cfg = FlowTrackerConfig::default();
cfg.infer_tcp_initiator = infer;
let mut tracker: FlowTracker<_, ()> = FlowTracker::with_config(FiveTuple::bidirectional(), cfg);
let mut started_side = None;
for (t, frame) in frames.iter().enumerate() {
for ev in tracker.track(PacketView::new(frame, Timestamp::new(t as u32, 0))) {
if let FlowEvent::Started { side, .. } = ev {
started_side = Some(side);
}
}
}
let mut stats = None;
for (_k, entry) in tracker.flows() {
stats = Some(entry.stats.clone());
}
(
stats.expect("flow created"),
started_side.expect("Started emitted"),
)
}
#[test]
fn syn_ack_first_flips_initiator_when_inference_on() {
let (stats, started_side) = drive_tcp(true, &[tcp_b_to_a(SYN_ACK, b""), tcp_a_to_b(SYN, b"")]);
assert_eq!(
stats.initiator_orientation,
Orientation::Forward,
"the SYN sender (A→B = Forward) is the initiator despite arriving second"
);
assert!(stats.direction_flipped, "a SYN+ACK-first flow was flipped");
assert_eq!(
stats.side_for(Orientation::Forward),
FlowSide::Initiator,
"Forward (the SYN direction) maps to Initiator"
);
assert_eq!(started_side, FlowSide::Responder);
}
#[test]
fn syn_ack_first_mislabels_without_inference() {
let (stats, started_side) = drive_tcp(false, &[tcp_b_to_a(SYN_ACK, b""), tcp_a_to_b(SYN, b"")]);
assert_eq!(stats.initiator_orientation, Orientation::Reverse);
assert!(!stats.direction_flipped, "no flip happened");
assert_eq!(started_side, FlowSide::Initiator);
}
#[test]
fn normal_syn_order_is_not_flipped() {
let (stats, started_side) = drive_tcp(true, &[tcp_a_to_b(SYN, b""), tcp_b_to_a(SYN_ACK, b"")]);
assert_eq!(stats.initiator_orientation, Orientation::Forward);
assert!(
!stats.direction_flipped,
"SYN arrived first — no flip needed"
);
assert_eq!(started_side, FlowSide::Initiator);
}
#[test]
fn non_handshake_first_packet_falls_back_to_arrival_order() {
const ACK_PSH: u8 = 0x18;
let (stats, _side) = drive_tcp(
true,
&[tcp_b_to_a(ACK_PSH, b"data"), tcp_a_to_b(ACK_PSH, b"x")],
);
assert_eq!(
stats.initiator_orientation,
Orientation::Reverse,
"first-seen (B→A) wins when no SYN is visible"
);
assert!(!stats.direction_flipped);
}
#[test]
fn orientation_helpers_round_trip() {
assert_eq!(Orientation::Forward.flipped(), Orientation::Reverse);
assert_eq!(Orientation::Reverse.flipped(), Orientation::Forward);
assert_eq!(
Orientation::Forward.flipped().flipped(),
Orientation::Forward
);
assert_eq!(Orientation::Forward.as_str(), "forward");
assert_eq!(Orientation::Reverse.as_str(), "reverse");
assert_eq!(Orientation::default(), Orientation::Forward);
}
fn packet_legs(frames: &[(Vec<u8>, u32)], opt_in: bool) -> (Vec<Option<u32>>, FlowStats) {
use flowscope::FlowTrackerConfig;
let mut config = FlowTrackerConfig::default();
config.emit_packet_source_idx = opt_in;
let mut tracker: FlowTracker<_, ()> =
FlowTracker::with_config(FiveTuple::bidirectional(), config);
let mut legs = Vec::new();
for (t, (frame, src)) in frames.iter().enumerate() {
let mut view = PacketView::new(frame, Timestamp::new(t as u32, 0));
if *src != 0 {
view = view.with_source_idx(*src);
}
for evt in tracker.track(view) {
if let FlowEvent::Packet { source_idx, .. } = evt {
legs.push(source_idx);
}
}
}
let mut stats = None;
for (_k, entry) in tracker.flows() {
assert!(stats.is_none(), "expected exactly one merged flow");
stats = Some(entry.stats.clone());
}
(legs, stats.expect("flow was created"))
}
#[test]
fn per_packet_source_idx_default_off() {
let frames = vec![(a_to_b(b"q1"), 1), (b_to_a(b"r1"), 2), (a_to_b(b"q2"), 1)];
let (legs, stats) = packet_legs(&frames, false);
assert_eq!(legs, vec![None, None, None]);
assert_eq!(stats.source_idx_forward, Some(1));
assert_eq!(stats.source_idx_reverse, Some(2));
}
#[test]
fn per_packet_source_idx_reconstructs_exact_leg_sequence() {
let frames = vec![
(a_to_b(b"q1"), 1),
(b_to_a(b"r1"), 2),
(a_to_b(b"q2"), 1),
(a_to_b(b"q3"), 2), (b_to_a(b"r2"), 2),
];
let (legs, stats) = packet_legs(&frames, true);
assert_eq!(
legs,
vec![Some(1), Some(2), Some(1), Some(2), Some(2)],
"per-packet legs reconstruct the exact sequence"
);
assert_eq!(stats.source_idx_forward, Some(1));
assert_eq!(stats.source_idx_reverse, Some(2));
assert!(stats.capture_leg_inconsistent);
}
#[test]
fn per_packet_source_idx_zero_sentinel_yields_none() {
let frames = vec![(a_to_b(b"q1"), 0), (b_to_a(b"r1"), 0)];
let (legs, stats) = packet_legs(&frames, true);
assert_eq!(legs, vec![None, None]);
assert_eq!(stats.source_idx_forward, None);
assert_eq!(stats.source_idx_reverse, None);
}