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/stdlib/push.rs

@@ -1,6 +1,19 @@
 use crate::compiler::prelude::*;
+use crate::value::depth::{depth_exceeds, MAX_VALUE_DEPTH};
 
 fn push(list: Value, item: Value) -> Resolved {
+    // OBE-10732: `v = push([], v)` inside a loop grows nesting one level per iteration, which is
+    // how a VRL program builds a `Value` deep enough to overflow the stack in `Clone`, `PartialEq`
+    // or `Display`. None of those can return an error, so the only place to stop it is before the
+    // value is built. The item lands one level below the resulting array, so it may be at most
+    // `MAX_VALUE_DEPTH - 1` deep.
+    if depth_exceeds(&item, MAX_VALUE_DEPTH - 1) {
+        return Err(format!(
+            "cannot push: the result would nest deeper than the limit of {MAX_VALUE_DEPTH}"
+        )
+        .into());
+    }
+
     let mut list = list.try_array()?;
     list.push(item);
     Ok(list.into())
@@ -129,3 +142,47 @@ mod tests {
         }
     ];
 }
+
+#[cfg(test)]
+mod depth_tests {
+    use super::*;
+    use crate::value::depth::MAX_VALUE_DEPTH;
+
+    /// Builds a `Value` nested `depth` levels. Iterative, so building it costs no stack —
+    /// which is the whole reason a deep `Value` is reachable from VRL in the first place.
+    /// 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: `v = push([], v)` in a loop grows nesting one level per iteration, with no cap.
+    // Past a few thousand levels the resulting `Value` crashes the process in `PartialEq`, `Clone`
+    // or `Display` — none of which can report an error — so the only place to stop it is here.
+    #[test]
+    fn push_rejects_an_item_that_would_exceed_the_depth_cap() {
+        let item = nested(MAX_VALUE_DEPTH);
+        assert!(
+            push(Value::Array(vec![]), item).is_err(),
+            "expected an error once the result would exceed MAX_VALUE_DEPTH"
+        );
+    }
+
+    #[test]
+    fn push_accepts_an_item_at_the_boundary() {
+        let item = nested(MAX_VALUE_DEPTH - 1);
+        assert!(
+            push(Value::Array(vec![]), item).is_ok(),
+            "expected a value landing exactly at MAX_VALUE_DEPTH to be accepted"
+        );
+    }
+
+    #[test]
+    fn push_leaves_ordinary_values_alone() {
+        assert!(push(Value::Array(vec![]), Value::Integer(1)).is_ok());
+        assert!(push(Value::Array(vec![]), nested(8)).is_ok());
+    }
+}

src/value/depth.rs

@@ -0,0 +1,115 @@
+//! Bounds how deeply a [`Value`] may be nested.
+//!
+//! `Value`'s `Clone`, `PartialEq`, `Hash` and drop glue are all structurally recursive and none of
+//! them can report an error — their signatures return `Self`, `bool`, a hash and nothing. So a
+//! deeply-nested `Value` cannot be handled safely once it exists; it has to not exist. This is the
+//! same defence `serde_json` uses (a depth limit in its *parser*, `de.rs`) and, contrary to
+//! OBE-10732's description, `serde_json` has no `impl Drop for Value` to copy.
+
+use super::Value;
+
+/// Largest nesting depth a VRL program may construct.
+///
+/// Derived from measurement rather than chosen. The cheapest traversal to overflow is
+/// `Display::fmt` at ~625 bytes of stack per level, so 512 levels costs ~320 KiB — 6.4x headroom
+/// inside the 2 MiB stack tokio gives Vector's workers (Vector never calls `thread_stack_size`,
+/// so the tokio default applies). Measured limits on a 2 MiB thread, for reference:
+///
+/// | traversal    | overflows at | bytes/level |
+/// |--------------|--------------|-------------|
+/// | `Display`    | 3,294        | ~625        |
+/// | `PartialEq`  | 9,415        | ~223        |
+/// | `Clone`      | 10,983       | ~190        |
+/// | `Serialize`  | 32,951       | ~64         |
+/// | drop glue    | 43,932       | ~48         |
+///
+/// 512 also sits well above every other cap in this crate (128) and above `serde_json`'s parser
+/// limit (128), so it cannot plausibly reject legitimate data.
+pub const MAX_VALUE_DEPTH: usize = 512;
+
+/// Returns `true` if `value` nests deeper than `limit`.
+///
+/// Iterative: it walks an explicit heap worklist instead of recursing, so the check itself can
+/// never overflow the stack it exists to protect. It stops as soon as the limit is passed, so for
+/// the shape this guards against — an accumulator wrapped one level per loop iteration — the cost
+/// is O(limit) rather than O(size of value).
+pub fn depth_exceeds(value: &Value, limit: usize) -> bool {
+    // Depth-first with an explicit stack of (node, depth-of-node).
+    let mut stack: Vec<(&Value, usize)> = vec![(value, 1)];
+
+    while let Some((node, depth)) = stack.pop() {
+        if depth > limit {
+            return true;
+        }
+        match node {
+            Value::Array(array) => stack.extend(array.iter().map(|child| (child, depth + 1))),
+            Value::Object(map) => stack.extend(map.values().map(|child| (child, depth + 1))),
+            _ => {}
+        }
+    }
+
+    false
+}
+
+#[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
+    }
+
+    #[test]
+    fn scalars_have_depth_one() {
+        assert!(!depth_exceeds(&Value::Integer(1), 1));
+        assert!(depth_exceeds(&Value::Integer(1), 0));
+    }
+
+    #[test]
+    fn reports_exactly_at_the_boundary() {
+        assert!(!depth_exceeds(&nested(9), 10));
+        assert!(!depth_exceeds(&nested(10), 10));
+        assert!(depth_exceeds(&nested(11), 10));
+    }
+
+    #[test]
+    fn finds_depth_nested_in_an_object() {
+        let mut v = Value::Null;
+        for _ in 0..20 {
+            let mut map = crate::value::ObjectMap::new();
+            map.insert("a".into(), v);
+            v = Value::Object(map);
+        }
+        assert!(depth_exceeds(&v, 10));
+        assert!(!depth_exceeds(&v, 30));
+    }
+
+    // The check must not be defeated by putting the deep branch behind a wide shallow one.
+    #[test]
+    fn finds_depth_behind_breadth() {
+        let mut children: Vec<Value> = (0..1_000).map(Value::Integer).collect();
+        children.push(nested(50));
+        assert!(depth_exceeds(&Value::Array(children), 20));
+    }
+
+    // It must never recurse, or it would overflow on exactly the input it is meant to reject.
+    #[test]
+    fn does_not_itself_overflow_on_a_very_deep_value() {
+        let deep = nested(100_000);
+        assert!(depth_exceeds(&deep, MAX_VALUE_DEPTH));
+        // Drop it iteratively too, so the test does not die tearing `deep` down.
+        let mut cur = deep;
+        loop {
+            let next = match &mut cur {
+                Value::Array(a) if !a.is_empty() => a.remove(0),
+                _ => break,
+            };
+            cur = next;
+        }
+    }
+}

src/value/mod.rs

@@ -40,6 +40,7 @@ pub mod secrets;
 pub mod value;
 
 mod btreemap;
+pub(crate) mod depth;
 mod keystring;
 
 pub use kind::Kind;