diff --git a/Cargo.lock b/Cargo.lock index 5d42990..d19e68b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -665,11 +665,12 @@ dependencies = [ [[package]] name = "dpp" -version = "0.3.1" +version = "0.3.2" dependencies = [ "aes", "anyhow", "arrayvec", + "byteorder_slice", "bytes", "clap", "core_affinity", diff --git a/Cargo.toml b/Cargo.toml index afb4984..6b7e8af 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,7 @@ [package] name = "dpp" -version = "0.3.1" +version = "0.3.2" edition = "2024" rust-version = "1.97.1" build = "build.rs" # Custom build script to include version information from git @@ -47,6 +47,7 @@ opt-level = 3 [dependencies] pcap = "2.5" # Library for reading pcap files pcap-file = "3.0.0-rc1" # Pure-Rust parser for classic pcap and pcapng files +byteorder_slice = "3.0" # Stateless pcapng block decoding without upstream timestamp conversion rayon = "1.12" # Parallelism library for multithreaded processing hickory-proto = { version = "0.26", default-features = false, features = ["std"] } # DNS protocol parsing library csv = "1.4" # CSV serialization and deserialization diff --git a/README.md b/README.md index bcc311b..7e4c215 100644 --- a/README.md +++ b/README.md @@ -245,9 +245,15 @@ Unsupported non-PCAP/non-PCAPNG stream magic on `stdin` is rejected explicitly i hidden temp-file or second ingest path. Repeated pending queries that share the same match identity (`id`, `name`, client IP, client port, -resolver IP, and `query_type`) inside the configured match window (`1200ms` by default) are +resolver IP, `query_type`, QCLASS, and OPCODE) inside the configured match window (`1200ms` by default) are deduplicated to the earliest canonical query. Deduplicated retries increment a separate counter and -do not emit extra timeout or matched records. +do not emit extra timeout or matched records. In default mode, pending retry timestamps are retained +so an earlier timestamp arriving in a later batch can split pending groups without widening the +match window. Finalized transactions are never reopened. With `--monotonic-capture`, no retry +history is needed or allocated. + +DNS QR determines query/response direction, including valid exchanges with UDP port 53 on both +endpoints. QCLASS and OPCODE distinguish transactions internally and do not add output columns. For QNAME matching, DPP preserves the observed presentation-form name bytes and does not lowercase them before building matcher identity keys. This is a deliberate Community Edition trade-off, not @@ -280,6 +286,10 @@ the absent `response_timestamp`; `response_code` is absent because no DNS respon extended RCODE `16` is exported as `EDNS_BADVERS`; `TSIG Failure` is used for code `16` only when a TSIG RR error field supplies that status. +Unknown query types retain their wire number as `TYPE` (for example, `TYPE65400`), +and unknown response codes use their decimal number (for example, `64`). CSV and Parquet +preserve these values instead of collapsing distinct codes into an `Unknown` label. + With `--report-format json` or `DPP_REPORT_FORMAT=json`, DPP suppresses routine `info`/`warn` reporting and emits one final JSON summary object to `stdout`. This keeps CSV or Parquet output in its own file while making the end-of-run report easy to capture and parse. @@ -493,7 +503,9 @@ Additional notes: ## Limitations - **UDP/53 only:** DPP currently processes DNS traffic over UDP port 53 only. -- DPP does not reassemble IPv4 fragments, so fragmented IPv4 datagrams are skipped. +- DPP does not reassemble IPv4 or IPv6 fragments. IPv6 Hop-by-Hop, Routing, Destination Options, + Authentication and atomic Fragment headers are traversed before UDP; fragments requiring + reassembly, ESP and IPv6 jumbograms are unsupported. Flow identity uses observed IP endpoints. - If capture parsing fails after processing begins, DPP flushes valid partial (not atomic) output from complete accepted batches and exits with an error; pending queries are not emitted as timeouts. - **PCAPNG support level:** DPP supports PCAPNG on stream input and via `libpcap` on regular-file fallback paths, but the performance-critical pure-Rust fast path remains focused on classic PCAP. diff --git a/benches/README.md b/benches/README.md index 27904d3..418f3d6 100644 --- a/benches/README.md +++ b/benches/README.md @@ -7,6 +7,8 @@ This directory is the canonical location for repeatable performance measurements - `benchmark.sh`: end-to-end benchmark harness for the `dpp` binary. - `allocator-benchmarking.md`: canonical build-and-measure protocol for global allocator comparisons. +- `protocol-correctness-audit.md`: final validation, restored throughput and measured memory costs + of the 2026-09-05 protocol and matcher fixes, with earlier experiments identified separately. ## Safety Contract diff --git a/benches/protocol-correctness-audit.md b/benches/protocol-correctness-audit.md new file mode 100644 index 0000000..333618e --- /dev/null +++ b/benches/protocol-correctness-audit.md @@ -0,0 +1,130 @@ +# Protocol correctness: final validation and performance + +Updated 2026-09-05 for implementation commit `8881539` on +`feature/protocol-correctness-audit`. + +The final implementation retains the protocol and matcher corrections and restores throughput +on the selected 18-million-packet capture. Default mode slightly exceeds the pre-audit version +in the final comparison; monotonic mode is at approximately the same level. The workspace suite +passes **254 tests**, and full CSV output is byte-identical to the correctness-fixed version in +both modes and with both DNS decoders. Peak RSS is higher, approximately 288–301 MiB in these runs. + +## Current performance + +Each comparison below used ten position-balanced pairs on one macOS ARM64 host. Speed ratios +are the median of paired reference-time / final-time ratios; values above 1 indicate improvement. +The two reference comparisons are separate measurement series, so their absolute timings and RSS +must not be combined as if they were one run. + +Against the pre-audit version, `3bb45397baff03a6c7043d2e6cff1508f0161b39`: + +| Configuration | Pre-audit median (s) | Final median (s) | Paired speed ratio | Median peak RSS, pre-audit / final (MiB) | +|---|---:|---:|---:|---:| +| Default | 0.9807 | 0.9677 | 1.0134 | 217.7 / 298.4 | +| Monotonic | 0.9580 | 0.9627 | 0.9962 | 221.3 / 287.9 | + +Default throughput is about 1.3% higher in this series. The roughly 0.4% monotonic difference is +within the observed run-to-run variation: paired speed ratios ranged from 0.9687 to 1.0394. +These results establish local recovery on this capture, not a throughput or portability guarantee. + +Against the correctness-fixed version before optimization, `2ef34a4`: + +| Configuration | Fixed median (s) | Final median (s) | Paired speed ratio | Median peak RSS, fixed / final (MiB) | +|---|---:|---:|---:|---:| +| Default | 1.0273 | 0.9758 | 1.0484 | 209.8 / 301.4 | +| Monotonic | 0.9980 | 0.9723 | 1.0290 | 219.0 / 288.4 | + +The optimizations improve throughput by about 4.8% and 2.9% respectively in this confirmation +series. Higher occupied pipeline backlog and allocator retention are hypotheses for the RSS +increase; their contributions were not isolated. Queue bounds and payload ownership are unchanged. + +## Final implementation + +- Routing metadata is **42 bytes on the measured host**. A checked `u16` offset is relative to + the minimum 42-byte Ethernet/IPv4/UDP prefix. This still represents DNS starts above 65535: + a maximum IPv6 frame ends at byte 65589 and must leave at least 12 DNS bytes. IP/UDP boundaries, + QR direction, QCLASS/OPCODE identity and compression validation remain enforced. +- Flow routing collects the existing tuple's `Hash` writes into a 64-byte stack buffer and + invokes SeaHash once over the bytes. Integer encoding, seeds and the complete digest are + unchanged, preserving shard assignment and output order. If the input exceeds the buffer + capacity, its contents are replayed into the original streaming hasher. No additional persistent + flow state is kept. +- Each pending canonical query owns one optional retry-history pointer. Default mode allocates + history when a pending retry arrives; later retries can grow its vector. Replacing a canonical + query with an earlier attempt regroups that identity's unresolved attempts in `O(n log n)`. + Monotonic mode allocates no retry history. Memory on duplicate-heavy captures therefore depends + on unresolved attempts; this capture's low duplicate rate does not bound that workload. +- PCAPNG stdin retains parser-owned framing, one section-local interface table and exact signed + timestamp conversion, including binary/decimal resolution and offsets. Conversion includes + `u128` division per packet; the reusable block buffer grows with received bytes. Regular-file + PCAPNG still uses libpcap, and classic PCAP retains its pure-Rust streaming path. +- Known exported DNS type/status labels remain borrowed strings; numeric fallback allocates for + unknown values. CSV and Parquet share the same canonical formatting contract. + +`build.rs` was reviewed and left unchanged. It requests executable stack size and supplies build +metadata. Optimization levels, LTO and codegen units belong to `Cargo.toml`; no compiler or linker +settings were changed during performance recovery. + +## Validation + +`cargo test --workspace --offline --locked` passes 254 tests. Coverage includes 30 capture-reader +tests, 13 DNS parser regressions, a 720-permutation matcher check, both pipeline execution models, +and CSV/Parquet preservation of unknown codes. Four differential hashing tests compare complete +64-bit digests for address families, ports, 4096 deterministic flows, integer encodings, buffer +boundaries, streaming fallback and repeated `finish` calls. + +`cargo fmt --all -- --check` and the CI Clippy command pass: + +```sh +cargo clippy --all-targets --no-deps --offline --locked -- \ + -D warnings \ + -A clippy::too_many_arguments \ + -A clippy::type_complexity \ + -A clippy::large_enum_variant +``` + +Every final comparison preserved packet/query/response/duplicate/match/timeout counters and +average matched RTT. The selected capture contains 18,006,441 packets, 9,013,611 queries, +13,610 duplicates, 8,992,830 responses and matched pairs, and 7,171 timeouts. + +Full CSV output was compared against `2ef34a4` with each combination of monotonic mode and +`--dns-wire-fast-path`: 854,848,168 bytes per output, with identical SHA-256 within each pair. +Both decoders produce the same output for a given mode. Default and monotonic modes have different +row ordering by their existing timeout-emission contracts. + +| Mode | SHA-256, both reference and final | +|---|---| +| Default | `02db37f51a60a84000b457c47339fb87bd725071c4e08899f3e8b90b430f9fc9` | +| Monotonic | `08f435bc3aa241615937e3c08b4c089570d5e3dbe9fb245a46c8e9bd3d680bc0` | + +## Measurement method + +The explicitly selected input was `synthetic/server1-like.pcap`. All compared binaries were built +with Rust 1.97.1 using `cargo build --profile perf`, default jemalloc and no additional RUSTFLAGS. +This profile inherits release's `opt-level=3`, LTO and one codegen unit, and disables overflow +checks for trusted benchmark input; the production release profile keeps overflow checks enabled. + +The host exposed 16 CPUs, giving 14 matcher workers in staged mode. Affinity was disabled, CSV +output went to `/dev/null`, and bounded output settings were identical. Runs alternated reference / +final and final / reference order, with an excluded warmup pair per mode and no subsequent outlier +removal. Timings include process startup and shutdown. The capture was read from the filesystem +cache, so these results do not measure cold-storage performance. Peak RSS was collected per child +with `wait4`. See [README.md](README.md) for the benchmark scaffold and explicit-input contract. + +## Earlier measurements and discarded experiments + +Before optimization, the correctness fixes showed about 5.6% lower default throughput and 4.8% +lower monotonic throughput in the initial comparison. Those figures describe the intermediate +`2ef34a4` implementation, not the current version. Compact relative offsets and buffered flow +hashing were subsequently retained on measurement evidence. + +Short four-pair screens did not justify retaining query-helper inlining/cold regrouping, CSV +formatter inlining, a fused borrowed matcher lookup, or a new packed hash input. The final code +contains none of those experimental changes. CPU sampling identified packet routing and its +streaming hash calls as the limiting stage on this capture. + +An earlier PCAPNG-reader check compared the pre-audit reader with the correctness-fixed reader +using the first 1,000,000 packets re-encoded as little-endian Enhanced Packet Blocks. Three pairs +measured 0.0780 / 0.0787 seconds, with identical processing counters. The roughly 80 ms runs were +too short to establish a timing change. This is reader-validation history, not a final PCAPNG +throughput benchmark. diff --git a/docs/architecture.md b/docs/architecture.md index 6d02374..65942a4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -83,6 +83,12 @@ and uses the remaining CPU budget for shard workers. Routed DNS packets also car metadata into shard workers so the worker path can reuse the first L3/L4 parse instead of repeating it for full DNS question decoding. +Flow routing buffers the existing tuple's `Hash` writes on the stack and runs SeaHash once over +the resulting bytes. The integer encoding and digest remain identical to the streaming hasher, +so shard assignment and output order are preserved. If a future tuple encoding exceeds the +buffer, routing replays its bytes into the streaming hasher; this adds no persistent flow state +or second encoding of canonical addresses and ports. + ## Core Components - `src/packet_parser.rs` @@ -90,8 +96,13 @@ repeating it for full DNS question decoding. files use a pure-Rust streaming reader; regular-file PCAPNG and other non-classic formats currently fall back to libpcap. EOF-terminated stdin capture streams are also supported through parser-owned stream-native readers: classic PCAP stdin uses the same pure-Rust fast path family - as classic file input, and PCAPNG stdin uses a pure-Rust pcapng reader so stdin probing does not - need a temp-file or second ingest owner. Unsupported stdin stream magic is rejected explicitly. + as classic file input, and PCAPNG stdin owns block framing and one section-local interface table. + PCAPNG uses the dependency's stateless structural validation, while the parser converts raw + high/low timestamp words using the interface's full binary/decimal resolution and signed offset. + Conversion scales the complete counter before rounding to microseconds and saturates only the + final signed timestamp. It never invokes the dependency's stateful timestamp conversion. + Buffers grow with received block bytes rather than untrusted declared lengths. Stdin probing + needs no temp-file or second ingest owner. Unsupported stdin stream magic is rejected explicitly. The pure-Rust classic-PCAP reader still relies on the upstream `pcap-file` `3.0.0-rc1` release candidate until a stable line with the required functionality is available. The parser can also enforce globally monotonic capture timestamps for the optional batched @@ -143,10 +154,19 @@ repeating it for full DNS question decoding. shard-local DNS decode, but that reuse must stay within the same ownership boundary so packet parsing does not gain a second source of truth for IP/port extraction. That metadata owns the exact DNS byte range validated against IPv4 Total Length or IPv6 Payload Length and then UDP - Length; capture padding and trailing IP payload cannot extend the DNS slice. Fragmented IPv4 - datagrams are skipped because this boundary has no IP reassembly stage. The optional runtime flag + Length; capture padding and trailing IP payload cannot extend the DNS slice. A compact relative + offset preserves DNS starts above 65535 without widening every packet's routing metadata: the + offset is measured from the minimum Ethernet/IPv4/UDP header length, and construction checks + that it fits. DNS QR determines + direction and the canonical client/resolver flow, including exchanges with UDP port 53 on both + ends. IPv6 Hop-by-Hop, Routing, Destination Options, AH and atomic Fragment headers are traversed + within the declared payload boundary. Non-atomic IPv6 and fragmented IPv4 datagrams are skipped + because this boundary has no IP reassembly stage. The optional runtime flag `--dns-wire-fast-path` may enable a custom question-only wire fast path, but `hickory` remains - the semantic fallback for rare DNS messages that the fast path does not accept. Both paths accept + the semantic fallback for rare DNS messages that the fast path does not accept. Compression + pointers must target prior, nonoverlapping names; enabling the fast path must not weaken that + validation. TSIG status extraction consumes and bounds-checks the declared Other Data, including + the six-byte server time in BADTIME responses. Both paths accept decompressed wire QNAMEs up to the RFC 1035 limit of 255 octets, including label-length octets and the terminating root octet. A valid name can expand to 1003 bytes in escaped presentation form and must remain distinct through matching and export. If any QNAME exceeds the wire limit, the @@ -283,9 +303,17 @@ The DNS matcher must preserve these invariants: - Routing and in-flight matching use the original observed client IP, client port, and resolver IP. The resolver remains internal; deterministic client-IP pseudonymization is applied exactly once when the matcher constructs a finalized `DnsRecord`. +- DNS ID, observed QNAME, QTYPE, QCLASS and OPCODE also distinguish in-flight identities. QCLASS, + OPCODE and resolver identity remain internal and do not change the exported record schema. - Repeated pending queries with the same match identity inside the configured timeout window (`1200ms` by default) are deduplicated to the earliest canonical query. Later duplicates are counted separately and must not create extra matched or timeout records. +- In default mode, each pending canonical query owns any retry timestamps needed to regroup + unresolved attempts when an earlier query arrives across a batch boundary. Regrouping uses + earliest-first timeout windows for that identity, so chains of regressions cannot transitively + swallow attempts outside the final canonical window. Already finalized transactions are never + reopened. Monotonic mode allocates no retry history. This adds one optional pointer per pending + query and storage for default-mode retries; only earlier-canonical replacements regroup attempts. - Match identity preserves the observed presentation-form QNAME bytes instead of lowercasing them. This is a deliberate Community Edition trade-off, not a protocol guarantee. RFC 4343 defines ASCII label comparison as case-insensitive, and a valid response is allowed to differ from the diff --git a/docs/rfc/0004-forward-only-matcher.md b/docs/rfc/0004-forward-only-matcher.md index 5512a2f..7d73d4a 100644 --- a/docs/rfc/0004-forward-only-matcher.md +++ b/docs/rfc/0004-forward-only-matcher.md @@ -21,8 +21,9 @@ order within each shard and never backtracks. The key design choices: based). There is no shared mutable matcher state between shards. 2. **Canonical flow routing.** Before full DNS decode, the routing stage extracts a cheap - `CanonicalFlowKey` (client IP, resolver IP, port pair — always ordered so that - `client < resolver`) and hashes it to a shard index. This guarantees that a query and its + `CanonicalFlowKey` (observed client IP, client port and resolver IP, oriented by DNS QR) + and hashes it to a shard index. This includes the valid case where both ports are 53 and + guarantees that a query and its matching response always land in the same shard. 3. **Deterministic ordering.** Within a shard, packets are processed in strict @@ -31,11 +32,21 @@ order within each shard and never backtracks. The key design choices: 4. **Retry deduplication.** If a query with the same identity arrives while an earlier one is still pending inside the match-timeout window (1200 ms by default), the duplicate is counted - but doesn't create a second in-flight entry. One query → one terminal outcome (matched or - timeout), always. - - Match identity includes the DNS ID, observed name, client IP and port, resolver IP, and query - type. Resolver identity remains internal and is not added to the exported `DnsRecord` schema. + but doesn't create a second canonical query. One canonical query → one terminal outcome + (matched or timeout), always. Default mode retains retry timestamps inside that pending query's + payload. If an earlier query arrives across batches, the matcher regroups the identity's + unresolved attempts into earliest-first timeout windows. For example, pending attempts observed + at 3s, 2s and then 1s with a 1.2s window leave canonicals at 1s and 3s, rather than losing the + latter through transitive deduplication. Finalized transactions are never reopened. + + Retry history is allocated only when a duplicate is observed in default mode; monotonic mode + needs none. Regrouping costs `O(n log n)` in the affected identity's unresolved attempts and is + restricted to earlier-canonical replacements. The canonical query payload remains the sole owner + of retry history, with no secondary matcher map. + + Match identity includes the DNS ID, observed name, client IP and port, resolver IP, query + type, query class and opcode. Resolver identity, query class and opcode remain internal and are + not added to the exported `DnsRecord` schema. The current Community Edition identity key preserves the observed presentation-form QNAME bytes and does not lowercase them before matching. This is a deliberate Community Edition trade-off, @@ -61,7 +72,9 @@ order within each shard and never backtracks. The key design choices: These must hold for any valid implementation: - Each query reaches exactly one terminal outcome: matched once, or emitted once as a timeout. -- For a fixed input PCAP and configuration, output is bit-for-bit deterministic. +- For a fixed input PCAP and runtime configuration, finalized record values and order are + deterministic. Output container layout, such as Parquet row-group boundaries, need not be + byte-identical across runs. - Internal sequencing metadata (`packet_ordinal`, `record_ordinal`) never leaks into the exported `DnsRecord`. - Duplicate responses remain distinguishable in matcher state until matched or discarded. diff --git a/docs/rfc/0005-dual-path-pcap-parsing.md b/docs/rfc/0005-dual-path-pcap-parsing.md index cee59b8..42f210b 100644 --- a/docs/rfc/0005-dual-path-pcap-parsing.md +++ b/docs/rfc/0005-dual-path-pcap-parsing.md @@ -40,9 +40,10 @@ file's magic bytes when the input source is a regular file: using parser-owned probe-and-replay so the same stdin byte stream remains the single source of truth. -- **EOF-terminated stdin PCAPNG streams** → pure-Rust `pcapng` reader via `pcap-file`, for the - same reason: once stdin bytes have been inspected, the parser stays stream-native instead of - trying to reopen the stream through a second owner. +- **EOF-terminated stdin PCAPNG streams** → parser-owned pure-Rust block framing and one + section-local interface table, with stateless block/body/option validation via `pcap-file`. + Once stdin bytes have been inspected, the parser stays stream-native instead of trying to + reopen the stream through a second owner. - **EOF-terminated stdin streams with unknown magic** → explicit rejection. DPP does not create a temp file or hidden second ingest path to recover fallback compatibility for unsupported stdin @@ -52,6 +53,29 @@ The detection is a simple 4-byte magic check (`0xa1b2c3d4` or `0xd4c3b2a1` for c `0xa1b23c4d` or `0x4d3cb2a1` for nanosecond-resolution classic, `0x0a0d0d0a` for PCAPNG). For regular files, everything else goes to `libpcap`. For stdin streams, everything else is rejected. +### PCAPNG timestamp ownership + +The parser reads timestamp high/low words separately in section byte order for both Enhanced +Packet Blocks and legacy Packet Blocks. It applies each interface's `if_tsresol` and signed +`if_tsoffset` exactly once. All seven-bit resolution exponents are accepted; the complete raw +counter is scaled before integer rounding to microseconds, including sub-nanosecond units. +The final timestamp saturates to `i64` only after adding the signed offset. New sections reset +the interface table and may change byte order. + +The stateful `pcap-file 3.0.0-rc1` reader is deliberately bypassed: it swaps legacy little-endian +timestamp words, mis-scales binary resolutions, ignores offsets, and panics on exponent 30. +The newer `3.0.0-rc.3`, inspected on 2026-09-05, fixes those calculations but still rejects +sub-nanosecond resolutions and dates before the Unix epoch. Upgrading alone does not satisfy +this timestamp contract. The existing stateless decoder accepts the wire resolution field +without applying the faulty conversion, and remains responsible for structural validation. + +The parser validates referenced interfaces, captured/original/snap lengths, and uniqueness of +timestamp options. Input buffers grow only as bytes arrive; a huge declared length on a truncated +stream cannot trigger allocation of that entire declared block. Independent manually encoded +fixtures cover both byte orders and packet-block formats, interface offsets, resolution extremes, +section resets, truncation, and malformed lengths/options. Tests must not encode expected +timestamp layouts with the same dependency writer whose decoder they are validating. + ### IP and UDP boundaries The routing stage and shard-local DNS decoder share one `ParsedUdpDnsMeta` value containing the @@ -61,6 +85,13 @@ Ethernet padding and trailing capture bytes, never reach a DNS decoder. DPP has no IPv4 reassembly stage. IPv4 datagrams with the More Fragments flag or a non-zero fragment offset are therefore skipped rather than interpreting a fragment body as a complete UDP datagram. +IPv6 extraction traverses Hop-by-Hop, Routing, Destination Options, Authentication and atomic +Fragment headers with per-header bounds checks. Non-atomic fragments require reassembly and are +skipped. The DNS offset can exceed 65535 after a long valid extension chain. Internal metadata +stores a checked `u16` delta from the minimum 42-byte Ethernet/IPv4/UDP prefix and reconstructs +the absolute offset when accessing packet bytes. A maximum IPv6 frame ends at byte 65589 and +must leave at least 12 DNS bytes, so this delta covers the full supported range. The UDP-bounded +DNS length remains `u16`; routing metadata is 42 bytes on the measured macOS ARM64 target. ### DNS QNAME boundary @@ -69,6 +100,8 @@ same RFC 1035 boundary after name decompression. A QNAME may occupy at most 255 including label-length octets and the terminating root octet. Its escaped presentation form can be as large as 1003 bytes and remains valid input; DPP preserves it for matching and export rather than replacing it with an empty name. +Both question decoders reject compression pointers that point forward or overlap the current name. +The fast path's fallback must not become a way to accept a message the semantic decoder rejects. If any decompressed QNAME exceeds the wire limit, the entire DNS message is rejected before matcher or writer handoff. The processing counter `oversized_qname_message_count` increments exactly once diff --git a/src/custom_types.rs b/src/custom_types.rs index 2121177..4cdb4e5 100644 --- a/src/custom_types.rs +++ b/src/custom_types.rs @@ -8,6 +8,7 @@ use arrayvec::ArrayString; use hickory_proto::op::ResponseCode as HickoryResponseCode; use hickory_proto::rr::RecordType as HickoryRecordType; use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use std::borrow::Cow; use std::cmp::Ordering; use std::fmt; use std::hash::{Hash, Hasher}; @@ -63,8 +64,11 @@ impl From for HickoryRecordType { } impl ProtoRecordType { - pub fn as_str(&self) -> &'static str { - Into::<&str>::into(self.0) + pub fn as_str(&self) -> Cow<'static, str> { + match self.0 { + HickoryRecordType::Unknown(code) => Cow::Owned(format!("TYPE{code}")), + known => Cow::Borrowed(Into::<&str>::into(known)), + } } } @@ -87,7 +91,7 @@ impl Serialize for ProtoRecordType { where S: Serializer, { - serializer.serialize_str(Into::<&str>::into(self.0)) + serializer.serialize_str(&self.as_str()) } } @@ -97,9 +101,14 @@ impl<'de> Deserialize<'de> for ProtoRecordType { D: Deserializer<'de>, { let value = String::deserialize(deserializer)?; - let record_type = value - .parse::() - .map_err(serde::de::Error::custom)?; + let record_type = if let Some(code) = value.strip_prefix("TYPE") { + HickoryRecordType::from(code.parse::().map_err(serde::de::Error::custom)?) + } else { + value + .to_ascii_uppercase() + .parse::() + .map_err(serde::de::Error::custom)? + }; Ok(ProtoRecordType(record_type)) } } @@ -205,10 +214,13 @@ impl ProtoResponseCode { self.code } - pub fn as_str(&self) -> &'static str { + pub fn as_str(&self) -> Cow<'static, str> { match (self.context, self.as_u16()) { - (ResponseCodeContext::EdnsOpt, 16) => "EDNS_BADVERS", - _ => Into::::into(*self).to_str(), + (ResponseCodeContext::EdnsOpt, 16) => Cow::Borrowed("EDNS_BADVERS"), + _ => match Into::::into(*self) { + HickoryResponseCode::Unknown(code) => Cow::Owned(code.to_string()), + known => Cow::Borrowed(known.to_str()), + }, } } } @@ -242,7 +254,7 @@ impl Serialize for ProtoResponseCode { where S: Serializer, { - serializer.serialize_str(self.as_str()) + serializer.serialize_str(&self.as_str()) } } @@ -594,9 +606,10 @@ impl<'de, const N: usize> Deserialize<'de> for FixedSizeString { #[cfg(test)] mod tests { use super::{ - DNS_NAME_PRESENTATION_MAX_LENGTH, DnsNameBuf, FixedSizeString, HickoryResponseCode, - ProtoResponseCode, + DNS_NAME_PRESENTATION_MAX_LENGTH, DnsNameBuf, FixedSizeString, HickoryRecordType, + HickoryResponseCode, ProtoRecordType, ProtoResponseCode, }; + use std::borrow::Cow; use std::mem::size_of; #[test] @@ -608,6 +621,43 @@ mod tests { assert_eq!(u16::from(parsed), 4095); } + #[test] + fn unknown_protocol_values_preserve_numeric_identity_through_serde() { + for value in [65400, 65401, u16::MAX] { + let record_type = ProtoRecordType::from(HickoryRecordType::Unknown(value)); + let encoded = serde_json::to_string(&record_type).expect("type serializes"); + assert_eq!(encoded, format!("\"TYPE{value}\"")); + let decoded: ProtoRecordType = serde_json::from_str(&encoded).expect("type parses"); + assert_eq!(decoded, record_type); + } + for value in [64, 65, 4095, u16::MAX] { + let response_code = ProtoResponseCode::from(HickoryResponseCode::Unknown(value)); + let encoded = serde_json::to_string(&response_code).expect("rcode serializes"); + assert_eq!(encoded, format!("\"{value}\"")); + let decoded: ProtoResponseCode = serde_json::from_str(&encoded).expect("rcode parses"); + assert_eq!(decoded.as_u16(), value); + } + } + + #[test] + fn known_protocol_labels_remain_borrowed() { + assert!(matches!( + ProtoRecordType::from(HickoryRecordType::A).as_str(), + Cow::Borrowed("A") + )); + assert!(matches!( + ProtoResponseCode::from(HickoryResponseCode::NoError).as_str(), + Cow::Borrowed("No Error") + )); + } + + #[test] + fn generic_record_type_rejects_out_of_range_or_missing_code() { + for value in ["TYPE", "TYPE65536", "TYPE-1", "TYPEbad"] { + assert!(serde_json::from_str::(&format!("\"{value}\"")).is_err()); + } + } + #[test] fn proto_response_code_round_trips_string_representation() { let serialized = diff --git a/src/dns_processor/matcher.rs b/src/dns_processor/matcher.rs index 05c192a..e2f4bd6 100644 --- a/src/dns_processor/matcher.rs +++ b/src/dns_processor/matcher.rs @@ -52,7 +52,7 @@ impl DnsProcessor { response_timestamp_micros: i64, response_code: ProtoResponseCode, ) -> DnsRecord { - let (id, name, client_ip, src_port, query_type, _resolver_ip) = query_identity; + let (id, name, client_ip, src_port, query_type, ..) = query_identity; DnsRecord { request_timestamp: query_key.timestamp_micros, response_timestamp: Some(response_timestamp_micros), @@ -70,7 +70,7 @@ impl DnsProcessor { query_identity: &QueryIdentityKey, query_key: TimelineKey, ) -> DnsRecord { - let (id, name, client_ip, src_port, query_type, _resolver_ip) = query_identity; + let (id, name, client_ip, src_port, query_type, ..) = query_identity; DnsRecord { request_timestamp: query_key.timestamp_micros, response_timestamp: None, @@ -103,28 +103,100 @@ impl DnsProcessor { &self, record: &ProcessedDnsRecord, state: &mut MatcherShardState, - output_records: &mut OutputRecordBatches, - dns_query_count: &mut usize, - duplicated_query_count: &mut usize, - matched_query_response_count: &mut usize, - matched_rtt_sum_micros: &mut u64, + result: &mut ShardProcessingResult, ) { let query_identity = self.query_identity_from_record(record); let query_key = self.timeline_key_from_record(record); - *dns_query_count += 1; + result.dns_query_count += 1; if let Some(pending_key) = self.pending_query_within_timeout(state, &query_identity, query_key.timestamp_micros) { - *duplicated_query_count += 1; if pending_key <= query_key { + result.duplicated_query_count += 1; + if !self.monotonic_capture { + state + .query_map + .get_mut(&query_identity) + .and_then(|timeline| timeline.get_mut(pending_key)) + .expect("pending retry handle must remain valid") + .retain_retry(query_key); + } return; } - self.remove_query_entry(state, &query_identity, pending_key) - .expect("pending retry handle must remain valid until replacement"); + // An earlier attempt can split a retry group and move every later group + // boundary for this identity. Repartition unresolved attempts, never an + // already finalized transaction, using the same earliest-query window. + debug_assert!(!self.monotonic_capture); + let timeline = state + .query_map + .remove(&query_identity) + .expect("pending retry identity must remain valid"); + let (queries, duplicate_increment) = + self.repartition_pending_queries(timeline, query_key); + result.duplicated_query_count += duplicate_increment; + for (key, payload) in queries { + self.match_or_insert_query(query_identity.clone(), key, payload, state, result); + } + return; + } + + self.match_or_insert_query( + query_identity, + query_key, + QueryEventPayload::default(), + state, + result, + ); + } + + fn repartition_pending_queries( + &self, + timeline: Timeline, + new_key: TimelineKey, + ) -> (Vec<(TimelineKey, QueryEventPayload)>, usize) { + let mut attempts = vec![new_key]; + let mut previous_query_count = 0_usize; + timeline.into_entries(|key, payload| { + previous_query_count += 1; + attempts.push(key); + payload.append_retries_to(&mut attempts); + }); + attempts.sort_unstable(); + + let mut queries: Vec<(TimelineKey, QueryEventPayload)> = Vec::new(); + for key in attempts { + if let Some((canonical_key, payload)) = queries.last_mut() + && key.timestamp_micros + <= canonical_key + .timestamp_micros + .saturating_add(self.match_timeout_micros) + { + payload.retain_retry(key); + } else { + queries.push((key, QueryEventPayload::default())); + } } + // Greedy earliest-first windows use the minimum number of fixed-width + // intervals. Adding one attempt adds either zero or one canonical query; + // duplicate counts therefore need no retroactive decrement. + let duplicate_increment = (previous_query_count + 1) + .checked_sub(queries.len()) + .expect("one inserted attempt cannot add multiple canonical queries"); + debug_assert!(duplicate_increment <= 1); + (queries, duplicate_increment) + } + + fn match_or_insert_query( + &self, + query_identity: QueryIdentityKey, + query_key: TimelineKey, + payload: QueryEventPayload, + state: &mut MatcherShardState, + result: &mut ShardProcessingResult, + ) { if let Some((_, response_handle)) = self.find_closest_response( state, &query_identity, @@ -137,19 +209,21 @@ impl DnsProcessor { let response_payload = self .remove_response_entry(state, &query_identity, response_handle) .expect("matched response handle must remain valid until removal"); - output_records.push(self.create_matched_record_from_query_parts( - &query_identity, - query_key, - response_handle.timestamp_micros, - response_payload.response_code, - )); - *matched_query_response_count += 1; - *matched_rtt_sum_micros += response_handle + result + .output_records + .push(self.create_matched_record_from_query_parts( + &query_identity, + query_key, + response_handle.timestamp_micros, + response_payload.response_code, + )); + result.matched_query_response_count += 1; + result.matched_rtt_sum_micros += response_handle .timestamp_micros .saturating_sub(query_key.timestamp_micros) .max(0) as u64; } else { - self.insert_query_entry(state, query_identity, query_key); + self.insert_query_entry_with_payload(state, query_identity, query_key, payload); } } @@ -293,17 +367,33 @@ impl DnsProcessor { self.evict_responses_before(state, threshold_timestamp_micros); } + #[cfg(test)] pub(super) fn insert_query_entry( &self, state: &mut MatcherShardState, identity: QueryIdentityKey, timeline_key: TimelineKey, + ) { + self.insert_query_entry_with_payload( + state, + identity, + timeline_key, + QueryEventPayload::default(), + ); + } + + fn insert_query_entry_with_payload( + &self, + state: &mut MatcherShardState, + identity: QueryIdentityKey, + timeline_key: TimelineKey, + payload: QueryEventPayload, ) { let replaced = state .query_map .entry(identity) .or_default() - .insert(timeline_key, QueryEventPayload); + .insert(timeline_key, payload); if replaced.is_some() { debug_assert!( @@ -402,6 +492,8 @@ impl DnsProcessor { client_port, record.query_type, resolver_ip, + record.query_class, + record.opcode, ) } @@ -419,6 +511,8 @@ impl DnsProcessor { client_port, record.query_type, resolver_ip, + record.query_class, + record.opcode, ) } @@ -431,6 +525,8 @@ impl DnsProcessor { query.src_port, query.query_type, query.resolver_ip, + query.query_class, + query.opcode, ) } @@ -470,15 +566,7 @@ impl DnsProcessor { &mut result.matched_rtt_sum_micros, ); } else { - self.process_query( - &record, - state, - &mut result.output_records, - &mut result.dns_query_count, - &mut result.duplicated_query_count, - &mut result.matched_query_response_count, - &mut result.matched_rtt_sum_micros, - ); + self.process_query(&record, state, &mut result); } } diff --git a/src/dns_processor/parser.rs b/src/dns_processor/parser.rs index 6759eda..6df10fe 100644 --- a/src/dns_processor/parser.rs +++ b/src/dns_processor/parser.rs @@ -24,9 +24,16 @@ const IPV4_MIN_HEADER_LEN: usize = 20; const IPV6_HEADER_LEN: usize = 40; const UDP_HEADER_LEN: usize = 8; const DNS_HEADER_LEN: usize = 12; +const MIN_DNS_OFFSET: usize = ETHERNET_HEADER_LEN + IPV4_MIN_HEADER_LEN + UDP_HEADER_LEN; const ETHER_TYPE_IPV4: u16 = 0x0800; const ETHER_TYPE_IPV6: u16 = 0x86dd; const IP_PROTOCOL_UDP: u8 = 17; +const IPV6_HOP_BY_HOP: u8 = 0; +const IPV6_ROUTING: u8 = 43; +const IPV6_FRAGMENT: u8 = 44; +const IP_AUTHENTICATION: u8 = 51; +const IPV6_DESTINATION_OPTIONS: u8 = 60; +const IPV6_FRAGMENT_OFFSET_AND_MORE: u16 = 0xfff9; const IPV4_MORE_FRAGMENTS: u16 = 0x2000; const IPV4_FRAGMENT_OFFSET_MASK: u16 = 0x1fff; const DNS_PORT: u16 = 53; @@ -48,12 +55,18 @@ pub(super) struct CanonicalFlowKey { #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(super) struct ParsedUdpDnsMeta { pub(super) flow_key: CanonicalFlowKey, - pub(super) dns_offset: u16, + // DNS starts at least at byte 42. A maximum IPv6 frame ends at byte 65589; + // reserving the validated 12-byte DNS header bounds this delta to 65535. + dns_offset_delta: u16, pub(super) dns_len: u16, pub(super) is_response: bool, } impl ParsedUdpDnsMeta { + fn dns_offset(self) -> usize { + MIN_DNS_OFFSET + usize::from(self.dns_offset_delta) + } + fn src_ip(self) -> IpAddr { if self.is_response { self.flow_key.resolver_ip @@ -87,7 +100,7 @@ impl ParsedUdpDnsMeta { } fn dns_data(self, data: &[u8]) -> Result<&[u8], &'static str> { - let start = usize::from(self.dns_offset); + let start = self.dns_offset(); let end = start .checked_add(usize::from(self.dns_len)) .ok_or("Failed to parse UDP packet")?; @@ -97,12 +110,14 @@ impl ParsedUdpDnsMeta { struct DecodedDnsHeader { id: u16, + opcode: u8, response_code: ProtoResponseCode, } struct DecodedDnsQuestion { name: DnsNameBuf, query_type: HickoryRecordType, + query_class: u16, } #[derive(Debug)] @@ -371,6 +386,8 @@ impl DnsProcessor { is_query: !meta.is_response, name: query.name, query_type: query.query_type, + query_class: query.query_class, + opcode: header.opcode, response_code, }); } @@ -425,10 +442,14 @@ impl DnsProcessor { "DNS question truncated", )?); cursor += 2; - let _query_class = Self::parse_u16_at(dns_data, cursor, "DNS question truncated")?; + let query_class = Self::parse_u16_at(dns_data, cursor, "DNS question truncated")?; cursor += 2; - queries.push(DecodedDnsQuestion { name, query_type }); + queries.push(DecodedDnsQuestion { + name, + query_type, + query_class, + }); } let mut response_code = ProtoResponseCode::from(HickoryResponseCode::from(0, low_response_code)); @@ -459,7 +480,11 @@ impl DnsProcessor { )?; } } - let header = DecodedDnsHeader { id, response_code }; + let header = DecodedDnsHeader { + id, + opcode: ((flags >> 11) & 0x0f) as u8, + response_code, + }; Ok((header, queries)) } @@ -489,6 +514,7 @@ impl DnsProcessor { Ok(( DecodedDnsHeader { id: header.id, + opcode: header.op_code.into(), response_code, }, queries @@ -498,6 +524,7 @@ impl DnsProcessor { name: Self::format_domain_name(query.name()) .map_err(|_| DnsQuestionDecodeError::Invalid)?, query_type: query.query_type(), + query_class: query.query_class().into(), }) }) .collect::, DnsQuestionDecodeError>>()?, @@ -606,7 +633,12 @@ impl DnsProcessor { let error = Self::parse_u16_at(dns_data, cursor, "TSIG record truncated")?; cursor += 2; - Self::skip_bytes_in_range(&mut cursor, rdata_end, 2)?; + let other_len = usize::from(Self::parse_u16_at( + dns_data, + cursor, + "TSIG record truncated", + )?); + Self::skip_bytes_in_range(&mut cursor, rdata_end, 2 + other_len)?; if cursor != rdata_end { return Err("TSIG record has trailing data"); } @@ -681,9 +713,14 @@ impl DnsProcessor { let mut position = *cursor; let mut resume_position = None; let mut jump_count = 0; + let mut segment_start = position; + let mut segment_end = dns_data.len(); loop { - let length = *dns_data.get(position).ok_or("DNS name truncated")?; + let length = *dns_data + .get(..segment_end) + .and_then(|segment| segment.get(position)) + .ok_or("DNS name truncated")?; match length { 0 => { @@ -693,13 +730,14 @@ impl DnsProcessor { } _ if (length & DNS_POINTER_MASK) == DNS_POINTER_TAG => { let next = *dns_data - .get(position + 1) + .get(..segment_end) + .and_then(|segment| segment.get(position + 1)) .ok_or("DNS compression pointer truncated")?; let offset = (((length & DNS_LABEL_LEN_MASK) as usize) << 8) | usize::from(next); - if offset >= dns_data.len() { - return Err("DNS compression pointer out of bounds"); + if offset >= segment_start { + return Err("DNS compression pointer is not prior to name"); } if resume_position.is_none() { @@ -711,6 +749,8 @@ impl DnsProcessor { return Err("DNS compression pointer loop"); } + segment_end = segment_start; + segment_start = offset; position = offset; } _ if (length & DNS_POINTER_MASK) != 0 => { @@ -719,7 +759,8 @@ impl DnsProcessor { _ => { let label_len = usize::from(length); dns_data - .get(position + 1..position + 1 + label_len) + .get(..segment_end) + .and_then(|segment| segment.get(position + 1..position + 1 + label_len)) .ok_or("DNS label truncated")?; position += 1 + label_len; } @@ -738,9 +779,14 @@ impl DnsProcessor { let mut jump_count = 0; let mut expanded_wire_len = 1_usize; let mut oversized = false; + let mut segment_start = position; + let mut segment_end = dns_data.len(); loop { - let length = *dns_data.get(position).ok_or("DNS name truncated")?; + let length = *dns_data + .get(..segment_end) + .and_then(|segment| segment.get(position)) + .ok_or("DNS name truncated")?; match length { 0 => { @@ -759,12 +805,15 @@ impl DnsProcessor { } _ if (length & DNS_POINTER_MASK) == DNS_POINTER_TAG => { let next = *dns_data - .get(position + 1) + .get(..segment_end) + .and_then(|segment| segment.get(position + 1)) .ok_or("DNS compression pointer truncated")?; let offset = (((length & DNS_LABEL_LEN_MASK) as usize) << 8) | usize::from(next); - if offset >= dns_data.len() { + // Each pointer must refer to an earlier, nonoverlapping name, + // matching Hickory's question decoder (RFC 1035 section 4.1.4). + if offset >= segment_start { return Err(DnsQuestionDecodeError::Invalid); } @@ -777,6 +826,8 @@ impl DnsProcessor { return Err(DnsQuestionDecodeError::Invalid); } + segment_end = segment_start; + segment_start = offset; position = offset; } _ if (length & DNS_POINTER_MASK) != 0 => { @@ -785,7 +836,8 @@ impl DnsProcessor { _ => { let label_len = usize::from(length); let label = dns_data - .get(position + 1..position + 1 + label_len) + .get(..segment_end) + .and_then(|segment| segment.get(position + 1..position + 1 + label_len)) .ok_or("DNS label truncated")?; expanded_wire_len = expanded_wire_len.saturating_add(1 + label_len); @@ -913,10 +965,6 @@ impl DnsProcessor { .get(IPV6_HEADER_LEN..packet_length) .ok_or("Failed to parse IPv6 packet")?; - if header[6] != IP_PROTOCOL_UDP { - return Ok(None); - } - let src_ip = IpAddr::V6(Ipv6Addr::from( <[u8; 16]>::try_from(&header[8..24]).unwrap(), )); @@ -924,7 +972,60 @@ impl DnsProcessor { <[u8; 16]>::try_from(&header[24..40]).unwrap(), )); - Self::extract_udp_dns_from_transport(payload, src_ip, dst_ip, l3_offset + IPV6_HEADER_LEN) + let mut next_header = header[6]; + let mut cursor = 0; + loop { + if next_header == IP_PROTOCOL_UDP { + return Self::extract_udp_dns_from_transport( + &payload[cursor..], + src_ip, + dst_ip, + l3_offset + IPV6_HEADER_LEN + cursor, + ); + } + + let minimum_length = match next_header { + IPV6_HOP_BY_HOP if cursor != 0 => { + return Err("IPv6 Hop-by-Hop header must follow IPv6 header"); + } + IPV6_HOP_BY_HOP | IPV6_ROUTING | IPV6_DESTINATION_OPTIONS | IPV6_FRAGMENT => 8, + IP_AUTHENTICATION => 12, + // Other transport protocols, ESP and No Next Header contain no + // directly decodable UDP payload at this extraction boundary. + _ => return Ok(None), + }; + let extension = payload + .get(cursor..) + .filter(|remaining| remaining.len() >= minimum_length) + .ok_or("IPv6 extension header truncated")?; + let extension_length = match next_header { + IPV6_FRAGMENT => { + let offset_and_flags = u16::from_be_bytes([extension[2], extension[3]]); + if offset_and_flags & IPV6_FRAGMENT_OFFSET_AND_MORE != 0 { + // An atomic fragment needs no reassembly; other fragments do. + return Ok(None); + } + 8 + } + IP_AUTHENTICATION => { + // AH counts 32-bit words excluding the first two words, + // while other extension lengths count 8-octet units. + let length = (usize::from(extension[1]) + 2) * 4; + if length < minimum_length || length % 8 != 0 { + return Err("Invalid IPv6 Authentication header length"); + } + length + } + _ => (usize::from(extension[1]) + 1) * 8, + }; + extension + .get(..extension_length) + .ok_or("IPv6 extension header truncated")?; + next_header = extension[0]; + // Every accepted extension consumes at least eight bytes, so even + // repeated headers are bounded by the declared IPv6 Payload Length. + cursor += extension_length; + } } fn extract_udp_dns_from_transport( @@ -954,18 +1055,26 @@ impl DnsProcessor { if dns_data.len() < DNS_HEADER_LEN { return Err("DNS data too short"); } + let is_response = dns_data[2] & 0x80 != 0; + if (is_response && src_port != DNS_PORT) || (!is_response && dst_port != DNS_PORT) { + return Ok(None); + } let dns_offset = l4_offset .checked_add(UDP_HEADER_LEN) .ok_or("Failed to parse UDP packet")?; Ok(Some(ParsedUdpDnsMeta { - flow_key: Self::canonical_flow_key(src_ip, dst_ip, src_port, dst_port), - dns_offset: u16::try_from(dns_offset) - .map_err(|_| "UDP DNS offset exceeds supported range")?, + flow_key: Self::canonical_flow_key(src_ip, dst_ip, src_port, dst_port, is_response), + dns_offset_delta: u16::try_from( + dns_offset + .checked_sub(MIN_DNS_OFFSET) + .ok_or("UDP DNS offset precedes minimum header length")?, + ) + .map_err(|_| "UDP DNS offset exceeds supported range")?, dns_len: u16::try_from(dns_data.len()) .map_err(|_| "UDP DNS payload exceeds supported range")?, - is_response: src_port == DNS_PORT, + is_response, })) } @@ -974,8 +1083,9 @@ impl DnsProcessor { dst_ip: IpAddr, src_port: u16, dst_port: u16, + is_response: bool, ) -> CanonicalFlowKey { - if src_port == DNS_PORT { + if is_response { CanonicalFlowKey { client_ip: dst_ip, client_port: dst_port, @@ -990,3 +1100,329 @@ impl DnsProcessor { } } } + +#[cfg(test)] +mod protocol_regression_tests { + use super::*; + use crate::test_support::{encode_dns_header, make_udp_dns_packet_with_payload}; + + fn processors() -> [DnsProcessor; 2] { + [ + DnsProcessor::new(None).unwrap(), + DnsProcessor::new_with_dns_wire_fast_path(None, true).unwrap(), + ] + } + + fn question(flags: u16, query_class: u16) -> Vec { + let mut dns = encode_dns_header(0xbeef, flags, 1); + dns.extend_from_slice(b"\x07example\x03com\0\0\x01"); + dns.extend_from_slice(&query_class.to_be_bytes()); + dns + } + + fn ipv4(dns: &[u8], response: bool, client_port: u16) -> Vec { + let (src, dst, src_port, dst_port) = if response { + ([8, 8, 8, 8], [10, 0, 0, 1], 53, client_port) + } else { + ([10, 0, 0, 1], [8, 8, 8, 8], client_port, 53) + }; + make_udp_dns_packet_with_payload(src, dst, src_port, dst_port, dns) + } + + fn ipv6(dns: &[u8], extensions: &[(u8, Vec)]) -> Vec { + let mut payload = Vec::new(); + for (index, (_, extension)) in extensions.iter().enumerate() { + let mut extension = extension.clone(); + extension[0] = extensions.get(index + 1).map_or(17, |next| next.0); + payload.extend_from_slice(&extension); + } + payload.extend_from_slice(&53000_u16.to_be_bytes()); + payload.extend_from_slice(&53_u16.to_be_bytes()); + payload.extend_from_slice(&u16::try_from(8 + dns.len()).unwrap().to_be_bytes()); + payload.extend_from_slice(&[0, 0]); + payload.extend_from_slice(dns); + + let mut packet = vec![0; ETHERNET_HEADER_LEN + IPV6_HEADER_LEN]; + packet[12..14].copy_from_slice(ÐER_TYPE_IPV6.to_be_bytes()); + packet[14] = 0x60; + packet[18..20].copy_from_slice(&u16::try_from(payload.len()).unwrap().to_be_bytes()); + packet[20] = extensions.first().map_or(17, |first| first.0); + packet[21] = 64; + packet[22..38].copy_from_slice(&"2001:db8::1".parse::().unwrap().octets()); + packet[38..54].copy_from_slice(&"2001:db8::2".parse::().unwrap().octets()); + packet.extend_from_slice(&payload); + packet + } + + #[test] + fn qr_routes_port_53_query_and_response_to_same_flow() { + let query = ipv4(&question(0x0100, 1), false, 53); + let response = ipv4(&question(0x8180, 1), true, 53); + let query_meta = DnsProcessor::packet_routing_meta(&query).unwrap(); + let response_meta = DnsProcessor::packet_routing_meta(&response).unwrap(); + assert!(!query_meta.is_response); + assert!(response_meta.is_response); + assert_eq!(query_meta.flow_key, response_meta.flow_key); + assert_eq!(query_meta.flow_key.client_ip, IpAddr::from([10, 0, 0, 1])); + assert_eq!(query_meta.flow_key.client_port, 53); + + for processor in processors() { + let queries = processor.process_packet_batch(&query, 0).unwrap(); + let responses = processor.process_packet_batch(&response, 100).unwrap(); + assert_eq!(queries.len(), 1); + assert_eq!(responses.len(), 1); + assert!(queries[0].is_query); + assert!(!responses[0].is_query); + assert_eq!(queries[0].src_ip, responses[0].dst_ip); + } + } + + #[test] + fn qr_does_not_create_candidates_on_wrong_server_port() { + for packet in [ + ipv4(&question(0x8180, 1), false, 53000), + ipv4(&question(0x0100, 1), true, 53000), + ] { + assert!(DnsProcessor::packet_routing_meta(&packet).is_none()); + } + } + + #[test] + fn both_decoders_preserve_query_class_and_every_opcode() { + for opcode in 0..16 { + for query_class in [1, 3, 4, 65535] { + let packet = ipv4( + &question(0x0100 | (opcode << 11), query_class), + false, + 53000, + ); + for processor in processors() { + let records = processor.process_packet_batch(&packet, 0).unwrap(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].query_class, query_class); + assert_eq!(records[0].opcode, opcode as u8); + } + } + } + } + + fn tsig_response(error: u16, other_len: u16, other_data: &[u8]) -> Vec { + let mut dns = question(if error == 0 { 0x8180 } else { 0x8189 }, 1); + dns[10..12].copy_from_slice(&1_u16.to_be_bytes()); + let mut rdata = b"\x0bhmac-sha256\0".to_vec(); + rdata.extend_from_slice(&[0; 6]); + rdata.extend_from_slice(&300_u16.to_be_bytes()); + rdata.extend_from_slice(&32_u16.to_be_bytes()); + rdata.extend_from_slice(&[0; 32]); + rdata.extend_from_slice(&0xbeef_u16.to_be_bytes()); + rdata.extend_from_slice(&error.to_be_bytes()); + rdata.extend_from_slice(&other_len.to_be_bytes()); + rdata.extend_from_slice(other_data); + dns.extend_from_slice(b"\x03key\x07example\0"); + dns.extend_from_slice(&DNS_TSIG_RECORD_TYPE.to_be_bytes()); + dns.extend_from_slice(&255_u16.to_be_bytes()); + dns.extend_from_slice(&[0; 4]); + dns.extend_from_slice(&u16::try_from(rdata.len()).unwrap().to_be_bytes()); + dns.extend_from_slice(&rdata); + dns + } + + #[test] + fn tsig_badtime_accepts_required_server_time_other_data() { + for (error, other_data) in [(0, &[][..]), (18, &[0, 0, 1, 2, 3, 4][..])] { + let dns = tsig_response(error, other_data.len() as u16, other_data); + let packet = ipv4(&dns, true, 53000); + for processor in processors() { + let records = processor.process_packet_batch(&packet, 100).unwrap(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].response_code.as_u16(), error); + } + } + } + + #[test] + fn tsig_other_data_must_match_declared_length() { + for (other_len, other_data) in [(6, &[0; 5][..]), (0, &[0; 6][..]), (65535, &[][..])] { + let packet = ipv4(&tsig_response(18, other_len, other_data), true, 53000); + for processor in processors() { + assert!(processor.process_packet_batch(&packet, 100).is_none()); + } + } + } + + #[test] + fn forward_qname_pointer_is_rejected_by_both_decoders() { + let mut dns = encode_dns_header(0xbeef, 0x8180, 1); + dns[6..8].copy_from_slice(&1_u16.to_be_bytes()); + // QNAME at offset 12 illegally refers forward to the answer owner at 18. + dns.extend_from_slice(b"\xc0\x12\0\x01\0\x01\x07example\x03com\0"); + dns.extend_from_slice(&[0, 1, 0, 1, 0, 0, 0, 60, 0, 4, 1, 2, 3, 4]); + let packet = ipv4(&dns, true, 53000); + for processor in processors() { + assert!(processor.process_packet_batch(&packet, 0).is_none()); + } + } + + #[test] + fn compression_rejects_overlapping_names_but_accepts_prior_questions() { + let overlap = [4, b'a', b'b', 0xc0, 0, 0]; + assert!(DnsProcessor::read_wire_domain_name(&overlap, &mut 3).is_err()); + assert!(Name::read(&mut BinDecoder::new(&overlap).clone(3)).is_err()); + + let mut dns = question(0x0100, 1); + dns[4..6].copy_from_slice(&2_u16.to_be_bytes()); + dns.extend_from_slice(b"\xc0\x0c\0\x1c\0\x01"); + for processor in processors() { + let packet = ipv4(&dns, false, 53000); + let records = processor.process_packet_batch(&packet, 0).unwrap(); + assert_eq!(records.len(), 2); + assert_eq!(records[0].name, records[1].name); + assert_eq!(records[1].query_type, HickoryRecordType::AAAA); + } + } + + #[test] + fn ipv6_walks_extension_chains_and_atomic_fragments() { + let hbh = (IPV6_HOP_BY_HOP, vec![0; 8]); + let dest = (IPV6_DESTINATION_OPTIONS, vec![0; 8]); + // An unrecognized Routing Type with Segments Left zero is skipped by IPv6. + let routing = (IPV6_ROUTING, vec![0, 0, 253, 0, 0, 0, 0, 0]); + let atomic = (IPV6_FRAGMENT, vec![0; 8]); + let mut ah = vec![0; 16]; + ah[1] = 2; + let ah = (IP_AUTHENTICATION, ah); + let chains = [ + vec![], + vec![hbh.clone()], + vec![dest.clone()], + vec![routing.clone()], + vec![atomic.clone()], + vec![ah.clone()], + vec![hbh, dest.clone(), routing, atomic, ah, dest], + ]; + let dns = question(0x0100, 1); + for chain in chains { + let packet = ipv6(&dns, &chain); + let meta = DnsProcessor::packet_routing_meta(&packet).unwrap(); + assert_eq!(meta.dns_data(&packet).unwrap(), dns); + for processor in processors() { + let records = processor.process_packet_batch(&packet, 0).unwrap(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].name.as_str(), "example.com"); + } + } + } + + #[test] + fn ipv6_skips_fragments_requiring_reassembly() { + for flags in [1_u16, 8, 9] { + let mut fragment = vec![0; 8]; + fragment[2..4].copy_from_slice(&flags.to_be_bytes()); + let packet = ipv6(&question(0x0100, 1), &[(IPV6_FRAGMENT, fragment)]); + assert!(DnsProcessor::packet_routing_meta(&packet).is_none()); + } + } + + #[test] + fn ipv6_extension_lengths_cannot_consume_capture_padding() { + let dns = question(0x0100, 1); + for extension_type in [ + IPV6_HOP_BY_HOP, + IPV6_ROUTING, + IPV6_FRAGMENT, + IP_AUTHENTICATION, + IPV6_DESTINATION_OPTIONS, + ] { + let packet = ipv6(&dns, &[(extension_type, vec![0; 8])]); + for captured_extension_bytes in 0..8_u16 { + let mut truncated = packet.clone(); + // Keep captured bytes intact while limiting the declared IP payload. + truncated[18..20].copy_from_slice(&captured_extension_bytes.to_be_bytes()); + assert!(DnsProcessor::extract_udp_dns_meta(&truncated).is_err()); + } + } + let mut oversized_extension = ipv6(&dns, &[(IPV6_DESTINATION_OPTIONS, vec![0; 8])]); + oversized_extension[55] = 255; + assert!(DnsProcessor::extract_udp_dns_meta(&oversized_extension).is_err()); + for ah_length in [0, 1, 255] { + let mut ah = vec![0; 16]; + ah[1] = ah_length; + let packet = ipv6(&dns, &[(IP_AUTHENTICATION, ah)]); + assert!(DnsProcessor::extract_udp_dns_meta(&packet).is_err()); + } + } + + #[test] + fn ipv6_dns_offset_can_exceed_u16_after_long_extension_chain() { + let dns = question(0x0100, 1); + let chain = vec![(IPV6_DESTINATION_OPTIONS, vec![0; 8]); 8186]; + let packet = ipv6(&dns, &chain); + let meta = DnsProcessor::packet_routing_meta(&packet).unwrap(); + assert!(meta.dns_offset() > usize::from(u16::MAX)); + assert_eq!(meta.dns_data(&packet).unwrap(), dns); + for processor in processors() { + assert_eq!(processor.process_packet_batch(&packet, 0).unwrap().len(), 1); + } + } + + #[test] + fn dns_offsets_cover_minimum_headers_and_maximum_ip_payloads() { + for is_ipv6 in [false, true] { + let ip_header_len = if is_ipv6 { + IPV6_HEADER_LEN + } else { + IPV4_MIN_HEADER_LEN + }; + let dns_len = + usize::from(u16::MAX) - UDP_HEADER_LEN - if is_ipv6 { 0 } else { ip_header_len }; + let mut dns = question(0x0100, 1); + dns[10..12].copy_from_slice(&1_u16.to_be_bytes()); + // Fill the maximum UDP payload with a valid EDNS Padding option. + let padding_len = dns_len - dns.len() - 11 - 4; + dns.push(0); + dns.extend_from_slice(&DNS_OPT_RECORD_TYPE.to_be_bytes()); + dns.extend_from_slice(&u16::MAX.to_be_bytes()); + dns.extend_from_slice(&[0; 4]); + dns.extend_from_slice(&u16::try_from(4 + padding_len).unwrap().to_be_bytes()); + dns.extend_from_slice(&12_u16.to_be_bytes()); + dns.extend_from_slice(&u16::try_from(padding_len).unwrap().to_be_bytes()); + dns.resize(dns_len, 0); + let packet = if is_ipv6 { + ipv6(&dns, &[]) + } else { + ipv4(&dns, false, 53000) + }; + let meta = DnsProcessor::packet_routing_meta(&packet).unwrap(); + assert_eq!( + meta.dns_offset(), + ETHERNET_HEADER_LEN + ip_header_len + UDP_HEADER_LEN + ); + assert_eq!(usize::from(meta.dns_len), dns_len); + assert_eq!(meta.dns_data(&packet).unwrap(), dns); + if !is_ipv6 { + assert_eq!(meta.dns_offset_delta, 0); + } + for processor in processors() { + assert_eq!(processor.process_packet_batch(&packet, 0).unwrap().len(), 1); + } + } + } + + #[test] + fn relative_dns_offset_covers_largest_ipv6_frame_and_extension_prefix() { + let dns = encode_dns_header(0xbeef, 0x0100, 0); + // Extension lengths are multiples of eight; this is the largest prefix + // leaving room for UDP and the minimum 12-byte DNS message. + let chain = vec![(IPV6_DESTINATION_OPTIONS, vec![0; 8]); 8189]; + let mut packet = ipv6(&dns, &chain); + packet.extend_from_slice(&[0; 3]); + packet[18..20].copy_from_slice(&u16::MAX.to_be_bytes()); + assert_eq!( + packet.len(), + ETHERNET_HEADER_LEN + IPV6_HEADER_LEN + usize::from(u16::MAX) + ); + let meta = DnsProcessor::packet_routing_meta(&packet).unwrap(); + assert_eq!(meta.dns_offset(), 65574); + assert_eq!(meta.dns_data(&packet).unwrap(), dns); + } +} diff --git a/src/dns_processor/pipeline.rs b/src/dns_processor/pipeline.rs index 9d26717..18c748e 100644 --- a/src/dns_processor/pipeline.rs +++ b/src/dns_processor/pipeline.rs @@ -5,6 +5,7 @@ * Commercial licensing options: . */ +use arrayvec::ArrayVec; use crossbeam::channel::{Receiver, Sender}; use rayon::prelude::*; use seahash::SeaHasher; @@ -409,13 +410,103 @@ fn routed_shard_capacity_hint(packet_count: usize, shard_count: usize) -> usize (packet_count / shard_count) + 1 } +// Buffer the tuple's existing Hash writes so routing runs SeaHash once per flow. +// Streaming fallback keeps future Hash implementations correct if they exceed this capacity. +enum BufferedFlowHasher { + Buffered(ArrayVec), + Streaming(SeaHasher), +} + +impl Default for BufferedFlowHasher { + fn default() -> Self { + Self::Buffered(ArrayVec::new()) + } +} + +impl Hasher for BufferedFlowHasher { + #[inline] + fn finish(&self) -> u64 { + match self { + Self::Buffered(bytes) => seahash::hash(bytes.as_slice()), + Self::Streaming(hasher) => hasher.finish(), + } + } + + #[inline] + fn write(&mut self, bytes: &[u8]) { + match self { + Self::Buffered(buffer) => { + if buffer.try_extend_from_slice(bytes).is_err() { + let mut hasher = SeaHasher::new(); + hasher.write(buffer.as_slice()); + hasher.write(bytes); + *self = Self::Streaming(hasher); + } + } + Self::Streaming(hasher) => hasher.write(bytes), + } + } + + // SeaHasher 4.1 encodes these integer writes as little endian. Its u128/i128 + // methods use Hasher's native-endian defaults, which this adapter also inherits. + #[inline] + fn write_u8(&mut self, value: u8) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_u16(&mut self, value: u16) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_u32(&mut self, value: u32) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_u64(&mut self, value: u64) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_usize(&mut self, value: usize) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_i8(&mut self, value: i8) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_i16(&mut self, value: i16) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_i32(&mut self, value: i32) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_i64(&mut self, value: i64) { + self.write(&value.to_le_bytes()); + } + + #[inline] + fn write_isize(&mut self, value: isize) { + self.write(&value.to_le_bytes()); + } +} + fn shard_map_index(flow_key: CanonicalFlowKey, shard_count: usize) -> usize { debug_assert!(shard_count > 0); if shard_count == 1 { return 0; } - let mut hasher = SeaHasher::new(); + let mut hasher = BufferedFlowHasher::default(); ( flow_key.client_ip, flow_key.client_port, @@ -1444,6 +1535,84 @@ mod tests { fs::remove_file(path).expect("test pcap removed"); } + #[test] + fn both_pipeline_models_preserve_matches_after_chained_retry_regressions() { + let path = temp_test_path("pipeline-chained-retry-regression", "pcap"); + let mut query_payload = encode_dns_header(0x1234, 0x0100, 1); + query_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, + ]); + query_payload.extend_from_slice(&1_u16.to_be_bytes()); + query_payload.extend_from_slice(&1_u16.to_be_bytes()); + let mut response_payload = query_payload.clone(); + response_payload[2..4].copy_from_slice(&0x8180_u16.to_be_bytes()); + let query = make_udp_dns_packet_with_payload( + [10, 0, 0, 1], + [8, 8, 8, 8], + 53_000, + 53, + &query_payload, + ); + let response = make_udp_dns_packet_with_payload( + [8, 8, 8, 8], + [10, 0, 0, 1], + 53, + 53_000, + &response_payload, + ); + let padding = make_udp_dns_packet([10, 0, 0, 1], [8, 8, 8, 8], 53_000, 123); + let mut packets = Vec::with_capacity(2 * PACKET_BATCH_SIZE + 2); + for timestamp in [3, 2] { + packets.push((timestamp, 0, query.as_slice())); + packets.extend(std::iter::repeat_n( + (timestamp, 0, padding.as_slice()), + PACKET_BATCH_SIZE - 1, + )); + } + packets.push((1, 0, query.as_slice())); + packets.push((3, 100_000, response.as_slice())); + fs::write(&path, classic_pcap_bytes(&packets)).expect("test pcap written"); + + for available_cpus in [1, 5] { + let mut parser = + PacketParser::new(&InputSource::File(path.clone()), false).expect("parser opens"); + let (output_tx, output_rx) = crossbeam::channel::unbounded(); + let counters = DnsProcessor::dns_processing_loop( + Arc::new(DnsProcessor::new(None).expect("processor initializes")), + &mut parser, + &Arc::new(AtomicUsize::new(0)), + &output_tx, + ExecutionBudget::from_available_cpus(available_cpus), + AffinityPlan::disabled(), + true, + Arc::new(AtomicBool::new(false)), + Arc::new(AtomicBool::new(false)), + ) + .expect("pipeline completes"); + assert_eq!(counters.total_packets_processed, 2 * PACKET_BATCH_SIZE + 2); + assert_eq!(counters.dns_query_count, 3); + assert_eq!(counters.duplicated_query_count, 1); + assert_eq!(counters.dns_response_count, 1); + assert_eq!(counters.matched_query_response_count, 1); + assert_eq!(counters.timeout_query_count, 1); + drop(output_tx); + let rows = output_rx + .into_iter() + .flat_map(|message| match message { + OutputMessage::Records(records) => records, + other => panic!("unexpected output message: {other:?}"), + }) + .map(|record| (record.request_timestamp, record.response_timestamp)) + .collect::>(); + assert_eq!( + rows, + vec![(3_000_000, Some(3_100_000)), (1_000_000, None)], + "available_cpus={available_cpus}" + ); + } + fs::remove_file(path).expect("test pcap removed"); + } + #[test] fn routed_worker_batches_use_global_batch_max_timestamp() { let path = temp_test_path("pipeline-routed-batch-watermark", "pcap"); @@ -1803,3 +1972,141 @@ mod tests { ); } } + +#[cfg(test)] +mod buffered_flow_hash_tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + + fn assert_flow_hash_matches_reference(flow: CanonicalFlowKey) { + let identity = (flow.client_ip, flow.client_port, flow.resolver_ip); + let mut reference = SeaHasher::new(); + identity.hash(&mut reference); + let expected = reference.finish(); + let mut buffered = BufferedFlowHasher::default(); + identity.hash(&mut buffered); + assert_eq!(buffered.finish(), expected, "{flow:?}"); + for shard_count in [1, 2, 3, 4, 7, 16, 64, 257, 1024] { + assert_eq!( + shard_map_index(flow, shard_count), + (expected as usize) % shard_count, + "{flow:?}, shard_count={shard_count}" + ); + } + } + + #[test] + fn buffered_flow_hash_preserves_address_families_ports_and_shards() { + let addresses = [ + IpAddr::V4(Ipv4Addr::UNSPECIFIED), + IpAddr::V4(Ipv4Addr::LOCALHOST), + IpAddr::V4(Ipv4Addr::BROADCAST), + IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), + IpAddr::V6(Ipv6Addr::UNSPECIFIED), + IpAddr::V6(Ipv6Addr::LOCALHOST), + IpAddr::V6(Ipv6Addr::from(u128::MAX)), + IpAddr::V6(Ipv6Addr::from( + 0x2001_0db8_0001_0203_0405_0607_0809_0a0b_u128, + )), + IpAddr::V6(Ipv4Addr::new(1, 2, 3, 4).to_ipv6_mapped()), + ]; + for client_ip in addresses { + for resolver_ip in addresses { + for client_port in [0, 1, 53, 1023, 1024, 32768, 65535] { + assert_flow_hash_matches_reference(CanonicalFlowKey { + client_ip, + client_port, + resolver_ip, + }); + } + } + } + } + + #[test] + fn buffered_flow_hash_matches_many_deterministic_flows() { + fn next(state: &mut u64) -> u64 { + *state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + *state + } + let mut state = 0x0123_4567_89ab_cdef; + for index in 0..4096 { + let client_v4 = Ipv4Addr::from(next(&mut state) as u32); + let client_v6 = + Ipv6Addr::from((u128::from(next(&mut state)) << 64) | u128::from(next(&mut state))); + let resolver_v4 = Ipv4Addr::from(next(&mut state) as u32); + let resolver_v6 = + Ipv6Addr::from((u128::from(next(&mut state)) << 64) | u128::from(next(&mut state))); + assert_flow_hash_matches_reference(CanonicalFlowKey { + client_ip: if index & 1 == 0 { + client_v4.into() + } else { + client_v6.into() + }, + client_port: next(&mut state) as u16, + resolver_ip: if index & 2 == 0 { + resolver_v4.into() + } else { + resolver_v6.into() + }, + }); + } + } + + #[test] + fn buffered_flow_hash_preserves_integer_write_encodings() { + macro_rules! check { + ($method:ident, $value:expr) => { + for prefix_len in [0, 63, 65] { + let prefix = [0x37; 65]; + let mut reference = SeaHasher::new(); + let mut buffered = BufferedFlowHasher::default(); + reference.write(&prefix[..prefix_len]); + buffered.write(&prefix[..prefix_len]); + reference.$method($value); + buffered.$method($value); + assert_eq!(buffered.finish(), reference.finish(), stringify!($method)); + } + }; + } + check!(write_u8, 0xef); + check!(write_u16, 0x1234); + check!(write_u32, 0x1234_5678); + check!(write_u64, 0x1234_5678_9abc_def0); + check!(write_usize, 0x1234_5678); + check!(write_i8, -17); + check!(write_i16, -0x1234); + check!(write_i32, -0x1234_5678); + check!(write_i64, -0x1234_5678_9abc_def0); + check!(write_isize, -0x1234_5678); + check!(write_u128, 0x0123_4567_89ab_cdef_fedc_ba98_7654_3210); + check!(write_i128, -0x0123_4567_89ab_cdef_fedc_ba98_7654_3210); + } + + #[test] + fn buffered_flow_hash_preserves_streaming_overflow_and_finish_semantics() { + let bytes: Vec = (0..257).map(|index| (index * 73 + 19) as u8).collect(); + for length in [ + 0, 1, 7, 8, 9, 31, 32, 33, 63, 64, 65, 66, 127, 128, 129, 256, 257, + ] { + for chunk_size in [1, 2, 3, 7, 8, 15, 16, 31, 63, 64, 65, 97, 257] { + let mut reference = SeaHasher::new(); + let mut buffered = BufferedFlowHasher::default(); + for chunk in bytes[..length].chunks(chunk_size) { + reference.write(chunk); + buffered.write(chunk); + assert_eq!(buffered.finish(), reference.finish()); + // finish is nondestructive, and an empty write cannot alter the stream. + buffered.write(&[]); + assert_eq!(buffered.finish(), reference.finish()); + } + assert_eq!(buffered.finish(), reference.finish()); + reference.write(&[1, 2, 3, 4, 5, 6, 7, 8, 9]); + buffered.write(&[1, 2, 3, 4, 5, 6, 7, 8, 9]); + assert_eq!(buffered.finish(), reference.finish()); + } + } + } +} diff --git a/src/dns_processor/tests.rs b/src/dns_processor/tests.rs index 19f175a..6f2f8f6 100644 --- a/src/dns_processor/tests.rs +++ b/src/dns_processor/tests.rs @@ -392,6 +392,8 @@ fn make_query_with_timestamp( packet_ordinal, record_ordinal, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, } } @@ -415,6 +417,8 @@ fn make_response_with_timestamp( record_ordinal, response_code: HickoryResponseCode::NoError.into(), query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, } } @@ -430,6 +434,8 @@ fn insert_query( query.src_port, query.query_type, query.resolver_ip, + query.query_class, + query.opcode, ); let key = super::types::TimelineKey::new( query.timestamp_micros, @@ -451,6 +457,8 @@ fn insert_response( response.dst_port, response.query_type, response.resolver_ip, + response.query_class, + response.opcode, ); let key = super::types::TimelineKey::new( response.timestamp_micros, @@ -481,6 +489,8 @@ fn make_query_record_with_timestamp( is_query: true, name: test_name(), query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, response_code: HickoryResponseCode::ServFail.into(), } } @@ -502,6 +512,8 @@ fn make_response_record_with_timestamp( is_query: false, name: test_name(), query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, response_code: HickoryResponseCode::NoError.into(), } } @@ -884,6 +896,70 @@ fn deduplicates_later_pending_queries_inside_timeout_window() { assert_eq!(pending_response_count(&state), 0); } +#[test] +fn query_class_and_opcode_keep_pending_transactions_independent() { + for (query_class, opcode) in [(3, 0), (1, 4)] { + let processor = test_processor(); + let mut state = test_shard_state(); + let mut other_query = make_query_record_with_timestamp(1_100, 2, 0); + other_query.query_class = query_class; + other_query.opcode = opcode; + let mut other_response = make_response_record_with_timestamp(1_200, 3, 0); + other_response.query_class = query_class; + other_response.opcode = opcode; + + let result = processor.process_shard_records_with_batch_watermark( + vec![ + make_query_record_with_timestamp(1_000, 1, 0), + other_query, + other_response, + make_response_record_with_timestamp(1_300, 4, 0), + ], + &mut state, + None, + ); + assert_eq!(result.duplicated_query_count, 0); + assert_eq!(result.matched_query_response_count, 2); + assert_eq!(result.output_records[0].request_timestamp, 1_100); + assert_eq!(result.output_records[0].response_timestamp, Some(1_200)); + assert_eq!(result.output_records[1].request_timestamp, 1_000); + assert_eq!(result.output_records[1].response_timestamp, Some(1_300)); + assert_eq!(pending_query_count(&state), 0); + assert_eq!(pending_response_count(&state), 0); + } +} + +#[test] +fn mismatched_query_class_or_opcode_never_pairs_in_either_arrival_order() { + for (query_class, opcode) in [(3, 0), (1, 4)] { + for response_first in [false, true] { + let processor = test_processor(); + let mut state = test_shard_state(); + let query = make_query_record_with_timestamp(1_000, 1, 0); + let mut response = make_response_record_with_timestamp(1_100, 2, 0); + response.query_class = query_class; + response.opcode = opcode; + let records = if response_first { + [response, query] + } else { + [query, response] + }; + for record in records { + let batch = processor.process_shard_records_with_batch_watermark( + vec![record], + &mut state, + None, + ); + assert_eq!(batch.matched_query_response_count, 0); + } + let finalization = processor.finalize_shard(&mut state); + assert_eq!(finalization.matched_query_response_count, 0); + assert_eq!(finalization.timeout_query_count, 1); + assert_eq!(pending_response_count(&state), 1); + } + } +} + #[test] fn timestamp_regression_keeps_earliest_pending_query_canonical() { let processor = test_processor(); @@ -914,6 +990,200 @@ fn timestamp_regression_keeps_earliest_pending_query_canonical() { assert_eq!(pending_query_count(&state), 0); } +#[test] +fn chained_timestamp_regressions_restore_queries_outside_the_canonical_window() { + let processor = test_processor(); + let mut state = test_shard_state(); + let mut duplicates = 0; + for (ordinal, timestamp) in [3_000_000, 2_000_000, 1_000_000].into_iter().enumerate() { + let batch = processor.process_shard_records_with_batch_watermark( + vec![make_query_record_with_timestamp( + timestamp, + ordinal as u64, + 0, + )], + &mut state, + None, + ); + duplicates += batch.duplicated_query_count; + assert!(batch.output_records.is_empty()); + } + assert_eq!(duplicates, 1); + assert_eq!(pending_query_count(&state), 2); + + let response = processor.process_shard_records_with_batch_watermark( + vec![make_response_record_with_timestamp(3_100_000, 3, 0)], + &mut state, + None, + ); + assert_eq!(response.matched_query_response_count, 1); + assert_eq!(response.output_records[0].request_timestamp, 3_000_000); + assert_eq!( + response.output_records[0].response_timestamp, + Some(3_100_000) + ); + let finalization = processor.finalize_shard(&mut state); + assert_eq!(finalization.timeout_query_count, 1); + assert_eq!(finalization.output_records[0].request_timestamp, 1_000_000); + assert_eq!(pending_query_count(&state), 0); + assert_eq!(pending_response_count(&state), 0); +} + +#[test] +fn earlier_attempt_repartitions_multiple_pending_retry_groups() { + let processor = test_processor(); + let mut state = test_shard_state(); + let mut duplicates = 0; + for (ordinal, timestamp) in [ + 3_000_000, 4_000_000, 4_300_000, 5_300_000, 5_600_000, 6_600_000, 2_000_000, + ] + .into_iter() + .enumerate() + { + let batch = processor.process_shard_records_with_batch_watermark( + vec![make_query_record_with_timestamp( + timestamp, + ordinal as u64, + 0, + )], + &mut state, + None, + ); + duplicates += batch.duplicated_query_count; + } + let finalization = processor.finalize_shard(&mut state); + let timestamps = finalization + .output_records + .into_batch_iter() + .flatten() + .map(|record| record.request_timestamp) + .collect::>(); + assert_eq!(timestamps, vec![2_000_000, 4_000_000, 5_300_000, 6_600_000]); + assert_eq!(duplicates, 3); + assert_eq!(finalization.timeout_query_count, 4); +} + +#[test] +fn repartition_matches_buffered_response_and_preserves_restored_query() { + let processor = test_processor(); + let mut state = test_shard_state(); + let first = processor.process_shard_records_with_batch_watermark( + vec![ + make_query_record_with_timestamp(3_000_000, 0, 0), + make_query_record_with_timestamp(4_000_000, 1, 0), + ], + &mut state, + None, + ); + assert_eq!(first.duplicated_query_count, 1); + let buffered = processor.process_shard_records_with_batch_watermark( + vec![make_response_record_with_timestamp(2_500_000, 2, 0)], + &mut state, + None, + ); + assert_eq!(buffered.matched_query_response_count, 0); + let regressed = processor.process_shard_records_with_batch_watermark( + vec![make_query_record_with_timestamp(2_000_000, 3, 0)], + &mut state, + None, + ); + assert_eq!(regressed.duplicated_query_count, 0); + assert_eq!(regressed.matched_query_response_count, 1); + assert_eq!(regressed.output_records[0].request_timestamp, 2_000_000); + assert_eq!( + regressed.output_records[0].response_timestamp, + Some(2_500_000) + ); + assert_eq!(pending_query_count(&state), 1); + let restored = processor.process_shard_records_with_batch_watermark( + vec![make_response_record_with_timestamp(4_100_000, 4, 0)], + &mut state, + None, + ); + assert_eq!(restored.matched_query_response_count, 1); + assert_eq!(restored.output_records[0].request_timestamp, 4_000_000); + assert_eq!(processor.finalize_shard(&mut state).timeout_query_count, 0); +} + +#[test] +fn pending_retry_groups_are_independent_of_query_arrival_order() { + fn check_permutations(timestamps: &mut [i64], next: usize) { + if next < timestamps.len() { + for index in next..timestamps.len() { + timestamps.swap(next, index); + check_permutations(timestamps, next + 1); + timestamps.swap(next, index); + } + return; + } + let processor = test_processor(); + let mut state = test_shard_state(); + let mut duplicates = 0; + for (ordinal, timestamp) in timestamps.iter().copied().enumerate() { + let batch = processor.process_shard_records_with_batch_watermark( + vec![make_query_record_with_timestamp( + timestamp, + ordinal as u64, + 0, + )], + &mut state, + None, + ); + duplicates += batch.duplicated_query_count; + } + let finalization = processor.finalize_shard(&mut state); + let actual = finalization + .output_records + .into_batch_iter() + .flatten() + .map(|record| record.request_timestamp) + .collect::>(); + assert_eq!( + actual, + vec![1_000_000, 3_000_000, 5_000_000], + "{timestamps:?}" + ); + assert_eq!(duplicates, 3, "{timestamps:?}"); + } + // Covers equal timestamps, exact timeout boundaries, and changes to multiple groups. + check_permutations( + &mut [ + 1_000_000, 2_200_000, 3_000_000, 3_000_000, 4_200_000, 5_000_000, + ], + 0, + ); +} + +#[test] +fn monotonic_queries_do_not_retain_retry_timestamps() { + for monotonic_capture in [false, true] { + let processor = + DnsProcessor::new_with_runtime_options(None, false, 1_200_000, monotonic_capture) + .expect("processor initializes"); + let mut state = test_shard_state(); + let result = processor.process_shard_records_with_batch_watermark( + vec![ + make_query_record_with_timestamp(1_000_000, 0, 0), + make_query_record_with_timestamp(1_100_000, 1, 0), + ], + &mut state, + None, + ); + assert_eq!(result.duplicated_query_count, 1); + let (_, payload) = state + .query_map + .values() + .next() + .expect("pending query") + .first_entry_in_range(0, i64::MAX) + .expect("pending canonical attempt"); + assert_eq!( + payload.retained_retry_count(), + usize::from(!monotonic_capture) + ); + } +} + #[test] fn duplicate_query_identity_requires_same_source_port() { let processor = test_processor(); @@ -1222,6 +1492,8 @@ fn finalization_preserves_full_key_order_across_identity_and_timestamp() { packet_ordinal: 3, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); insert_query( @@ -1237,6 +1509,8 @@ fn finalization_preserves_full_key_order_across_identity_and_timestamp() { packet_ordinal: 2, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); insert_query( @@ -1252,6 +1526,8 @@ fn finalization_preserves_full_key_order_across_identity_and_timestamp() { packet_ordinal: 1, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); @@ -1291,6 +1567,8 @@ fn timeline_overflow_preserves_sorted_finalization_order() { packet_ordinal: 4, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); insert_query( @@ -1306,6 +1584,8 @@ fn timeline_overflow_preserves_sorted_finalization_order() { packet_ordinal: 2, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); insert_query( @@ -1321,6 +1601,8 @@ fn timeline_overflow_preserves_sorted_finalization_order() { packet_ordinal: 5, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); insert_query( @@ -1336,6 +1618,8 @@ fn timeline_overflow_preserves_sorted_finalization_order() { packet_ordinal: 3, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); insert_query( @@ -1351,6 +1635,8 @@ fn timeline_overflow_preserves_sorted_finalization_order() { packet_ordinal: 1, record_ordinal: 0, query_type: HickoryRecordType::A, + query_class: 1, + opcode: 0, }, ); diff --git a/src/dns_processor/types.rs b/src/dns_processor/types.rs index 2d9b700..7d3c490 100644 --- a/src/dns_processor/types.rs +++ b/src/dns_processor/types.rs @@ -26,8 +26,17 @@ const INLINE_TIMELINE_CAPACITY: usize = 1; // Edition intentionally does not canonicalize names here: byte-preserving matching aligns better // with the real behavior we target on offline caching-resolver workloads. As a result, a // query/response pair that differs only by case may fail to match even on otherwise valid DNS -// traffic. Tuple order is id, name, client IP, client port, query type, resolver IP. -pub(super) type MatcherIdentityKey = (u16, DnsNameBuf, IpAddr, u16, HickoryRecordType, IpAddr); +// traffic. Tuple order is id, name, client IP, client port, query type, resolver IP, query class, opcode. +pub(super) type MatcherIdentityKey = ( + u16, + DnsNameBuf, + IpAddr, + u16, + HickoryRecordType, + IpAddr, + u16, + u8, +); pub(super) type QueryIdentityKey = MatcherIdentityKey; pub(super) type ResponseIdentityKey = MatcherIdentityKey; @@ -56,8 +65,46 @@ impl TimelineKey { } } -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub(super) struct QueryEventPayload; +#[derive(Debug, Default)] +pub(super) struct QueryEventPayload { + // Only non-monotonic captures need to retain pending retries. Keeping the optional + // allocation in the canonical query preserves a single owner for every attempt. + retries: Option>, +} + +#[derive(Debug)] +struct QueryRetries { + // The common first retry needs only this box; additional retries allocate a vector. + first: TimelineKey, + later: Vec, +} + +impl QueryEventPayload { + pub(super) fn retain_retry(&mut self, key: TimelineKey) { + if let Some(retries) = self.retries.as_mut() { + retries.later.push(key); + } else { + self.retries = Some(Box::new(QueryRetries { + first: key, + later: Vec::new(), + })); + } + } + + pub(super) fn append_retries_to(self, attempts: &mut Vec) { + if let Some(retries) = self.retries { + attempts.push(retries.first); + attempts.extend(retries.later); + } + } + + #[cfg(test)] + pub(super) fn retained_retry_count(&self) -> usize { + self.retries + .as_ref() + .map_or(0, |retries| 1 + retries.later.len()) + } +} #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub(super) struct ResponseEventPayload { @@ -151,6 +198,18 @@ impl Timeline { } } + pub(super) fn get_mut(&mut self, key: TimelineKey) -> Option<&mut Record> { + match self { + Self::Inline(entries) => { + let index = entries + .binary_search_by_key(&key, |(entry_key, _)| *entry_key) + .ok()?; + Some(&mut entries[index].1) + } + Self::Tree(tree) => tree.get_mut(&key), + } + } + pub(super) fn remove(&mut self, key: TimelineKey) -> Option { match self { Self::Inline(entries) => { @@ -361,6 +420,8 @@ pub(super) struct ProcessedDnsRecord { pub(super) is_query: bool, pub(super) name: DnsNameBuf, pub(super) query_type: HickoryRecordType, + pub(super) query_class: u16, + pub(super) opcode: u8, pub(super) response_code: ProtoResponseCode, } @@ -376,6 +437,8 @@ pub(super) struct DnsQuery { pub(super) packet_ordinal: u64, pub(super) record_ordinal: u32, pub(super) query_type: HickoryRecordType, + pub(super) query_class: u16, + pub(super) opcode: u8, } #[cfg(test)] @@ -391,6 +454,8 @@ pub(super) struct DnsResponse { pub(super) record_ordinal: u32, pub(super) response_code: ProtoResponseCode, pub(super) query_type: HickoryRecordType, + pub(super) query_class: u16, + pub(super) opcode: u8, } #[derive(Default)] diff --git a/src/packet_parser.rs b/src/packet_parser.rs index 0e28f17..7ad6c15 100644 --- a/src/packet_parser.rs +++ b/src/packet_parser.rs @@ -7,16 +7,16 @@ use crate::config::{InputSource, PACKET_BATCH_SIZE}; use anyhow::{Context, Result}; +use byteorder_slice::{BigEndian, LittleEndian}; use pcap::{Capture, Error as LibpcapError, Linktype, Offline}; -use pcap_file::DataLink; use pcap_file::pcap::PcapReader; -use pcap_file::pcapng::PcapNgReader; use pcap_file::pcapng::{ Block as PcapNgBlock, - blocks::{interface_description::InterfaceDescriptionBlock, packet::PacketBlock}, + blocks::interface_description::{InterfaceDescriptionBlock, InterfaceDescriptionOption}, }; +use pcap_file::{DataLink, Endianness}; use std::fs::File; -use std::io::{BufReader, Cursor, ErrorKind, Read}; +use std::io::{BufRead, BufReader, Cursor, ErrorKind, Read}; use std::ops::Deref; use std::path::Path; use std::time::Duration; @@ -85,7 +85,7 @@ type BufferedCaptureInput = BufReader; enum PacketBackend { Classic(PcapReader), - PcapNg(PcapNgReader), + PcapNg(PcapNgStreamReader), Libpcap(Capture), } @@ -141,10 +141,11 @@ impl PacketBackend { Ok(Self::Classic(reader)) } StreamFormat::PcapNg => { - let reader = PcapNgReader::new(BufReader::new(CaptureInputReader::Stream(reader))) - .with_context(|| { - format!("Unable to parse pcapng section header from '{source_name}'") - })?; + let reader = + PcapNgStreamReader::new(BufReader::new(CaptureInputReader::Stream(reader))) + .with_context(|| { + format!("Unable to parse pcapng section header from '{source_name}'") + })?; Ok(Self::PcapNg(reader)) } StreamFormat::Unknown => anyhow::bail!( @@ -176,20 +177,7 @@ impl PacketBackend { Err(LibpcapError::NoMorePackets) => Ok(None), Err(err) => Err(err.into()), }, - PacketBackend::PcapNg(reader) => loop { - let next_block = reader - .next_block() - .map(|result| result.map(|block| block.into_owned())); - match next_block { - Some(Ok(block)) => { - if let Some(packet) = pcapng_block_to_packet_data(reader, block)? { - break Ok(Some(packet)); - } - } - Some(Err(err)) => break Err(err.into()), - None => break Ok(None), - } - }, + PacketBackend::PcapNg(reader) => reader.next_packet_data(), } } } @@ -198,7 +186,7 @@ impl PacketParser { /// Creates a new `PacketParser` instance by opening the specified capture source. /// /// Classic pcap files use a pure-Rust streaming reader. Other formats fall back to libpcap to - /// preserve existing compatibility assumptions. Stdin uses libpcap's offline stream reader. + /// preserve existing compatibility assumptions. Stdin uses parser-owned streaming readers. pub fn new(input_source: &InputSource, enforce_monotonic_timestamps: bool) -> Result { Ok(Self { backend: PacketBackend::from_input_source(input_source)?, @@ -428,59 +416,200 @@ fn ensure_libpcap_ethernet_linktype( Ok(()) } -fn pcapng_block_to_packet_data( - reader: &PcapNgReader, - block: PcapNgBlock<'_>, -) -> Result> { - match block { - PcapNgBlock::EnhancedPacket(packet) => { - pcapng_packet_interface(reader, packet.interface_id)?; - Ok(Some(PacketData { - data: PacketPayload::owned(packet.data.into_owned().into_boxed_slice()), - timestamp_micros: duration_to_micros(packet.timestamp), - packet_ordinal: 0, - })) +/// Owns stdin PCAPNG framing and its sole section-local interface table. The dependency's +/// stateful reader converts timestamps incorrectly; even `next_raw_block` rejects valid +/// resolutions while updating interfaces. Only its stateless block validation is used here. +struct PcapNgStreamReader { + input: BufferedCaptureInput, + endianness: Endianness, + interfaces: Vec, + block_bytes: Vec, +} + +impl PcapNgStreamReader { + fn new(input: BufferedCaptureInput) -> Result { + let mut reader = Self { + input, + endianness: Endianness::Big, + interfaces: Vec::new(), + block_bytes: Vec::new(), + }; + anyhow::ensure!(reader.read_block_bytes()?, "Missing pcapng section header"); + anyhow::ensure!( + matches!(reader.decode_block()?, PcapNgBlock::SectionHeader(_)), + "Missing pcapng section header" + ); + Ok(reader) + } + + fn read_block_bytes(&mut self) -> Result { + if self.input.fill_buf()?.is_empty() { + return Ok(false); + } + + // Read only the fixed prefix before trusting the declared block length. The body + // grows with bytes actually received, so a huge length in a truncated stream cannot + // force an equally huge allocation. Memory remains bounded by the largest read block. + let mut prefix = [0_u8; 12]; + self.input.read_exact(&mut prefix)?; + if is_pcapng_magic(prefix[..4].try_into().expect("four-byte block type")) { + self.endianness = match &prefix[8..12] { + [0x1a, 0x2b, 0x3c, 0x4d] => Endianness::Big, + [0x4d, 0x3c, 0x2b, 0x1a] => Endianness::Little, + _ => anyhow::bail!("Invalid pcapng section byte-order magic"), + }; + } + let block_len = self.u32(&prefix[4..8]); + anyhow::ensure!( + block_len >= 12 && block_len.is_multiple_of(4), + "Invalid pcapng block length {block_len}: expected a multiple of four, at least 12" + ); + self.block_bytes.clear(); + self.block_bytes.extend_from_slice(&prefix); + self.input + .by_ref() + .take(u64::from(block_len - 12)) + .read_to_end(&mut self.block_bytes)?; + anyhow::ensure!( + self.block_bytes.len() == block_len as usize, + "Truncated pcapng block: expected {block_len} bytes, read {}", + self.block_bytes.len() + ); + Ok(true) + } + + fn decode_block(&self) -> Result> { + // Retain upstream validation of the trailing block length, body bounds, options, + // and all known non-packet blocks, without invoking timestamp-resolution conversion. + let (_, block) = match self.endianness { + Endianness::Big => PcapNgBlock::from_slice::(&self.block_bytes)?, + Endianness::Little => PcapNgBlock::from_slice::(&self.block_bytes)?, + }; + Ok(block) + } + + fn u32(&self, bytes: &[u8]) -> u32 { + let bytes = bytes.try_into().expect("validated four-byte pcapng field"); + match self.endianness { + Endianness::Big => u32::from_be_bytes(bytes), + Endianness::Little => u32::from_le_bytes(bytes), } - PcapNgBlock::Packet(packet) => Ok(Some(packet_block_to_packet_data(reader, packet)?)), - PcapNgBlock::SimplePacket(_) => anyhow::bail!( - "Unsupported pcapng Simple Packet Block: packet timestamps are required for offline DNS matching." - ), - _ => Ok(None), } -} -fn pcapng_packet_interface( - reader: &PcapNgReader, - interface_id: u32, -) -> Result<&InterfaceDescriptionBlock<'static>> { - let interface = reader - .interfaces() - .get(interface_id as usize) - .ok_or_else(|| { + fn next_packet_data(&mut self) -> Result> { + while self.read_block_bytes()? { + match self.decode_block()? { + PcapNgBlock::SectionHeader(_) => self.interfaces.clear(), + PcapNgBlock::InterfaceDescription(interface) => { + let interface = PcapNgInterface::new(&interface)?; + self.interfaces.push(interface); + } + PcapNgBlock::EnhancedPacket(packet) => { + return self + .packet_data(packet.interface_id, packet.original_len, &packet.data) + .map(Some); + } + PcapNgBlock::Packet(packet) => { + return self + .packet_data( + u32::from(packet.interface_id), + packet.original_len, + &packet.data, + ) + .map(Some); + } + PcapNgBlock::SimplePacket(_) => anyhow::bail!( + "Unsupported pcapng Simple Packet Block: packet timestamps are required for offline DNS matching." + ), + _ => {} + } + } + Ok(None) + } + + fn packet_data(&self, interface_id: u32, original_len: u32, data: &[u8]) -> Result { + let interface = self.interfaces.get(interface_id as usize).ok_or_else(|| { anyhow::anyhow!("pcapng packet references unknown interface {interface_id}") })?; - ensure_ethernet_datalink( - interface.linktype, - format_args!("pcapng interface {interface_id}"), - )?; + ensure_ethernet_datalink( + interface.linktype, + format_args!("pcapng interface {interface_id}"), + )?; + anyhow::ensure!( + data.len() <= original_len as usize, + "pcapng captured packet length exceeds original packet length" + ); + anyhow::ensure!( + interface.snaplen == 0 || data.len() <= interface.snaplen as usize, + "pcapng captured packet length exceeds interface snaplen" + ); - Ok(interface) + // Both EPB and legacy PB store two section-endian u32 words, high then low. + // The stateless decoder has already validated this fixed header and the payload. + // Ignore its timestamp representation: legacy PB incorrectly decodes a single u64. + let ticks = (u64::from(self.u32(&self.block_bytes[12..16])) << 32) + | u64::from(self.u32(&self.block_bytes[16..20])); + Ok(PacketData { + data: PacketPayload::owned(Box::from(data)), + timestamp_micros: interface.timestamp_micros(ticks), + packet_ordinal: 0, + }) + } } -fn packet_block_to_packet_data( - reader: &PcapNgReader, - packet: PacketBlock<'_>, -) -> Result { - let interface = pcapng_packet_interface(reader, u32::from(packet.interface_id))?; - let nanos_per_unit = u128::from(interface.ts_resolution()?.to_nano_secs()); - let timestamp_nanos = u128::from(packet.timestamp).saturating_mul(nanos_per_unit); - let timestamp_micros = timestamp_nanos.saturating_div(1_000).min(i64::MAX as u128) as i64; - - Ok(PacketData { - data: PacketPayload::owned(packet.data.into_owned().into_boxed_slice()), - timestamp_micros, - packet_ordinal: 0, - }) +struct PcapNgInterface { + linktype: DataLink, + snaplen: u32, + // None means 10^exponent exceeds u128. Even u64::MAX ticks then round down to 0us. + units_per_second: Option, + offset_seconds: i64, +} + +impl PcapNgInterface { + fn new(interface: &InterfaceDescriptionBlock<'_>) -> Result { + let mut resolution = None; + let mut offset_seconds = None; + for option in &interface.options { + match option { + InterfaceDescriptionOption::IfTsResol(value) => { + anyhow::ensure!(resolution.is_none(), "Duplicate pcapng if_tsresol option"); + resolution = Some(*value); + } + InterfaceDescriptionOption::IfTsOffset(value) => { + anyhow::ensure!( + offset_seconds.is_none(), + "Duplicate pcapng if_tsoffset option" + ); + // The dependency exposes u64, but the option is a signed two's-complement i64. + offset_seconds = Some(*value as i64); + } + _ => {} + } + } + let resolution = resolution.unwrap_or(6); + let exponent = u32::from(resolution & 0x7f); + let units_per_second = if resolution & 0x80 == 0 { + 10_u128.checked_pow(exponent) + } else { + Some(1_u128 << exponent) + }; + Ok(Self { + linktype: interface.linktype, + snaplen: interface.snaplen, + units_per_second, + offset_seconds: offset_seconds.unwrap_or(0), + }) + } + + fn timestamp_micros(&self, ticks: u64) -> i64 { + // Scale the full counter before division: fractional nanosecond units, including + // every binary resolution, must not be rounded individually before multiplication. + let micros = self + .units_per_second + .map_or(0, |units| u128::from(ticks) * 1_000_000 / units); + let adjusted = micros as i128 + i128::from(self.offset_seconds) * 1_000_000; + adjusted.clamp(i128::from(i64::MIN), i128::from(i64::MAX)) as i64 + } } fn duration_to_micros(duration: Duration) -> i64 { @@ -568,24 +697,7 @@ mod tests { timestamp_micros: u64, payload: &[u8], ) -> Vec { - let mut writer = PcapNgWriter::new(Vec::new()).expect("pcapng writer initializes"); - for linktype in linktypes { - writer - .write_pcapng_block(InterfaceDescriptionBlock::new(*linktype, 0xFFFF)) - .expect("pcapng interface block writes"); - } - writer - .write_pcapng_block(PacketBlock { - interface_id, - drop_count: 0, - timestamp: timestamp_micros, - captured_len: payload.len() as u32, - original_len: payload.len() as u32, - data: Cow::Borrowed(payload), - options: Vec::new(), - }) - .expect("pcapng legacy packet block writes"); - writer.into_inner() + super::wire_tests::legacy_packet_bytes(linktypes, interface_id, timestamp_micros, payload) } #[test] @@ -959,3 +1071,7 @@ mod tests { ); } } + +#[cfg(test)] +#[path = "packet_parser_wire_tests.rs"] +mod wire_tests; diff --git a/src/packet_parser_wire_tests.rs b/src/packet_parser_wire_tests.rs new file mode 100644 index 0000000..e8bde61 --- /dev/null +++ b/src/packet_parser_wire_tests.rs @@ -0,0 +1,361 @@ +/* + * Nameto Oy © 2026. All rights reserved. + * + * This software is licensed under the GNU General Public License (GPL) version 3. + * Commercial licensing options: . + */ + +use super::*; + +// Independent wire fixtures: do not round-trip the pcap-file writer, which shares the +// timestamp decoding bugs these tests guard against. +#[derive(Clone, Copy)] +enum Order { + Little, + Big, +} + +impl Order { + fn u16(self, value: u16) -> [u8; 2] { + match self { + Self::Little => value.to_le_bytes(), + Self::Big => value.to_be_bytes(), + } + } + + fn u32(self, value: u32) -> [u8; 4] { + match self { + Self::Little => value.to_le_bytes(), + Self::Big => value.to_be_bytes(), + } + } + + fn u64(self, value: u64) -> [u8; 8] { + match self { + Self::Little => value.to_le_bytes(), + Self::Big => value.to_be_bytes(), + } + } +} + +struct WireCapture { + order: Order, + bytes: Vec, +} + +impl WireCapture { + fn new(order: Order) -> Self { + let mut capture = Self { + order, + bytes: Vec::new(), + }; + let mut section = order.u32(0x1a2b3c4d).to_vec(); + section.extend(order.u16(1)); + section.extend(order.u16(0)); + section.extend(order.u64(u64::MAX)); + capture.block(0x0a0d0d0a, §ion); + capture + } + + fn block(&mut self, kind: u32, body: &[u8]) { + let length = (body.len() + 12) as u32; + self.bytes.extend(self.order.u32(kind)); + self.bytes.extend(self.order.u32(length)); + self.bytes.extend(body); + self.bytes.extend(self.order.u32(length)); + } + + fn option(&self, code: u16, value: &[u8]) -> Vec { + let mut option = self.order.u16(code).to_vec(); + option.extend(self.order.u16(value.len() as u16)); + option.extend(value); + option.resize(option.len().next_multiple_of(4), 0); + option + } + + fn interface_options(&mut self, linktype: u16, snaplen: u32, options: &[u8]) { + let mut body = self.order.u16(linktype).to_vec(); + body.extend(self.order.u16(0)); + body.extend(self.order.u32(snaplen)); + body.extend(options); + self.block(1, &body); + } + + fn interface( + &mut self, + linktype: u16, + snaplen: u32, + resolution: Option, + offset: Option, + ) { + let mut options = Vec::new(); + if let Some(resolution) = resolution { + options.extend(self.option(9, &[resolution])); + } + if let Some(offset) = offset { + options.extend(self.option(14, &self.order.u64(offset as u64))); + } + if !options.is_empty() { + options.extend([0; 4]); + } + self.interface_options(linktype, snaplen, &options); + } + + fn packet(&mut self, kind: u32, interface_id: u32, ticks: u64, payload: &[u8]) { + let mut body = if kind == 2 { + let mut fields = self.order.u16(interface_id as u16).to_vec(); + fields.extend(self.order.u16(0xffff)); + fields + } else { + self.order.u32(interface_id).to_vec() + }; + body.extend(self.order.u32((ticks >> 32) as u32)); + body.extend(self.order.u32(ticks as u32)); + body.extend(self.order.u32(payload.len() as u32)); + body.extend(self.order.u32(payload.len() as u32)); + body.extend(payload); + body.resize(body.len().next_multiple_of(4), 0); + self.block(kind, &body); + } +} + +fn parser(bytes: Vec) -> PacketParser { + PacketParser { + backend: PacketBackend::from_stream(Box::new(Cursor::new(bytes)), "wire-fixture") + .expect("stream opens"), + enforce_monotonic_timestamps: false, + packet_ordinal: 0, + last_timestamp_micros: None, + first_non_monotonic_timestamp: None, + non_monotonic_timestamp_count: 0, + } +} + +pub(super) fn legacy_packet_bytes( + linktypes: &[DataLink], + interface_id: u16, + timestamp_micros: u64, + payload: &[u8], +) -> Vec { + let mut capture = WireCapture::new(Order::Little); + for linktype in linktypes { + capture.interface(u32::from(*linktype) as u16, 65535, None, None); + } + capture.packet(2, u32::from(interface_id), timestamp_micros, payload); + capture.bytes +} + +#[test] +fn reads_all_timestamp_resolution_boundaries_in_both_packet_formats_and_byte_orders() { + let cases = [ + (None, 1_234_567, 1_234_567), + (Some(0), 1, 1_000_000), + (Some(3), 12_345, 12_345_000), + (Some(6), 1_234_567, 1_234_567), + (Some(9), 1_234_567_890, 1_234_567), + (Some(9), u64::MAX, 18_446_744_073_709_551), + (Some(10), 10_000_000_000, 1_000_000), + (Some(24), u64::MAX, 18), + (Some(25), u64::MAX, 1), + (Some(26), u64::MAX, 0), + (Some(38), u64::MAX, 0), + (Some(39), u64::MAX, 0), + (Some(127), u64::MAX, 0), + (Some(0x80), 1, 1_000_000), + (Some(0x8a), 1024, 1_000_000), + (Some(0x8a), 2201, 2_149_414), + (Some(0x9e), 1 << 30, 1_000_000), + (Some(0x9f), 1 << 31, 1_000_000), + (Some(0xbf), 1 << 63, 1_000_000), + (Some(0xc0), u64::MAX, 999_999), + (Some(0xff), u64::MAX, 0), + ]; + for order in [Order::Little, Order::Big] { + for kind in [2, 6] { + for (resolution, ticks, expected_micros) in cases { + let mut capture = WireCapture::new(order); + capture.interface(1, 65535, resolution, None); + capture.packet(kind, 0, ticks, &[1, 2, 3]); + let mut parser = parser(capture.bytes); + let batch = parser.next_batch(8).unwrap().unwrap(); + assert_eq!(batch.len(), 1); + assert_eq!( + batch[0].timestamp_micros, expected_micros, + "resolution {resolution:?}" + ); + assert_eq!(batch[0].data.as_slice(), &[1, 2, 3]); + assert!(parser.next_batch(8).unwrap().is_none()); + } + } + } +} + +#[test] +fn applies_signed_offsets_per_interface_before_monotonic_checks() { + for order in [Order::Little, Order::Big] { + for kind in [2, 6] { + let mut capture = WireCapture::new(order); + capture.interface(1, 65535, None, Some(100)); + capture.interface(1, 65535, None, Some(101)); + capture.interface(1, 65535, None, Some(-1)); + capture.packet(kind, 0, 1_000_000, &[1]); + capture.packet(kind, 1, 100_000, &[2]); + capture.packet(kind, 2, 102_200_000, &[3]); + let mut parser = parser(capture.bytes); + parser.enforce_monotonic_timestamps = true; + let batch = parser.next_batch(8).unwrap().unwrap(); + assert_eq!( + batch.iter().map(|p| p.timestamp_micros).collect::>(), + [101_000_000, 101_100_000, 101_200_000] + ); + } + } +} + +#[test] +fn preserves_negative_timestamps_and_saturates_only_after_adding_offset() { + let cases = [ + (6, 100_000, -1, -900_000), + (0, 1, -1, 0), + (0, u64::MAX, i64::MIN, i64::MAX), + (0, u64::MAX, 0, i64::MAX), + (6, 0, i64::MIN, i64::MIN), + (6, 0, i64::MAX, i64::MAX), + // Conversion must not saturate the raw timestamp before applying its offset. + (0, 10_000_000_000_000, -10_000_000_000_000, 0), + ]; + for kind in [2, 6] { + for (resolution, ticks, offset, expected) in cases { + let mut capture = WireCapture::new(Order::Little); + capture.interface(1, 65535, Some(resolution), Some(offset)); + capture.packet(kind, 0, ticks, &[1]); + assert_eq!( + parser(capture.bytes).next_batch(8).unwrap().unwrap()[0].timestamp_micros, + expected + ); + } + } +} + +#[test] +fn new_sections_reset_interface_metadata_and_can_change_endianness() { + let mut first = WireCapture::new(Order::Little); + first.interface(1, 65535, Some(0x8a), Some(100)); + first.packet(2, 0, 1024, &[1]); + let mut second = WireCapture::new(Order::Big); + second.interface(1, 65535, None, None); + second.packet(6, 0, 102_000_000, &[2]); + first.bytes.extend(second.bytes); + let batch = parser(first.bytes).next_batch(8).unwrap().unwrap(); + assert_eq!(batch[0].timestamp_micros, 101_000_000); + assert_eq!(batch[1].timestamp_micros, 102_000_000); +} + +#[test] +fn rejects_unknown_interfaces_in_each_section_and_packet_format() { + for kind in [2, 6] { + let mut capture = WireCapture::new(Order::Little); + capture.interface(1, 65535, None, None); + let mut second = WireCapture::new(Order::Big); + second.packet(kind, 0, 1, &[1]); + capture.bytes.extend(second.bytes); + assert!( + parser(capture.bytes) + .next_batch(8) + .unwrap_err() + .to_string() + .contains("unknown interface 0") + ); + } +} + +#[test] +fn rejects_duplicate_timestamp_options_and_invalid_option_lengths() { + for (code, values) in [ + (9, vec![vec![6], vec![9]]), + (14, vec![vec![0; 8], vec![0; 8]]), + (9, vec![vec![6, 6]]), + (14, vec![vec![0; 4]]), + ] { + let mut capture = WireCapture::new(Order::Little); + let mut options = Vec::new(); + for value in values { + options.extend(capture.option(code, &value)); + } + options.extend([0; 4]); + capture.interface_options(1, 65535, &options); + assert!(parser(capture.bytes).next_batch(8).is_err()); + } +} + +#[test] +fn preserves_block_and_packet_length_validation() { + for kind in [2, 6] { + for invalid_field in [ + "trailer", + "captured_len", + "original_len", + "snaplen", + "truncated", + ] { + let mut capture = WireCapture::new(Order::Little); + capture.interface( + 1, + if invalid_field == "snaplen" { 1 } else { 65535 }, + None, + None, + ); + let packet_start = capture.bytes.len(); + capture.packet(kind, 0, 1, &[1, 2, 3, 4]); + match invalid_field { + "trailer" => { + let last = capture.bytes.len() - 4; + capture.bytes[last] = 0; + } + "captured_len" => capture.bytes[packet_start + 20..packet_start + 24] + .copy_from_slice(&100_u32.to_le_bytes()), + "original_len" => capture.bytes[packet_start + 24..packet_start + 28] + .copy_from_slice(&1_u32.to_le_bytes()), + "truncated" => { + capture.bytes.pop(); + } + _ => {} + } + assert!( + parser(capture.bytes).next_batch(8).is_err(), + "{invalid_field}" + ); + } + } +} + +#[test] +fn rejects_huge_truncated_blocks_without_reserving_the_declared_length() { + let mut capture = WireCapture::new(Order::Little); + capture.bytes.extend(6_u32.to_le_bytes()); + capture.bytes.extend(0xffff_fffc_u32.to_le_bytes()); + capture.bytes.extend([0; 4]); + let mut parser = parser(capture.bytes); + assert!( + parser + .next_batch(8) + .unwrap_err() + .to_string() + .contains("Truncated pcapng block") + ); + let PacketBackend::PcapNg(reader) = &parser.backend else { + panic!("pcapng backend") + }; + assert!(reader.block_bytes.capacity() < 1024 * 1024); +} + +#[test] +fn validates_options_even_on_ignored_known_blocks() { + let mut capture = WireCapture::new(Order::Little); + // Interface Statistics Block, followed by an incomplete eight-byte counter option. + let mut body = vec![0; 12]; + body.extend(capture.option(4, &[1, 2, 3, 4])); + body.extend([0; 4]); + capture.block(5, &body); + assert!(parser(capture.bytes).next_batch(8).is_err()); +} diff --git a/src/parquet_writer.rs b/src/parquet_writer.rs index c53812b..280846a 100644 --- a/src/parquet_writer.rs +++ b/src/parquet_writer.rs @@ -16,6 +16,7 @@ use parquet::file::writer::SerializedFileWriter; use parquet::schema::parser::parse_message_type; use parquet::schema::types::ColumnPath; use parquet::schema::types::TypePtr; +use std::borrow::Cow; use std::error::Error; use std::fmt::{self, Write as _}; use std::io::{self, Write}; @@ -138,6 +139,13 @@ fn max_formatted_ip_length(ip: &IpAddr) -> usize { } } +fn protocol_text_bytes(text: Cow<'static, str>) -> ByteArray { + ByteArray::from(match text { + Cow::Borrowed(value) => Bytes::from_static(value.as_bytes()), + Cow::Owned(value) => Bytes::from(value), + }) +} + fn write_column( row_group_writer: &mut parquet::file::writer::SerializedRowGroupWriter<'_, impl Write + Send>, values: &[Type::T], @@ -216,13 +224,9 @@ where source_ip_builder.push_display(record.source_ip); name_builder.push_bytes(record.name.as_bytes()); - query_types.push(ByteArray::from(Bytes::from_static( - record.query_type.as_str().as_bytes(), - ))); + query_types.push(protocol_text_bytes(record.query_type.as_str())); if let Some(response_code) = &record.response_code { - response_codes.push(ByteArray::from(Bytes::from_static( - response_code.as_str().as_bytes(), - ))); + response_codes.push(protocol_text_bytes(response_code.as_str())); response_code_definition_levels.push(1); } else { response_code_definition_levels.push(0); diff --git a/src/test_support.rs b/src/test_support.rs index bc78796..bc8233d 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -123,7 +123,11 @@ pub(crate) fn make_udp_dns_packet( src_port: u16, dst_port: u16, ) -> Vec { - make_udp_dns_packet_with_payload(src_ip, dst_ip, src_port, dst_port, &[0_u8; 12]) + let mut header = [0_u8; 12]; + if src_port == 53 { + header[2] = 0x80; + } + make_udp_dns_packet_with_payload(src_ip, dst_ip, src_port, dst_port, &header) } pub(crate) fn test_dns_record() -> DnsRecord { diff --git a/tests/end_to_end.rs b/tests/end_to_end.rs index ef2c763..c22ba14 100644 --- a/tests/end_to_end.rs +++ b/tests/end_to_end.rs @@ -666,3 +666,96 @@ fn closed_stderr_does_not_abort_file_output() { fs::remove_file(&input_path).expect("remove input pcap"); fs::remove_file(&output_path).expect("remove output csv"); } + +#[test] +fn unknown_protocol_values_round_trip_through_csv_and_parquet() { + use parquet::file::reader::{FileReader, SerializedFileReader}; + use parquet::record::RowAccessor; + + let input_path = temp_test_path("unknown-protocol-values", "pcap"); + let mut frames = Vec::new(); + for (id, query_type, response_code) in [(1_u16, 65400_u16, 64_u16), (2, 65401, 65)] { + let mut query = encode_dns_header(id, 0x0100, 1); + append_example_a_query(&mut query); + let type_offset = query.len() - 4; + query[type_offset..type_offset + 2].copy_from_slice(&query_type.to_be_bytes()); + let mut response = query.clone(); + response[2..4].copy_from_slice(&(0x8180 | (response_code & 0xf)).to_be_bytes()); + response[10..12].copy_from_slice(&1_u16.to_be_bytes()); + append_opt_record(&mut response, (response_code >> 4) as u8, 0); + frames.push(make_udp_dns_packet_with_payload( + [10, 0, 0, 1], + [8, 8, 8, 8], + 53000, + 53, + &query, + )); + frames.push(make_udp_dns_packet_with_payload( + [8, 8, 8, 8], + [10, 0, 0, 1], + 53, + 53000, + &response, + )); + } + let packets: Vec<_> = frames + .iter() + .enumerate() + .map(|(index, frame)| (1, index as u32 * 10_000, frame.as_slice())) + .collect(); + fs::write(&input_path, classic_pcap_bytes(&packets)).expect("writes capture"); + + for fast in [false, true] { + for format in ["csv", "parquet"] { + let output_path = temp_test_path("unknown-protocol-values", format); + let mut command = Command::new(dpp_binary()); + command.args(["-s", "-f", format]); + if fast { + command.arg("--dns-wire-fast-path"); + } + let output = command + .arg(&input_path) + .arg(&output_path) + .output() + .expect("runs dpp"); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let values: Vec<(String, String)> = if format == "csv" { + csv::Reader::from_path(&output_path) + .expect("opens csv") + .records() + .map(|row| { + let row = row.expect("reads row"); + (row[6].to_owned(), row[7].to_owned()) + }) + .collect() + } else { + SerializedFileReader::new(fs::File::open(&output_path).expect("opens parquet")) + .expect("reads parquet") + .get_row_iter(None) + .expect("reads rows") + .map(|row| { + let row = row.expect("reads row"); + ( + row.get_string(6).unwrap().clone(), + row.get_string(7).unwrap().clone(), + ) + }) + .collect() + }; + assert_eq!( + values, + vec![ + ("TYPE65400".to_owned(), "64".to_owned()), + ("TYPE65401".to_owned(), "65".to_owned()) + ], + "fast={fast}, format={format}" + ); + fs::remove_file(output_path).expect("removes output"); + } + } + fs::remove_file(input_path).expect("removes input"); +}