Skip to content

Commit 442b0b6

Browse files
benoitfbenoitf
authored
feat(exec): add bidirectional streaming for interactive TTY sessions (#1331)
* feat(exec): add bidirectional streaming for interactive TTY sessions The existing ExecSandbox RPC sends stdin upfront in the request body, making interactive programs (bash, top, vim) unusable with --tty. This adds an ExecSandboxInteractive bidirectional streaming RPC that forwards live keystrokes, terminal resize events, and EOF to the sandbox. - Add ExecSandboxInteractive RPC, ExecSandboxInput and ExecSandboxWindowResize messages to the proto definition - Implement server handler with relay bridging, PTY allocation via russh channel.split(), and timeout support - Implement CLI client with raw mode, spawn_blocking stdin reader, SIGWINCH resize forwarding, and process::exit for clean shutdown - Route to interactive path only when --tty is explicitly passed Signed-off-by: Florent Benoit <fbenoit@redhat.com> * fix(exec): address interactive TTY streaming review feedback Server-side (sandbox.rs): - Close SSH channel after EOF so PTY-backed programs (bash, vim, top) terminate on client disconnect instead of leaking the session - Break output loop when tx.send() fails (client gone) to tear down the streaming pipeline promptly - Default zero cols/rows to 80x24 before request_pty, matching the proto "0 = use default" contract and preventing 1x1 terminals for non-CLI clients Client-side (run.rs): - Replace spawn_blocking stdin reader with a detached std::thread so the tokio runtime shutdown does not hang on a thread blocked in stdin.read(). This removes the need for std::process::exit(), allowing save_last_sandbox() in the caller to execute normally - Switch from futures::channel::mpsc with try_send() to tokio::sync::mpsc with blocking_send() for proper backpressure — try_send silently stopped forwarding all input when the channel filled during paste or slow network - Add TaskGuard (abort-on-drop) for the resize task to ensure cleanup on all return paths including early errors --------- Signed-off-by: Florent Benoit <fbenoit@redhat.com>
1 parent 63bdcd1 commit 442b0b6

15 files changed

Lines changed: 809 additions & 52 deletions

‎crates/openshell-cli/src/run.rs‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2497,6 +2497,11 @@ pub async fn sandbox_exec_grpc(
24972497
let tty = tty_override
24982498
.unwrap_or_else(|| std::io::stdin().is_terminal() && std::io::stdout().is_terminal());
24992499

2500+
if tty_override == Some(true) && std::io::stdin().is_terminal() {
2501+
return sandbox_exec_interactive_grpc(client, &sandbox, command, workdir, timeout_seconds)
2502+
.await;
2503+
}
2504+
25002505
// Make the streaming gRPC call.
25012506
let mut stream = client
25022507
.exec_sandbox(ExecSandboxRequest {
@@ -2507,6 +2512,7 @@ pub async fn sandbox_exec_grpc(
25072512
timeout_seconds,
25082513
stdin: stdin_payload,
25092514
tty,
2515+
..Default::default()
25102516
})
25112517
.await
25122518
.into_diagnostic()?
@@ -2826,6 +2832,154 @@ async fn drain_and_shutdown_local_socket(mut socket: tokio::net::TcpStream) {
28262832
let _ = socket.shutdown().await;
28272833
}
28282834

2835+
struct RawModeGuard;
2836+
2837+
impl Drop for RawModeGuard {
2838+
fn drop(&mut self) {
2839+
let _ = crossterm::terminal::disable_raw_mode();
2840+
}
2841+
}
2842+
2843+
struct TaskGuard(tokio::task::JoinHandle<()>);
2844+
2845+
impl Drop for TaskGuard {
2846+
fn drop(&mut self) {
2847+
self.0.abort();
2848+
}
2849+
}
2850+
2851+
async fn sandbox_exec_interactive_grpc(
2852+
mut client: crate::tls::GrpcClient,
2853+
sandbox: &Sandbox,
2854+
command: &[String],
2855+
workdir: Option<&str>,
2856+
timeout_seconds: u32,
2857+
) -> Result<i32> {
2858+
use openshell_core::proto::{ExecSandboxInput, ExecSandboxWindowResize, exec_sandbox_input};
2859+
use tokio_stream::wrappers::ReceiverStream;
2860+
2861+
let (cols, rows) = crossterm::terminal::size().unwrap_or((80, 24));
2862+
2863+
let (input_tx, input_rx) = tokio::sync::mpsc::channel::<ExecSandboxInput>(4096);
2864+
2865+
// Send the start message with exec metadata.
2866+
input_tx
2867+
.send(ExecSandboxInput {
2868+
payload: Some(exec_sandbox_input::Payload::Start(ExecSandboxRequest {
2869+
sandbox_id: sandbox.object_id().to_string(),
2870+
command: command.to_vec(),
2871+
workdir: workdir.unwrap_or_default().to_string(),
2872+
environment: HashMap::new(),
2873+
timeout_seconds,
2874+
stdin: Vec::new(),
2875+
tty: true,
2876+
cols: u32::from(cols),
2877+
rows: u32::from(rows),
2878+
})),
2879+
})
2880+
.await
2881+
.into_diagnostic()?;
2882+
2883+
let mut stream = client
2884+
.exec_sandbox_interactive(ReceiverStream::new(input_rx))
2885+
.await
2886+
.into_diagnostic()?
2887+
.into_inner();
2888+
2889+
// Enable raw mode so keystrokes are forwarded immediately.
2890+
crossterm::terminal::enable_raw_mode().into_diagnostic()?;
2891+
let raw_guard = RawModeGuard;
2892+
2893+
// Stdin reader on a detached OS thread. Using std::thread (not
2894+
// spawn_blocking) so the tokio runtime shutdown doesn't wait for a
2895+
// thread blocked on stdin.read(). The thread exits when the channel
2896+
// closes (blocking_send returns Err) or stdin hits EOF.
2897+
#[cfg(unix)]
2898+
{
2899+
let stdin_tx = input_tx.clone();
2900+
std::thread::spawn(move || {
2901+
let mut stdin = std::io::stdin().lock();
2902+
let mut buf = [0u8; 4096];
2903+
loop {
2904+
match stdin.read(&mut buf) {
2905+
Ok(0) | Err(_) => break,
2906+
Ok(n) => {
2907+
if stdin_tx
2908+
.blocking_send(ExecSandboxInput {
2909+
payload: Some(exec_sandbox_input::Payload::Stdin(
2910+
buf[..n].to_vec(),
2911+
)),
2912+
})
2913+
.is_err()
2914+
{
2915+
break;
2916+
}
2917+
}
2918+
}
2919+
}
2920+
});
2921+
}
2922+
2923+
// SIGWINCH handler: forward terminal resize events.
2924+
#[cfg(unix)]
2925+
let resize_task = {
2926+
let resize_tx = input_tx.clone();
2927+
tokio::spawn(async move {
2928+
let mut sig =
2929+
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::window_change())
2930+
.expect("failed to register SIGWINCH handler");
2931+
while sig.recv().await.is_some() {
2932+
if let Ok((c, r)) = crossterm::terminal::size() {
2933+
let msg = ExecSandboxInput {
2934+
payload: Some(exec_sandbox_input::Payload::Resize(
2935+
ExecSandboxWindowResize {
2936+
cols: u32::from(c),
2937+
rows: u32::from(r),
2938+
},
2939+
)),
2940+
};
2941+
if resize_tx.send(msg).await.is_err() {
2942+
break;
2943+
}
2944+
}
2945+
}
2946+
})
2947+
};
2948+
#[cfg(unix)]
2949+
let _resize_guard = TaskGuard(resize_task);
2950+
2951+
let mut exit_code = 0i32;
2952+
let stdout = std::io::stdout();
2953+
2954+
while let Some(event) = stream.next().await {
2955+
let event = event.into_diagnostic()?;
2956+
match event.payload {
2957+
Some(exec_sandbox_event::Payload::Stdout(out)) => {
2958+
let mut handle = stdout.lock();
2959+
handle.write_all(&out.data).into_diagnostic()?;
2960+
handle.flush().into_diagnostic()?;
2961+
}
2962+
Some(exec_sandbox_event::Payload::Stderr(err)) => {
2963+
let mut handle = stdout.lock();
2964+
handle.write_all(&err.data).into_diagnostic()?;
2965+
handle.flush().into_diagnostic()?;
2966+
}
2967+
Some(exec_sandbox_event::Payload::Exit(exit)) => {
2968+
exit_code = exit.exit_code;
2969+
break;
2970+
}
2971+
None => {}
2972+
}
2973+
}
2974+
2975+
drop(input_tx);
2976+
2977+
// Drop the raw mode guard to restore the terminal before returning.
2978+
drop(raw_guard);
2979+
2980+
Ok(exit_code)
2981+
}
2982+
28292983
/// Print a single YAML line with dimmed keys and regular values.
28302984
fn print_yaml_line(line: &str) {
28312985
// Find leading whitespace

‎crates/openshell-cli/tests/ensure_providers_integration.rs‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -18,14 +18,15 @@ use openshell_core::proto::{
1818
CreateSandboxRequest, CreateSshSessionRequest, CreateSshSessionResponse, DeleteProviderRequest,
1919
DeleteProviderResponse, DeleteSandboxRequest, DeleteSandboxResponse,
2020
DetachSandboxProviderRequest, DetachSandboxProviderResponse, ExecSandboxEvent,
21-
ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest, GetGatewayConfigResponse,
22-
GetProviderRequest, GetSandboxConfigRequest, GetSandboxConfigResponse,
23-
GetSandboxProviderEnvironmentRequest, GetSandboxProviderEnvironmentResponse, GetSandboxRequest,
24-
HealthRequest, HealthResponse, ListProvidersRequest, ListProvidersResponse,
25-
ListSandboxProvidersRequest, ListSandboxProvidersResponse, ListSandboxesRequest,
26-
ListSandboxesResponse, Provider, ProviderResponse, RevokeSshSessionRequest,
27-
RevokeSshSessionResponse, SandboxResponse, SandboxStreamEvent, ServiceStatus,
28-
SupervisorMessage, UpdateProviderRequest, WatchSandboxRequest,
21+
ExecSandboxInput, ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest,
22+
GetGatewayConfigResponse, GetProviderRequest, GetSandboxConfigRequest,
23+
GetSandboxConfigResponse, GetSandboxProviderEnvironmentRequest,
24+
GetSandboxProviderEnvironmentResponse, GetSandboxRequest, HealthRequest, HealthResponse,
25+
ListProvidersRequest, ListProvidersResponse, ListSandboxProvidersRequest,
26+
ListSandboxProvidersResponse, ListSandboxesRequest, ListSandboxesResponse, Provider,
27+
ProviderResponse, RevokeSshSessionRequest, RevokeSshSessionResponse, SandboxResponse,
28+
SandboxStreamEvent, ServiceStatus, SupervisorMessage, UpdateProviderRequest,
29+
WatchSandboxRequest,
2930
};
3031
use openshell_core::{ObjectId, ObjectName};
3132
use std::collections::HashMap;
@@ -373,6 +374,15 @@ impl OpenShell for TestOpenShell {
373374
)))
374375
}
375376

377+
type ExecSandboxInteractiveStream =
378+
tokio_stream::wrappers::ReceiverStream<Result<ExecSandboxEvent, Status>>;
379+
async fn exec_sandbox_interactive(
380+
&self,
381+
_request: tonic::Request<tonic::Streaming<ExecSandboxInput>>,
382+
) -> Result<Response<Self::ExecSandboxInteractiveStream>, Status> {
383+
Err(Status::unimplemented("not implemented in test"))
384+
}
385+
376386
async fn update_config(
377387
&self,
378388
_request: tonic::Request<openshell_core::proto::UpdateConfigRequest>,

‎crates/openshell-cli/tests/mtls_integration.rs‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -9,10 +9,10 @@ use helpers::{
99
use openshell_cli::tls::{TlsOptions, grpc_client};
1010
use openshell_core::proto::{
1111
CreateProviderRequest, CreateSshSessionRequest, CreateSshSessionResponse,
12-
DeleteProviderRequest, DeleteProviderResponse, ExecSandboxEvent, ExecSandboxRequest,
13-
GetProviderRequest, HealthRequest, HealthResponse, ListProvidersRequest, ListProvidersResponse,
14-
ProviderResponse, RevokeSshSessionRequest, RevokeSshSessionResponse, ServiceStatus,
15-
UpdateProviderRequest,
12+
DeleteProviderRequest, DeleteProviderResponse, ExecSandboxEvent, ExecSandboxInput,
13+
ExecSandboxRequest, GetProviderRequest, HealthRequest, HealthResponse, ListProvidersRequest,
14+
ListProvidersResponse, ProviderResponse, RevokeSshSessionRequest, RevokeSshSessionResponse,
15+
ServiceStatus, UpdateProviderRequest,
1616
open_shell_server::{OpenShell, OpenShellServer},
1717
};
1818
use tempfile::tempdir;
@@ -283,6 +283,15 @@ impl OpenShell for TestOpenShell {
283283
)))
284284
}
285285

286+
type ExecSandboxInteractiveStream =
287+
tokio_stream::wrappers::ReceiverStream<Result<ExecSandboxEvent, Status>>;
288+
async fn exec_sandbox_interactive(
289+
&self,
290+
_request: tonic::Request<tonic::Streaming<ExecSandboxInput>>,
291+
) -> Result<Response<Self::ExecSandboxInteractiveStream>, Status> {
292+
Err(Status::unimplemented("not implemented in test"))
293+
}
294+
286295
async fn update_config(
287296
&self,
288297
_request: tonic::Request<openshell_core::proto::UpdateConfigRequest>,

‎crates/openshell-cli/tests/provider_commands_integration.rs‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,14 +14,15 @@ use openshell_core::proto::{
1414
CreateSandboxRequest, CreateSshSessionRequest, CreateSshSessionResponse, DeleteProviderRequest,
1515
DeleteProviderResponse, DeleteSandboxRequest, DeleteSandboxResponse,
1616
DetachSandboxProviderRequest, DetachSandboxProviderResponse, ExecSandboxEvent,
17-
ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest, GetGatewayConfigResponse,
18-
GetProviderRequest, GetSandboxConfigRequest, GetSandboxConfigResponse,
19-
GetSandboxProviderEnvironmentRequest, GetSandboxProviderEnvironmentResponse, GetSandboxRequest,
20-
HealthRequest, HealthResponse, ListProvidersRequest, ListProvidersResponse,
21-
ListSandboxProvidersRequest, ListSandboxProvidersResponse, ListSandboxesRequest,
22-
ListSandboxesResponse, Provider, ProviderProfile, ProviderResponse, RevokeSshSessionRequest,
23-
RevokeSshSessionResponse, SandboxResponse, SandboxStreamEvent, ServiceStatus,
24-
SupervisorMessage, UpdateProviderRequest, WatchSandboxRequest,
17+
ExecSandboxInput, ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest,
18+
GetGatewayConfigResponse, GetProviderRequest, GetSandboxConfigRequest,
19+
GetSandboxConfigResponse, GetSandboxProviderEnvironmentRequest,
20+
GetSandboxProviderEnvironmentResponse, GetSandboxRequest, HealthRequest, HealthResponse,
21+
ListProvidersRequest, ListProvidersResponse, ListSandboxProvidersRequest,
22+
ListSandboxProvidersResponse, ListSandboxesRequest, ListSandboxesResponse, Provider,
23+
ProviderProfile, ProviderResponse, RevokeSshSessionRequest, RevokeSshSessionResponse,
24+
SandboxResponse, SandboxStreamEvent, ServiceStatus, SupervisorMessage, UpdateProviderRequest,
25+
WatchSandboxRequest,
2526
};
2627
use openshell_core::{ObjectId, ObjectName};
2728
use std::collections::HashMap;
@@ -505,6 +506,15 @@ impl OpenShell for TestOpenShell {
505506
)))
506507
}
507508

509+
type ExecSandboxInteractiveStream =
510+
tokio_stream::wrappers::ReceiverStream<Result<ExecSandboxEvent, Status>>;
511+
async fn exec_sandbox_interactive(
512+
&self,
513+
_request: tonic::Request<tonic::Streaming<ExecSandboxInput>>,
514+
) -> Result<Response<Self::ExecSandboxInteractiveStream>, Status> {
515+
Err(Status::unimplemented("not implemented in test"))
516+
}
517+
508518
async fn update_config(
509519
&self,
510520
_request: tonic::Request<openshell_core::proto::UpdateConfigRequest>,

‎crates/openshell-cli/tests/sandbox_create_lifecycle_integration.rs‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,15 @@ use openshell_core::proto::{
1515
CreateSandboxRequest, CreateSshSessionRequest, CreateSshSessionResponse, DeleteProviderRequest,
1616
DeleteProviderResponse, DeleteSandboxRequest, DeleteSandboxResponse,
1717
DetachSandboxProviderRequest, DetachSandboxProviderResponse, ExecSandboxEvent,
18-
ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest, GetGatewayConfigResponse,
19-
GetProviderRequest, GetSandboxConfigRequest, GetSandboxConfigResponse,
20-
GetSandboxProviderEnvironmentRequest, GetSandboxProviderEnvironmentResponse, GetSandboxRequest,
21-
HealthRequest, HealthResponse, ListProvidersRequest, ListProvidersResponse,
22-
ListSandboxProvidersRequest, ListSandboxProvidersResponse, ListSandboxesRequest,
23-
ListSandboxesResponse, PlatformEvent, ProviderResponse, RevokeSshSessionRequest,
24-
RevokeSshSessionResponse, Sandbox, SandboxPhase, SandboxResponse, SandboxStreamEvent,
25-
ServiceStatus, SupervisorMessage, UpdateProviderRequest, WatchSandboxRequest,
26-
sandbox_stream_event,
18+
ExecSandboxInput, ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest,
19+
GetGatewayConfigResponse, GetProviderRequest, GetSandboxConfigRequest,
20+
GetSandboxConfigResponse, GetSandboxProviderEnvironmentRequest,
21+
GetSandboxProviderEnvironmentResponse, GetSandboxRequest, HealthRequest, HealthResponse,
22+
ListProvidersRequest, ListProvidersResponse, ListSandboxProvidersRequest,
23+
ListSandboxProvidersResponse, ListSandboxesRequest, ListSandboxesResponse, PlatformEvent,
24+
ProviderResponse, RevokeSshSessionRequest, RevokeSshSessionResponse, Sandbox, SandboxPhase,
25+
SandboxResponse, SandboxStreamEvent, ServiceStatus, SupervisorMessage, UpdateProviderRequest,
26+
WatchSandboxRequest, sandbox_stream_event,
2727
};
2828
use std::collections::HashMap;
2929
use std::fs;
@@ -357,6 +357,15 @@ impl OpenShell for TestOpenShell {
357357
)))
358358
}
359359

360+
type ExecSandboxInteractiveStream =
361+
tokio_stream::wrappers::ReceiverStream<Result<ExecSandboxEvent, Status>>;
362+
async fn exec_sandbox_interactive(
363+
&self,
364+
_request: tonic::Request<tonic::Streaming<ExecSandboxInput>>,
365+
) -> Result<Response<Self::ExecSandboxInteractiveStream>, Status> {
366+
Err(Status::unimplemented("not implemented in test"))
367+
}
368+
360369
async fn update_config(
361370
&self,
362371
_request: tonic::Request<openshell_core::proto::UpdateConfigRequest>,

‎crates/openshell-cli/tests/sandbox_name_fallback_integration.rs‎

Lines changed: 17 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,14 @@ use openshell_core::proto::{
1515
CreateSandboxRequest, CreateSshSessionRequest, CreateSshSessionResponse, DeleteProviderRequest,
1616
DeleteProviderResponse, DeleteSandboxRequest, DeleteSandboxResponse,
1717
DetachSandboxProviderRequest, DetachSandboxProviderResponse, ExecSandboxEvent,
18-
ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest, GetGatewayConfigResponse,
19-
GetProviderRequest, GetSandboxConfigRequest, GetSandboxConfigResponse,
20-
GetSandboxProviderEnvironmentRequest, GetSandboxProviderEnvironmentResponse, GetSandboxRequest,
21-
HealthRequest, HealthResponse, ListProvidersRequest, ListProvidersResponse,
22-
ListSandboxProvidersRequest, ListSandboxProvidersResponse, ListSandboxesRequest,
23-
ListSandboxesResponse, ProviderResponse, Sandbox, SandboxPolicy, SandboxResponse,
24-
SandboxStreamEvent, ServiceStatus, SupervisorMessage, UpdateProviderRequest,
25-
WatchSandboxRequest,
18+
ExecSandboxInput, ExecSandboxRequest, GatewayMessage, GetGatewayConfigRequest,
19+
GetGatewayConfigResponse, GetProviderRequest, GetSandboxConfigRequest,
20+
GetSandboxConfigResponse, GetSandboxProviderEnvironmentRequest,
21+
GetSandboxProviderEnvironmentResponse, GetSandboxRequest, HealthRequest, HealthResponse,
22+
ListProvidersRequest, ListProvidersResponse, ListSandboxProvidersRequest,
23+
ListSandboxProvidersResponse, ListSandboxesRequest, ListSandboxesResponse, ProviderResponse,
24+
Sandbox, SandboxPolicy, SandboxResponse, SandboxStreamEvent, ServiceStatus, SupervisorMessage,
25+
UpdateProviderRequest, WatchSandboxRequest,
2626
};
2727
use std::sync::Arc;
2828
use tempfile::TempDir;
@@ -295,6 +295,15 @@ impl OpenShell for TestOpenShell {
295295
)))
296296
}
297297

298+
type ExecSandboxInteractiveStream =
299+
tokio_stream::wrappers::ReceiverStream<Result<ExecSandboxEvent, Status>>;
300+
async fn exec_sandbox_interactive(
301+
&self,
302+
_request: tonic::Request<tonic::Streaming<ExecSandboxInput>>,
303+
) -> Result<Response<Self::ExecSandboxInteractiveStream>, Status> {
304+
Err(Status::unimplemented("not implemented in test"))
305+
}
306+
298307
async fn update_config(
299308
&self,
300309
_request: tonic::Request<openshell_core::proto::UpdateConfigRequest>,

0 commit comments

Comments
 (0)