Merge branch 'security': robust parsing invariants and auth cache poison prevention

Reconciles security's UTF-8/panic-safety parsing hardening with main's DDL
parser refactor (collection_type split into a directory module; CREATE ...
INDEX parsing unified through the shared options grammar).

- collection_type: main moved build_strict_schema -> strict.rs and
  parse_column_type_str_full -> type_str.rs carrying the old panic-prone
  byte slicing; ported security's hardening there (find_ascii_case_insensitive
  for GENERATED / PRIMARY KEY / NOT NULL / DEFAULT, safe .get(..) slicing,
  stronger unicode test).
- search_index: main deleted the standalone parser; the ANALYZER name is now
  tokenized by the shared options lexer, which already slices on char
  boundaries (quote.len_utf8()), so security's parse_analyzer_name fix is
  subsumed. Preserved its regression intent with a unicode-quoted-value test
  in options/lex.rs.

Author: farhan-syah

.github/workflows/test.yml

@@ -74,6 +74,10 @@ jobs:
         run: |
           python3 scripts/ci/check_authorized_dispatch.py --self-test
           python3 scripts/ci/check_authorized_dispatch.py
+      - name: Robust-parsing gate
+        run: |
+          python3 scripts/ci/check_robust_parsing.py --self-test
+          python3 scripts/ci/check_robust_parsing.py
       - name: Install cargo-deny
         uses: taiki-e/install-action@v2
         with:

Cargo.lock

@@ -65,6 +65,9 @@ nodedb-test-support = { path = "nodedb-test-support" }
 # Async runtimes
 tokio = { version = "1", features = ["full"] }
 
+# Poison-free synchronization for security-critical in-memory state.
+parking_lot = "0.12"
+
 # Error handling & diagnostics
 thiserror = "2.0"
 tracing = "0.1"

Cargo.toml

@@ -11,7 +11,7 @@
 use nodedb_codec::error::CodecError;
 use nodedb_types::Surrogate;
 
-use crate::codec::limits::{MAX_COLUMN_ENTRIES, check_decoded_size};
+use crate::codec::limits::{MAX_COLUMN_ENTRIES, checked_capacity};
 use crate::error::{ArrayError, ArrayResult};
 use crate::types::cell_value::value::CellValue;
 
@@ -182,9 +182,15 @@ pub fn decode_attr_col(data: &[u8]) -> ArrayResult<Vec<CellValue>> {
                     .try_into()
                     .expect("invariant: bounds-checked above (body.len() >= 4)"),
             ) as usize;
-            check_decoded_size(count, MAX_COLUMN_ENTRIES, "attr_col_msgpack count")?;
+            let capacity = checked_capacity(
+                count,
+                MAX_COLUMN_ENTRIES,
+                body.len() - 4,
+                4,
+                "attr_col_msgpack count",
+            )?;
             let mut pos = 4;
-            let mut values = Vec::with_capacity(count);
+            let mut values = Vec::with_capacity(capacity);
             for _ in 0..count {
                 if pos + 4 > body.len() {
                     return Err(ArrayError::SegmentCorruption {
@@ -300,6 +306,17 @@ mod tests {
         assert_eq!(out, vals);
     }
 
+    #[test]
+    fn huge_msgpack_count_in_tiny_payload_is_rejected_before_allocation() {
+        let mut data = vec![ATTR_TAG_MSGPACK];
+        data.extend_from_slice(&(MAX_COLUMN_ENTRIES as u32).to_le_bytes());
+        let result = decode_attr_col(&data);
+        assert!(matches!(
+            result,
+            Err(crate::error::ArrayError::SegmentCorruption { .. })
+        ));
+    }
+
     #[test]
     fn attr_col_empty_roundtrip() {
         let data = encode_attr_col(&[]).unwrap();

nodedb-array/src/codec/column_codec.rs

@@ -14,7 +14,9 @@
 //   [u32 LE] index count (= cell count)
 //   [zigzag-varint deltas ...] index deltas (first value as-is, then deltas)
 
-use crate::codec::limits::{MAX_CELLS_PER_TILE, MAX_DICT_CARDINALITY, check_decoded_size};
+use crate::codec::limits::{
+    MAX_CELLS_PER_TILE, MAX_DICT_CARDINALITY, check_decoded_size, checked_capacity,
+};
 use crate::error::{ArrayError, ArrayResult};
 use crate::tile::sparse_tile::DimDict;
 use crate::types::coord::value::CoordValue;
@@ -150,9 +152,15 @@ pub fn decode_coord_axis(data: &[u8], pos: &mut usize) -> ArrayResult<DimDict> {
                         .expect("invariant: preceding bounds check guarantees 4 bytes available"),
                 ) as usize;
                 *pos += 4;
-                check_decoded_size(dict_count, MAX_DICT_CARDINALITY, "coord_delta dict_count")?;
+                let capacity = checked_capacity(
+                    dict_count,
+                    MAX_DICT_CARDINALITY,
+                    data.len() - *pos,
+                    4,
+                    "coord_delta dict_count",
+                )?;
 
-                let mut out: Vec<CoordValue> = Vec::with_capacity(dict_count);
+                let mut out: Vec<CoordValue> = Vec::with_capacity(capacity);
                 for _ in 0..dict_count {
                     if *pos + 4 > data.len() {
                         return Err(ArrayError::SegmentCorruption {
@@ -238,9 +246,15 @@ pub fn decode_coord_axis(data: &[u8], pos: &mut usize) -> ArrayResult<DimDict> {
             .expect("invariant: preceding bounds check guarantees 4 bytes available"),
     ) as usize;
     *pos += 4;
-    check_decoded_size(idx_count, MAX_CELLS_PER_TILE, "coord_delta idx_count")?;
+    let capacity = checked_capacity(
+        idx_count,
+        MAX_CELLS_PER_TILE,
+        data.len() - *pos,
+        1,
+        "coord_delta idx_count",
+    )?;
 
-    let mut indices: Vec<u32> = Vec::with_capacity(idx_count);
+    let mut indices: Vec<u32> = Vec::with_capacity(capacity);
     let mut prev: i64 = 0;
     for _ in 0..idx_count {
         let (zz, consumed) =
@@ -363,6 +377,18 @@ mod tests {
         assert_eq!(out.values, d.values);
     }
 
+    #[test]
+    fn huge_msgpack_dictionary_count_in_tiny_payload_is_rejected_before_allocation() {
+        let mut data = Vec::new();
+        data.push(DICT_TAG_MSGPACK);
+        data.extend_from_slice(&(MAX_DICT_CARDINALITY as u32).to_le_bytes());
+        let result = decode_coord_axis(&data, &mut 0);
+        assert!(matches!(
+            result,
+            Err(crate::error::ArrayError::SegmentCorruption { .. })
+        ));
+    }
+
     #[test]
     fn zero_tag_byte_is_corruption() {
         // Tag 0 is reserved/forbidden. A payload starting with 0x00 must

nodedb-array/src/codec/coord_delta.rs

@@ -18,6 +18,7 @@
 use crate::codec::coord_delta::{decode_coord_axis, encode_coord_axis};
 use crate::codec::limits::{
     MAX_CELLS_PER_TILE, MAX_DICT_CARDINALITY, MAX_RLE_RUN_LEN, MAX_RLE_RUNS, check_decoded_size,
+    checked_capacity,
 };
 use crate::error::{ArrayError, ArrayResult};
 use crate::tile::sparse_tile::DimDict;
@@ -127,7 +128,13 @@ fn decode_rle(data: &[u8], pos: &mut usize) -> ArrayResult<DimDict> {
             .expect("invariant: bounds check at line 113 guarantees 4 bytes available"),
     ) as usize;
     *pos += 4;
-    check_decoded_size(run_count, MAX_RLE_RUNS, "rle run_count")?;
+    checked_capacity(
+        run_count,
+        MAX_RLE_RUNS,
+        data.len() - *pos,
+        8,
+        "rle run_count",
+    )?;
 
     let mut indices: Vec<u32> = Vec::new();
     let mut total_len: usize = 0;
@@ -167,9 +174,15 @@ fn decode_rle(data: &[u8], pos: &mut usize) -> ArrayResult<DimDict> {
             .expect("invariant: bounds check at line 143 guarantees 4 bytes available"),
     ) as usize;
     *pos += 4;
-    check_decoded_size(dict_count, MAX_DICT_CARDINALITY, "rle dict_count")?;
+    let capacity = checked_capacity(
+        dict_count,
+        MAX_DICT_CARDINALITY,
+        data.len() - *pos,
+        4,
+        "rle dict_count",
+    )?;
 
-    let mut values: Vec<CoordValue> = Vec::with_capacity(dict_count);
+    let mut values: Vec<CoordValue> = Vec::with_capacity(capacity);
     for _ in 0..dict_count {
         if *pos + 4 > data.len() {
             return Err(ArrayError::SegmentCorruption {
@@ -215,6 +228,18 @@ mod tests {
         decode_coord_axis_rle(&buf, &mut pos).unwrap()
     }
 
+    #[test]
+    fn huge_run_count_in_tiny_payload_is_rejected_before_allocation() {
+        let mut data = Vec::new();
+        data.extend_from_slice(&RLE_MARKER.to_le_bytes());
+        data.extend_from_slice(&(MAX_RLE_RUNS as u32).to_le_bytes());
+        let result = decode_coord_axis_rle(&data, &mut 0);
+        assert!(matches!(
+            result,
+            Err(crate::error::ArrayError::SegmentCorruption { .. })
+        ));
+    }
+
     #[test]
     fn empty_dict_roundtrip() {
         let d = make_dict(vec![], vec![]);

nodedb-array/src/codec/coord_rle.rs

@@ -43,12 +43,60 @@ pub const MAX_COLUMN_ENTRIES: usize = MAX_CELLS_PER_TILE;
 /// Reject a length value that exceeds `cap`. Used everywhere a `usize`
 /// length is decoded from segment bytes.
 pub fn check_decoded_size(value: usize, cap: usize, what: &str) -> ArrayResult<()> {
+    checked_capacity(value, cap, 0, 0, what).map(|_| ())
+}
+
+/// Validate a decoded collection count before it drives an allocation.
+///
+/// `min_encoded_bytes` is the smallest on-wire representation per entry. It
+/// rejects a tiny frame carrying a huge count before reserving memory, while
+/// the caller still validates each variable-sized entry as it is consumed.
+pub fn checked_capacity(
+    value: usize,
+    cap: usize,
+    remaining_bytes: usize,
+    min_encoded_bytes: usize,
+    what: &str,
+) -> ArrayResult<usize> {
     if value > cap {
         return Err(ArrayError::SegmentCorruption {
             detail: format!(
                 "decoded {what} = {value} exceeds hard cap {cap} (segment likely corrupt)"
             ),
         });
     }
-    Ok(())
+    let minimum_bytes =
+        value
+            .checked_mul(min_encoded_bytes)
+            .ok_or_else(|| ArrayError::SegmentCorruption {
+                detail: format!("decoded {what} minimum encoded size overflows"),
+            })?;
+    if minimum_bytes > remaining_bytes {
+        return Err(ArrayError::SegmentCorruption {
+            detail: format!(
+                "decoded {what} requires at least {minimum_bytes} encoded bytes, only {remaining_bytes} remain"
+            ),
+        });
+    }
+    Ok(value)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn checked_capacity_rejects_count_without_minimum_encoded_bytes() {
+        assert!(checked_capacity(2, 8, 7, 4, "test entries").is_err());
+    }
+
+    #[test]
+    fn checked_capacity_rejects_count_over_hard_cap() {
+        assert!(checked_capacity(9, 8, 36, 4, "test entries").is_err());
+    }
+
+    #[test]
+    fn checked_capacity_returns_validated_count() {
+        assert_eq!(checked_capacity(2, 8, 8, 4, "test entries").unwrap(), 2);
+    }
 }

nodedb-array/src/codec/limits.rs

@@ -12,7 +12,7 @@ use crate::codec::column_codec::{
 };
 use crate::codec::coord_rle::decode_coord_axis_rle;
 use crate::codec::limits::{
-    MAX_ATTRS_PER_TILE, MAX_AXES_PER_TILE, MAX_CELLS_PER_TILE, check_decoded_size,
+    MAX_ATTRS_PER_TILE, MAX_AXES_PER_TILE, MAX_CELLS_PER_TILE, check_decoded_size, checked_capacity,
 };
 use crate::codec::tag::{CodecTag, peek_tag};
 use crate::error::{ArrayError, ArrayResult};
@@ -108,10 +108,16 @@ fn decode_structural(body: &[u8]) -> ArrayResult<SparseTile> {
             .expect("invariant: bounds-checked above (pos + 4 <= body.len())"),
     ) as usize;
     pos += 4;
-    check_decoded_size(axis_count, MAX_AXES_PER_TILE, "axis_count")?;
+    let axis_capacity = checked_capacity(
+        axis_count,
+        MAX_AXES_PER_TILE,
+        body.len() - pos,
+        4,
+        "axis_count",
+    )?;
 
     // Coordinate axes.
-    let mut dim_dicts = Vec::with_capacity(axis_count);
+    let mut dim_dicts = Vec::with_capacity(axis_capacity);
     for _ in 0..axis_count {
         let axis_bytes = read_framed(body, &mut pos)?;
         let mut inner_pos = 0;
@@ -150,9 +156,15 @@ fn decode_structural(body: &[u8]) -> ArrayResult<SparseTile> {
             .expect("invariant: bounds-checked above (pos + 4 <= body.len())"),
     ) as usize;
     pos += 4;
-    check_decoded_size(attr_count, MAX_ATTRS_PER_TILE, "attr_count")?;
-
-    let mut attr_cols = Vec::with_capacity(attr_count);
+    let attr_capacity = checked_capacity(
+        attr_count,
+        MAX_ATTRS_PER_TILE,
+        body.len() - pos,
+        4,
+        "attr_count",
+    )?;
+
+    let mut attr_cols = Vec::with_capacity(attr_capacity);
     for _ in 0..attr_count {
         let col_bytes = read_framed(body, &mut pos)?;
         let col = decode_attr_col(col_bytes)?;
@@ -302,6 +314,15 @@ mod tests {
         assert_eq!(out.row_kinds, tile.row_kinds);
     }
 
+    #[test]
+    fn huge_axis_count_in_tiny_structural_payload_is_rejected_before_allocation() {
+        let mut body = Vec::new();
+        body.extend_from_slice(&0u32.to_le_bytes());
+        body.extend_from_slice(&(MAX_AXES_PER_TILE as u32).to_le_bytes());
+        let error = decode_structural(&body).unwrap_err();
+        assert!(matches!(error, ArrayError::SegmentCorruption { .. }));
+    }
+
     #[test]
     fn invalid_version_returns_error() {
         let s = schema();

nodedb-array/src/codec/tile_decode.rs

@@ -27,8 +27,8 @@
 use std::mem::size_of;
 
 use crate::bounds::{
-    checked_add, checked_mul, checked_range, decoded_len, encode_input_len, encode_u32_len,
-    u32_to_usize,
+    checked_add, checked_capacity, checked_mul, checked_range, decoded_len, encode_input_len,
+    encode_u32_len, u32_to_usize,
 };
 use crate::error::CodecError;
 use crate::fastlanes;
@@ -239,7 +239,12 @@ pub fn decode(data: &[u8]) -> Result<Vec<f64>, CodecError> {
     let exceptions_end = checked_add(HEADER_SIZE, exceptions_size, "ALP exceptions")?;
     checked_range(data, HEADER_SIZE, exceptions_size, "ALP exceptions")?;
 
-    let mut exceptions = Vec::with_capacity(exception_count);
+    let exception_capacity = checked_capacity(
+        exception_count,
+        size_of::<(usize, u64)>(),
+        "ALP exception allocation",
+    )?;
+    let mut exceptions = Vec::with_capacity(exception_capacity);
     let mut pos = HEADER_SIZE;
     let mut previous_index = None;
     for _ in 0..exception_count {
@@ -283,7 +288,8 @@ pub fn decode(data: &[u8]) -> Result<Vec<f64>, CodecError> {
     }
 
     // Reconstruct f64 values using the same decode mode that was selected during encode.
-    let mut values = Vec::with_capacity(count);
+    let value_capacity = checked_capacity(count, size_of::<f64>(), "ALP values")?;
+    let mut values = Vec::with_capacity(value_capacity);
     for &int_val in &encoded_ints {
         values.push(alp_decode_value(int_val, decode_exp, mode));
     }