flowscope 0.26.0

//! `PcapFlowSource` / `CaptureReader` read classic pcap and pcapng
//! (0.25), applying pcapng per-interface timestamp resolution.

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

use std::borrow::Cow;
use std::io::Cursor;
use std::time::Duration;

use flowscope::extract::parse::test_frames::ipv4_udp;
use flowscope::pcap::{CaptureFormat, CaptureReader, PcapFlowSource};
use flowscope::{FlowSide, FlowState, Timestamp};
use pcap_file::DataLink;
use pcap_file::pcapng::PcapNgWriter;
use pcap_file::pcapng::blocks::enhanced_packet::EnhancedPacketBlock;
use pcap_file::pcapng::blocks::interface_description::{
    InterfaceDescriptionBlock, InterfaceDescriptionOption,
};

/// A pcapng with one interface at `tsresol` and one EPB whose raw
/// timestamp is `raw_ticks`.
fn pcapng(tsresol: Option<u8>, raw_ticks: u64, frame: &[u8]) -> Vec<u8> {
    let mut w = PcapNgWriter::new(Vec::new()).unwrap();
    let options = tsresol
        .map(|v| vec![InterfaceDescriptionOption::IfTsResol(v)])
        .unwrap_or_default();
    w.write_pcapng_block(InterfaceDescriptionBlock {
        linktype: DataLink::ETHERNET,
        snaplen: 65535,
        options,
    })
    .unwrap();
    w.write_pcapng_block(EnhancedPacketBlock {
        interface_id: 0,
        // pcap-file stores the raw tick count in a Duration's nanos.
        timestamp: Duration::from_nanos(raw_ticks),
        original_len: frame.len() as u32,
        data: Cow::Borrowed(frame),
        options: vec![],
    })
    .unwrap();
    w.into_inner()
}

#[test]
fn pcapng_default_resolution_is_microseconds() {
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, b"q");
    // 1_700_000_000.250_000 s in µs.
    let bytes = pcapng(None, 1_700_000_000_250_000, &frame);
    let mut r = CaptureReader::new(Cursor::new(bytes)).unwrap();
    assert_eq!(r.format(), CaptureFormat::PcapNg);
    let p = r.next_packet().unwrap().unwrap();
    assert_eq!(p.timestamp, Duration::new(1_700_000_000, 250_000_000));
    assert_eq!(p.data, frame);
    assert_eq!(p.datalink, DataLink::ETHERNET);
    assert!(r.next_packet().is_none());
}

#[test]
fn pcapng_nanosecond_and_binary_resolutions() {
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, b"q");
    let ns = pcapng(Some(9), 1_700_000_000_000_000_123, &frame);
    let p = CaptureReader::new(Cursor::new(ns))
        .unwrap()
        .next_packet()
        .unwrap()
        .unwrap();
    assert_eq!(p.timestamp, Duration::new(1_700_000_000, 123));

    // 2^-10 s units: 3 * 1024 + 512 ticks = 3.5 s.
    let bin = pcapng(Some(0x80 | 10), 3 * 1024 + 512, &frame);
    let p = CaptureReader::new(Cursor::new(bin))
        .unwrap()
        .next_packet()
        .unwrap()
        .unwrap();
    assert_eq!(p.timestamp, Duration::from_millis(3_500));
}

#[test]
fn pcap_flow_source_replays_pcapng() {
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, b"q");
    let bytes = pcapng(Some(6), 2_000_000, &frame);
    let source = PcapFlowSource::from_reader(Cursor::new(bytes)).unwrap();
    assert_eq!(source.format(), CaptureFormat::PcapNg);
    let views: Vec<_> = source.views().collect::<Result<_, _>>().unwrap();
    assert_eq!(views.len(), 1);
    assert_eq!(views[0].timestamp, Timestamp::new(2, 0));
}

#[test]
fn flow_side_and_state_have_stable_labels() {
    assert_eq!(FlowSide::Initiator.as_str(), "initiator");
    assert_eq!(FlowSide::Responder.to_string(), "responder");
    assert_eq!(FlowSide::Initiator.opposite(), FlowSide::Responder);
    assert_eq!(FlowState::Established.as_str(), "established");
    assert_eq!(FlowState::SynSent.to_string(), "syn_sent");
    assert_eq!(FlowState::ClosingTcp.as_str(), "closing_tcp");
}

// ── #195: link types and direction ───────────────────────────────

/// A classic pcap with `datalink` and the given records.
fn pcap_with(datalink: DataLink, records: &[Vec<u8>]) -> Vec<u8> {
    use pcap_file::pcap::{PcapHeader, PcapPacket, PcapWriter};
    let header = PcapHeader {
        datalink,
        ..Default::default()
    };
    let mut w = PcapWriter::with_header(Vec::new(), header).unwrap();
    for (i, r) in records.iter().enumerate() {
        w.write_packet(&PcapPacket::new(
            Duration::from_secs(1 + i as u64),
            r.len() as u32,
            r,
        ))
        .unwrap();
    }
    w.into_writer()
}

/// IPv4/UDP bytes of an Ethernet frame built by the test helpers.
fn ip_of(frame: &[u8]) -> Vec<u8> {
    frame[14..].to_vec()
}

fn flows_of(bytes: Vec<u8>) -> usize {
    use flowscope::FlowEvent;
    use flowscope::extract::FiveTuple;
    PcapFlowSource::from_reader(Cursor::new(bytes))
        .unwrap()
        .with_extractor(FiveTuple::bidirectional())
        .filter(|e| matches!(e, Ok(FlowEvent::Started { .. })))
        .count()
}

#[test]
fn linux_cooked_capture_is_tracked() {
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, b"q");
    // SLL: outgoing (4), ARPHRD_ETHER, 6-byte address, IPv4.
    let mut sll = vec![0, 4, 0, 1, 0, 6, 2, 2, 2, 2, 2, 2, 0, 0, 0x08, 0x00];
    sll.extend_from_slice(&ip_of(&frame));
    let bytes = pcap_with(DataLink::LINUX_SLL, &[sll]);
    let p = CaptureReader::new(Cursor::new(bytes.clone()))
        .unwrap()
        .next_packet()
        .unwrap()
        .unwrap();
    assert_eq!(
        p.direction,
        Some(flowscope::pcap::CaptureDirection::Outbound)
    );
    assert_eq!(
        flows_of(bytes),
        1,
        "tcpdump -i any captures are no longer unmatched"
    );
}

#[test]
fn raw_ip_capture_is_tracked() {
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, b"q");
    assert_eq!(flows_of(pcap_with(DataLink::RAW, &[ip_of(&frame)])), 1);
}

#[test]
fn unsupported_link_types_are_skipped_and_counted() {
    let bytes = pcap_with(DataLink::IEEE802_11, &[vec![0u8; 40]]);
    let mut views = PcapFlowSource::from_reader(Cursor::new(bytes))
        .unwrap()
        .views();
    assert!(views.next().is_none());
    assert_eq!(views.unsupported(), 1);
}

#[test]
fn pcapng_epb_flags_give_the_direction() {
    use pcap_file::pcapng::blocks::enhanced_packet::EnhancedPacketOption;
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, b"q");
    let mut w = PcapNgWriter::new(Vec::new()).unwrap();
    w.write_pcapng_block(InterfaceDescriptionBlock {
        linktype: DataLink::ETHERNET,
        snaplen: 65535,
        options: vec![],
    })
    .unwrap();
    for flags in [1u32, 2u32] {
        w.write_pcapng_block(EnhancedPacketBlock {
            interface_id: 0,
            timestamp: Duration::from_nanos(1),
            original_len: frame.len() as u32,
            data: Cow::Borrowed(&frame),
            options: vec![EnhancedPacketOption::Flags(flags)],
        })
        .unwrap();
    }
    let dirs: Vec<_> = CaptureReader::new(Cursor::new(w.into_inner()))
        .unwrap()
        .map(|p| p.unwrap().direction)
        .collect();
    use flowscope::pcap::CaptureDirection::{Inbound, Outbound};
    assert_eq!(dirs, vec![Some(Inbound), Some(Outbound)]);
}

/// A classic pcap taken with a snap length (`tcpdump -s 96 -w x.pcap`)
/// stores truncated records whose original length exceeds the header's
/// snaplen. That is the format working as intended, so every record must
/// replay — pcap-file's validating reader rejected each one as "orig_len >
/// snap_len", ending the replay at the first full-size packet.
#[test]
fn truncated_records_of_a_snaplen_capture_replay() {
    use pcap_file::pcap::{PcapHeader, PcapPacket, PcapWriter};
    let mut frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1000, 53, &[0x42; 1400]);
    frame.truncate(frame.len().min(1514));
    let header = PcapHeader {
        snaplen: 96,
        ..Default::default()
    };
    let mut w = PcapWriter::with_header(Vec::new(), header).unwrap();
    for i in 0..3u64 {
        w.write_packet(&PcapPacket::new(
            Duration::new(1 + i, 500_000),
            frame.len() as u32,
            &frame[..96],
        ))
        .unwrap();
    }
    let bytes = w.into_writer();

    let mut r = CaptureReader::new(Cursor::new(bytes.clone())).unwrap();
    let mut n = 0;
    while let Some(p) = r.next_packet() {
        let p = p.expect("a truncated record is valid pcap");
        assert_eq!(p.data.len(), 96);
        assert_eq!(p.original_len as usize, frame.len());
        assert_eq!(p.timestamp, Duration::new(1 + n, 500_000));
        n += 1;
    }
    assert_eq!(n, 3);

    // #212: the truncated records are tracked — one flow, not none.
    assert_eq!(flows_of(bytes.clone()), 1);
    // #219: counted at their wire length, not the 96 bytes captured.
    let stats = PcapFlowSource::from_reader(Cursor::new(bytes.clone()))
        .unwrap()
        .with_extractor(flowscope::extract::FiveTuple::bidirectional())
        .find_map(|e| match e {
            Ok(flowscope::FlowEvent::Ended { stats, .. }) => Some(stats),
            _ => None,
        })
        .expect("the flow ends");
    assert_eq!(stats.packets_initiator, 3);
    assert_eq!(stats.bytes_initiator, 3 * frame.len() as u64);
    assert_eq!(stats.ip_bytes_initiator, 3 * (frame.len() - 14) as u64);
    assert_eq!(stats.packets_truncated_initiator, 3);

    // #219: the replayed views carry what the snap length cut, so the
    // packet's wire length is known.
    let views: Vec<_> = PcapFlowSource::from_reader(Cursor::new(bytes))
        .unwrap()
        .views()
        .collect::<Result<_, _>>()
        .unwrap();
    assert_eq!(views.len(), 3);
    for v in &views {
        assert_eq!(v.frame.len(), 96);
        assert_eq!(v.snap_cut as usize, frame.len() - 96);
        assert_eq!(v.wire_len(), frame.len());
        assert!(v.is_truncated());
        assert_eq!(v.as_view().wire_len(), frame.len());
    }
}

/// #219: a snap-length cut survives link-header normalisation — the
/// cooked header is replaced at the front, the cut is at the end.
#[test]
fn snap_cut_survives_link_normalisation() {
    use pcap_file::pcap::{PcapHeader, PcapPacket, PcapWriter};
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1000, 53, &[0x42; 400]);
    // SLL: outgoing (4), ARPHRD_ETHER, 6-byte address, IPv4.
    let mut sll = vec![0, 4, 0, 1, 0, 6, 2, 2, 2, 2, 2, 2, 0, 0, 0x08, 0x00];
    sll.extend_from_slice(&ip_of(&frame));
    let header = PcapHeader {
        datalink: DataLink::LINUX_SLL,
        snaplen: 80,
        ..Default::default()
    };
    let mut w = PcapWriter::with_header(Vec::new(), header).unwrap();
    w.write_packet(&PcapPacket::new(
        Duration::from_secs(1),
        sll.len() as u32,
        &sll[..80],
    ))
    .unwrap();
    let views: Vec<_> = PcapFlowSource::from_reader(Cursor::new(w.into_writer()))
        .unwrap()
        .views()
        .collect::<Result<_, _>>()
        .unwrap();
    assert_eq!(views.len(), 1);
    // 16-byte cooked header → 14-byte Ethernet header.
    assert_eq!(views[0].frame.len(), 78);
    assert_eq!(views[0].snap_cut as usize, sll.len() - 80);
    // The wire length is the Ethernet frame's: 14 + the IP packet.
    assert_eq!(views[0].wire_len(), frame.len());
}

/// #219: a pcapng EPB never reports fewer bytes on the wire than it
/// captured (an `original_len` of 0 from a broken writer), as the
/// classic-pcap arm already did. An SPB has no captured length: its data
/// is the original length, padded to 32 bits — the padding is not part of
/// the frame.
#[test]
fn pcapng_original_len_is_at_least_the_captured_length() {
    use pcap_file::pcapng::blocks::simple_packet::SimplePacketBlock;
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, b"q");
    let mut w = PcapNgWriter::new(Vec::new()).unwrap();
    w.write_pcapng_block(InterfaceDescriptionBlock {
        linktype: DataLink::ETHERNET,
        snaplen: 65535,
        options: vec![],
    })
    .unwrap();
    w.write_pcapng_block(EnhancedPacketBlock {
        interface_id: 0,
        timestamp: Duration::from_nanos(1),
        original_len: 0,
        data: Cow::Borrowed(&frame),
        options: vec![],
    })
    .unwrap();
    // 43 bytes: written with one byte of padding.
    assert_eq!(frame.len() % 4, 3);
    w.write_pcapng_block(SimplePacketBlock {
        original_len: frame.len() as u32,
        data: Cow::Borrowed(&frame),
    })
    .unwrap();
    let lens: Vec<_> = CaptureReader::new(Cursor::new(w.into_inner()))
        .unwrap()
        .map(|p| {
            let p = p.unwrap();
            (p.original_len as usize, p.data.len())
        })
        .collect();
    assert_eq!(
        lens,
        vec![(frame.len(), frame.len()), (frame.len(), frame.len())]
    );
}

/// An SPB of a capture with a snap length holds min(original length,
/// interface snaplen) bytes, padded to 32 bits: the padding is not frame
/// data, so the cut is reported in full.
#[test]
fn pcapng_spb_cut_by_the_interface_snaplen() {
    use pcap_file::pcapng::blocks::simple_packet::SimplePacketBlock;
    let frame = ipv4_udp([10, 0, 0, 1], [10, 0, 0, 2], 1, 53, &[3; 30]);
    assert_eq!(frame.len(), 72);
    let mut w = PcapNgWriter::new(Vec::new()).unwrap();
    w.write_pcapng_block(InterfaceDescriptionBlock {
        linktype: DataLink::ETHERNET,
        snaplen: 41,
        options: vec![],
    })
    .unwrap();
    w.write_pcapng_block(SimplePacketBlock {
        original_len: frame.len() as u32,
        data: Cow::Borrowed(&frame[..41]), // written with 3 bytes of padding
    })
    .unwrap();
    let bytes = w.into_inner();
    let p = CaptureReader::new(Cursor::new(bytes.clone()))
        .unwrap()
        .next_packet()
        .unwrap()
        .unwrap();
    assert_eq!(p.data, &frame[..41]);
    assert_eq!(p.original_len as usize, frame.len());
    let views: Vec<_> = PcapFlowSource::from_reader(Cursor::new(bytes))
        .unwrap()
        .views()
        .collect::<Result<_, _>>()
        .unwrap();
    assert_eq!(views[0].snap_cut, 31);
}