test: add gather tests and fix fd inheritance

Signed-off-by: Cong Wang <cwang@multikernel.io>

Author: congwang-mk

crates/sandlock-core/src/context.rs

@@ -647,7 +647,7 @@ fn write_id_maps_overflow() {
 ///
 /// This function **never returns**: it calls `execvp` on success or
 /// `_exit(127)` on any error.
-pub(crate) fn confine_child(policy: &Policy, cmd: &[CString], pipes: &PipePair, cow_config: Option<&ChildMountConfig>, nested: bool) -> ! {
+pub(crate) fn confine_child(policy: &Policy, cmd: &[CString], pipes: &PipePair, cow_config: Option<&ChildMountConfig>, nested: bool, keep_fds: &[RawFd]) -> ! {
     // Helper: abort child on error. Includes the OS error automatically.
     macro_rules! fail {
         ($msg:expr) => {{
@@ -884,11 +884,11 @@ pub(crate) fn confine_child(policy: &Policy, cmd: &[CString], pipes: &PipePair,
     }
 
     // 12. Close all fds above stderr (always on for isolation)
+    let mut fds_to_keep: Vec<RawFd> = keep_fds.to_vec();
     if keep_fd >= 0 {
-        close_fds_above(2, &[keep_fd]);
-    } else {
-        close_fds_above(2, &[]);
+        fds_to_keep.push(keep_fd);
     }
+    close_fds_above(2, &fds_to_keep);
 
     // 13. Apply environment
     if policy.clean_env {

crates/sandlock-core/src/sandbox.rs

@@ -825,8 +825,11 @@ impl Sandbox {
             drop(stdout_r);
             drop(stderr_r);
 
+            // Collect target fds from gather that must survive close_fds_above
+            let gather_keep_fds: Vec<i32> = self.extra_fds.iter().map(|&(target, _)| target).collect();
+
             // This never returns.
-            context::confine_child(&self.policy, &c_cmd, &pipes, cow_config.as_ref(), nested);
+            context::confine_child(&self.policy, &c_cmd, &pipes, cow_config.as_ref(), nested, &gather_keep_fds);
         }
 
         // ===== PARENT PROCESS =====

crates/sandlock-core/tests/integration/test_pipeline.rs

@@ -1,5 +1,5 @@
 use sandlock_core::policy::Policy;
-use sandlock_core::pipeline::{Stage, Pipeline};
+use sandlock_core::pipeline::{Stage, Pipeline, Gather};
 use std::time::Duration;
 
 fn base_policy() -> Policy {
@@ -194,3 +194,106 @@ async fn test_xoa_data_flow() {
 
     let _ = std::fs::remove_dir_all(&tmp);
 }
+
+// ============================================================
+// Gather tests
+// ============================================================
+
+#[tokio::test]
+async fn test_gather_two_sources() {
+    let policy = base_policy();
+    // greeting → fd 3, name → stdin (last source)
+    // Consumer: cat fd 3 first, then cat stdin
+    let result = Gather::new()
+        .source("greeting", Stage::new(&policy, &["echo", "hello"]))
+        .source("name", Stage::new(&policy, &["echo", "world"]))
+        .consumer(Stage::new(&policy, &["sh", "-c",
+            "cat <&3; cat"
+        ]))
+        .run(None).await.unwrap();
+
+    assert!(result.success(), "exit={:?} stderr={}", result.code(),
+            result.stderr_str().unwrap_or(""));
+    let stdout = result.stdout_str().unwrap_or("");
+    assert!(stdout.contains("hello"), "missing greeting, got: {}", stdout);
+    assert!(stdout.contains("world"), "missing name, got: {}", stdout);
+}
+
+#[tokio::test]
+async fn test_gather_env_var() {
+    let policy = base_policy();
+    let result = Gather::new()
+        .source("alpha", Stage::new(&policy, &["echo", "aaa"]))
+        .source("beta", Stage::new(&policy, &["echo", "bbb"]))
+        .consumer(Stage::new(&policy, &["sh", "-c",
+            "echo $_SANDLOCK_GATHER"
+        ]))
+        .run(None).await.unwrap();
+
+    assert!(result.success());
+    let stdout = result.stdout_str().unwrap_or("");
+    assert!(stdout.contains("alpha:"), "got: {}", stdout);
+    assert!(stdout.contains("beta:"), "got: {}", stdout);
+}
+
+#[tokio::test]
+async fn test_gather_disjoint_policies() {
+    let tmp = std::env::temp_dir().join(format!("sandlock-test-gather-{}", std::process::id()));
+    let _ = std::fs::create_dir_all(&tmp);
+    let secret = tmp.join("secret.txt");
+    std::fs::write(&secret, "secret data").unwrap();
+
+    // Data source: can read the file
+    let data_policy = Policy::builder()
+        .fs_read("/usr").fs_read("/lib").fs_read("/lib64").fs_read("/bin")
+        .fs_read("/etc").fs_read("/proc").fs_read("/dev")
+        .fs_read(&tmp)
+        .build()
+        .unwrap();
+
+    // Code source: no access to tmp (like a planner)
+    let code_policy = base_policy();
+
+    // Consumer: no access to tmp either (gets data via pipe)
+    let consumer_policy = base_policy();
+
+    let result = Gather::new()
+        .source("data", Stage::new(&data_policy, &[
+            "cat", secret.to_str().unwrap()
+        ]))
+        .source("code", Stage::new(&code_policy, &[
+            "echo", "tr a-z A-Z <&3"
+        ]))
+        .consumer(Stage::new(&consumer_policy, &[
+            // code is on stdin, data is on fd 3
+            "sh", "-c", "code=$(cat); eval \"$code\""
+        ]))
+        .run(None).await.unwrap();
+
+    assert!(result.success(), "exit={:?} stderr={}", result.code(),
+            result.stderr_str().unwrap_or(""));
+    let stdout = result.stdout_str().unwrap_or("");
+    assert!(stdout.contains("SECRET DATA"), "got: {}", stdout);
+
+    let _ = std::fs::remove_dir_all(&tmp);
+}
+
+#[tokio::test]
+async fn test_gather_requires_consumer() {
+    let policy = base_policy();
+    let result = Gather::new()
+        .source("a", Stage::new(&policy, &["echo", "hello"]))
+        .run(None).await;
+
+    assert!(result.is_err());
+}
+
+#[tokio::test]
+async fn test_gather_requires_sources() {
+    let policy = base_policy();
+    let result = Gather::new()
+        .consumer(Stage::new(&policy, &["cat"]))
+        .run(None).await;
+
+    assert!(result.is_err());
+}

python/tests/test_pipeline.py

@@ -7,7 +7,7 @@
 
 import pytest
 
-from sandlock import Sandbox, Policy, Stage, Pipeline
+from sandlock import Sandbox, Policy, Stage, Pipeline, NamedStage, Gather, GatherPipeline
 
 
 # --- Helpers ---
@@ -231,3 +231,112 @@ def test_xoa_executor_no_network(self):
 
         # Executor tried to connect but was blocked
         assert not result.success
+
+
+# --- Gather ---
+
+class TestGather:
+    def test_as_returns_named_stage(self):
+        stage = Sandbox(_policy()).cmd(["echo", "hello"])
+        named = stage.as_("greeting")
+        assert isinstance(named, NamedStage)
+        assert named.name == "greeting"
+
+    def test_add_returns_gather(self):
+        a = Sandbox(_policy()).cmd(["echo", "a"]).as_("a")
+        b = Sandbox(_policy()).cmd(["echo", "b"]).as_("b")
+        g = a + b
+        assert isinstance(g, Gather)
+        assert len(g.sources) == 2
+
+    def test_gather_or_stage_returns_pipeline(self):
+        g = (
+            Sandbox(_policy()).cmd(["echo", "a"]).as_("a")
+            + Sandbox(_policy()).cmd(["echo", "b"]).as_("b")
+        )
+        gp = g | Sandbox(_policy()).cmd(["cat"])
+        assert isinstance(gp, GatherPipeline)
+
+    def test_gather_two_sources(self):
+        """Two producers pipe into one consumer via gather."""
+        result = (
+            Sandbox(_policy()).cmd(["echo", "hello"]).as_("greeting")
+            + Sandbox(_policy()).cmd(["echo", "world"]).as_("name")
+            | Sandbox(_policy()).cmd(
+                ["sh", "-c",
+                 'read name; greeting=$(cat <&3); echo "$greeting $name"']
+            )
+        ).run()
+        assert result.success, f"stderr={result.stderr}"
+        assert b"hello" in result.stdout
+        assert b"world" in result.stdout
+
+    def test_gather_with_python_inputs(self):
+        """Consumer reads gather inputs via sandlock.inputs."""
+        python_paths = [p for p in sys.path if p and os.path.isdir(p)]
+        policy = _policy(fs_readable=list(dict.fromkeys([
+            "/usr", "/lib", "/lib64", "/etc", "/bin", "/sbin",
+            "/home", _PYTHON_PREFIX,
+        ] + python_paths)))
+
+        result = (
+            Sandbox(policy).cmd(
+                [sys.executable, "-c", "print('DATA_CONTENT')"]
+            ).as_("data")
+            + Sandbox(policy).cmd(
+                [sys.executable, "-c", "print('CODE_CONTENT')"]
+            ).as_("code")
+            | Sandbox(policy).cmd(
+                [sys.executable, "-c",
+                 "from sandlock import inputs; "
+                 "print(f'code={inputs[\"code\"].strip()}'); "
+                 "print(f'data={inputs[\"data\"].strip()}')"]
+            )
+        ).run()
+        assert result.success, f"stderr={result.stderr}"
+        assert b"code=CODE_CONTENT" in result.stdout
+        assert b"data=DATA_CONTENT" in result.stdout
+
+    def test_gather_disjoint_policies(self):
+        """Sources have independent policies — data source can read
+        a file the code source cannot."""
+        with tempfile.TemporaryDirectory() as tmp:
+            secret = os.path.join(tmp, "secret.txt")
+            with open(secret, "w") as f:
+                f.write("sensitive data")
+
+            data_policy = _policy(fs_readable=list(dict.fromkeys([
+                tmp, "/usr", "/lib", "/lib64", "/etc", "/bin", "/sbin",
+                _PYTHON_PREFIX,
+            ])))
+            code_policy = _policy()
+            consumer_policy = _policy()
+
+            result = (
+                Sandbox(data_policy).cmd(["cat", secret]).as_("data")
+                + Sandbox(code_policy).cmd(
+                    ["echo", "tr a-z A-Z <&3"]
+                ).as_("code")
+                | Sandbox(consumer_policy).cmd(
+                    ["sh", "-c", 'eval "$(cat)"']
+                )
+            ).run()
+            assert result.success, f"stderr={result.stderr}"
+            assert b"SENSITIVE DATA" in result.stdout
+
+    def test_gather_three_sources(self):
+        """Three producers fan into one consumer."""
+        result = (
+            Sandbox(_policy()).cmd(["echo", "aaa"]).as_("a")
+            + Sandbox(_policy()).cmd(["echo", "bbb"]).as_("b")
+            + Sandbox(_policy()).cmd(["echo", "ccc"]).as_("c")
+            | Sandbox(_policy()).cmd(
+                ["sh", "-c",
+                 # c on stdin, a on fd 3, b on fd 4
+                 'read c; a=$(cat <&3); b=$(cat <&4); echo "$a $b $c"']
+            )
+        ).run()
+        assert result.success, f"stderr={result.stderr}"
+        assert b"aaa" in result.stdout
+        assert b"bbb" in result.stdout
+        assert b"ccc" in result.stdout