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/.github/workflows/branch-e2e.yml b/.github/workflows/branch-e2e.yml index 99c68b9793..2935042776 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 @@ -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/.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 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/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/architecture/sandbox.md b/architecture/sandbox.md index e6f31436a4..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 @@ -648,7 +654,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-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-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/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/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/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() { 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/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(); 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/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/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/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/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/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/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/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/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; }); } 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/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/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 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/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. 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 { 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 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() { 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]]