fix(value): [OBE-10732] bound the depth a VRL program can nest a Value to

examples/depth_probe.rs

@@ -0,0 +1,93 @@
+//! OBE-10732 spike: measure which `Value` traversal overflows first, and at what depth.
+//!
+//! Deep values are built iteratively (O(1) stack per level) so that construction itself never
+//! recurses — this isolates the traversal under test. Values we are not measuring are leaked with
+//! `mem::forget` so a stray recursive drop cannot be mistaken for the mode's own overflow.
+//!
+//! Usage: depth_probe <mode> <depth> <stack_bytes>
+//! Modes: build | drop | clone | display | serialize | partial_eq
+//!
+//! Exits 0 and prints OK when the traversal survives. A stack overflow aborts the process
+//! (SIGSEGV/SIGABRT), which is the signal the caller measures.
+
+use std::mem;
+use vrl::value::Value;
+
+fn build(depth: usize) -> Value {
+    let mut v = Value::Null;
+    for _ in 0..depth {
+        v = Value::Array(vec![v]);
+    }
+    v
+}
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() != 4 {
+        eprintln!("usage: depth_probe <mode> <depth> <stack_bytes>");
+        std::process::exit(2);
+    }
+    let mode = args[1].clone();
+    let depth: usize = args[2].parse().expect("depth");
+    let stack: usize = args[3].parse().expect("stack_bytes");
+
+    let handle = std::thread::Builder::new()
+        .stack_size(stack)
+        .spawn(move || {
+            let v = build(depth);
+
+            match mode.as_str() {
+                // Control: construction only. Should never overflow.
+                "build" => {
+                    mem::forget(v);
+                }
+                // Recursive drop glue on Vec<Value>.
+                "drop" => {
+                    drop(v);
+                }
+                // Derived Clone. The clone is measured; both values leak so drop can't confound.
+                "clone" => {
+                    let c = v.clone();
+                    mem::forget(c);
+                    mem::forget(v);
+                }
+                // Hand-written recursive Display::fmt.
+                "display" => {
+                    let s = v.to_string();
+                    mem::forget(v);
+                    mem::forget(s);
+                }
+                // Serialize -> serde_json (write side has no recursion limit).
+                "serialize" => {
+                    let s = serde_json::to_string(&v).expect("serialize");
+                    mem::forget(v);
+                    mem::forget(s);
+                }
+                // Derived PartialEq.
+                "partial_eq" => {
+                    let c = v.clone();
+                    let eq = v == c;
+                    mem::forget(c);
+                    mem::forget(v);
+                    if !eq {
+                        eprintln!("unexpected inequality");
+                        std::process::exit(3);
+                    }
+                }
+                other => {
+                    eprintln!("unknown mode: {other}");
+                    std::process::exit(2);
+                }
+            }
+            println!("OK");
+        })
+        .expect("spawn");
+
+    match handle.join() {
+        Ok(()) => std::process::exit(0),
+        Err(_) => {
+            eprintln!("PANIC");
+            std::process::exit(1)
+        }
+    }
+}

examples/vrl_depth_probe.rs

@@ -0,0 +1,83 @@
+//! OBE-10732 spike, part 2: what depth can a real VRL program actually reach?
+//!
+//! Runs the ticket's own exploit shape — `v = push([], v)` inside `for_each`, which grows nesting
+//! one level per iteration — and optionally applies a sink afterwards. Answers the reachability
+//! question that decides whether the unguardable traversals (Clone/PartialEq/Drop) need a
+//! construction cap at all.
+//!
+//! Usage: vrl_depth_probe <sink> <iterations> <stack_bytes>
+//! Sinks: none | eq | display | encode_json
+
+use std::collections::BTreeMap;
+use vrl::compiler::{state::RuntimeState, Context, TargetValue, TimeZone};
+use vrl::value::{Secrets, Value};
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() != 4 {
+        eprintln!("usage: vrl_depth_probe <sink> <iterations> <stack_bytes>");
+        std::process::exit(2);
+    }
+    let sink = args[1].clone();
+    let iters: usize = args[2].parse().expect("iterations");
+    let stack: usize = args[3].parse().expect("stack_bytes");
+
+    let sink_src = match sink.as_str() {
+        "none" => "",
+        "eq" => "if v == v { .hit = true }",
+        "display" => ".hit = to_string!(v)",
+        "encode_json" => ".hit = encode_json(v)",
+        other => {
+            eprintln!("unknown sink: {other}");
+            std::process::exit(2);
+        }
+    };
+
+    // `v = push([], v)` wraps the accumulator once per iteration: depth grows to `iters`.
+    let src = format!(
+        r#"
+v = []
+for_each(array!(.items)) -> |_i, _x| {{ v = push([], v) }}
+{sink_src}
+.depth_built = length(v)
+"#
+    );
+
+    let handle = std::thread::Builder::new()
+        .stack_size(stack)
+        .spawn(move || {
+            let fns = vrl::stdlib::all();
+            let result = match vrl::compiler::compile(&src, &fns) {
+                Ok(r) => r,
+                Err(e) => {
+                    println!("COMPILE_ERROR: {e:?}");
+                    return;
+                }
+            };
+
+            let items = Value::Array(vec![Value::Integer(0); iters]);
+            let mut target = TargetValue {
+                value: Value::Object(BTreeMap::from([("items".into(), items)])),
+                metadata: Value::Object(BTreeMap::new()),
+                secrets: Secrets::default(),
+            };
+            let mut state = RuntimeState::default();
+            let timezone = TimeZone::default();
+            let mut ctx = Context::new(&mut target, &mut state, &timezone);
+
+            match result.program.resolve(&mut ctx) {
+                Ok(_) => println!("OK"),
+                Err(e) => println!("RUNTIME_ERROR: {e}"),
+            }
+            // Falling out of scope here drops the runtime state, including the deep `v`.
+        })
+        .expect("spawn");
+
+    match handle.join() {
+        Ok(()) => std::process::exit(0),
+        Err(_) => {
+            eprintln!("PANIC");
+            std::process::exit(1)
+        }
+    }
+}

src/compiler/expression/array.rs

@@ -1,5 +1,6 @@
 use std::{collections::BTreeMap, fmt, ops::Deref};
 
+use crate::value::depth::{depth_exceeds, MAX_VALUE_DEPTH};
 use crate::value::Value;
 use crate::{
     compiler::{
@@ -29,13 +30,35 @@ impl Deref for Array {
     }
 }
 
+// OBE-10732: `v = [v]` in a loop grows nesting one level per iteration, same shape `push` closed.
+// Literal syntax can't be made fallible without breaking every array literal in existence, so —
+// as with the array-index cap in `crud/mod.rs` — an over-limit item is dropped and logged instead.
+fn cap_depth(items: Vec<Value>) -> Vec<Value> {
+    items
+        .into_iter()
+        .map(|item| {
+            if depth_exceeds(&item, MAX_VALUE_DEPTH - 1) {
+                tracing::warn!(
+                    max_depth = MAX_VALUE_DEPTH,
+                    "array literal element exceeds max value depth, replaced with null"
+                );
+                Value::Null
+            } else {
+                item
+            }
+        })
+        .collect()
+}
+
 impl Expression for Array {
     fn resolve(&self, ctx: &mut Context) -> Resolved {
-        self.inner
+        let items = self
+            .inner
             .iter()
             .map(|expr| expr.resolve(ctx))
-            .collect::<Result<Vec<_>, _>>()
-            .map(Value::Array)
+            .collect::<Result<Vec<_>, _>>()?;
+
+        Ok(Value::Array(cap_depth(items)))
     }
 
     fn resolve_constant(&self, state: &TypeState) -> Option<Value> {
@@ -139,4 +162,28 @@ mod tests {
             ])),
         }
     ];
+
+    /// A `Value` nested exactly `depth` levels: `nested(1)` is a scalar, `nested(2)` is `[scalar]`.
+    fn nested(depth: usize) -> Value {
+        let mut v = Value::Null;
+        for _ in 1..depth {
+            v = Value::Array(vec![v]);
+        }
+        v
+    }
+
+    // OBE-10732: an over-limit item is dropped, the boundary and ordinary items are untouched.
+    #[test]
+    fn cap_depth_drops_only_the_over_limit_item() {
+        let at_boundary = nested(MAX_VALUE_DEPTH - 1);
+        let items = vec![
+            Value::Integer(1),
+            at_boundary.clone(),
+            nested(MAX_VALUE_DEPTH),
+        ];
+        assert_eq!(
+            cap_depth(items),
+            vec![Value::Integer(1), at_boundary, Value::Null]
+        );
+    }
 }

src/compiler/expression/object.rs

@@ -1,5 +1,6 @@
 use std::{collections::BTreeMap, fmt, ops::Deref};
 
+use crate::value::depth::{depth_exceeds, MAX_VALUE_DEPTH};
 use crate::value::{KeyString, Value};
 use crate::{
     compiler::{
@@ -30,13 +31,34 @@ impl Deref for Object {
     }
 }
 
+// OBE-10732: `v = { "a": v }` in a loop grows nesting one level per iteration. Same tradeoff as
+// the array literal cap in `array.rs`: an over-limit field value is dropped and logged.
+fn cap_depth(fields: BTreeMap<KeyString, Value>) -> BTreeMap<KeyString, Value> {
+    fields
+        .into_iter()
+        .map(|(key, value)| {
+            if depth_exceeds(&value, MAX_VALUE_DEPTH - 1) {
+                tracing::warn!(
+                    max_depth = MAX_VALUE_DEPTH,
+                    "object literal field exceeds max value depth, replaced with null"
+                );
+                (key, Value::Null)
+            } else {
+                (key, value)
+            }
+        })
+        .collect()
+}
+
 impl Expression for Object {
     fn resolve(&self, ctx: &mut Context) -> Resolved {
-        self.inner
+        let fields: BTreeMap<_, _> = self
+            .inner
             .iter()
             .map(|(key, expr)| expr.resolve(ctx).map(|v| (key.clone(), v)))
-            .collect::<Result<BTreeMap<_, _>, _>>()
-            .map(Value::Object)
+            .collect::<Result<BTreeMap<_, _>, _>>()?;
+
+        Ok(Value::Object(cap_depth(fields)))
     }
 
     fn resolve_constant(&self, state: &TypeState) -> Option<Value> {
@@ -102,3 +124,32 @@ impl From<BTreeMap<KeyString, Expr>> for Object {
         Self { inner }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A `Value` nested exactly `depth` levels: `nested(1)` is a scalar, `nested(2)` is `[scalar]`.
+    fn nested(depth: usize) -> Value {
+        let mut v = Value::Null;
+        for _ in 1..depth {
+            v = Value::Array(vec![v]);
+        }
+        v
+    }
+
+    // OBE-10732: an over-limit field is dropped, the boundary and ordinary fields are untouched.
+    #[test]
+    fn cap_depth_drops_only_the_over_limit_field() {
+        let at_boundary = nested(MAX_VALUE_DEPTH - 1);
+        let fields = BTreeMap::from([
+            (KeyString::from("a"), Value::Integer(1)),
+            (KeyString::from("b"), at_boundary.clone()),
+            (KeyString::from("c"), nested(MAX_VALUE_DEPTH)),
+        ]);
+        let capped = cap_depth(fields);
+        assert_eq!(capped[&KeyString::from("a")], Value::Integer(1));
+        assert_eq!(capped[&KeyString::from("b")], at_boundary);
+        assert_eq!(capped[&KeyString::from("c")], Value::Null);
+    }
+}

src/stdlib/append.rs

@@ -1,8 +1,23 @@
 use crate::compiler::prelude::*;
+use crate::value::depth::{depth_exceeds, MAX_VALUE_DEPTH};
 
 fn append(value: Value, items: Value) -> Resolved {
     let mut value = value.try_array()?;
     let mut items = items.try_array()?;
+
+    // OBE-10732: same reasoning as `push` — every element of both arrays becomes a direct child
+    // of the result, so each is checked.
+    if value
+        .iter()
+        .chain(items.iter())
+        .any(|item| depth_exceeds(item, MAX_VALUE_DEPTH - 1))
+    {
+        return Err(format!(
+            "cannot append: the result would nest deeper than the limit of {MAX_VALUE_DEPTH}"
+        )
+        .into());
+    }
+
     value.append(&mut items);
     Ok(value.into())
 }
@@ -142,3 +157,38 @@ mod tests {
         }
     ];
 }
+
+#[cfg(test)]
+mod depth_tests {
+    use super::*;
+
+    /// A `Value` nested exactly `depth` levels: `nested(1)` is a scalar, `nested(2)` is `[scalar]`.
+    fn nested(depth: usize) -> Value {
+        let mut v = Value::Null;
+        for _ in 1..depth {
+            v = Value::Array(vec![v]);
+        }
+        v
+    }
+
+    #[test]
+    fn append_rejects_only_past_the_depth_cap() {
+        let boundary = Value::Array(vec![nested(MAX_VALUE_DEPTH - 1)]);
+        let over = Value::Array(vec![nested(MAX_VALUE_DEPTH)]);
+        assert!(append(Value::Array(vec![]), boundary).is_ok());
+        assert!(append(Value::Array(vec![]), over).is_err());
+    }
+
+    // The type error is more fundamental, so it must win when both are wrong.
+    #[test]
+    fn append_reports_the_type_error_before_the_depth_error() {
+        let too_deep_items = Value::Array(vec![nested(MAX_VALUE_DEPTH)]);
+        let err = append(Value::Integer(1), too_deep_items)
+            .expect_err("expected an error")
+            .to_string();
+        assert!(
+            !err.contains("nest deeper"),
+            "expected the try_array type error, got the depth error instead: {err}"
+        );
+    }
+}