From 8b111ddfdcf5ec7aefcf37d1039c87d1dc17e70f Mon Sep 17 00:00:00 2001 From: Eric Harding Date: Tue, 1 Sep 2026 08:38:44 -0400 Subject: [PATCH] allow inflight filter --- src/chunker.rs | 395 ++++++++++++++++++++++++++++++++++++++++++------ src/cli.rs | 25 +++ src/client.rs | 33 ++++ src/filter.rs | 317 ++++++++++++++++++++++++++++++++++++++ src/main.rs | 1 + src/naming.rs | 36 +++++ src/pipeline.rs | 173 ++++++++++++++++++++- src/server.rs | 8 + src/tail.rs | 23 ++- 9 files changed, 956 insertions(+), 55 deletions(-) create mode 100644 src/filter.rs diff --git a/src/chunker.rs b/src/chunker.rs index e65e913..8b19efc 100644 --- a/src/chunker.rs +++ b/src/chunker.rs @@ -1,3 +1,4 @@ +use crate::filter::{IpFilter, LINKTYPE_ETHERNET}; use anyhow::{Result, bail}; use std::time::{Duration, Instant}; @@ -146,10 +147,37 @@ pub struct RecordAlignedChunker { header_ready: bool, endianness: Option, snaplen: u32, + /// Link-layer type from global-header bytes [20..24], needed to know + /// where the IP header starts. Only meaningful once `header_done`. + linktype: u32, + /// Source bytes read but not yet confirmed as complete records. Every + /// record that `rescan` confirms is consumed out of here immediately, + /// so this holds only the unconfirmed tail -- bounded by one read + /// window, no matter how little the filter keeps. pending: Vec, - complete_len: usize, - emitted_len: u64, - last_growth: Instant, + /// Cumulative source bytes consumed (confirmed and removed from + /// `pending`), NOT bytes emitted downstream -- the two differ whenever + /// the filter drops a record. Source-offset accounting depends only on + /// this. + consumed_len: u64, + /// Confirmed records the filter kept, accumulating across many + /// `pending` recycles until they reach `target`. This is what gets + /// emitted, so chunk (and therefore zstd frame) size is measured in + /// KEPT bytes -- a filter keeping 1% of the traffic still produces + /// full-size frames rather than 1%-size ones. + kept: Vec, + filter: IpFilter, + /// When the oldest byte currently in `kept` was confirmed, i.e. how + /// long the pending output has been waiting. `None` when `kept` is + /// empty. + /// + /// This deliberately tracks the age of the OUTPUT, not of the last + /// source growth: with a selective filter the source can grow + /// continuously (resetting any source-based timer forever) while the + /// handful of matching records sit here unsent. Timing from the kept + /// bytes is what keeps `flush_after` an actual latency bound rather + /// than an idle-only one. + kept_since: Option, target: usize, flush_after: Duration, /// Whether this source is a still-growing (partial) file. Gates every @@ -161,17 +189,34 @@ pub struct RecordAlignedChunker { } impl RecordAlignedChunker { + /// Unfiltered: keeps every record. Equivalent to `with_filter` with + /// `IpFilter::match_all()` -- which is not a separate code path, just + /// a filter that says yes, so this and the filtered case exercise the + /// same record-confirmation logic. Production goes through + /// `with_filter`; this is the convenience the tests use. + #[allow(dead_code)] pub fn new(target: usize, flush_after: Duration, partial: bool) -> Self { + Self::with_filter(target, flush_after, partial, IpFilter::match_all()) + } + + pub fn with_filter( + target: usize, + flush_after: Duration, + partial: bool, + filter: IpFilter, + ) -> Self { Self { header_buf: Vec::with_capacity(GLOBAL_HEADER_LEN), header_done: false, header_ready: false, endianness: None, snaplen: 0, + linktype: 0, pending: Vec::new(), - complete_len: 0, - emitted_len: 0, - last_growth: Instant::now(), + consumed_len: 0, + kept: Vec::new(), + filter, + kept_since: None, target, flush_after, partial, @@ -191,19 +236,49 @@ impl RecordAlignedChunker { } let e = detect_endianness(&self.header_buf)?; self.snaplen = read_u32(&self.header_buf[16..20], e); + self.linktype = read_u32(&self.header_buf[20..24], e); self.endianness = Some(e); self.header_done = true; self.header_ready = true; + // A filter can only find addresses in an Ethernet frame, so on any + // other link type it keeps nothing and the transfer silently + // produces a header-only pcap. `tcpdump -i any` (LINUX_SLL, 113) is + // the easy way to hit this, so say so out loud. stderr, never + // stdout: in --server mode stdout is the wire. + if !self.filter.is_match_all() && self.linktype != LINKTYPE_ETHERNET { + eprintln!( + "scpcap: warning: --ip filtering needs Ethernet (link type {LINKTYPE_ETHERNET}), \ + but this capture is link type {}; no packets will match", + self.linktype + ); + } Ok(()) } + /// Walks `pending` confirming whole records, hands each confirmed + /// record to the filter (keeping the matches in `kept`), and consumes + /// the confirmed bytes out of `pending`. + /// + /// Confirming/consuming and *emitting* are deliberately separate. If + /// consumption were tied to emission, a filter keeping 1% of the + /// traffic would hold the other 99% in `pending` for as long as `kept` + /// stayed under `target` -- the whole un-emitted source accumulating in + /// memory, gigabytes on a large capture. Consuming here caps `pending` + /// at the unconfirmed tail while `kept` fills up across many passes. + /// + /// `cursor` is deliberately a local, not a field: every confirmed byte + /// leaves `pending` before this returns, so there is no persistent + /// "confirmed but still buffered" region for the source-offset + /// accounting (or `refresh_tail`) to have to reason about. fn rescan(&mut self, trust_trailing_record: bool) { let e = match self.endianness { Some(e) => e, None => return, }; + let max_incl_len = self.max_incl_len(); + let mut cursor = 0usize; loop { - let remaining = &self.pending[self.complete_len..]; + let remaining = &self.pending[cursor..]; if remaining.len() < RECORD_HEADER_LEN { break; } @@ -224,7 +299,7 @@ impl RecordAlignedChunker { if ts_sec == 0 { break; } - if incl_len > self.max_incl_len() { + if incl_len > max_incl_len { break; } @@ -235,13 +310,13 @@ impl RecordAlignedChunker { // Require positive proof the writer has moved past this // record -- the start of the NEXT record's header, itself // looking real -- before confirming it. This makes - // complete_len always sit (at least) one record behind + // confirmed_len() always sit (at least) one record behind // whatever's currently visible, by construction, for every // record, not just "the last one" as a special case. let after = &remaining[record_total..]; let next_is_proven = after.len() >= RECORD_HEADER_LEN && read_u32(&after[0..4], e) != 0 - && (read_u32(&after[8..12], e) as usize) <= self.max_incl_len(); + && (read_u32(&after[8..12], e) as usize) <= max_incl_len; // Escape hatch (only when the caller is certain nothing // more will EVER arrive): a record that is itself plausible @@ -259,15 +334,30 @@ impl RecordAlignedChunker { break; } } - self.complete_len += record_total; + + // Confirmed. Show the captured bytes (everything after the + // record header) to the filter, and keep the whole record -- + // header included -- if it matches, so `kept` stays a valid + // sequence of pcap records. + let start = cursor; + let end = cursor + record_total; + if self.filter.matches(&self.pending[start + RECORD_HEADER_LEN..end], self.linktype) { + self.kept.extend_from_slice(&self.pending[start..end]); + } + cursor = end; + } + if cursor > 0 { + self.pending.drain(..cursor); + self.consumed_len += cursor as u64; + } + if !self.kept.is_empty() && self.kept_since.is_none() { + self.kept_since = Some(Instant::now()); } } - fn drain_complete(&mut self) -> (ChunkKind, Vec) { - let chunk: Vec = self.pending.drain(..self.complete_len).collect(); - self.emitted_len += self.complete_len as u64; - self.complete_len = 0; - (ChunkKind::Data, chunk) + fn take_kept(&mut self) -> (ChunkKind, Vec) { + self.kept_since = None; + (ChunkKind::Data, std::mem::take(&mut self.kept)) } } @@ -296,7 +386,6 @@ impl Chunker for RecordAlignedChunker { if !rest.is_empty() { self.pending.extend_from_slice(rest); self.rescan(false); - self.last_growth = Instant::now(); } Ok(()) } @@ -306,19 +395,27 @@ impl Chunker for RecordAlignedChunker { self.header_ready = false; return Some((ChunkKind::Header, std::mem::take(&mut self.header_buf))); } - if self.complete_len >= self.target { - return Some(self.drain_complete()); + // All three cut points gate on `kept`, never on how much source was + // consumed: a stretch of source where the filter matched nothing + // must produce no chunk at all, rather than an empty one. An empty + // chunk would become a zero-content zstd frame, which + // STREAMING_ZSTD_FORMAT.md §3 does not allow for. + if self.kept.len() >= self.target { + return Some(self.take_kept()); } None } fn should_flush(&self, now: Instant) -> bool { - self.complete_len > 0 && now.duration_since(self.last_growth) >= self.flush_after + // `kept_since` is Some exactly when `kept` is non-empty, so this + // can never ask for a flush that would produce an empty chunk. + self.kept_since + .is_some_and(|since| now.duration_since(since) >= self.flush_after) } fn flush(&mut self) -> Option<(ChunkKind, Vec)> { - if self.complete_len > 0 { - Some(self.drain_complete()) + if !self.kept.is_empty() { + Some(self.take_kept()) } else { None } @@ -328,8 +425,8 @@ impl Chunker for RecordAlignedChunker { if !self.header_done { bail!("source ended before a complete pcap global header was read"); } - if self.complete_len > 0 { - Ok(Some(self.drain_complete())) + if !self.kept.is_empty() { + Ok(Some(self.take_kept())) } else { Ok(None) } @@ -339,7 +436,7 @@ impl Chunker for RecordAlignedChunker { if !self.header_done { self.header_buf.len() as u64 } else { - GLOBAL_HEADER_LEN as u64 + self.emitted_len + self.pending.len() as u64 + GLOBAL_HEADER_LEN as u64 + self.consumed_len + self.pending.len() as u64 } } @@ -352,7 +449,14 @@ impl Chunker for RecordAlignedChunker { // treating the stuck zero bytes as already consumed. 0 } else { - GLOBAL_HEADER_LEN as u64 + self.emitted_len + self.complete_len as u64 + // Source-offset accounting, entirely independent of what the + // filter kept: every confirmed record is consumed the moment + // `rescan` proves it, so `consumed_len` alone is the confirmed + // prefix. (This is the same number the old + // `emitted_len + complete_len` produced; consuming on + // confirmation just moves the quantity from the second term to + // the first.) + GLOBAL_HEADER_LEN as u64 + self.consumed_len } } @@ -373,30 +477,24 @@ impl Chunker for RecordAlignedChunker { if !self.header_done { return Ok(()); } - // header_done just flipped true, so complete_len is still 0 here - // (nothing beyond the header has been considered yet) -- - // pending is simply the fresh post-header bytes. + // header_done just flipped true, so nothing beyond the header + // has been considered yet -- pending is simply the fresh + // post-header bytes. self.pending = bytes[GLOBAL_HEADER_LEN..].to_vec(); self.rescan(trust_trailing_record); - self.last_growth = Instant::now(); return Ok(()); } - // `pending[..complete_len]` is CONFIRMED content that just hasn't - // been drained (emitted) yet -- e.g. because it's still under - // `target`. `bytes` starts exactly at `confirmed_len()` (checked - // above), i.e. exactly where that confirmed prefix ends, so it must - // be appended after it, never used to replace the whole buffer -- - // doing that would silently discard already-confirmed, already - // real, not-yet-emitted content on every single re-peek, which is - // catastrophic for a long-running growing transfer (each poll would - // re-derive from scratch instead of building on prior confirmed - // progress, and once the fresh read window is too small to - // re-establish the same progress alone -- e.g. capped by - // READ_WINDOW -- confirmed_len can never recover). - self.pending.truncate(self.complete_len); + // `pending` holds only the UNCONFIRMED tail -- `rescan` consumes + // every record it confirms, so nothing confirmed is still sitting + // here waiting to be emitted. `bytes` is a fresh read starting at + // `confirmed_len()` (checked above), i.e. exactly where that + // unconfirmed tail begins, so replacing the buffer wholesale is + // correct and discards nothing: anything already confirmed lives in + // `consumed_len` (for source accounting) and, if the filter kept + // it, in `kept` (for emission). Neither is touched here. + self.pending.clear(); self.pending.extend_from_slice(bytes); self.rescan(trust_trailing_record); - self.last_growth = Instant::now(); Ok(()) } } @@ -771,4 +869,213 @@ mod tests { "confirmed_len must never regress below what was already established" ); } + + // --- IP filtering ------------------------------------------------- + // + // Note that every test above this point already exercises the + // filtering code path, with a match-all filter: there is no separate + // unfiltered implementation to regress against. + + /// A global header declaring Ethernet (link type 1) at bytes [20..24], + /// which the filter needs in order to find an IP header at all. + fn eth_header_bytes() -> Vec { + let mut h = header_bytes(0xA1B2C3D4); + h[20..24].copy_from_slice(&1u32.to_le_bytes()); + h + } + + /// A pcap record holding an Ethernet/IPv4 packet between `src` and `dst`, + /// padded out to `payload_len` total captured bytes. + fn ip_record(src: [u8; 4], dst: [u8; 4], payload_len: usize) -> Vec { + let mut pkt = vec![0u8; payload_len.max(34)]; + pkt[12..14].copy_from_slice(&0x0800u16.to_be_bytes()); // ethertype IPv4 + pkt[14] = 0x45; // version 4, IHL 5 + pkt[26..30].copy_from_slice(&src); + pkt[30..34].copy_from_slice(&dst); + record_bytes_with_ts(&pkt, false, 1_700_000_000) + } + + fn filter_for(specs: &[&str]) -> IpFilter { + IpFilter::parse(&specs.iter().map(|s| s.to_string()).collect::>()).unwrap() + } + + fn drain_all(c: &mut RecordAlignedChunker) -> Vec<(ChunkKind, Vec)> { + let mut out = Vec::new(); + while let Some(chunk) = c.ready_chunk() { + out.push(chunk); + } + out + } + + const WANTED: [u8; 4] = [1, 1, 1, 1]; + const OTHER: [u8; 4] = [9, 9, 9, 9]; + const THIRD: [u8; 4] = [8, 8, 8, 8]; + + #[test] + fn keeps_only_matching_records_and_keeps_them_whole() { + let mut c = RecordAlignedChunker::with_filter(1, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"])); + let hit_src = ip_record(WANTED, OTHER, 60); + let hit_dst = ip_record(OTHER, WANTED, 60); + let miss = ip_record(OTHER, THIRD, 60); + + let mut stream = eth_header_bytes(); + stream.extend_from_slice(&miss); + stream.extend_from_slice(&hit_src); + stream.extend_from_slice(&miss); + stream.extend_from_slice(&hit_dst); + stream.extend_from_slice(&miss); + c.feed(&stream).unwrap(); + + let chunks = drain_all(&mut c); + assert_eq!(chunks[0].0, ChunkKind::Header); + assert_eq!(chunks[0].1, eth_header_bytes(), "header is never filtered"); + + // Records come out whole (16-byte header included) and in order. + let data: Vec = chunks[1..].iter().flat_map(|(_, d)| d.clone()).collect(); + let mut expected = hit_src.clone(); + expected.extend_from_slice(&hit_dst); + assert_eq!(data, expected); + } + + #[test] + fn target_counts_kept_bytes_not_source_bytes() { + // The reason filtering lives at record confirmation rather than in a + // post-pass over emitted chunks: with 1% of traffic kept, chunks + // (and therefore zstd frames) must still reach `target`, not shrink + // to 1% of it. + let hit = ip_record(WANTED, OTHER, 100); + let miss = ip_record(OTHER, THIRD, 100); + let target = 10 * hit.len(); + let mut c = RecordAlignedChunker::with_filter(target, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"])); + + c.feed(ð_header_bytes()).unwrap(); + assert_eq!(c.ready_chunk().unwrap().0, ChunkKind::Header); + // 1 kept in every 100 records, 1000 records total => 10 kept. + for i in 0..1000 { + c.feed(if i % 100 == 0 { &hit } else { &miss }).unwrap(); + } + + let chunks = drain_all(&mut c); + assert_eq!(chunks.len(), 1, "expected one full-size chunk, got {}", chunks.len()); + assert!( + chunks[0].1.len() >= target, + "chunk is {} bytes, below the {target}-byte target", + chunks[0].1.len() + ); + assert_eq!(chunks[0].1.len(), 10 * hit.len()); + } + + #[test] + fn source_buffer_does_not_grow_when_nothing_matches() { + // Confirmed records are consumed from `pending` immediately rather + // than when a chunk is emitted. Without that, a filter that keeps + // nothing would accumulate the entire source in memory while + // waiting for `kept` to reach `target`. + let miss = ip_record(OTHER, THIRD, 100); + let mut c = RecordAlignedChunker::with_filter( + DEFAULT_TARGET_BYTES, + DEFAULT_FLUSH, + false, + filter_for(&["1.1.1.1"]), + ); + c.feed(ð_header_bytes()).unwrap(); + let mut source_len = GLOBAL_HEADER_LEN as u64; + for _ in 0..5000 { + c.feed(&miss).unwrap(); + source_len += miss.len() as u64; + } + + assert!( + c.pending.len() < miss.len() * 2, + "unconfirmed tail grew to {} bytes; confirmed records are not being consumed", + c.pending.len() + ); + assert!(c.kept.is_empty()); + // ...and the source is nonetheless fully accounted for. + assert_eq!(c.confirmed_len(), source_len); + assert_eq!(c.read_from(), source_len); + } + + #[test] + fn a_fully_dropped_region_yields_no_chunk_not_an_empty_one() { + // An empty chunk would be compressed into a zero-content zstd frame, + // which STREAMING_ZSTD_FORMAT.md §3 does not provide for. + let miss = ip_record(OTHER, THIRD, 100); + let mut c = RecordAlignedChunker::with_filter(1, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"])); + c.feed(ð_header_bytes()).unwrap(); + assert_eq!(c.ready_chunk().unwrap().0, ChunkKind::Header); + for _ in 0..20 { + c.feed(&miss).unwrap(); + } + // target is 1 byte, so anything kept at all would be emitted. + assert!(c.ready_chunk().is_none()); + assert!(c.flush().is_none()); + assert!(!c.should_flush(Instant::now() + Duration::from_secs(60))); + assert!(c.finish().unwrap().is_none()); + } + + #[test] + fn source_accounting_is_unaffected_by_what_the_filter_keeps() { + // The direct statement of the invariant the rest of the system + // depends on: confirmed_len()/read_from() are SOURCE offsets, which + // tail.rs compares against the real file length to decide finalize. + // Dropping packets must not move them by a single byte. + let mut stream = eth_header_bytes(); + for i in 0..40u8 { + let src = if i % 3 == 0 { WANTED } else { OTHER }; + stream.extend_from_slice(&ip_record(src, THIRD, 60 + i as usize)); + } + + let mut all = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, false, IpFilter::match_all()); + let mut some = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, false, filter_for(&["1.1.1.1"])); + let mut none = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, false, filter_for(&["203.0.113.7"])); + for c in [&mut all, &mut some, &mut none] { + c.feed(&stream).unwrap(); + } + + for c in [&some, &none] { + assert_eq!(c.confirmed_len(), all.confirmed_len()); + assert_eq!(c.read_from(), all.read_from()); + } + assert_eq!(all.confirmed_len(), stream.len() as u64); + + // Meanwhile the kept content really does differ. + assert_eq!(all.kept.len(), stream.len() - GLOBAL_HEADER_LEN); + assert!(!some.kept.is_empty() && some.kept.len() < all.kept.len()); + assert!(none.kept.is_empty()); + } + + #[test] + fn filtering_a_growing_source_confirms_and_drops_across_refreshes() { + // The partial path reaches records through refresh_tail, which + // replaces the unconfirmed tail wholesale. A record must be filtered + // exactly once even though the bytes around it are re-read. + let hit = ip_record(WANTED, OTHER, 60); + let miss = ip_record(OTHER, THIRD, 60); + let mut c = RecordAlignedChunker::with_filter(1 << 20, DEFAULT_FLUSH, true, filter_for(&["1.1.1.1"])); + + let mut stream = eth_header_bytes(); + stream.extend_from_slice(&miss); + stream.extend_from_slice(&hit); + stream.extend_from_slice(&miss); + stream.extend_from_slice(&hit); + + // Feed it in growing prefixes, each time re-reading from the current + // confirmed offset, exactly as tail.rs does. + for end in [40, 100, stream.len() - 10, stream.len()] { + let offset = c.confirmed_len() as usize; + c.refresh_tail(offset as u64, &stream[offset..end], false).unwrap(); + } + // The trailing record has no next-record proof yet, so only the + // first `hit` is confirmed. + assert_eq!(c.kept, hit); + + // Source stops growing and the caller vouches for the tail. + let offset = c.confirmed_len() as usize; + c.refresh_tail(offset as u64, &stream[offset..], true).unwrap(); + assert_eq!(c.confirmed_len(), stream.len() as u64); + let mut both = hit.clone(); + both.extend_from_slice(&hit); + assert_eq!(c.kept, both, "each kept record appears exactly once"); + } } diff --git a/src/cli.rs b/src/cli.rs index 90d1410..3be61ac 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -33,6 +33,21 @@ pub struct ClientCli { #[arg(long, default_value = "scpcap")] pub remote_exe: String, + /// How long a partial chunk may wait before being sent anyway, bounding + /// latency on a still-growing source: "100ms", "2s", or a bare number of + /// milliseconds. Chunks that reach --chunk sooner are sent immediately; + /// this only governs the leftover. Matters most with --ip, where a + /// selective filter would otherwise take a long time to fill a chunk. + #[arg(long, value_name = "DURATION", default_value = "100ms")] + pub flush_interval: String, + + /// Only forward packets whose source or destination IP matches. May be + /// repeated; a packet is kept if it matches any entry. Accepts a bare + /// address ("1.1.1.1", "2606:4700::1111") or a CIDR prefix + /// ("10.0.0.0/8"). Requires an Ethernet-linktype pcap source. + #[arg(long = "ip", value_name = "ADDR")] + pub ip: Vec, + /// Append a diagnostic trace of this (local) process's transfer events /// to this file -- positions, rename detection, finalize -- for /// figuring out where a stalled transfer is stuck. @@ -80,6 +95,16 @@ pub struct ServerCli { #[arg(long, default_value_t = crate::zstd_frame::DEFAULT_LEVEL)] pub level: i32, + /// Flush interval for a partial chunk (--send mode). See the client's + /// --flush-interval. + #[arg(long, value_name = "DURATION", default_value = "100ms")] + pub flush_interval: String, + + /// IP filter for --send mode; repeated, one per address/prefix. See the + /// client's --ip. + #[arg(long = "ip", value_name = "ADDR")] + pub ip: Vec, + /// Destination final path (--recv mode). #[arg(long)] pub dest_final: Option, diff --git a/src/client.rs b/src/client.rs index 46085fd..859d369 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,5 +1,6 @@ use crate::channel::{ChannelSink, ChannelSource}; use crate::cli::ClientCli; +use crate::filter::IpFilter; use crate::naming::{self, resolve_dest_paths, strip_name, Target}; use crate::pipeline::{run_recv, run_send, RecvOptions, SendOptions}; use crate::protocol::{MessageSink, MessageSource, WireSink, WireSource}; @@ -35,6 +36,28 @@ pub fn run(cli: ClientCli) -> Result<()> { let is_partial = source_stripped.was_partial; let effective_compress = cli.compress.is_some() && !source_stripped.was_zst; let chunk_target = naming::parse_size(&cli.chunk)?; + let flush_after = naming::parse_duration(&cli.flush_interval)?; + // Parsed up front so a bad --ip fails before anything is spawned or + // opened. Only the SENDING side filters, so this is used directly when + // we are the sender (local source) and forwarded as args when the + // remote is (see the Remote->Local branch). It is never passed to a + // --recv server: the receiver writes bytes verbatim and never parses + // them. + let filter = IpFilter::parse(&cli.ip)?; + // Filtering means reading pcap records, which an already-compressed + // source doesn't expose. Without this the transfer would still pick the + // record-aligned chunker (a filter forces it) and abort deep inside with + // "unrecognized pcap global header magic: 0xfd2fb528" -- the zstd magic. + // Such a copy is verbatim and works fine unfiltered, so refuse the + // combination rather than the source. + if !filter.is_match_all() && source_stripped.was_zst { + bail!( + "cannot apply --ip to an already-compressed source ({}): \ + filtering needs to read pcap records. Copy it without --ip, \ + or filter the uncompressed capture.", + cli.source + ); + } let dest = resolve_dest_paths(dest_target.path(), &source_stripped, effective_compress, &cli.extension); @@ -46,7 +69,9 @@ pub fn run(cli: ClientCli) -> Result<()> { final_source_path: PathBuf::from(&final_source_path), compress: effective_compress, chunk_target, + flush_after, level: cli.level, + filter: filter.clone(), log: log.clone(), }; let recv_opts = RecvOptions { @@ -83,7 +108,9 @@ pub fn run(cli: ClientCli) -> Result<()> { final_source_path: PathBuf::from(&final_source_path), compress: effective_compress, chunk_target, + flush_after, level: cli.level, + filter: filter.clone(), log: log.clone(), }; let sink: Box = Box::new(WireSink(stdin)); @@ -104,12 +131,18 @@ pub fn run(cli: ClientCli) -> Result<()> { final_source_path.clone(), "--chunk".to_string(), cli.chunk.clone(), + "--flush-interval".to_string(), + cli.flush_interval.clone(), "--level".to_string(), cli.level.to_string(), ]; if is_partial { server_args.push("--partial".to_string()); } + for spec in &cli.ip { + server_args.push("--ip".to_string()); + server_args.push(spec.clone()); + } if effective_compress { server_args.push("--compress".to_string()); server_args.push("zstd".to_string()); diff --git a/src/filter.rs b/src/filter.rs new file mode 100644 index 0000000..ce5e5d6 --- /dev/null +++ b/src/filter.rs @@ -0,0 +1,317 @@ +use anyhow::{Result, bail}; +use std::net::IpAddr; + +/// The only link-layer type the IP peek understands. Anything else matches +/// nothing -- see `matches`. +pub const LINKTYPE_ETHERNET: u32 = 1; + +const ETH_HEADER_LEN: usize = 14; +const ETHERTYPE_IPV4: u16 = 0x0800; +const ETHERTYPE_IPV6: u16 = 0x86DD; +const ETHERTYPE_VLAN: u16 = 0x8100; +const ETHERTYPE_QINQ: u16 = 0x88A8; +const ETHERTYPE_QINQ_ALT: u16 = 0x9100; +/// 802.1Q and one layer of QinQ. Deeper stacking is real but vanishingly +/// rare in capture files, and the bound keeps a corrupt/truncated packet +/// from walking off into the payload looking for an ethertype. +const MAX_VLAN_TAGS: usize = 2; + +const IPV4_HEADER_MIN: usize = 20; +const IPV6_HEADER_LEN: usize = 40; + +/// A set of IP addresses/prefixes to keep. A packet matches if *either* its +/// source or destination address falls in *any* entry. +/// +/// Sized for the expected 1-4 entries: two plain vectors, split by address +/// family so a v4 packet never compares against a v6 entry, scanned +/// linearly. No sorting (a prefix match isn't an equality test, so ordering +/// buys nothing) and no hashing. +#[derive(Debug, Clone)] +pub struct IpFilter { + /// Set when no `--ip` was given: every packet is kept, without looking + /// at it at all. This is the default so that the filtering code path is + /// the *only* code path -- see the chunker. + match_all: bool, + /// (network, mask) pairs, network already masked at parse time. + v4: Vec<(u32, u32)>, + v6: Vec<(u128, u128)>, +} + +impl IpFilter { + /// The no-`--ip` default: keeps everything, matches without inspecting. + pub fn match_all() -> Self { + Self { match_all: true, v4: Vec::new(), v6: Vec::new() } + } + + /// Parses `--ip` specs: a bare address (`1.1.1.1`, `2606:4700::1111`) + /// or a CIDR prefix (`10.0.0.0/8`). An empty list yields `match_all`. + pub fn parse(specs: &[String]) -> Result { + if specs.is_empty() { + return Ok(Self::match_all()); + } + let mut f = Self { match_all: false, v4: Vec::new(), v6: Vec::new() }; + for spec in specs { + let (addr_str, prefix_str) = match spec.split_once('/') { + Some((a, p)) => (a, Some(p)), + None => (spec.as_str(), None), + }; + let addr: IpAddr = addr_str + .parse() + .map_err(|_| anyhow::anyhow!("--ip {spec:?}: not an IP address"))?; + let width = if addr.is_ipv4() { 32u32 } else { 128u32 }; + let prefix = match prefix_str { + Some(p) => p + .parse::() + .map_err(|_| anyhow::anyhow!("--ip {spec:?}: prefix length is not a number"))?, + None => width, + }; + if prefix > width { + bail!("--ip {spec:?}: prefix length {prefix} exceeds {width} bits"); + } + match addr { + IpAddr::V4(v4) => { + let mask = if prefix == 0 { 0 } else { u32::MAX << (32 - prefix) }; + f.v4.push((u32::from(v4) & mask, mask)); + } + IpAddr::V6(v6) => { + let mask = if prefix == 0 { 0 } else { u128::MAX << (128 - prefix) }; + f.v6.push((u128::from(v6) & mask, mask)); + } + } + } + Ok(f) + } + + /// Whether this filter keeps everything, i.e. no `--ip` was given. + /// Callers use it to skip work (and warnings) that only a real filter + /// needs. + pub fn is_match_all(&self) -> bool { + self.match_all + } + + /// Whether to keep `packet` -- the captured bytes of one pcap record, + /// starting at the link-layer header. + /// + /// Everything unmatched is dropped: non-IP frames (ARP, LLDP), packets + /// truncated by snaplen before the addresses, and any link type other + /// than Ethernet. + pub fn matches(&self, packet: &[u8], linktype: u32) -> bool { + if self.match_all { + return true; + } + if linktype != LINKTYPE_ETHERNET || packet.len() < ETH_HEADER_LEN { + return false; + } + let mut ethertype = be16(&packet[12..14]); + let mut off = ETH_HEADER_LEN; + let mut tags = 0; + // A VLAN tag is TPID(2) + TCI(2); the TPID is the ethertype we just + // read, so the *next* type field sits 2 bytes into the tag. + while matches!(ethertype, ETHERTYPE_VLAN | ETHERTYPE_QINQ | ETHERTYPE_QINQ_ALT) { + tags += 1; + if tags > MAX_VLAN_TAGS || packet.len() < off + 4 { + return false; + } + ethertype = be16(&packet[off + 2..off + 4]); + off += 4; + } + match ethertype { + ETHERTYPE_IPV4 => { + if packet.len() < off + IPV4_HEADER_MIN || packet[off] >> 4 != 4 { + return false; + } + let src = be32(&packet[off + 12..off + 16]); + let dst = be32(&packet[off + 16..off + 20]); + self.v4.iter().any(|&(net, mask)| src & mask == net || dst & mask == net) + } + ETHERTYPE_IPV6 => { + if packet.len() < off + IPV6_HEADER_LEN || packet[off] >> 4 != 6 { + return false; + } + let src = be128(&packet[off + 8..off + 24]); + let dst = be128(&packet[off + 24..off + 40]); + self.v6.iter().any(|&(net, mask)| src & mask == net || dst & mask == net) + } + _ => false, + } + } +} + +fn be16(b: &[u8]) -> u16 { + u16::from_be_bytes(b.try_into().unwrap()) +} + +fn be32(b: &[u8]) -> u32 { + u32::from_be_bytes(b.try_into().unwrap()) +} + +fn be128(b: &[u8]) -> u128 { + u128::from_be_bytes(b.try_into().unwrap()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn specs(list: &[&str]) -> Vec { + list.iter().map(|s| s.to_string()).collect() + } + + fn filter(list: &[&str]) -> IpFilter { + IpFilter::parse(&specs(list)).unwrap() + } + + /// An Ethernet frame with `vlan_tags` 802.1Q tags in front of `ethertype`. + fn eth(vlan_tags: usize, ethertype: u16, payload: &[u8]) -> Vec { + let mut f = vec![0u8; 12]; // dst + src MAC + for _ in 0..vlan_tags { + f.extend_from_slice(ÐERTYPE_VLAN.to_be_bytes()); + f.extend_from_slice(&[0x00, 0x64]); // TCI: VLAN 100 + } + f.extend_from_slice(ðertype.to_be_bytes()); + f.extend_from_slice(payload); + f + } + + fn ipv4(src: &str, dst: &str) -> Vec { + let mut p = vec![0u8; IPV4_HEADER_MIN]; + p[0] = 0x45; // version 4, IHL 5 + let s: std::net::Ipv4Addr = src.parse().unwrap(); + let d: std::net::Ipv4Addr = dst.parse().unwrap(); + p[12..16].copy_from_slice(&s.octets()); + p[16..20].copy_from_slice(&d.octets()); + p + } + + fn ipv6(src: &str, dst: &str) -> Vec { + let mut p = vec![0u8; IPV6_HEADER_LEN]; + p[0] = 0x60; // version 6 + let s: std::net::Ipv6Addr = src.parse().unwrap(); + let d: std::net::Ipv6Addr = dst.parse().unwrap(); + p[8..24].copy_from_slice(&s.octets()); + p[24..40].copy_from_slice(&d.octets()); + p + } + + fn v4_packet(src: &str, dst: &str) -> Vec { + eth(0, ETHERTYPE_IPV4, &ipv4(src, dst)) + } + + #[test] + fn match_all_keeps_everything_including_non_ip_and_odd_linktypes() { + let f = IpFilter::match_all(); + assert!(f.is_match_all()); + assert!(f.matches(&[], 113)); + assert!(f.matches(ð(0, 0x0806, &[1, 2, 3]), LINKTYPE_ETHERNET)); + // An empty --ip list is the same thing. + assert!(IpFilter::parse(&[]).unwrap().is_match_all()); + } + + #[test] + fn exact_v4_matches_either_direction_only() { + let f = filter(&["1.1.1.1"]); + assert!(!f.is_match_all()); + assert!(f.matches(&v4_packet("1.1.1.1", "9.9.9.9"), LINKTYPE_ETHERNET)); + assert!(f.matches(&v4_packet("9.9.9.9", "1.1.1.1"), LINKTYPE_ETHERNET)); + assert!(!f.matches(&v4_packet("9.9.9.9", "8.8.8.8"), LINKTYPE_ETHERNET)); + // Neighbouring address must not match a bare (/32) entry. + assert!(!f.matches(&v4_packet("1.1.1.2", "9.9.9.9"), LINKTYPE_ETHERNET)); + } + + #[test] + fn exact_v6_matches_either_direction() { + let f = filter(&["2606:4700::1111"]); + let pkt = eth(0, ETHERTYPE_IPV6, &ipv6("2001:db8::1", "2606:4700::1111")); + assert!(f.matches(&pkt, LINKTYPE_ETHERNET)); + let miss = eth(0, ETHERTYPE_IPV6, &ipv6("2001:db8::1", "2606:4700::1112")); + assert!(!f.matches(&miss, LINKTYPE_ETHERNET)); + } + + #[test] + fn cidr_boundaries_are_exact() { + let f = filter(&["10.0.0.0/8"]); + assert!(f.matches(&v4_packet("10.0.0.0", "9.9.9.9"), LINKTYPE_ETHERNET)); + assert!(f.matches(&v4_packet("10.255.255.255", "9.9.9.9"), LINKTYPE_ETHERNET)); + assert!(!f.matches(&v4_packet("11.0.0.0", "9.9.9.9"), LINKTYPE_ETHERNET)); + assert!(!f.matches(&v4_packet("9.255.255.255", "9.9.9.9"), LINKTYPE_ETHERNET)); + + // /0 keeps every packet of that family, but still not non-IP. + let all = filter(&["0.0.0.0/0"]); + assert!(all.matches(&v4_packet("203.0.113.9", "9.9.9.9"), LINKTYPE_ETHERNET)); + assert!(!all.matches(ð(0, 0x0806, &[0u8; 28]), LINKTYPE_ETHERNET)); + + let v6 = filter(&["2606:4700::/32"]); + assert!(v6.matches(ð(0, ETHERTYPE_IPV6, &ipv6("2606:4700:dead::1", "::1")), LINKTYPE_ETHERNET)); + assert!(!v6.matches(ð(0, ETHERTYPE_IPV6, &ipv6("2606:4701::1", "::1")), LINKTYPE_ETHERNET)); + } + + #[test] + fn families_do_not_cross_match() { + // A v4-only filter must never keep a v6 packet, and vice versa -- + // the split vectors mean the comparison is never even attempted. + let v4only = filter(&["1.1.1.1"]); + assert!(!v4only.matches(ð(0, ETHERTYPE_IPV6, &ipv6("::1", "::2")), LINKTYPE_ETHERNET)); + let v6only = filter(&["::1"]); + assert!(!v6only.matches(&v4_packet("1.1.1.1", "9.9.9.9"), LINKTYPE_ETHERNET)); + } + + #[test] + fn vlan_tags_are_skipped_single_and_double() { + let f = filter(&["1.1.1.1"]); + let one = eth(1, ETHERTYPE_IPV4, &ipv4("1.1.1.1", "9.9.9.9")); + let two = eth(2, ETHERTYPE_IPV4, &ipv4("9.9.9.9", "1.1.1.1")); + assert!(f.matches(&one, LINKTYPE_ETHERNET)); + assert!(f.matches(&two, LINKTYPE_ETHERNET)); + // Beyond the tag bound we stop rather than hunt through the payload. + let three = eth(3, ETHERTYPE_IPV4, &ipv4("1.1.1.1", "9.9.9.9")); + assert!(!f.matches(&three, LINKTYPE_ETHERNET)); + } + + #[test] + fn unparseable_packets_are_dropped() { + let f = filter(&["1.1.1.1"]); + // ARP. + assert!(!f.matches(ð(0, 0x0806, &[0u8; 28]), LINKTYPE_ETHERNET)); + // Truncated before the addresses (snaplen too small). + let full = v4_packet("1.1.1.1", "9.9.9.9"); + assert!(!f.matches(&full[..ETH_HEADER_LEN + 12], LINKTYPE_ETHERNET)); + // Shorter than an Ethernet header, and empty. + assert!(!f.matches(&full[..8], LINKTYPE_ETHERNET)); + assert!(!f.matches(&[], LINKTYPE_ETHERNET)); + // Right ethertype, wrong version nibble. + let mut bad_version = full.clone(); + bad_version[ETH_HEADER_LEN] = 0x65; + assert!(!f.matches(&bad_version, LINKTYPE_ETHERNET)); + } + + #[test] + fn non_ethernet_linktypes_match_nothing() { + let f = filter(&["1.1.1.1"]); + let pkt = v4_packet("1.1.1.1", "9.9.9.9"); + // 113 = LINUX_SLL, what `tcpdump -i any` produces; 101 = RAW; 0 = NULL. + for linktype in [0u32, 101, 113, 276] { + assert!(!f.matches(&pkt, linktype), "linktype {linktype} should match nothing"); + } + } + + #[test] + fn bad_specs_are_rejected_at_parse() { + for spec in ["1.1.1.1/33", "::1/129", "not-an-ip", "1.1.1.1/x", "1.1.1.256"] { + assert!( + IpFilter::parse(&specs(&[spec])).is_err(), + "{spec:?} should be rejected" + ); + } + // /0 and full-width prefixes are fine. + assert!(IpFilter::parse(&specs(&["1.1.1.1/32", "0.0.0.0/0", "::/0", "::1/128"])).is_ok()); + } + + #[test] + fn multiple_entries_are_all_considered() { + let f = filter(&["1.1.1.1", "8.8.8.8", "10.0.0.0/8", "2606:4700::/32"]); + assert!(f.matches(&v4_packet("8.8.8.8", "9.9.9.9"), LINKTYPE_ETHERNET)); + assert!(f.matches(&v4_packet("9.9.9.9", "10.1.2.3"), LINKTYPE_ETHERNET)); + assert!(f.matches(ð(0, ETHERTYPE_IPV6, &ipv6("::1", "2606:4700::1")), LINKTYPE_ETHERNET)); + assert!(!f.matches(&v4_packet("9.9.9.9", "203.0.113.1"), LINKTYPE_ETHERNET)); + } +} diff --git a/src/main.rs b/src/main.rs index 00f794b..501fd77 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ mod channel; mod chunker; mod cli; mod client; +mod filter; mod naming; mod pipeline; mod protocol; diff --git a/src/naming.rs b/src/naming.rs index cd129b9..ed89d97 100644 --- a/src/naming.rs +++ b/src/naming.rs @@ -1,4 +1,5 @@ use anyhow::{bail, Result}; +use std::time::Duration; use std::path::Path; /// Either a local filesystem path or a `[user@]host:path` remote target. @@ -128,6 +129,28 @@ pub fn parse_size(s: &str) -> Result { Ok(n * mult) } +/// Parses a flush/idle interval: a bare number is milliseconds, or an +/// explicit "ms"/"s" suffix. "100", "100ms" and "0.1s" are not all spellings +/// of the same thing -- only integers are accepted, so use "100ms" or "1s". +pub fn parse_duration(s: &str) -> Result { + let s = s.trim(); + if s.is_empty() { + bail!("empty interval"); + } + let (digits, mult_ms) = if let Some(d) = s.strip_suffix("ms") { + (d, 1) + } else if let Some(d) = s.strip_suffix('s') { + (d, 1000) + } else { + (s, 1) + }; + let n: u64 = digits + .trim() + .parse() + .map_err(|_| anyhow::anyhow!("invalid interval: {s:?} (try \"100ms\" or \"1s\")"))?; + Ok(Duration::from_millis(n * mult_ms)) +} + /// Single-quote a string for safe inclusion as one argument in a remote shell /// command line, e.g. for `ssh host sh -c '...'`-style invocations. pub fn shell_quote(s: &str) -> String { @@ -251,6 +274,19 @@ mod tests { assert_eq!(d.temp_path, "cap.pcap.partial"); } + #[test] + fn duration_parsing() { + assert_eq!(parse_duration("100").unwrap(), Duration::from_millis(100)); + assert_eq!(parse_duration("100ms").unwrap(), Duration::from_millis(100)); + assert_eq!(parse_duration("2s").unwrap(), Duration::from_secs(2)); + assert_eq!(parse_duration(" 250ms ").unwrap(), Duration::from_millis(250)); + // 0 is legal: cut a chunk as soon as anything is kept. + assert_eq!(parse_duration("0").unwrap(), Duration::ZERO); + for bad in ["", "abc", "0.1s", "100m", "-5"] { + assert!(parse_duration(bad).is_err(), "{bad:?} should be rejected"); + } + } + #[test] fn size_parsing() { assert_eq!(parse_size("256k").unwrap(), 256 * 1024); diff --git a/src/pipeline.rs b/src/pipeline.rs index 0c2ae2b..ebe441d 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -1,4 +1,5 @@ -use crate::chunker::{Chunker, RawChunker, RecordAlignedChunker, DEFAULT_FLUSH}; +use crate::chunker::{Chunker, RawChunker, RecordAlignedChunker}; +use crate::filter::IpFilter; use crate::protocol::{Message, MessageSink, MessageSource}; use crate::tail::{run_tail, SourceEvent}; use crate::tracelog::Logger; @@ -8,6 +9,7 @@ use std::fs::File; use std::io::Write; use std::path::PathBuf; use std::sync::{mpsc, Arc}; +use std::time::Duration; use std::thread; /// Bound on the reader->compressor and compressor->sender channels: enough to @@ -22,7 +24,14 @@ pub struct SendOptions { pub final_source_path: PathBuf, pub compress: bool, pub chunk_target: usize, + /// How long kept-but-undersized output may wait before being sent + /// anyway. See `Chunker::should_flush`. + pub flush_after: Duration, pub level: i32, + /// Which packets to forward. `IpFilter::match_all()` when no `--ip` was + /// given -- not a disabled filter, just one that says yes, so there is + /// only ever one code path through the chunker. + pub filter: IpFilter, pub log: Arc, } @@ -39,8 +48,32 @@ enum CompressedItem { /// serializing -- see the plan's Performance section. Non-compress mode /// collapses to 2 stages since there's no compression work to overlap. pub fn run_send(opts: SendOptions, mut sink: Box) -> Result<()> { - let chunker: Box = if opts.compress || opts.is_partial { - Box::new(RecordAlignedChunker::new(opts.chunk_target, DEFAULT_FLUSH, opts.is_partial)) + // Two independent reasons to parse records, either of which forces the + // record-aligned chunker: + // * `compress`: chunk boundaries become zstd frame boundaries, and the + // format requires every frame to hold a whole number of records + // (STREAMING_ZSTD_FORMAT.md invariant 3). The cut point *is* the format. + // * `is_partial`: a growing source needs `confirmed_len()` -- record + // structure is the only way to tell real content from a preallocating + // writer's not-yet-written tail. Here the chunker is a validator, not + // a framer. + // * a real `--ip` filter: deciding whether to forward a packet means + // finding the packet, and record boundaries are the only way to know + // where one starts. Here the chunker is a demultiplexer. + // A final, uncompressed copy has neither: `run_recv` writes every Data + // payload verbatim and never inspects a boundary, so alignment would buy + // nothing, while requiring pcap structure would break copying pcapng, + // truncated, or otherwise non-pcap files that transfer fine today. So it + // gets the passthrough chunker, framed on the wire only by tail.rs's + // READ_WINDOW. + let chunker: Box = if opts.compress || opts.is_partial || !opts.filter.is_match_all() + { + Box::new(RecordAlignedChunker::with_filter( + opts.chunk_target, + opts.flush_after, + opts.is_partial, + opts.filter.clone(), + )) } else { Box::new(RawChunker::new()) }; @@ -238,7 +271,9 @@ mod tests { final_source_path: src.clone(), compress: false, chunk_target: crate::chunker::DEFAULT_TARGET_BYTES, + flush_after: crate::chunker::DEFAULT_FLUSH, level: crate::zstd_frame::DEFAULT_LEVEL, + filter: IpFilter::match_all(), log: Arc::new(Logger::none()), }; run_pipe(opts, dest_final.clone(), dest_temp.clone()).unwrap(); @@ -274,7 +309,9 @@ mod tests { final_source_path: final_src.clone(), compress: false, chunk_target: crate::chunker::DEFAULT_TARGET_BYTES, + flush_after: crate::chunker::DEFAULT_FLUSH, level: crate::zstd_frame::DEFAULT_LEVEL, + filter: IpFilter::match_all(), log: Arc::new(Logger::none()), }; @@ -321,7 +358,9 @@ mod tests { final_source_path: src.clone(), compress: true, chunk_target: 512, // small target to force multiple frames + flush_after: crate::chunker::DEFAULT_FLUSH, level: crate::zstd_frame::DEFAULT_LEVEL, + filter: IpFilter::match_all(), log: Arc::new(Logger::none()), }; run_pipe(opts, dest_final.clone(), dest_temp.clone()).unwrap(); @@ -357,10 +396,138 @@ mod tests { final_source_path: src.clone(), compress: true, chunk_target: crate::chunker::DEFAULT_TARGET_BYTES, + flush_after: crate::chunker::DEFAULT_FLUSH, level: crate::zstd_frame::DEFAULT_LEVEL, + filter: IpFilter::match_all(), log: Arc::new(Logger::none()), }; assert!(run_pipe(opts, dest_final.clone(), dest_temp).is_err()); assert!(!dest_final.exists()); } + + fn eth_header_bytes() -> Vec { + let mut h = header_bytes(0xA1B2C3D4); + h[20..24].copy_from_slice(&1u32.to_le_bytes()); // LINKTYPE_ETHERNET + h + } + + /// A record holding an Ethernet/IPv4 packet, padded to `len` captured bytes. + fn ip_record(src: [u8; 4], dst: [u8; 4], len: usize) -> Vec { + let mut pkt = vec![0u8; len.max(34)]; + pkt[12..14].copy_from_slice(&0x0800u16.to_be_bytes()); + pkt[14] = 0x45; + pkt[26..30].copy_from_slice(&src); + pkt[30..34].copy_from_slice(&dst); + record_bytes(&pkt, 1_700_000_000) + } + + fn ip_filter(specs: &[&str]) -> IpFilter { + IpFilter::parse(&specs.iter().map(|s| s.to_string()).collect::>()).unwrap() + } + + /// Source stream plus the exact pcap a `--ip 1.1.1.1` transfer should + /// produce from it: same global header, only the matching records. + fn filtered_fixture(n: usize) -> (Vec, Vec) { + let mut source = eth_header_bytes(); + let mut expected = eth_header_bytes(); + for i in 0..n { + if i % 5 == 0 { + let hit = ip_record([1, 1, 1, 1], [9, 9, 9, 9], 100 + i); + source.extend_from_slice(&hit); + expected.extend_from_slice(&hit); + } else { + source.extend_from_slice(&ip_record([9, 9, 9, 9], [8, 8, 8, 8], 100 + i)); + } + } + (source, expected) + } + + #[test] + fn filtered_transfer_carries_only_matching_packets() { + let dir = tempdir().unwrap(); + let src = dir.path().join("cap.pcap"); + let (source, expected) = filtered_fixture(50); + std::fs::write(&src, &source).unwrap(); + let dest_final = dir.path().join("out.pcap"); + let dest_temp = dir.path().join("out.pcap.partial"); + + let opts = SendOptions { + source_path: src.clone(), + is_partial: false, + final_source_path: src.clone(), + compress: false, + chunk_target: crate::chunker::DEFAULT_TARGET_BYTES, + flush_after: crate::chunker::DEFAULT_FLUSH, + level: crate::zstd_frame::DEFAULT_LEVEL, + filter: ip_filter(&["1.1.1.1"]), + log: Arc::new(Logger::none()), + }; + run_pipe(opts, dest_final.clone(), dest_temp).unwrap(); + + // A filter forces the record-aligned chunker even though this is a + // final, uncompressed copy -- so the output is a real pcap, not a + // byte-for-byte copy of the source. + assert_eq!(std::fs::read(&dest_final).unwrap(), expected); + assert!(expected.len() < source.len()); + } + + #[test] + fn filtered_compressed_transfer_round_trips_to_the_filtered_pcap() { + let dir = tempdir().unwrap(); + let src = dir.path().join("cap.pcap"); + let (source, expected) = filtered_fixture(200); + std::fs::write(&src, &source).unwrap(); + let dest_final = dir.path().join("out.pcap.zst"); + let dest_temp = dir.path().join("out.pcap.zst.partial"); + + let opts = SendOptions { + source_path: src.clone(), + is_partial: false, + final_source_path: src.clone(), + compress: true, + chunk_target: 512, + flush_after: crate::chunker::DEFAULT_FLUSH, + level: crate::zstd_frame::DEFAULT_LEVEL, + filter: ip_filter(&["1.1.1.1"]), + log: Arc::new(Logger::none()), + }; + run_pipe(opts, dest_final.clone(), dest_temp).unwrap(); + + let out_bytes = std::fs::read(&dest_final).unwrap(); + assert_eq!(&out_bytes[0..14], &MARKER_FRAME[..]); + // The whole point of the container: decompressing yields a valid + // pcap, here the filtered one. + assert_eq!(zstd::stream::decode_all(out_bytes.as_slice()).unwrap(), expected); + // Header frame still declares exactly the 24-byte global header. + let header_content_size = zstd::zstd_safe::get_frame_content_size(&out_bytes[14..]) + .unwrap() + .unwrap(); + assert_eq!(header_content_size, 24); + } + + #[test] + fn a_filter_matching_nothing_produces_a_header_only_pcap() { + // Not an error: the transfer completes and the destination is a + // valid, empty capture. + let dir = tempdir().unwrap(); + let src = dir.path().join("cap.pcap"); + let (source, _) = filtered_fixture(20); + std::fs::write(&src, &source).unwrap(); + let dest_final = dir.path().join("out.pcap"); + let dest_temp = dir.path().join("out.pcap.partial"); + + let opts = SendOptions { + source_path: src.clone(), + is_partial: false, + final_source_path: src.clone(), + compress: false, + chunk_target: crate::chunker::DEFAULT_TARGET_BYTES, + flush_after: crate::chunker::DEFAULT_FLUSH, + level: crate::zstd_frame::DEFAULT_LEVEL, + filter: ip_filter(&["203.0.113.7"]), + log: Arc::new(Logger::none()), + }; + run_pipe(opts, dest_final.clone(), dest_temp).unwrap(); + assert_eq!(std::fs::read(&dest_final).unwrap(), eth_header_bytes()); + } } diff --git a/src/server.rs b/src/server.rs index a8b2c42..8cddc38 100644 --- a/src/server.rs +++ b/src/server.rs @@ -1,4 +1,5 @@ use crate::cli::{ServerCli, ServerRole}; +use crate::filter::IpFilter; use crate::naming; use crate::pipeline::{run_recv, run_send, RecvOptions, SendOptions}; use crate::protocol::{MessageSink, MessageSource, WireSink, WireSource}; @@ -35,6 +36,11 @@ fn run_send_server(cli: ServerCli) -> Result<()> { None => false, }; let chunk_target = naming::parse_size(&cli.chunk)?; + let flush_after = naming::parse_duration(&cli.flush_interval)?; + // The sending side is the only side that filters -- dropping packets + // before they hit the wire is the whole point. `run_recv` never parses + // what it writes, so --recv has no --ip. + let filter = IpFilter::parse(&cli.ip)?; let opts = SendOptions { source_path: PathBuf::from(source), @@ -42,7 +48,9 @@ fn run_send_server(cli: ServerCli) -> Result<()> { final_source_path: PathBuf::from(final_source), compress, chunk_target, + flush_after, level: cli.level, + filter, log, }; // `Stdout`/`Stdin` (not the `.lock()` guards) are used here: the guards diff --git a/src/tail.rs b/src/tail.rs index a713f76..234fbba 100644 --- a/src/tail.rs +++ b/src/tail.rs @@ -170,10 +170,6 @@ fn drain( last_stuck_peek = Some(preview.to_vec()); } } - } else if chunker.should_flush(Instant::now()) - && let Some((kind, data)) = chunker.flush() - { - send_chunk(events, kind, data)?; } if file_len < confirmed && !shrunk_below_anchor { shrunk_below_anchor = true; @@ -193,13 +189,24 @@ fn drain( emit_ready(chunker, events)?; made_progress = true; } - } else if chunker.should_flush(Instant::now()) - && let Some((kind, data)) = chunker.flush() - { - send_chunk(events, kind, data)?; } } + // Checked on EVERY iteration, not just ones that read nothing. The + // flush timer measures how long output has been waiting, so it has + // to be able to fire while the source is still growing: with a + // selective filter, a busy source produces a steady trickle of + // matching records that would otherwise sit here until they reached + // the size target -- which at a 1% match rate means ~100x the target + // in source bytes, seconds of added latency on a live capture. + // `should_flush` is false whenever there is nothing kept, so this + // never cuts an empty chunk. + if chunker.should_flush(Instant::now()) + && let Some((kind, data)) = chunker.flush() + { + send_chunk(events, kind, data)?; + } + if !complete_signal && final_path.try_exists()? { // Rename observed -- signal now, well before finalization, per // section 7.4's repoint-then-drain-then-finalize ordering.