flowscope 0.26.0

//! Issue #118 — the canonical [`Orientation`] axis on flow events is
//! deterministic, while [`FlowSide`] is arrival-order-relative.
//!
//! This is the regression test for the #71 tap-merge problem: when two
//! capture legs feed one tracker and a scheduling race delivers the
//! *response* packet first, `FlowSide::Initiator` binds to the server
//! on some flows and the client on others. The deterministic,
//! address-sorted `Orientation` does **not** move — a given wire
//! direction always carries the same `Orientation` regardless of which
//! packet of the flow was observed first.
//!
//! Issue #120 — the physical capture leg (`RxMetadata::source_idx`) is
//! folded to a per-canonical-orientation binding on `FlowStats`, so a
//! merged bidirectional flow still reports which NIC each direction
//! arrived on (IPFIX biflow-merge model) without splitting into two
//! flows.

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

use flowscope::{
    FlowEvent, FlowSide, FlowStats, FlowTracker, Orientation, PacketView, Timestamp,
    extract::{FiveTuple, parse::test_frames},
};

// Two endpoints. As `SocketAddr`, A < B, so the bidirectional
// extractor canonicalizes the key to `a = A`, `b = B`. A packet whose
// src→dst is A→B is therefore `Orientation::Forward`; B→A is `Reverse`.
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)
}

/// Drive a tracker over the given frames in order, returning the
/// `(side, orientation)` of every `Started`/`Packet` event, plus the
/// `initiator_orientation` recorded on the flow's stats.
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));
                }
                _ => {}
            }
        }
        // Capture initiator_orientation off a live snapshot.
        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() {
    // Run A — the client's A→B packet is seen first (the normal case).
    let (a_events, a_init) = run(&[a_to_b(b"q1"), b_to_a(b"r1"), a_to_b(b"q2")]);
    // Run B — the server's B→A packet is seen first (tap-merge race).
    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());

    // ── The headline invariant: orientation is a pure function of the
    // wire direction, independent of arrival order. ─────────────────
    // A→B is always Forward and B→A is always Reverse, in both runs.
    // In run A the first event (Started) is the A→B packet → Forward.
    assert_eq!(
        a_events[0].1,
        Orientation::Forward,
        "run A: first packet A→B must be Forward"
    );
    // In run B the first event (Started) is the B→A packet → Reverse.
    assert_eq!(
        b_events[0].1,
        Orientation::Reverse,
        "run B: first packet B→A must be Reverse"
    );

    // ── side flips between runs for the SAME wire direction. ────────
    // Run A: the A→B packet is the initiator (seen first).
    assert_eq!(a_events[0].0, FlowSide::Initiator);
    // Run B: the A→B packet now arrives second, so it is the responder.
    // Find the A→B (Forward) event in run B — it must be Responder.
    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"
    );

    // ── initiator_orientation records which canonical direction the
    // first-seen packet had — and it differs between the two runs. ──
    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;

    // A flow whose initiator went A→B (Forward).
    let mut stats = FlowStats::default();
    stats.initiator_orientation = Orientation::Forward;
    // Forward orientation == initiator side; Reverse == responder.
    assert_eq!(stats.side_for(Orientation::Forward), FlowSide::Initiator);
    assert_eq!(stats.side_for(Orientation::Reverse), FlowSide::Responder);
    // And the inverse mapping round-trips.
    assert_eq!(
        stats.orientation_for(FlowSide::Initiator),
        Orientation::Forward
    );
    assert_eq!(
        stats.orientation_for(FlowSide::Responder),
        Orientation::Reverse
    );

    // A flow whose initiator went B→A (Reverse) — the tap-merge case.
    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);
}

/// Feed `(frame, source_idx)` pairs through one merged bidirectional
/// flow and return the single flow's live `FlowStats`.
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() {
    // A→B (Forward) arrives on NIC 1; B→A (Reverse) on NIC 2. Both
    // legs of the SAME flow — they must merge into one flow while each
    // direction remembers its leg.
    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"
    );
    // Still a single bidirectional flow.
    assert_eq!(stats.total_packets(), 4);
}

#[test]
fn inconsistent_leg_flips_the_ioc() {
    // The A→B direction shows up on TWO different NICs (1 then 3) —
    // tap miswire / asymmetric routing. The first binding is kept, the
    // IOC flips.
    let stats = drive_with_legs(&[
        (a_to_b(b"q1"), 1),
        (b_to_a(b"r1"), 2),
        (a_to_b(b"q2"), 3), // same orientation, different leg
    ]);

    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() {
    // pcap / synthetic: source_idx stays at the `0` "unused" sentinel,
    // so the per-direction bindings stay None and the IOC stays clear.
    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);
}

// ── Issue #122 — SYN-based initiator inference ─────────────────────

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;

/// Drive TCP frames through a tracker built with `infer`, returning the
/// single flow's stats and the `side` carried on its `Started` event.
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() {
    // Tap-merge race: the server's SYN+ACK (B→A) is delivered before
    // the client's SYN (A→B). With inference ON, the SYN sender (A→B,
    // Forward) must still be the initiator.
    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"
    );
    // The first packet (the SYN+ACK) is the responder, so Started says so.
    assert_eq!(started_side, FlowSide::Responder);
}

#[test]
fn syn_ack_first_mislabels_without_inference() {
    // Same race, inference OFF (default): first-seen wins, so the
    // server (B→A, Reverse) is wrongly labelled the initiator. This is
    // the bug #122 fixes — asserted here so a regression is visible.
    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() {
    // SYN seen first (the common case): inference ON changes nothing.
    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() {
    // Mid-stream capture: first packet is a bare ACK+PSH (no SYN). Even
    // with inference ON, fall back to arrival order (can't know better).
    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);
}

// ── issue #121 — opt-in per-packet capture leg on Packet events ─────

/// Drive `(frame, source_idx)` pairs and collect every `Packet`
/// event's `source_idx` field.
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() {
    // Default config: the field stays `None` even though the frames
    // carry real leg info.
    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]);
    // The per-direction #120 binding still works — the knob only
    // gates the per-packet field.
    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() {
    // Two NICs feeding one merged flow: the per-packet field must
    // reproduce the exact leg sequence, including a stray Forward
    // packet that arrived on the "wrong" NIC (asymmetric routing) —
    // precisely what the per-direction #120 summary cannot express.
    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), // wrong-leg packet
        (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"
    );
    // Cross-check against the folded per-direction view: Forward
    // bound to its FIRST leg (1), the stray leg-2 Forward packet
    // flips the inconsistency IOC instead of rebinding.
    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() {
    // Sources without leg info (pcap / synthetic) report the `0`
    // sentinel — never surfaced, even with the knob on.
    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);
}

// ── #217 / #218 — outputs and detectors read flows from the initiator ──

/// A TCP flow `client → server` as frames: SYN, SYN-ACK, ACK, a
/// request, the FIN exchange.
fn tcp_conversation(client: [u8; 4], cport: u16, server: [u8; 4], sport: u16) -> Vec<Vec<u8>> {
    let c = |seq: u32, ack: u32, flags: u8, p: &[u8]| {
        test_frames::ipv4_tcp(
            [0; 6], [0; 6], client, server, cport, sport, seq, ack, flags, p,
        )
    };
    let s = |seq: u32, ack: u32, flags: u8| {
        test_frames::ipv4_tcp(
            [0; 6], [0; 6], server, client, sport, cport, seq, ack, flags, b"",
        )
    };
    const ACK: u8 = 0x10;
    const FIN: u8 = 0x01;
    vec![
        c(100, 0, SYN, b""),
        s(500, 101, SYN_ACK),
        c(101, 501, ACK, b""),
        c(101, 501, ACK | 0x08, b"GET / HTTP/1.1\r\n\r\n"),
        c(119, 501, FIN | ACK, b""),
        s(501, 120, FIN | ACK),
        c(120, 502, ACK, b""),
    ]
}

/// A scanner whose address sorts *above* its targets: every flow key
/// puts the target first. The port-scan detector must name the scanner
/// — before 0.26 it scored each target as the source of the failed
/// connections, and flagged the targets.
#[test]
fn scanner_with_higher_address_is_the_source() {
    use flowscope::detect::patterns::PortScanDetector;
    use flowscope::detect::{DetectorRegistry, SrcHost};
    use flowscope::extract::FiveTupleKey;

    let mut tracker: FlowTracker<_, ()> = FlowTracker::new(FiveTuple::bidirectional());
    let mut registry: DetectorRegistry<FiveTupleKey> = DetectorRegistry::new();
    registry.register(PortScanDetector::<SrcHost>::new());
    let mut anomalies = Vec::new();
    let mut t = 0;
    // 10.0.0.9 probes six ports on each of 10.0.0.1 ..= 10.0.0.5, all
    // unanswered SYNs.
    for target in 1..=5u8 {
        for port in 22..28u16 {
            t += 1;
            let syn = test_frames::ipv4_tcp(
                [0; 6],
                [0; 6],
                [10, 0, 0, 9],
                [10, 0, 0, target],
                40_000 + t as u16,
                port,
                1,
                0,
                SYN,
                b"",
            );
            for ev in tracker.track(PacketView::new(&syn, Timestamp::new(t, 0))) {
                registry.observe(&ev, &mut anomalies);
            }
        }
    }
    for ev in tracker.sweep(Timestamp::MAX) {
        registry.observe(&ev, &mut anomalies);
    }
    let scanners: Vec<_> = anomalies.iter().filter_map(|a| a.src_ip).collect();
    assert!(
        !scanners.is_empty(),
        "30 failed connections from one host is a scan"
    );
    assert!(
        scanners
            .iter()
            .all(|ip| *ip == "10.0.0.9".parse::<std::net::IpAddr>().unwrap()),
        "the scanner is 10.0.0.9: {scanners:?}"
    );
}

/// Zeek's `id.orig_*` is the connection's originator, also when its
/// address sorts above the responder's.
#[cfg(feature = "emit")]
#[test]
fn zeek_orig_is_the_initiator_with_the_higher_address() {
    use flowscope::emit::{ZeekConnLogWriter, ZeekOptions};

    let frames = tcp_conversation([10, 0, 0, 9], 40000, [10, 0, 0, 1], 80);
    let mut tracker: FlowTracker<_, ()> = FlowTracker::new(FiveTuple::bidirectional());
    let mut opts = ZeekOptions::default();
    opts.emit_headers = false;
    let mut w = ZeekConnLogWriter::with_options(Vec::new(), opts).unwrap();
    for (t, f) in frames.iter().enumerate() {
        for ev in tracker.track(PacketView::new(f, Timestamp::new(t as u32 + 1, 0))) {
            w.write_event(&ev).unwrap();
        }
    }
    let out = String::from_utf8(w.finish().unwrap()).unwrap();
    let cols: Vec<&str> = out.trim_end().split('\t').collect();
    assert_eq!(
        (cols[2], cols[3], cols[4], cols[5]),
        ("10.0.0.9", "40000", "10.0.0.1", "80"),
        "id.orig_h/p id.resp_h/p: {out}"
    );
    // orig_pkts orig_ip_bytes resp_pkts resp_ip_bytes follow the same
    // orientation (four 40-byte segments + one carrying 18 bytes).
    assert_eq!(cols[12..], ["5", "218", "2", "80"], "{out}");
}

/// #218: with initiator inference, a flow whose first captured packet
/// is the server's SYN+ACK reports the *client's* orientation on the
/// typed driver's `Started`, the one `FlowStats::initiator_orientation`
/// carries.
#[cfg(all(feature = "reassembler", feature = "session"))]
#[test]
fn started_orientation_is_the_initiators_after_inference() {
    use flowscope::FlowTrackerConfig;
    use flowscope::driver::{Driver, Event};

    // B (server, higher address) answers before A's SYN is seen.
    let frames = [tcp_b_to_a(SYN_ACK, b""), tcp_a_to_b(SYN, b"")];
    let mut cfg = FlowTrackerConfig::default();
    cfg.infer_tcp_initiator = true;
    let mut b = Driver::builder(FiveTuple::bidirectional());
    b.config(cfg);
    let mut d = b.build();
    let mut events = Vec::new();
    for (t, f) in frames.iter().enumerate() {
        d.track_into(
            PacketView::new(f, Timestamp::new(t as u32 + 1, 0)),
            &mut events,
        );
    }
    d.finish_into(&mut events);
    let started = events
        .iter()
        .find_map(|e| match e {
            Event::Started { orientation, .. } => Some(*orientation),
            _ => None,
        })
        .expect("Started");
    let initiator = events
        .iter()
        .find_map(|e| match e {
            Event::Ended { stats, .. } => Some(stats.initiator_orientation),
            _ => None,
        })
        .expect("Ended");
    assert_eq!(initiator, Orientation::Forward, "A sent the SYN");
    assert_eq!(started, initiator, "Started names the initiator");
}