fix(state): drain, cap and time-bound a state hook's output

FORK.md

@@ -192,7 +192,7 @@ Signing is unaffected either way.
 
 - `cargo fmt --all -- --check` — clean.
 - `cargo clippy --all-features --all-targets` — clean.
-- `cargo test --all-features -- --test-threads 1` — 64 unit + 23 integration tests
+- `cargo test --all-features -- --test-threads 1` — 68 unit + 23 integration tests
   pass (10 ignored, all hardware-dependent).
 - The published image is pulled and checked on every release: it reports its own
   version, keeps `tmkms` as the entrypoint, and links the USB libraries the

src/chain/state/hook.rs

@@ -8,14 +8,21 @@ use crate::{
 use cometbft::block;
 use serde::Deserialize;
 use std::{
+    io::{self, Read},
     process::{Command, Stdio},
-    time::Duration,
+    sync::mpsc::{self, RecvTimeoutError},
+    thread,
+    time::{Duration, Instant},
 };
 use wait_timeout::ChildExt;
 
 /// Default timeout to use when a user one is unspecified
 const DEFAULT_TIMEOUT_SECS: u64 = 1;
 
+/// Most output a hook may produce. Valid output is one short JSON object, so the
+/// limit exists only to stop a runaway hook from growing the signer's memory.
+const MAX_OUTPUT_BYTES: u64 = 1024 * 1024;
+
 /// Sanity limit on how far the block height from the hook can diverge from the
 /// last known state
 pub const BLOCK_HEIGHT_SANITY_LIMIT: u64 = 9000;
@@ -32,27 +39,72 @@ pub fn run(config: &HookConfig) -> Result<Output, Error> {
         .args(args)
         .stdout(Stdio::piped())
         .spawn()?;
+
+    // Drained while the hook runs rather than once it exits: a hook that writes
+    // more than the pipe holds blocks on that write until something reads it
+    let stdout = match child.stdout.take() {
+        Some(stdout) => stdout,
+        None => fail!(HookError, "couldn't consume stdout from child"),
+    };
+    let (sender, receiver) = mpsc::channel();
+    thread::Builder::new()
+        .name("state-hook-output".to_owned())
+        .spawn(move || {
+            // Fails only once `run` has stopped waiting, so nobody is left to tell
+            let _ = sender.send(read_capped(stdout));
+        })?;
+
     let timeout = Duration::from_secs(config.timeout_secs.unwrap_or(DEFAULT_TIMEOUT_SECS));
+    let deadline = Instant::now() + timeout;
 
-    match child.wait_timeout(timeout)? {
-        Some(status) => {
-            if status.success() {
-                if let Some(stdout) = child.stdout {
-                    Ok(serde_json::from_reader(stdout)?)
-                } else {
-                    fail!(HookError, "couldn't consume stdout from child");
-                }
-            } else {
-                fail!(HookError, "subcommand returned status {:?}", status.code())
-            }
-        }
+    let status = match child.wait_timeout(timeout)? {
+        Some(status) => status,
         None => {
             // timeout
             child.kill()?;
             child.wait()?;
             fail!(HookError, "subcommand timed out after {:?}", timeout)
         }
+    };
+
+    // Bounded by the hook's own deadline: a process the hook leaves behind can
+    // hold stdout open long after the hook has exited
+    let remaining = deadline.saturating_duration_since(Instant::now());
+    let output = receiver.recv_timeout(remaining);
+
+    // Checked before the exit status: a hook cut off at the limit dies of
+    // SIGPIPE, and its status alone would not say why
+    if matches!(&output, Ok(Ok(output)) if output.len() as u64 > MAX_OUTPUT_BYTES) {
+        fail!(
+            HookError,
+            "hook wrote more than {} bytes of output",
+            MAX_OUTPUT_BYTES
+        )
+    }
+
+    if !status.success() {
+        fail!(HookError, "subcommand returned status {:?}", status.code())
     }
+
+    let output = match output {
+        Ok(read) => read?,
+        Err(RecvTimeoutError::Timeout) => {
+            fail!(HookError, "hook output still open after {:?}", timeout)
+        }
+        Err(RecvTimeoutError::Disconnected) => {
+            fail!(HookError, "couldn't read the hook's output")
+        }
+    };
+
+    Ok(serde_json::from_slice(&output)?)
+}
+
+/// Read a hook's output, stopping one byte past the limit: going over it can then
+/// be told apart without holding everything a runaway hook writes
+fn read_capped(output: impl Read) -> io::Result<Vec<u8>> {
+    let mut read = Vec::new();
+    output.take(MAX_OUTPUT_BYTES + 1).read_to_end(&mut read)?;
+    Ok(read)
 }
 
 /// JSON output from the hook command (parsed with serde)
@@ -64,7 +116,32 @@ pub struct Output {
 
 #[cfg(test)]
 mod tests {
-    use crate::{config::chain::HookConfig, error::ErrorKind};
+    use crate::{
+        config::chain::HookConfig,
+        error::{Error, ErrorKind},
+    };
+    use std::{
+        io::{self, Read},
+        sync::{Mutex, MutexGuard, PoisonError},
+        time::{Duration, Instant},
+    };
+
+    /// Held while a test runs a hook. On macOS a pipe is created before it is
+    /// marked close-on-exec, so a child spawned on another test thread in that
+    /// window inherits it. A child that outlives its own test, like the `sleep &`
+    /// below, would then hold another test's output open past its deadline.
+    static SPAWNING: Mutex<()> = Mutex::new(());
+
+    /// Wait until no other test is running a hook
+    fn spawning() -> MutexGuard<'static, ()> {
+        SPAWNING.lock().unwrap_or_else(PoisonError::into_inner)
+    }
+
+    /// Run a hook with no other test spawning one at the same time
+    fn run(config: &HookConfig) -> Result<super::Output, Error> {
+        let _spawning = spawning();
+        super::run(config)
+    }
 
     /// Build a config for a hook that runs the given shell snippet
     fn shell_hook(script: &str) -> HookConfig {
@@ -77,15 +154,77 @@ mod tests {
 
     #[test]
     fn reads_latest_block_height_from_hook_stdout() {
-        let output = super::run(&shell_hook("echo '{\"latest_block_height\":\"12345\"}'"))
+        let output = run(&shell_hook("echo '{\"latest_block_height\":\"12345\"}'"))
             .expect("hook should produce output");
 
         assert_eq!(output.latest_block_height.value(), 12345);
     }
 
+    /// Nothing may wait on the hook to exit before draining its stdout: a hook
+    /// that writes more than a pipe holds blocks on the write, never exits, and
+    /// the run would be reported as a timeout instead of read
+    #[test]
+    fn a_hook_writing_more_than_a_pipe_holds_is_still_read() {
+        // 256 KiB of leading whitespace, four times the 64 KiB pipe buffer of macOS
+        // and of Linux on 4 KiB pages, yet short of the output limit, then the JSON
+        // a hook is expected to print
+        let output = run(&shell_hook(
+            "head -c 262144 /dev/zero | tr '\\0' ' '; echo '{\"latest_block_height\":\"5\"}'",
+        ))
+        .expect("a large but well-formed output should be read");
+
+        assert_eq!(output.latest_block_height.value(), 5);
+    }
+
+    /// Output past the limit is refused rather than held, and reported as such:
+    /// a hook cut off at the limit dies of SIGPIPE, so the cause must not be
+    /// reported as a bare exit status
+    #[test]
+    fn output_past_the_limit_is_an_error() {
+        // Two mebibytes of leading whitespace before otherwise valid JSON
+        let err = run(&shell_hook(
+            "head -c 2097152 /dev/zero | tr '\\0' ' '; echo '{\"latest_block_height\":\"5\"}'",
+        ))
+        .expect_err("output past the limit should be refused");
+
+        assert_eq!(*err.kind(), ErrorKind::HookError);
+        assert!(err.to_string().contains("more than"), "{err}");
+    }
+
+    /// The cap is what bounds the signer's memory. Without it the length check
+    /// still rejects an oversized hook, but only after holding all of its output
+    #[test]
+    fn reading_stops_one_byte_past_the_limit() {
+        let source = io::repeat(b' ').take(4 * super::MAX_OUTPUT_BYTES);
+
+        let read = super::read_capped(source).unwrap();
+
+        assert_eq!(read.len() as u64, super::MAX_OUTPUT_BYTES + 1);
+    }
+
+    /// The timeout bounds the whole run, reading the output included: a process
+    /// the hook leaves behind holding stdout open must not hold up startup
+    #[test]
+    fn output_held_open_past_the_timeout_is_an_error() {
+        let mut config = shell_hook("echo '{\"latest_block_height\":\"5\"}'; sleep 5 &");
+        config.timeout_secs = Some(1);
+
+        // Timed from once the lock is held, so waiting on other tests does not count
+        let _spawning = spawning();
+        let started = Instant::now();
+        let err = super::run(&config).expect_err("output held open past the timeout should fail");
+
+        assert_eq!(*err.kind(), ErrorKind::HookError);
+        assert!(
+            started.elapsed() < Duration::from_secs(4),
+            "took {:?}",
+            started.elapsed()
+        );
+    }
+
     #[test]
     fn nonzero_exit_status_is_an_error() {
-        let err = super::run(&shell_hook("exit 3")).expect_err("expected hook failure");
+        let err = run(&shell_hook("exit 3")).expect_err("expected hook failure");
 
         assert_eq!(*err.kind(), ErrorKind::HookError);
     }
@@ -98,7 +237,7 @@ mod tests {
             fail_closed: true,
         };
 
-        let err = super::run(&config).expect_err("expected an error for an empty command");
+        let err = run(&config).expect_err("expected an error for an empty command");
 
         assert_eq!(*err.kind(), ErrorKind::HookError);
     }
@@ -108,7 +247,7 @@ mod tests {
         let mut config = shell_hook("sleep 5");
         config.timeout_secs = Some(1);
 
-        let err = super::run(&config).expect_err("expected a timeout error");
+        let err = run(&config).expect_err("expected a timeout error");
 
         assert_eq!(*err.kind(), ErrorKind::HookError);
     }

src/config/chain/hook.rs

@@ -7,7 +7,7 @@ pub struct HookConfig {
     /// Command (with arguments) to invoke
     pub cmd: Vec<String>,
 
-    /// Timeout (in seconds) to wait when executing the command (default 5)
+    /// Time limit (in seconds) for the command to exit and close its stdout (default 1)
     pub timeout_secs: Option<u64>,
 
     /// Whether or not to fail open or closed if this command fails to execute.