mirror of
https://github.com/NVIDIA/OpenShell.git
synced 2026-11-04 08:25:55 +08:00
refactor(runtime): keep supervisor process access boundary driven
Signed-off-by: Drew Newberry <anewberry@nvidia.com>
This commit is contained in:
@@ -26,6 +26,10 @@ checks and tests. Their gateway session, boundary attachment, and TCP readiness
|
||||
are portable; only the optional Unix SSH/readiness socket adapters are gated.
|
||||
This is compile and control-plane coverage, not Windows isolation qualification.
|
||||
Preserve readiness gating on authenticated gateway acceptance and reconnection.
|
||||
The supervisor process-access multiplexer consumes boundary-provided streams;
|
||||
do not restore local PID, PTY, or process-group operations there. Native process
|
||||
mechanics belong to the sandbox or isolation backend. Keep the optional Unix
|
||||
SSH access adapter separate from portable session orchestration.
|
||||
Shared Sandbox Protocol audit validation defaults to strict Linux evidence;
|
||||
concrete platform validators must be selected by the implementing backend.
|
||||
|
||||
|
||||
Generated
-1
@@ -5132,7 +5132,6 @@ dependencies = [
|
||||
"hex",
|
||||
"libc",
|
||||
"miette",
|
||||
"nix 0.29.0",
|
||||
"openshell-core",
|
||||
"openshell-isolation-interface",
|
||||
"openshell-ocsf",
|
||||
|
||||
@@ -24,6 +24,7 @@ the execution outcome unknown.
|
||||
|
||||
Exec envelopes without an expiration time are rejected. Update the supervisor
|
||||
and sandbox runtime together when deploying this protocol change.
|
||||
|
||||
## Boundary audit validation
|
||||
|
||||
`OpenShellRuntimeBackend` implements the authenticated host side of the shared
|
||||
@@ -37,3 +38,11 @@ not bypass generation, session, resource, identity, or outer-fence checks.
|
||||
|
||||
Concrete platform validators belong to the implementing backend, not this
|
||||
shared transport library.
|
||||
|
||||
## Boundary transport establishment
|
||||
|
||||
Backends may supply a `BoundaryTransportConnector` for raw stream establishment.
|
||||
The shared client retains TLS peer verification, session bearer authentication,
|
||||
generation checks, and reconnection handling. The connector is reused for policy
|
||||
discovery and attachment; it does not define platform launch configuration or
|
||||
replace the isolation interface.
|
||||
|
||||
@@ -14,7 +14,7 @@ pub mod mediation;
|
||||
mod runtime;
|
||||
pub mod sandbox_auth;
|
||||
|
||||
pub use runtime::OpenShellRuntimeBackend;
|
||||
pub use runtime::{BoundaryTransportConnector, OpenShellRuntimeBackend};
|
||||
|
||||
/// Stable isolation backend name implemented by `openshell-sandbox`.
|
||||
pub const BACKEND_NAME: &str = "openshell-sandbox";
|
||||
|
||||
@@ -70,9 +70,20 @@ fn begin_recovery_window(
|
||||
*deadline.get_or_insert(failure_time + CONNECT_RETRY_TIMEOUT)
|
||||
}
|
||||
|
||||
/// Backend-owned raw transport establishment; shared code still authenticates
|
||||
/// the resulting stream with generation-pinned TLS and Sandbox Protocol JWTs.
|
||||
#[async_trait]
|
||||
pub trait BoundaryTransportConnector: std::fmt::Debug + Send + Sync {
|
||||
async fn connect(
|
||||
&self,
|
||||
descriptor: &SandboxRuntimeDescriptor,
|
||||
) -> Result<BoundaryDuplexStream, BackendError>;
|
||||
}
|
||||
|
||||
/// Host-side `OpenShell` Sandbox Protocol implementation registered with the supervisor.
|
||||
#[derive(Debug)]
|
||||
pub struct OpenShellRuntimeBackend {
|
||||
connector: Option<Arc<dyn BoundaryTransportConnector>>,
|
||||
audit_validator: Arc<dyn crate::audit::BoundaryAuditValidator>,
|
||||
ca_file_paths: Arc<std::sync::Mutex<Option<(PathBuf, PathBuf)>>>,
|
||||
provider_credentials: openshell_core::provider_credentials::ProviderCredentialState,
|
||||
@@ -85,7 +96,16 @@ impl OpenShellRuntimeBackend {
|
||||
descriptor: SandboxRuntimeDescriptor,
|
||||
bearer: openshell_core::jwt::SessionBearerTokenSlot,
|
||||
) -> Result<(Option<String>, bool), BackendError> {
|
||||
let client = BoundaryClient::new(descriptor, bearer);
|
||||
Self::discover_policy_with_connector(descriptor, bearer, None).await
|
||||
}
|
||||
|
||||
pub async fn discover_policy_with_connector(
|
||||
descriptor: SandboxRuntimeDescriptor,
|
||||
bearer: openshell_core::jwt::SessionBearerTokenSlot,
|
||||
connector: Option<Arc<dyn BoundaryTransportConnector>>,
|
||||
) -> Result<(Option<String>, bool), BackendError> {
|
||||
let mut client = BoundaryClient::new(descriptor, bearer);
|
||||
client.connector = connector;
|
||||
match client.call_idempotent(Request::DiscoverPolicy).await? {
|
||||
Response::ImagePolicy { yaml, invalid } => Ok((yaml, invalid)),
|
||||
_ => Err(BackendError::Descriptor(
|
||||
@@ -100,6 +120,7 @@ impl OpenShellRuntimeBackend {
|
||||
sandbox_bearer: openshell_core::jwt::SessionBearerTokenSlot,
|
||||
) -> Self {
|
||||
Self {
|
||||
connector: None,
|
||||
audit_validator: Arc::new(crate::audit::LinuxBoundaryAuditValidator),
|
||||
ca_file_paths,
|
||||
provider_credentials,
|
||||
@@ -107,6 +128,15 @@ impl OpenShellRuntimeBackend {
|
||||
}
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn with_transport_connector(
|
||||
mut self,
|
||||
connector: Arc<dyn BoundaryTransportConnector>,
|
||||
) -> Self {
|
||||
self.connector = Some(connector);
|
||||
self
|
||||
}
|
||||
|
||||
/// Select the backend implementation that validates opaque audit evidence.
|
||||
#[must_use]
|
||||
pub fn with_audit_validator(
|
||||
@@ -139,10 +169,9 @@ impl IsolationBackend for OpenShellRuntimeBackend {
|
||||
let generation = runtime_descriptor.generation.clone();
|
||||
let session_id = runtime_descriptor.session_id;
|
||||
let outer_fence = runtime_descriptor.outer_fence.clone();
|
||||
let client = Arc::new(BoundaryClient::new(
|
||||
runtime_descriptor,
|
||||
self.sandbox_bearer.clone(),
|
||||
));
|
||||
let mut client = BoundaryClient::new(runtime_descriptor, self.sandbox_bearer.clone());
|
||||
client.connector = self.connector.clone();
|
||||
let client = Arc::new(client);
|
||||
let response = client
|
||||
.call_idempotent(Request::Attach {
|
||||
supervisor_instance_id: client.supervisor_instance_id,
|
||||
@@ -1096,6 +1125,7 @@ async fn dispatch_client_mediation_frame(
|
||||
}
|
||||
|
||||
struct BoundaryClient {
|
||||
connector: Option<Arc<dyn BoundaryTransportConnector>>,
|
||||
runtime_descriptor: SandboxRuntimeDescriptor,
|
||||
supervisor_instance_id: crate::boundary_protocol::SupervisorInstanceId,
|
||||
sandbox_bearer: openshell_core::jwt::SessionBearerTokenSlot,
|
||||
@@ -1259,6 +1289,7 @@ impl BoundaryClient {
|
||||
sandbox_bearer: openshell_core::jwt::SessionBearerTokenSlot,
|
||||
) -> Self {
|
||||
Self {
|
||||
connector: None,
|
||||
runtime_descriptor,
|
||||
supervisor_instance_id: crate::boundary_protocol::SupervisorInstanceId::new(),
|
||||
sandbox_bearer,
|
||||
@@ -1880,6 +1911,7 @@ impl BoundaryClient {
|
||||
&self,
|
||||
) -> Result<(tonic::transport::Channel, Arc<TransportAbort>), BackendError> {
|
||||
let runtime_descriptor = self.runtime_descriptor.clone();
|
||||
let connector = self.connector.clone();
|
||||
let transport = Arc::new(TransportAbort::default());
|
||||
let connector_transport = transport.clone();
|
||||
let endpoint =
|
||||
@@ -1893,6 +1925,7 @@ impl BoundaryClient {
|
||||
let channel = endpoint
|
||||
.connect_with_connector(tower::service_fn(move |_: tonic::transport::Uri| {
|
||||
let runtime_descriptor = runtime_descriptor.clone();
|
||||
let connector = connector.clone();
|
||||
let abort = connector_transport.clone();
|
||||
async move {
|
||||
// A channel is one authenticated connection. Never let it
|
||||
@@ -1900,7 +1933,7 @@ impl BoundaryClient {
|
||||
if abort.is_aborted() || abort.connected.swap(true, Ordering::AcqRel) {
|
||||
return Err(AbortableTransport::aborted_error());
|
||||
}
|
||||
connect_boundary_with_retry(&runtime_descriptor)
|
||||
connect_boundary_with_retry(&runtime_descriptor, connector.clone())
|
||||
.await
|
||||
.map(|inner| TokioIo::new(AbortableTransport { inner, abort }))
|
||||
.map_err(|error| std::io::Error::other(error.to_string()))
|
||||
@@ -1921,10 +1954,11 @@ impl BoundaryClient {
|
||||
|
||||
async fn connect_boundary_with_retry(
|
||||
runtime_descriptor: &SandboxRuntimeDescriptor,
|
||||
connector: Option<Arc<dyn BoundaryTransportConnector>>,
|
||||
) -> Result<BoundaryDuplexStream, BackendError> {
|
||||
let deadline = tokio::time::Instant::now() + CONNECT_RETRY_TIMEOUT;
|
||||
loop {
|
||||
match connect_boundary_once(runtime_descriptor).await {
|
||||
match connect_boundary_using_connector(runtime_descriptor, connector.as_deref()).await {
|
||||
Ok(stream) => return Ok(stream),
|
||||
Err(error) if tokio::time::Instant::now() >= deadline => return Err(error),
|
||||
Err(_) => tokio::time::sleep(Duration::from_millis(25)).await,
|
||||
@@ -1932,41 +1966,53 @@ async fn connect_boundary_with_retry(
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
async fn connect_boundary_once(
|
||||
runtime_descriptor: &SandboxRuntimeDescriptor,
|
||||
) -> Result<BoundaryDuplexStream, BackendError> {
|
||||
let stream: BoundaryDuplexStream = match &runtime_descriptor.transport {
|
||||
#[cfg(unix)]
|
||||
SandboxTransport::Unix { socket_path } => {
|
||||
let stream = UnixStream::connect(socket_path).await.map_err(|error| {
|
||||
BackendError::Unavailable(format!(
|
||||
"connect to mapped boundary control socket {}: {error}",
|
||||
socket_path.display()
|
||||
))
|
||||
})?;
|
||||
Box::new(stream)
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
SandboxTransport::Unix { .. } => {
|
||||
return Err(BackendError::Unavailable(
|
||||
"Unix boundary transport requires a Unix host".to_string(),
|
||||
));
|
||||
}
|
||||
SandboxTransport::Tcp {
|
||||
authority,
|
||||
addresses,
|
||||
} => {
|
||||
let stream = openshell_core::net::connect_tcp_nodelay_best_effort(addresses)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
connect_boundary_using_connector(runtime_descriptor, None).await
|
||||
}
|
||||
|
||||
async fn connect_boundary_using_connector(
|
||||
runtime_descriptor: &SandboxRuntimeDescriptor,
|
||||
connector: Option<&dyn BoundaryTransportConnector>,
|
||||
) -> Result<BoundaryDuplexStream, BackendError> {
|
||||
let stream: BoundaryDuplexStream = if let Some(connector) = connector {
|
||||
connector.connect(runtime_descriptor).await?
|
||||
} else {
|
||||
match &runtime_descriptor.transport {
|
||||
#[cfg(unix)]
|
||||
SandboxTransport::Unix { socket_path } => {
|
||||
let stream = UnixStream::connect(socket_path).await.map_err(|error| {
|
||||
BackendError::Unavailable(format!(
|
||||
"connect to boundary TLS endpoint {authority}: {error}"
|
||||
"connect to mapped boundary control socket {}: {error}",
|
||||
socket_path.display()
|
||||
))
|
||||
})?;
|
||||
enable_boundary_tcp_keepalive(&stream);
|
||||
Box::new(stream)
|
||||
Box::new(stream)
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
SandboxTransport::Unix { .. } => {
|
||||
return Err(BackendError::Unavailable(
|
||||
"Unix boundary transport requires a Unix host".to_string(),
|
||||
));
|
||||
}
|
||||
SandboxTransport::Tcp {
|
||||
authority,
|
||||
addresses,
|
||||
} => {
|
||||
let stream = openshell_core::net::connect_tcp_nodelay_best_effort(addresses)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
BackendError::Unavailable(format!(
|
||||
"connect to boundary TLS endpoint {authority}: {error}"
|
||||
))
|
||||
})?;
|
||||
enable_boundary_tcp_keepalive(&stream);
|
||||
Box::new(stream)
|
||||
}
|
||||
SandboxTransport::Vsock { guest_cid, port } => connect_host_vsock(*guest_cid, *port)?,
|
||||
}
|
||||
SandboxTransport::Vsock { guest_cid, port } => connect_host_vsock(*guest_cid, *port)?,
|
||||
};
|
||||
let tls = &runtime_descriptor.tls;
|
||||
let server_name =
|
||||
@@ -3165,6 +3211,83 @@ mod tests {
|
||||
server.abort();
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct ReverseTestConnector(tokio::net::TcpListener);
|
||||
|
||||
#[async_trait]
|
||||
impl BoundaryTransportConnector for ReverseTestConnector {
|
||||
async fn connect(
|
||||
&self,
|
||||
_: &SandboxRuntimeDescriptor,
|
||||
) -> Result<BoundaryDuplexStream, BackendError> {
|
||||
let (stream, _) = self
|
||||
.0
|
||||
.accept()
|
||||
.await
|
||||
.map_err(|error| BackendError::Unavailable(error.to_string()))?;
|
||||
Ok(Box::new(stream))
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn custom_reverse_transport_preserves_tls_and_bearer_authentication() {
|
||||
let certificate = test_certificate();
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let address = listener.local_addr().unwrap();
|
||||
let connector = Arc::new(ReverseTestConnector(listener));
|
||||
let server = tokio::spawn(async move {
|
||||
let stream = tokio::net::TcpStream::connect(address).await.unwrap();
|
||||
let stream = tokio_rustls::TlsAcceptor::from(certificate.server_config)
|
||||
.accept(stream)
|
||||
.await
|
||||
.unwrap();
|
||||
serve_test_grpc(Box::new(stream), "a".repeat(32)).await;
|
||||
});
|
||||
let mut client = BoundaryClient::new(
|
||||
tls_runtime_descriptor(address, certificate.client_tls),
|
||||
test_bearer(&"a".repeat(32)),
|
||||
);
|
||||
client.connector = Some(connector);
|
||||
let response =
|
||||
tokio::time::timeout(Duration::from_secs(5), client.exchange(Request::Confirm))
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(
|
||||
response,
|
||||
Response::Confirmed {
|
||||
confirmation: Box::new(test_confirmation())
|
||||
}
|
||||
);
|
||||
server.abort();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn custom_transport_cannot_bypass_tls_peer_verification() {
|
||||
let server_certificate = test_certificate();
|
||||
let different_trust = test_certificate();
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let address = listener.local_addr().unwrap();
|
||||
let connector = ReverseTestConnector(listener);
|
||||
let server = tokio::spawn(async move {
|
||||
let stream = tokio::net::TcpStream::connect(address).await.unwrap();
|
||||
let _ = tokio_rustls::TlsAcceptor::from(server_certificate.server_config)
|
||||
.accept(stream)
|
||||
.await;
|
||||
});
|
||||
let descriptor = tls_runtime_descriptor(address, different_trust.client_tls);
|
||||
assert!(
|
||||
tokio::time::timeout(
|
||||
Duration::from_secs(5),
|
||||
connect_boundary_using_connector(&descriptor, Some(&connector))
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.is_err()
|
||||
);
|
||||
server.await.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tls_tcp_rejects_tls12_only_server() {
|
||||
let certificate = test_certificate_with_protocol_versions(&[&rustls::version::TLS12]);
|
||||
|
||||
@@ -21,7 +21,6 @@ base64 = { workspace = true }
|
||||
bytes = { workspace = true }
|
||||
hex = "0.4"
|
||||
miette = { workspace = true }
|
||||
nix = { workspace = true }
|
||||
rand = "0.10"
|
||||
russh = "0.63.1"
|
||||
serde_json = { workspace = true }
|
||||
|
||||
@@ -4,20 +4,10 @@
|
||||
//! Retained I/O multiplexer for the canonical sandbox process.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
#[cfg(unix)]
|
||||
use std::io::{Read, Write};
|
||||
#[cfg(unix)]
|
||||
use std::os::fd::AsRawFd;
|
||||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use bytes::Bytes;
|
||||
#[cfg(unix)]
|
||||
use nix::fcntl::{FcntlArg, OFlag, fcntl};
|
||||
#[cfg(unix)]
|
||||
use nix::pty::Winsize;
|
||||
#[cfg(unix)]
|
||||
use tokio::io::unix::AsyncFd;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::sync::Notify;
|
||||
use tokio::sync::watch;
|
||||
@@ -28,17 +18,6 @@ use openshell_isolation_interface::contract::{
|
||||
|
||||
const OUTPUT_BUFFER_BYTES: usize = 1024 * 1024;
|
||||
|
||||
/// Canonical-process I/O retained by the supervisor session multiplexer.
|
||||
#[cfg(unix)]
|
||||
pub enum ProcessIo {
|
||||
Pty(std::fs::File),
|
||||
Pipes {
|
||||
stdin: tokio::process::ChildStdin,
|
||||
stdout: tokio::process::ChildStdout,
|
||||
stderr: tokio::process::ChildStderr,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub enum MainOutput {
|
||||
Stdout(Bytes),
|
||||
@@ -197,15 +176,11 @@ impl MainOutputCursor {
|
||||
}
|
||||
|
||||
pub struct MainSession {
|
||||
#[cfg(unix)]
|
||||
pid: u32,
|
||||
terminal: bool,
|
||||
input: tokio::sync::mpsc::Sender<Vec<u8>>,
|
||||
output: Arc<OutputLog>,
|
||||
input_owner: Mutex<Option<u64>>,
|
||||
next_owner: AtomicU64,
|
||||
#[cfg(unix)]
|
||||
pty_master: Option<Arc<std::fs::File>>,
|
||||
boundary_process: Option<Arc<dyn BoundaryProcess>>,
|
||||
boundary_terminal: Option<Arc<dyn BoundaryTerminal>>,
|
||||
readers_remaining: AtomicUsize,
|
||||
@@ -226,15 +201,11 @@ impl MainSession {
|
||||
pub fn inert_with_input() -> (Arc<Self>, tokio::sync::mpsc::Receiver<Vec<u8>>) {
|
||||
let (input, input_rx) = tokio::sync::mpsc::channel(64);
|
||||
let session = Arc::new(Self {
|
||||
#[cfg(unix)]
|
||||
pid: 1,
|
||||
terminal: false,
|
||||
input,
|
||||
output: OutputLog::new(),
|
||||
input_owner: Mutex::new(None),
|
||||
next_owner: AtomicU64::new(1),
|
||||
#[cfg(unix)]
|
||||
pty_master: None,
|
||||
boundary_process: None,
|
||||
boundary_terminal: None,
|
||||
readers_remaining: AtomicUsize::new(0),
|
||||
@@ -250,63 +221,6 @@ impl MainSession {
|
||||
(session, input_rx)
|
||||
}
|
||||
|
||||
#[cfg(all(test, unix))]
|
||||
pub fn terminal_for_test() -> (Arc<Self>, std::fs::File) {
|
||||
let pty = nix::pty::openpty(None, None).expect("open test PTY");
|
||||
let slave = std::fs::File::from(pty.slave);
|
||||
(
|
||||
Self::new(ProcessIo::Pty(std::fs::File::from(pty.master)), 1),
|
||||
slave,
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(all(test, unix))]
|
||||
#[allow(unsafe_code)]
|
||||
pub fn terminal_size_for_test(&self) -> (u16, u16) {
|
||||
let master = self.pty_master.as_ref().expect("terminal PTY master");
|
||||
let mut winsize: libc::winsize = unsafe { std::mem::zeroed() };
|
||||
let result = unsafe { libc::ioctl(master.as_raw_fd(), libc::TIOCGWINSZ, &mut winsize) };
|
||||
assert_eq!(result, 0, "read terminal dimensions");
|
||||
(winsize.ws_col, winsize.ws_row)
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
#[cfg(unix)]
|
||||
pub fn new(io: ProcessIo, pid: u32) -> Arc<Self> {
|
||||
let terminal = matches!(io, ProcessIo::Pty(_));
|
||||
let (input, input_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(64);
|
||||
let pty_master = match &io {
|
||||
ProcessIo::Pty(master) => {
|
||||
set_nonblocking(master).expect("set canonical PTY master nonblocking");
|
||||
master.try_clone().ok().map(Arc::new)
|
||||
}
|
||||
ProcessIo::Pipes { .. } => None,
|
||||
};
|
||||
let session = Arc::new(Self {
|
||||
pid,
|
||||
terminal,
|
||||
input,
|
||||
output: OutputLog::new(),
|
||||
input_owner: Mutex::new(None),
|
||||
next_owner: AtomicU64::new(1),
|
||||
#[cfg(unix)]
|
||||
pty_master,
|
||||
boundary_process: None,
|
||||
boundary_terminal: None,
|
||||
readers_remaining: AtomicUsize::new(if terminal { 1 } else { 2 }),
|
||||
readers_done: Notify::new(),
|
||||
finished: std::sync::atomic::AtomicBool::new(false),
|
||||
terminal_attachments: Mutex::new(TerminalAttachmentState {
|
||||
active: 0,
|
||||
process_finished: false,
|
||||
expectation: AttachmentExpectation::None,
|
||||
}),
|
||||
terminal_attachments_done: Notify::new(),
|
||||
});
|
||||
Self::start_io(&session, io, input_rx);
|
||||
session
|
||||
}
|
||||
|
||||
/// Build the control-side multiplexer around a boundary-owned admitted
|
||||
/// process. Process lifecycle and PTY operations remain delegated to the
|
||||
/// boundary process handle.
|
||||
@@ -324,15 +238,11 @@ impl MainSession {
|
||||
let terminal_mode = terminal.is_some();
|
||||
let (input, mut input_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(64);
|
||||
let session = Arc::new(Self {
|
||||
#[cfg(unix)]
|
||||
pid: 0,
|
||||
terminal: terminal_mode,
|
||||
input,
|
||||
output: OutputLog::new(),
|
||||
input_owner: Mutex::new(None),
|
||||
next_owner: AtomicU64::new(1),
|
||||
#[cfg(unix)]
|
||||
pty_master: None,
|
||||
boundary_process: Some(process),
|
||||
boundary_terminal: terminal,
|
||||
readers_remaining: AtomicUsize::new(if terminal_mode { 1 } else { 2 }),
|
||||
@@ -384,100 +294,9 @@ impl MainSession {
|
||||
session
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn start_io(
|
||||
this: &Arc<Self>,
|
||||
io: ProcessIo,
|
||||
mut input_rx: tokio::sync::mpsc::Receiver<Vec<u8>>,
|
||||
) {
|
||||
match io {
|
||||
ProcessIo::Pty(master) => {
|
||||
let master = Arc::new(AsyncFd::new(master).expect("register canonical PTY master"));
|
||||
let reader = Arc::clone(&master);
|
||||
let output = Arc::clone(this);
|
||||
tokio::spawn(async move {
|
||||
let mut buffer = [0u8; 4096];
|
||||
loop {
|
||||
let Ok(mut ready) = reader.readable().await else {
|
||||
break;
|
||||
};
|
||||
match ready.try_io(|inner| {
|
||||
let mut file = inner.get_ref();
|
||||
file.read(&mut buffer)
|
||||
}) {
|
||||
Ok(Ok(0) | Err(_)) => break,
|
||||
Ok(Ok(read)) => output.publish(MainOutput::Stdout(
|
||||
Bytes::copy_from_slice(&buffer[..read]),
|
||||
)),
|
||||
Err(_would_block) => {}
|
||||
}
|
||||
}
|
||||
output.reader_finished();
|
||||
});
|
||||
tokio::spawn(async move {
|
||||
while let Some(data) = input_rx.recv().await {
|
||||
let mut remaining = data.as_slice();
|
||||
while !remaining.is_empty() {
|
||||
let Ok(mut ready) = master.writable().await else {
|
||||
return;
|
||||
};
|
||||
match ready.try_io(|inner| {
|
||||
let mut file = inner.get_ref();
|
||||
file.write(remaining)
|
||||
}) {
|
||||
Ok(Ok(0) | Err(_)) => return,
|
||||
Ok(Ok(written)) => remaining = &remaining[written..],
|
||||
Err(_would_block) => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
ProcessIo::Pipes {
|
||||
mut stdin,
|
||||
mut stdout,
|
||||
mut stderr,
|
||||
} => {
|
||||
let stdout_session = Arc::clone(this);
|
||||
tokio::spawn(async move {
|
||||
let mut buffer = [0u8; 4096];
|
||||
loop {
|
||||
match stdout.read(&mut buffer).await {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(read) => {
|
||||
stdout_session.publish(MainOutput::Stdout(Bytes::copy_from_slice(
|
||||
&buffer[..read],
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
stdout_session.reader_finished();
|
||||
});
|
||||
let stderr_session = Arc::clone(this);
|
||||
tokio::spawn(async move {
|
||||
let mut buffer = [0u8; 4096];
|
||||
loop {
|
||||
match stderr.read(&mut buffer).await {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(read) => {
|
||||
stderr_session.publish(MainOutput::Stderr(Bytes::copy_from_slice(
|
||||
&buffer[..read],
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
stderr_session.reader_finished();
|
||||
});
|
||||
tokio::spawn(async move {
|
||||
while let Some(data) = input_rx.recv().await {
|
||||
if stdin.write_all(&data).await.is_err() {
|
||||
break;
|
||||
}
|
||||
let _ = stdin.flush().await;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
#[cfg(all(test, unix))]
|
||||
pub(crate) fn publish_test_output(&self, data: &'static [u8]) {
|
||||
self.publish(MainOutput::Stdout(Bytes::from_static(data)));
|
||||
}
|
||||
|
||||
fn publish(&self, event: MainOutput) {
|
||||
@@ -660,7 +479,7 @@ impl MainSession {
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn resize(&self, columns: u32, rows: u32, pixel_width: u32, pixel_height: u32) {
|
||||
pub async fn resize(&self, columns: u32, rows: u32, _pixel_width: u32, _pixel_height: u32) {
|
||||
if let Some(terminal) = self.boundary_terminal.as_ref() {
|
||||
let _ = terminal
|
||||
.resize(
|
||||
@@ -668,67 +487,19 @@ impl MainSession {
|
||||
u16::try_from(rows.max(1)).unwrap_or(u16::MAX),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
let _ = (columns, rows, pixel_width, pixel_height);
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let Some(master) = self.pty_master.as_ref() else {
|
||||
return;
|
||||
};
|
||||
let winsize = Winsize {
|
||||
ws_row: u16::try_from(rows.max(1)).unwrap_or(u16::MAX),
|
||||
ws_col: u16::try_from(columns.max(1)).unwrap_or(u16::MAX),
|
||||
ws_xpixel: u16::try_from(pixel_width).unwrap_or(u16::MAX),
|
||||
ws_ypixel: u16::try_from(pixel_height).unwrap_or(u16::MAX),
|
||||
};
|
||||
#[allow(unsafe_code)]
|
||||
unsafe {
|
||||
libc::ioctl(master.as_raw_fd(), libc::TIOCSWINSZ, &winsize);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Preserve the local Unix signal surface, including SIGQUIT.
|
||||
#[cfg(unix)]
|
||||
pub async fn signal_group(&self, signal: nix::sys::signal::Signal) -> Result<(), String> {
|
||||
if self.boundary_process.is_some() {
|
||||
let boundary_signal = match signal {
|
||||
nix::sys::signal::Signal::SIGHUP => BoundarySignal::Hup,
|
||||
nix::sys::signal::Signal::SIGINT => BoundarySignal::Int,
|
||||
nix::sys::signal::Signal::SIGKILL => BoundarySignal::Kill,
|
||||
nix::sys::signal::Signal::SIGTERM => BoundarySignal::Term,
|
||||
other => return Err(format!("boundary signal {other:?} is unsupported")),
|
||||
};
|
||||
return self.signal_boundary_group(boundary_signal).await;
|
||||
}
|
||||
let pid = i32::try_from(self.pid).unwrap_or(i32::MAX);
|
||||
nix::sys::signal::kill(nix::unistd::Pid::from_raw(-pid), signal)
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
|
||||
/// Delegate process-group signaling to the isolation backend.
|
||||
pub async fn signal_boundary_group(&self, signal: BoundarySignal) -> Result<(), String> {
|
||||
if let Some(process) = self.boundary_process.as_ref() {
|
||||
return process
|
||||
.signal(signal)
|
||||
.await
|
||||
.map_err(|error| error.to_string());
|
||||
}
|
||||
#[cfg(unix)]
|
||||
{
|
||||
let pid = i32::try_from(self.pid).unwrap_or(i32::MAX);
|
||||
let signal = match signal {
|
||||
BoundarySignal::Hup => nix::sys::signal::Signal::SIGHUP,
|
||||
BoundarySignal::Int => nix::sys::signal::Signal::SIGINT,
|
||||
BoundarySignal::Kill => nix::sys::signal::Signal::SIGKILL,
|
||||
BoundarySignal::Term => nix::sys::signal::Signal::SIGTERM,
|
||||
};
|
||||
nix::sys::signal::kill(nix::unistd::Pid::from_raw(-pid), signal)
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
Err("local process-group signaling is unsupported on Windows".to_string())
|
||||
let process = self
|
||||
.boundary_process
|
||||
.as_ref()
|
||||
.ok_or_else(|| "no boundary process is attached".to_string())?;
|
||||
process
|
||||
.signal(signal)
|
||||
.await
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
@@ -742,17 +513,6 @@ impl MainSession {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn set_nonblocking(file: &std::fs::File) -> Result<(), nix::errno::Errno> {
|
||||
let flags = fcntl(file.as_raw_fd(), FcntlArg::F_GETFL)?;
|
||||
let flags = OFlag::from_bits_truncate(flags);
|
||||
fcntl(
|
||||
file.as_raw_fd(),
|
||||
FcntlArg::F_SETFL(flags | OFlag::O_NONBLOCK),
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -998,45 +758,15 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn terminal_pump_reads_output_and_writes_input() {
|
||||
let (session, mut slave) = MainSession::terminal_for_test();
|
||||
set_nonblocking(&slave).expect("set test PTY slave nonblocking");
|
||||
let mut output = session.subscribe();
|
||||
|
||||
slave
|
||||
.write_all(b"process output")
|
||||
.expect("write PTY output");
|
||||
let event = tokio::time::timeout(std::time::Duration::from_secs(1), output.recv())
|
||||
.await
|
||||
.expect("PTY output timed out")
|
||||
.expect("PTY output was retained");
|
||||
assert!(matches!(
|
||||
event,
|
||||
MainOutput::Stdout(data) if data == b"process output"[..]
|
||||
));
|
||||
|
||||
let (owner, input) = session.acquire_input().expect("acquire PTY input");
|
||||
input
|
||||
.send(b"client input\n".to_vec())
|
||||
.await
|
||||
.expect("queue PTY input");
|
||||
let mut received = [0; 64];
|
||||
let read = tokio::time::timeout(std::time::Duration::from_secs(1), async {
|
||||
loop {
|
||||
match slave.read(&mut received) {
|
||||
Ok(read) => break read,
|
||||
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
Err(error) => panic!("read PTY input: {error}"),
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("PTY input timed out");
|
||||
assert_eq!(&received[..read], b"client input\n");
|
||||
session.release_input(owner);
|
||||
async fn signaling_without_boundary_is_rejected() {
|
||||
let session = MainSession::inert();
|
||||
assert_eq!(
|
||||
session
|
||||
.signal_boundary_group(BoundarySignal::Term)
|
||||
.await
|
||||
.unwrap_err(),
|
||||
"no boundary process is attached"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -954,15 +954,14 @@ impl russh::server::Handler for SshHandler {
|
||||
.is_some_and(|state| state.main_attached)
|
||||
{
|
||||
let signal = match signal {
|
||||
Sig::HUP => Some(nix::sys::signal::Signal::SIGHUP),
|
||||
Sig::INT => Some(nix::sys::signal::Signal::SIGINT),
|
||||
Sig::KILL => Some(nix::sys::signal::Signal::SIGKILL),
|
||||
Sig::QUIT => Some(nix::sys::signal::Signal::SIGQUIT),
|
||||
Sig::TERM => Some(nix::sys::signal::Signal::SIGTERM),
|
||||
Sig::HUP => Some(openshell_isolation_interface::contract::BoundarySignal::Hup),
|
||||
Sig::INT => Some(openshell_isolation_interface::contract::BoundarySignal::Int),
|
||||
Sig::KILL => Some(openshell_isolation_interface::contract::BoundarySignal::Kill),
|
||||
Sig::TERM => Some(openshell_isolation_interface::contract::BoundarySignal::Term),
|
||||
_ => None,
|
||||
};
|
||||
if let (Some(signal), Some(main_session)) = (signal, self.main_session.as_ref())
|
||||
&& let Err(error) = main_session.signal_group(signal).await
|
||||
&& let Err(error) = main_session.signal_boundary_group(signal).await
|
||||
{
|
||||
warn!(%error, ?signal, "failed to signal canonical main process group");
|
||||
}
|
||||
@@ -1315,7 +1314,6 @@ fn direct_tcpip_target(
|
||||
)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::io::Write as _;
|
||||
|
||||
pub(super) struct AcceptAnyServerKey;
|
||||
|
||||
@@ -1792,14 +1790,13 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn main_attachment_accepts_declared_session_after_process_exit() {
|
||||
let (main_session, mut slave) = MainSession::terminal_for_test();
|
||||
let main_session = MainSession::inert();
|
||||
let mut output = main_session.subscribe();
|
||||
slave.write_all(b"retained output").unwrap();
|
||||
main_session.publish_test_output(b"retained output");
|
||||
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
|
||||
|
||||
@@ -15,6 +15,7 @@ The supervisor admits policy and prepares credentials before constructing and at
|
||||
The client receives the supervisor's live provider state, bearer-token slot, and CA-path slot. Provider refresh, token rotation, and later CA publication must remain visible through those shared handles. Startup does not create independent copies of their current values.
|
||||
|
||||
The setup interface stays private to the supervisor. It adds no runtime backend registration, endpoint configuration, or public factory API. The public `run_sandbox` signature and standard backend selection remain unchanged.
|
||||
|
||||
## Portable supervisor access
|
||||
|
||||
The supervisor consumes the shared isolation backend contract. Its gateway
|
||||
@@ -27,8 +28,10 @@ readiness guard closes the listener. A requested Unix readiness endpoint fails
|
||||
explicitly on unsupported hosts, even before session acceptance.
|
||||
|
||||
The optional Unix SSH adapter remains separate from boundary-based process I/O.
|
||||
Local Unix process signaling retains its existing signal surface, including
|
||||
`SIGQUIT`. Remote signals use the backend-neutral `BoundarySignal` contract.
|
||||
The process-access multiplexer consumes boundary-provided streams and delegates
|
||||
terminal resizing and signals through the existing isolation contract. It does
|
||||
not own local PIDs, PTYs, or process-group signaling; native process operations
|
||||
belong to the sandbox or isolation backend. Signals use `BoundarySignal`.
|
||||
|
||||
These portable control-plane foundations do not qualify a platform isolation
|
||||
backend or enable a Windows workload runtime.
|
||||
|
||||
@@ -176,14 +176,16 @@ fn arm_parent_liveness(raw_fd: Option<i32>) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn backend_descriptor(args: &Args) -> Result<BackendDescriptor> {
|
||||
fn backend_descriptor(args: &Args, admitted_backend: Option<&str>) -> Result<BackendDescriptor> {
|
||||
let path = args.backend_descriptor_file.as_deref().ok_or_else(|| {
|
||||
miette::miette!("--backend-descriptor-file is required for --role=isolation-backend")
|
||||
})?;
|
||||
let payload = std::fs::read(path)
|
||||
.map_err(|error| miette::miette!("read backend descriptor {}: {error}", path.display()))?;
|
||||
Ok(BackendDescriptor {
|
||||
backend_name: openshell_sandbox_backend::BACKEND_NAME.to_string(),
|
||||
backend_name: admitted_backend
|
||||
.unwrap_or(openshell_sandbox_backend::BACKEND_NAME)
|
||||
.to_string(),
|
||||
payload,
|
||||
})
|
||||
}
|
||||
@@ -298,8 +300,10 @@ fn main() -> Result<()> {
|
||||
validate_role_arguments(&args)?;
|
||||
arm_parent_liveness(args.parent_liveness_fd)?;
|
||||
validate_main_exit_marker(args.main_exit_marker.as_deref())?;
|
||||
let admitted_isolation_backend =
|
||||
std::env::var(openshell_core::sandbox_env::ADMITTED_ISOLATION_BACKEND).ok();
|
||||
let isolation_inputs = if args.role == SupervisorRole::IsolationBackend {
|
||||
let descriptor = backend_descriptor(&args)?;
|
||||
let descriptor = backend_descriptor(&args, admitted_isolation_backend.as_deref())?;
|
||||
let auth = auth_bundle(&args)?;
|
||||
// Install the driver-provisioned session before starting log push or
|
||||
// any other gateway client. `run_sandbox` obtains the same Sandbox
|
||||
@@ -429,8 +433,6 @@ fn main() -> Result<()> {
|
||||
"isolation-backend role started without validated runtime inputs"
|
||||
));
|
||||
};
|
||||
let admitted_isolation_backend =
|
||||
std::env::var(openshell_core::sandbox_env::ADMITTED_ISOLATION_BACKEND).ok();
|
||||
Box::pin(openshell_supervisor::run_sandbox(
|
||||
command,
|
||||
workdir,
|
||||
@@ -502,7 +504,14 @@ mod tests {
|
||||
assert_eq!(args.role, SupervisorRole::IsolationBackend);
|
||||
assert!(validate_role_arguments(&args).is_ok());
|
||||
assert_eq!(
|
||||
backend_descriptor(&args)
|
||||
backend_descriptor(&args, None).unwrap().backend_name,
|
||||
openshell_sandbox_backend::BACKEND_NAME
|
||||
);
|
||||
let admitted = backend_descriptor(&args, Some("openshell-test-backend")).unwrap();
|
||||
assert_eq!(admitted.backend_name, "openshell-test-backend");
|
||||
assert_eq!(admitted.payload, vec![0]);
|
||||
assert_eq!(
|
||||
backend_descriptor(&args, None)
|
||||
.expect("runtime descriptor")
|
||||
.payload,
|
||||
vec![0]
|
||||
|
||||
Reference in New Issue
Block a user