Validate packet burst bounds before writes

AGENTS.md

@@ -19,6 +19,7 @@ CMake options (full table in `docs/getting-started.md`):
 - `DAQIRI_BUILD_PYTHON` — builds `pybind11` bindings from `python/`.
 - `DAQIRI_BUILD_EXAMPLES` — builds the benchmark executables (default `ON`).
 - `DAQIRI_ENABLE_OTEL_METRICS` — enables OpenTelemetry metrics instrumentation (default `OFF`).
+- `DAQIRI_ENABLE_BURST_VALIDATION` — enables defensive burst bounds and backing-capacity validation in public packet accessors and setters (default `ON`; disable only for trusted-input performance testing).
 - `DAQIRI_REORDER_GPU_PROFILE` — enable CUDA event timing in the DPDK reorder kernels (off by default).
 - `DAQIRI_ENABLE_S3` — enable AWS SDK-backed asynchronous raw packet writes to S3 (off by default).
 - `DAQIRI_PREFER_SYSTEM_YAML_CPP` — prefer system `yaml-cpp` over the vendored `third_party/yaml-cpp` submodule (off by default; keep off when a conda/miniforge env is on `PATH`).
@@ -124,7 +125,7 @@ The web docs live in `docs/` and are built with [MkDocs Material](https://squidf
 
 **Keeping docs in sync with code:** before committing changes, scan for the recurring drift hotspots:
 - **Stream-type list** (`src/engines/*/`) — README Engines table, `docs/getting-started.md`, `docs/concepts.md` (Stream Types section + Support and testing admonition), `docs/api-reference/configuration.md`
-- **CMake options / `DAQIRI_ENGINE` default** (`src/CMakeLists.txt:137`) — README Quick Start, `docs/getting-started.md`, this file's Build & run section
+- **CMake options / `DAQIRI_ENGINE` default** (`src/CMakeLists.txt`) — README Quick Start, `docs/getting-started.md`, this file's Build & run section
 - **Benchmark binary or YAML names** (`examples/`) — the benchmark table above, `docs/benchmarks/raw_benchmarking.md`, the "Choosing an example config" decision tree in `docs/tutorials/configuration-walkthrough.md` (every YAML must have a leaf; CI's `scripts/check_doc_refs.py` enforces coverage), and per-platform performance docs (`docs/benchmarks/performance-*.md`)
 - **Public API include** (`#include <daqiri/daqiri.h>`; source files under `include/daqiri/`) — `docs/api-reference/index.md`, `docs/api-reference/cpp.md`, `docs/api-reference/python.md`; if the change adds or renames a user-facing concept, also `docs/concepts.md`
 - **Python bindings** (`python/daqiri_common_pybind.cpp`) — `docs/api-reference/python.md` (function reference tables, enums/classes tables, GIL Behavior section)

CMakeLists.txt

@@ -33,6 +33,8 @@ endif()
 option(DAQIRI_BUILD_PYTHON "Build Python bindings" OFF)
 option(DAQIRI_BUILD_EXAMPLES "Build standalone examples" ON)
 option(DAQIRI_ENABLE_OTEL_METRICS "Enable OpenTelemetry metrics instrumentation" OFF)
+option(DAQIRI_ENABLE_BURST_VALIDATION
+       "Enable defensive burst bounds and backing capacity validation" ON)
 set(DAQIRI_ENGINE "dpdk ibverbs" CACHE STRING "Optional engine implementations to build: dpdk, ibverbs (Linux sockets are always built in)")
 
 add_subdirectory(src)

docs/getting-started.md

@@ -195,6 +195,7 @@ Both methods use the same public C++ include:
 | `DAQIRI_ENGINE` | `"dpdk ibverbs"` | Space-separated list of optional engine implementations to compile in. Valid values: `dpdk` (Raw Ethernet) and `ibverbs`. `ibverbs` builds two libibverbs-based engines: RDMA/RoCE (for `stream_type: "socket"` with `roce://` endpoints) and a Mellanox/mlx5 Multi-Packet (striding) Receive Queue engine for `stream_type: "raw"` (opt in per stream with `engine: "ibverbs"`). Linux UDP/TCP sockets are always built in, so there is no `socket` value. |
 | `DAQIRI_BUILD_PYTHON` | `OFF` | Build pybind11 Python bindings. |
 | `DAQIRI_BUILD_EXAMPLES` | `ON` | Build benchmark executables. |
+| `DAQIRI_ENABLE_BURST_VALIDATION` | `ON` | Enable defensive burst bounds and backing-capacity validation in public packet accessors and setters. Disable only for trusted-input performance testing after benchmarking. |
 | `DAQIRI_ENABLE_GDS` | `OFF` | Enable cuFile-backed burst file writes from CUDA device memory. Host-memory writes use POSIX APIs without GDS. |
 | `DAQIRI_ENABLE_OTEL_METRICS` | `OFF` | Enable OpenTelemetry C++ metrics instrumentation. When enabled, OpenTelemetry C++ API package metadata must be available to CMake. |
 | `DAQIRI_ENABLE_S3` | `OFF` | Enable AWS SDK-backed asynchronous raw packet writes to S3. |

docs/tutorials/bare-metal-cmake-build.md

@@ -278,6 +278,7 @@ The override is honored because `src/CMakeLists.txt` only sets `CMAKE_CUDA_ARCHI
 
 - `-DBUILD_SHARED_LIBS=ON`: produces `libdaqiri.so` (recommended). With `OFF`, you get a static library.
 - `-DDAQIRI_BUILD_EXAMPLES=ON`: builds the `daqiri_bench_*` executables under `build/examples/`. Required for the smoke test in [Step 5.3](#53-smoke-test). On by default.
+- `-DDAQIRI_ENABLE_BURST_VALIDATION=ON`: enables defensive burst bounds and backing-capacity validation in packet accessors and setters. On by default; turn off only for trusted-input performance testing after benchmarking.
 - `-DDAQIRI_REORDER_GPU_PROFILE=ON`: instruments the CUDA reorder kernels with CUDA event timing. Off by default; turn on only when profiling.
 
 ## Step 5: Build, install, and verify

src/CMakeLists.txt

@@ -64,6 +64,8 @@ endif()
 option(DAQIRI_ENABLE_GDS "Enable cuFile-backed burst file writes from CUDA device memory" OFF)
 option(DAQIRI_ENABLE_OTEL_METRICS "Enable OpenTelemetry metrics instrumentation" OFF)
 option(DAQIRI_ENABLE_S3 "Enable AWS SDK-backed raw packet writes to S3" OFF)
+option(DAQIRI_ENABLE_BURST_VALIDATION
+       "Enable defensive burst bounds and backing capacity validation" ON)
 set(DOCA_PATH /opt/mellanox/doca)
 
 if(DAQIRI_ENABLE_OTEL_METRICS)
@@ -72,6 +74,11 @@ if(DAQIRI_ENABLE_OTEL_METRICS)
 else()
   add_compile_definitions(DAQIRI_ENABLE_OTEL_METRICS=0)
 endif()
+if(DAQIRI_ENABLE_BURST_VALIDATION)
+  add_compile_definitions(DAQIRI_ENABLE_BURST_VALIDATION=1)
+else()
+  add_compile_definitions(DAQIRI_ENABLE_BURST_VALIDATION=0)
+endif()
 
 if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/../third_party/spdlog/include/spdlog/spdlog.h")
   message(STATUS "Using vendored spdlog headers from third_party/spdlog")

src/burst_validation.h

@@ -0,0 +1,276 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <initializer_list>
+#include <limits>
+
+#include <daqiri/logging.hpp>
+#include <daqiri/types.h>
+
+#ifndef DAQIRI_ENABLE_BURST_VALIDATION
+#define DAQIRI_ENABLE_BURST_VALIDATION 1
+#endif
+
+namespace daqiri::burst_validation {
+
+struct BurstLimits {
+  size_t num_pkts = 0;
+  int num_segs = 0;
+};
+
+static inline constexpr size_t kUnknownCapacity = std::numeric_limits<size_t>::max();
+
+inline constexpr bool strict_enabled() {
+  return DAQIRI_ENABLE_BURST_VALIDATION != 0;
+}
+
+inline BurstLimits header_limits(const BurstParams* burst) {
+  if (burst == nullptr) { return {}; }
+  return {burst->hdr.hdr.num_pkts, burst->hdr.hdr.num_segs};
+}
+
+inline BurstLimits transport_limits(const BurstParams* burst) {
+  if (burst == nullptr) { return {}; }
+  return {burst->transport_hdr.num_pkts, burst->transport_hdr.num_segs};
+}
+
+inline Status validate_burst(const BurstParams* burst, const char* op_name) {
+  if (burst != nullptr) { return Status::SUCCESS; }
+  DAQIRI_LOG_ERROR("{}: burst is null", op_name);
+  return Status::NULL_PTR;
+}
+
+inline Status validate_packet_index(const BurstParams* burst,
+                                    BurstLimits limits,
+                                    int idx,
+                                    const char* op_name) {
+  Status status = validate_burst(burst, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  if (idx < 0 || (strict_enabled() && static_cast<size_t>(idx) >= limits.num_pkts)) {
+    DAQIRI_LOG_ERROR("{}: packet index {} out of range [0, {})", op_name, idx, limits.num_pkts);
+    return Status::INVALID_PARAMETER;
+  }
+  return Status::SUCCESS;
+}
+
+inline Status validate_packet_count(const BurstParams* burst,
+                                    BurstLimits limits,
+                                    size_t max_pkts,
+                                    const char* op_name) {
+  Status status = validate_burst(burst, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  if (!strict_enabled()) { return Status::SUCCESS; }
+  if (limits.num_pkts == 0 || limits.num_pkts > max_pkts) {
+    DAQIRI_LOG_ERROR("{}: invalid packet count {} (max {})",
+                     op_name,
+                     limits.num_pkts,
+                     max_pkts);
+    return Status::INVALID_PARAMETER;
+  }
+  return Status::SUCCESS;
+}
+
+inline Status validate_segment_count(const BurstParams* burst,
+                                     BurstLimits limits,
+                                     const char* op_name) {
+  Status status = validate_burst(burst, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  if (!strict_enabled()) { return Status::SUCCESS; }
+  if (limits.num_segs <= 0 || limits.num_segs > MAX_NUM_SEGS) {
+    DAQIRI_LOG_ERROR("{}: invalid segment count {}", op_name, limits.num_segs);
+    return Status::INVALID_PARAMETER;
+  }
+  return Status::SUCCESS;
+}
+
+inline Status validate_segment_index(const BurstParams* burst,
+                                     BurstLimits limits,
+                                     int seg,
+                                     const char* op_name) {
+  Status status = validate_burst(burst, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  if (!strict_enabled()) {
+    if (seg < 0 || seg >= MAX_NUM_SEGS) {
+      DAQIRI_LOG_ERROR("{}: segment index {} out of range [0, {})", op_name, seg, MAX_NUM_SEGS);
+      return Status::INVALID_PARAMETER;
+    }
+    return Status::SUCCESS;
+  }
+  status = validate_segment_count(burst, limits, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  if (seg < 0 || seg >= limits.num_segs) {
+    DAQIRI_LOG_ERROR("{}: segment index {} out of range [0, {})", op_name, seg, limits.num_segs);
+    return Status::INVALID_PARAMETER;
+  }
+  return Status::SUCCESS;
+}
+
+inline Status validate_segment_packet_storage(const BurstParams* burst,
+                                              BurstLimits limits,
+                                              int seg,
+                                              int idx,
+                                              bool require_packet_ptr,
+                                              bool require_length_ptr,
+                                              const char* op_name) {
+  Status status = validate_packet_index(burst, limits, idx, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  status = validate_segment_index(burst, limits, seg, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  if (!strict_enabled()) { return Status::SUCCESS; }
+
+  if (require_packet_ptr) {
+    if (burst->pkts[seg] == nullptr) {
+      DAQIRI_LOG_ERROR("{}: packet pointer array is null for segment {}", op_name, seg);
+      return Status::NULL_PTR;
+    }
+    if (burst->pkts[seg][idx] == nullptr) {
+      DAQIRI_LOG_ERROR("{}: packet pointer is null for packet {} segment {}", op_name, idx, seg);
+      return Status::NULL_PTR;
+    }
+  }
+
+  if (require_length_ptr && burst->pkt_lens[seg] == nullptr) {
+    DAQIRI_LOG_ERROR("{}: packet length array is null for segment {}", op_name, seg);
+    return Status::NULL_PTR;
+  }
+
+  return Status::SUCCESS;
+}
+
+inline Status validate_payload_write(const BurstParams* burst,
+                                     BurstLimits limits,
+                                     int idx,
+                                     const void* data,
+                                     int len,
+                                     size_t payload_capacity,
+                                     const char* op_name) {
+  Status status = validate_segment_packet_storage(
+      burst, limits, 0, idx, true, false, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  if (!strict_enabled()) { return Status::SUCCESS; }
+  if (len < 0) {
+    DAQIRI_LOG_ERROR("{}: payload length {} is negative", op_name, len);
+    return Status::INVALID_PARAMETER;
+  }
+  if (len > 0 && data == nullptr) {
+    DAQIRI_LOG_ERROR("{}: payload source is null for non-empty payload", op_name);
+    return Status::NULL_PTR;
+  }
+  const size_t payload_len = static_cast<size_t>(len);
+  if (payload_capacity != kUnknownCapacity && payload_len > payload_capacity) {
+    DAQIRI_LOG_ERROR("{}: payload length {} exceeds backing capacity {}",
+                     op_name,
+                     payload_len,
+                     payload_capacity);
+    return Status::INVALID_PARAMETER;
+  }
+  return Status::SUCCESS;
+}
+
+inline Status validate_packet_lengths(const BurstParams* burst,
+                                      BurstLimits limits,
+                                      int idx,
+                                      const std::initializer_list<int>& lens,
+                                      const std::array<size_t, MAX_NUM_SEGS>& capacities,
+                                      bool require_packet_ptrs,
+                                      bool require_length_ptrs,
+                                      size_t max_segment_len,
+                                      const char* op_name,
+                                      uint32_t* total_len) {
+  if (total_len != nullptr) { *total_len = 0; }
+
+  Status status = validate_packet_index(burst, limits, idx, op_name);
+  if (status != Status::SUCCESS) { return status; }
+  status = validate_segment_count(burst, limits, op_name);
+  if (status != Status::SUCCESS) { return status; }
+
+  if (!strict_enabled()) {
+    uint64_t total = 0;
+    for (int len : lens) {
+      total += static_cast<uint32_t>(len);
+      if (total > std::numeric_limits<uint32_t>::max()) {
+        DAQIRI_LOG_ERROR("{}: total packet length {} exceeds uint32_t", op_name, total);
+        return Status::INVALID_PARAMETER;
+      }
+    }
+    if (total_len != nullptr) { *total_len = static_cast<uint32_t>(total); }
+    return Status::SUCCESS;
+  }
+
+  if (lens.size() != static_cast<size_t>(limits.num_segs)) {
+    DAQIRI_LOG_ERROR("{}: got {} packet length(s), expected {} segment length(s)",
+                     op_name,
+                     lens.size(),
+                     limits.num_segs);
+    return Status::INVALID_PARAMETER;
+  }
+
+  uint64_t total = 0;
+  int seg = 0;
+  for (int len : lens) {
+    status = validate_segment_packet_storage(
+        burst, limits, seg, idx, require_packet_ptrs, require_length_ptrs, op_name);
+    if (status != Status::SUCCESS) { return status; }
+    if (len < 0) {
+      DAQIRI_LOG_ERROR("{}: packet length {} is negative for segment {}",
+                       op_name,
+                       len,
+                       seg);
+      return Status::INVALID_PARAMETER;
+    }
+
+    const size_t seg_len = static_cast<size_t>(len);
+    if (seg_len > max_segment_len) {
+      DAQIRI_LOG_ERROR("{}: packet length {} exceeds maximum segment length {} for segment {}",
+                       op_name,
+                       seg_len,
+                       max_segment_len,
+                       seg);
+      return Status::INVALID_PARAMETER;
+    }
+    if (capacities[seg] != kUnknownCapacity && seg_len > capacities[seg]) {
+      DAQIRI_LOG_ERROR("{}: packet length {} exceeds backing capacity {} for segment {}",
+                       op_name,
+                       seg_len,
+                       capacities[seg],
+                       seg);
+      return Status::INVALID_PARAMETER;
+    }
+    total += seg_len;
+    if (total > std::numeric_limits<uint32_t>::max()) {
+      DAQIRI_LOG_ERROR("{}: total packet length {} exceeds uint32_t", op_name, total);
+      return Status::INVALID_PARAMETER;
+    }
+    ++seg;
+  }
+
+  if (total_len != nullptr) { *total_len = static_cast<uint32_t>(total); }
+  return Status::SUCCESS;
+}
+
+inline std::array<size_t, MAX_NUM_SEGS> unknown_capacities() {
+  std::array<size_t, MAX_NUM_SEGS> capacities{};
+  capacities.fill(kUnknownCapacity);
+  return capacities;
+}
+
+}  // namespace daqiri::burst_validation