From 8d1fd3e4c3547aec0feea035ac61ddd1e96d9cd2 Mon Sep 17 00:00:00 2001 From: Mrunal Patel Date: Mon, 14 Sep 2026 13:19:05 -0700 Subject: [PATCH] feat(api): add durable exec launch admission Fence duplicate exec launches with keyed durable admission and producer-owned terminal completion. Keep uncertain launches unresolved and never replay output or interactive input. Part of #3051 (phase 4a). Signed-off-by: Mrunal Patel --- architecture/gateway.md | 15 +- crates/openshell-cli/src/run.rs | 2 + crates/openshell-sdk/src/client.rs | 2 + crates/openshell-server/src/grpc/mod.rs | 2 +- .../src/grpc/mutation_replay.rs | 79 +++- .../src/grpc/mutation_replay/streaming.rs | 130 +++++ .../grpc/mutation_replay/streaming/tests.rs | 445 ++++++++++++++++++ crates/openshell-server/src/grpc/sandbox.rs | 193 +++++--- crates/openshell-server/src/storage_proto.rs | 2 +- docs/reference/api-errors.mdx | 39 +- e2e/python/test_exec_admission.py | 154 ++++++ proto/openshell.proto | 6 + sdk/go/proto/openshellv1/openshell.pb.go | 20 +- 13 files changed, 1001 insertions(+), 88 deletions(-) create mode 100644 crates/openshell-server/src/grpc/mutation_replay/streaming.rs create mode 100644 crates/openshell-server/src/grpc/mutation_replay/streaming/tests.rs create mode 100644 e2e/python/test_exec_admission.py diff --git a/architecture/gateway.md b/architecture/gateway.md index 62eceed3eb..d39b711514 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -91,7 +91,20 @@ in-memory extension. Replay reauthorizes original and current effective scopes, requires the same effective payload, and reruns current interceptor validation. Interceptors cannot mutate the request UUID. Server-marked replay suppresses post-commit observation, which remains best-effort rather than an outbox. -Credential capabilities and streaming execution require separate contracts. +Credential capabilities require separate contracts. + +Both exec RPCs share keyed admission but defer completion to the SSH producer. +The initial interactive Start uses the exec request schema under a distinct RPC +namespace; later stdin and resize frames are not replayable inputs. Admission +precedes relay opening and hands its CAS-only finalizer to the owned producer, +releasing the shared admission-worker permit after handoff. Only a confirmed +remote exit records a terminal marker and starts 24-hour retention. Synthetic +timeouts, disconnects without exit confirmation, and persistence failures leave +permanent unresolved claims. Duplicates never launch, attach, or replay output: +pending records report uncertainty, and terminal records report stream +unavailability. Existing transport cancellation behavior remains unchanged. +Exec timeouts retain the public duration's precision and presence: an absent +timeout is unbounded, while an explicit zero is a finite timeout. The gateway listens on one service port and multiplexes gRPC and HTTP traffic. The default local single-user deployment mode is mTLS user authentication: diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index ffd0fa2206..77784fed2a 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -1805,6 +1805,7 @@ pub async fn sandbox_exec_grpc( // Make the streaming gRPC call. let mut stream = client .exec_sandbox(ExecSandboxRequest { + request_id: String::new(), sandbox_id: sandbox.object_id().to_string(), command: command.to_vec(), workdir: workdir.unwrap_or_default().to_string(), @@ -2189,6 +2190,7 @@ async fn sandbox_exec_interactive_grpc( input_tx .send(ExecSandboxInput { payload: Some(exec_sandbox_input::Payload::Start(ExecSandboxRequest { + request_id: String::new(), sandbox_id: sandbox.object_id().to_string(), command: command.to_vec(), workdir: workdir.unwrap_or_default().to_string(), diff --git a/crates/openshell-sdk/src/client.rs b/crates/openshell-sdk/src/client.rs index 473f34d625..6f00dea319 100644 --- a/crates/openshell-sdk/src/client.rs +++ b/crates/openshell-sdk/src/client.rs @@ -608,6 +608,7 @@ impl OpenShellClient { pub async fn exec(&self, name: &str, cmd: &[String], opts: ExecOptions) -> Result { let sandbox = self.get_sandbox(name).await?; let request = proto::ExecSandboxRequest { + request_id: String::new(), sandbox_id: sandbox.id, command: cmd.to_vec(), workdir: opts.workdir.unwrap_or_default(), @@ -1068,6 +1069,7 @@ impl WorkspaceScopedClient { pub async fn exec(&self, name: &str, cmd: &[String], opts: ExecOptions) -> Result { let sandbox = self.get_sandbox(name).await?; let request = proto::ExecSandboxRequest { + request_id: String::new(), sandbox_id: sandbox.id, command: cmd.to_vec(), workdir: opts.workdir.unwrap_or_default(), diff --git a/crates/openshell-server/src/grpc/mod.rs b/crates/openshell-server/src/grpc/mod.rs index c0f6f926d5..6e51cf02b8 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -378,7 +378,7 @@ impl OpenShell for OpenShellService { &self, request: Request, ) -> Result, Status> { - sandbox::handle_exec_sandbox(&self.state, request).await + mutation_replay::run(&self.state, request).await } type ForwardTcpStream = diff --git a/crates/openshell-server/src/grpc/mutation_replay.rs b/crates/openshell-server/src/grpc/mutation_replay.rs index 34153a9ad8..4ceefe8a77 100644 --- a/crates/openshell-server/src/grpc/mutation_replay.rs +++ b/crates/openshell-server/src/grpc/mutation_replay.rs @@ -78,6 +78,7 @@ pub(super) enum Success { Resource { id: String, version: u64 }, Deletion { outcome: i32 }, Ordinary(ordinary::Outcome), + StreamTerminal, } pub(super) struct Scope { @@ -87,10 +88,15 @@ pub(super) struct Scope { #[tonic::async_trait] pub(super) trait Mutation: Message + Default + Send + Sync + 'static { - type Output: Message + Default + Send + 'static; + type Output: Send + 'static; const METHOD: &'static str; const PROTECTED: bool = false; + // Streaming producers, not returned stream handles, confirm completion. + const DEFERRED: bool = false; fn request_id(&self) -> &str; + fn canonical_message(&self) -> Result { + decode_request_message(&format!("{}Request", Self::METHOD), self) + } async fn authorize(&self, state: &ServerState, principal: &Principal) -> Result; async fn execute( state: &Arc, @@ -207,7 +213,7 @@ async fn execute_owned( protection: Option, scope: Scope, ) -> Result, Status> { - let mut admission = Admission { + let admission = Admission { format_version: 1, payload_hash, protection, @@ -302,32 +308,70 @@ async fn execute_owned( }; // Any error or panic from here leaves the claim unresolved. Status codes // do not establish that a handler performed no effects. + let completion = Completion { + store: state.store.clone(), + claim_id, + key: key.into(), + bucket: bucket.into(), + version, + admission: payload, + }; + if M::DEFERRED { + request.extensions_mut().insert(completion.clone()); + } let facts = ordinary::Facts::default(); request.extensions_mut().insert(facts.clone()); let mut response = M::execute(state, request).await?; + if M::DEFERRED { + return Ok(response); + } response.extensions_mut().insert(facts); - admission.success = Some(M::capture(&response)?); + completion.complete(M::capture(&response)?).await?; + return Ok(response); + } + Err(uncertain()) +} + +/// Only the admitted owner can finalize this exact incarnation of a claim. +/// A dropped finalizer leaves a permanent unresolved admission, never a lease. +#[derive(Clone)] +pub(super) struct Completion { + store: Arc, + claim_id: String, + key: String, + bucket: String, + version: u64, + admission: Vec, +} + +impl Completion { + pub(super) async fn stream_terminal(&self) -> Result<(), Status> { + self.complete(Success::StreamTerminal).await + } + + async fn complete(&self, success: Success) -> Result<(), Status> { + let mut admission: Admission = + serde_json::from_slice(&self.admission).map_err(|_| uncertain())?; + admission.success = Some(success); admission.completed_at_ms = Some(current_time_ms()); let payload = serde_json::to_vec(&admission).map_err(|_| uncertain())?; if payload.len() > 64 * 1024 { return Err(uncertain()); } - state - .store + self.store .put_if( OBJECT_TYPE, - &claim_id, - key, - bucket, + &self.claim_id, + &self.key, + &self.bucket, &payload, None, - WriteCondition::MatchResourceVersion(version), + WriteCondition::MatchResourceVersion(self.version), ) .await .map_err(|_| uncertain())?; - return Ok(response); + Ok(()) } - Err(uncertain()) } async fn prune_expired(store: &Store, bucket: &str) -> Result { @@ -377,12 +421,16 @@ fn validate_request_id(value: &str) -> Result { Ok(id.hyphenated().to_string()) } -fn fingerprint(request: &M) -> Result { +fn decode_request_message(name: &str, request: &impl Message) -> Result { let descriptor = DESCRIPTORS - .get_message_by_name(&format!("openshell.v1.{}Request", M::METHOD)) + .get_message_by_name(&format!("openshell.v1.{name}")) .ok_or_else(|| Status::internal("mutation request descriptor missing"))?; - let mut message = DynamicMessage::decode(descriptor, request.encode_to_vec().as_slice()) - .map_err(|_| Status::internal("decode mutation request"))?; + DynamicMessage::decode(descriptor, request.encode_to_vec().as_slice()) + .map_err(|_| Status::internal("decode mutation request")) +} + +fn fingerprint(request: &M) -> Result { + let mut message = request.canonical_message()?; message.clear_field_by_name("request_id"); let value = serde_json::to_value(message) .map_err(|_| Status::internal("canonicalize mutation request"))?; @@ -706,3 +754,4 @@ deletion_mutation!( mod tests; pub(super) mod ordinary; +pub(super) mod streaming; diff --git a/crates/openshell-server/src/grpc/mutation_replay/streaming.rs b/crates/openshell-server/src/grpc/mutation_replay/streaming.rs new file mode 100644 index 0000000000..b42601f4cd --- /dev/null +++ b/crates/openshell-server/src/grpc/mutation_replay/streaming.rs @@ -0,0 +1,130 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Launch admission only: output and interactive input are never replayed. + +#[cfg(test)] +mod tests; + +use std::sync::Arc; + +use openshell_core::ObjectWorkspace; +use openshell_core::proto::{ + ExecSandboxEvent, ExecSandboxInput, ExecSandboxRequest, exec_sandbox_input, +}; +use openshell_core::rpc_error; +use prost_reflect::DynamicMessage; +use tokio::sync::Mutex; +use tokio_stream::wrappers::ReceiverStream; +use tonic::{Request, Response, Status}; + +use super::{Mutation, Scope, Success, decode_request_message, named_scope, uncertain}; +use crate::ServerState; +use crate::auth::principal::Principal; +use crate::grpc::sandbox; +use crate::persistence::Store; + +pub(in crate::grpc) type ExecStream = ReceiverStream>; + +/// Private, one-owner transport handoff. The fingerprint includes only Start. +#[derive(Clone)] +pub(in crate::grpc) struct InteractiveInput( + pub Arc>>>, +); + +pub(in crate::grpc) fn start(input: &ExecSandboxInput) -> Result<&ExecSandboxRequest, Status> { + match input.payload.as_ref() { + Some(exec_sandbox_input::Payload::Start(request)) => Ok(request), + _ => Err(rpc_error::invalid_argument( + "start", + "first message must be a start payload", + )), + } +} + +async fn authorize( + state: &ServerState, + principal: &Principal, + request: &ExecSandboxRequest, +) -> Result { + sandbox::validate_exec_start(request)?; + let sandbox = + sandbox::fetch_and_authorize_sandbox(state, principal, &request.sandbox_id).await?; + named_scope(state, sandbox.object_workspace()).await +} + +fn stream_unavailable(success: Success) -> Status { + if !matches!(success, Success::StreamTerminal) { + return super::replay_unavailable(); + } + rpc_error::failed_precondition( + "REQUEST_STREAM_UNAVAILABLE", + "execution terminated, but its output stream is not stored; this request was not launched again", + ) +} + +#[tonic::async_trait] +impl Mutation for ExecSandboxRequest { + type Output = ExecStream; + const METHOD: &'static str = "ExecSandbox"; + const PROTECTED: bool = true; + const DEFERRED: bool = true; + + fn request_id(&self) -> &str { + &self.request_id + } + + async fn authorize(&self, state: &ServerState, principal: &Principal) -> Result { + authorize(state, principal, self).await + } + + async fn execute( + state: &Arc, + request: Request, + ) -> Result, Status> { + sandbox::handle_exec_sandbox(state, request).await + } + + fn capture(_: &Response) -> Result { + Err(uncertain()) + } + + async fn restore(_: &Store, success: Success) -> Result { + Err(stream_unavailable(success)) + } +} + +#[tonic::async_trait] +impl Mutation for ExecSandboxInput { + type Output = ExecStream; + const METHOD: &'static str = "ExecSandboxInteractive"; + const PROTECTED: bool = true; + const DEFERRED: bool = true; + + fn request_id(&self) -> &str { + start(self).map_or("", |request| request.request_id.as_str()) + } + + fn canonical_message(&self) -> Result { + decode_request_message("ExecSandboxRequest", start(self)?) + } + + async fn authorize(&self, state: &ServerState, principal: &Principal) -> Result { + authorize(state, principal, start(self)?).await + } + + async fn execute( + state: &Arc, + request: Request, + ) -> Result, Status> { + sandbox::handle_exec_sandbox_interactive_start(state, request).await + } + + fn capture(_: &Response) -> Result { + Err(uncertain()) + } + + async fn restore(_: &Store, success: Success) -> Result { + Err(stream_unavailable(success)) + } +} diff --git a/crates/openshell-server/src/grpc/mutation_replay/streaming/tests.rs b/crates/openshell-server/src/grpc/mutation_replay/streaming/tests.rs new file mode 100644 index 0000000000..e6aa544eab --- /dev/null +++ b/crates/openshell-server/src/grpc/mutation_replay/streaming/tests.rs @@ -0,0 +1,445 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use crate::grpc::mutation_replay::tests::{reason, state_for}; +use crate::grpc::mutation_replay::{ + Admission, Completion, OBJECT_TYPE, SUCCESS_TTL_MS, fingerprint, run, +}; +use crate::grpc::test_support::authed_request; +use crate::persistence::{ObjectType, WriteCondition, current_time_ms}; +use openshell_core::ObjectId; +use openshell_core::proto::datamodel::v1::ObjectMeta; +use openshell_core::proto::{ + GatewayMessage, Sandbox, SandboxPhase, SandboxStatus, gateway_message, +}; +use tokio::sync::{mpsc, oneshot}; +use tokio_stream::StreamExt; + +async fn setup(url: &str, directory: &tempfile::TempDir) -> Arc { + let mut state = state_for(Store::connect(url).await.unwrap()).await; + let key = directory.path().join("key"); + std::fs::write(&key, b"exec-test-only-key").unwrap(); + let state_mut = Arc::get_mut(&mut state).unwrap(); + state_mut.admin_role = "openshell-admin".into(); + state_mut.config.gateway_jwt = Some(openshell_core::config::GatewayJwtConfig { + signing_key_path: key, + public_key_path: directory.path().join("public"), + kid_path: directory.path().join("kid"), + gateway_id: "test".into(), + ttl_secs: None, + }); + state +} + +async fn request(state: &ServerState) -> ExecSandboxRequest { + let sandbox = Sandbox { + metadata: Some(ObjectMeta { + id: uuid::Uuid::new_v4().to_string(), + name: "exec-admission".into(), + workspace: "default".into(), + ..Default::default() + }), + status: Some(SandboxStatus { + phase: SandboxPhase::Ready.into(), + ..Default::default() + }), + ..Default::default() + }; + state.store.put_message(&sandbox).await.unwrap(); + ExecSandboxRequest { + sandbox_id: sandbox.object_id().into(), + request_id: uuid::Uuid::new_v4().to_string(), + command: vec!["echo".into(), "secret-command-argument".into()], + stdin: b"secret-stdin".to_vec(), + environment: [("PRIVATE".into(), "secret-environment".into())].into(), + ..Default::default() + } +} + +fn register(state: &ServerState, id: &str) -> mpsc::Receiver { + let (tx, rx) = mpsc::channel(64); + let (shutdown, _) = oneshot::channel(); + state + .supervisor_sessions + .register(id.into(), "session".into(), tx, shutdown); + rx +} + +async fn receive_relay(state: &ServerState, messages: &mut mpsc::Receiver) { + let message = tokio::time::timeout(std::time::Duration::from_secs(2), messages.recv()) + .await + .unwrap() + .unwrap(); + let Some(gateway_message::Payload::RelayOpen(open)) = message.payload else { + panic!("expected relay open") + }; + // No SSH transport in these admission tests. Release the detached producer. + state + .supervisor_sessions + .fail_pending_relay(&open.channel_id, "test relay ended".into()); +} + +async fn completion(state: &ServerState) -> Completion { + let rows = state + .store + .list_by_type_after(OBJECT_TYPE, None, 100) + .await + .unwrap(); + assert_eq!(rows.len(), 1); + let row = rows.into_iter().next().unwrap(); + Completion { + store: state.store.clone(), + claim_id: row.id, + key: row.name, + bucket: row.workspace, + version: row.resource_version, + admission: row.payload, + } +} + +#[test] +fn interactive_fingerprint_uses_start_schema_and_ignores_request_id() { + let mut req = ExecSandboxRequest { + request_id: uuid::Uuid::new_v4().to_string(), + command: vec!["echo".into()], + ..Default::default() + }; + let first = fingerprint(&req).unwrap(); + req.request_id = uuid::Uuid::new_v4().to_string(); + let input = ExecSandboxInput { + payload: Some(exec_sandbox_input::Payload::Start(req.clone())), + }; + assert_eq!(first, fingerprint(&input).unwrap()); + assert_ne!(ExecSandboxRequest::METHOD, ExecSandboxInput::METHOD); + req.stdin.push(42); + assert_ne!(first, fingerprint(&req).unwrap()); + assert_eq!( + ExecSandboxInput::default() + .canonical_message() + .unwrap_err() + .code(), + tonic::Code::InvalidArgument + ); +} + +#[tokio::test] +async fn concurrent_exec_claims_open_one_relay_and_store_no_secrets() { + let directory = tempfile::tempdir().unwrap(); + let state = setup("sqlite::memory:", &directory).await; + let req = request(&state).await; + let mut messages = register(&state, &req.sandbox_id); + let mut tasks = Vec::new(); + for _ in 0..16 { + let state = state.clone(); + let req = req.clone(); + tasks.push(tokio::spawn(async move { + run(&state, authed_request(req)).await + })); + } + let mut admitted = 0; + for task in tasks { + match task.await.unwrap() { + Ok(_) => admitted += 1, + Err(status) => assert_eq!(reason(&status), "REQUEST_OUTCOME_UNCERTAIN"), + } + } + assert_eq!(admitted, 1); + receive_relay(&state, &mut messages).await; + assert!(messages.try_recv().is_err()); + let pending = completion(&state).await; + let text = String::from_utf8(pending.admission).unwrap(); + for secret in [ + "secret-command-argument", + "secret-stdin", + "secret-environment", + &fingerprint(&req).unwrap(), + ] { + assert!(!text.contains(secret)); + } + let mut changed = req.clone(); + changed.command.push("different".into()); + assert_eq!( + reason(&run(&state, authed_request(changed)).await.unwrap_err()), + "REQUEST_ID_PAYLOAD_MISMATCH" + ); +} + +#[tokio::test] +async fn confirmed_exit_including_nonzero_and_124_finalizes_before_output_delivery() { + for code in [0, 7, 124] { + let directory = tempfile::tempdir().unwrap(); + let state = setup("sqlite::memory:", &directory).await; + let req = request(&state).await; + let mut messages = register(&state, &req.sandbox_id); + drop(run(&state, authed_request(req.clone())).await.unwrap()); + receive_relay(&state, &mut messages).await; + let owner = completion(&state).await; + assert_eq!( + sandbox::wait_for_exec_terminal(async { Ok(code) }, None, Some(owner.clone())) + .await + .unwrap(), + Some(code) + ); + assert_eq!( + reason(&run(&state, authed_request(req)).await.unwrap_err()), + "REQUEST_STREAM_UNAVAILABLE" + ); + assert!( + owner.stream_terminal().await.is_err(), + "CAS rejects a second finalization" + ); + let row = completion(&state).await; + let admission: Admission = serde_json::from_slice(&row.admission).unwrap(); + assert!(matches!(admission.success, Some(Success::StreamTerminal))); + assert!(admission.completed_at_ms.is_some()); + } +} + +#[tokio::test] +async fn timeout_and_lost_exit_status_never_finalize_or_expire_pending_launches() { + let directory = tempfile::tempdir().unwrap(); + let state = setup("sqlite::memory:", &directory).await; + let req = request(&state).await; + let mut messages = register(&state, &req.sandbox_id); + drop(run(&state, authed_request(req.clone())).await.unwrap()); + receive_relay(&state, &mut messages).await; + let owner = completion(&state).await; + for execution_timeout in [ + std::time::Duration::ZERO, + std::time::Duration::from_nanos(1), + std::time::Duration::from_secs(1), + ] { + assert_eq!( + tokio::time::timeout( + std::time::Duration::from_secs(5), + sandbox::wait_for_exec_terminal( + std::future::pending(), + Some(execution_timeout), + Some(owner.clone()), + ), + ) + .await + .expect("a present timeout must not wait indefinitely") + .unwrap(), + None + ); + } + assert!( + sandbox::wait_for_exec_terminal( + async { Err(Status::unavailable("lost exit")) }, + None, + Some(owner.clone()) + ) + .await + .is_err() + ); + let mut admission: Admission = serde_json::from_slice(&owner.admission).unwrap(); + assert!(admission.success.is_none()); + admission.completed_at_ms = Some(current_time_ms() - SUCCESS_TTL_MS - 1); + state + .store + .put_if( + OBJECT_TYPE, + &owner.claim_id, + &owner.key, + &owner.bucket, + &serde_json::to_vec(&admission).unwrap(), + None, + WriteCondition::MatchResourceVersion(owner.version), + ) + .await + .unwrap(); + // A real exit cannot be exposed as confirmed if the finalizer loses its CAS. + let status = sandbox::wait_for_exec_terminal(async { Ok(0) }, None, Some(owner)) + .await + .unwrap_err(); + assert_eq!(reason(&status), "REQUEST_OUTCOME_UNCERTAIN"); + assert_eq!( + reason(&run(&state, authed_request(req)).await.unwrap_err()), + "REQUEST_OUTCOME_UNCERTAIN" + ); + assert!(messages.try_recv().is_err()); +} + +#[tokio::test] +async fn cancellation_before_admission_does_nothing_and_after_admission_keeps_owner() { + let directory = tempfile::tempdir().unwrap(); + let state = setup("sqlite::memory:", &directory).await; + let req = request(&state).await; + drop(run(&state, authed_request(req.clone()))); + assert!( + state + .store + .list_by_type_after(OBJECT_TYPE, None, 100) + .await + .unwrap() + .is_empty() + ); + let task_state = state.clone(); + let task_req = req.clone(); + let task = tokio::spawn(async move { run(&task_state, authed_request(task_req)).await }); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while state + .store + .list_by_type_after(OBJECT_TYPE, None, 100) + .await + .unwrap() + .is_empty() + { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + task.abort(); + let mut messages = register(&state, &req.sandbox_id); + receive_relay(&state, &mut messages).await; + assert_eq!( + reason(&run(&state, authed_request(req)).await.unwrap_err()), + "REQUEST_OUTCOME_UNCERTAIN" + ); + assert!(messages.try_recv().is_err()); +} + +#[tokio::test] +async fn exec_restart_preserves_pending_and_terminal_fences_and_reauthorizes() { + let directory = tempfile::tempdir().unwrap(); + let url = format!( + "sqlite://{}?mode=rwc", + directory.path().join("store.db").display() + ); + let state = setup(&url, &directory).await; + let req = request(&state).await; + let mut messages = register(&state, &req.sandbox_id); + let mut output = run(&state, authed_request(req.clone())) + .await + .unwrap() + .into_inner(); + receive_relay(&state, &mut messages).await; + assert!(output.next().await.unwrap().is_err()); + assert!(output.next().await.is_none()); + drop(output); + drop(messages); + drop(state); + let state = setup(&url, &directory).await; + assert_eq!( + reason(&run(&state, authed_request(req.clone())).await.unwrap_err()), + "REQUEST_OUTCOME_UNCERTAIN" + ); + completion(&state).await.stream_terminal().await.unwrap(); + drop(state); + let state = setup(&url, &directory).await; + let mut sandbox: Sandbox = state + .store + .get_message(&req.sandbox_id) + .await + .unwrap() + .unwrap(); + sandbox.status.as_mut().unwrap().phase = SandboxPhase::Stopped.into(); + state.store.put_message(&sandbox).await.unwrap(); + assert_eq!( + reason(&run(&state, authed_request(req.clone())).await.unwrap_err()), + "REQUEST_STREAM_UNAVAILABLE" + ); + let mut unauthorized = authed_request(req.clone()); + if let Principal::User(user) = unauthorized + .extensions_mut() + .get_mut::() + .unwrap() + { + user.identity.roles.clear(); + } + assert_eq!( + run(&state, unauthorized).await.unwrap_err().code(), + tonic::Code::NotFound + ); + state + .store + .delete(Sandbox::object_type(), &req.sandbox_id) + .await + .unwrap(); + assert_eq!( + run(&state, authed_request(req)).await.unwrap_err().code(), + tonic::Code::NotFound + ); +} + +#[tokio::test] +async fn interactive_wire_start_is_admitted_under_its_own_method_namespace() { + use crate::grpc::OpenShellService; + use openshell_core::proto::{ + open_shell_client::OpenShellClient, open_shell_server::OpenShellServer, + }; + use tokio_stream::wrappers::TcpListenerStream; + let directory = tempfile::tempdir().unwrap(); + let state = setup("sqlite::memory:", &directory).await; + let req = request(&state).await; + let mut messages = register(&state, &req.sandbox_id); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let principal = authed_request(()) + .extensions() + .get::() + .unwrap() + .clone(); + let service = OpenShellServer::with_interceptor( + OpenShellService::new(state.clone()), + move |mut request: Request<()>| { + request.extensions_mut().insert(principal.clone()); + Ok(request) + }, + ); + let server = tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(service) + .serve_with_incoming(TcpListenerStream::new(listener)) + .await + .unwrap(); + }); + let mut client = OpenShellClient::connect(format!("http://{address}")) + .await + .unwrap(); + let start = ExecSandboxInput { + payload: Some(exec_sandbox_input::Payload::Start(req.clone())), + }; + let output = client + .exec_sandbox_interactive(tokio_stream::iter([start.clone()])) + .await + .unwrap(); + receive_relay(&state, &mut messages).await; + let owner = completion(&state).await; + assert_eq!( + reason( + &client + .exec_sandbox_interactive(tokio_stream::iter([start.clone()])) + .await + .unwrap_err() + ), + "REQUEST_OUTCOME_UNCERTAIN" + ); + owner.stream_terminal().await.unwrap(); + assert_eq!( + reason( + &client + .exec_sandbox_interactive(tokio_stream::iter([start])) + .await + .unwrap_err() + ), + "REQUEST_STREAM_UNAVAILABLE" + ); + // The same UUID belongs to a different RPC namespace, not another stdin stream. + drop(client.exec_sandbox(req).await.unwrap()); + receive_relay(&state, &mut messages).await; + assert_eq!( + state + .store + .list_by_type_after(OBJECT_TYPE, None, 100) + .await + .unwrap() + .len(), + 2 + ); + drop(output); + server.abort(); +} diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index ea470a7408..410124fae7 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -128,7 +128,7 @@ impl Drop for WatchSandboxStream { /// `NOT_FOUND` for both missing and unauthorized sandboxes so that callers /// cannot distinguish the two cases (CWE-203). pub(super) async fn fetch_and_authorize_sandbox( - state: &Arc, + state: &ServerState, principal: &crate::auth::principal::Principal, sandbox_id: &str, ) -> Result { @@ -1874,19 +1874,12 @@ pub(super) async fn handle_exec_sandbox( use openshell_core::ObjectId; let principal = super::extract_principal(&request)?; + let completion = request + .extensions() + .get::() + .cloned(); let req = request.into_inner(); - if req.sandbox_id.is_empty() { - return Err(Status::invalid_argument("sandbox_id is required")); - } - if req.command.is_empty() { - return Err(Status::invalid_argument("command is required")); - } - if req.environment.keys().any(|key| !is_valid_env_key(key)) { - return Err(Status::invalid_argument( - "environment keys must match ^[A-Za-z_][A-Za-z0-9_]*$", - )); - } - validate_exec_request_fields(&req)?; + validate_exec_start(&req)?; let sandbox = fetch_and_authorize_sandbox(state, &principal, &req.sandbox_id).await?; @@ -1940,6 +1933,7 @@ pub(super) async fn handle_exec_sandbox( no_login_shell, cols, rows, + completion, ) .await { @@ -2290,6 +2284,29 @@ async fn bridge_forward_tcp_stream( // Interactive exec handler (bidirectional stdin streaming) // --------------------------------------------------------------------------- +pub(super) fn validate_exec_start(req: &ExecSandboxRequest) -> Result<(), Status> { + use openshell_core::rpc_error; + if req.sandbox_id.is_empty() { + return Err(rpc_error::invalid_argument( + "sandbox_id", + "sandbox_id is required", + )); + } + if req.command.is_empty() { + return Err(rpc_error::invalid_argument( + "command", + "command is required", + )); + } + if req.environment.keys().any(|key| !is_valid_env_key(key)) { + return Err(rpc_error::invalid_argument( + "environment", + "environment keys must match ^[A-Za-z_][A-Za-z0-9_]*$", + )); + } + validate_exec_request_fields(req) +} + fn validate_interactive_exec_start( msg: Option, ) -> Result { @@ -2304,18 +2321,7 @@ fn validate_interactive_exec_start( )); }; - if req.sandbox_id.is_empty() { - return Err(Status::invalid_argument("sandbox_id is required")); - } - if req.command.is_empty() { - return Err(Status::invalid_argument("command is required")); - } - if req.environment.keys().any(|key| !is_valid_env_key(key)) { - return Err(Status::invalid_argument( - "environment keys must match ^[A-Za-z_][A-Za-z0-9_]*$", - )); - } - validate_exec_request_fields(&req)?; + validate_exec_start(&req)?; Ok(req) } @@ -2324,10 +2330,7 @@ pub(super) async fn handle_exec_sandbox_interactive( state: &Arc, request: Request>, ) -> Result>>, Status> { - use openshell_core::ObjectId; - - let principal = super::extract_principal(&request)?; - let mut input_stream = request.into_inner(); + let (metadata, extensions, mut input_stream) = request.into_parts(); let first_msg = input_stream .message() @@ -2336,6 +2339,46 @@ pub(super) async fn handle_exec_sandbox_interactive( let req = validate_interactive_exec_start(first_msg)?; + let mut request = Request::from_parts( + metadata, + extensions, + ExecSandboxInput { + payload: Some(openshell_core::proto::exec_sandbox_input::Payload::Start( + req, + )), + }, + ); + request + .extensions_mut() + .insert(super::mutation_replay::streaming::InteractiveInput( + Arc::new(tokio::sync::Mutex::new(Some(input_stream))), + )); + super::mutation_replay::run(state, request).await +} + +pub(super) async fn handle_exec_sandbox_interactive_start( + state: &Arc, + request: Request, +) -> Result>>, Status> { + let principal = super::extract_principal(&request)?; + let completion = request + .extensions() + .get::() + .cloned(); + let input = request + .extensions() + .get::() + .ok_or_else(|| Status::internal("interactive input handoff missing"))? + .clone(); + let input_stream = input + .0 + .lock() + .await + .take() + .ok_or_else(|| Status::internal("interactive input already claimed"))?; + let req = super::mutation_replay::streaming::start(request.get_ref())?; + validate_exec_start(req)?; + let sandbox = fetch_and_authorize_sandbox(state, &principal, &req.sandbox_id).await?; if SandboxPhase::try_from(sandbox.phase()).ok() != Some(SandboxPhase::Ready) { @@ -2348,7 +2391,7 @@ pub(super) async fn handle_exec_sandbox_interactive( .await .map_err(|e| Status::unavailable(format!("supervisor relay failed: {e}")))?; - let command_str = build_remote_exec_command(&req) + let command_str = build_remote_exec_command(req) .map_err(|e| Status::invalid_argument(format!("command construction failed: {e}")))?; let request_tty = req.tty; let no_login_shell = req.no_login_shell; @@ -2388,6 +2431,7 @@ pub(super) async fn handle_exec_sandbox_interactive( execution_timeout, cols, rows, + completion, ) .await { @@ -2695,6 +2739,7 @@ async fn stream_exec_over_relay( no_login_shell: bool, cols: u32, rows: u32, + completion: Option, ) -> Result<(), Status> { let command_preview: String = command .chars() @@ -2724,26 +2769,22 @@ async fn stream_exec_over_relay( tx.clone(), ); - let exec_result = if let Some(execution_timeout) = execution_timeout { - if let Ok(result) = tokio::time::timeout(execution_timeout, exec).await { - result - } else { - let _ = tx - .send(Ok(ExecSandboxEvent { - payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit( - ExecSandboxExit { exit_code: 124 }, - )), - })) - .await; - let _ = proxy_task.await; - return Ok(()); - } - } else { - exec.await - }; + let exec_result = wait_for_exec_terminal(exec, execution_timeout, completion).await; + if matches!(exec_result, Ok(None)) { + let _ = tx + .send(Ok(ExecSandboxEvent { + payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit( + ExecSandboxExit { exit_code: 124 }, + )), + })) + .await; + let _ = proxy_task.await; + return Ok(()); + } let exit_code = match exec_result { - Ok(code) => code, + Ok(Some(code)) => code, + Ok(None) => unreachable!("timeout returned above"), Err(status) => { let _ = proxy_task.await; return Err(status); @@ -2776,6 +2817,7 @@ async fn stream_interactive_exec_over_relay( execution_timeout: Option, cols: u32, rows: u32, + completion: Option, ) -> Result<(), Status> { let command_preview: String = command .chars() @@ -2805,26 +2847,22 @@ async fn stream_interactive_exec_over_relay( tx.clone(), ); - let exec_result = if let Some(execution_timeout) = execution_timeout { - if let Ok(result) = tokio::time::timeout(execution_timeout, exec).await { - result - } else { - let _ = tx - .send(Ok(ExecSandboxEvent { - payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit( - ExecSandboxExit { exit_code: 124 }, - )), - })) - .await; - let _ = proxy_task.await; - return Ok(()); - } - } else { - exec.await - }; + let exec_result = wait_for_exec_terminal(exec, execution_timeout, completion).await; + if matches!(exec_result, Ok(None)) { + let _ = tx + .send(Ok(ExecSandboxEvent { + payload: Some(openshell_core::proto::exec_sandbox_event::Payload::Exit( + ExecSandboxExit { exit_code: 124 }, + )), + })) + .await; + let _ = proxy_task.await; + return Ok(()); + } let exit_code = match exec_result { - Ok(code) => code, + Ok(Some(code)) => code, + Ok(None) => unreachable!("timeout returned above"), Err(status) => { let _ = proxy_task.await; return Err(status); @@ -2844,6 +2882,29 @@ async fn stream_interactive_exec_over_relay( Ok(()) } +/// `Ok(Some(code))` requires a real SSH `ExitStatus`, including nonzero codes. +/// Synthetic gateway timeouts, lost status, and dropped futures leave the claim +/// unresolved. Finalization precedes terminal delivery to the client. +pub(super) async fn wait_for_exec_terminal( + exec: impl Future>, + execution_timeout: Option, + completion: Option, +) -> Result, Status> { + let result = if let Some(execution_timeout) = execution_timeout { + match tokio::time::timeout(execution_timeout, exec).await { + Ok(result) => result, + Err(_) => return Ok(None), + } + } else { + exec.await + }; + let exit_code = result?; + if let Some(completion) = completion { + completion.stream_terminal().await?; + } + Ok(Some(exit_code)) +} + #[allow(clippy::too_many_arguments)] async fn run_interactive_exec_with_russh( local_proxy_port: u16, diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index 5d168322e4..2dbc4d3f08 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -118,7 +118,7 @@ mod tests { const STORAGE_V1_SCHEMA_SHA256: &str = "574bf5fcff731bd6e3fd84ed3f124161035bd236ef0fb7e32b4d8a8c55ceba5e"; const PUBLIC_RPC_SCHEMA_SHA256: &str = - "1c72f65167a5ceb00cd17043079b92324ca55d19b547c86827b51bd7376cb23c"; + "9df91739f47f743a271536847486b7be457f66bb37370e5537ad4e7e69b91fdc"; const DURABLE_SCHEMA_SHA256: &str = "65066c0b0eef57a4c708f20fcbbb8e8f47376da9f4bf73dfc3bca0b3df174ba8"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = diff --git a/docs/reference/api-errors.mdx b/docs/reference/api-errors.mdx index 3ad2c98a3f..93e51b5239 100644 --- a/docs/reference/api-errors.mdx +++ b/docs/reference/api-errors.mdx @@ -34,6 +34,7 @@ Recognized gateway reasons include the following. | `REQUEST_ID_PAYLOAD_MISMATCH` | `FAILED_PRECONDITION` | Keep the original payload for that request ID. Inspect the original operation before submitting a different one. | | `REQUEST_OUTCOME_UNCERTAIN` | `FAILED_PRECONDITION` | An attempt is admitted but has no confirmed replayable success. Observe resource state and reconcile effects. Do not switch to a new ID to bypass the admission. | | `REQUEST_REPLAY_UNAVAILABLE` | `FAILED_PRECONDITION` | The original scope, resource, interceptor transformation, or fingerprint key is no longer replayable. Reconcile effects; the gateway does not execute the request again. Missing private-key material can also reject admission before work starts. | +| `REQUEST_STREAM_UNAVAILABLE` | `FAILED_PRECONDITION` | The exec attempt terminated, but its output was not stored. Reconcile command effects; the gateway does not launch it again or recreate its stream. | ## Status and retry guidance @@ -183,12 +184,48 @@ The gateway also limits detached admission workers to 64 per process and these request messages to 4 MiB. Supply IDs through generated/raw RPC clients. Curated SDK request-ID helpers and -automatic retry policies are not part of this contract yet. Exec, SSH sessions, +automatic retry policies are not part of this contract yet. SSH sessions, rootfs staging, and supervisor-authenticated mutations do not gain deduplication from this feature. Sandbox principals cannot opt into `UpdateConfig` admission. Check gateway compatibility before relying on IDs; older protobuf servers can silently ignore unknown fields. +## Exec launch admission + +`ExecSandbox` and `ExecSandboxInteractive` accept an optional `request_id` in +`ExecSandboxRequest`. For interactive exec, supply it in the initial `start` +message. The UUID rules, keyed fingerprints, current authorization checks, +per-caller quota, and request-size limit above also apply to exec. + +The gateway admits one owner before opening the supervisor relay. Both RPCs +fingerprint the command, sandbox UUID, environment, initial stdin, and other +start fields, excluding the request ID. They use separate method namespaces. +Later interactive stdin and resize messages are not part of the fingerprint and +cannot be submitted through a duplicate stream. + +| Admitted attempt | Duplicate response | +|---|---| +| Still running, interrupted, timed out, or missing confirmed exit status | `REQUEST_OUTCOME_UNCERTAIN`. The record does not expire or permit another launch. | +| Confirmed remote exit, including a nonzero exit code | `REQUEST_STREAM_UNAVAILABLE` for 24 hours after durable completion. No output, exit code, stream attachment, or stdin is replayed. | +| Changed start payload | `REQUEST_ID_PAYLOAD_MISMATCH`. The changed command does not run. | + +The terminal receipt contains only a marker and completion time, not commands, +environment values, stdin, stdout, or stderr. Receiving a stream handle is not +completion. A gateway-generated timeout exit code of `124` does not prove the +remote process stopped, so it leaves the admission unresolved. A confirmed remote +exit code of `124` is a terminal result. + +Cancellation does not undo admission. Existing transport behavior still applies +to the original execution, including interactive SSH closure on disconnect; the +gateway does not promise that a disconnected command continues. When it loses +exit confirmation, the record remains unresolved even after restart. Current +authorization and the original sandbox UUID are checked before duplicate lookup. + +After confirmed terminal retention expires, the same ID can start a new command. +Do not reuse an old ID without reconciling its effects. Empty IDs keep the existing +streaming behavior without launch deduplication. Use generated/raw RPC clients +for this opt-in contract; curated SDK helpers and automatic retries are separate. + ## Deletion outcomes Delete, membership-removal, and SSH-revocation RPCs return a typed outcome. diff --git a/e2e/python/test_exec_admission.py b/e2e/python/test_exec_admission.py new file mode 100644 index 0000000000..5d81b7c681 --- /dev/null +++ b/e2e/python/test_exec_admission.py @@ -0,0 +1,154 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import threading +import time +import uuid +from typing import TYPE_CHECKING + +import grpc +import pytest +from google.protobuf import duration_pb2 + +from openshell._proto import openshell_pb2 +from openshell.errors import from_grpc_error + +if TYPE_CHECKING: + from collections.abc import Callable + + from openshell import Sandbox, SandboxClient + + +def assert_blocked(stream, reason: str) -> None: + with pytest.raises(grpc.RpcError) as caught: + list(stream) + error = from_grpc_error(caught.value) + assert error.code() == grpc.StatusCode.FAILED_PRECONDITION + assert error.error_info is not None + assert error.error_info.reason == reason + + +@pytest.mark.parametrize("interactive", [False, True]) +def test_exec_request_id_never_relaunches_or_replays_output( + sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient, interactive: bool +) -> None: + with sandbox(delete_on_exit=True) as sb: + path = f"/sandbox/exec-admission-{uuid.uuid4().hex}" + request = openshell_pb2.ExecSandboxRequest( + sandbox_id=sb.id, + command=[ + "/bin/sh", + "-c", + f"printf x >> {path}; printf started; " + f"while [ ! -e {path}.release ]; do sleep 0.1; done; " + "printf finished; exit 7", + ], + request_id=str(uuid.uuid4()), + ) + done = threading.Event() + + def invoke(value): + if not interactive: + return sandbox_client._stub.ExecSandbox(value, timeout=30) + + def inputs(): + yield openshell_pb2.ExecSandboxInput(start=value) + done.wait(timeout=30) + + return sandbox_client._stub.ExecSandboxInteractive(inputs(), timeout=30) + + stream = invoke(request) + try: + first = next(stream) + assert first.WhichOneof("payload") == "stdout" + assert b"started" in first.stdout.data + assert_blocked(invoke(request), "REQUEST_OUTCOME_UNCERTAIN") + changed = openshell_pb2.ExecSandboxRequest() + changed.CopyFrom(request) + changed.command.append("changed-payload") + assert_blocked(invoke(changed), "REQUEST_ID_PAYLOAD_MISMATCH") + assert sb.exec(["touch", f"{path}.release"]).exit_code == 0 + events = list(stream) + assert events[-1].WhichOneof("payload") == "exit" + assert events[-1].exit.exit_code == 7 + assert_blocked(invoke(request), "REQUEST_STREAM_UNAVAILABLE") + result = sb.exec(["cat", path]) + assert result.exit_code == 0 + assert result.stdout == "x" + finally: + done.set() + stream.cancel() + + +@pytest.mark.parametrize("interactive", [False, True]) +def test_exec_request_timeout_keeps_launch_unresolved( + sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient, interactive: bool +) -> None: + with sandbox(delete_on_exit=True) as sb: + path = f"/sandbox/exec-timeout-{uuid.uuid4().hex}" + request = openshell_pb2.ExecSandboxRequest( + sandbox_id=sb.id, + command=["/bin/sh", "-c", f"printf x >> {path}; sleep 5"], + execution_timeout=duration_pb2.Duration(seconds=1), + request_id=str(uuid.uuid4()), + ) + done = threading.Event() + + def invoke(): + if not interactive: + return sandbox_client._stub.ExecSandbox(request, timeout=30) + + def inputs(): + yield openshell_pb2.ExecSandboxInput(start=request) + done.wait(timeout=30) + + return sandbox_client._stub.ExecSandboxInteractive(inputs(), timeout=30) + + stream = invoke() + try: + events = [] + for event in stream: + events.append(event) + if event.WhichOneof("payload") == "exit": + # A synthetic timeout leaves the interactive input task + # alive until the client closes its side of the stream. + done.set() + assert events[-1].exit.exit_code == 124 + assert_blocked(invoke(), "REQUEST_OUTCOME_UNCERTAIN") + assert sb.exec(["cat", path]).stdout == "x" + finally: + done.set() + stream.cancel() + + +def test_exec_client_cancellation_does_not_clear_launch_admission( + sandbox: Callable[..., Sandbox], sandbox_client: SandboxClient +) -> None: + with sandbox(delete_on_exit=True) as sb: + path = f"/sandbox/exec-cancel-{uuid.uuid4().hex}" + request = openshell_pb2.ExecSandboxRequest( + sandbox_id=sb.id, + command=[ + "/bin/sh", + "-c", + f"printf x >> {path}; printf started; sleep 1; printf finished", + ], + request_id=str(uuid.uuid4()), + ) + stream = sandbox_client._stub.ExecSandbox(request, timeout=30) + assert b"started" in next(stream).stdout.data + stream.cancel() + deadline = time.monotonic() + 15 + while True: + with pytest.raises(grpc.RpcError) as caught: + list(sandbox_client._stub.ExecSandbox(request, timeout=30)) + error = from_grpc_error(caught.value) + assert error.error_info is not None + if error.error_info.reason == "REQUEST_STREAM_UNAVAILABLE": + break + assert error.error_info.reason == "REQUEST_OUTCOME_UNCERTAIN" + assert time.monotonic() < deadline + time.sleep(0.1) + assert sb.exec(["cat", path]).stdout == "x" diff --git a/proto/openshell.proto b/proto/openshell.proto index 0e9fe429a6..111eda6952 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1613,6 +1613,12 @@ message ExecSandboxRequest { // sourced by them) are applied. When true, the command runs without those // files (`bash -c`), for automation that needs predictable startup behavior. bool no_login_shell = 10; + + // Optional nonzero UUID for durable launch admission. Also applies to the + // initial ExecSandboxInteractive start message. Duplicates never relaunch, + // reattach, or replay output/stdin. Unconfirmed executions remain fenced; + // confirmed terminal executions retain the fence for 24 hours. + string request_id = 11; } // One stdout chunk from a sandbox exec. diff --git a/sdk/go/proto/openshellv1/openshell.pb.go b/sdk/go/proto/openshellv1/openshell.pb.go index 310df01110..9fbaf8cc2e 100644 --- a/sdk/go/proto/openshellv1/openshell.pb.go +++ b/sdk/go/proto/openshellv1/openshell.pb.go @@ -5014,7 +5014,12 @@ type ExecSandboxRequest struct { // (`bash -lc`) so user startup files (.bash_profile/.profile, and .bashrc if // sourced by them) are applied. When true, the command runs without those // files (`bash -c`), for automation that needs predictable startup behavior. - NoLoginShell bool `protobuf:"varint,10,opt,name=no_login_shell,json=noLoginShell,proto3" json:"no_login_shell,omitempty"` + NoLoginShell bool `protobuf:"varint,10,opt,name=no_login_shell,json=noLoginShell,proto3" json:"no_login_shell,omitempty"` + // Optional nonzero UUID for durable launch admission. Also applies to the + // initial ExecSandboxInteractive start message. Duplicates never relaunch, + // reattach, or replay output/stdin. Unconfirmed executions remain fenced; + // confirmed terminal executions retain the fence for 24 hours. + RequestId string `protobuf:"bytes,11,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -5119,6 +5124,13 @@ func (x *ExecSandboxRequest) GetNoLoginShell() bool { return false } +func (x *ExecSandboxRequest) GetRequestId() string { + if x != nil { + return x.RequestId + } + return "" +} + // One stdout chunk from a sandbox exec. type ExecSandboxStdout struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -15506,7 +15518,7 @@ const file_openshell_proto_rawDesc = "" + "\x05token\x18\x01 \x01(\tB\x04\x88\xb5\x18\x01R\x05token\x12#\n" + "\rallow_missing\x18\x02 \x01(\bR\fallowMissing\"b\n" + "\x18RevokeSshSessionResponse\x127\n" + - "\aoutcome\x18\x02 \x01(\x0e2\x1d.openshell.v1.DeletionOutcomeR\aoutcomeJ\x04\b\x01\x10\x02R\arevoked\"\xd1\x03\n" + + "\aoutcome\x18\x02 \x01(\x0e2\x1d.openshell.v1.DeletionOutcomeR\aoutcomeJ\x04\b\x01\x10\x02R\arevoked\"\xf0\x03\n" + "\x12ExecSandboxRequest\x12\x1d\n" + "\n" + "sandbox_id\x18\x01 \x01(\tR\tsandboxId\x12\x18\n" + @@ -15519,7 +15531,9 @@ const file_openshell_proto_rawDesc = "" + "\x04cols\x18\b \x01(\rR\x04cols\x12\x12\n" + "\x04rows\x18\t \x01(\rR\x04rows\x12$\n" + "\x0eno_login_shell\x18\n" + - " \x01(\bR\fnoLoginShell\x1a>\n" + + " \x01(\bR\fnoLoginShell\x12\x1d\n" + + "\n" + + "request_id\x18\v \x01(\tR\trequestId\x1a>\n" + "\x10EnvironmentEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01J\x04\b\x05\x10\x06R\x0ftimeout_seconds\"'\n" +