From d9810d2609ec0185a760fd5a44ee426686f0e08e Mon Sep 17 00:00:00 2001 From: Konstantin Mikhailov Date: Fri, 2 Oct 2026 04:52:02 +0300 Subject: [PATCH] fix: allow opt-in recovery of stale IPv4 fragment IDs --- README.md | 23 +- docs/architecture.md | 40 +- docs/rfc/0005-dual-path-pcap-parsing.md | 22 + src/app.rs | 20 +- src/cli.rs | 211 +++++++- src/config.rs | 27 + src/dns_processor.rs | 7 + src/dns_processor/pipeline.rs | 190 ++++++- src/dns_processor/reassembly.rs | 687 +++++++++++++++++++++++- src/parquet_writer.rs | 6 + tests/end_to_end.rs | 294 +++++++++- 11 files changed, 1486 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index 9bdd2d7..5ab6f88 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,7 @@ dpp --report-format json input.pcap output.csv > dpp-summary.json | `--max-dns-compression-jumps ` | Maximum compression-pointer jumps per DNS name; defaults to `32`; `0` disables the jump limit | | `--allow-fragments` | Opt in to matching complete queries with first IPv4 response fragments when the DNS header and question are present; no reassembly | | `--full-fragments` | Reassemble IPv4 fragments into complete UDP datagrams; also enables `--allow-fragments` for incomplete responses | +| `--ipv4-fragment-reuse-gap-ms ` | Recover a pending IPv4 ID when a new first fragment is separated by more than this capture-time interval; requires `--full-fragments`, conflicts with `--monotonic-capture`; range `1..=9223372036854775`, unset by default | | `--anonymize ` | Path to the pseudonymization key file | | `-h, --help` | Print help | | `-V, --version` | Print version | @@ -182,6 +183,7 @@ provide supported per-core thread pinning through the affinity backend. | `DPP_MAX_DNS_COMPRESSION_JUMPS` | Maximum compression-pointer jumps per DNS name when `--max-dns-compression-jumps` is not set; defaults to `32`; `0` disables the jump limit | | `DPP_ALLOW_FRAGMENTS` | Enable first-fragment IPv4 response matching without reassembly. | | `DPP_FULL_FRAGMENTS` | Enable IPv4 reassembly and first-fragment response matching. | +| `DPP_IPV4_FRAGMENT_REUSE_GAP_MS` | IPv4 pending-ID recovery interval in milliseconds when `--ipv4-fragment-reuse-gap-ms` is absent; same range and mode requirements as the CLI option; unset by default. | | `DPP_ANONYMIZE` | Path to the key file used for pseudonymization | | `DPP_SILENT` | Suppress info-level log output | @@ -284,10 +286,27 @@ mode. Fallback uses the first fragment's timestamp. Fragmented queries need complete reassembly; overlapping, inconsistent, or UDP-length-mismatched fragment sets are rejected. +In ordinary full-fragment mode, an unresolved or rejected fragment set can retain its IPv4 ID +until capacity eviction or EOF. For captures where a chosen time interval can separate reused +IDs, opt in with `--ipv4-fragment-reuse-gap-ms ` or `DPP_IPV4_FRAGMENT_REUSE_GAP_MS`. +The option requires full reassembly and cannot be combined with monotonic-capture mode. + +Recovery starts only when a valid DNS first fragment arrives more than the configured interval +after the maximum timestamp admitted to the old pending set, including duplicates and rejected +fragments. It also works after a collision has rejected that set. The old usable response prefix +may fall back once. The retained cutoff is the old maximum plus the interval; fragments with +timestamps at or before it are ignored while the key's history remains retained. Completed +datagrams alone do not trigger recovery, and the interval does not expire a set by itself. + +The interval expresses an operator's hypothesis about capture timestamps and generation +separation; it is not a universal maximum datagram lifetime. Recovery does not guarantee +tail-before-first handling across generations, and protection against old replay ends after all +bounded history for the key has been evicted. Without the option, existing behavior is unchanged. + Duplicate suppression uses bounded history and compares the fragment key and UDP payload bytes: -- A reused IPv4 ID with different content can form a new datagram, even with an identical first - fragment. +- After completion, a reused IPv4 ID with different content can form a new datagram, even with an + identical first fragment. - A fully identical new datagram, or an incomplete one whose observed fragments match a recent completion, may be suppressed because it is indistinguishable from a duplicate. - A separate bounded history drops tails after a non-DNS first fragment. If a new DNS datagram diff --git a/docs/architecture.md b/docs/architecture.md index b41d226..4908e53 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -356,8 +356,8 @@ The datagram bounds must agree: Incomplete responses may fall back to first-fragment inference on capacity eviction or EOF. With monotonic capture, entries can also expire after the match timeout. Without it, timestamp -gaps between fragments of the same datagram do not expire the entry; capacity and EOF still bound -its lifetime. +gaps alone do not expire an entry; capacity and EOF still bound its lifetime. Explicit pending-ID +recovery is described below. - Complete datagrams use the final fragment's (`MF=0`) capture timestamp. - Inferred responses use the first fragment's timestamp. @@ -372,7 +372,7 @@ no record. ```mermaid flowchart TB Input["Packet batches"] --> Reassembly["IPv4 reassembly
pending fragment sets"] - Resolve["EOF / capacity pressure
expiry only in monotonic mode"] -.-> Reassembly + Resolve["EOF / capacity pressure
monotonic expiry / opted-in first-led reuse"] -.-> Reassembly Reassembly -->|"unfragmented packets, complete datagrams,
or fallback prefixes"| Ready["Retained ready packets
held while fragments are unresolved"] Ready -.->|"backlog limit exceeded:
resolve pending sets"| Reassembly Ready -->|"after pending sets are resolved"| Route["Sort by timestamp and ordinal
validate packet metadata and route"] @@ -395,8 +395,8 @@ Reassembly keeps a bounded history of completed datagrams. Fragment keys and UDP are compared before suppressing repeated complete datagrams or prefix fallback matching a recent completion. -A reused IPv4 ID with different content can form a new datagram, even if its first -fragment is identical. An incomplete new datagram whose observed fragments match a recent +After completion, a reused IPv4 ID with different content can form a new datagram, even if its +first fragment is identical. An incomplete new datagram whose observed fragments match a recent completion is indistinguishable from a duplicate and may be suppressed, as may a fully identical new datagram. @@ -411,6 +411,36 @@ The canonical VLAN context is part of each fragment key: different tagged segmen assembled together, while PCP or DEI changes do not split a datagram. All reassembly state must remain bounded in memory. +### Opt-in pending-ID recovery + +`--ipv4-fragment-reuse-gap-ms` (environment: `DPP_IPV4_FRAGMENT_REUSE_GAP_MS`) supplies an explicit +capture-time interval for recovering an unresolved or rejected IPv4 fragment set. It requires +`--full-fragments`, conflicts with `--monotonic-capture`, and is unset by default. CLI values take +precedence over the environment; the supported range is `1..=9223372036854775` milliseconds. + +The routing stage retains one reassembler across batches. Each pending candidate records the +maximum timestamp of admitted fragments, including duplicate and rejection events. A valid DNS +first fragment here means a structurally valid IPv4/UDP first fragment with a port 53 endpoint; +DNS semantic validation remains downstream. Interval-based recovery starts a new candidate only +when its timestamp is strictly greater than the old maximum plus the configured interval. +Identical first-fragment bytes can also initiate this recovery. Completed and non-DNS history +alone cannot initiate it. + +Recovery removes the old segments and completion history. An old usable first-response prefix +passes through the existing fallback path once; rejected candidates and candidates whose bytes +still match a recent completion produce no prefix. A retained cutoff, equal to the old maximum +plus the interval, rejects first fragments, tails, and non-DNS first fragments at or before it +before they can change state. The existing pending, completed, and non-DNS records carry this +cutoff, taking the maximum when boundaries meet. No additional history table is required. +First-seen timestamps continue to govern the existing ordering policy. Saturated timestamps at +the signed 64-bit limits cannot initiate recovery. + +This policy relies on the operator's hypothesis that capture timestamps and the chosen interval +separate generations; it does not establish a universal datagram lifetime. Tail-before-first +recovery across generations is not guaranteed. The cutoff is retained only while some bounded +history for the key remains; after its eviction, old replay can be accepted again. Structural +fragment validation and exported record formats retain their existing semantics. + ## DNS validation ### Decoder paths and message boundaries diff --git a/docs/rfc/0005-dual-path-pcap-parsing.md b/docs/rfc/0005-dual-path-pcap-parsing.md index 4a8c7fa..0277a63 100644 --- a/docs/rfc/0005-dual-path-pcap-parsing.md +++ b/docs/rfc/0005-dual-path-pcap-parsing.md @@ -115,6 +115,28 @@ With `--full-fragments`, incomplete responses may use prefix inference on capaci EOF. Elapsed capture-time timeout can also trigger fallback when `--monotonic-capture` is enabled. Fragmented queries require complete reassembly. +`--ipv4-fragment-reuse-gap-ms ` opts in to recovering an existing pending fragment set when +a valid DNS first fragment arrives strictly more than `MS` after its maximum admitted timestamp. +The maximum includes duplicates and rejection events, so rejected sets can also recover. The +first fragment must be structurally valid IPv4/UDP with a port 53 endpoint; DNS semantic +validation remains downstream. The option is unset by default, accepts `1..=9223372036854775` +milliseconds, requires full reassembly, and conflicts with monotonic-capture mode. +`DPP_IPV4_FRAGMENT_REUSE_GAP_MS` supplies the value when the CLI option is absent. Mode +requirements are checked after CLI and environment resolution. + +Recovery retires old segments and completion history, and releases an old usable response prefix +through the existing fallback at most once. Rejected sets and candidates whose observed bytes +match a recent completion do not release a prefix. The old maximum plus the interval becomes a +cutoff: first fragments, tails, and non-DNS first fragments at or before it are ignored before +changing state. Existing bounded pending, completed, and non-DNS history carries the maximum +retained cutoff. Completed or non-DNS records alone never initiate interval-based recovery. + +The interval is an operator's hypothesis about timestamps and generation separation, not a +universal maximum datagram lifetime or an automatic expiry timer. Tail-before-first recovery +across generations is not guaranteed; replay protection ends when all retained history for the +key is evicted. Saturated signed 64-bit timestamps cannot initiate recovery. Without this option, +the existing fragment policy is unchanged. + The reassembler keys fragments by source, destination, protocol, and IPv4 identification. Payload bytes use the declared eight-octet offsets. A datagram reaches the existing UDP/DNS parser only when the first fragment, final fragment, and contiguous coverage are present. Its fragmented IP diff --git a/src/app.rs b/src/app.rs index 0f559d7..bd32722 100644 --- a/src/app.rs +++ b/src/app.rs @@ -164,6 +164,11 @@ fn display_fragment_mode(args: &AppConfig) { info!( "IPv4 fragment reassembly: enabled (incomplete responses may use first-fragment matching)" ); + if let Some(gap_ms) = args.ipv4_fragment_reuse_gap_ms { + info!( + "IPv4 ID reuse recovery gap: {gap_ms} ms (per-key forward capture timestamp assumption)" + ); + } } else if args.allow_fragments { info!( "IPv4 response fragment prefix matching: enabled (no reassembly; full UDP/DNS content unverified)" @@ -203,6 +208,8 @@ struct RunConfigSummary { output_format: OutputFormat, report_format: ReportFormat, match_timeout_ms: u64, + #[serde(skip_serializing_if = "Option::is_none")] + ipv4_fragment_reuse_gap_ms: Option, monotonic_capture: bool, dns_wire_fast_path: bool, max_dns_compression_jumps: usize, @@ -404,6 +411,7 @@ fn build_run_summary( output_format: args.format, report_format: args.report_format, match_timeout_ms: args.match_timeout_ms, + ipv4_fragment_reuse_gap_ms: args.ipv4_fragment_reuse_gap_ms, monotonic_capture: args.monotonic_capture, dns_wire_fast_path: args.dns_wire_fast_path, max_dns_compression_jumps: args.max_dns_compression_jumps, @@ -626,7 +634,8 @@ pub(crate) fn run(args: AppConfig) -> Result<(), AppRunError> { .map_err(|source| AppRunError::DnsProcessorInit { source })? .with_max_dns_compression_jumps(args.max_dns_compression_jumps) .with_allow_fragments(args.allow_fragments) - .with_full_fragments(args.full_fragments), + .with_full_fragments(args.full_fragments) + .with_ipv4_fragment_reuse_gap_micros(args.ipv4_fragment_reuse_gap_micros()), ); let mut packet_parser = match PacketParser::new_with_shutdown( &args.input_source, @@ -779,6 +788,7 @@ mod tests { format: OutputFormat::Csv, report_format: crate::config::ReportFormat::Text, match_timeout_ms: crate::config::DEFAULT_MATCH_TIMEOUT_MS, + ipv4_fragment_reuse_gap_ms: None, monotonic_capture: false, zstd: false, v2: false, @@ -829,6 +839,11 @@ mod tests { ); let serialized = serde_json::to_value(&summary).expect("summary serializes"); + assert!( + serialized["config"] + .get("ipv4_fragment_reuse_gap_ms") + .is_none() + ); let config = serialized .get("config") .and_then(serde_json::Value::as_object) @@ -857,6 +872,7 @@ mod tests { let mut config = test_config(); config.allow_fragments = true; config.full_fragments = true; + config.ipv4_fragment_reuse_gap_ms = Some(120_000); let summary = build_run_summary( &config, config.execution_budget(), @@ -872,6 +888,7 @@ mod tests { let serialized = serde_json::to_value(&summary).expect("summary serializes"); assert_eq!(serialized["config"]["allow_fragments"], true); assert_eq!(serialized["config"]["full_fragments"], true); + assert_eq!(serialized["config"]["ipv4_fragment_reuse_gap_ms"], 120_000); assert_eq!(serialized["metrics"]["fragmented_response_prefix_count"], 2); } @@ -999,6 +1016,7 @@ mod tests { format, report_format: crate::config::ReportFormat::Text, match_timeout_ms: crate::config::DEFAULT_MATCH_TIMEOUT_MS, + ipv4_fragment_reuse_gap_ms: None, monotonic_capture: false, zstd: false, v2: false, diff --git a/src/cli.rs b/src/cli.rs index 7e527f3..183abd8 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -7,7 +7,8 @@ use crate::config::{ AppConfig, DEFAULT_MATCH_TIMEOUT_MS, DEFAULT_MAX_DNS_COMPRESSION_JUMPS, InputSource, - MAX_MATCH_TIMEOUT_MS, OutputFormat, OutputTarget, ReportFormat, output_target_for_path, + MAX_IPV4_FRAGMENT_REUSE_GAP_MS, MAX_MATCH_TIMEOUT_MS, OutputFormat, OutputTarget, ReportFormat, + output_target_for_path, }; use anyhow::{Context, Result, anyhow, bail}; use clap::{Arg, ArgAction, ArgMatches, Command}; @@ -33,6 +34,7 @@ pub(crate) fn parse_args() -> Result { let env_max_dns_compression_jumps = env::var_os("DPP_MAX_DNS_COMPRESSION_JUMPS"); let env_allow_fragments = parse_env_bool("DPP_ALLOW_FRAGMENTS"); let env_full_fragments = parse_env_bool("DPP_FULL_FRAGMENTS"); + let env_ipv4_fragment_reuse_gap_ms = env::var_os("DPP_IPV4_FRAGMENT_REUSE_GAP_MS"); let env_monotonic_capture = parse_env_bool("DPP_MONOTONIC_CAPTURE"); let env_threads = env::var_os("DPP_THREADS"); let env_bonded = env::var_os("DPP_BONDED"); @@ -94,6 +96,12 @@ pub(crate) fn parse_args() -> Result { let (allow_fragments, full_fragments) = resolve_fragment_flags(&matches, env_allow_fragments, env_full_fragments); let monotonic_capture = matches.get_flag("monotonic_capture") || env_monotonic_capture; + let ipv4_fragment_reuse_gap_ms = resolve_ipv4_fragment_reuse_gap_ms( + &matches, + env_ipv4_fragment_reuse_gap_ms.as_deref(), + full_fragments, + monotonic_capture, + )?; validate_parquet_only_flags(format, zstd, v2)?; @@ -103,6 +111,7 @@ pub(crate) fn parse_args() -> Result { format, report_format, match_timeout_ms, + ipv4_fragment_reuse_gap_ms, monotonic_capture, zstd, v2, @@ -136,6 +145,8 @@ fn build_cli(version: &'static str) -> Command { Maximum compression-pointer jumps per DNS name; default is 32, 0 disables the limit (used if --max-dns-compression-jumps is not specified) DPP_ALLOW_FRAGMENTS Set to 'true' to match queries with observable first IPv4 response fragments without reassembly DPP_FULL_FRAGMENTS Set to 'true' to reassemble IPv4 fragments and enable first-fragment response matching + DPP_IPV4_FRAGMENT_REUSE_GAP_MS + Optional per-key forward timestamp gap for IPv4 ID reuse recovery in milliseconds; range 1..=9223372036854775, disabled by default; requires full fragments and conflicts with monotonic capture (used if --ipv4-fragment-reuse-gap-ms is not specified) DPP_MONOTONIC_CAPTURE Set to 'true' to assume globally monotonic packet timestamps, enable batched timeout eviction, and abort on timestamp regressions DPP_THREADS Maximum CPU execution budget (positive integer, capped to available CPUs; used if --threads is not specified) @@ -264,6 +275,14 @@ LICENSE INFORMATION: .help("Reassemble IPv4 fragments into complete datagrams; also enables --allow-fragments") .action(ArgAction::SetTrue), ) + .arg( + Arg::new("ipv4_fragment_reuse_gap_ms") + .long("ipv4-fragment-reuse-gap-ms") + .help("Recover reused IPv4 IDs after a per-key forward timestamp gap in milliseconds (opt-in capture assumption; requires --full-fragments and conflicts with --monotonic-capture)") + .value_name("MS") + .num_args(1) + .value_parser(parse_ipv4_fragment_reuse_gap_ms), + ) .arg( Arg::new("anonymize") .long("anonymize") @@ -322,6 +341,50 @@ fn resolve_match_timeout_ms(matches: &ArgMatches, env_value: Option<&str>) -> Re } } +fn resolve_ipv4_fragment_reuse_gap_ms( + matches: &ArgMatches, + env_value: Option<&OsStr>, + full_fragments: bool, + monotonic_capture: bool, +) -> Result> { + let gap_ms = if let Some(value) = matches.get_one::("ipv4_fragment_reuse_gap_ms") { + Some(*value) + } else if let Some(value) = env_value { + let value = value + .to_str() + .ok_or_else(|| anyhow!("DPP_IPV4_FRAGMENT_REUSE_GAP_MS must contain valid UTF-8"))?; + Some(parse_ipv4_fragment_reuse_gap_ms(value)?) + } else { + None + }; + if gap_ms.is_some() { + if !full_fragments { + bail!( + "'--ipv4-fragment-reuse-gap-ms' requires '--full-fragments' (or DPP_FULL_FRAGMENTS)." + ); + } + if monotonic_capture { + bail!( + "'--ipv4-fragment-reuse-gap-ms' cannot be combined with '--monotonic-capture' (or DPP_MONOTONIC_CAPTURE)." + ); + } + } + Ok(gap_ms) +} + +fn parse_ipv4_fragment_reuse_gap_ms(value: &str) -> Result { + let gap_ms = value + .parse::() + .with_context(|| format!("Failed to parse IPv4 fragment reuse gap from '{value}'"))?; + if gap_ms == 0 || gap_ms > MAX_IPV4_FRAGMENT_REUSE_GAP_MS { + bail!( + "IPv4 fragment reuse gap must be in 1..={} milliseconds.", + MAX_IPV4_FRAGMENT_REUSE_GAP_MS + ); + } + Ok(gap_ms) +} + fn resolve_thread_limit(matches: &ArgMatches, env_value: Option<&OsStr>) -> Result> { let value = if let Some(cli_value) = matches.get_one::("threads") { cli_value.as_str() @@ -896,6 +959,152 @@ mod tests { ); } + #[test] + fn ipv4_fragment_reuse_gap_is_disabled_by_default_and_cli_overrides_env() { + let defaults = build_cli("test") + .try_get_matches_from(["dpp", "input.pcap"]) + .expect("cli parses"); + assert_eq!( + resolve_ipv4_fragment_reuse_gap_ms(&defaults, None, false, true).unwrap(), + None + ); + assert_eq!( + resolve_ipv4_fragment_reuse_gap_ms(&defaults, Some(OsStr::new("120000")), true, false) + .unwrap(), + Some(120_000) + ); + let cli = build_cli("test") + .try_get_matches_from(["dpp", "--ipv4-fragment-reuse-gap-ms", "1000", "input.pcap"]) + .expect("cli parses"); + for env_value in ["120000", "invalid", "0"] { + assert_eq!( + resolve_ipv4_fragment_reuse_gap_ms(&cli, Some(OsStr::new(env_value)), true, false) + .unwrap(), + Some(1_000) + ); + } + } + + #[test] + fn ipv4_fragment_reuse_gap_accepts_exact_conversion_range_in_cli_and_env() { + let defaults = build_cli("test") + .try_get_matches_from(["dpp", "input.pcap"]) + .expect("cli parses"); + for expected in [1, MAX_IPV4_FRAGMENT_REUSE_GAP_MS] { + let value = expected.to_string(); + let cli = build_cli("test") + .try_get_matches_from([ + "dpp", + "--ipv4-fragment-reuse-gap-ms", + value.as_str(), + "input.pcap", + ]) + .expect("boundary parses"); + assert_eq!( + resolve_ipv4_fragment_reuse_gap_ms(&cli, None, true, false).unwrap(), + Some(expected) + ); + assert_eq!( + resolve_ipv4_fragment_reuse_gap_ms( + &defaults, + Some(OsStr::new(&value)), + true, + false + ) + .unwrap(), + Some(expected) + ); + } + } + + #[test] + fn invalid_ipv4_fragment_reuse_gaps_are_rejected_in_cli_and_env() { + let defaults = build_cli("test") + .try_get_matches_from(["dpp", "input.pcap"]) + .expect("cli parses"); + for value in [ + "0", + "-1", + "9223372036854776", + "18446744073709551616", + "invalid", + "", + ] { + assert!( + build_cli("test") + .try_get_matches_from([ + "dpp", + &format!("--ipv4-fragment-reuse-gap-ms={value}"), + "input.pcap", + ]) + .is_err(), + "invalid CLI value {value:?} must not fall back to ENV" + ); + assert!( + resolve_ipv4_fragment_reuse_gap_ms(&defaults, Some(OsStr::new(value)), true, false) + .is_err(), + "invalid ENV value {value:?}" + ); + } + } + + #[test] + fn ipv4_fragment_reuse_gap_validates_resolved_cli_and_env_modes() { + for cli_full in [false, true] { + for env_full in [false, true] { + for cli_monotonic in [false, true] { + for env_monotonic in [false, true] { + for env_gap in [false, true] { + let mut argv = vec!["dpp", "input.pcap"]; + if cli_full { + argv.push("--full-fragments"); + } + if cli_monotonic { + argv.push("--monotonic-capture"); + } + if !env_gap { + argv.extend(["--ipv4-fragment-reuse-gap-ms", "120000"]); + } + let matches = build_cli("test") + .try_get_matches_from(argv) + .expect("cli parses"); + let (_, full) = resolve_fragment_flags(&matches, false, env_full); + let monotonic = matches.get_flag("monotonic_capture") || env_monotonic; + let result = resolve_ipv4_fragment_reuse_gap_ms( + &matches, + env_gap.then_some(OsStr::new("120000")), + full, + monotonic, + ); + assert_eq!(result.is_ok(), full && !monotonic); + if full && !monotonic { + assert_eq!(result.unwrap(), Some(120_000)); + } + } + } + } + } + } + } + + #[cfg(unix)] + #[test] + fn non_utf8_ipv4_fragment_reuse_gap_env_requires_cli_override() { + use std::os::unix::ffi::OsStrExt; + let invalid = OsStr::from_bytes(&[0xff]); + let defaults = build_cli("test") + .try_get_matches_from(["dpp", "input.pcap"]) + .expect("cli parses"); + assert!(resolve_ipv4_fragment_reuse_gap_ms(&defaults, Some(invalid), true, false).is_err()); + let cli = build_cli("test") + .try_get_matches_from(["dpp", "--ipv4-fragment-reuse-gap-ms", "1", "input.pcap"]) + .expect("cli parses"); + assert_eq!( + resolve_ipv4_fragment_reuse_gap_ms(&cli, Some(invalid), true, false).unwrap(), + Some(1) + ); + } + #[test] fn zero_match_timeout_is_rejected() { let error = parse_match_timeout_ms("0").expect_err("zero timeout is invalid"); diff --git a/src/config.rs b/src/config.rs index 78676b8..26c5ecc 100644 --- a/src/config.rs +++ b/src/config.rs @@ -23,6 +23,9 @@ pub(crate) const DEFAULT_MAX_DNS_COMPRESSION_JUMPS: usize = 32; /// Maximum supported DNS match timeout, expressed in milliseconds. pub(crate) const MAX_MATCH_TIMEOUT_MS: u64 = 5_000; +/// Maximum IPv4 reuse gap that can be converted to signed microseconds. +pub(crate) const MAX_IPV4_FRAGMENT_REUSE_GAP_MS: u64 = (i64::MAX / 1_000) as u64; + /// Logical shard multiplier used to reduce hot-shard skew relative to worker count. pub(crate) const MATCHER_SHARD_FACTOR: usize = 4; @@ -159,6 +162,7 @@ pub(crate) struct AppConfig { pub(crate) format: OutputFormat, pub(crate) report_format: ReportFormat, pub(crate) match_timeout_ms: u64, + pub(crate) ipv4_fragment_reuse_gap_ms: Option, pub(crate) monotonic_capture: bool, pub(crate) zstd: bool, pub(crate) v2: bool, @@ -239,6 +243,15 @@ impl AppConfig { .expect("validated timeout milliseconds fit into microseconds") } + pub(crate) fn ipv4_fragment_reuse_gap_micros(&self) -> Option { + self.ipv4_fragment_reuse_gap_ms.map(|gap_ms| { + i64::try_from(gap_ms) + .expect("validated IPv4 fragment reuse gap fits into i64") + .checked_mul(1_000) + .expect("validated IPv4 fragment reuse gap fits into microseconds") + }) + } + pub(crate) fn anonymize_key_path(&self) -> Option<&Path> { self.anonymize.as_deref() } @@ -282,6 +295,7 @@ mod tests { format: OutputFormat::Csv, report_format: ReportFormat::Text, match_timeout_ms: DEFAULT_MATCH_TIMEOUT_MS, + ipv4_fragment_reuse_gap_ms: None, monotonic_capture: false, zstd: false, v2: false, @@ -341,6 +355,19 @@ mod tests { assert_eq!(config.match_timeout_micros(), 1_200_000); } + #[test] + fn ipv4_fragment_reuse_gap_conversion_preserves_option_and_range_boundaries() { + let mut config = test_config(); + assert_eq!(config.ipv4_fragment_reuse_gap_micros(), None); + config.ipv4_fragment_reuse_gap_ms = Some(1); + assert_eq!(config.ipv4_fragment_reuse_gap_micros(), Some(1_000)); + config.ipv4_fragment_reuse_gap_ms = Some(MAX_IPV4_FRAGMENT_REUSE_GAP_MS); + assert_eq!( + config.ipv4_fragment_reuse_gap_micros(), + Some(i64::MAX - i64::MAX % 1_000) + ); + } + #[test] fn stdout_output_sentinel_is_detected() { assert_eq!(output_target_for_path(Path::new("-")), OutputTarget::Stdout); diff --git a/src/dns_processor.rs b/src/dns_processor.rs index ba14592..ce3e48f 100644 --- a/src/dns_processor.rs +++ b/src/dns_processor.rs @@ -28,6 +28,7 @@ pub struct DnsProcessor { dns_wire_fast_path: bool, max_dns_compression_jumps: usize, match_timeout_micros: i64, + ipv4_fragment_reuse_gap_micros: Option, monotonic_capture: bool, allow_fragments: bool, full_fragments: bool, @@ -71,6 +72,7 @@ impl DnsProcessor { dns_wire_fast_path, max_dns_compression_jumps: DEFAULT_MAX_DNS_COMPRESSION_JUMPS, match_timeout_micros, + ipv4_fragment_reuse_gap_micros: None, monotonic_capture, allow_fragments: false, full_fragments: false, @@ -87,6 +89,11 @@ impl DnsProcessor { self } + pub fn with_ipv4_fragment_reuse_gap_micros(mut self, gap_micros: Option) -> Self { + self.ipv4_fragment_reuse_gap_micros = gap_micros; + self + } + pub fn with_full_fragments(mut self, full_fragments: bool) -> Self { self.full_fragments = full_fragments; self.allow_fragments |= full_fragments; diff --git a/src/dns_processor/pipeline.rs b/src/dns_processor/pipeline.rs index 107d421..3344d7e 100644 --- a/src/dns_processor/pipeline.rs +++ b/src/dns_processor/pipeline.rs @@ -46,6 +46,7 @@ struct FragmentProcessingConfig { allow_fragments: bool, full_fragments: bool, match_timeout_micros: i64, + ipv4_fragment_reuse_gap_micros: Option, monotonic_capture: bool, } @@ -636,9 +637,14 @@ struct FragmentRoutingState { } impl FragmentRoutingState { - fn new(match_timeout_micros: i64, monotonic_capture: bool) -> Self { + fn new( + match_timeout_micros: i64, + monotonic_capture: bool, + reuse_gap_micros: Option, + ) -> Self { Self { - reassembler: Ipv4FragmentReassembler::new(match_timeout_micros, monotonic_capture), + reassembler: Ipv4FragmentReassembler::new(match_timeout_micros, monotonic_capture) + .with_reuse_gap_micros(reuse_gap_micros), ready: Vec::new(), } } @@ -746,6 +752,7 @@ fn run_phase_processing_worker( FragmentRoutingState::new( dns_processor.match_timeout_micros, dns_processor.monotonic_capture, + dns_processor.ipv4_fragment_reuse_gap_micros, ) }); @@ -847,11 +854,17 @@ fn run_parser_stage( allow_fragments, full_fragments, match_timeout_micros, + ipv4_fragment_reuse_gap_micros, monotonic_capture, } = fragment_config; let mut batch_seq = 0_u64; - let mut fragment_routing = - full_fragments.then(|| FragmentRoutingState::new(match_timeout_micros, monotonic_capture)); + let mut fragment_routing = full_fragments.then(|| { + FragmentRoutingState::new( + match_timeout_micros, + monotonic_capture, + ipv4_fragment_reuse_gap_micros, + ) + }); loop { let received = batch_rx.recv(); @@ -1161,6 +1174,7 @@ fn run_staged_processing_pipeline_with_spawner( let allow_fragments = dns_processor.allow_fragments; let full_fragments = dns_processor.full_fragments; let match_timeout_micros = dns_processor.match_timeout_micros; + let ipv4_fragment_reuse_gap_micros = dns_processor.ipv4_fragment_reuse_gap_micros; let monotonic_capture = dns_processor.monotonic_capture; let affinity_plan = affinity_plan.clone(); move || { @@ -1177,6 +1191,7 @@ fn run_staged_processing_pipeline_with_spawner( allow_fragments, full_fragments, match_timeout_micros, + ipv4_fragment_reuse_gap_micros, monotonic_capture, }, ) @@ -1579,11 +1594,22 @@ mod tests { staged: bool, monotonic: bool, signals: WorkerShutdownSignals, + ) -> (PipelineCounters, Vec) { + run_fragment_test_batches_with_reuse_gap(batches, staged, monotonic, signals, None) + } + + fn run_fragment_test_batches_with_reuse_gap( + batches: Vec, + staged: bool, + monotonic: bool, + signals: WorkerShutdownSignals, + reuse_gap_micros: Option, ) -> (PipelineCounters, Vec) { let processor = Arc::new( DnsProcessor::new_with_runtime_options(None, false, 1_200_000, monotonic) .expect("processor initializes") - .with_full_fragments(true), + .with_full_fragments(true) + .with_ipv4_fragment_reuse_gap_micros(reuse_gap_micros), ); let (batch_tx, batch_rx) = crossbeam::channel::unbounded(); for batch in batches { @@ -1615,6 +1641,47 @@ mod tests { (counters, records) } + #[test] + fn ipv4_fragment_reuse_gap_reaches_both_pipeline_models() { + let (mut new_query, old_response) = fragment_ordering_query_response(); + let mut new_response = old_response.clone(); + // Same flow and fragment key; a different QNAME distinguishes the DNS outcomes. + new_query[55] = b'a'; + new_response[55] = b'a'; + let (old_first, _) = split_ipv4_test_packet(&old_response); + let (new_first, new_tail) = split_ipv4_test_packet(&new_response); + for staged in [false, true] { + for reuse_gap_micros in [None, Some(120_000_000)] { + let (counters, records) = run_fragment_test_batches_with_reuse_gap( + vec![ + vec![test_packet(&old_first, 1_100_000, 0)], + vec![ + test_packet(&new_query, 122_000_000, 1), + test_packet(&new_first, 122_100_000, 2), + test_packet(&new_tail, 122_100_001, 3), + ], + ], + staged, + false, + test_worker_signals(false), + reuse_gap_micros, + ); + let recovered = reuse_gap_micros.is_some(); + assert_eq!( + counters.matched_query_response_count, + usize::from(recovered) + ); + assert_eq!(counters.timeout_query_count, usize::from(!recovered)); + assert_eq!(records.len(), 1); + assert_eq!(records[0].name.as_str(), "axample.com"); + assert_eq!( + records[0].response_timestamp, + recovered.then_some(122_100_001) + ); + } + } + } + #[test] fn full_fragments_preserve_delayed_query_order_in_both_pipeline_models() { let (query, response) = fragment_ordering_query_response(); @@ -1715,7 +1782,7 @@ mod tests { let (query, _) = fragment_ordering_query_response(); let (_, tail) = split_ipv4_test_packet(&query); for byte_bound in [false, true] { - let mut routing = FragmentRoutingState::new(1_200_000, false); + let mut routing = FragmentRoutingState::new(1_200_000, false, None); let plan = ShardRoutingPlan::new(1, 1); let blocked = routing.route_batch( vec![test_packet(&tail, 0, 0), test_packet(&query, 1, 1)], @@ -1741,29 +1808,108 @@ mod tests { fn full_fragments_drain_accepted_backlog_after_signal_or_intake_error() { let (query, response) = fragment_ordering_query_response(); let (_, tail) = split_ipv4_test_packet(&query); + for staged in [false, true] { + for signal in [false, true] { + for (monotonic, reuse_gap_micros) in + [(true, None), (false, None), (false, Some(120_000_000))] + { + let signals = test_worker_signals(signal); + signals.intake_failed.store(!signal, AtomicOrdering::SeqCst); + let mut unmatched = query.clone(); + unmatched[42..44].copy_from_slice(&0x9876_u16.to_be_bytes()); + let (counters, records) = run_fragment_test_batches_with_reuse_gap( + vec![vec![ + test_packet(&tail, 1_000_000, 0), + test_packet(&query, 1_100_000, 1), + test_packet(&response, 1_200_000, 2), + test_packet(&unmatched, 1_300_000, 3), + ]], + staged, + monotonic, + signals, + reuse_gap_micros, + ); + assert_eq!(counters.dns_query_count, 2); + assert_eq!(counters.matched_query_response_count, 1); + assert_eq!(counters.timeout_query_count, 0); + assert_eq!(records.len(), 1); + assert_eq!(records[0].request_timestamp, 1_100_000); + assert_eq!(records[0].response_timestamp, Some(1_200_000)); + } + } + } + } + + #[test] + fn ipv4_reuse_recovery_drains_both_prefixes_without_timeouts_after_interruption() { + let (old_query, response) = fragment_ordering_query_response(); + let mut dns = response[42..].to_vec(); + dns[10..12].copy_from_slice(&1_u16.to_be_bytes()); + // A complete question fits in the first fragment; the OPT tail is absent. + dns.extend_from_slice(&[0, 0, 41, 4, 208, 0, 0, 0, 0, 0, 0]); + let old_response = + make_udp_dns_packet_with_payload([8, 8, 8, 8], [10, 0, 0, 1], 53, 53_000, &dns); + let mut new_query = old_query.clone(); + let mut new_response = old_response.clone(); + new_query[55] = b'a'; + new_response[55] = b'a'; + let first_prefix = |response: &[u8]| { + let mut first = response[..74].to_vec(); + first[16..18].copy_from_slice(&60_u16.to_be_bytes()); + first[18..20].copy_from_slice(&0x4321_u16.to_be_bytes()); + first[20..22].copy_from_slice(&0x2000_u16.to_be_bytes()); + first + }; + let old_first = first_prefix(&old_response); + let new_first = first_prefix(&new_response); + let mut unmatched = new_query.clone(); + unmatched[42..44].copy_from_slice(&0x9876_u16.to_be_bytes()); + for staged in [false, true] { for signal in [false, true] { let signals = test_worker_signals(signal); signals.intake_failed.store(!signal, AtomicOrdering::SeqCst); - let mut unmatched = query.clone(); - unmatched[42..44].copy_from_slice(&0x9876_u16.to_be_bytes()); - let (counters, records) = run_fragment_test_batches( - vec![vec![ - test_packet(&tail, 1_000_000, 0), - test_packet(&query, 1_100_000, 1), - test_packet(&response, 1_200_000, 2), - test_packet(&unmatched, 1_300_000, 3), - ]], + let (counters, records) = run_fragment_test_batches_with_reuse_gap( + vec![ + vec![ + test_packet(&old_query, 1_000_000, 0), + test_packet(&old_first, 1_100_000, 1), + ], + vec![ + test_packet(&new_query, 122_000_000, 2), + test_packet(&new_first, 122_100_000, 3), + test_packet(&unmatched, 122_200_000, 4), + ], + ], staged, - true, + false, signals, + Some(120_000_000), ); - assert_eq!(counters.dns_query_count, 2); - assert_eq!(counters.matched_query_response_count, 1); + assert_eq!(counters.dns_query_count, 3); + assert_eq!(counters.dns_response_count, 2); + assert_eq!(counters.fragmented_response_prefix_count, 2); + assert_eq!(counters.matched_query_response_count, 2); assert_eq!(counters.timeout_query_count, 0); - assert_eq!(records.len(), 1); - assert_eq!(records[0].request_timestamp, 1_100_000); - assert_eq!(records[0].response_timestamp, Some(1_200_000)); + assert_eq!(records.len(), 2); + let outcomes = records + .iter() + .map(|record| { + assert!(record.response_code.is_none()); + ( + record.name.as_str(), + record.request_timestamp, + record.response_timestamp, + ) + }) + .collect::>(); + assert_eq!( + outcomes, + vec![ + ("example.com", 1_000_000, Some(1_100_000)), + ("axample.com", 122_000_000, Some(122_100_000)), + ] + ); } } } @@ -2338,7 +2484,7 @@ mod tests { .next_batch(PACKET_BATCH_SIZE) .expect("batch reads") .expect("first batch exists"); - let mut fragment_routing = FragmentRoutingState::new(1_200_000, true); + let mut fragment_routing = FragmentRoutingState::new(1_200_000, true, None); let routed = fragment_routing.route_batch(first_batch, &ShardRoutingPlan::new(1, 1)); assert_eq!(routed.batch_max_timestamp_micros, Some(1_000_000)); diff --git a/src/dns_processor/reassembly.rs b/src/dns_processor/reassembly.rs index 02dd07d..b4c6273 100644 --- a/src/dns_processor/reassembly.rs +++ b/src/dns_processor/reassembly.rs @@ -9,7 +9,7 @@ //! //! Later fragments have no UDP ports, so this state has one owner in the routing stage. //! A completed datagram replaces its first fragment with one ordinary IPv4 packet. An -//! incomplete first fragment is released only when it expires, is evicted, or input ends; +//! incomplete first fragment is released on expiry, eviction, input end, or opted-in reuse; //! the existing `--allow-fragments` parser then decides whether its DNS prefix is usable. use std::collections::{BTreeMap, HashMap}; @@ -69,6 +69,7 @@ struct CompletedDatagram { payload: Arc<[u8]>, observed_timestamp: i64, arrival_order: u64, + reuse_cutoff_timestamp: Option, } impl CompletedDatagram { @@ -80,6 +81,7 @@ impl CompletedDatagram { struct NonDnsKey { observed_timestamp: i64, arrival_order: u64, + reuse_cutoff_timestamp: Option, } enum FragmentInput { @@ -93,6 +95,8 @@ enum FragmentInput { struct PendingDatagram { arrival_order: u64, first_seen_timestamp: i64, + max_observed_timestamp: i64, + reuse_cutoff_timestamp: Option, first: Option, first_header_len: Option, first_l3_offset: Option, @@ -110,6 +114,8 @@ impl PendingDatagram { Self { arrival_order, first_seen_timestamp: timestamp, + max_observed_timestamp: timestamp, + reuse_cutoff_timestamp: None, first: None, first_header_len: None, first_l3_offset: None, @@ -372,6 +378,7 @@ pub(super) struct Ipv4FragmentReassembler { max_seen_timestamp: Option, timeout_micros: i64, monotonic_capture: bool, + reuse_gap_micros: Option, } impl Ipv4FragmentReassembler { @@ -392,9 +399,35 @@ impl Ipv4FragmentReassembler { max_seen_timestamp: None, timeout_micros: match_timeout_micros.max(1), monotonic_capture, + reuse_gap_micros: None, } } + pub(super) fn with_reuse_gap_micros(mut self, gap: Option) -> Self { + debug_assert!(gap.is_none_or(|gap| gap > 0)); + debug_assert!(gap.is_none() || !self.monotonic_capture); + self.reuse_gap_micros = gap; + self + } + + fn retained_reuse_cutoff(&self, key: &FragmentKey) -> Option { + // Preserve the default path's lookups when recovery is disabled. + self.reuse_gap_micros?; + self.pending + .get(key) + .and_then(|entry| entry.reuse_cutoff_timestamp) + .max( + self.recently_completed + .get(key) + .and_then(|entry| entry.reuse_cutoff_timestamp), + ) + .max( + self.known_non_dns + .get(key) + .and_then(|entry| entry.reuse_cutoff_timestamp), + ) + } + /// Returns decodable packets and the oldest unresolved fragment timestamp. /// The caller must cap monotonic matcher eviction at that timestamp until resolution. pub(super) fn process_batch(&mut self, input: PacketBatch) -> (PacketBatch, Option) { @@ -410,12 +443,20 @@ impl Ipv4FragmentReassembler { FragmentInput::Ordinary => output.push(packet), FragmentInput::Skip => {} FragmentInput::NonDnsFirst(key) => { + let cutoff = self.retained_reuse_cutoff(&key); + if cutoff.is_some_and(|cutoff| packet.timestamp_micros <= cutoff) { + continue; + } // Later fragments have no ports. Drop any tails retained before // this first fragment established that the datagram is not DNS. self.take_entry(&key); - self.remember_non_dns(key, packet.timestamp_micros); + self.remember_non_dns(key, packet.timestamp_micros, cutoff); } FragmentInput::Part(part) => { + let cutoff = self.retained_reuse_cutoff(&part.key); + if cutoff.is_some_and(|cutoff| packet.timestamp_micros <= cutoff) { + continue; + } self.expire_completed_key_at(&part.key, packet.timestamp_micros); self.expire_non_dns_key_at(&part.key, packet.timestamp_micros); self.expire_key_at(&part.key, packet.timestamp_micros, &mut output); @@ -428,7 +469,7 @@ impl Ipv4FragmentReassembler { // fragments may now be assembled again. self.take_non_dns(&part.key); } - self.accept_part(part, packet, &mut output); + self.accept_part(part, packet, cutoff, &mut output); // One input batch may contain many large fragments. Enforce the // memory bound per fragment, not only after the whole batch. self.enforce_limits(&mut output); @@ -469,10 +510,30 @@ impl Ipv4FragmentReassembler { output } - fn accept_part(&mut self, part: FragmentPart, packet: PacketData, output: &mut PacketBatch) { + fn accept_part( + &mut self, + part: FragmentPart, + packet: PacketData, + mut cutoff: Option, + output: &mut PacketBatch, + ) { let key = part.key.clone(); let observed_timestamp = packet.timestamp_micros; let mut existing = self.take_entry(&key); + if let Some(gap) = self.reuse_gap_micros + && part.offset == 0 + && let Some(entry) = existing.as_ref() + && let Some(recovered_cutoff) = + recovery_cutoff(entry.max_observed_timestamp, observed_timestamp, gap) + { + // Compare before admitting this first fragment, including exact duplicates. + // The old candidate owns the fallback decision and all its retained bytes. + if let Some(first) = existing.take().and_then(PendingDatagram::into_fallback) { + output.push(first); + } + self.take_completion(&key); + cutoff = cutoff.max(Some(recovered_cutoff)); + } if existing.as_ref().is_some_and(|entry| { entry.recent_payload.is_some() && entry.first.is_some() @@ -495,9 +556,15 @@ impl Ipv4FragmentReassembler { self.next_arrival_order = self.next_arrival_order.wrapping_add(1); PendingDatagram::new(arrival_order, observed_timestamp, recent_payload) }); + if self.reuse_gap_micros.is_some() { + entry.reuse_cutoff_timestamp = entry.reuse_cutoff_timestamp.max(cutoff); + // Exact duplicates and insert errors are admitted observations too. + entry.max_observed_timestamp = entry.max_observed_timestamp.max(observed_timestamp); + } match entry.insert(part, packet) { Ok(true) => { let previous_payload = entry.recent_payload.clone(); + let cutoff = entry.reuse_cutoff_timestamp; let header_len = entry .first_header_len .expect("complete datagram has first header"); @@ -514,8 +581,23 @@ impl Ipv4FragmentReassembler { key.clone(), Arc::<[u8]>::from(payload), observed_timestamp, + cutoff, ); output.push(packet); + } else if cutoff.is_some() { + if let Some(completed) = self.recently_completed.get_mut(&key) { + completed.reuse_cutoff_timestamp = + completed.reuse_cutoff_timestamp.max(cutoff); + } else { + // A pending duplicate can outlive its bounded cache entry. + // Preserve its fence when completion suppresses the output. + self.remember_completion( + key, + previous_payload.expect("duplicate has a recent payload"), + observed_timestamp, + cutoff, + ); + } } } } @@ -547,14 +629,19 @@ impl Ipv4FragmentReassembler { key: FragmentKey, payload: Arc<[u8]>, observed_timestamp: i64, + reuse_cutoff_timestamp: Option, ) { - self.take_completion(&key); + let reuse_cutoff_timestamp = reuse_cutoff_timestamp.max( + self.take_completion(&key) + .and_then(|entry| entry.reuse_cutoff_timestamp), + ); let arrival_order = self.next_completed_order; self.next_completed_order = self.next_completed_order.wrapping_add(1); let completed = CompletedDatagram { payload, observed_timestamp, arrival_order, + reuse_cutoff_timestamp, }; self.completed_bytes += completed.memory_bytes() + key.vlan_context.memory_bytes(); self.completed_order.insert(arrival_order, key.clone()); @@ -577,8 +664,16 @@ impl Ipv4FragmentReassembler { Some(completed) } - fn remember_non_dns(&mut self, key: FragmentKey, observed_timestamp: i64) { - self.take_non_dns(&key); + fn remember_non_dns( + &mut self, + key: FragmentKey, + observed_timestamp: i64, + reuse_cutoff_timestamp: Option, + ) { + let reuse_cutoff_timestamp = reuse_cutoff_timestamp.max( + self.take_non_dns(&key) + .and_then(|entry| entry.reuse_cutoff_timestamp), + ); let arrival_order = self.next_non_dns_order; self.next_non_dns_order = self.next_non_dns_order.wrapping_add(1); self.non_dns_bytes += RECENT_ENTRY_OVERHEAD_ESTIMATE + key.vlan_context.memory_bytes(); @@ -588,6 +683,7 @@ impl Ipv4FragmentReassembler { NonDnsKey { observed_timestamp, arrival_order, + reuse_cutoff_timestamp, }, ); while self.known_non_dns.len() > MAX_NON_DNS_KEYS || self.non_dns_bytes > MAX_NON_DNS_BYTES @@ -737,6 +833,19 @@ impl Ipv4FragmentReassembler { } } +/// A checked exclusive fence; saturated capture timestamps have unknown provenance. +fn recovery_cutoff(old_max: i64, timestamp: i64, gap: i64) -> Option { + if matches!(old_max, i64::MIN | i64::MAX) || matches!(timestamp, i64::MIN | i64::MAX) { + return None; + } + let old_max = i128::from(old_max); + let gap = i128::from(gap); + if i128::from(timestamp) - old_max <= gap { + return None; + } + i64::try_from(old_max + gap).ok() +} + fn classify_fragment(packet: &PacketData) -> FragmentInput { let bytes = packet.data.as_slice(); let Ok((ethertype, l3_offset, vlan_context)) = @@ -876,6 +985,569 @@ mod tests { packet } + fn changed_first(ordinal: u64, timestamp: i64) -> PacketData { + let mut first = packet(0, true, ordinal, timestamp); + let mut bytes = first.data.as_slice().to_vec(); + bytes[ETHERNET_HEADER_LEN + IPV4_MIN_HEADER_LEN + UDP_HEADER_LEN] ^= 1; + first.data = PacketPayload::owned(bytes.into_boxed_slice()); + first + } + + fn with_id(mut packet: PacketData, id: u16) -> PacketData { + let mut bytes = packet.data.as_slice().to_vec(); + bytes[ETHERNET_HEADER_LEN + 4..ETHERNET_HEADER_LEN + 6].copy_from_slice(&id.to_be_bytes()); + packet.data = PacketPayload::owned(bytes.into_boxed_slice()); + packet + } + + fn recovering() -> Ipv4FragmentReassembler { + Ipv4FragmentReassembler::new(1_200_000, false).with_reuse_gap_micros(Some(1_000)) + } + + #[test] + fn late_first_recovers_live_and_rejected_candidates_only_when_enabled() { + for enabled in [false, true] { + for rejected in [false, true] { + let mut assembler = Ipv4FragmentReassembler::new(1_200_000, false) + .with_reuse_gap_micros(enabled.then_some(120_000_000)); + assembler.process_batch(vec![packet(0, true, 1, 100)]); + if rejected { + assert!( + assembler + .process_batch(vec![changed_first(2, 101)]) + .0 + .is_empty() + ); + } + let (output, pending) = assembler.process_batch(vec![ + changed_first(3, 121_000_100), + packet(24, false, 4, 121_000_101), + ]); + assert_eq!( + output.len(), + if enabled { + 1 + usize::from(!rejected) + } else { + 0 + } + ); + assert_eq!(pending, None); + if enabled { + assert_eq!(output.last().unwrap().packet_ordinal, 4); + assert_eq!(&output.last().unwrap().data[20..22], &[0, 0]); + if !rejected { + assert_eq!(output[0].packet_ordinal, 1); + assert_eq!(&output[0].data[20..22], &MORE_FRAGMENTS.to_be_bytes()); + } + } + assert!(assembler.finish().is_empty()); + assert!(assembler.finish().is_empty()); + } + } + } + + #[test] + fn reuse_gap_is_strict_and_can_restart_an_identical_first() { + for identical in [false, true] { + for extra_micros in [0, 1] { + let mut assembler = recovering(); + assembler.process_batch(vec![packet(0, true, 1, 100)]); + let timestamp = 1_100 + extra_micros; + let first = if identical { + packet(0, true, 2, timestamp) + } else { + changed_first(2, timestamp) + }; + let (output, _) = assembler.process_batch(vec![first]); + assert_eq!(output.len(), extra_micros as usize); + let entry = assembler.pending.values().next().unwrap(); + assert_eq!( + entry.reuse_cutoff_timestamp, + (extra_micros == 1).then_some(1_100) + ); + assert_eq!( + entry.first_seen_timestamp, + if extra_micros == 1 { timestamp } else { 100 } + ); + let (complete, _) = + assembler.process_batch(vec![packet(24, false, 3, timestamp + 1)]); + assert_eq!(complete.len(), usize::from(identical || extra_micros == 1)); + assert!(assembler.finish().is_empty()); + } + } + } + + #[test] + fn admitted_duplicates_tails_and_rejections_advance_candidate_max() { + for observation in [ + packet(0, true, 2, 150), + packet(32, false, 2, 150), + packet(16, false, 2, 150), + ] { + let mut assembler = recovering(); + assembler.process_batch(vec![packet(0, true, 1, 100), observation]); + assert_eq!( + assembler + .pending + .values() + .next() + .unwrap() + .max_observed_timestamp, + 150 + ); + assert!( + assembler + .process_batch(vec![changed_first(3, 1_101)]) + .0 + .is_empty() + ); + let entry = assembler.pending.values().next().unwrap(); + assert!( + entry.rejected, + "a fresh observation prevents premature recovery" + ); + assert_eq!(entry.max_observed_timestamp, 1_101); + assembler.process_batch(vec![packet(16, false, 4, 1_250)]); + assert_eq!( + assembler + .pending + .values() + .next() + .unwrap() + .max_observed_timestamp, + 1_250 + ); + assert!( + assembler + .process_batch(vec![changed_first(5, 2_250)]) + .0 + .is_empty() + ); + let (output, _) = + assembler.process_batch(vec![changed_first(6, 3_251), packet(24, false, 7, 3_252)]); + assert_eq!(output.len(), 1, "rejected generations have no fallback"); + assert_eq!(output[0].packet_ordinal, 7); + assert!(assembler.finish().is_empty()); + } + } + + #[test] + fn saved_cutoff_discards_old_first_tail_and_non_dns_before_and_after_completion() { + let mut assembler = recovering(); + assembler.process_batch(vec![packet(0, true, 1, 100)]); + assert_eq!( + assembler + .process_batch(vec![changed_first(2, 1_101)]) + .0 + .len(), + 1 + ); + for completed in [false, true] { + for timestamp in [-10_000, 100, 1_100] { + for replay in [ + packet(0, true, 10, timestamp), + packet(24, false, 11, timestamp), + non_dns_first(12, timestamp), + ] { + let pending_bytes = assembler.pending_bytes; + let completed_bytes = assembler.completed_bytes; + assert!(assembler.process_batch(vec![replay]).0.is_empty()); + assert_eq!(assembler.pending_bytes, pending_bytes); + assert_eq!(assembler.completed_bytes, completed_bytes); + assert!(assembler.known_non_dns.is_empty()); + if !completed { + let entry = assembler.pending.values().next().unwrap(); + assert!(!entry.rejected); + assert_eq!(entry.max_observed_timestamp, 1_101); + } else { + assert!(assembler.pending.is_empty()); + } + } + } + if !completed { + assert_eq!( + assembler + .process_batch(vec![packet(24, false, 3, 1_102)]) + .0 + .len(), + 1 + ); + assert_eq!( + assembler + .recently_completed + .values() + .next() + .unwrap() + .reuse_cutoff_timestamp, + Some(1_100) + ); + } + } + assert!(assembler.finish().is_empty()); + } + + #[test] + fn cutoff_survives_non_dns_and_duplicate_completion_transfers() { + let mut assembler = recovering(); + assembler.process_batch(vec![packet(0, true, 1, 100), changed_first(2, 1_101)]); + assembler.process_batch(vec![non_dns_first(3, 1_102)]); + assert!(assembler.pending.is_empty()); + assert_eq!( + assembler + .known_non_dns + .values() + .next() + .unwrap() + .reuse_cutoff_timestamp, + Some(1_100) + ); + assert!( + assembler + .process_batch(vec![packet(0, true, 4, 1_100)]) + .0 + .is_empty() + ); + assert_eq!(assembler.known_non_dns.len(), 1); + assert_eq!( + assembler + .process_batch(vec![changed_first(5, 1_103), packet(24, false, 6, 1_104)]) + .0 + .len(), + 1 + ); + assembler.process_batch(vec![non_dns_first(7, 1_105)]); + assert!( + assembler + .process_batch(vec![changed_first(8, 1_106), packet(24, false, 9, 1_107)]) + .0 + .is_empty() + ); + assert!(assembler.pending.is_empty()); + assert!(assembler.known_non_dns.is_empty()); + assert_eq!( + assembler + .recently_completed + .values() + .next() + .unwrap() + .reuse_cutoff_timestamp, + Some(1_100) + ); + assert!( + assembler + .process_batch(vec![ + packet(0, true, 10, 1_100), + packet(24, false, 11, 1_100) + ]) + .0 + .is_empty() + ); + assert!(assembler.finish().is_empty()); + } + + #[test] + fn completed_history_alone_does_not_restart_but_pending_duplicate_can() { + let mut assembler = recovering(); + assert_eq!( + assembler + .process_batch(vec![packet(0, true, 1, 100), packet(24, false, 2, 101)]) + .0 + .len(), + 1 + ); + assert!( + assembler + .process_batch(vec![packet(0, true, 3, 1_102), packet(24, false, 4, 1_103)]) + .0 + .is_empty() + ); + assert_eq!( + assembler + .recently_completed + .values() + .next() + .unwrap() + .reuse_cutoff_timestamp, + None + ); + assembler.process_batch(vec![packet(0, true, 5, 1_104)]); + assert!( + assembler + .process_batch(vec![packet(0, true, 6, 2_105)]) + .0 + .is_empty(), + "duplicate candidate has no prefix fallback" + ); + assert!( + assembler.recently_completed.is_empty(), + "restart retires the old payload history" + ); + assert_eq!( + assembler + .process_batch(vec![packet(24, false, 7, 2_106)]) + .0 + .len(), + 1 + ); + assert!(assembler.finish().is_empty()); + } + + #[test] + fn pending_duplicate_retains_fence_when_completion_cache_is_evicted() { + let mut assembler = recovering(); + assembler.process_batch(vec![ + packet(0, true, 1, 100), + changed_first(2, 1_101), + packet(24, false, 3, 1_102), + ]); + let key = assembler.recently_completed.keys().next().unwrap().clone(); + assembler.process_batch(vec![changed_first(4, 1_103)]); + for id in 0..MAX_RECENT_COMPLETIONS { + let mut other = key.clone(); + other.identification = id as u16; + assembler.remember_completion(other, Arc::from(&b"payload"[..]), 2_000, None); + } + assert!(!assembler.recently_completed.contains_key(&key)); + assert!( + assembler + .process_batch(vec![packet(24, false, 5, 1_104)]) + .0 + .is_empty() + ); + assert_eq!( + assembler + .recently_completed + .get(&key) + .unwrap() + .reuse_cutoff_timestamp, + Some(1_100) + ); + assert!( + assembler + .process_batch(vec![packet(0, true, 6, 1_100), packet(24, false, 7, 1_100)]) + .0 + .is_empty() + ); + assert!(assembler.finish().is_empty()); + assert!(assembler.recently_completed.len() <= MAX_RECENT_COMPLETIONS); + assert_eq!( + assembler.completed_bytes, + assembler + .recently_completed + .iter() + .map(|(key, entry)| entry.memory_bytes() + key.vlan_context.memory_bytes()) + .sum::() + ); + } + + #[test] + fn capacity_eviction_releases_new_prefix_once_and_ends_its_replay_fence() { + let mut assembler = recovering(); + assert_eq!( + assembler + .process_batch(vec![packet(0, true, 1, 100), changed_first(2, 1_101)]) + .0 + .len(), + 1 + ); + let input = (0..MAX_PENDING_DATAGRAMS) + .map(|id| with_id(packet(0, true, id as u64 + 1_000, 300), id as u16 + 20_000)) + .collect(); + let (evicted, _) = assembler.process_batch(input); + assert_eq!(evicted.len(), 1); + assert_eq!(evicted[0].packet_ordinal, 2); + assert_eq!(assembler.pending.len(), MAX_PENDING_DATAGRAMS); + assert!(assembler.pending_bytes <= MAX_PENDING_BYTES); + // All owners of this key were evicted. Older replay is now eligible again. + let (output, _) = assembler.process_batch(vec![ + packet(0, true, 90_000, 100), + packet(24, false, 90_001, 101), + ]); + assert_eq!( + output + .iter() + .filter(|packet| packet.packet_ordinal == 90_001) + .count(), + 1 + ); + assert!( + assembler + .finish() + .iter() + .all(|packet| packet.packet_ordinal != 2) + ); + assert_eq!(assembler.pending_bytes, 0); + } + + #[test] + fn recovery_is_independent_of_other_addresses_vlan_and_global_future_time() { + for dimension in 0..3 { + let mut assembler = recovering(); + assembler.process_batch(vec![packet(0, true, 1, 100)]); + // Admitted timestamp regressions preserve the candidate's maximum. + assembler.process_batch(vec![packet(0, true, 2, -100)]); + let mut unrelated = packet(0, true, 3, i64::MAX - 1); + if dimension == 2 { + unrelated = tagged_packet(unrelated, &[(0x8100, 100)]); + } else { + let mut bytes = unrelated.data.as_slice().to_vec(); + bytes[if dimension == 0 { 29 } else { 33 }] ^= 1; + unrelated.data = PacketPayload::owned(bytes.into_boxed_slice()); + } + assembler.process_batch(vec![unrelated]); + let (output, pending) = + assembler.process_batch(vec![changed_first(4, 1_101), packet(24, false, 5, 1_102)]); + assert_eq!(output.len(), 2); + assert_eq!(output[0].packet_ordinal, 1); + assert_eq!(output[1].packet_ordinal, 5); + assert_eq!(pending, Some(i64::MAX - 1)); + let leftover = assembler.finish(); + assert_eq!(leftover.len(), 1); + assert_eq!(leftover[0].packet_ordinal, 3); + } + } + + #[test] + fn recovery_and_eof_are_independent_of_batch_boundaries() { + for chunk_size in [1, 2, 5] { + let mut assembler = recovering(); + let input = [ + packet(0, true, 1, 100), + changed_first(2, 1_101), + packet(24, false, 3, 1_100), + non_dns_first(4, 100), + packet(24, false, 5, 1_102), + ]; + let mut output = Vec::new(); + for chunk in input.chunks(chunk_size) { + output.extend(assembler.process_batch(chunk.to_vec()).0); + } + assert_eq!( + output + .iter() + .map(|packet| packet.packet_ordinal) + .collect::>(), + vec![1, 5] + ); + assert!(assembler.finish().is_empty()); + assert!(assembler.finish().is_empty()); + } + let mut assembler = recovering(); + assert_eq!( + assembler + .process_batch(vec![packet(0, true, 1, 100), changed_first(2, 1_101)]) + .0 + .len(), + 1 + ); + let fallback = assembler.finish(); + assert_eq!(fallback.len(), 1); + assert_eq!(fallback[0].packet_ordinal, 2); + assert!(assembler.finish().is_empty()); + } + + #[test] + fn reuse_preserves_overlap_and_incomplete_coverage_checks() { + for tail in [packet(16, false, 3, 1_102), packet(32, false, 3, 1_102)] { + let mut assembler = recovering(); + assembler.process_batch(vec![packet(0, true, 1, 100), changed_first(2, 1_101)]); + let (output, _) = assembler.process_batch(vec![tail]); + assert!(output.is_empty()); + // A missing byte range remains incomplete; an overlap becomes rejected. + let (output, _) = assembler.process_batch(vec![packet(16, false, 4, 1_103)]); + assert!(output.is_empty()); + assert!(assembler.pending.values().next().unwrap().rejected); + assert!(assembler.finish().is_empty()); + } + } + + #[test] + fn reuse_keeps_udp_mf_and_final_boundary_validation() { + for (offset, payload_len, more) in [(32_u16, 16, false), (32, 8, true), (24, 8, false)] { + let mut assembler = recovering(); + assembler.process_batch(vec![packet(0, true, 1, 100), changed_first(2, 1_101)]); + let mut tail = packet(24, false, 3, 1_102); + let mut bytes = tail.data.as_slice().to_vec(); + bytes.truncate(ETHERNET_HEADER_LEN + IPV4_MIN_HEADER_LEN + payload_len); + bytes[16..18] + .copy_from_slice(&((IPV4_MIN_HEADER_LEN + payload_len) as u16).to_be_bytes()); + let flags = (offset / 8) | if more { MORE_FRAGMENTS } else { 0 }; + bytes[20..22].copy_from_slice(&flags.to_be_bytes()); + tail.data = PacketPayload::owned(bytes.into_boxed_slice()); + assert!(assembler.process_batch(vec![tail]).0.is_empty()); + assert!(assembler.pending.values().next().unwrap().rejected); + assert!(assembler.finish().is_empty()); + } + } + + #[test] + fn recovery_timestamp_arithmetic_handles_extremes_without_saturation() { + let cases = [ + (i64::MIN, 0, 1_000, false), + (0, i64::MAX, 1_000, false), + (i64::MAX, i64::MAX - 1, 1_000, false), + (i64::MAX - 10, i64::MAX - 1, 1_000, false), + (i64::MIN + 1, i64::MIN + 1_002, 1_000, true), + (i64::MAX - 2_000, i64::MAX - 999, 1_000, true), + (i64::MIN + 1, i64::MAX - 1, 1_000, true), + (0, i64::MAX - 1, (i64::MAX / 1_000) * 1_000, true), + ]; + for (old, new, gap, allowed) in cases { + let mut assembler = + Ipv4FragmentReassembler::new(1_200_000, false).with_reuse_gap_micros(Some(gap)); + assembler.process_batch(vec![packet(0, true, 1, old)]); + let (output, _) = + assembler.process_batch(vec![changed_first(2, new), packet(24, false, 3, new)]); + assert_eq!( + output.len(), + if allowed { 2 } else { 0 }, + "old={old}, new={new}, gap={gap}" + ); + assert!(assembler.finish().is_empty()); + } + } + + #[test] + fn reuse_metadata_fits_existing_accounting_estimates() { + use std::mem::size_of; + assert!(size_of::() + size_of::() <= ENTRY_OVERHEAD_ESTIMATE); + assert!( + size_of::() + size_of::() + <= RECENT_ENTRY_OVERHEAD_ESTIMATE + ); + assert!( + size_of::() + size_of::() <= RECENT_ENTRY_OVERHEAD_ESTIMATE + ); + let mut assembler = recovering(); + assembler.process_batch(vec![ + tagged_packet(packet(0, true, 1, 100), &[(0x8100, 10)]), + tagged_packet(changed_first(2, 1_101), &[(0x8100, 10)]), + ]); + assert_eq!( + assembler.pending_bytes, + assembler + .pending + .iter() + .map(|(key, entry)| entry.memory_bytes() + key.vlan_context.memory_bytes()) + .sum::() + ); + assembler.process_batch(vec![tagged_packet( + non_dns_first(3, 1_102), + &[(0x8100, 10)], + )]); + assert_eq!(assembler.pending_bytes, 0); + assert_eq!(assembler.non_dns_bytes, RECENT_ENTRY_OVERHEAD_ESTIMATE + 4); + assert_eq!( + assembler + .known_non_dns + .values() + .next() + .unwrap() + .reuse_cutoff_timestamp, + Some(1_100) + ); + } + #[test] fn non_dns_first_discards_tails_before_and_after_it() { let mut reassembler = Ipv4FragmentReassembler::new(1_200_000, true); @@ -1140,6 +1812,7 @@ mod tests { }, Arc::<[u8]>::from(&b"payload"[..]), 5_000_103, + None, ); } assert!(reassembler.recently_completed.len() <= MAX_RECENT_COMPLETIONS); diff --git a/src/parquet_writer.rs b/src/parquet_writer.rs index 2fd5e37..c78fbb9 100644 --- a/src/parquet_writer.rs +++ b/src/parquet_writer.rs @@ -390,6 +390,7 @@ mod tests { max_dns_compression_jumps: crate::config::DEFAULT_MAX_DNS_COMPRESSION_JUMPS, allow_fragments: false, full_fragments: false, + ipv4_fragment_reuse_gap_ms: None, }; let file = File::create(&filename).expect("creates parquet file"); let writer = create_parquet_writer(file, &config).expect("creates parquet writer"); @@ -431,6 +432,7 @@ mod tests { max_dns_compression_jumps: crate::config::DEFAULT_MAX_DNS_COMPRESSION_JUMPS, allow_fragments: false, full_fragments: false, + ipv4_fragment_reuse_gap_ms: None, }; let file = File::create(&filename).expect("creates parquet file"); let writer = create_parquet_writer(file, &config).expect("creates parquet writer"); @@ -472,6 +474,7 @@ mod tests { max_dns_compression_jumps: crate::config::DEFAULT_MAX_DNS_COMPRESSION_JUMPS, allow_fragments: false, full_fragments: false, + ipv4_fragment_reuse_gap_ms: None, }; let writer = create_parquet_writer( SharedSink { @@ -515,6 +518,7 @@ mod tests { max_dns_compression_jumps: crate::config::DEFAULT_MAX_DNS_COMPRESSION_JUMPS, allow_fragments: false, full_fragments: false, + ipv4_fragment_reuse_gap_ms: None, }; let file = File::create(&filename).expect("creates parquet file"); let writer = create_parquet_writer(file, &config).expect("creates parquet writer"); @@ -557,6 +561,7 @@ mod tests { max_dns_compression_jumps: crate::config::DEFAULT_MAX_DNS_COMPRESSION_JUMPS, allow_fragments: false, full_fragments: false, + ipv4_fragment_reuse_gap_ms: None, }; let file = File::create(&filename).expect("creates parquet file"); let writer = create_parquet_writer(file, &config).expect("creates parquet writer"); @@ -624,6 +629,7 @@ mod tests { max_dns_compression_jumps: crate::config::DEFAULT_MAX_DNS_COMPRESSION_JUMPS, allow_fragments: false, full_fragments: false, + ipv4_fragment_reuse_gap_ms: None, }; let file = File::create(&filename).expect("creates parquet file"); let mut writer = create_parquet_writer(file, &config).expect("creates parquet writer"); diff --git a/tests/end_to_end.rs b/tests/end_to_end.rs index 7a844bd..cf417dd 100644 --- a/tests/end_to_end.rs +++ b/tests/end_to_end.rs @@ -36,9 +36,13 @@ fn dpp_binary() -> &'static str { } fn append_example_a_query(dns_payload: &mut Vec) { - dns_payload.extend_from_slice(&[ - 7, b'e', b'x', b'a', b'm', b'p', b'l', b'e', 3, b'c', b'o', b'm', 0, - ]); + append_com_a_query(dns_payload, b"example"); +} + +fn append_com_a_query(dns_payload: &mut Vec, label: &[u8]) { + dns_payload.push(u8::try_from(label.len()).expect("test label length fits into u8")); + dns_payload.extend_from_slice(label); + dns_payload.extend_from_slice(&[3, b'c', b'o', b'm', 0]); dns_payload.extend_from_slice(&1_u16.to_be_bytes()); dns_payload.extend_from_slice(&1_u16.to_be_bytes()); } @@ -94,6 +98,92 @@ fn fragment_ipv4_udp_packet(packet: &[u8], first_payload_len: usize) -> (Vec (first_fragment, later_fragment) } +fn fragment_reuse_transaction(id: u16, label: &[u8]) -> (Vec, Vec, Vec) { + assert_eq!( + label.len(), + 7, + "fixture question must fit in the first fragment" + ); + let mut query_payload = encode_dns_header(id, 0x0100, 1); + append_com_a_query(&mut query_payload, label); + let query_packet = + make_udp_dns_packet_with_payload([10, 0, 0, 1], [8, 8, 8, 8], 53_000, 53, &query_payload); + + let mut response_payload = encode_dns_header(id, 0x8180, 1); + response_payload[10..12].copy_from_slice(&1_u16.to_be_bytes()); + append_com_a_query(&mut response_payload, label); + append_opt_record(&mut response_payload, 1, 0); + let response_packet = make_udp_dns_packet_with_payload( + [8, 8, 8, 8], + [10, 0, 0, 1], + 53, + 53_000, + &response_payload, + ); + let (first, tail) = fragment_ipv4_udp_packet(&response_packet, 40); + (query_packet, first, tail) +} + +fn run_fragment_reuse_capture( + name: &str, + capture: &[u8], + threads: usize, + extra_args: &[&str], + env_gap: Option<&str>, +) -> (String, serde_json::Value) { + let input_path = temp_test_path(name, "pcap"); + let output_path = temp_test_path(name, "csv"); + fs::write(&input_path, capture).expect("fragment reuse capture written"); + let mut command = Command::new(dpp_binary()); + command + .args(["-s", "-f", "csv", "--full-fragments", "--threads"]) + .arg(threads.to_string()) + .args(["--report-format", "json"]) + .args(extra_args) + .env_remove("DPP_IPV4_FRAGMENT_REUSE_GAP_MS") + .env_remove("DPP_MONOTONIC_CAPTURE") + .env_remove("DPP_MATCH_TIMEOUT_MS"); + if let Some(gap) = env_gap { + command.env("DPP_IPV4_FRAGMENT_REUSE_GAP_MS", gap); + } + let output = command + .arg(&input_path) + .arg(&output_path) + .output() + .expect("DPP executes the fragment reuse capture"); + assert!( + output.status.success(), + "DPP failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + let csv = fs::read_to_string(&output_path).expect("fragment reuse CSV readable"); + assert_eq!( + csv.lines().next(), + Some( + "request_timestamp,response_timestamp,source_ip,source_port,id,name,query_type,response_code" + ) + ); + let report: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("JSON report parses"); + let available_cpus = report["execution"]["available_cpus"] + .as_u64() + .expect("execution budget is reported"); + assert_eq!( + report["execution"]["model"], + if available_cpus >= 5 { + "staged" + } else { + "phase_parallel" + } + ); + if threads == 1 { + assert_eq!(available_cpus, 1); + } + fs::remove_file(input_path).expect("remove fragment reuse capture"); + fs::remove_file(output_path).expect("remove fragment reuse CSV"); + (csv, report) +} + #[cfg(unix)] fn wait_for_child_exit(child: &mut Child) { let deadline = Instant::now() + Duration::from_secs(5); @@ -568,6 +658,204 @@ fn full_ipv4_fragment_mode_falls_back_to_first_fragment_at_eof() { fs::remove_file(output_path).expect("remove output csv"); } +#[test] +fn ipv4_fragment_reuse_gap_recovers_late_id_from_cli_and_environment() { + let (query_a, first_a, _) = fragment_reuse_transaction(0x2345, b"example"); + let (query_b, first_b, tail_b) = fragment_reuse_transaction(0x3456, b"another"); + let capture = classic_pcap_bytes(&[ + (1, 0, &query_a), + (1, 100_000, &first_a), + (122, 0, &query_b), + (122, 100_000, &first_b), + (122, 100_001, &tail_b), + ]); + + for threads in [1, 5] { + let (csv, report) = + run_fragment_reuse_capture("ipv4-reuse-default", &capture, threads, &[], None); + assert_eq!( + csv, + concat!( + "request_timestamp,response_timestamp,source_ip,source_port,id,name,query_type,response_code\n", + "1000000,,10.0.0.1,53000,9029,example.com,A,\n", + "122000000,,10.0.0.1,53000,13398,another.com,A,\n" + ) + ); + assert_eq!(report["metrics"]["total_dns_responses_processed"], 0); + assert_eq!(report["metrics"]["timed_out_queries"], 2); + + for env_gap in [None, Some("120000")] { + let extra_args: &[&str] = if env_gap.is_some() { + &[] + } else { + &["--ipv4-fragment-reuse-gap-ms", "120000"] + }; + let (csv, report) = run_fragment_reuse_capture( + "ipv4-reuse-enabled", + &capture, + threads, + extra_args, + env_gap, + ); + assert_eq!( + csv, + concat!( + "request_timestamp,response_timestamp,source_ip,source_port,id,name,query_type,response_code\n", + "1000000,1100000,10.0.0.1,53000,9029,example.com,A,\n", + "122000000,122100001,10.0.0.1,53000,13398,another.com,A,EDNS_BADVERS\n" + ) + ); + assert_eq!(report["metrics"]["fragmented_response_prefix_count"], 1); + assert_eq!(report["metrics"]["total_dns_responses_processed"], 2); + assert_eq!(report["metrics"]["total_matched_query_response_pairs"], 2); + assert_eq!(report["metrics"]["timed_out_queries"], 0); + } + } +} + +#[test] +fn ipv4_fragment_reuse_gap_is_strict_and_cli_overrides_environment() { + let (query_a, first_a, _) = fragment_reuse_transaction(0x2345, b"example"); + let (query_b, first_b, tail_b) = fragment_reuse_transaction(0x3456, b"another"); + + for delta in [0, 1] { + let capture = classic_pcap_bytes(&[ + (1, 0, &query_a), + (1, 100_000, &first_a), + (121, 0, &query_b), + (121, 100_000 + delta, &first_b), + (121, 100_001 + delta, &tail_b), + ]); + let (csv, report) = run_fragment_reuse_capture( + "ipv4-reuse-boundary", + &capture, + 1, + &["--ipv4-fragment-reuse-gap-ms", "120000"], + Some("1"), + ); + let expected_b = if delta == 0 { + "121000000,,10.0.0.1,53000,13398,another.com,A,".to_owned() + } else { + "121000000,121100002,10.0.0.1,53000,13398,another.com,A,EDNS_BADVERS".to_owned() + }; + assert!(csv.lines().any(|line| line == expected_b)); + assert_eq!( + report["metrics"]["total_matched_query_response_pairs"], + if delta == 0 { 0 } else { 2 } + ); + } +} + +#[test] +fn ipv4_fragment_reuse_gap_recovers_tombstone_without_rejected_prefix() { + let (query_a, first_a, _) = fragment_reuse_transaction(0x2345, b"example"); + let (query_b, first_b, tail_b) = fragment_reuse_transaction(0x3456, b"another"); + let capture = classic_pcap_bytes(&[ + (1, 0, &query_a), + (1, 100_000, &first_a), + // A close collision rejects the old candidate before the late reuse. + (1, 200_000, &first_b), + (122, 100_000, &query_b), + (122, 200_000, &first_b), + (122, 200_001, &tail_b), + ]); + for threads in [1, 5] { + let (csv, report) = run_fragment_reuse_capture( + "ipv4-reuse-tombstone", + &capture, + threads, + &["--ipv4-fragment-reuse-gap-ms", "120000"], + None, + ); + assert_eq!(csv.lines().count(), 3); + assert!(csv.contains("1000000,,10.0.0.1,53000,9029,example.com,A,\n")); + assert!( + csv.contains("122100000,122200001,10.0.0.1,53000,13398,another.com,A,EDNS_BADVERS\n") + ); + assert_eq!(report["metrics"]["fragmented_response_prefix_count"], 0); + assert_eq!(report["metrics"]["total_dns_responses_processed"], 1); + assert_eq!(report["metrics"]["total_matched_query_response_pairs"], 1); + assert_eq!(report["metrics"]["timed_out_queries"], 1); + } +} + +#[test] +fn ipv4_fragment_reuse_gap_filters_old_replay_across_history_and_batches() { + let (query_a, first_a, tail_a) = fragment_reuse_transaction(0x2345, b"example"); + let (query_b, first_b, tail_b) = fragment_reuse_transaction(0x3456, b"another"); + let mut non_dns_first = first_a.clone(); + non_dns_first[34..36].copy_from_slice(&123_u16.to_be_bytes()); + let mut padding = first_a.clone(); + padding[14 + 9] = 1; + + for padded in [false, true] { + let mut packets = vec![(1, 0, query_a.as_slice()), (1, 100_000, first_a.as_slice())]; + if padded { + // Unsupported ICMP fragments fill an ingest batch without a ready backlog + // or another pending datagram that would change fallback timing. + packets.extend(std::iter::repeat_n( + (1, 200_000, padding.as_slice()), + 65_536 - packets.len(), + )); + } + packets.extend([ + (122, 0, query_b.as_slice()), + (122, 100_000, first_b.as_slice()), + ]); + if padded { + packets.extend(std::iter::repeat_n( + (122, 200_000, padding.as_slice()), + 65_536 - packets.len() % 65_536, + )); + } + packets.extend([ + // These old events are exactly on the retired generation's cutoff. + (121, 100_000, first_a.as_slice()), + (121, 100_000, tail_a.as_slice()), + (121, 100_000, non_dns_first.as_slice()), + (122, 100_001, tail_b.as_slice()), + // The cutoff must survive completion and a later non-DNS mark. + (1, 100_000, first_a.as_slice()), + (1, 100_001, tail_a.as_slice()), + (1, 100_002, non_dns_first.as_slice()), + (122, 150_000, non_dns_first.as_slice()), + (1, 100_000, first_a.as_slice()), + (1, 100_001, tail_a.as_slice()), + (1, 100_002, non_dns_first.as_slice()), + // A fully identical completed response remains a duplicate. + (122, 200_000, query_b.as_slice()), + (122, 300_000, first_b.as_slice()), + (122, 300_001, tail_b.as_slice()), + (1, 100_000, first_a.as_slice()), + (1, 100_001, tail_a.as_slice()), + (1, 100_002, non_dns_first.as_slice()), + ]); + let capture = classic_pcap_bytes(&packets); + for threads in [1, 5] { + let (csv, report) = run_fragment_reuse_capture( + "ipv4-reuse-replay", + &capture, + threads, + &["--ipv4-fragment-reuse-gap-ms", "120000"], + None, + ); + assert_eq!( + csv, + concat!( + "request_timestamp,response_timestamp,source_ip,source_port,id,name,query_type,response_code\n", + "1000000,1100000,10.0.0.1,53000,9029,example.com,A,\n", + "122000000,122100001,10.0.0.1,53000,13398,another.com,A,EDNS_BADVERS\n", + "122200000,,10.0.0.1,53000,13398,another.com,A,\n" + ) + ); + assert_eq!(report["metrics"]["fragmented_response_prefix_count"], 1); + assert_eq!(report["metrics"]["total_dns_responses_processed"], 2); + assert_eq!(report["metrics"]["total_matched_query_response_pairs"], 2); + assert_eq!(report["metrics"]["timed_out_queries"], 1); + } + } +} + #[test] fn maximum_wire_qname_round_trips_with_full_escaped_presentation() { let input_path = temp_test_path("maximum-wire-qname", "pcap");