From 5b589abd5afadd6cd56b983c0af34f980882cf83 Mon Sep 17 00:00:00 2001 From: Drew Newberry <385+drew@users.noreply.github.com> Date: Wed, 26 Aug 2026 04:53:20 +0000 Subject: [PATCH] refactor(supervisor): add backend-neutral boundary primitives Signed-off-by: Drew Newberry <385+drew@users.noreply.github.com> --- Cargo.lock | 4 + .../openshell-supervisor-network/Cargo.toml | 3 + .../src/identity_source.rs | 168 +++++ .../src/l7/tls.rs | 10 +- .../openshell-supervisor-network/src/lib.rs | 1 + .../openshell-supervisor-process/Cargo.toml | 2 + .../src/boundary_exec.rs | 695 ++++++++++++++++++ .../src/boundary_io.rs | 317 ++++++++ .../openshell-supervisor-process/src/lib.rs | 2 + .../src/managed_children.rs | 154 +++- .../src/netns/nft_ruleset.rs | 71 +- .../src/sandbox/linux/seccomp.rs | 39 + .../openshell-supervisor-process/src/ssh.rs | 4 +- 13 files changed, 1427 insertions(+), 43 deletions(-) create mode 100644 crates/openshell-supervisor-network/src/identity_source.rs create mode 100644 crates/openshell-supervisor-process/src/boundary_exec.rs create mode 100644 crates/openshell-supervisor-process/src/boundary_io.rs diff --git a/Cargo.lock b/Cargo.lock index 549b1fc834..b9d4b7a32f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4449,6 +4449,7 @@ name = "openshell-supervisor-network" version = "0.0.0" dependencies = [ "apollo-parser", + "async-trait", "aws-credential-types", "aws-sigv4", "aws-smithy-runtime-api", @@ -4464,6 +4465,7 @@ dependencies = [ "libc", "miette", "openshell-core", + "openshell-isolation", "openshell-ocsf", "openshell-policy", "openshell-router", @@ -4502,6 +4504,7 @@ name = "openshell-supervisor-process" version = "0.0.0" dependencies = [ "anyhow", + "async-trait", "base64 0.22.1", "bytes", "capctl", @@ -4512,6 +4515,7 @@ dependencies = [ "miette", "nix 0.29.0", "openshell-core", + "openshell-isolation", "openshell-ocsf", "openshell-policy", "rand 0.10.2", diff --git a/crates/openshell-supervisor-network/Cargo.toml b/crates/openshell-supervisor-network/Cargo.toml index 34d9c32a47..5ae4bf1f94 100644 --- a/crates/openshell-supervisor-network/Cargo.toml +++ b/crates/openshell-supervisor-network/Cargo.toml @@ -12,11 +12,14 @@ rust-version.workspace = true [dependencies] openshell-core = { path = "../openshell-core", features = ["oauth"] } +openshell-isolation = { path = "../openshell-isolation" } openshell-ocsf = { path = "../openshell-ocsf" } openshell-policy = { path = "../openshell-policy" } openshell-router = { path = "../openshell-router" } openshell-supervisor-middleware = { path = "../openshell-supervisor-middleware" } +async-trait = "0.1" + apollo-parser = { workspace = true } aws-sigv4 = { version = "1", features = ["sign-http", "http1"] } aws-credential-types = { version = "1", features = ["hardcoded-credentials"] } diff --git a/crates/openshell-supervisor-network/src/identity_source.rs b/crates/openshell-supervisor-network/src/identity_source.rs new file mode 100644 index 0000000000..0e5960aa8e --- /dev/null +++ b/crates/openshell-supervisor-network/src/identity_source.rs @@ -0,0 +1,168 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The in-pod binary-identity resolver (RFC 0012 runtime contract). +//! +//! RFC 0012 delivers executable identity on every +//! [`MediatedConnection`](openshell_isolation::contract::MediatedConnection): +//! the backend resolves identity for the accepted connection before mediation. +//! An unresolved identity denies that connection. This is the in-pod +//! resolution mechanism — procfs, keyed by the workload-side TCP peer port — +//! kept in this crate on purpose: the proxy that consumes identity is here, and +//! so are procfs and the binary identity cache. Stronger backends may use a +//! different resolution mechanism without changing the contract. The result +//! type lives in the lower `openshell-isolation` crate (network -> isolation -> +//! core, acyclic). +//! +//! The legacy listener still resolves identity in the proxy hot path. The RFC +//! 0012 co-located source invokes this resolver before returning each accepted +//! connection, so mediation consumes the bound identity result. + +use std::sync::Arc; +use std::sync::atomic::AtomicU32; + +use openshell_isolation::contract::{BinaryIdentity, ResolveError, Sha256Digest}; + +/// In-pod binary-identity resolver: reads and hashes the executable resolved +/// for an accepted connection from procfs. Resolution fails closed; it never +/// fabricates identity fields. +#[derive(Clone)] +pub struct ProcfsIdentityResolver { + /// The workload entrypoint PID, whose network namespace owns the peer + /// sockets the proxy resolves. Published once the agent starts. + pub entrypoint_pid: Arc, +} + +impl ProcfsIdentityResolver { + /// Resolve the executable identity behind an accepted workload connection. + pub fn resolve_connection( + &self, + workload_addr: std::net::SocketAddr, + proxy_addr: std::net::SocketAddr, + ) -> Result { + // procfs resolution is Linux-only; on other targets the supervisor has + // no procfs to read, so resolution fails closed. + #[cfg(target_os = "linux")] + { + self.resolve_via_procfs(workload_addr, proxy_addr) + } + #[cfg(not(target_os = "linux"))] + { + let _ = (workload_addr, proxy_addr); + Err(ResolveError::Failed( + "no procfs on this platform; identity resolution unavailable".to_string(), + )) + } + } +} + +#[cfg(target_os = "linux")] +impl ProcfsIdentityResolver { + fn resolve_via_procfs( + &self, + workload_addr: std::net::SocketAddr, + proxy_addr: std::net::SocketAddr, + ) -> Result { + use std::sync::atomic::Ordering; + + let entrypoint_pid = self.entrypoint_pid.load(Ordering::Acquire); + if entrypoint_pid == 0 { + // No workload yet: nothing to attribute the connection to. Fail + // closed so a binary-scoped rule cannot match an unattributed peer. + return Err(ResolveError::NotFound); + } + + let connection = crate::procfs::WorkloadProxyTcpConnection::new(workload_addr, proxy_addr); + let owners = crate::procfs::resolve_tcp_peer_socket_owners(entrypoint_pid, connection) + .map_err(|_| ResolveError::NotFound)?; + let mut identities = Vec::with_capacity(owners.owners.len()); + for owner in owners.owners { + identities.push(Self::resolve_owner(owner.pid, entrypoint_pid)?); + } + let Some(identity) = identities.first().cloned() else { + return Err(ResolveError::NotFound); + }; + if identities.iter().skip(1).any(|candidate| { + candidate.binary_path != identity.binary_path + || candidate.binary_digest != identity.binary_digest + || candidate.ancestors != identity.ancestors + || candidate.cmdline_paths != identity.cmdline_paths + }) { + return Err(ResolveError::Failed( + "shared socket owners have different policy identities".to_string(), + )); + } + Ok(identity) + } + + fn resolve_owner(owner_pid: u32, entrypoint_pid: u32) -> Result { + let binary_path = crate::procfs::binary_path(owner_pid.cast_signed()) + .map_err(|error| ResolveError::Failed(error.to_string()))?; + + // Hash the live `/proc//exe` object, not the reopened resolved + // path: opening the magic symlink pins the inode the process is actually + // executing, so a post-resolution swap of the path cannot launder the + // hash. A missing digest is `None`, never an empty string, and an + // unhashable binary fails closed rather than asserting an identity the + // resolver could not verify. + let exe = std::path::PathBuf::from(format!("/proc/{owner_pid}/exe")); + let binary_digest = match crate::procfs::file_sha256(&exe) { + Ok(digest) => Some(digest.parse::()?), + Err(_) => { + return Err(ResolveError::Failed( + "could not hash resolved executable; refusing to assert identity".to_string(), + )); + } + }; + + let ancestors = crate::procfs::collect_ancestor_binaries(owner_pid, entrypoint_pid); + let mut exclude = ancestors.clone(); + exclude.push(binary_path.clone()); + let cmdline_paths = + crate::procfs::collect_cmdline_paths(owner_pid, entrypoint_pid, &exclude); + + Ok(BinaryIdentity { + binary_path, + binary_digest, + ancestors, + cmdline_paths, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Stands in for the mediation service: a binary-scoped rule can only be + /// authorized by a resolved identity carrying the fields it requires. + fn admits_binary_rule(result: Result) -> bool { + matches!(result, Ok(identity) if identity.binary_digest.is_some()) + } + + #[test] + fn fails_closed_before_the_workload_starts() { + // entrypoint_pid == 0 means no agent yet; identity must fail closed so a + // binary-scoped rule cannot be satisfied by an unattributed connection. + let resolver = ProcfsIdentityResolver { + entrypoint_pid: Arc::new(AtomicU32::new(0)), + }; + assert!(!admits_binary_rule(resolver.resolve_connection( + "127.0.0.1:12345".parse().unwrap(), + "127.0.0.1:3128".parse().unwrap(), + ))); + } + + #[test] + fn unknown_peer_fails_closed() { + // A peer port no live workload connection owns must resolve to an error, + // never a fabricated identity. + let resolver = ProcfsIdentityResolver { + entrypoint_pid: Arc::new(AtomicU32::new(u32::MAX - 1)), + }; + assert!(!admits_binary_rule(resolver.resolve_connection( + "127.0.0.1:1".parse().unwrap(), + "127.0.0.1:3128".parse().unwrap(), + ))); + } +} diff --git a/crates/openshell-supervisor-network/src/l7/tls.rs b/crates/openshell-supervisor-network/src/l7/tls.rs index 2275a60d34..d3def44743 100644 --- a/crates/openshell-supervisor-network/src/l7/tls.rs +++ b/crates/openshell-supervisor-network/src/l7/tls.rs @@ -17,7 +17,6 @@ use std::io::BufReader; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use tokio::io::{AsyncRead, AsyncWrite}; -use tokio::net::TcpStream; use tokio_rustls::{TlsAcceptor, TlsConnector}; const MAX_CACHED_CERTS: usize = 256; @@ -170,11 +169,14 @@ impl ProxyTlsState { /// Accept TLS from a sandbox client, presenting a dynamic cert for the hostname. /// /// Returns a TLS stream that can be used for plaintext HTTP inspection. -pub async fn tls_terminate_client( - client: TcpStream, +pub async fn tls_terminate_client( + client: S, tls_state: &ProxyTlsState, hostname: &str, -) -> Result { +) -> Result +where + S: AsyncRead + AsyncWrite + Unpin + Send, +{ let acceptor = tls_state.acceptor_for(hostname)?; let tls_stream = acceptor.accept(client).await.into_diagnostic()?; Ok(tls_stream) diff --git a/crates/openshell-supervisor-network/src/lib.rs b/crates/openshell-supervisor-network/src/lib.rs index 4fec48b300..a828f75fba 100644 --- a/crates/openshell-supervisor-network/src/lib.rs +++ b/crates/openshell-supervisor-network/src/lib.rs @@ -9,6 +9,7 @@ //! aggregate them. pub mod identity; +pub mod identity_source; pub mod inference_routes; pub mod l7; pub mod opa; diff --git a/crates/openshell-supervisor-process/Cargo.toml b/crates/openshell-supervisor-process/Cargo.toml index 2e2120f1d0..5a2ba05f64 100644 --- a/crates/openshell-supervisor-process/Cargo.toml +++ b/crates/openshell-supervisor-process/Cargo.toml @@ -12,10 +12,12 @@ rust-version.workspace = true [dependencies] openshell-core = { path = "../openshell-core" } +openshell-isolation = { path = "../openshell-isolation" } openshell-ocsf = { path = "../openshell-ocsf" } openshell-policy = { path = "../openshell-policy" } anyhow = { workspace = true } +async-trait = "0.1" base64 = { workspace = true } bytes = { workspace = true } hex = "0.4" diff --git a/crates/openshell-supervisor-process/src/boundary_exec.rs b/crates/openshell-supervisor-process/src/boundary_exec.rs new file mode 100644 index 0000000000..00d5239f29 --- /dev/null +++ b/crates/openshell-supervisor-process/src/boundary_exec.rs @@ -0,0 +1,695 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Co-located implementation of RFC 0012 in-boundary exec. + +use std::collections::HashMap; +use std::os::fd::{AsRawFd, OwnedFd}; +use std::process::{Child, Command, Stdio}; +use std::sync::Arc; + +use async_trait::async_trait; +use nix::pty::{Winsize, openpty}; +use nix::sys::signal::{Signal, killpg}; +use nix::unistd::Pid; +use openshell_core::policy::SandboxPolicy; +use openshell_core::provider_credentials::ProviderCredentialState; +use openshell_isolation::contract::{ + BackendError, BoundaryExec, BoundaryExitStatus, BoundaryInput, BoundaryOutput, BoundaryProcess, + BoundarySignal, BoundaryTerminal, ExecSession, ExecSpec, +}; + +use crate::process::{ProcessEnforcementMode, ResolvedProcessIdentity}; + +/// The co-located executor. Every spawn reuses the same admitted policy and +/// execution-environment controls while taking a fresh provider credential +/// snapshot. +#[derive(Clone)] +pub struct LocalBoundaryExec { + policy: SandboxPolicy, + base_workdir: Option, + netns_fd: Option>, + proxy_url: Option, + ca_file_paths: Option>, + provider_credentials: ProviderCredentialState, + user_environment: HashMap, + resolved_identity: ResolvedProcessIdentity, + enforcement_mode: ProcessEnforcementMode, + runtime: Arc, +} + +impl LocalBoundaryExec { + /// Construct one executor for an active co-located boundary. + #[allow(clippy::too_many_arguments)] + #[must_use] + pub fn new( + policy: SandboxPolicy, + base_workdir: Option, + netns_fd: Option>, + proxy_url: Option, + ca_file_paths: Option>, + provider_credentials: ProviderCredentialState, + user_environment: HashMap, + resolved_identity: ResolvedProcessIdentity, + enforcement_mode: ProcessEnforcementMode, + runtime: Arc, + ) -> Self { + Self { + policy, + base_workdir, + netns_fd, + proxy_url, + ca_file_paths, + provider_credentials, + user_environment, + resolved_identity, + enforcement_mode, + runtime, + } + } + + fn command(&self, spec: &ExecSpec) -> Result { + if spec.program.is_empty() { + return Err(BackendError::Process("exec program is empty".to_string())); + } + let mut command = Command::new(&spec.program); + command.args(&spec.args); + let effective_workdir = spec.workdir.as_deref().or(self.base_workdir.as_deref()); + let (session_user, session_home) = + crate::process::session_user_and_home(&self.policy, effective_workdir); + crate::ssh::apply_child_env( + &mut command, + &session_home, + &session_user, + if spec.pty { "xterm-256color" } else { "dumb" }, + self.proxy_url.as_deref(), + self.ca_file_paths.as_deref(), + &self.provider_credentials.child_env_with_gcp_resolved(), + &self.user_environment, + ); + for (key, value) in &spec.env { + if !key.starts_with("OPENSHELL_") { + command.env(key, value); + } + } + if let Some(workdir) = spec.workdir.as_deref().or(self.base_workdir.as_deref()) { + command.current_dir(workdir); + } + Ok(command) + } + + fn prepare_sandbox( + &self, + workdir: Option<&str>, + ) -> Result, BackendError> { + #[cfg(target_os = "linux")] + { + if self.enforcement_mode.enforces_child_sandbox() { + crate::sandbox::linux::log_sandbox_readiness(&self.policy, workdir); + } + crate::process::prepare_child_sandbox(&self.policy, workdir, self.enforcement_mode) + .map_err(|error| BackendError::Process(error.to_string())) + } + #[cfg(not(target_os = "linux"))] + { + let _ = workdir; + Ok(None) + } + } + + fn spawn_piped(&self, spec: &ExecSpec) -> Result { + self.runtime.ensure_active()?; + let mut command = self.command(spec)?; + command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let effective_workdir = spec.workdir.as_deref().or(self.base_workdir.as_deref()); + #[cfg(target_os = "linux")] + let prepared = self.prepare_sandbox(effective_workdir)?; + crate::ssh::unsafe_pty::install_pre_exec_no_pty( + &mut command, + self.policy.clone(), + effective_workdir.map(str::to_string), + self.netns_fd.as_deref().map(AsRawFd::as_raw_fd), + self.resolved_identity, + self.enforcement_mode, + #[cfg(target_os = "linux")] + prepared, + ); + #[cfg(target_os = "linux")] + let mut child_registry = crate::managed_children::lock(); + let mut child = command + .spawn() + .map_err(|error| BackendError::Process(error.to_string()))?; + let pid = child.id(); + let process_terminal = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let signal_lock = Arc::new(std::sync::Mutex::new(())); + if let Err(error) = + self.runtime + .register_process_group(pid, process_terminal.clone(), signal_lock.clone()) + { + let _ = killpg( + Pid::from_raw(i32::try_from(pid).unwrap_or(i32::MAX)), + Signal::SIGKILL, + ); + let _ = child.wait(); + return Err(error); + } + #[cfg(target_os = "linux")] + let managed_child = child_registry.register(pid); + #[cfg(target_os = "linux")] + drop(child_registry); + let stdin = child.stdin.take().map(|file| -> BoundaryInput { + let fd: OwnedFd = file.into(); + Box::new(tokio::fs::File::from_std(std::fs::File::from(fd))) + }); + let stdout = child + .stdout + .take() + .map(|file| -> BoundaryOutput { + let fd: OwnedFd = file.into(); + Box::new(tokio::fs::File::from_std(std::fs::File::from(fd))) + }) + .ok_or_else(|| BackendError::Process("exec stdout pipe missing".to_string()))?; + let stderr = child.stderr.take().map(|file| -> BoundaryOutput { + let fd: OwnedFd = file.into(); + Box::new(tokio::fs::File::from_std(std::fs::File::from(fd))) + }); + let process = Arc::new(LocalExecProcess::new( + child, + pid, + self.runtime.clone(), + process_terminal, + signal_lock, + #[cfg(target_os = "linux")] + managed_child, + )); + Ok(SpawnedExec { + session: Some(ExecSession { + process: process.clone(), + stdin, + stdout, + stderr, + terminal: None, + }), + process, + armed: true, + }) + } + + fn spawn_pty(&self, spec: &ExecSpec) -> Result { + self.runtime.ensure_active()?; + let winsize = Winsize { + ws_row: 24, + ws_col: 80, + ws_xpixel: 0, + ws_ypixel: 0, + }; + let pty = openpty(Some(&winsize), None) + .map_err(|error| BackendError::Process(error.to_string()))?; + let master = std::fs::File::from(pty.master); + let slave = std::fs::File::from(pty.slave); + let slave_fd = slave.as_raw_fd(); + let input = master + .try_clone() + .map_err(|error| BackendError::Process(error.to_string()))?; + let output = master + .try_clone() + .map_err(|error| BackendError::Process(error.to_string()))?; + let stdin = slave + .try_clone() + .map_err(|error| BackendError::Process(error.to_string()))?; + let stdout = slave + .try_clone() + .map_err(|error| BackendError::Process(error.to_string()))?; + let mut command = self.command(spec)?; + command.stdin(stdin).stdout(stdout).stderr(slave); + let effective_workdir = spec.workdir.as_deref().or(self.base_workdir.as_deref()); + #[cfg(target_os = "linux")] + let prepared = self.prepare_sandbox(effective_workdir)?; + crate::ssh::unsafe_pty::install_pre_exec( + &mut command, + self.policy.clone(), + effective_workdir.map(str::to_string), + slave_fd, + self.netns_fd.as_deref().map(AsRawFd::as_raw_fd), + self.resolved_identity, + self.enforcement_mode, + #[cfg(target_os = "linux")] + prepared, + ); + #[cfg(target_os = "linux")] + let mut child_registry = crate::managed_children::lock(); + let mut child = command + .spawn() + .map_err(|error| BackendError::Process(error.to_string()))?; + let pid = child.id(); + let process_terminal = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let signal_lock = Arc::new(std::sync::Mutex::new(())); + if let Err(error) = + self.runtime + .register_process_group(pid, process_terminal.clone(), signal_lock.clone()) + { + let _ = killpg( + Pid::from_raw(i32::try_from(pid).unwrap_or(i32::MAX)), + Signal::SIGKILL, + ); + let _ = child.wait(); + return Err(error); + } + #[cfg(target_os = "linux")] + let managed_child = child_registry.register(pid); + #[cfg(target_os = "linux")] + drop(child_registry); + let terminal: Arc = Arc::new(LocalTerminal { master }); + let process = Arc::new(LocalExecProcess::new( + child, + pid, + self.runtime.clone(), + process_terminal, + signal_lock, + #[cfg(target_os = "linux")] + managed_child, + )); + Ok(SpawnedExec { + session: Some(ExecSession { + process: process.clone(), + stdin: Some(Box::new(tokio::fs::File::from_std(input))), + stdout: Box::new(tokio::fs::File::from_std(output)), + stderr: None, + terminal: Some(terminal), + }), + process, + armed: true, + }) + } +} + +struct SpawnedExec { + session: Option, + process: Arc, + armed: bool, +} + +impl SpawnedExec { + fn into_session(mut self) -> ExecSession { + self.armed = false; + self.session.take().expect("spawned exec session") + } +} + +impl Drop for SpawnedExec { + fn drop(&mut self) { + if self.armed { + let _ = self.process.deliver(Signal::SIGKILL); + } + } +} + +#[async_trait] +impl BoundaryExec for LocalBoundaryExec { + async fn exec(&self, spec: ExecSpec) -> Result { + let executor = self.clone(); + let (send, receive) = tokio::sync::oneshot::channel(); + tokio::task::spawn_blocking(move || { + let result = if spec.pty { + executor.spawn_pty(&spec) + } else { + executor.spawn_piped(&spec) + }; + // If the caller cancelled, either send fails and drops the armed + // process guard here, or the queued guard is dropped with the + // receiver. Both paths terminate an unobservable exec process. + let _ = send.send(result); + }); + receive + .await + .map_err(|_| BackendError::Process("exec spawn task failed".to_string()))? + .map(SpawnedExec::into_session) + } +} + +struct LocalTerminal { + master: std::fs::File, +} + +#[async_trait] +impl BoundaryTerminal for LocalTerminal { + async fn resize(&self, cols: u16, rows: u16) -> Result<(), BackendError> { + crate::ssh::unsafe_pty::set_winsize( + self.master.as_raw_fd(), + Winsize { + ws_row: rows.max(1), + ws_col: cols.max(1), + ws_xpixel: 0, + ws_ypixel: 0, + }, + ) + .map_err(|error| BackendError::Process(error.to_string())) + } +} + +struct LocalExecProcess { + pid: u32, + result: Arc>>>, + exited: Arc, + runtime: Arc, + terminal: Arc, + signal_lock: Arc>, +} + +impl LocalExecProcess { + fn new( + child: Child, + pid: u32, + runtime: Arc, + terminal: Arc, + signal_lock: Arc>, + #[cfg(target_os = "linux")] managed_child: Option, + ) -> Self { + let result = Arc::new(std::sync::Mutex::new(None)); + let exited = Arc::new(tokio::sync::Notify::new()); + let result_for_wait = result.clone(); + let exited_for_wait = exited.clone(); + let runtime_for_wait = runtime.clone(); + let terminal_for_wait = terminal.clone(); + let registration_terminal = terminal.clone(); + #[cfg(target_os = "linux")] + let signal_lock_for_wait = signal_lock.clone(); + tokio::spawn(async move { + let waited = tokio::task::spawn_blocking(move || { + let mut child = child; + #[cfg(target_os = "linux")] + { + crate::managed_children::wait_until_terminal(pid)?; + let _signal_guard = signal_lock_for_wait + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + terminal_for_wait.store(true, std::sync::atomic::Ordering::Release); + let result = child.wait(); + if let Some(managed_child) = managed_child { + crate::managed_children::unregister(managed_child); + } + result + } + #[cfg(not(target_os = "linux"))] + { + let result = child.wait(); + terminal_for_wait.store(true, std::sync::atomic::Ordering::Release); + result + } + }) + .await + .map_err(|error| error.to_string()) + .and_then(|status| status.map_err(|error| error.to_string())) + .map(|status| { + #[cfg(unix)] + { + use std::os::unix::process::ExitStatusExt; + if let Some(signal) = status.signal() { + return BoundaryExitStatus::Signaled(signal); + } + } + BoundaryExitStatus::Exited(status.code().unwrap_or(1)) + }); + runtime_for_wait.unregister_process_group(pid, ®istration_terminal); + if let Ok(mut slot) = result_for_wait.lock() { + *slot = Some(waited); + } + exited_for_wait.notify_waiters(); + }); + Self { + pid, + result, + exited, + runtime, + terminal, + signal_lock, + } + } + + fn deliver(&self, signal: Signal) -> Result<(), BackendError> { + self.runtime.ensure_active()?; + let _signal_guard = self + .signal_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if self.terminal.load(std::sync::atomic::Ordering::Acquire) { + return Err(BackendError::Terminated("process has exited".to_string())); + } + let pid = i32::try_from(self.pid).unwrap_or(i32::MAX); + killpg(Pid::from_raw(pid), signal).map_err(|error| BackendError::Process(error.to_string())) + } +} + +#[async_trait] +impl BoundaryProcess for LocalExecProcess { + async fn wait(&self) -> Result { + loop { + let notified = self.exited.notified(); + let result = self + .result + .lock() + .map_err(|_| BackendError::Process("exec result lock poisoned".to_string()))? + .clone(); + if let Some(result) = result { + return result.map_err(BackendError::Process); + } + notified.await; + } + } + + async fn signal(&self, signal: BoundarySignal) -> Result<(), BackendError> { + self.deliver(match signal { + BoundarySignal::Term => Signal::SIGTERM, + BoundarySignal::Kill => Signal::SIGKILL, + BoundarySignal::Int => Signal::SIGINT, + BoundarySignal::Hup => Signal::SIGHUP, + }) + } + + async fn terminate(&self) -> Result<(), BackendError> { + self.deliver(Signal::SIGKILL) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + fn executor() -> LocalBoundaryExec { + LocalBoundaryExec::new( + SandboxPolicy { + version: 1, + filesystem: openshell_core::policy::FilesystemPolicy::default(), + network: openshell_core::policy::NetworkPolicy::default(), + landlock: openshell_core::policy::LandlockPolicy::default(), + process: openshell_core::policy::ProcessPolicy::default(), + }, + None, + None, + None, + None, + ProviderCredentialState::from_environment( + 0, + HashMap::new(), + HashMap::new(), + HashMap::new(), + ), + HashMap::new(), + ResolvedProcessIdentity::default(), + ProcessEnforcementMode::NetworkOnly, + crate::boundary_io::BoundaryRuntimeState::new(), + ) + } + + #[tokio::test] + async fn non_pty_exec_preserves_stdin_stdout_and_stderr() { + let mut session = executor() + .exec(ExecSpec { + program: "/bin/sh".to_string(), + args: vec![ + "-c".to_string(), + "read line; printf 'out:%s' \"$line\"; printf 'err:%s' \"$line\" >&2" + .to_string(), + ], + env: vec![], + workdir: None, + pty: false, + }) + .await + .expect("spawn exec"); + let mut stdin = session.stdin.take().expect("stdin"); + stdin.write_all(b"value\n").await.expect("write stdin"); + drop(stdin); + let mut stdout = String::new(); + let mut stderr = String::new(); + session + .stdout + .read_to_string(&mut stdout) + .await + .expect("read stdout"); + session + .stderr + .take() + .expect("stderr") + .read_to_string(&mut stderr) + .await + .expect("read stderr"); + assert_eq!( + session.process.wait().await.unwrap(), + BoundaryExitStatus::Exited(0) + ); + assert_eq!(stdout, "out:value"); + assert_eq!(stderr, "err:value"); + } + + #[tokio::test] + async fn exec_rejects_after_boundary_end() { + let executor = executor(); + executor.runtime.deactivate(); + let result = executor + .exec(ExecSpec { + program: "/bin/true".to_string(), + args: vec![], + env: vec![], + workdir: None, + pty: false, + }) + .await; + assert!(matches!(result, Err(BackendError::Terminated(_)))); + } + + #[tokio::test] + async fn failed_exec_leaves_boundary_active_without_registered_processes() { + let executor = executor(); + let runtime = executor.runtime.clone(); + let result = executor + .exec(ExecSpec { + program: "/definitely/missing/openshell-exec".to_string(), + args: vec![], + env: vec![], + workdir: None, + pty: false, + }) + .await; + assert!(matches!(result, Err(BackendError::Process(_)))); + runtime.ensure_active().expect("boundary remains active"); + assert_eq!(runtime.registered_process_group_count(), 0); + } + + #[tokio::test] + async fn cancelled_exec_does_not_leave_a_registered_process() { + let executor = executor(); + let runtime = executor.runtime.clone(); + let task = tokio::spawn(async move { + executor + .exec(ExecSpec { + program: "/bin/sleep".to_string(), + args: vec!["30".to_string()], + env: vec![], + workdir: None, + pty: false, + }) + .await + }); + tokio::task::yield_now().await; + task.abort(); + let _ = task.await; + + // Give the detached blocking setup time to reach its cancelled + // handoff, including the case where cancellation won before spawn. + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while runtime.registered_process_group_count() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("cancelled exec process must be terminated and reaped"); + runtime.ensure_active().expect("boundary remains active"); + } + + #[tokio::test] + async fn dropping_undelivered_exec_guard_terminates_process() { + let executor = executor(); + let runtime = executor.runtime.clone(); + let spawned = tokio::task::spawn_blocking(move || { + executor.spawn_piped(&ExecSpec { + program: "/bin/sleep".to_string(), + args: vec!["30".to_string()], + env: vec![], + workdir: None, + pty: false, + }) + }) + .await + .expect("spawn task") + .expect("spawn exec"); + assert_eq!(runtime.registered_process_group_count(), 1); + + // This is the post-send/pre-receive cancellation case: dropping the + // queued ownership guard must kill the process before it is observable. + drop(spawned); + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while runtime.registered_process_group_count() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("undelivered exec process must be terminated and reaped"); + runtime.ensure_active().expect("boundary remains active"); + } + + #[tokio::test] + async fn completed_exec_removes_its_process_group_registration() { + let executor = executor(); + let runtime = executor.runtime.clone(); + let session = executor + .exec(ExecSpec { + program: "/bin/true".to_string(), + args: vec![], + env: vec![], + workdir: None, + pty: false, + }) + .await + .expect("spawn exec"); + assert_eq!( + session.process.wait().await.unwrap(), + BoundaryExitStatus::Exited(0) + ); + assert_eq!(runtime.registered_process_group_count(), 0); + } + + #[tokio::test] + async fn pty_exec_exposes_resize_and_stable_wait() { + let session = executor() + .exec(ExecSpec { + program: "/bin/sh".to_string(), + args: vec!["-c".to_string(), "exit 7".to_string()], + env: vec![], + workdir: None, + pty: true, + }) + .await + .expect("spawn pty exec"); + session + .terminal + .as_ref() + .expect("terminal") + .resize(120, 40) + .await + .expect("resize"); + assert_eq!( + session.process.wait().await.unwrap(), + BoundaryExitStatus::Exited(7) + ); + assert_eq!( + session.process.wait().await.unwrap(), + BoundaryExitStatus::Exited(7) + ); + } +} diff --git a/crates/openshell-supervisor-process/src/boundary_io.rs b/crates/openshell-supervisor-process/src/boundary_io.rs new file mode 100644 index 0000000000..c09974d2df --- /dev/null +++ b/crates/openshell-supervisor-process/src/boundary_io.rs @@ -0,0 +1,317 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The in-pod [`BoundaryPortForward`] interface (RFC 0012 runtime contract). +//! +//! This is the live in-boundary port-forward for the in-pod placement. It lives +//! in this crate on purpose: the SSH server and supervisor session that consume +//! it are here, and so is the primitive it wraps +//! ([`connect_in_netns`](crate::ssh::connect_in_netns)). The interface trait +//! lives in the lower `openshell-isolation` crate, so this crate depends on the +//! trait (process -> isolation -> core, acyclic) and the SSH server drives a +//! `&dyn BoundaryPortForward` without depending on the backend. +//! +//! The SSH server and supervisor session are wired to this through the +//! `RunningBoundary::port_forward()` accessor: swapping in a kernel-separated +//! backend swaps this implementation (where `connect` tunnels into the guest) +//! and touches no consumer code. + +use async_trait::async_trait; +use openshell_isolation::contract::{ + BackendError, BoundaryDuplexStream, BoundaryPortForward, LoopbackTarget, +}; +use std::collections::HashMap; +use std::os::fd::{AsRawFd, OwnedFd}; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::{Arc, Mutex}; + +/// Shared liveness and child-process ownership for one active boundary. +pub struct BoundaryRuntimeState { + state: AtomicU8, + process_groups: Mutex>, + exclusive_pid_namespace: bool, +} + +impl BoundaryRuntimeState { + #[must_use] + pub fn new() -> Arc { + Arc::new(Self { + state: AtomicU8::new(0), + process_groups: Mutex::new(HashMap::new()), + exclusive_pid_namespace: false, + }) + } + + /// Construct state for a boundary that exclusively owns its PID namespace. + #[must_use] + pub fn new_exclusive_pid_namespace() -> Arc { + Arc::new(Self { + state: AtomicU8::new(0), + process_groups: Mutex::new(HashMap::new()), + exclusive_pid_namespace: true, + }) + } + + #[must_use] + pub const fn requires_dedicated_process_group(&self) -> bool { + self.exclusive_pid_namespace + } + + pub fn ensure_active(&self) -> Result<(), BackendError> { + if self.state.load(Ordering::Acquire) == 0 { + Ok(()) + } else { + Err(BackendError::Terminated("boundary has ended".to_string())) + } + } + + #[must_use] + pub fn is_active(&self) -> bool { + self.state.load(Ordering::Acquire) == 0 + } + + #[must_use] + pub fn enforcement_was_lost(&self) -> bool { + self.state.load(Ordering::Acquire) == 2 + } + + pub fn register_process_group( + &self, + pid: u32, + terminal: Arc, + signal_lock: Arc>, + ) -> Result<(), BackendError> { + let mut groups = self + .process_groups + .lock() + .map_err(|_| BackendError::Process("boundary process registry poisoned".to_string()))?; + self.ensure_active()?; + groups.insert( + pid, + RegisteredProcessGroup { + pid, + terminal, + signal_lock, + }, + ); + Ok(()) + } + + pub fn unregister_process_group( + &self, + pid: u32, + terminal: &Arc, + ) { + if let Ok(mut groups) = self.process_groups.lock() + && groups + .get(&pid) + .is_some_and(|group| Arc::ptr_eq(&group.terminal, terminal)) + { + groups.remove(&pid); + } + } + + #[cfg(test)] + pub fn registered_process_group_count(&self) -> usize { + self.process_groups.lock().map_or(0, |groups| groups.len()) + } + + /// End the boundary and terminate every registered workload process group. + pub fn deactivate(&self) { + if self + .state + .compare_exchange(0, 1, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + { + self.terminate_registered_processes(); + } + } + + /// End the boundary because required standing enforcement was lost. + /// + /// Returns `true` only to the caller that won the active-to-terminated + /// transition. A concurrent normal teardown cannot later be reclassified + /// as enforcement loss. + pub fn deactivate_for_enforcement_loss(&self) -> bool { + if self + .state + .compare_exchange(0, 2, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return false; + } + self.terminate_registered_processes(); + true + } + + fn terminate_registered_processes(&self) { + let groups = self + .process_groups + .lock() + .map(|groups| groups.values().cloned().collect::>()) + .unwrap_or_default(); + for group in groups { + group.terminate(); + } + } +} + +#[derive(Clone)] +struct RegisteredProcessGroup { + pid: u32, + terminal: Arc, + signal_lock: Arc>, +} + +impl RegisteredProcessGroup { + fn terminate(&self) { + let _signal_guard = self + .signal_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if self.terminal.load(Ordering::Acquire) { + return; + } + if let Ok(pid) = i32::try_from(self.pid) { + let _ = nix::sys::signal::killpg( + nix::unistd::Pid::from_raw(pid), + nix::sys::signal::Signal::SIGKILL, + ); + } + } +} + +/// In-pod loopback port-forward: connects to a loopback target from inside the +/// workload's network namespace via [`connect_in_netns`](crate::ssh::connect_in_netns). +pub struct NetnsPortForward { + /// File descriptor of the boundary's network namespace, or `None` to + /// connect from the supervisor's own namespace. + netns_fd: Option>, + runtime: Option>, +} + +impl NetnsPortForward { + #[must_use] + pub fn new(netns_fd: Option>, runtime: Option>) -> Self { + Self { netns_fd, runtime } + } +} + +#[async_trait] +impl BoundaryPortForward for NetnsPortForward { + async fn connect(&self, target: LoopbackTarget) -> Result { + if let Some(runtime) = &self.runtime { + runtime.ensure_active()?; + } + let addr = std::net::SocketAddr::new(target.host(), target.port()); + let addr_string = addr.to_string(); + let stream = crate::ssh::connect_in_netns( + &addr_string, + self.netns_fd.as_deref().map(AsRawFd::as_raw_fd), + ) + .await + .map_err(|e| BackendError::Process(format!("port-forward connect to {addr}: {e}")))?; + if let Some(runtime) = &self.runtime { + runtime.ensure_active()?; + } + Ok(Box::new(stream)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::Ipv4Addr; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + /// Stands in for the SSH server's port-forward path: connect through the + /// interface, write, read the echo. With `netns_fd: None` the connect happens in + /// the supervisor's namespace, so this exercises the real primitive without + /// requiring a network namespace. + #[tokio::test] + async fn port_forward_connects_and_round_trips() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let (mut sock, _) = listener.accept().await.unwrap(); + let mut buf = [0u8; 4]; + sock.read_exact(&mut buf).await.unwrap(); + sock.write_all(&buf).await.unwrap(); + }); + + let pf = NetnsPortForward::new(None, None); + let target = + LoopbackTarget::new(Ipv4Addr::LOCALHOST.into(), addr.port()).expect("loopback target"); + let mut conn = pf.connect(target).await.expect("connect through interface"); + conn.write_all(b"ping").await.unwrap(); + let mut buf = [0u8; 4]; + conn.read_exact(&mut buf).await.unwrap(); + assert_eq!(&buf, b"ping"); + } + + /// Drive the port-forward interface through a generic `&dyn` consumer, proving a + /// kernel-separated backend (tunneling into a guest) would use the same call. + #[tokio::test] + async fn port_forward_is_driven_via_dyn() { + async fn forward_one(pf: &dyn BoundaryPortForward, target: LoopbackTarget) -> bool { + pf.connect(target).await.is_ok() + } + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + let _ = listener.accept().await; + }); + let pf = NetnsPortForward::new(None, None); + let target = LoopbackTarget::new(Ipv4Addr::LOCALHOST.into(), addr.port()).unwrap(); + assert!(forward_one(&pf, target).await); + } + + #[tokio::test] + async fn port_forward_rejects_after_boundary_end() { + let runtime = BoundaryRuntimeState::new(); + let pf = NetnsPortForward::new(None, Some(runtime.clone())); + runtime.deactivate(); + let target = LoopbackTarget::new(Ipv4Addr::LOCALHOST.into(), 1).unwrap(); + assert!(matches!( + pf.connect(target).await, + Err(BackendError::Terminated(_)) + )); + } + + #[tokio::test] + async fn failed_port_forward_keeps_boundary_active() { + let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)) + .await + .unwrap(); + let port = listener.local_addr().unwrap().port(); + drop(listener); + let runtime = BoundaryRuntimeState::new(); + let pf = NetnsPortForward::new(None, Some(runtime.clone())); + let target = LoopbackTarget::new(Ipv4Addr::LOCALHOST.into(), port).unwrap(); + assert!(matches!( + pf.connect(target).await, + Err(BackendError::Process(_)) + )); + runtime.ensure_active().expect("boundary remains active"); + } + + #[test] + fn stale_unregister_preserves_reused_process_group_registration() { + let runtime = BoundaryRuntimeState::new(); + let first_terminal = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let second_terminal = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let pid = 42; + runtime + .register_process_group(pid, first_terminal.clone(), Arc::new(Mutex::new(()))) + .expect("first registration"); + runtime + .register_process_group(pid, second_terminal.clone(), Arc::new(Mutex::new(()))) + .expect("replacement registration"); + + runtime.unregister_process_group(pid, &first_terminal); + assert_eq!(runtime.registered_process_group_count(), 1); + + runtime.unregister_process_group(pid, &second_terminal); + assert_eq!(runtime.registered_process_group_count(), 0); + } +} diff --git a/crates/openshell-supervisor-process/src/lib.rs b/crates/openshell-supervisor-process/src/lib.rs index 743942faa4..ee6bedeb22 100644 --- a/crates/openshell-supervisor-process/src/lib.rs +++ b/crates/openshell-supervisor-process/src/lib.rs @@ -8,6 +8,8 @@ //! and log push. Populated by follow-up commits as modules migrate out of //! `openshell-sandbox`. +pub mod boundary_exec; +pub mod boundary_io; pub mod child_env; pub mod debug_rpc; #[cfg(unix)] diff --git a/crates/openshell-supervisor-process/src/managed_children.rs b/crates/openshell-supervisor-process/src/managed_children.rs index 311c80693f..04f4114a04 100644 --- a/crates/openshell-supervisor-process/src/managed_children.rs +++ b/crates/openshell-supervisor-process/src/managed_children.rs @@ -10,44 +10,146 @@ #![cfg(target_os = "linux")] -use std::collections::HashSet; -use std::sync::{LazyLock, Mutex}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{LazyLock, Mutex, MutexGuard}; -static MANAGED_CHILDREN: LazyLock>> = - LazyLock::new(|| Mutex::new(HashSet::new())); +static MANAGED_CHILDREN: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); +static NEXT_GENERATION: AtomicU64 = AtomicU64::new(1); -/// Add `pid` to the supervised-child set. Non-positive or out-of-range values -/// are silently ignored. -pub fn register(pid: u32) { - let Ok(pid) = i32::try_from(pid) else { - return; - }; - if pid <= 0 { - return; +/// Identity of one registry entry. The generation prevents an old waiter from +/// removing a newer child that reused the same numeric PID after reap. +#[derive(Clone, Copy)] +pub struct ManagedChild { + pid: i32, + generation: u64, +} + +/// A managed-child registration accepted by [`unregister`]. +/// +/// New boundary-owned processes retain a generation-bearing token. Legacy +/// supervisor paths still identify their child by PID; supporting both keeps +/// the registry race-safe for new code without forcing an unrelated rewrite +/// of the canonical main-process and SSH paths. +pub enum ManagedChildRegistration { + Token(ManagedChild), + Pid(u32), +} + +impl From for ManagedChildRegistration { + fn from(value: ManagedChild) -> Self { + Self::Token(value) } - if let Ok(mut children) = MANAGED_CHILDREN.lock() { - children.insert(pid); +} + +impl From for ManagedChildRegistration { + fn from(value: u32) -> Self { + Self::Pid(value) } } -/// Remove `pid` from the supervised-child set. Non-positive or out-of-range -/// values are silently ignored. -pub fn unregister(pid: u32) { - let Ok(pid) = i32::try_from(pid) else { - return; - }; - if pid <= 0 { - return; +/// Exclusive access to the managed-child registry. +/// +/// A process spawner holds this guard from immediately before `spawn` or +/// `fork` until the returned PID is registered. The orphan reaper holds the +/// same guard while deciding whether to reap an exited child. This closes the +/// otherwise unavoidable window in which a fast-exiting managed child exists +/// but its PID has not yet been published. +pub struct RegistryGuard(MutexGuard<'static, HashMap>); + +impl RegistryGuard { + /// Add a newly spawned managed child. + pub fn register(&mut self, pid: u32) -> Option { + let Ok(pid) = i32::try_from(pid) else { + return None; + }; + if pid <= 0 { + return None; + } + let generation = NEXT_GENERATION.fetch_add(1, Ordering::Relaxed); + self.0.insert(pid, generation); + Some(ManagedChild { pid, generation }) + } + + /// Return whether the PID belongs to an explicit waiter. + #[must_use] + pub fn contains(&self, pid: i32) -> bool { + self.0.contains_key(&pid) } +} + +/// Lock the registry for an atomic spawn-and-register or inspect-and-reap +/// operation. +pub fn lock() -> RegistryGuard { + RegistryGuard( + MANAGED_CHILDREN + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ) +} + +/// Register a child for a legacy caller that cannot retain a generation token. +pub fn register(pid: u32) { + let _ = lock().register(pid); +} + +/// Remove exactly this supervised-child registration. A newer registration +/// for a reused PID is preserved. +pub fn unregister(child: impl Into) { if let Ok(mut children) = MANAGED_CHILDREN.lock() { - children.remove(&pid); + match child.into() { + ManagedChildRegistration::Token(child) + if children.get(&child.pid) == Some(&child.generation) => + { + children.remove(&child.pid); + } + ManagedChildRegistration::Pid(pid) => { + if let Ok(pid) = i32::try_from(pid) { + children.remove(&pid); + } + } + ManagedChildRegistration::Token(_) => {} + } } } /// Return `true` if `pid` is currently in the supervised-child set. #[must_use] pub fn is_managed(pid: i32) -> bool { - MANAGED_CHILDREN - .lock() - .is_ok_and(|children| children.contains(&pid)) + lock().contains(pid) +} + +/// Wait until a managed child is terminal without reaping it. +/// +/// Keeping the child as a zombie prevents PID/process-group reuse until the +/// owner publishes terminal state and performs the final wait. +pub fn wait_until_terminal(pid: u32) -> std::io::Result<()> { + use nix::sys::wait::{Id, WaitPidFlag, waitid}; + let pid = i32::try_from(pid) + .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "PID out of range"))?; + waitid( + Id::Pid(nix::unistd::Pid::from_raw(pid)), + WaitPidFlag::WEXITED | WaitPidFlag::WNOWAIT, + ) + .map(|_| ()) + .map_err(std::io::Error::other) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stale_unregister_preserves_reused_pid_registration() { + let pid = i32::MAX as u32; + let first = lock().register(pid).expect("first registration"); + let second = lock().register(pid).expect("replacement registration"); + + unregister(first); + assert!(is_managed(i32::try_from(pid).expect("test pid"))); + + unregister(second); + assert!(!is_managed(i32::try_from(pid).expect("test pid"))); + } } diff --git a/crates/openshell-supervisor-process/src/netns/nft_ruleset.rs b/crates/openshell-supervisor-process/src/netns/nft_ruleset.rs index aef95b6068..2fb075b420 100644 --- a/crates/openshell-supervisor-process/src/netns/nft_ruleset.rs +++ b/crates/openshell-supervisor-process/src/netns/nft_ruleset.rs @@ -24,7 +24,7 @@ pub struct NftCommand { pub required: bool, } -/// Generate nft commands for sandbox network bypass enforcement. +/// Generate the legacy nft commands for sandbox bypass detection. /// /// Creates an `inet` family table (handles both IPv4 and IPv6) with rules that: /// 1. Accept traffic to the proxy (IPv4 only) @@ -34,11 +34,34 @@ pub struct NftCommand { /// /// If `log_prefix` is provided, log rules are inserted before each reject rule /// so that bypass attempts are recorded in the kernel ring buffer before being -/// rejected. Log rules are always non-required since they need `nf_log` support. +/// rejected. Log rules are non-required since they need `nf_log` support. pub fn generate_bypass_commands( host_ip: &str, proxy_port: u16, log_prefix: Option<&str>, +) -> Vec { + generate_commands(host_ip, proxy_port, log_prefix, false) +} + +/// Generate the RFC 0012 default-deny egress ceiling. +/// +/// Only the exact proxy destination and loopback are accepted. TCP and UDP +/// rejects are optional fast-fail behavior; the base-chain drop policy covers +/// every address family and protocol. No blanket conntrack exception is +/// installed because pre-existing or related flows must not bypass mediation. +pub fn generate_egress_ceiling_commands( + host_ip: &str, + proxy_port: u16, + log_prefix: Option<&str>, +) -> Vec { + generate_commands(host_ip, proxy_port, log_prefix, true) +} + +fn generate_commands( + host_ip: &str, + proxy_port: u16, + log_prefix: Option<&str>, + default_deny: bool, ) -> Vec { let table = "openshell_bypass"; let mut cmds = vec![ @@ -52,7 +75,11 @@ pub fn generate_bypass_commands( "inet", table, "output", - "{ type filter hook output priority 0; policy accept; }", + if default_deny { + "{ type filter hook output priority 0; policy drop; }" + } else { + "{ type filter hook output priority 0; policy accept; }" + }, ], ), nft_cmd( @@ -78,7 +105,10 @@ pub fn generate_bypass_commands( "add", "rule", "inet", table, "output", "oifname", "lo", "accept", ], ), - nft_cmd( + ]; + + if !default_deny { + cmds.push(nft_cmd( false, &[ "add", @@ -91,8 +121,8 @@ pub fn generate_bypass_commands( "established,related", "accept", ], - ), - ]; + )); + } if let Some(prefix) = log_prefix { let quoted = nft_quote(prefix); @@ -106,7 +136,7 @@ pub fn generate_bypass_commands( } cmds.push(nft_cmd( - true, + !default_deny, &[ "add", "rule", @@ -127,7 +157,7 @@ pub fn generate_bypass_commands( ], )); cmds.push(nft_cmd( - true, + !default_deny, &[ "add", "rule", @@ -160,7 +190,7 @@ pub fn generate_bypass_commands( } cmds.push(nft_cmd( - true, + !default_deny, &[ "add", "rule", @@ -181,7 +211,7 @@ pub fn generate_bypass_commands( ], )); cmds.push(nft_cmd( - true, + !default_deny, &[ "add", "rule", @@ -598,6 +628,25 @@ mod tests { assert!(text.contains("type filter hook output priority 0; policy accept;")); } + #[test] + fn in_pod_ceiling_is_default_deny_for_all_protocols() { + let text = all_strs(&generate_egress_ceiling_commands("10.0.2.2", 3128, None)); + assert!(text.contains("policy drop")); + assert!(!text.contains("policy accept")); + assert!(!text.contains("ct state")); + } + + #[test] + fn in_pod_reject_rules_are_optional_fast_fail_over_default_drop() { + let commands = generate_egress_ceiling_commands("10.0.2.2", 3128, None); + for command in commands + .iter() + .filter(|command| command.args.iter().any(|argument| argument == "reject")) + { + assert!(!command.required); + } + } + #[test] fn proxy_accept_rule_uses_provided_ip_and_port() { let cmds = generate_bypass_commands("172.16.0.1", 9999, None); @@ -611,7 +660,7 @@ mod tests { let text = all_strs(&cmds); let proxy_pos = text.find("ip daddr").unwrap(); let lo_pos = text.find("oifname lo").unwrap(); - let ct_pos = text.find("ct state established,related").unwrap(); + let ct_pos = text.find("ct state established").unwrap(); let reject_pos = text.find("reject with icmp type").unwrap(); assert!(proxy_pos < lo_pos); diff --git a/crates/openshell-supervisor-process/src/sandbox/linux/seccomp.rs b/crates/openshell-supervisor-process/src/sandbox/linux/seccomp.rs index a9c67af95a..ddd37a502d 100644 --- a/crates/openshell-supervisor-process/src/sandbox/linux/seccomp.rs +++ b/crates/openshell-supervisor-process/src/sandbox/linux/seccomp.rs @@ -838,4 +838,43 @@ mod tests { "socket(AF_NETLINK, SOCK_RAW, NETLINK_SOCK_DIAG) should be blocked with EPERM" ); } + + #[test] + fn behavioral_block_mode_denies_inet_and_packet_sockets() { + let filter = build_filter(false).unwrap(); + let pid = unsafe { libc::fork() }; + assert!(pid >= 0, "fork failed"); + if pid == 0 { + unsafe { + libc::prctl(libc::PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0); + apply_filter(&filter).expect("apply block-mode filter"); + for (domain, socket_type, protocol) in [ + (libc::AF_INET, libc::SOCK_STREAM, 0), + (libc::AF_INET6, libc::SOCK_DGRAM, 0), + (libc::AF_PACKET, libc::SOCK_RAW, 0), + ] { + let fd = libc::socket(domain, socket_type, protocol); + let errno = *libc::__errno_location(); + if fd >= 0 || errno != libc::EPERM { + if fd >= 0 { + libc::close(fd); + } + libc::_exit(1); + } + } + let unix_fd = libc::socket(libc::AF_UNIX, libc::SOCK_STREAM, 0); + if unix_fd < 0 { + libc::_exit(1); + } + libc::close(unix_fd); + libc::_exit(0); + } + } + let mut status: libc::c_int = 0; + unsafe { libc::waitpid(pid, &mut status, 0) }; + assert!( + unsafe { libc::WIFEXITED(status) && libc::WEXITSTATUS(status) == 0 }, + "block mode must deny IPv4, IPv6, and packet sockets while retaining Unix IPC" + ); + } } diff --git a/crates/openshell-supervisor-process/src/ssh.rs b/crates/openshell-supervisor-process/src/ssh.rs index 893967b2ac..c0b5a2c30f 100644 --- a/crates/openshell-supervisor-process/src/ssh.rs +++ b/crates/openshell-supervisor-process/src/ssh.rs @@ -1134,7 +1134,7 @@ impl Default for PtyRequest { } #[allow(clippy::too_many_arguments)] -fn apply_child_env( +pub(crate) fn apply_child_env( cmd: &mut Command, session_home: &str, session_user: &str, @@ -1530,7 +1530,7 @@ fn spawn_pipe_exec( Ok(sender) } -mod unsafe_pty { +pub(crate) mod unsafe_pty { #[cfg(not(target_os = "linux"))] use super::sandbox; use super::{