From 6cb1140c66e2a2e402f289f3aa273f38f70545c3 Mon Sep 17 00:00:00 2001 From: Evan Lezar Date: Wed, 23 Sep 2026 14:30:27 +0000 Subject: [PATCH 01/10] fix(drivers): normalize label namespace (#3609) Signed-off-by: Evan Lezar --- crates/openshell-driver-podman/NETWORKING.md | 4 ++-- crates/openshell-driver-podman/src/isolation.rs | 4 ++-- crates/openshell-driver-podman/src/watcher.rs | 4 ++-- crates/openshell-driver-vm/src/lifecycle.rs | 2 +- e2e/rust/tests/podman_gateway_start.rs | 2 +- e2e/rust/tests/podman_oci_identity.rs | 2 +- e2e/with-podman-gateway.sh | 2 +- 7 files changed, 10 insertions(+), 10 deletions(-) diff --git a/crates/openshell-driver-podman/NETWORKING.md b/crates/openshell-driver-podman/NETWORKING.md index 3e52cccd3c..be86ac335e 100644 --- a/crates/openshell-driver-podman/NETWORKING.md +++ b/crates/openshell-driver-podman/NETWORKING.md @@ -39,8 +39,8 @@ does not share the workload's PID, mount, or network namespaces. ## Troubleshooting Inspect both containers with the same sandbox-ID label, distinguishing -`openshell.io/isolation-role=sandbox` from -`openshell.io/isolation-role=supervisor`. +`openshell.ai/isolation-role=sandbox` from +`openshell.ai/isolation-role=supervisor`. - Sandbox fails its qualification probe: use its log to identify the denied kernel/runtime primitive. Do not add capabilities or disable runtime seccomp. diff --git a/crates/openshell-driver-podman/src/isolation.rs b/crates/openshell-driver-podman/src/isolation.rs index 421ae32757..2524186cbe 100644 --- a/crates/openshell-driver-podman/src/isolation.rs +++ b/crates/openshell-driver-podman/src/isolation.rs @@ -21,8 +21,8 @@ use openshell_sandbox_backend::boundary_protocol::{ }; use serde::{Deserialize, Serialize}; -pub const LABEL_ROLE: &str = "openshell.io/isolation-role"; -pub const WORKLOAD_FILTER: &str = "openshell.io/isolation-role=sandbox"; +pub const LABEL_ROLE: &str = "openshell.ai/isolation-role"; +pub const WORKLOAD_FILTER: &str = "openshell.ai/isolation-role=sandbox"; pub const CHANNEL_ROOT: &str = "/.openshell/channel"; pub const BOOTSTRAP_PATH: &str = "/.openshell/channel/sandbox/bootstrap.json"; pub const RUNTIME_DESCRIPTOR_PATH: &str = "/.openshell/supervisor/runtime-descriptor.json"; diff --git a/crates/openshell-driver-podman/src/watcher.rs b/crates/openshell-driver-podman/src/watcher.rs index ff0d0b12e4..d7364d4f6c 100644 --- a/crates/openshell-driver-podman/src/watcher.rs +++ b/crates/openshell-driver-podman/src/watcher.rs @@ -618,7 +618,7 @@ mod tests { vec![ StubResponse::new( StatusCode::OK, - r#"{"Id":"workload","Name":"workload","State":{"Status":"running","Running":true},"Config":{"Labels":{"openshell.ai/sandbox-id":"test","openshell.io/isolation-role":"sandbox"}}}"#, + r#"{"Id":"workload","Name":"workload","State":{"Status":"running","Running":true},"Config":{"Labels":{"openshell.ai/sandbox-id":"test","openshell.ai/isolation-role":"sandbox"}}}"#, ), StubResponse::new(StatusCode::NOT_FOUND, "missing companion"), StubResponse::new(StatusCode::NO_CONTENT, ""), @@ -651,7 +651,7 @@ mod tests { vec![ StubResponse::new( StatusCode::OK, - r#"{"Id":"workload","Name":"workload","State":{"Status":"running","Running":true},"Config":{"Labels":{"openshell.ai/sandbox-id":"test","openshell.io/isolation-role":"sandbox"}}}"#, + r#"{"Id":"workload","Name":"workload","State":{"Status":"running","Running":true},"Config":{"Labels":{"openshell.ai/sandbox-id":"test","openshell.ai/isolation-role":"sandbox"}}}"#, ), StubResponse::new( StatusCode::OK, diff --git a/crates/openshell-driver-vm/src/lifecycle.rs b/crates/openshell-driver-vm/src/lifecycle.rs index 51977ccef2..f92082d99e 100644 --- a/crates/openshell-driver-vm/src/lifecycle.rs +++ b/crates/openshell-driver-vm/src/lifecycle.rs @@ -18,7 +18,7 @@ use crate::runtime::VmBackend; /// gateway), so a guest cannot self-activate an extension. This contract /// lives with the lifecycle framework that consumes it rather than in the /// shared settings registry. -pub const SANDBOX_EXTENSION_LABEL_PREFIX: &str = "openshell.io/extension."; +pub const SANDBOX_EXTENSION_LABEL_PREFIX: &str = "openshell.ai/extension."; #[derive(Debug, Clone, PartialEq, Eq)] pub enum LaunchAbortReason { diff --git a/e2e/rust/tests/podman_gateway_start.rs b/e2e/rust/tests/podman_gateway_start.rs index 7e159ffa78..2dee7c333e 100644 --- a/e2e/rust/tests/podman_gateway_start.rs +++ b/e2e/rust/tests/podman_gateway_start.rs @@ -55,7 +55,7 @@ fn sandbox_container_running(sandbox_name: &str) -> Result { "--filter", MANAGED_BY_LABEL_FILTER, "--filter", - "label=openshell.io/isolation-role=sandbox", + "label=openshell.ai/isolation-role=sandbox", "--filter", ]) .arg(sandbox_name_filter) diff --git a/e2e/rust/tests/podman_oci_identity.rs b/e2e/rust/tests/podman_oci_identity.rs index c56c700d01..54a5437397 100644 --- a/e2e/rust/tests/podman_oci_identity.rs +++ b/e2e/rust/tests/podman_oci_identity.rs @@ -137,7 +137,7 @@ fn container_id_for_role( role: &str, ) -> Result { let name_filter = format!("label=openshell.ai/sandbox-name={sandbox_name}"); - let role_filter = format!("label=openshell.io/isolation-role={role}"); + let role_filter = format!("label=openshell.ai/isolation-role={role}"); let stdout = run_engine( engine, &[ diff --git a/e2e/with-podman-gateway.sh b/e2e/with-podman-gateway.sh index 97cb750dde..338217453c 100755 --- a/e2e/with-podman-gateway.sh +++ b/e2e/with-podman-gateway.sh @@ -196,7 +196,7 @@ cleanup() { local workload_ids workload_id workload_ids="$(podman_cmd ps -aq --filter "label=openshell.managed=true" \ --filter "label=openshell.ai/sandbox-id=${sandbox_id}" \ - --filter "label=openshell.io/isolation-role=sandbox" 2>/dev/null || true)" + --filter "label=openshell.ai/isolation-role=sandbox" 2>/dev/null || true)" for workload_id in ${workload_ids}; do podman_cmd rm -f "${workload_id}" >/dev/null 2>&1 || true done From 11f1fe5806797d8c211c29b411017197a3d810e8 Mon Sep 17 00:00:00 2001 From: Florent BENOIT Date: Wed, 23 Sep 2026 14:36:57 +0000 Subject: [PATCH 02/10] fix(vm): relocate per-sandbox Unix sockets to /tmp to fit macOS sun_path (#3544) The VM driver bound control.sock and ssh.sock inside the per-sandbox state directory, which defaults to $HOME/.local/state/openshell/vm-driver/sandboxes//. On macOS the sun_path field of sockaddr_un holds 104 bytes, so a normal home directory plus the sandbox UUID already pushes the socket path past the limit and bind fails with "path too long". Bind both sockets under a short driver-owned namespace instead: /tmp/os--<128-bit-hex>//. The worst case with a UUID sandbox id is 101 bytes. Security properties of the new location: - The root is created with mkdir mode 0700 and a random 128-bit name, retrying on EEXIST, so a local user cannot pre-create the path to block the driver or plant a symlink. - The root's directory FD is held for the driver lifetime. Per-sandbox leaves are created and removed with mkdirat/unlinkat relative to that FD, so there is no path re-resolution between check and mutation. - Leaf creation rejects any sandbox id whose socket path would still exceed sun_path, returning a clear error instead of a bind failure. The driver removes the whole root on shutdown so restarts do not accumulate orphaned roots in /tmp. VM children are kill_on_drop, so no socket is live at that point. Restore-on-restart goes through the normal launch path and recreates leaves under the new root. Signed-off-by: Florent Benoit --- crates/openshell-driver-vm/src/driver.rs | 269 ++++++++++++++++++++++- crates/openshell-driver-vm/src/main.rs | 7 +- 2 files changed, 272 insertions(+), 4 deletions(-) diff --git a/crates/openshell-driver-vm/src/driver.rs b/crates/openshell-driver-vm/src/driver.rs index 243a0ba0ca..78e0b7b63c 100644 --- a/crates/openshell-driver-vm/src/driver.rs +++ b/crates/openshell-driver-vm/src/driver.rs @@ -77,6 +77,8 @@ use std::io::{BufRead, BufReader, BufWriter, Read, Seek, SeekFrom, Write}; #[cfg(unix)] use std::os::fd::AsRawFd as _; #[cfg(unix)] +use std::os::fd::OwnedFd; +#[cfg(unix)] use std::os::unix::fs::PermissionsExt; use std::path::{Component, Path, PathBuf}; use std::pin::Pin; @@ -663,6 +665,8 @@ fn provisioning_span( #[derive(Clone)] pub struct VmDriver { config: VmDriverConfig, + socket_root: PathBuf, + socket_root_fd: Arc, launcher_bin: PathBuf, registry: Arc>>, image_cache_lock: Arc>, @@ -742,9 +746,14 @@ impl VmDriver { None }; + let (socket_root, socket_root_fd) = allocate_socket_root() + .map_err(|err| format!("failed to allocate socket root in /tmp: {err}"))?; + let (events, _) = broadcast::channel(WATCH_BUFFER); let driver = Self { config, + socket_root, + socket_root_fd: Arc::new(socket_root_fd), launcher_bin, registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -931,7 +940,7 @@ impl VmDriver { .env(openshell_core::sandbox_env::SANDBOX, &sandbox.name) .env( openshell_core::sandbox_env::SSH_SOCKET_PATH, - state_dir.join("ssh.sock"), + sandbox_socket_dir(&self.socket_root, &sandbox.id).join("ssh.sock"), ) .env( openshell_core::sandbox_env::PROXY_TLS_DIR, @@ -1145,10 +1154,20 @@ impl VmDriver { return Err(Status::internal(format!("create state dir failed: {err}"))); } + if let Err(err) = + create_sandbox_socket_dir(&self.socket_root_fd, &self.socket_root, &sandbox.id) + { + let mut registry = self.registry.lock().await; + registry.remove(&sandbox.id); + let _ = tokio::fs::remove_dir_all(&state_dir).await; + return Err(Status::internal(format!("create socket dir failed: {err}"))); + } + if let Err(err) = self.ensure_extension_state_dirs(&state_dir).await { let mut registry = self.registry.lock().await; registry.remove(&sandbox.id); let _ = tokio::fs::remove_dir_all(&state_dir).await; + remove_sandbox_socket_dir(&self.socket_root_fd, &sandbox.id); return Err(err); } @@ -1156,6 +1175,7 @@ impl VmDriver { let mut registry = self.registry.lock().await; registry.remove(&sandbox.id); let _ = tokio::fs::remove_dir_all(&state_dir).await; + remove_sandbox_socket_dir(&self.socket_root_fd, &sandbox.id); return Err(Status::internal(format!( "write sandbox start metadata failed: {err}" ))); @@ -1231,6 +1251,7 @@ impl VmDriver { if overlay_preparation == OverlayPreparation::Fresh { let _ = tokio::fs::remove_dir_all(&state_dir).await; } + remove_sandbox_socket_dir(&self.socket_root_fd, &sandbox_id); return; } @@ -1502,7 +1523,10 @@ impl VmDriver { } let console_output = state_dir.join("rootfs-console.log"); - let control_socket = state_dir.join(VM_CONTROL_SOCKET); + create_sandbox_socket_dir(&self.socket_root_fd, &self.socket_root, &sandbox.id) + .map_err(|err| Status::internal(format!("create socket dir failed: {err}")))?; + let control_socket = + sandbox_socket_dir(&self.socket_root, &sandbox.id).join(VM_CONTROL_SOCKET); let session_id = launch_authentication.supervisor.session_id; let channel_tls = generate_sandbox_tls_material(session_id) .map_err(|error| Status::internal(error.to_string()))?; @@ -2014,6 +2038,7 @@ impl VmDriver { } remove_sandbox_state_dir(&self.config.state_dir, &state_dir).await?; + remove_sandbox_socket_dir(&self.socket_root_fd, &record_id); { let mut registry = self.registry.lock().await; @@ -2347,6 +2372,12 @@ impl VmDriver { true } + /// Best-effort removal of the per-driver socket root. VM children are + /// `kill_on_drop`, so no socket is live once the server has stopped. + pub fn remove_socket_root(&self) { + let _ = fs::remove_dir_all(&self.socket_root); + } + fn release_gpu(&self, sandbox_id: &str) { if let Some(inventory) = self.gpu_inventory.as_ref() && let Ok(mut inv) = inventory.lock() @@ -2639,6 +2670,7 @@ impl VmDriver { if remove_state { let _ = tokio::fs::remove_dir_all(state_dir).await; + remove_sandbox_socket_dir(&self.socket_root_fd, sandbox_id); } self.publish_platform_event( sandbox_id.to_string(), @@ -5686,6 +5718,88 @@ async fn restrict_owner_only_dir(_path: &Path) -> Result<(), std::io::Error> { Ok(()) } +const SOCKET_ROOT_ALLOC_RETRIES: usize = 16; +/// macOS `sun_path` is 104 bytes including the NUL terminator. +const MAX_SUN_PATH_LEN: usize = 103; + +fn sandbox_socket_dir(socket_root: &Path, sandbox_id: &str) -> PathBuf { + socket_root.join(sandbox_id) +} + +fn allocate_socket_root() -> Result<(PathBuf, OwnedFd), std::io::Error> { + let uid = rustix::process::geteuid().as_raw(); + allocate_socket_root_with(Path::new("/tmp"), || { + let random: u128 = rand::random(); + format!("os-{uid}-{random:032x}") + }) +} + +fn allocate_socket_root_with( + base: &Path, + mut gen_name: impl FnMut() -> String, +) -> Result<(PathBuf, OwnedFd), std::io::Error> { + for _ in 0..SOCKET_ROOT_ALLOC_RETRIES { + let root = base.join(gen_name()); + match rustix::fs::mkdir(&root, rustix::fs::Mode::from_raw_mode(0o700)) { + Ok(()) => { + let fd = rustix::fs::open( + &root, + rustix::fs::OFlags::RDONLY + | rustix::fs::OFlags::DIRECTORY + | rustix::fs::OFlags::NOFOLLOW, + rustix::fs::Mode::empty(), + ) + .map_err(std::io::Error::from)?; + return Ok((root, fd)); + } + Err(rustix::io::Errno::EXIST) => {} + Err(e) => return Err(e.into()), + } + } + Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!( + "failed to allocate socket root under {} after {SOCKET_ROOT_ALLOC_RETRIES} attempts", + base.display() + ), + )) +} + +fn create_sandbox_socket_dir( + root_fd: &OwnedFd, + socket_root: &Path, + sandbox_id: &str, +) -> Result { + let longest = socket_root.join(sandbox_id).join(VM_CONTROL_SOCKET); + if longest.as_os_str().len() > MAX_SUN_PATH_LEN { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!( + "socket path {} exceeds sun_path limit ({MAX_SUN_PATH_LEN} bytes)", + longest.display() + ), + )); + } + match rustix::fs::mkdirat(root_fd, sandbox_id, rustix::fs::Mode::from_raw_mode(0o700)) { + Ok(()) | Err(rustix::io::Errno::EXIST) => {} + Err(e) => return Err(e.into()), + } + Ok(socket_root.join(sandbox_id)) +} + +fn remove_sandbox_socket_dir(root_fd: &OwnedFd, sandbox_id: &str) { + if let Ok(leaf_fd) = rustix::fs::openat( + root_fd, + sandbox_id, + rustix::fs::OFlags::RDONLY | rustix::fs::OFlags::DIRECTORY | rustix::fs::OFlags::NOFOLLOW, + rustix::fs::Mode::empty(), + ) { + let _ = rustix::fs::unlinkat(&leaf_fd, VM_CONTROL_SOCKET, rustix::fs::AtFlags::empty()); + let _ = rustix::fs::unlinkat(&leaf_fd, "ssh.sock", rustix::fs::AtFlags::empty()); + } + let _ = rustix::fs::unlinkat(root_fd, sandbox_id, rustix::fs::AtFlags::REMOVEDIR); +} + #[allow(clippy::result_large_err)] fn sandbox_state_dir(root: &Path, sandbox_id: &str) -> Result { validate_sandbox_id(sandbox_id)?; @@ -6988,6 +7102,22 @@ mod tests { use std::time::{SystemTime, UNIX_EPOCH}; use tonic::Code; + fn test_socket_root() -> (PathBuf, Arc) { + let dir = std::env::temp_dir().join(format!("os-test-{:016x}", rand::random::())); + std::fs::create_dir(&dir).expect("create test socket root"); + std::fs::set_permissions(&dir, fs::Permissions::from_mode(0o700)) + .expect("set test socket root permissions"); + let fd = rustix::fs::open( + &dir, + rustix::fs::OFlags::RDONLY + | rustix::fs::OFlags::DIRECTORY + | rustix::fs::OFlags::NOFOLLOW, + rustix::fs::Mode::empty(), + ) + .expect("open test socket root"); + (dir, Arc::new(fd)) + } + static ENV_LOCK: std::sync::LazyLock> = std::sync::LazyLock::new(|| std::sync::Mutex::new(())); @@ -8526,6 +8656,105 @@ mod tests { let _ = std::fs::remove_dir_all(root); } + #[test] + fn create_sandbox_socket_dir_rejects_over_long_ids() { + let (root, fd) = test_socket_root(); + let err = create_sandbox_socket_dir(&fd, &root, &"x".repeat(128)).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); + assert!(!root.join("x".repeat(128)).exists()); + let _ = std::fs::remove_dir_all(root); + } + + #[test] + fn sandbox_socket_dir_fits_macos_sun_path() { + let sandbox_id = "3eb2ad45-bead-4c2e-bd10-1a4a7f3a2721"; + let worst_root = "/tmp/os-4294967295-00000000000000000000000000000000"; + let control = sandbox_socket_dir(Path::new(worst_root), sandbox_id).join(VM_CONTROL_SOCKET); + let ssh = sandbox_socket_dir(Path::new(worst_root), sandbox_id).join("ssh.sock"); + assert!( + control.as_os_str().len() < 104, + "control socket path exceeds macOS sun_path: {}", + control.display() + ); + assert!( + ssh.as_os_str().len() < 104, + "ssh socket path exceeds macOS sun_path: {}", + ssh.display() + ); + } + + #[test] + fn allocate_socket_root_returns_distinct_roots() { + let tmp = tempfile::tempdir().unwrap(); + let mut counter = 0u64; + let (root_a, _fd_a) = allocate_socket_root_with(tmp.path(), || { + counter += 1; + format!("root-{counter}") + }) + .unwrap(); + let (root_b, _fd_b) = allocate_socket_root_with(tmp.path(), || { + counter += 1; + format!("root-{counter}") + }) + .unwrap(); + assert_ne!( + root_a, root_b, + "two allocations must produce distinct roots" + ); + } + + #[test] + fn allocate_socket_root_retries_on_occupied_name() { + let tmp = tempfile::tempdir().unwrap(); + let victim = tmp.path().join("victim"); + std::fs::create_dir(&victim).unwrap(); + let original_mode = std::fs::metadata(&victim).unwrap().permissions().mode() & 0o777; + + // attempt-0: symlink to victim directory + std::os::unix::fs::symlink(&victim, tmp.path().join("attempt-0")).unwrap(); + // attempt-1: existing directory (simulates foreign-owned pre-creation) + std::fs::create_dir(tmp.path().join("attempt-1")).unwrap(); + + let call_count = AtomicUsize::new(0); + let (root, _fd) = allocate_socket_root_with(tmp.path(), || { + let n = call_count.fetch_add(1, Ordering::SeqCst); + format!("attempt-{n}") + }) + .expect("should succeed after retries"); + + assert_eq!(root, tmp.path().join("attempt-2")); + assert_eq!(call_count.load(Ordering::SeqCst), 3); + assert!( + tmp.path() + .join("attempt-0") + .symlink_metadata() + .unwrap() + .file_type() + .is_symlink(), + "pre-created symlink must not be removed" + ); + assert!( + tmp.path().join("attempt-1").exists(), + "pre-created directory must not be removed" + ); + let after_mode = std::fs::metadata(&victim).unwrap().permissions().mode() & 0o777; + assert_eq!( + original_mode, after_mode, + "victim directory permissions must not change" + ); + } + + #[test] + fn allocate_socket_root_exhaustion_returns_error() { + let tmp = tempfile::tempdir().unwrap(); + let blocked = tmp.path().join("blocked"); + std::fs::create_dir(&blocked).unwrap(); + + let err = allocate_socket_root_with(tmp.path(), || "blocked".to_string()) + .expect_err("should fail when all names are occupied"); + assert_eq!(err.kind(), std::io::ErrorKind::AlreadyExists); + } + #[test] fn sandbox_state_dir_rejects_path_unsafe_ids() { let err = sandbox_state_dir(Path::new("/tmp/openshell-vm"), "../escape") @@ -8815,11 +9044,14 @@ mod tests { #[test] fn capabilities_report_configured_default_image() { + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { default_image: "openshell/sandbox:dev".to_string(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("/tmp/openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -8851,11 +9083,14 @@ mod tests { #[test] fn resolved_sandbox_image_prefers_template_image() { + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { default_image: "openshell/sandbox:default".to_string(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("/tmp/openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -8882,11 +9117,14 @@ mod tests { #[test] fn resolved_sandbox_image_falls_back_to_driver_default() { + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { default_image: "openshell/sandbox:default".to_string(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("/tmp/openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -8910,8 +9148,11 @@ mod tests { #[test] fn resolved_sandbox_image_returns_none_without_template_or_default() { + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig::default(), + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("/tmp/openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -8932,12 +9173,15 @@ mod tests { #[test] fn bootstrap_image_ref_prefers_explicit_bootstrap_image() { + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { default_image: "openshell/sandbox:default".to_string(), bootstrap_image: "openshell/sandbox-bootstrap:latest".to_string(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("/tmp/openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -8954,11 +9198,14 @@ mod tests { #[test] fn bootstrap_image_ref_falls_back_to_default_image() { + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { default_image: "openshell/sandbox:default".to_string(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("/tmp/openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -8975,8 +9222,11 @@ mod tests { #[test] fn bootstrap_image_ref_falls_back_to_requested_image() { + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig::default(), + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("/tmp/openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -9378,11 +9628,14 @@ mod tests { let base = unique_temp_dir(); let driver_state = base.join("driver-state"); let (events, _) = broadcast::channel(WATCH_BUFFER); + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { state_dir: driver_state.clone(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -9440,11 +9693,14 @@ mod tests { let base = unique_temp_dir(); let driver_state = base.join("driver-state"); let (events, _) = broadcast::channel(WATCH_BUFFER); + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { state_dir: driver_state.clone(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -9490,12 +9746,15 @@ mod tests { let base = unique_temp_dir(); let driver_state = base.join("driver-state"); let (events, _) = broadcast::channel(WATCH_BUFFER); + let (socket_root, socket_root_fd) = test_socket_root(); let driver = VmDriver { config: VmDriverConfig { state_dir: driver_state.clone(), default_image: "ghcr.io/example/sandbox:latest".to_string(), ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -9889,12 +10148,15 @@ mod tests { /// Driver whose rootfs tar staging root is an isolated temp directory. fn rootfs_tar_test_driver(staging_root: &Path, max_bytes: Option) -> VmDriver { let (events, _) = broadcast::channel(WATCH_BUFFER); + let (socket_root, socket_root_fd) = test_socket_root(); VmDriver { config: VmDriverConfig { rootfs_tar_staging_dir: Some(staging_root.to_path_buf()), rootfs_tar_max_bytes: max_bytes, ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), @@ -10210,6 +10472,7 @@ mod tests { fn test_driver_with_extensions(extensions: LifecycleExtensionRegistry) -> VmDriver { let (events, _) = broadcast::channel(WATCH_BUFFER); + let (socket_root, socket_root_fd) = test_socket_root(); VmDriver { config: VmDriverConfig { grpc_endpoint: "http://127.0.0.1:8080".to_string(), @@ -10219,6 +10482,8 @@ mod tests { gpu_mem_mib: 16384, ..Default::default() }, + socket_root, + socket_root_fd, launcher_bin: PathBuf::from("openshell-driver-vm"), registry: Arc::new(Mutex::new(HashMap::new())), image_cache_lock: Arc::new(Mutex::new(())), diff --git a/crates/openshell-driver-vm/src/main.rs b/crates/openshell-driver-vm/src/main.rs index eedb86fecb..fe0b035023 100644 --- a/crates/openshell-driver-vm/src/main.rs +++ b/crates/openshell-driver-vm/src/main.rs @@ -321,7 +321,8 @@ async fn main() -> Result<()> { .await .map_err(|err| miette::miette!("{err}"))?; - match listen_mode { + let socket_cleanup = driver.clone(); + let result = match listen_mode { ComputeDriverListenMode::Unix { socket_path, expected_peer_pid, @@ -352,7 +353,9 @@ async fn main() -> Result<()> { .await .into_diagnostic() } - } + }; + socket_cleanup.remove_socket_root(); + result } async fn shutdown_signal() { From fd49df4f41a33802fdee45694f5ef27b827db700 Mon Sep 17 00:00:00 2001 From: Jim Meyer Date: Wed, 23 Sep 2026 15:35:08 +0000 Subject: [PATCH 03/10] fix(ci): use approved setup-oras revision (#3625) Signed-off-by: Jim Meyer --- .github/workflows/release-tag.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/release-tag.yml b/.github/workflows/release-tag.yml index 10465d251d..98fd1394e1 100644 --- a/.github/workflows/release-tag.yml +++ b/.github/workflows/release-tag.yml @@ -729,7 +729,7 @@ jobs: name: ${{ needs.qualification-result.outputs.artifact-name }} - name: Install ORAS - uses: oras-project/setup-oras@38de303aac69abb66f3e6255b7198bff35f323e3 # v2.0.0 + uses: oras-project/setup-oras@1d808f7d7f6995cc68b7bf507bfe5c5446e1dc9d # v2.0.0 - name: Log in to GHCR run: echo "${{ secrets.GITHUB_TOKEN }}" | oras login ghcr.io -u "${GITHUB_ACTOR}" --password-stdin From c02683688f1111801df00929723a051e3c727d5f Mon Sep 17 00:00:00 2001 From: krishicks Date: Wed, 23 Sep 2026 15:47:35 +0000 Subject: [PATCH 04/10] ci(e2e): run the Kubernetes HA and credential-driver suites on test:e2e-kubernetes (#3626) PRs labeled test:e2e-kubernetes now run the Kubernetes HA suite and the Kubernetes credential-driver suite (Kubernetes Secrets and Vault). Both stay optional and off in merge groups. - Read the test:e2e-kubernetes label for the HA and credential-driver lanes in Branch E2E Checks. - Point gator at test:e2e for Helm and Kubernetes coverage and at test:e2e-kubernetes for gateway high availability and credential driver storage. Refs #3481 Signed-off-by: Kris Hicks --- .github/workflows/branch-e2e.yml | 4 ++-- scripts/agents/gator/skills/gator-gate/SKILL.md | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/.github/workflows/branch-e2e.yml b/.github/workflows/branch-e2e.yml index 99c68b9793..bac688c88b 100644 --- a/.github/workflows/branch-e2e.yml +++ b/.github/workflows/branch-e2e.yml @@ -47,8 +47,8 @@ jobs: push) run_core_e2e="$(jq -r 'index("test:e2e") != null' <<< "$LABELS_JSON")" run_gpu_e2e="$(jq -r 'index("test:e2e-gpu") != null' <<< "$LABELS_JSON")" - run_kubernetes_ha_e2e=false - run_kubernetes_credential_drivers_e2e=false + run_kubernetes_ha_e2e="$(jq -r 'index("test:e2e-kubernetes") != null' <<< "$LABELS_JSON")" + run_kubernetes_credential_drivers_e2e="$(jq -r 'index("test:e2e-kubernetes") != null' <<< "$LABELS_JSON")" ;; merge_group) # Merge groups have no PR labels. When GPU E2E is required as documented diff --git a/scripts/agents/gator/skills/gator-gate/SKILL.md b/scripts/agents/gator/skills/gator-gate/SKILL.md index 0d70551af8..494946286d 100644 --- a/scripts/agents/gator/skills/gator-gate/SKILL.md +++ b/scripts/agents/gator/skills/gator-gate/SKILL.md @@ -918,7 +918,9 @@ Use `test:e2e` for changes that affect: Use `test:e2e-gpu` for GPU runtime, CDI, CUDA, GPU driver, or GPU policy behavior. -Use `test:e2e-kubernetes` for Kubernetes HA, Helm, Agent Sandbox CRDs, Kubernetes scheduling, namespace, or controller behavior when the Kubernetes-specific suite is needed. +Use `test:e2e` for Helm, Agent Sandbox CRDs, Kubernetes scheduling, namespace, workspace mode, or controller behavior; it runs the Kubernetes, managed-workspace, and operator-workspace suites. + +Add `test:e2e-kubernetes` for gateway high availability or Kubernetes Secrets and Vault credential driver storage. It runs only the HA and credential-driver suites, so apply it alongside `test:e2e` when other Kubernetes coverage is also needed. Apply `test:windows` whenever a PR affects Windows support. This includes Windows or MSVC CI and `mise` tasks, Windows lock-file platform entries, the MXC driver, and `cfg(windows)` implementations or tests in otherwise cross-platform crates. Apply it alongside any applicable E2E label; it is not a substitute for runtime coverage. From f3e097d6c1a251c645a847cf900742ec2864344e Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Wed, 23 Sep 2026 17:28:36 +0000 Subject: [PATCH 05/10] chore(deps): bump anyio from 4.13.0 to 4.14.2 (#3474) Bumps [anyio](https://github.com/agronholm/anyio) from 4.13.0 to 4.14.2. - [Release notes](https://github.com/agronholm/anyio/releases) - [Commits](https://github.com/agronholm/anyio/compare/4.13.0...4.14.2) --- updated-dependencies: - dependency-name: anyio dependency-version: 4.14.2 dependency-type: indirect ... Signed-off-by: dependabot[bot] Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> --- uv.lock | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/uv.lock b/uv.lock index 063ef739e6..fa223a9c7a 100644 --- a/uv.lock +++ b/uv.lock @@ -4,15 +4,15 @@ requires-python = ">=3.11" [[package]] name = "anyio" -version = "4.13.0" +version = "4.14.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "idna" }, { name = "typing-extensions", marker = "python_full_version < '3.13'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/19/14/2c5dd9f512b66549ae92767a9c7b330ae88e1932ca57876909410251fe13/anyio-4.13.0.tar.gz", hash = "sha256:334b70e641fd2221c1505b3890c69882fe4a2df910cba14d97019b90b24439dc", size = 231622, upload-time = "2026-03-24T12:59:09.671Z" } +sdist = { url = "https://files.pythonhosted.org/packages/61/cc/a381afa6efea9f496eff839d4a6a1aed3bfafc7b3ab4b0d1b243a12573dd/anyio-4.14.2.tar.gz", hash = "sha256:cfa139f3ed1a23ee8f88a145ddb5ac7605b8bbfd8592baacd7ce3d8bb4313c7f", size = 260176, upload-time = "2026-07-12T20:29:07.082Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/da/42/e921fccf5015463e32a3cf6ee7f980a6ed0f395ceeaa45060b61d86486c2/anyio-4.13.0-py3-none-any.whl", hash = "sha256:08b310f9e24a9594186fd75b4f73f4a4152069e3853f1ed8bfbf58369f4ad708", size = 114353, upload-time = "2026-03-24T12:59:08.246Z" }, + { url = "https://files.pythonhosted.org/packages/da/35/f2287558c17e29fafc8ef3daf819bb9834061cfa43bff8014f7df7f63bdc/anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494", size = 125813, upload-time = "2026-07-12T20:29:05.763Z" }, ] [[package]] From bed9e5eafcc86e45f0ec837a9787556bacae75cf Mon Sep 17 00:00:00 2001 From: Piotr Mlocek Date: Wed, 23 Sep 2026 17:31:13 +0000 Subject: [PATCH 06/10] fix(supervisor): use better error message when sandbox connect is not available (#3572) * fix(supervisor): explain unavailable main terminal attachments Signed-off-by: Piotr Mlocek * fix(supervisor): reset cursor after terminal attachment errors Signed-off-by: Piotr Mlocek * fix(supervisor): format read-only warnings for terminal clients Signed-off-by: Piotr Mlocek * docs: drop terminal attachment documentation additions Signed-off-by: Piotr Mlocek * fix(supervisor): exit read-only viewers on Ctrl-C Signed-off-by: Piotr Mlocek * docs: simplify read-only viewer Ctrl-C guidance Signed-off-by: Piotr Mlocek --------- Signed-off-by: Piotr Mlocek --- architecture/sandbox.md | 4 +- .../src/main_session.rs | 12 +- .../openshell-supervisor-process/src/ssh.rs | 320 +++++++++++++++++- docs/sandboxes/manage-sandboxes.mdx | 3 +- e2e/rust/tests/sandbox_lifecycle.rs | 37 +- skills/openshell-cli/SKILL.md | 12 +- 6 files changed, 360 insertions(+), 28 deletions(-) diff --git a/architecture/sandbox.md b/architecture/sandbox.md index e6f31436a4..e269634547 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -648,7 +648,9 @@ sandbox workload directly. The relay supports: - Attachment to the canonical main process through the `openshell-main` SSH subsystem. The supervisor owns its retained PTY or pipes, a 1 MiB replay - buffer, and a single stdin lease across client disconnects. + buffer, and a single stdin lease across client disconnects. Ctrl-C interrupts + the foreground process. For read-only attachments, Ctrl-C only exits the + current viewer. - Independent interactive shell sessions. - Command execution. Commands run through a login shell (`bash -lc`) by default, so the first of the user's `.bash_profile`, `.bash_login`, or `.profile` is diff --git a/crates/openshell-supervisor-process/src/main_session.rs b/crates/openshell-supervisor-process/src/main_session.rs index 3ffa24d1d0..1e59aa5c6e 100644 --- a/crates/openshell-supervisor-process/src/main_session.rs +++ b/crates/openshell-supervisor-process/src/main_session.rs @@ -211,8 +211,13 @@ impl MainSession { const REMOTE_OUTPUT_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); #[cfg(test)] pub fn inert() -> Arc { - let (input, _input_rx) = tokio::sync::mpsc::channel(64); - Arc::new(Self { + Self::inert_with_input().0 + } + + #[cfg(test)] + pub fn inert_with_input() -> (Arc, tokio::sync::mpsc::Receiver>) { + let (input, input_rx) = tokio::sync::mpsc::channel(64); + let session = Arc::new(Self { pid: 1, terminal: false, input, @@ -231,7 +236,8 @@ impl MainSession { expectation: AttachmentExpectation::None, }), terminal_attachments_done: Notify::new(), - }) + }); + (session, input_rx) } #[cfg(test)] diff --git a/crates/openshell-supervisor-process/src/ssh.rs b/crates/openshell-supervisor-process/src/ssh.rs index 1eca60d933..911cf0a072 100644 --- a/crates/openshell-supervisor-process/src/ssh.rs +++ b/crates/openshell-supervisor-process/src/ssh.rs @@ -635,14 +635,45 @@ impl russh::server::Handler for SshHandler { session: &mut Session, ) -> Result<(), Self::Error> { if name == "openshell-main" { - let Some(main_session) = self.main_session.clone() else { + let Some(state) = self.channels.get(&channel) else { session.channel_failure(channel)?; return Ok(()); }; - if !begin_main_attachment(&main_session, self.channels.contains_key(&channel)) { - session.channel_failure(channel)?; - return Ok(()); - } + // Supervisor diagnostics bypass the PTY's output processing. SSH + // puts terminal clients in raw mode, so send the carriage return too. + let line_ending = if state.pty_request.is_some() { + "\r\n" + } else { + "\n" + }; + let main_session = self + .main_session + .clone() + .ok_or("no main session is available") + .and_then(|main_session| { + main_session.begin_terminal_attachment()?; + Ok(main_session) + }); + let main_session = match main_session { + Ok(main_session) => main_session, + Err(error) => { + // The subsystem is supported, but its terminal is unavailable. + // A bare request failure hides this reason in OpenSSH clients. + session.channel_success(channel)?; + session.extended_data( + channel, + 1, + format!( + "openshell: cannot connect to the canonical main process: {error}{line_ending}" + ) + .into_bytes(), + )?; + session.exit_status_request(channel, 1)?; + session.eof(channel)?; + session.close(channel)?; + return Ok(()); + } + }; let state = self .channels .get_mut(&channel) @@ -680,7 +711,7 @@ impl russh::server::Handler for SshHandler { .extended_data( channel, 1, - format!("openshell: {error}; attached read-only\n").into_bytes(), + format!("openshell: {error}; attached read-only; press Ctrl-C to exit{line_ending}").into_bytes(), ) .await; } @@ -789,6 +820,13 @@ impl russh::server::Handler for SshHandler { warn!("data on unknown channel {channel:?}"); return Ok(()); }; + // A viewer has no process stdin to interrupt. Ctrl-C closes only its + // attachment; the input owner's Ctrl-C still reaches the process. + if state.main_attached && state.main_input_owner.is_none() && data.contains(&0x03) { + self.close_main_attachment(channel, session.handle(), None) + .await; + return Ok(()); + } let (forward, detach) = if state.main_attached { filter_main_detach_sequence(&mut state.main_detach_prefix_pending, data) } else { @@ -1052,10 +1090,6 @@ impl SshHandler { } } -fn begin_main_attachment(main_session: &MainSession, channel_exists: bool) -> bool { - channel_exists && main_session.begin_terminal_attachment().is_ok() -} - async fn send_main_output(handle: &Handle, channel: ChannelId, event: MainOutput) -> bool { match event { MainOutput::Stdout(data) => handle.data(channel, data).await.is_ok(), @@ -1226,6 +1260,12 @@ mod tests { } async fn authenticated_test_client() -> russh::client::Handle { + main_test_client(Some(MainSession::inert())).await + } + + async fn main_test_client( + main_session: Option>, + ) -> russh::client::Handle { let host_key = { let mut rng = rand::rng(); PrivateKey::random(&mut rng, Algorithm::Ed25519).expect("host key") @@ -1239,7 +1279,7 @@ mod tests { let handler = SshHandler::new( Arc::new(TestLoopbackConnector), Arc::new(RejectingExec), - Some(MainSession::inert()), + main_session, ); let (server_stream, client_stream) = tokio::io::duplex(64 * 1024); tokio::spawn(async move { @@ -1373,19 +1413,263 @@ mod tests { assert_eq!(&echoed, b"ping"); } + async fn next_main_event( + channel: &mut russh::Channel, + ) -> russh::ChannelMsg { + tokio::time::timeout(Duration::from_secs(5), channel.wait()) + .await + .expect("main attachment event timeout") + .expect("main attachment channel closed before expected event") + } + + async fn assert_main_rejection( + main_session: Option>, + terminal: bool, + message: &str, + ) { + let client = main_test_client(main_session).await; + let mut channel = client.channel_open_session().await.unwrap(); + if terminal { + channel + .request_pty(true, "xterm", 80, 24, 0, 0, &[]) + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + } + channel + .request_subsystem(true, "openshell-main") + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + match next_main_event(&mut channel).await { + russh::ChannelMsg::ExtendedData { data, ext: 1 } => { + assert_eq!(String::from_utf8_lossy(&data), message); + } + event => panic!("expected actionable stderr, got {event:?}"), + } + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::ExitStatus { exit_status: 1 } + )); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Eof + )); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Close + )); + } + #[tokio::test] - async fn main_attachment_accepts_declared_session_after_process_exit() { + async fn main_attachment_missing_session_reports_terminal_error() { + assert_main_rejection(None, false, "openshell: cannot connect to the canonical main process: no main session is available\n").await; + } + + #[tokio::test] + async fn main_attachment_finished_session_reports_terminal_error() { + let main_session = MainSession::inert(); + assert!(!main_session.finish(130, false).await); + assert_main_rejection(Some(main_session), false, "openshell: cannot connect to the canonical main process: canonical main process already finished\n").await; + } + + #[tokio::test] + async fn main_attachment_pty_rejections_return_cursor_to_start_of_line() { + assert_main_rejection(None, true, "openshell: cannot connect to the canonical main process: no main session is available\r\n").await; let main_session = MainSession::inert(); - assert!(main_session.finish(23, true).await); - assert!(main_session.finished()); + assert!(!main_session.finish(130, false).await); + assert_main_rejection(Some(main_session), true, "openshell: cannot connect to the canonical main process: canonical main process already finished\r\n").await; + } + + #[tokio::test] + async fn main_attachment_occupied_stdin_still_attaches_read_only() { + assert_read_only_attachment(false, "\n").await; + } + + #[tokio::test] + async fn main_attachment_read_only_pty_warning_returns_cursor_to_start_of_line() { + assert_read_only_attachment(true, "\r\n").await; + } - assert!(begin_main_attachment(&main_session, true)); + async fn assert_read_only_attachment(terminal: bool, line_ending: &str) { + let main_session = MainSession::inert(); + let (owner, _input) = main_session.acquire_input().unwrap(); + let client = main_test_client(Some(main_session.clone())).await; + let mut channel = client.channel_open_session().await.unwrap(); + if terminal { + channel + .request_pty(true, "xterm", 80, 24, 0, 0, &[]) + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + } + channel + .request_subsystem(true, "openshell-main") + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + match next_main_event(&mut channel).await { + russh::ChannelMsg::ExtendedData { data, ext: 1 } => { + assert_eq!( + String::from_utf8_lossy(&data), + format!( + "openshell: canonical main process already has an input owner; attached read-only; press Ctrl-C to exit{line_ending}" + ) + ); + } + event => panic!("expected read-only warning, got {event:?}"), + } + assert!(main_session.acquire_input().is_err()); + channel.data(&b"ignored\x03also ignored"[..]).await.unwrap(); + assert_viewer_closed(&mut channel).await; + assert!(!main_session.finished()); + assert!( + main_session.acquire_input().is_err(), + "viewer must not release the owner's input lease" + ); + main_session.release_input(owner); + } + + async fn assert_viewer_closed(channel: &mut russh::Channel) { + assert!(matches!( + next_main_event(channel).await, + russh::ChannelMsg::Eof + )); + assert!(matches!( + next_main_event(channel).await, + russh::ChannelMsg::ExitStatus { exit_status: 0 } + )); + assert!(matches!( + next_main_event(channel).await, + russh::ChannelMsg::Close + )); + } + + #[tokio::test] + async fn main_attachment_explicit_read_only_ctrl_c_exits_without_input_lease() { + let (main_session, mut input) = MainSession::inert_with_input(); + let client = main_test_client(Some(main_session.clone())).await; + let mut channel = client.channel_open_session().await.unwrap(); + channel + .set_env(true, "OPENSHELL_MAIN_READ_ONLY", "1") + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + channel + .request_subsystem(true, "openshell-main") + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + channel.data(&b"\x03"[..]).await.unwrap(); + assert_viewer_closed(&mut channel).await; + assert!(!main_session.finished()); + assert!(matches!( + input.try_recv(), + Err(tokio::sync::mpsc::error::TryRecvError::Empty) + )); + let (owner, _) = main_session + .acquire_input() + .expect("viewer leaves stdin available"); + main_session.release_input(owner); + } + + #[tokio::test] + async fn main_attachment_input_owner_ctrl_c_reaches_main_stdin() { + let (main_session, mut input) = MainSession::inert_with_input(); + let client = main_test_client(Some(main_session)).await; + let mut channel = client.channel_open_session().await.unwrap(); + channel + .request_subsystem(true, "openshell-main") + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + for data in [b"\x03".as_slice(), b"still attached".as_slice()] { + channel.data(data).await.unwrap(); + let forwarded = tokio::time::timeout(Duration::from_secs(5), input.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(forwarded, data); + } + channel.close().await.unwrap(); + } + + #[tokio::test] + async fn main_attachment_accepts_declared_session_after_process_exit() { + let (main_session, mut slave) = MainSession::terminal_for_test(); let mut output = main_session.subscribe(); + slave.write_all(b"retained output").unwrap(); + assert!(matches!( + tokio::time::timeout(Duration::from_secs(5), output.recv()).await.unwrap().unwrap(), + MainOutput::Stdout(data) if data == b"retained output"[..] + )); + drop(slave); + assert!( + tokio::time::timeout(Duration::from_secs(5), main_session.finish(23, true)) + .await + .unwrap() + ); + main_session.mark_terminal_reported(); + let client = main_test_client(Some(main_session)).await; + let mut channel = client.channel_open_session().await.unwrap(); + channel + .request_subsystem(true, "openshell-main") + .await + .unwrap(); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Success + )); + assert!( + matches!(next_main_event(&mut channel).await, russh::ChannelMsg::Data { data } if data == b"retained output"[..]) + ); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Eof + )); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::ExitStatus { exit_status: 23 } + )); + assert!(matches!( + next_main_event(&mut channel).await, + russh::ChannelMsg::Close + )); + } + + #[tokio::test] + async fn main_attachment_unknown_subsystem_still_fails_request() { + let client = main_test_client(None).await; + let mut channel = client.channel_open_session().await.unwrap(); + channel + .request_subsystem(true, "unknown-subsystem") + .await + .unwrap(); assert!(matches!( - output.recv().await.expect("retained terminal status"), - MainOutput::Exit(23) + next_main_event(&mut channel).await, + russh::ChannelMsg::Failure )); - main_session.end_terminal_attachment(); } #[cfg(unix)] diff --git a/docs/sandboxes/manage-sandboxes.mdx b/docs/sandboxes/manage-sandboxes.mdx index 3cf3fdf838..a5dc0880c0 100644 --- a/docs/sandboxes/manage-sandboxes.mdx +++ b/docs/sandboxes/manage-sandboxes.mdx @@ -314,7 +314,8 @@ output. One attachment owns stdin at a time. Use `sandbox exec --tty -- Press `Ctrl-P`, then `Ctrl-Q` to disconnect without terminating the main process. `Ctrl-C` retains its normal terminal behavior and interrupts the -foreground process. +foreground process. For read-only attachments, `Ctrl-C` only exits the +current viewer. Launch VS Code or Cursor directly into the sandbox workspace: diff --git a/e2e/rust/tests/sandbox_lifecycle.rs b/e2e/rust/tests/sandbox_lifecycle.rs index 3b1883f11f..de50bc2d34 100644 --- a/e2e/rust/tests/sandbox_lifecycle.rs +++ b/e2e/rust/tests/sandbox_lifecycle.rs @@ -652,10 +652,43 @@ async fn canonical_main_disconnect_reconnect_replays_history_for_same_process() "read-only attachment should observe output: {observer_output_line}" ); + observer + .stdin + .as_mut() + .expect("observer stdin") + .write_all(b"\x03") + .await + .expect("send Ctrl-C to viewer"); + let observer_status = tokio::time::timeout(Duration::from_secs(30), observer.wait()) + .await + .expect("Ctrl-C should exit viewer") + .expect("wait for viewer"); + assert!(observer_status.success(), "viewer should exit successfully"); + + owner + .stdin + .as_mut() + .expect("owner stdin") + .write_all(b"owner-after-viewer-exit\n") + .await + .expect("send input after viewer exits"); + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let line = owner_lines + .next_line() + .await + .expect("read owner output after viewer exits") + .expect("main process should remain running"); + if line.contains("input=owner-after-viewer-exit") { + break; + } + } + }) + .await + .expect("owner should retain working stdin after viewer exits"); + owner.kill().await.expect("disconnect input owner"); owner.wait().await.expect("wait for input owner disconnect"); - observer.kill().await.expect("disconnect observer"); - observer.wait().await.expect("wait for observer disconnect"); let (mut reconnect, replay) = reconnect_with_input_ownership(&sandbox.name).await; diff --git a/skills/openshell-cli/SKILL.md b/skills/openshell-cli/SKILL.md index 89c707d424..eea2867a23 100644 --- a/skills/openshell-cli/SKILL.md +++ b/skills/openshell-cli/SKILL.md @@ -353,14 +353,20 @@ openshell sandbox connect my-sandbox --editor vscode Attaches to the sandbox's existing canonical main process. Disconnecting leaves that process running; reconnecting targets the same process instance and replays recent output. Use `sandbox exec --tty -- /bin/bash -l` for a new shell. Press -`Ctrl-P`, then `Ctrl-Q` to disconnect without terminating main. `Ctrl-C` retains -its normal terminal behavior and interrupts the foreground process. Configure -VS Code Remote-SSH with: +`Ctrl-P`, then `Ctrl-Q` to disconnect without terminating main. When you own +stdin, `Ctrl-C` interrupts the foreground process. In a read-only attachment, +`Ctrl-C` exits the viewer and leaves main and other attachments running. +Configure VS Code Remote-SSH with: ```bash openshell sandbox ssh-config my-sandbox >> ~/.ssh/config ``` +If `connect` reports `canonical main process already finished`, inspect the +result with `sandbox get`. A pending +foreground attachment can still retrieve retained output in `Completed` or +`Error`; phase alone does not determine whether attachment is available. + ### Upload and download files ```bash From a408f5dd083a9a4be259bb44478598002f9387a6 Mon Sep 17 00:00:00 2001 From: Mrunal Patel Date: Wed, 23 Sep 2026 18:30:24 +0000 Subject: [PATCH 07/10] chore(kubernetes): update Agent Sandbox to v1.0.3 (#3578) Signed-off-by: Mrunal Patel --- .github/workflows/branch-e2e.yml | 2 +- .github/workflows/e2e-kubernetes-test.yml | 2 +- .github/workflows/release-canary.yml | 2 +- architecture/compute-runtimes.md | 4 +++- deploy/helm/openshell/README.md | 2 +- deploy/helm/openshell/README.md.gotmpl | 2 +- docs/kubernetes/setup.mdx | 2 +- e2e/support/install-agent-sandbox.sh | 11 +++++++++-- e2e/with-kube-gateway.sh | 2 +- tasks/scripts/helm-k3s-local.sh | 10 +++++----- 10 files changed, 24 insertions(+), 15 deletions(-) diff --git a/.github/workflows/branch-e2e.yml b/.github/workflows/branch-e2e.yml index bac688c88b..2935042776 100644 --- a/.github/workflows/branch-e2e.yml +++ b/.github/workflows/branch-e2e.yml @@ -354,7 +354,7 @@ jobs: matrix: include: - agent_sandbox_api: v1beta1 - agent_sandbox_version: v0.5.0 + agent_sandbox_version: v1.0.3 - agent_sandbox_api: v1alpha1 agent_sandbox_version: v0.4.6 permissions: diff --git a/.github/workflows/e2e-kubernetes-test.yml b/.github/workflows/e2e-kubernetes-test.yml index 4aaa932644..0d018187a8 100644 --- a/.github/workflows/e2e-kubernetes-test.yml +++ b/.github/workflows/e2e-kubernetes-test.yml @@ -36,7 +36,7 @@ on: description: "Agent Sandbox release to install before OpenShell" required: false type: string - default: "v0.5.0" + default: "v1.0.3" e2e-task: description: "mise task to run for the Kubernetes e2e job" required: false diff --git a/.github/workflows/release-canary.yml b/.github/workflows/release-canary.yml index 7a458e7ef7..a8a8df7b51 100644 --- a/.github/workflows/release-canary.yml +++ b/.github/workflows/release-canary.yml @@ -276,7 +276,7 @@ jobs: RELEASE_NAME: openshell RELEASE_NAMESPACE: openshell KIND_GATEWAY_NAME: kind - AGENT_SANDBOX_VERSION: v0.5.0 + AGENT_SANDBOX_VERSION: v1.0.3 steps: - name: Checkout Agent Sandbox helper uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 diff --git a/architecture/compute-runtimes.md b/architecture/compute-runtimes.md index 3f28c3f0a4..53d24aa0cc 100644 --- a/architecture/compute-runtimes.md +++ b/architecture/compute-runtimes.md @@ -439,7 +439,9 @@ database-backed installs can render a Deployment with `workload.kind=deployment` HA deployments must point `server.externalDbSecret` at an operator-managed PostgreSQL database. Agent Sandbox CRDs and controller lifecycle remain operator-owned; the chart can optionally preflight for a served supported API -but does not install the cluster-scoped dependency. +but does not install the cluster-scoped dependency. OpenShell's Kubernetes test +clusters install the upstream core manifest; Agent Sandbox extensions are not +required by the gateway. Standalone local deployments start the gateway with a selected runtime such as Docker, Podman, or VM. The CLI can register multiple gateways and switch between them without changing the sandbox architecture. diff --git a/deploy/helm/openshell/README.md b/deploy/helm/openshell/README.md index 70f0a56940..e2f7b3e6af 100644 --- a/deploy/helm/openshell/README.md +++ b/deploy/helm/openshell/README.md @@ -22,7 +22,7 @@ multiple pre-provisioned workspace namespaces. The Kubernetes Agent Sandbox CRDs and controller must be installed on the cluster before deploying OpenShell. Install them with: ```shell -kubectl apply -f https://github.com/kubernetes-sigs/agent-sandbox/releases/latest/download/manifest.yaml +kubectl apply -f https://github.com/kubernetes-sigs/agent-sandbox/releases/latest/download/sandbox.yaml ``` The chart does not install this cluster-scoped dependency. By default, it diff --git a/deploy/helm/openshell/README.md.gotmpl b/deploy/helm/openshell/README.md.gotmpl index 07af6e78b4..59e8118f3b 100644 --- a/deploy/helm/openshell/README.md.gotmpl +++ b/deploy/helm/openshell/README.md.gotmpl @@ -22,7 +22,7 @@ multiple pre-provisioned workspace namespaces. The Kubernetes Agent Sandbox CRDs and controller must be installed on the cluster before deploying OpenShell. Install them with: ```shell -kubectl apply -f https://github.com/kubernetes-sigs/agent-sandbox/releases/latest/download/manifest.yaml +kubectl apply -f https://github.com/kubernetes-sigs/agent-sandbox/releases/latest/download/sandbox.yaml ``` The chart does not install this cluster-scoped dependency. By default, it diff --git a/docs/kubernetes/setup.mdx b/docs/kubernetes/setup.mdx index 494e14aa22..4a3539c450 100644 --- a/docs/kubernetes/setup.mdx +++ b/docs/kubernetes/setup.mdx @@ -33,7 +33,7 @@ OpenShell uses the [Agent Sandbox](https://agent-sandbox.sigs.k8s.io) Kubernetes Apply the latest release manifest: ```shell -kubectl apply -f https://github.com/kubernetes-sigs/agent-sandbox/releases/latest/download/manifest.yaml +kubectl apply -f https://github.com/kubernetes-sigs/agent-sandbox/releases/latest/download/sandbox.yaml ``` This creates the `agent-sandbox-system` namespace, installs the `sandboxes.agents.x-k8s.io` CRD, and starts the controller. diff --git a/e2e/support/install-agent-sandbox.sh b/e2e/support/install-agent-sandbox.sh index 7b33a53eea..2ff93c12af 100644 --- a/e2e/support/install-agent-sandbox.sh +++ b/e2e/support/install-agent-sandbox.sh @@ -6,7 +6,14 @@ # context arguments (for example, --context kind-e2e) as script arguments. set -euo pipefail -agent_sandbox_version="${AGENT_SANDBOX_VERSION:-v0.5.0}" +agent_sandbox_version="${AGENT_SANDBOX_VERSION:-v1.0.3}" + +# Agent Sandbox renamed its core release manifest after v0.5.1. Keep the +# v1alpha1 compatibility lane on the asset published with that release. +case "${agent_sandbox_version}" in + v0.[0-4].*|v0.5.[01]) agent_sandbox_manifest="manifest.yaml" ;; + *) agent_sandbox_manifest="sandbox.yaml" ;; +esac wait_for_agent_sandbox_crd() { local deadline @@ -32,7 +39,7 @@ wait_for_agent_sandbox_crd() { echo "Installing agent-sandbox CRDs and controller (${agent_sandbox_version})..." agent_sandbox_base="https://github.com/kubernetes-sigs/agent-sandbox/releases/download/${agent_sandbox_version}" -kubectl "$@" apply -f "${agent_sandbox_base}/manifest.yaml" +kubectl "$@" apply -f "${agent_sandbox_base}/${agent_sandbox_manifest}" wait_for_agent_sandbox_crd "$@" kubectl "$@" -n agent-sandbox-system rollout status \ deployment/agent-sandbox-controller --timeout=300s diff --git a/e2e/with-kube-gateway.sh b/e2e/with-kube-gateway.sh index 8c6659d638..cb9a0f8820 100755 --- a/e2e/with-kube-gateway.sh +++ b/e2e/with-kube-gateway.sh @@ -75,7 +75,7 @@ source "${ROOT}/e2e/support/gateway-common.sh" # Upstream agent-sandbox release. The Kubernetes driver supports the v1beta1 # Sandbox API introduced in v0.5.0 and falls back to v1alpha1 for v0.4.6 # clusters. Override this env var to exercise the v1alpha1 controller release. -AGENT_SANDBOX_VERSION="${AGENT_SANDBOX_VERSION:-v0.5.0}" +AGENT_SANDBOX_VERSION="${AGENT_SANDBOX_VERSION:-v1.0.3}" e2e_preserve_mise_dirs e2e_align_docker_host_with_cli_context diff --git a/tasks/scripts/helm-k3s-local.sh b/tasks/scripts/helm-k3s-local.sh index e24893e7d5..2a5aaf272d 100755 --- a/tasks/scripts/helm-k3s-local.sh +++ b/tasks/scripts/helm-k3s-local.sh @@ -32,11 +32,11 @@ HOST_LB_PORT="${HELM_K3S_LB_HOST_PORT:-8080}" DEFAULT_SANDBOX_PRELOAD_IMAGE="nvcr.io/nvidia/base/ubuntu:24.04" PRELOAD_SANDBOX_IMAGE="${HELM_K3S_PRELOAD_SANDBOX_IMAGE-${DEFAULT_SANDBOX_PRELOAD_IMAGE}}" -# Upstream agent-sandbox release pinned for both CRDs/controller and extensions. +# Upstream agent-sandbox release pinned for the core CRDs and controller. # The Kubernetes driver supports the v1beta1 Sandbox API introduced in v0.5.0 # and falls back to v1alpha1 for v0.4.6 clusters. Override this env var to # exercise the v1alpha1 controller release. -AGENT_SANDBOX_VERSION="${AGENT_SANDBOX_VERSION:-v0.5.0}" +AGENT_SANDBOX_VERSION="${AGENT_SANDBOX_VERSION:-v1.0.3}" # Local OTLP receiver and trace UI. The current Aspire implementation accepts # OTLP/gRPC directly, so the development cluster needs only one deployment. @@ -164,9 +164,9 @@ merge_kubeconfig() { apply_base_manifests() { require_kubectl - local base="https://github.com/kubernetes-sigs/agent-sandbox/releases/download/${AGENT_SANDBOX_VERSION}" - echo "Applying agent-sandbox manifest (${AGENT_SANDBOX_VERSION})..." - kubectl --kubeconfig="${KUBECONFIG_TARGET}" apply -f "${base}/manifest.yaml" + AGENT_SANDBOX_VERSION="${AGENT_SANDBOX_VERSION}" \ + bash "${ROOT}/e2e/support/install-agent-sandbox.sh" \ + --kubeconfig="${KUBECONFIG_TARGET}" } install_trace_collector() { From 95632406cdfaec6cbb71027d62d01c425a4e3bbe Mon Sep 17 00:00:00 2001 From: krishicks Date: Wed, 23 Sep 2026 18:45:07 +0000 Subject: [PATCH 08/10] fix(server): make HA sandbox create and HA e2e tests reliable (#3635) * fix(server): retry runtime identity persistence on sandbox create With multiple gateway replicas, another replica can update a new sandbox record between the create path's read and its compare-and-swap write of the runtime identity. The create path made one attempt, so the conflict failed the request and deleted the backend sandbox. Persist the identity through the same retrying helper that start and startup recovery use, which rereads the record and retries while the sandbox stays in the same generation and a Provisioning or Ready phase. Signed-off-by: Kris Hicks * ci(e2e): run Kubernetes HA tests one at a time The two HA tests scale and roll the shared gateway Deployment. Their in-process lock does not apply under nextest, which runs each test in its own process, so one test could delete a gateway pod while the other was executing through it. Put both tests in a nextest test group limited to one thread in the e2e-kubernetes profile. The override matches test names because a binary() filter fails in workspaces that lack the HA test binary. Signed-off-by: Kris Hicks --------- Signed-off-by: Kris Hicks --- .config/nextest.toml | 10 +++ crates/openshell-server/src/compute/mod.rs | 89 +++++++++++++++++----- 2 files changed, 81 insertions(+), 18 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index b8267fb5f5..19b4c21769 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -17,6 +17,16 @@ failure-output = "immediate-final" # Flag slow tests after 60s and terminate them after five minutes. slow-timeout = { period = "60s", terminate-after = 5 } +# The HA tests scale and roll the shared gateway Deployment, so they cannot +# overlap. Nextest runs each test in its own process. The filter matches test +# names because binary() fails in workspaces without the HA test binary. +[test-groups] +kubernetes-ha = { max-threads = 1 } + +[[profile.e2e-kubernetes.overrides]] +filter = "test(/gateway_(scale_and_rollout|pod_rolls)$/)" +test-group = "kubernetes-ha" + # Relative to the profile store dir (`e2e/rust/target/nextest/e2e-kubernetes/`). [profile.e2e-kubernetes.junit] path = "../../../../../results/e2e-kubernetes.xml" diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 2a5ea8e4f2..b578333560 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -1128,21 +1128,14 @@ impl ComputeRuntime { .await); } if self.supports_sandbox_authentication() { - let driver_name = self.configured_driver_name().to_string(); let persisted = self - .store - .update_message_cas::(&sandbox_id, 0, move |sandbox| { - if let Some(metadata) = sandbox.metadata.as_mut() { - metadata.annotations.insert( - COMPUTE_DRIVER_ANNOTATION.to_string(), - driver_name.clone(), - ); - metadata.annotations.insert( - COMPUTE_RUNTIME_IDENTITY_ANNOTATION.to_string(), - runtime_identity.clone(), - ); - } - }) + .persist_runtime_binding( + &sandbox_id, + &sandbox, + self.configured_driver_name(), + &runtime_identity, + &[SandboxPhase::Provisioning, SandboxPhase::Ready], + ) .await; sandbox = match persisted { Ok(sandbox) => sandbox, @@ -1790,7 +1783,7 @@ impl ComputeRuntime { let latest = if self.supports_sandbox_authentication() { let driver_name = self.configured_driver_name().to_string(); let persisted = self - .persist_start_runtime_binding( + .persist_runtime_binding( &sandbox_id, &starting, &driver_name, @@ -1841,7 +1834,7 @@ impl ComputeRuntime { } } - async fn persist_start_runtime_binding( + async fn persist_runtime_binding( &self, sandbox_id: &str, starting: &Sandbox, @@ -1900,7 +1893,7 @@ impl ComputeRuntime { sandbox_id, attempt, expected_resource_version, - "Retrying runtime identity persistence after concurrent start progress" + "Retrying runtime identity persistence after a concurrent sandbox update" ); } Err(error) => return Err(error.to_string()), @@ -3072,7 +3065,7 @@ impl ComputeRuntime { continue; } match self - .persist_start_runtime_binding( + .persist_runtime_binding( &sandbox_id, &sandbox, self.configured_driver_name(), @@ -7528,6 +7521,66 @@ mod tests { assert_eq!(driver.delete_calls(), 0); } + #[tokio::test] + async fn create_retries_runtime_binding_after_concurrent_replica_write() { + let directory = tempfile::tempdir().expect("temporary database directory"); + let database_url = format!("sqlite://{}", directory.path().join("gateway.db").display()); + let store = Arc::new(Store::connect(&database_url).await.expect("connect store")); + let pool = sqlx::SqlitePool::connect(&database_url) + .await + .expect("connect conflict injector"); + sqlx::query("CREATE TABLE injected_conflicts (id TEXT PRIMARY KEY)") + .execute(&pool) + .await + .expect("create conflict marker table"); + // Another replica writes the record between the binding read and its + // CAS write, once. + sqlx::query( + "CREATE TRIGGER inject_replica_write \ + BEFORE UPDATE OF payload ON objects \ + WHEN instr(NEW.payload, CAST('new-runtime-identity' AS BLOB)) > 0 \ + AND NOT EXISTS (SELECT 1 FROM injected_conflicts WHERE id = OLD.id) \ + BEGIN \ + INSERT INTO injected_conflicts (id) VALUES (OLD.id); \ + UPDATE objects SET resource_version = resource_version + 1 \ + WHERE object_type = OLD.object_type AND id = OLD.id; \ + SELECT RAISE(IGNORE); \ + END", + ) + .execute(&pool) + .await + .expect("install conflict trigger"); + pool.close().await; + + let driver = ControlledDriver::new(); + driver.set_runtime_identity("new-runtime-identity"); + let mut runtime = test_runtime(driver.clone()).await; + runtime.store = store; + enable_runtime_identity_binding(&mut runtime); + let sandbox = sandbox_record( + "sb-create-replica-race", + "create-replica-race", + SandboxPhase::Provisioning, + ); + + runtime + .create_sandbox(sandbox.clone(), None, false) + .await + .expect("create must retry the runtime binding after a concurrent write"); + + let stored = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); + assert_eq!( + stored.metadata.unwrap().annotations[COMPUTE_RUNTIME_IDENTITY_ANNOTATION], + "new-runtime-identity" + ); + assert_eq!(driver.delete_calls(), 0); + } + #[tokio::test] async fn empty_start_runtime_identity_stops_backend_and_restores_record() { let driver = ControlledDriver::new(); From 52aac37866aa2a2ffba7c4a198d7fb3c20dc5bbe Mon Sep 17 00:00:00 2001 From: Drew Newberry Date: Wed, 23 Sep 2026 12:01:46 -0700 Subject: [PATCH 09/10] fix(network): honor HTTP response connection closure (#3581) * fix(network): honor HTTP response connection closure Signed-off-by: Drew Newberry * test(network): fix EOF fixture socket setup Signed-off-by: Drew Newberry * test(network): keep pipeline probe response reusable Signed-off-by: Drew Newberry --------- Signed-off-by: Drew Newberry --- architecture/sandbox.md | 6 + .../src/l7/rest.rs | 136 ++++++++++++++++-- .../src/l7/rest/http_response.rs | 49 ++++--- .../src/proxy/relay.rs | 95 ++++++++++++ docs/reference/policy-schema.mdx | 5 + e2e/rust/tests/host_gateway_alias.rs | 72 ++++++++++ e2e/rust/tests/proxy_egress_pipeline.rs | 5 +- 7 files changed, 336 insertions(+), 32 deletions(-) diff --git a/architecture/sandbox.md b/architecture/sandbox.md index e269634547..f68c1b67a7 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -280,6 +280,12 @@ the shared raw byte relay after the existing adapter gates. Forward HTTP retains its guarded single-request relay while sharing authorization, request context, policy-pinning, and destination boundaries. Adapter-specific response and OCSF event shapes remain at the protocol boundary. +HTTP response framing and connection persistence are separate decisions. After +forwarding a complete closing response (explicit `Connection: close` or HTTP/1.0 +without keep-alive), the relay flushes and shuts down downstream writes before +ending the exchange, including TLS close notification. Response middleware +preserves this lifetime rule; persistent responses remain eligible for reuse. + An explicit `protocol: tcp` endpoint with a valid DNS hostname opts into native DNS and transparent TCP when the selected runtime advertises that substrate. Hostless `allowed_ips` and literal-IP selectors remain available only to the diff --git a/crates/openshell-supervisor-network/src/l7/rest.rs b/crates/openshell-supervisor-network/src/l7/rest.rs index 399db42301..4592600231 100644 --- a/crates/openshell-supervisor-network/src/l7/rest.rs +++ b/crates/openshell-supervisor-network/src/l7/rest.rs @@ -3663,7 +3663,7 @@ mod tests { const VALID_WS_ACCEPT: &str = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="; const TEXT_OPCODE: u8 = 0x1; - #[derive(Clone, Copy)] + #[derive(Debug, Clone, Copy)] enum ResponseRelayScript { HeadersOnly, WholeBody, @@ -7154,7 +7154,7 @@ mod tests { ResponseRelayScript::HeadersOnly, ) .await; - assert!(matches!(outcome.unwrap(), RelayOutcome::Reusable)); + assert!(matches!(outcome.unwrap(), RelayOutcome::Consumed)); let (outcome, delivered) = run_response_middleware_relay( b"HTTP/1.1 200 OK\r\n\r\n", @@ -7931,6 +7931,117 @@ mod tests { assert!(received_str.contains("hello")); } + #[tokio::test] + async fn response_persistence_is_independent_of_framing_and_middleware() { + for script in [ + None, + Some(ResponseRelayScript::HeadersOnly), + Some(ResponseRelayScript::Stream), + Some(ResponseRelayScript::WholeBody), + ] { + for (method, response, closes) in [ + ( + "GET", + "HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\nhello", + true, + ), + ( + "GET", + "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 5\r\n\r\nhello", + true, + ), + ( + "GET", + "HTTP/1.1 200 OK\r\nConnection: close\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n", + true, + ), + ("HEAD", "HTTP/1.0 200 OK\r\nContent-Length: 5\r\n\r\n", true), + ("GET", "HTTP/1.0 204 No Content\r\n\r\n", true), + ( + "GET", + "HTTP/1.1 304 Not Modified\r\nConnection: close\r\n\r\n", + true, + ), + ( + "GET", + "HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello", + false, + ), + ( + "GET", + "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n0\r\n\r\n", + false, + ), + ( + "HEAD", + "HTTP/1.0 200 OK\r\nConnection: keep-alive\r\nContent-Length: 5\r\n\r\n", + false, + ), + ] { + // Body-inspecting scripts reject bodiless responses at preflight. + if (method == "HEAD" + || response.contains("204 No Content") + || response.contains("304 Not Modified")) + && matches!( + script, + Some(ResponseRelayScript::Stream | ResponseRelayScript::WholeBody) + ) + { + continue; + } + let fixture = script.map(response_middleware_fixture); + let middleware = fixture + .as_ref() + .map(|(runner, chain)| response_middleware_context(runner, chain, method)); + let mut upstream = response.as_bytes(); + let (mut reader, mut writer) = tokio::io::duplex(4096); + let outcome = relay_response( + method, + &mut upstream, + &mut writer, + RelayResponseOptions::default(), + middleware, + ) + .await + .unwrap(); + assert_eq!( + matches!(outcome, RelayOutcome::Consumed), + closes, + "{response}" + ); + let mut delivered = Vec::new(); + if closes { + tokio::time::timeout( + std::time::Duration::from_secs(2), + reader.read_to_end(&mut delivered), + ) + .await + .unwrap_or_else(|error| panic!("{script:?} {method} {response:?}: {error}")) + .unwrap(); + } else { + // A second write demonstrates that persistent output is still open. + writer.write_all(b"next response").await.unwrap(); + writer.shutdown().await.unwrap(); + reader.read_to_end(&mut delivered).await.unwrap(); + assert!(delivered.ends_with(b"next response")); + } + if let Some(script) = script { + let expected = match script { + ResponseRelayScript::HeadersOnly => "hello", + ResponseRelayScript::Stream => "HELLO", + ResponseRelayScript::WholeBody => "whole:hello", + _ => unreachable!(), + }; + if response.ends_with("hello") { + assert!(String::from_utf8_lossy(&delivered).contains(expected)); + } + } else { + assert!(delivered.starts_with(response.as_bytes())); + } + } + } + } + #[tokio::test] async fn relay_response_connection_close_with_content_length() { let response = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello"; @@ -7956,18 +8067,19 @@ mod tests { .expect("relay must not deadlock"); let outcome = result.expect("relay_response should succeed"); - // With explicit framing, Connection: close is still reported as reusable - // so the relay loop continues. The *next* upstream write will fail and - // exit the loop via the normal error path. - assert!( - matches!(outcome, RelayOutcome::Reusable), - "explicit framing keeps loop alive despite Connection: close" - ); + assert!(matches!(outcome, RelayOutcome::Consumed)); - client_write.shutdown().await.unwrap(); + // Keep the relay-side stream alive: EOF must come from shutdown, + // not from dropping the stream or the test closing it manually. let mut received = Vec::new(); - client_read.read_to_end(&mut received).await.unwrap(); - assert!(String::from_utf8_lossy(&received).contains("hello")); + tokio::time::timeout( + std::time::Duration::from_secs(2), + client_read.read_to_end(&mut received), + ) + .await + .expect("closing response must deliver EOF") + .unwrap(); + assert_eq!(received, response); } #[tokio::test] diff --git a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs index 9a04244282..8b637ef263 100644 --- a/crates/openshell-supervisor-network/src/l7/rest/http_response.rs +++ b/crates/openshell-supervisor-network/src/l7/rest/http_response.rs @@ -178,7 +178,11 @@ where )) .await? { - return Ok(outcome); + return if matches!(outcome, RelayOutcome::Consumed) { + finish_response(client, true).await + } else { + Ok(outcome) + }; } // Bodiless responses (HEAD, 1xx, 204, 304): forward headers only, skip body @@ -187,12 +191,7 @@ where .write_all(&buf[..header_end]) .await .into_diagnostic()?; - client.flush().await.into_diagnostic()?; - return if server_wants_close { - Ok(RelayOutcome::Consumed) - } else { - Ok(RelayOutcome::Reusable) - }; + return finish_response(client, server_wants_close || http_10_closes_by_default).await; } // No explicit framing (no Content-Length, no Transfer-Encoding). @@ -261,13 +260,24 @@ where "relay_response complete (explicit framing)" ); - // When body framing is explicit (Content-Length / Chunked), always report - // the connection as reusable so the relay loop continues. If the server - // sent `Connection: close`, the *next* upstream write will fail and the - // loop will exit via the normal error path. Exiting early here would - // tear down the CONNECT tunnel before the client can detect the close, - // causing ~30 s retry delays in clients like `gh`. - Ok(RelayOutcome::Reusable) + finish_response(client, server_wants_close || http_10_closes_by_default).await +} + +/// Body framing determines when delivery finishes, not whether another +/// request is permitted. Signal EOF (including TLS `close_notify`) before the +/// caller tears down a closing CONNECT tunnel; waiting for another request +/// deadlocks clients that are themselves waiting for EOF. +async fn finish_response(client: &mut C, close: bool) -> Result +where + C: AsyncWrite + Unpin, +{ + client.flush().await.into_diagnostic()?; + if close { + client.shutdown().await.into_diagnostic()?; + Ok(RelayOutcome::Consumed) + } else { + Ok(RelayOutcome::Reusable) + } } #[allow(clippy::too_many_arguments)] @@ -758,8 +768,9 @@ where } client.flush().await.into_diagnostic()?; Ok(Some( - if (committed && close_delimited_output) - || (matches!(body_length, BodyLength::None) && (server_wants_close || event_stream)) + if server_wants_close + || (committed && close_delimited_output) + || (matches!(body_length, BodyLength::None) && event_stream) { RelayOutcome::Consumed } else { @@ -827,7 +838,11 @@ where BodyLength::None => {} } client.flush().await.into_diagnostic()?; - Ok(RelayOutcome::Reusable) + Ok(if server_wants_close { + RelayOutcome::Consumed + } else { + RelayOutcome::Reusable + }) } fn emit_http_response_middleware_invocations( diff --git a/crates/openshell-supervisor-network/src/proxy/relay.rs b/crates/openshell-supervisor-network/src/proxy/relay.rs index 4c89e5e382..77055245a5 100644 --- a/crates/openshell-supervisor-network/src/proxy/relay.rs +++ b/crates/openshell-supervisor-network/src/proxy/relay.rs @@ -374,6 +374,101 @@ mod tests { } } + async fn assert_response_lifecycle(mut caller: C, mut client: P) + where + C: AsyncRead + AsyncWrite + Unpin, + P: AsyncRead + AsyncWrite + Unpin + Send, + { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let (mut upstream, mut server) = tokio::io::duplex(1024); + let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + let decision = decision(engine.current_generation()); + let request = request_context(); + let context = prepare_http_relay(None, &engine, &decision, &request).unwrap(); + let persistent = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello"; + // Larger than either transport buffer: closing must drain the body. + let body = vec![b'x'; 64 * 1024]; + let mut closing = format!( + "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ) + .into_bytes(); + closing.extend_from_slice(&body); + let relay = Box::pin(relay_http_stream(&mut client, &mut upstream, context)); + let serve = async { + for response in [persistent.as_slice(), closing.as_slice()] { + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + request.push(server.read_u8().await.unwrap()); + assert!(request.len() < 4096); + } + assert!(request.starts_with(b"GET / HTTP/1.1\r\n")); + server.write_all(response).await.unwrap(); + server.flush().await.unwrap(); + } + // Retain the upstream socket: response headers decide persistence. + }; + let receive = async { + let request = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"; + caller.write_all(request).await.unwrap(); + caller.flush().await.unwrap(); + let mut first = vec![0; persistent.len()]; + caller.read_exact(&mut first).await.unwrap(); + assert_eq!(first, persistent); + // A real second exchange verifies reuse, without a timing assertion. + caller.write_all(request).await.unwrap(); + caller.flush().await.unwrap(); + let mut second = Vec::new(); + // TLS must return clean EOF, not UnexpectedEof from a dropped socket. + caller.read_to_end(&mut second).await.unwrap(); + assert_eq!(second, closing); + }; + tokio::time::timeout(std::time::Duration::from_secs(5), async { + let (result, (), ()) = tokio::join!(relay, serve, receive); + result.unwrap(); + }) + .await + .expect("response delivery and EOF must not await another request"); + } + + #[tokio::test] + async fn http_relay_reuses_then_closes_after_complete_response() { + let (caller, client) = tokio::io::duplex(1024); + assert_response_lifecycle(caller, client).await; + } + + #[tokio::test] + async fn tls_http_relay_reuses_then_sends_close_notify_after_complete_response() { + use crate::l7::tls::{CertCache, ProxyTlsState, SandboxCa, tls_terminate_client}; + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let ca = SandboxCa::generate().unwrap(); + let mut roots = rustls::RootCertStore::empty(); + for cert in rustls_pemfile::certs(&mut ca.cert_pem().as_bytes()) { + roots.add(cert.unwrap()).unwrap(); + } + let config = Arc::new( + rustls::ClientConfig::builder() + .with_root_certificates(roots) + .with_no_client_auth(), + ); + let state = ProxyTlsState::new(CertCache::new(ca), config.clone()); + let connector = tokio_rustls::TlsConnector::from(config); + let (caller, client) = tokio::io::duplex(1024); + let (caller, client) = tokio::time::timeout(std::time::Duration::from_secs(5), async { + tokio::join!( + connector.connect( + rustls::pki_types::ServerName::try_from("example.com").unwrap(), + caller + ), + tls_terminate_client(client, &state, "example.com"), + ) + }) + .await + .expect("TLS handshake must complete"); + assert_response_lifecycle(caller.unwrap(), client.unwrap()).await; + } + #[test] fn relay_without_route_pins_l4_decision_generation() { let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); diff --git a/docs/reference/policy-schema.mdx b/docs/reference/policy-schema.mdx index 3e1539b7b0..a72a21bb2c 100644 --- a/docs/reference/policy-schema.mdx +++ b/docs/reference/policy-schema.mdx @@ -267,6 +267,11 @@ access selects no preset. Unknown enum numbers are rejected before activation. - `deny_rules: []` (empty list) is rejected; remove it if no denials are needed. - `credential_signing` requires a resolvable AWS credential source before a sandbox policy can activate. Use an attached endpoint-bearing profile that declares `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` and covers the signed endpoint, or bind an attached endpointless profile that declares those keys with `credential_binding.provider`. +For proxied HTTP responses, OpenShell completes delivery and signals EOF when +the server sends `Connection: close` or an HTTP/1.0 response without keep-alive. +This also applies to HTTPS and responses processed by middleware. A +`Content-Length` or chunked body does not override the server's close decision. + Credential rewrite recognizes the canonical `openshell:resolve:env:KEY` placeholder form and whole-token provider-shaped aliases such as `provider-OPENSHELL-RESOLVE-ENV-API_TOKEN` when the referenced environment key exists in the configured provider credentials. Static provider placeholders also require the request host, port, and path to diff --git a/e2e/rust/tests/host_gateway_alias.rs b/e2e/rust/tests/host_gateway_alias.rs index e4ae811472..8d355f97fb 100644 --- a/e2e/rust/tests/host_gateway_alias.rs +++ b/e2e/rust/tests/host_gateway_alias.rs @@ -340,6 +340,78 @@ async fn sandbox_reaches_host_openshell_internal_via_host_gateway_alias() { ); } +#[tokio::test] +async fn sandbox_receives_eof_after_closing_http_response() { + for response in [ + "HTTP/1.0 200 OK\r\nContent-Length: 3\r\n\r\nOK\n", + "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 3\r\n\r\nOK\n", + "HTTP/1.1 200 OK\r\nConnection: close\r\nTransfer-Encoding: chunked\r\n\r\n3\r\nOK\n\r\n0\r\n\r\n", + ] { + let listener = TcpListener::bind(("0.0.0.0", 0)).await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = HostServer { + port, + task: tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.set_nodelay(true).expect("disable Nagle on fixture"); + let mut request = Vec::new(); + while !request.ends_with(b"\r\n\r\n") { + request.push(stream.read_u8().await.unwrap()); + assert!(request.len() < 4096); + } + stream.write_all(response.as_bytes()).await.unwrap(); + stream.shutdown().await.unwrap(); + }), + }; + let policy = write_policy(server.port).unwrap(); + let command = format!( + r#"set -eu +exec 3<>/dev/tcp/host.openshell.internal/{port} +printf 'GET / HTTP/1.1\r\nHost: host.openshell.internal:{port}\r\n\r\n' >&3 +while true; do + line= + if IFS= read -r -t 5 line <&3; then + printf '%s\n' "$line" + else + status=$? + [ "$status" -eq 1 ] || {{ echo EOF_TIMEOUT; exit 1; }} + [ -z "$line" ] || printf '%s\n' "$line" + break + fi +done +printf 'RESPONSE_EOF\n' +"#, + ); + let mut sandbox = SandboxGuard::create(&[ + "--policy", + policy.path().to_str().unwrap(), + "--no-auto-providers", + "--", + "/usr/bin/bash", + "-c", + &command, + ]) + .await + .expect("closing response must finish without a client timeout"); + assert!( + sandbox.create_output.lines().any(|line| line == "OK"), + "{}", + sandbox.create_output + ); + assert!( + sandbox.create_output.contains("RESPONSE_EOF"), + "{}", + sandbox.create_output + ); + assert!( + !sandbox.create_output.contains("EOF_TIMEOUT"), + "{}", + sandbox.create_output + ); + sandbox.cleanup().await; + } +} + #[tokio::test] async fn static_provider_credentials_are_bound_to_profile_endpoints() { let server = HostServer::start_with_auth_check("", Some("Bearer e2e-bound-secret")) diff --git a/e2e/rust/tests/proxy_egress_pipeline.rs b/e2e/rust/tests/proxy_egress_pipeline.rs index 9556c16429..09eda47615 100644 --- a/e2e/rust/tests/proxy_egress_pipeline.rs +++ b/e2e/rust/tests/proxy_egress_pipeline.rs @@ -614,10 +614,9 @@ impl PipelineProbeServer { } } observed.lock().unwrap().extend_from_slice(&request); + // Keep the first response reusable so the queued request reaches policy evaluation. let _ = stream - .write_all( - b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok", - ) + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") .await; }); } From 123d95e2edda002004ab17a46e3477d182059fa7 Mon Sep 17 00:00:00 2001 From: Gaizka Menendez Date: Wed, 23 Sep 2026 18:51:59 +0000 Subject: [PATCH 10/10] fix(pagination): document list contract and harden SDK pagers (#3279) * fix(pagination): document list contract and harden SDK pagers Signed-off-by: Gaizka Menendez Hernandez * fix(pagination): bound pager token history Signed-off-by: Gaizka Menendez Hernandez * fix(pagination): preflight token history limits Signed-off-by: Gaizka Menendez Hernandez * fix(pagination): paginate sandbox providers Signed-off-by: Gaizka Menendez Hernandez * fix(pagination): harden TypeScript pager Signed-off-by: Gaizka Menendez Hernandez * fix(pagination): expose provider pagers in SDKs Signed-off-by: Gaizka Menendez Hernandez * docs(pagination): describe a uniform list contract Signed-off-by: Gaizka Menendez Hernandez * fix(pagination): use stable provider cursors Signed-off-by: Gaizka Menendez Hernandez * docs(pagination): clarify mutation semantics Signed-off-by: Gaizka Menendez Hernandez * fix(cli): expose sandbox provider pagination Signed-off-by: Gaizka Menendez Hernandez * fix(proto): refresh pagination schema inventory Signed-off-by: Gaizka Menendez Hernandez * fix(proto): refresh rebased schema inventory Signed-off-by: Gaizka Menendez Hernandez Signed-off-by: Kris Hicks --------- Signed-off-by: Gaizka Menendez Hernandez Signed-off-by: Kris Hicks --- architecture/gateway.md | 24 ++- crates/openshell-cli/src/commands/provider.rs | 17 ++- crates/openshell-cli/src/main.rs | 17 ++- .../tests/provider_commands_integration.rs | 110 ++++++++++++- crates/openshell-sdk/README.md | 8 +- crates/openshell-sdk/src/pagination.rs | 4 +- crates/openshell-server/src/grpc/sandbox.rs | 144 +++++++++++++++++- crates/openshell-server/src/pagination.rs | 87 ++++++++++- crates/openshell-server/src/storage_proto.rs | 2 +- proto/openshell.proto | 8 + proto/pagination.proto | 6 + python/openshell/sandbox.py | 34 ++++- python/openshell/sandbox_test.py | 43 ++++++ sdk/go/openshell/v1/exec_client_test.go | 5 +- sdk/go/openshell/v1/fake/sandbox.go | 23 ++- sdk/go/openshell/v1/fake/sandbox_test.go | 14 +- sdk/go/openshell/v1/pager.go | 49 +++++- sdk/go/openshell/v1/pager_test.go | 40 +++++ sdk/go/openshell/v1/sandbox.go | 5 +- sdk/go/openshell/v1/sandbox_client.go | 39 +++-- sdk/go/openshell/v1/sandbox_client_test.go | 98 +++++++++--- sdk/go/openshell/v1/ssh_client_test.go | 5 +- sdk/go/openshell/v1/tcp_client_test.go | 5 +- sdk/go/proto/openshellv1/openshell.pb.go | 49 +++++- sdk/typescript/src/client.test.ts | 119 +++++++++++++-- sdk/typescript/src/client.ts | 68 +++++++-- 26 files changed, 909 insertions(+), 114 deletions(-) diff --git a/architecture/gateway.md b/architecture/gateway.md index e065fd8e1a..3de3154033 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -858,10 +858,16 @@ modes: and `page_token` fields, and responses carry `next_page_token`. The gateway clamps page sizes to 1,000 and returns opaque base64url continuation tokens. Tokens bind the RPC and every request parameter except `page_size`, contain no -authorization grant, and use immutable keyset cursors rather than database -offsets. Each page repeats normal authentication and authorization. Pagination -is weakly consistent under concurrent writes and deletes; it does not provide a -historical snapshot. +authorization grant, and use immutable keyset cursors. Each page repeats normal +authentication and authorization. Pagination is resumable and keyset-based; it +does not provide a historical snapshot while the collection is mutated +concurrently. + +Object-backed lists (`ListSandboxTemplates`, `ListSandboxes`, `ListServices`, +`ListProviders`, `ListWorkspaces`, and `ListWorkspaceMembers`) sort ascending +by `(created_at_ms, name, workspace, id)`. `ListProviderProfiles` sorts +ascending by `(id, scope)`, and `ListSandboxPolicies` sorts by descending policy +version. Provider attachments sort ascending by provider name. The token wire format is a private shared protobuf used only by the gateway. Public request and response messages repeat the standard AIP fields directly @@ -878,6 +884,16 @@ Curated Rust, Python, Go, and TypeScript SDK list methods return lazy pagers. Advancing a pager issues one list RPC and exposes its continuation token; explicit `list_all` helpers are the only curated APIs that exhaust a collection. +**Migration.** This pagination contract is a breaking replacement for the +former `limit`/`offset` list APIs. Protocol clients must send `page_size` and +resume only with the returned `next_page_token`; an empty token is the sole +completion signal. SDK callers that need every item must use the explicit +full-iteration helper (for example, Rust's `list_all_sandboxes`) rather than +awaiting `list_sandboxes`, which now returns one-page `Pager` state. Callers +that need one page should advance that pager once and retain its token. Internal +full scans use the reusable iteration helpers from the persistence pagination +audit rather than manually advancing offsets. + Persistence distinguishes one-page operations from exhaustive scans. `list_object_page` and `list_message_page` return one keyset page and its next cursor. `collect_records` and `collect_messages` exhaust those pages, fail on diff --git a/crates/openshell-cli/src/commands/provider.rs b/crates/openshell-cli/src/commands/provider.rs index d7b3e8fcf1..3e889e025c 100644 --- a/crates/openshell-cli/src/commands/provider.rs +++ b/crates/openshell-cli/src/commands/provider.rs @@ -87,6 +87,8 @@ fn aggregate_delete_failures(resource: &str, failures: &[String]) -> Result<()> pub async fn sandbox_provider_list( server: &str, name: &str, + page_size: i32, + page_token: &str, output: &str, workspace: &str, tls: &TlsOptions, @@ -98,14 +100,25 @@ pub async fn sandbox_provider_list( workspace_scope: Some(openshell_core::proto::workspace_selector( (workspace).to_string(), )), + page_size, + page_token: page_token.to_string(), }) .await .into_diagnostic()?; - let providers = response.into_inner().providers; + let response = response.into_inner(); + let next_page_token = response.next_page_token; + let providers = response.providers; - if crate::output::print_output_collection(output, &providers, attached_provider_to_json)? { + if crate::output::print_paginated_output_collection( + output, + "providers", + &providers, + &next_page_token, + attached_provider_to_json, + )? { return Ok(()); } + crate::output::print_next_page_token(&next_page_token); if providers.is_empty() { println!("No providers attached to sandbox {name}."); diff --git a/crates/openshell-cli/src/main.rs b/crates/openshell-cli/src/main.rs index 06ca33c9dc..eb52e3f253 100644 --- a/crates/openshell-cli/src/main.rs +++ b/crates/openshell-cli/src/main.rs @@ -1799,6 +1799,14 @@ enum SandboxProviderCommands { #[arg(add = ArgValueCompleter::new(completers::complete_sandbox_names))] name: Option, + /// Maximum number of attached providers to return in this page. + #[arg(long, default_value_t = 100)] + page_size: i32, + + /// Opaque continuation token from a previous page. + #[arg(long, default_value = "")] + page_token: String, + /// Output format. #[arg(short = 'o', long = "output", value_enum, default_value_t = OutputFormat::Table)] output: OutputFormat, @@ -3573,11 +3581,18 @@ async fn run_async() -> Result<()> { run::print_ssh_config(&ctx.name, &name, &cli.workspace); } SandboxCommands::Provider(command) => match command { - SandboxProviderCommands::List { name, output } => { + SandboxProviderCommands::List { + name, + page_size, + page_token, + output, + } => { let name = resolve_sandbox_name(name, &ctx.name, &cli.workspace)?; run::sandbox_provider_list( endpoint, &name, + page_size, + &page_token, output.as_str(), &cli.workspace, &tls, diff --git a/crates/openshell-cli/tests/provider_commands_integration.rs b/crates/openshell-cli/tests/provider_commands_integration.rs index 9d06a02035..9e8782b878 100644 --- a/crates/openshell-cli/tests/provider_commands_integration.rs +++ b/crates/openshell-cli/tests/provider_commands_integration.rs @@ -103,6 +103,7 @@ struct ProviderState { fail_delete_provider_profile_message: Arc>>, sandbox_providers: Arc>>>, sandbox_provider_requests: Arc>>, + sandbox_provider_next_page_token: Arc>, readiness_receipts: Arc>>, readiness_scripts: Arc>, readiness_requests: Arc>>, @@ -139,6 +140,8 @@ enum ProviderRefreshRequestLog { enum SandboxProviderRequestLog { List { sandbox_name: String, + page_size: i32, + page_token: String, }, Attach { sandbox_name: String, @@ -376,6 +379,8 @@ impl OpenShell for TestOpenShell { .await .push(SandboxProviderRequestLog::List { sandbox_name: sandbox_name.clone(), + page_size: request.page_size, + page_token: request.page_token, }); let provider_names = self .state @@ -390,7 +395,15 @@ impl OpenShell for TestOpenShell { .iter() .filter_map(|name| providers_by_name.get(name).cloned()) .collect(); - Ok(Response::new(ListSandboxProvidersResponse { providers })) + Ok(Response::new(ListSandboxProvidersResponse { + providers, + next_page_token: self + .state + .sandbox_provider_next_page_token + .lock() + .await + .clone(), + })) } async fn attach_sandbox_provider( @@ -4160,9 +4173,17 @@ async fn sandbox_provider_cli_run_functions_wire_requests_and_idempotent_results ) .await .expect("sandbox provider attach is idempotent"); - run::sandbox_provider_list(&ts.endpoint, "dev-sandbox", "table", "default", &ts.tls) - .await - .expect("sandbox provider list"); + run::sandbox_provider_list( + &ts.endpoint, + "dev-sandbox", + 100, + "", + "table", + "default", + &ts.tls, + ) + .await + .expect("sandbox provider list"); run::sandbox_provider_detach( &ts.endpoint, "dev-sandbox", @@ -4198,6 +4219,8 @@ async fn sandbox_provider_cli_run_functions_wire_requests_and_idempotent_results }, SandboxProviderRequestLog::List { sandbox_name: "dev-sandbox".to_string(), + page_size: 100, + page_token: String::new(), }, SandboxProviderRequestLog::Detach { sandbox_name: "dev-sandbox".to_string(), @@ -4214,6 +4237,85 @@ async fn sandbox_provider_cli_run_functions_wire_requests_and_idempotent_results assert!(providers.get("dev-sandbox").is_none_or(Vec::is_empty)); } +#[tokio::test] +async fn sandbox_provider_list_cli_forwards_pagination_and_emits_envelopes() { + let ts = run_server().await; + ts.state.providers.lock().await.insert( + "work-github".to_string(), + Provider { + metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { + id: "provider-work-github".to_string(), + name: "work-github".to_string(), + workspace: "default".to_string(), + resource_version: 1, + ..Default::default() + }), + r#type: "github".to_string(), + ..Default::default() + }, + ); + ts.state + .sandbox_providers + .lock() + .await + .insert("dev-sandbox".to_string(), vec!["work-github".to_string()]); + *ts.state.sandbox_provider_next_page_token.lock().await = "next-page".to_string(); + + for format in ["json", "yaml"] { + let output = run_readiness_cli( + &ts, + &[ + "sandbox", + "provider", + "list", + "dev-sandbox", + "--page-size", + "1", + "--page-token", + "prior-page", + "--output", + format, + ], + ) + .await; + assert!( + output.status.success(), + "sandbox provider list {format} failed:\n{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!( + output.stderr.is_empty(), + "structured output wrote to stderr" + ); + + let value: serde_json::Value = match format { + "json" => serde_json::from_slice(&output.stdout).expect("parse JSON envelope"), + "yaml" => serde_yml::from_slice(&output.stdout).expect("parse YAML envelope"), + _ => unreachable!(), + }; + assert_eq!(value["next_page_token"], "next-page"); + assert_eq!(value["providers"].as_array().map(Vec::len), Some(1)); + assert_eq!(value["providers"][0]["name"], "work-github"); + } + + let requests = ts.state.sandbox_provider_requests.lock().await.clone(); + assert_eq!( + requests, + vec![ + SandboxProviderRequestLog::List { + sandbox_name: "dev-sandbox".to_string(), + page_size: 1, + page_token: "prior-page".to_string(), + }, + SandboxProviderRequestLog::List { + sandbox_name: "dev-sandbox".to_string(), + page_size: 1, + page_token: "prior-page".to_string(), + }, + ] + ); +} + #[tokio::test] async fn sandbox_provider_attach_cli_surfaces_server_errors() { let ts = run_server().await; diff --git a/crates/openshell-sdk/README.md b/crates/openshell-sdk/README.md index 47805e38ee..4f721cbece 100644 --- a/crates/openshell-sdk/README.md +++ b/crates/openshell-sdk/README.md @@ -81,9 +81,11 @@ if deletion.outcome == openshell_sdk::DeletionOutcome::Accepted { ``` Curated `list_*` methods return a lazy `Pager`. Each `next_page()` call -issues at most one RPC and returns a `Page` with its opaque continuation -token. The explicit `list_all_*` conveniences exhaust that pager; `page_size` -always controls one gateway request, and `page_token` resumes a saved traversal. +fetches one logical page and returns a `Page` with its opaque continuation +token. OIDC authentication can retry that page once after an `Unauthenticated` +response. The explicit `list_all_*` conveniences exhaust that pager; +`page_size` always controls one gateway request, and `page_token` resumes a +saved traversal. ```rust let mut pages = client.list_sandboxes(ListOptions { diff --git a/crates/openshell-sdk/src/pagination.rs b/crates/openshell-sdk/src/pagination.rs index 8f35d03dab..94943436fd 100644 --- a/crates/openshell-sdk/src/pagination.rs +++ b/crates/openshell-sdk/src/pagination.rs @@ -22,7 +22,9 @@ pub struct Page { /// A lazy, single-pass iterator over response pages. /// /// Constructed by curated `list_*` methods. No RPC is issued until -/// [`Pager::next_page`] is called, and each call fetches at most one page. +/// [`Pager::next_page`] is called, and each call fetches one logical page. +/// A refreshed OIDC credential may retry that page once after an +/// `Unauthenticated` response. pub struct Pager { fetch: PageFetcher, next_page_token: Option, diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 1428b766db..db98b18bd9 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -1221,8 +1221,43 @@ pub(super) async fn handle_list_sandbox_providers( ) .await?; let workspace = sandbox.object_workspace().to_string(); - let providers = providers_for_sandbox(state, &sandbox, &workspace).await?; - Ok(Response::new(ListSandboxProvidersResponse { providers })) + let pagination = Pagination::new( + req.page_size, + &req.page_token, + "ListSandboxProviders", + &[&workspace, &req.sandbox], + )?; + let mut providers = providers_for_sandbox(state, &sandbox, &workspace) + .await? + .into_iter() + .map(|provider| { + let name = provider + .metadata + .as_ref() + .filter(|metadata| !metadata.name.is_empty()) + .map(|metadata| metadata.name.clone()) + .ok_or_else(|| Status::internal("provider metadata name is missing"))?; + Ok((name, provider)) + }) + .collect::, Status>>()?; + providers.sort_unstable_by(|left, right| left.0.cmp(&right.0)); + let start = pagination.provider_cursor()?.map_or(0, |cursor| { + providers.partition_point(|(name, _)| name.as_str() <= cursor) + }); + let end = start + .saturating_add(usize::try_from(pagination.page_size()).expect("u32 fits in usize")) + .min(providers.len()); + let next_page_token = pagination + .next_provider_token((end < providers.len()).then(|| providers[end - 1].0.as_str())); + Ok(Response::new(ListSandboxProvidersResponse { + providers: providers + .into_iter() + .skip(start) + .take(end - start) + .map(|(_, provider)| provider) + .collect(), + next_page_token, + })) } pub(super) async fn handle_attach_sandbox_provider( @@ -5671,6 +5706,8 @@ mod tests { #[tokio::test] async fn list_sandbox_providers_returns_attached_provider_records() { + use openshell_core::proto::CreateWorkspaceRequest; + let state = test_server_state().await; state .store @@ -5679,29 +5716,120 @@ mod tests { .unwrap(); state .store - .put_message(&test_sandbox("work", vec!["work-github".to_string()])) + .put_message(&test_provider("work-gitlab", "gitlab")) + .await + .unwrap(); + state + .store + .put_message(&test_sandbox( + "work", + vec!["work-github".to_string(), "work-gitlab".to_string()], + )) + .await + .unwrap(); + state + .store + .put_message(&test_sandbox("other", vec!["work-github".to_string()])) .await .unwrap(); - let response = handle_list_sandbox_providers( + let first_page = handle_list_sandbox_providers( &state, authed_request(ListSandboxProvidersRequest { sandbox: "work".to_string(), workspace_scope: Some(openshell_core::proto::workspace_selector( "default".to_string(), )), + page_size: 1, + page_token: String::new(), }), ) .await .unwrap() .into_inner(); - assert_eq!(response.providers.len(), 1); - assert_eq!(response.providers[0].r#type, "github"); + assert_eq!(first_page.providers.len(), 1); + assert_eq!(first_page.providers[0].r#type, "github"); assert_eq!( - response.providers[0].credentials.get("TOKEN"), + first_page.providers[0].credentials.get("TOKEN"), Some(&"REDACTED".to_string()) ); + assert!(!first_page.next_page_token.is_empty()); + + crate::grpc::workspace::handle_create_workspace( + &state, + Request::new(CreateWorkspaceRequest { + request_id: String::new(), + name: "beta".to_string(), + labels: HashMap::new(), + }), + ) + .await + .expect("beta workspace should be created"); + let mut beta_provider = test_provider("work-github", "github"); + beta_provider.metadata.as_mut().unwrap().id = "provider-beta-work-github".to_string(); + beta_provider.metadata.as_mut().unwrap().workspace = "beta".to_string(); + state.store.put_message(&beta_provider).await.unwrap(); + let mut beta_sandbox = test_sandbox("work", vec!["work-github".to_string()]); + beta_sandbox.metadata.as_mut().unwrap().id = "sandbox-beta-work".to_string(); + beta_sandbox.metadata.as_mut().unwrap().workspace = "beta".to_string(); + state.store.put_message(&beta_sandbox).await.unwrap(); + + let err = handle_list_sandbox_providers( + &state, + authed_request(ListSandboxProvidersRequest { + sandbox: "work".to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("beta")), + page_size: 1, + page_token: first_page.next_page_token.clone(), + }), + ) + .await + .expect_err("a token for one workspace must not list a same-named sandbox elsewhere"); + assert_eq!(err.code(), tonic::Code::InvalidArgument); + + let mut sandbox = state + .store + .get_message_by_name::("default", "work") + .await + .unwrap() + .expect("sandbox exists"); + sandbox + .spec + .as_mut() + .expect("sandbox has a spec") + .providers + .retain(|name| name != "work-github"); + state.store.put_message(&sandbox).await.unwrap(); + + let err = handle_list_sandbox_providers( + &state, + authed_request(ListSandboxProvidersRequest { + sandbox: "other".to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + page_size: 1, + page_token: first_page.next_page_token.clone(), + }), + ) + .await + .expect_err("a token for one sandbox must not list another sandbox"); + assert_eq!(err.code(), tonic::Code::InvalidArgument); + + let second_page = handle_list_sandbox_providers( + &state, + authed_request(ListSandboxProvidersRequest { + sandbox: "work".to_string(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + page_size: 100, + page_token: first_page.next_page_token, + }), + ) + .await + .unwrap() + .into_inner(); + assert_eq!(second_page.providers.len(), 1); + assert_eq!(second_page.providers[0].r#type, "gitlab"); + assert!(second_page.next_page_token.is_empty()); } #[tokio::test] @@ -8913,6 +9041,8 @@ mod tests { workspace_scope: Some(openshell_core::proto::workspace_selector( "no-such-ws".to_string(), )), + page_size: 0, + page_token: String::new(), }), ) .await diff --git a/crates/openshell-server/src/pagination.rs b/crates/openshell-server/src/pagination.rs index 66dd1d092e..50751d00ce 100644 --- a/crates/openshell-server/src/pagination.rs +++ b/crates/openshell-server/src/pagination.rs @@ -5,7 +5,8 @@ use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use openshell_core::proto::pagination::v1::{ - ObjectCursor as ProtoObjectCursor, PageToken, PolicyCursor, ProfileCursor, page_token::Cursor, + ObjectCursor as ProtoObjectCursor, PageToken, PolicyCursor, ProfileCursor, ProviderCursor, + page_token::Cursor, }; use prost::Message; use sha2::{Digest, Sha256}; @@ -110,6 +111,16 @@ impl Pagination { } } + pub fn provider_cursor(&self) -> Result, Status> { + match &self.cursor { + None => Ok(None), + Some(Cursor::Provider(cursor)) => Ok(Some(&cursor.name)), + Some(_) => Err(Status::invalid_argument( + "page_token has the wrong cursor type", + )), + } + } + pub fn next_object_token(&self, cursor: Option<&ObjectCursor>) -> String { cursor.map_or_else(String::new, |cursor| { self.encode(Cursor::Object(ProtoObjectCursor { @@ -135,6 +146,14 @@ impl Pagination { }) } + pub fn next_provider_token(&self, name: Option<&str>) -> String { + name.map_or_else(String::new, |name| { + self.encode(Cursor::Provider(ProviderCursor { + name: name.to_string(), + })) + }) + } + fn encode(&self, cursor: Cursor) -> String { URL_SAFE_NO_PAD.encode( PageToken { @@ -165,6 +184,72 @@ fn fingerprint(parameters: &[&str]) -> Vec { mod tests { use super::*; + fn message_body<'a>(proto: &'a str, message: &str) -> &'a str { + let marker = format!("message {message} {{"); + let start = proto + .find(&marker) + .unwrap_or_else(|| panic!("{message} message must exist")) + + marker.len(); + proto[start..] + .split_once("\n}") + .unwrap_or_else(|| panic!("{message} message must close")) + .0 + } + + #[test] + fn every_public_list_rpc_is_paginated() { + let proto = include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../proto/openshell.proto" + )); + let normalized = proto.lines().map(str::trim).collect::>().join(" "); + let mut remaining = normalized.as_str(); + + while let Some(start) = remaining.find("rpc List") { + remaining = &remaining[start + "rpc ".len()..]; + let (method, after_request) = remaining + .split_once('(') + .expect("List RPC must declare a request type"); + let (_, after_returns) = after_request + .split_once("returns (") + .expect("List RPC must declare a response type"); + let (response, after_response) = after_returns + .split_once(')') + .expect("List RPC response type must close"); + let request = format!("{method}Request"); + + let request_body = message_body(proto, &request); + assert!( + request_body.contains("int32 page_size"), + "{request} must declare page_size" + ); + assert!( + request_body.contains("string page_token"), + "{request} must declare page_token" + ); + assert!( + message_body(proto, response).contains("string next_page_token"), + "{response} must declare next_page_token" + ); + + remaining = after_response; + } + } + + #[test] + fn provider_token_round_trips() { + let first = + Pagination::new(1, "", "ListSandboxProviders", &["default", "sandbox"]).unwrap(); + let token = first.next_provider_token(Some("provider-a")); + assert_eq!( + Pagination::new(1, &token, "ListSandboxProviders", &["default", "sandbox"]) + .unwrap() + .provider_cursor() + .unwrap(), + Some("provider-a") + ); + } + #[test] fn page_size_defaults_clamps_and_rejects_negative_values() { assert_eq!( diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index 196b7be1cd..16a733415c 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -123,7 +123,7 @@ mod tests { // no messages or enums and touch no stored type, so the durable and overlap // fingerprints below remain unchanged. const PUBLIC_RPC_SCHEMA_SHA256: &str = - "e9533b91b8ead17666c43b17ca5bca3bf84050346e4855e5e5a261cbb56a26bb"; + "8fb59b0932ec2f227fdec2d46b6204925e79595695810a247bef731ddd632594"; const DURABLE_SCHEMA_SHA256: &str = "9eeaa29dfba187bff69fb7bc4f9a13a0f1d7be3f7049a38c8f0e20ce77ec7d8b"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = diff --git a/proto/openshell.proto b/proto/openshell.proto index ed18c86b25..f90e5335b5 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1373,6 +1373,12 @@ message ListSandboxProvidersRequest { // Workspace scope. Only a named workspace selection is accepted. openshell.datamodel.v1.WorkspaceSelector workspace_scope = 2; string sandbox = 1; + // The maximum number of providers to return. Zero uses 100. Values above + // 1000 are coerced to 1000; negative values are invalid. + int32 page_size = 3; + // Token from a previous ListSandboxProviders response. All other request + // parameters except page_size must match the request that produced it. + string page_token = 4; } // Attach provider to sandbox request. @@ -1461,6 +1467,8 @@ message ListSandboxesResponse { // List providers attached to a sandbox response. message ListSandboxProvidersResponse { repeated openshell.datamodel.v1.Provider providers = 1; + // Token for the next page. Empty when there are no subsequent pages. + string next_page_token = 2; } // Attach provider to sandbox response. diff --git a/proto/pagination.proto b/proto/pagination.proto index 05fcaef17f..febef64df4 100644 --- a/proto/pagination.proto +++ b/proto/pagination.proto @@ -18,6 +18,7 @@ message PageToken { ObjectCursor object = 4; PolicyCursor policy = 5; ProfileCursor profile = 6; + ProviderCursor provider = 7; } } @@ -38,3 +39,8 @@ message PolicyCursor { message ProfileCursor { string key = 1; } + +// Cursor for providers attached to one sandbox, ordered by provider name. +message ProviderCursor { + string name = 1; +} diff --git a/python/openshell/sandbox.py b/python/openshell/sandbox.py index e304038493..d85d4c70fa 100644 --- a/python/openshell/sandbox.py +++ b/python/openshell/sandbox.py @@ -37,6 +37,8 @@ ) _OAUTH_MAX_RESPONSE_BYTES = 1 << 20 +_PAGER_MAX_CONSUMED_TOKENS = 10_000 +_PAGER_MAX_CONSUMED_TOKEN_BYTES = 1 << 20 T = TypeVar("T") @@ -49,20 +51,46 @@ class Page(Generic[T]): class Pager(Generic[T]): - """Lazy, single-pass iterator that fetches one RPC page per advance.""" + """Lazy, single-pass iterator over the continuation-token contract. + + The repeated-token guard has bounded memory and raises ``SandboxError`` if + the traversal exceeds that guard's token-count or byte budget. + """ def __init__(self, fetch: Callable[[str], Page[T]], page_token: str = "") -> None: self._fetch = fetch self._page_token: str | None = page_token + self._consumed_page_tokens: set[str] = set() + self._consumed_page_token_bytes = 0 def __iter__(self) -> Pager[T]: return self + def _validate_page_token_budget(self, page_token: str) -> int: + if not page_token: + return 0 + token_bytes = len(page_token.encode("utf-8")) + if ( + len(self._consumed_page_tokens) >= _PAGER_MAX_CONSUMED_TOKENS + or self._consumed_page_token_bytes + token_bytes + > _PAGER_MAX_CONSUMED_TOKEN_BYTES + ): + raise SandboxError("pager continuation token history limit exceeded") + return token_bytes + def __next__(self) -> Page[T]: if self._page_token is None: raise StopIteration - page = self._fetch(self._page_token) - self._page_token = page.next_page_token or None + page_token = self._page_token + token_bytes = self._validate_page_token_budget(page_token) + page = self._fetch(page_token) + if page_token: + self._consumed_page_tokens.add(page_token) + self._consumed_page_token_bytes += token_bytes + next_page_token = page.next_page_token + if next_page_token and next_page_token in self._consumed_page_tokens: + raise SandboxError("pager received a repeated continuation token") + self._page_token = next_page_token or None return page def all(self) -> builtins.list[T]: diff --git a/python/openshell/sandbox_test.py b/python/openshell/sandbox_test.py index 2980450b7b..1d32ea3a80 100644 --- a/python/openshell/sandbox_test.py +++ b/python/openshell/sandbox_test.py @@ -2652,6 +2652,49 @@ def fetch(token: str) -> Page[int]: assert tokens == ["resume", "resume"] +def test_pager_rejects_a_repeated_continuation_token() -> None: + pager = Pager( + lambda token: Page(items=[1], next_page_token=token), + page_token="resume", + ) + + with pytest.raises(SandboxError, match="repeated continuation token"): + next(pager) + + +def test_pager_bounds_consumed_token_count(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(sandbox_module, "_PAGER_MAX_CONSUMED_TOKENS", 1) + requests: list[str] = [] + pager = Pager( + lambda token: ( + requests.append(token), + Page(items=[token], next_page_token="next"), + )[1], + page_token="first", + ) + + assert next(pager).items == ["first"] + with pytest.raises(SandboxError, match="token history limit exceeded"): + next(pager) + assert requests == ["first"] + + +def test_pager_bounds_consumed_token_bytes(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(sandbox_module, "_PAGER_MAX_CONSUMED_TOKEN_BYTES", 1) + requests: list[str] = [] + pager = Pager( + lambda token: ( + requests.append(token), + Page(items=[token], next_page_token="next"), + )[1], + page_token="too-large", + ) + + with pytest.raises(SandboxError, match="token history limit exceeded"): + next(pager) + assert requests == [] + + def test_list_ids_forwards_label_selector() -> None: stub = _FakeSandboxStub(listed=[_make_sandbox_proto("sandbox-1", "job-1")]) client = _client_with_fake_stub(stub) diff --git a/sdk/go/openshell/v1/exec_client_test.go b/sdk/go/openshell/v1/exec_client_test.go index 8354fb115c..34f090498c 100644 --- a/sdk/go/openshell/v1/exec_client_test.go +++ b/sdk/go/openshell/v1/exec_client_test.go @@ -52,7 +52,10 @@ func (r *stubSandboxResolver) AttachProvider(context.Context, string, string, st func (r *stubSandboxResolver) DetachProvider(context.Context, string, string, string, uint64) (*DetachProviderResult, error) { panic("not implemented") } -func (r *stubSandboxResolver) ListProviders(context.Context, string, string) ([]*Provider, error) { +func (r *stubSandboxResolver) ListProviders(string, string, ...ListOptions) (*Pager[*Provider], error) { + panic("not implemented") +} +func (r *stubSandboxResolver) ListAllProviders(context.Context, string, string, ...ListOptions) ([]*Provider, error) { panic("not implemented") } func (r *stubSandboxResolver) WaitReady(context.Context, string, string, ...WaitOptions) (*Sandbox, error) { diff --git a/sdk/go/openshell/v1/fake/sandbox.go b/sdk/go/openshell/v1/fake/sandbox.go index 432c859af8..2932457c25 100644 --- a/sdk/go/openshell/v1/fake/sandbox.go +++ b/sdk/go/openshell/v1/fake/sandbox.go @@ -814,11 +814,10 @@ func (c *fakeSandboxClient) GetLogs(_ context.Context, _, _ string, _ ...v1.LogO return nil, &types.StatusError{Code: types.ErrorUnimplemented, Message: "GetLogs not implemented in fake client"} } -// ListProviders returns stub Provider objects for each provider name -// attached to the sandbox. The returned providers contain only the Name -// field, since the fake client does not maintain a full provider registry -// per sandbox. -func (c *fakeSandboxClient) ListProviders(_ context.Context, workspace, sandboxName string) ([]*types.Provider, error) { +// ListProviders returns a pager over stub Provider objects attached to the +// sandbox. The returned providers contain only the Name field, since the fake +// client does not maintain a full provider registry per sandbox. +func (c *fakeSandboxClient) ListProviders(workspace, sandboxName string, opts ...v1.ListOptions) (*v1.Pager[*types.Provider], error) { if c.closedFunc() { return nil, &types.StatusError{Code: types.ErrorUnavailable, Message: "client is closed"} } @@ -832,5 +831,17 @@ func (c *fakeSandboxClient) ListProviders(_ context.Context, workspace, sandboxN for i, name := range sb.Spec.Providers { result[i] = &types.Provider{Name: name} } - return result, nil + var options v1.ListOptions + if len(opts) > 0 { + options = opts[0] + } + return newSlicePager(result, options.PageSize, options.PageToken) +} + +func (c *fakeSandboxClient) ListAllProviders(ctx context.Context, workspace, sandboxName string, opts ...v1.ListOptions) ([]*types.Provider, error) { + pager, err := c.ListProviders(workspace, sandboxName, opts...) + if err != nil { + return nil, err + } + return pager.All(ctx) } diff --git a/sdk/go/openshell/v1/fake/sandbox_test.go b/sdk/go/openshell/v1/fake/sandbox_test.go index 5303cdfdf3..4329652c13 100644 --- a/sdk/go/openshell/v1/fake/sandbox_test.go +++ b/sdk/go/openshell/v1/fake/sandbox_test.go @@ -667,7 +667,7 @@ func TestSandbox_ListProviders(t *testing.T) { require.NoError(t, err) // No providers yet - providers, err := sc.ListProviders(ctx, "default", "test-sb") + providers, err := sc.ListAllProviders(ctx, "default", "test-sb") require.NoError(t, err) assert.Empty(t, providers) @@ -678,7 +678,14 @@ func TestSandbox_ListProviders(t *testing.T) { _, err = sc.AttachProvider(ctx, "default", "test-sb", "anthropic", result.Sandbox.ResourceVersion) require.NoError(t, err) - providers, err = sc.ListProviders(ctx, "default", "test-sb") + pager, err := sc.ListProviders("default", "test-sb", v1.ListOptions{PageSize: 1}) + require.NoError(t, err) + firstPage, err := pager.NextPage(ctx) + require.NoError(t, err) + require.Len(t, firstPage.Items, 1) + require.NotEmpty(t, firstPage.NextPageToken) + + providers, err = sc.ListAllProviders(ctx, "default", "test-sb") require.NoError(t, err) assert.Len(t, providers, 2) @@ -692,9 +699,8 @@ func TestSandbox_ListProviders(t *testing.T) { func TestSandbox_ListProviders_SandboxNotFound(t *testing.T) { sc := newTestSandboxClient() - ctx := context.Background() - _, err := sc.ListProviders(ctx, "default", "nonexistent") + _, err := sc.ListProviders("default", "nonexistent") require.Error(t, err) assert.True(t, types.IsNotFound(err)) } diff --git a/sdk/go/openshell/v1/pager.go b/sdk/go/openshell/v1/pager.go index 558fda4694..a43dfdb957 100644 --- a/sdk/go/openshell/v1/pager.go +++ b/sdk/go/openshell/v1/pager.go @@ -8,6 +8,11 @@ import ( "errors" ) +const ( + maxConsumedPageTokens = 10_000 + maxConsumedPageTokenBytes = 1 << 20 +) + // Page is one response page from a list operation. type Page[T any] struct { Items []T @@ -16,28 +21,55 @@ type Page[T any] struct { type pageFetcher[T any] func(context.Context, string) (*Page[T], error) -// Pager lazily fetches one RPC page per call to NextPage. +// Pager lazily fetches pages from the continuation-token contract. // -// A Pager is single-pass and must not be used concurrently. +// A Pager is single-pass and must not be used concurrently. Its repeated-token +// guard has bounded memory and returns an error if the traversal exceeds its +// token-count or byte budget. type Pager[T any] struct { - fetch pageFetcher[T] - nextPageToken *string + fetch pageFetcher[T] + nextPageToken *string + consumedTokens map[string]struct{} + consumedTokenBytes int + maxConsumedTokens int + maxConsumedTokenBytes int } // NewPager constructs a pager from an RPC page fetcher. func NewPager[T any](pageToken string, fetch func(context.Context, string) (*Page[T], error)) *Pager[T] { - return &Pager[T]{fetch: fetch, nextPageToken: &pageToken} + return &Pager[T]{ + fetch: fetch, + nextPageToken: &pageToken, + consumedTokens: make(map[string]struct{}), + maxConsumedTokens: maxConsumedPageTokens, + maxConsumedTokenBytes: maxConsumedPageTokenBytes, + } } func newPager[T any](pageToken string, fetch pageFetcher[T]) *Pager[T] { return NewPager(pageToken, fetch) } +func (p *Pager[T]) validateCurrentTokenBudget() (int, error) { + if p.nextPageToken == nil || *p.nextPageToken == "" { + return 0, nil + } + tokenBytes := len(*p.nextPageToken) + if len(p.consumedTokens) >= p.maxConsumedTokens || tokenBytes > p.maxConsumedTokenBytes-p.consumedTokenBytes { + return 0, errors.New("pager continuation token history limit exceeded") + } + return tokenBytes, nil +} + // NextPage fetches the next page. It returns nil after the final page. func (p *Pager[T]) NextPage(ctx context.Context) (*Page[T], error) { if p.nextPageToken == nil { return nil, nil } + tokenBytes, err := p.validateCurrentTokenBudget() + if err != nil { + return nil, err + } page, err := p.fetch(ctx, *p.nextPageToken) if err != nil { return nil, err @@ -48,6 +80,13 @@ func (p *Pager[T]) NextPage(ctx context.Context) (*Page[T], error) { if page.Items == nil { page.Items = make([]T, 0) } + if *p.nextPageToken != "" { + p.consumedTokens[*p.nextPageToken] = struct{}{} + p.consumedTokenBytes += tokenBytes + } + if _, seen := p.consumedTokens[page.NextPageToken]; page.NextPageToken != "" && seen { + return nil, errors.New("pager received a repeated continuation token") + } if page.NextPageToken == "" { p.nextPageToken = nil } else { diff --git a/sdk/go/openshell/v1/pager_test.go b/sdk/go/openshell/v1/pager_test.go index 55dc74eb28..77d2e3227f 100644 --- a/sdk/go/openshell/v1/pager_test.go +++ b/sdk/go/openshell/v1/pager_test.go @@ -58,6 +58,46 @@ func TestPagerAllRetriesTheCurrentTokenAfterError(t *testing.T) { assert.Equal(t, []int{1, 2}, items) } +func TestPagerRejectsRepeatedContinuationToken(t *testing.T) { + pager := NewPager("resume-token", func(_ context.Context, token string) (*Page[string], error) { + return &Page[string]{Items: []string{"first"}, NextPageToken: token}, nil + }) + + page, err := pager.NextPage(context.Background()) + assert.Nil(t, page) + assert.EqualError(t, err, "pager received a repeated continuation token") +} + +func TestPagerBoundsConsumedTokenCount(t *testing.T) { + requests := 0 + pager := NewPager("first", func(_ context.Context, token string) (*Page[string], error) { + requests++ + return &Page[string]{Items: []string{token}, NextPageToken: "next"}, nil + }) + pager.maxConsumedTokens = 1 + + _, err := pager.NextPage(context.Background()) + require.NoError(t, err) + page, err := pager.NextPage(context.Background()) + assert.Nil(t, page) + assert.EqualError(t, err, "pager continuation token history limit exceeded") + assert.Equal(t, 1, requests) +} + +func TestPagerBoundsConsumedTokenBytes(t *testing.T) { + requests := 0 + pager := NewPager("too-large", func(_ context.Context, token string) (*Page[string], error) { + requests++ + return &Page[string]{Items: []string{token}, NextPageToken: "next"}, nil + }) + pager.maxConsumedTokenBytes = 1 + + page, err := pager.NextPage(context.Background()) + assert.Nil(t, page) + assert.EqualError(t, err, "pager continuation token history limit exceeded") + assert.Zero(t, requests) +} + func TestPagerNormalizesEmptyItems(t *testing.T) { pager := NewPager("", func(_ context.Context, _ string) (*Page[string], error) { return &Page[string]{}, nil diff --git a/sdk/go/openshell/v1/sandbox.go b/sdk/go/openshell/v1/sandbox.go index 89ba2859d7..02daad876b 100644 --- a/sdk/go/openshell/v1/sandbox.go +++ b/sdk/go/openshell/v1/sandbox.go @@ -80,7 +80,10 @@ type SandboxInterface interface { Delete(ctx context.Context, workspace, name string, opts ...DeleteOptions) (*DeletionResult, error) AttachProvider(ctx context.Context, workspace, sandboxName, providerName string, expectedResourceVersion uint64) (*AttachProviderResult, error) DetachProvider(ctx context.Context, workspace, sandboxName, providerName string, expectedResourceVersion uint64) (*DetachProviderResult, error) - ListProviders(ctx context.Context, workspace, sandboxName string) ([]*Provider, error) + // ListProviders returns a lazy pager over providers attached to a sandbox. + ListProviders(workspace, sandboxName string, opts ...ListOptions) (*Pager[*Provider], error) + // ListAllProviders exhausts ListProviders for callers that need every provider. + ListAllProviders(ctx context.Context, workspace, sandboxName string, opts ...ListOptions) ([]*Provider, error) WaitReady(ctx context.Context, workspace, name string, opts ...WaitOptions) (*Sandbox, error) WaitStopped(ctx context.Context, workspace, name string, opts ...WaitOptions) (*Sandbox, error) Watch(ctx context.Context, workspace, name string, opts ...WatchOptions) (WatchInterface[*Sandbox], error) diff --git a/sdk/go/openshell/v1/sandbox_client.go b/sdk/go/openshell/v1/sandbox_client.go index 2d22fd62fe..9868d11b93 100644 --- a/sdk/go/openshell/v1/sandbox_client.go +++ b/sdk/go/openshell/v1/sandbox_client.go @@ -228,20 +228,39 @@ func (s *sandboxClient) DetachProvider(ctx context.Context, workspace, sandboxNa }, nil } -func (s *sandboxClient) ListProviders(ctx context.Context, workspace, sandboxName string) ([]*Provider, error) { - resp, err := s.client.ListSandboxProviders(ctx, &pb.ListSandboxProvidersRequest{ - Sandbox: sandboxName, - WorkspaceScope: namedWorkspaceScope(workspace), - }) +func (s *sandboxClient) ListProviders(workspace, sandboxName string, opts ...ListOptions) (*Pager[*Provider], error) { + pageSize, err := listPageSize(opts) if err != nil { - return nil, converter.FromGRPCError(err) + return nil, err } + pageToken := "" + if len(opts) > 0 { + pageToken = opts[0].PageToken + } + return newPager(pageToken, func(ctx context.Context, pageToken string) (*Page[*Provider], error) { + resp, err := s.client.ListSandboxProviders(ctx, &pb.ListSandboxProvidersRequest{ + Sandbox: sandboxName, + WorkspaceScope: namedWorkspaceScope(workspace), + PageSize: pageSize, + PageToken: pageToken, + }) + if err != nil { + return nil, converter.FromGRPCError(err) + } + providers := make([]*Provider, 0, len(resp.GetProviders())) + for _, proto := range resp.GetProviders() { + providers = append(providers, converter.ProviderFromProto(proto)) + } + return &Page[*Provider]{Items: providers, NextPageToken: resp.GetNextPageToken()}, nil + }), nil +} - providers := make([]*Provider, 0, len(resp.GetProviders())) - for _, proto := range resp.GetProviders() { - providers = append(providers, converter.ProviderFromProto(proto)) +func (s *sandboxClient) ListAllProviders(ctx context.Context, workspace, sandboxName string, opts ...ListOptions) ([]*Provider, error) { + pager, err := s.ListProviders(workspace, sandboxName, opts...) + if err != nil { + return nil, err } - return providers, nil + return pager.All(ctx) } func (s *sandboxClient) WaitReady(ctx context.Context, workspace, name string, opts ...WaitOptions) (*Sandbox, error) { diff --git a/sdk/go/openshell/v1/sandbox_client_test.go b/sdk/go/openshell/v1/sandbox_client_test.go index c7c2421e0a..8ae398f9e0 100644 --- a/sdk/go/openshell/v1/sandbox_client_test.go +++ b/sdk/go/openshell/v1/sandbox_client_test.go @@ -25,26 +25,28 @@ import ( type mockSandboxServer struct { pb.UnimplementedOpenShellServer - mu sync.Mutex - sandboxes map[string]*pb.Sandbox - providers map[string][]*dm.Provider - createErr error - getErr error - listErr error - listPages [][]*pb.Sandbox - listRequests []*pb.ListSandboxesRequest - deleteErr error - deleteResponse *pb.DeleteSandboxResponse - deleteRequest *pb.DeleteSandboxRequest - attachErr error - detachErr error - listProvErr error - createRequest *pb.CreateSandboxRequest - watchEvents []*pb.SandboxStreamEvent - watchErr error - watchPostEventsErr error - watchKeepOpen chan struct{} // if non-nil, WatchSandbox blocks after sending events until closed - watchRequest *pb.WatchSandboxRequest // recorded request + mu sync.Mutex + sandboxes map[string]*pb.Sandbox + providers map[string][]*dm.Provider + createErr error + getErr error + listErr error + listPages [][]*pb.Sandbox + listRequests []*pb.ListSandboxesRequest + listProviderPages [][]*dm.Provider + listProviderRequests []*pb.ListSandboxProvidersRequest + deleteErr error + deleteResponse *pb.DeleteSandboxResponse + deleteRequest *pb.DeleteSandboxRequest + attachErr error + detachErr error + listProvErr error + createRequest *pb.CreateSandboxRequest + watchEvents []*pb.SandboxStreamEvent + watchErr error + watchPostEventsErr error + watchKeepOpen chan struct{} // if non-nil, WatchSandbox blocks after sending events until closed + watchRequest *pb.WatchSandboxRequest // recorded request // GetLogs fields getLogsResp *pb.GetSandboxLogsResponse @@ -209,9 +211,26 @@ func (s *mockSandboxServer) DetachSandboxProvider(_ context.Context, req *pb.Det } func (s *mockSandboxServer) ListSandboxProviders(_ context.Context, req *pb.ListSandboxProvidersRequest) (*pb.ListSandboxProvidersResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() if s.listProvErr != nil { return nil, s.listProvErr } + s.listProviderRequests = append(s.listProviderRequests, proto.Clone(req).(*pb.ListSandboxProvidersRequest)) + if s.listProviderPages != nil { + page := 0 + if req.GetPageToken() == "page-2" { + page = 1 + } + nextPageToken := "" + if page+1 < len(s.listProviderPages) { + nextPageToken = "page-2" + } + return &pb.ListSandboxProvidersResponse{ + Providers: s.listProviderPages[page], + NextPageToken: nextPageToken, + }, nil + } provs := s.providers[req.GetSandbox()] return &pb.ListSandboxProvidersResponse{Providers: provs}, nil } @@ -661,17 +680,44 @@ func TestSandboxDetachProvider_NotFound(t *testing.T) { func TestSandboxListProviders(t *testing.T) { mock := newMockSandboxServer() - mock.providers["my-sb"] = []*dm.Provider{ + mock.listProviderPages = [][]*dm.Provider{{ {Metadata: &dm.ObjectMeta{Name: "claude-prov"}, Type: "claude"}, + }, { {Metadata: &dm.ObjectMeta{Name: "github-prov"}, Type: "github"}, - } + }} client, cleanup := setupSandboxTest(t, mock) defer cleanup() - result, err := client.ListProviders(context.Background(), "default", "my-sb") + pager, err := client.ListProviders("default", "my-sb", ListOptions{PageSize: 1}) + + require.NoError(t, err) + page, err := pager.NextPage(context.Background()) + require.NoError(t, err) + require.NotNil(t, page) + assert.Len(t, page.Items, 1) + assert.Equal(t, "page-2", page.NextPageToken) + require.Len(t, mock.listProviderRequests, 1) + assert.Equal(t, "", mock.listProviderRequests[0].GetPageToken()) + assert.Equal(t, int32(1), mock.listProviderRequests[0].GetPageSize()) +} + +func TestSandboxListAllProviders(t *testing.T) { + mock := newMockSandboxServer() + mock.listProviderPages = [][]*dm.Provider{{ + {Metadata: &dm.ObjectMeta{Name: "claude-prov"}, Type: "claude"}, + }, { + {Metadata: &dm.ObjectMeta{Name: "github-prov"}, Type: "github"}, + }} + client, cleanup := setupSandboxTest(t, mock) + defer cleanup() + + result, err := client.ListAllProviders(context.Background(), "default", "my-sb", ListOptions{PageSize: 1}) require.NoError(t, err) assert.Len(t, result, 2) + require.Len(t, mock.listProviderRequests, 2) + assert.Equal(t, "", mock.listProviderRequests[0].GetPageToken()) + assert.Equal(t, "page-2", mock.listProviderRequests[1].GetPageToken()) } func TestSandboxListProviders_Empty(t *testing.T) { @@ -679,7 +725,7 @@ func TestSandboxListProviders_Empty(t *testing.T) { client, cleanup := setupSandboxTest(t, mock) defer cleanup() - result, err := client.ListProviders(context.Background(), "default", "empty-sb") + result, err := client.ListAllProviders(context.Background(), "default", "empty-sb") require.NoError(t, err) assert.Empty(t, result) @@ -691,7 +737,9 @@ func TestSandboxListProviders_Error(t *testing.T) { client, cleanup := setupSandboxTest(t, mock) defer cleanup() - _, err := client.ListProviders(context.Background(), "default", "sb") + pager, err := client.ListProviders("default", "sb") + require.NoError(t, err) + _, err = pager.NextPage(context.Background()) require.Error(t, err) assert.True(t, IsUnavailable(err)) diff --git a/sdk/go/openshell/v1/ssh_client_test.go b/sdk/go/openshell/v1/ssh_client_test.go index 9a00be634f..7225c058bb 100644 --- a/sdk/go/openshell/v1/ssh_client_test.go +++ b/sdk/go/openshell/v1/ssh_client_test.go @@ -170,7 +170,10 @@ func (m *mockSandboxResolver) AttachProvider(_ context.Context, _, _, _ string, func (m *mockSandboxResolver) DetachProvider(_ context.Context, _, _, _ string, _ uint64) (*DetachProviderResult, error) { return nil, nil } -func (m *mockSandboxResolver) ListProviders(_ context.Context, _, _ string) ([]*Provider, error) { +func (m *mockSandboxResolver) ListProviders(_, _ string, _ ...ListOptions) (*Pager[*Provider], error) { + return nil, nil +} +func (m *mockSandboxResolver) ListAllProviders(_ context.Context, _, _ string, _ ...ListOptions) ([]*Provider, error) { return nil, nil } func (m *mockSandboxResolver) WaitReady(_ context.Context, _, _ string, _ ...WaitOptions) (*Sandbox, error) { diff --git a/sdk/go/openshell/v1/tcp_client_test.go b/sdk/go/openshell/v1/tcp_client_test.go index f31f7a69b2..7bfdc60c45 100644 --- a/sdk/go/openshell/v1/tcp_client_test.go +++ b/sdk/go/openshell/v1/tcp_client_test.go @@ -1011,7 +1011,10 @@ func (r *flippableResolver) AttachProvider(context.Context, string, string, stri func (r *flippableResolver) DetachProvider(context.Context, string, string, string, uint64) (*DetachProviderResult, error) { panic("not implemented") } -func (r *flippableResolver) ListProviders(context.Context, string, string) ([]*Provider, error) { +func (r *flippableResolver) ListProviders(string, string, ...ListOptions) (*Pager[*Provider], error) { + panic("not implemented") +} +func (r *flippableResolver) ListAllProviders(context.Context, string, string, ...ListOptions) ([]*Provider, error) { panic("not implemented") } func (r *flippableResolver) WaitReady(context.Context, string, string, ...WaitOptions) (*Sandbox, error) { diff --git a/sdk/go/proto/openshellv1/openshell.pb.go b/sdk/go/proto/openshellv1/openshell.pb.go index e40dbd8082..66c1bb0c11 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -4094,8 +4094,14 @@ type ListSandboxProvidersRequest struct { // Workspace scope. Only a named workspace selection is accepted. WorkspaceScope *datamodelv1.WorkspaceSelector `protobuf:"bytes,2,opt,name=workspace_scope,json=workspaceScope,proto3" json:"workspace_scope,omitempty"` Sandbox string `protobuf:"bytes,1,opt,name=sandbox,proto3" json:"sandbox,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // The maximum number of providers to return. Zero uses 100. Values above + // 1000 are coerced to 1000; negative values are invalid. + PageSize int32 `protobuf:"varint,3,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"` + // Token from a previous ListSandboxProviders response. All other request + // parameters except page_size must match the request that produced it. + PageToken string `protobuf:"bytes,4,opt,name=page_token,json=pageToken,proto3" json:"page_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ListSandboxProvidersRequest) Reset() { @@ -4142,6 +4148,20 @@ func (x *ListSandboxProvidersRequest) GetSandbox() string { return "" } +func (x *ListSandboxProvidersRequest) GetPageSize() int32 { + if x != nil { + return x.PageSize + } + return 0 +} + +func (x *ListSandboxProvidersRequest) GetPageToken() string { + if x != nil { + return x.PageToken + } + return "" +} + // Attach provider to sandbox request. type AttachSandboxProviderRequest struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -4626,8 +4646,10 @@ func (x *ListSandboxesResponse) GetNextPageToken() string { // List providers attached to a sandbox response. type ListSandboxProvidersResponse struct { - state protoimpl.MessageState `protogen:"open.v1"` - Providers []*datamodelv1.Provider `protobuf:"bytes,1,rep,name=providers,proto3" json:"providers,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + Providers []*datamodelv1.Provider `protobuf:"bytes,1,rep,name=providers,proto3" json:"providers,omitempty"` + // Token for the next page. Empty when there are no subsequent pages. + NextPageToken string `protobuf:"bytes,2,opt,name=next_page_token,json=nextPageToken,proto3" json:"next_page_token,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -4669,6 +4691,13 @@ func (x *ListSandboxProvidersResponse) GetProviders() []*datamodelv1.Provider { return nil } +func (x *ListSandboxProvidersResponse) GetNextPageToken() string { + if x != nil { + return x.NextPageToken + } + return "" +} + // Attach provider to sandbox response. type AttachSandboxProviderResponse struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -17780,10 +17809,13 @@ const file_openshell_proto_rawDesc = "" + "\tpage_size\x18\x01 \x01(\x05R\bpageSize\x12\x1d\n" + "\n" + "page_token\x18\x02 \x01(\tR\tpageToken\x12%\n" + - "\x0elabel_selector\x18\x03 \x01(\tR\rlabelSelector\"\x8b\x01\n" + + "\x0elabel_selector\x18\x03 \x01(\tR\rlabelSelector\"\xc7\x01\n" + "\x1bListSandboxProvidersRequest\x12R\n" + "\x0fworkspace_scope\x18\x02 \x01(\v2).openshell.datamodel.v1.WorkspaceSelectorR\x0eworkspaceScope\x12\x18\n" + - "\asandbox\x18\x01 \x01(\tR\asandbox\"\x83\x02\n" + + "\asandbox\x18\x01 \x01(\tR\asandbox\x12\x1b\n" + + "\tpage_size\x18\x03 \x01(\x05R\bpageSize\x12\x1d\n" + + "\n" + + "page_token\x18\x04 \x01(\tR\tpageToken\"\x83\x02\n" + "\x1cAttachSandboxProviderRequest\x12R\n" + "\x0fworkspace_scope\x18\x04 \x01(\v2).openshell.datamodel.v1.WorkspaceSelectorR\x0eworkspaceScope\x12\x18\n" + "\asandbox\x18\x01 \x01(\tR\asandbox\x12\x1a\n" + @@ -17822,9 +17854,10 @@ const file_openshell_proto_rawDesc = "" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"t\n" + "\x15ListSandboxesResponse\x123\n" + "\tsandboxes\x18\x01 \x03(\v2\x15.openshell.v1.SandboxR\tsandboxes\x12&\n" + - "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"^\n" + + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"\x86\x01\n" + "\x1cListSandboxProvidersResponse\x12>\n" + - "\tproviders\x18\x01 \x03(\v2 .openshell.datamodel.v1.ProviderR\tproviders\"\xad\x01\n" + + "\tproviders\x18\x01 \x03(\v2 .openshell.datamodel.v1.ProviderR\tproviders\x12&\n" + + "\x0fnext_page_token\x18\x02 \x01(\tR\rnextPageToken\"\xad\x01\n" + "\x1dAttachSandboxProviderResponse\x12/\n" + "\asandbox\x18\x01 \x01(\v2\x15.openshell.v1.SandboxR\asandbox\x12\x1a\n" + "\battached\x18\x02 \x01(\bR\battached\x12?\n" + diff --git a/sdk/typescript/src/client.test.ts b/sdk/typescript/src/client.test.ts index dcc85f328e..7fef9b0db9 100644 --- a/sdk/typescript/src/client.test.ts +++ b/sdk/typescript/src/client.test.ts @@ -540,7 +540,7 @@ describe('create', () => { await sandbox.createSshSession('ssh', { workspace: 'staging' }); const attached = await sandbox.attachProvider('lookup', 'github', { workspace: 'staging' }); const detached = await sandbox.detachProvider('lookup', 'github', { workspace: 'staging' }); - await sandbox.listProviders('lookup', { workspace: 'staging' }); + await sandbox.listProviders('lookup', { workspace: 'staging' }).nextPage(); await sandbox.getConfig('config', { workspace: 'staging' }); await sandbox.setPolicy('lookup', { version: 1, networkPolicies: {} }, { workspace: 'staging' }); await sandbox.setSetting( @@ -617,6 +617,44 @@ describe('create', () => { expect(tokens).toEqual(['resume', 'resume']); }); + it('rejects a repeated continuation token', async () => { + const pager = new Pager(async (token) => ({ items: [1], nextPageToken: token }), 'resume'); + + await expect(pager.nextPage()).rejects.toThrow('pager received a repeated continuation token'); + }); + + it('prevents a request when the token-count budget is exhausted', async () => { + const requests: string[] = []; + const pager = new Pager( + async (token) => { + requests.push(token); + return { items: [1], nextPageToken: 'next' }; + }, + 'first', + 1, + ); + + await expect(pager.nextPage()).resolves.toEqual({ items: [1], nextPageToken: 'next' }); + await expect(pager.nextPage()).rejects.toThrow('pager continuation token history limit exceeded'); + expect(requests).toEqual(['first']); + }); + + it('prevents a request when the token-byte budget is exhausted', async () => { + const requests: string[] = []; + const pager = new Pager( + async (token) => { + requests.push(token); + return { items: [1], nextPageToken: '' }; + }, + 'too-large', + 10, + 1, + ); + + await expect(pager.nextPage()).rejects.toThrow('pager continuation token history limit exceeded'); + expect(requests).toEqual([]); + }); + it('createFromTemplate rejects an empty template name locally', async () => { const sandbox = client({}); await expect(sandbox.createFromTemplate({ workloadTemplate: ' ' })).rejects.toMatchObject({ @@ -1308,23 +1346,66 @@ describe('providers', () => { expect(detach.changed).toBe(false); }); - it('lists providers with u64 resourceVersion rendered as a string', async () => { + it('lists one provider page with its continuation token', async () => { + const pageTokens: string[] = []; const sandbox = client({ - listSandboxProviders: () => ({ - providers: [ - { - metadata: { - id: 'p1', - name: 'claude', - labels: { a: 'b' }, - resourceVersion: 99n, + listSandboxProviders: ({ pageToken, pageSize }) => { + pageTokens.push(pageToken); + expect(pageSize).toBe(1); + return { + providers: [ + { + metadata: { id: 'p1', name: 'claude', resourceVersion: 99n }, + type: 'claude', }, - type: 'claude', - }, - ], - }), + ], + nextPageToken: 'page-2', + }; + }, }); - const providers = await sandbox.listProviders('sb'); + + const page = await sandbox.listProviders('sb', { pageSize: 1 }).nextPage(); + expect(page).toMatchObject({ nextPageToken: 'page-2' }); + expect(page?.items.map((provider) => provider.name)).toEqual(['claude']); + expect(pageTokens).toEqual(['']); + }); + + it('lists all providers with u64 resourceVersion rendered as a string', async () => { + const pageTokens: string[] = []; + const sandbox = client({ + listSandboxProviders: ({ pageToken }) => { + pageTokens.push(pageToken); + return pageToken === '' + ? { + providers: [ + { + metadata: { + id: 'p1', + name: 'claude', + labels: { a: 'b' }, + resourceVersion: 99n, + }, + type: 'claude', + }, + ], + nextPageToken: 'page-2', + } + : { + providers: [ + { + metadata: { + id: 'p2', + name: 'github', + resourceVersion: 100n, + }, + type: 'github', + }, + ], + nextPageToken: '', + }; + }, + }); + const providers = await sandbox.listAllProviders('sb'); expect(providers).toEqual([ { id: 'p1', @@ -1333,7 +1414,15 @@ describe('providers', () => { labels: { a: 'b' }, resourceVersion: '99', }, + { + id: 'p2', + name: 'github', + type: 'github', + labels: {}, + resourceVersion: '100', + }, ]); + expect(pageTokens).toEqual(['', 'page-2']); }); }); diff --git a/sdk/typescript/src/client.ts b/sdk/typescript/src/client.ts index b0595f0c0a..93ad8ba6d1 100644 --- a/sdk/typescript/src/client.ts +++ b/sdk/typescript/src/client.ts @@ -224,6 +224,14 @@ export interface SandboxWorkspaceOptions { workspace?: string; } +/** Pagination and workspace scope for providers attached to one sandbox. */ +export type SandboxProviderListOptions = SandboxWorkspaceOptions & { + /** Maximum providers requested per page. */ + pageSize?: number; + /** Opaque token from a previous page. Omit to start at the beginning. */ + pageToken?: string; +}; + export type SandboxCallOptions = CallOptions & SandboxWorkspaceOptions; export interface SandboxTemplateWorkspaceOptions { @@ -781,21 +789,49 @@ export interface Page { readonly nextPageToken: string; } +const maxConsumedPageTokens = 10_000; +const maxConsumedPageTokenBytes = 1 << 20; + /** Lazy, single-pass iterator that fetches one RPC page per advance. */ export class Pager implements AsyncIterable> { private nextToken: string | undefined; + private readonly consumedTokens = new Set(); + private consumedTokenBytes = 0; constructor( private readonly fetch: (pageToken: string) => Promise>, pageToken = '', + private readonly maxConsumedTokens = maxConsumedPageTokens, + private readonly maxConsumedTokenBytes = maxConsumedPageTokenBytes, ) { this.nextToken = pageToken; } + private validateCurrentTokenBudget(pageToken: string): number { + if (pageToken === '') return 0; + const tokenBytes = new TextEncoder().encode(pageToken).byteLength; + if ( + this.consumedTokens.size >= this.maxConsumedTokens || + tokenBytes > this.maxConsumedTokenBytes - this.consumedTokenBytes + ) { + throw new Error('pager continuation token history limit exceeded'); + } + return tokenBytes; + } + /** Fetch the next page, or return undefined after the final page. */ async nextPage(): Promise | undefined> { if (this.nextToken === undefined) return undefined; - const page = await this.fetch(this.nextToken); + const pageToken = this.nextToken; + const tokenBytes = this.validateCurrentTokenBudget(pageToken); + const page = await this.fetch(pageToken); + if (pageToken !== '') { + this.consumedTokens.add(pageToken); + this.consumedTokenBytes += tokenBytes; + } + if (page.nextPageToken !== '' && this.consumedTokens.has(page.nextPageToken)) { + throw new Error('pager received a repeated continuation token'); + } this.nextToken = page.nextPageToken === '' ? undefined : page.nextPageToken; return page; } @@ -1606,15 +1642,27 @@ export class SandboxClient { } } - async listProviders(name: string, options?: SandboxWorkspaceOptions | null): Promise { - try { - const resp = await this.grpc.listSandboxProviders({ - ...sandboxTarget(name, options), - }); - return resp.providers.map((p) => providerRef(p)); - } catch (e) { - throw fromConnect(e); - } + listProviders(name: string, options?: SandboxProviderListOptions | null): Pager { + return new Pager(async (pageToken) => { + try { + const resp = await this.grpc.listSandboxProviders({ + ...sandboxTarget(name, options), + pageSize: options?.pageSize ?? 0, + pageToken, + }); + return { + items: resp.providers.map((provider) => providerRef(provider)), + nextPageToken: resp.nextPageToken, + }; + } catch (e) { + throw fromConnect(e); + } + }, options?.pageToken ?? ''); + } + + /** List and collect every provider attached to this sandbox. */ + async listAllProviders(name: string, options?: SandboxProviderListOptions | null): Promise { + return this.listProviders(name, options).all(); } async getConfig(name: string, options?: SandboxCallOptions | null): Promise {