From ac3cd28864c336639780bbc4e1451d453b469730 Mon Sep 17 00:00:00 2001 From: Jordan Ganoff Date: Tue, 21 Jul 2026 09:37:24 -0700 Subject: [PATCH] feat(isolation): implement RFC 0012 isolation backend Implement the Isolation Backend interface defined by RFC 0012 and migrate the in-pod logic that was coupled to the supervisor behind an in-pod backend. Squashed from the RFC 0012 implementation series (POC + subsequent RFC alignment revisions). Signed-off-by: Jordan Ganoff --- AGENTS.md | 1 + Cargo.lock | 19 + architecture/sandbox.md | 16 + crates/openshell-core/src/sandbox_env.rs | 11 + crates/openshell-driver-docker/Cargo.toml | 1 + crates/openshell-driver-docker/README.md | 1 + crates/openshell-driver-docker/src/lib.rs | 6 + crates/openshell-driver-kubernetes/Cargo.toml | 1 + .../openshell-driver-kubernetes/src/driver.rs | 7 + crates/openshell-driver-podman/Cargo.toml | 1 + crates/openshell-driver-podman/README.md | 1 + .../openshell-driver-podman/src/container.rs | 6 + crates/openshell-driver-vm/Cargo.toml | 1 + crates/openshell-driver-vm/src/driver.rs | 8 + crates/openshell-isolation/Cargo.toml | 22 + crates/openshell-isolation/src/contract.rs | 675 ++++++++++++++ .../openshell-isolation/src/contract/tests.rs | 633 +++++++++++++ crates/openshell-isolation/src/lib.rs | 99 ++ crates/openshell-sandbox/Cargo.toml | 7 + crates/openshell-sandbox/src/inpod.rs | 867 ++++++++++++++++++ crates/openshell-sandbox/src/lib.rs | 787 ++++++++++------ .../openshell-supervisor-network/Cargo.toml | 3 + .../src/identity_source.rs | 126 +++ .../openshell-supervisor-network/src/lib.rs | 1 + .../openshell-supervisor-process/Cargo.toml | 2 + .../src/boundary_io.rs | 90 ++ .../openshell-supervisor-process/src/lib.rs | 1 + .../src/process.rs | 3 + .../openshell-supervisor-process/src/run.rs | 237 ++++- .../openshell-supervisor-process/src/ssh.rs | 46 +- .../src/supervisor_session.rs | 94 +- 31 files changed, 3361 insertions(+), 412 deletions(-) create mode 100644 crates/openshell-isolation/Cargo.toml create mode 100644 crates/openshell-isolation/src/contract.rs create mode 100644 crates/openshell-isolation/src/contract/tests.rs create mode 100644 crates/openshell-isolation/src/lib.rs create mode 100644 crates/openshell-sandbox/src/inpod.rs create mode 100644 crates/openshell-supervisor-network/src/identity_source.rs create mode 100644 crates/openshell-supervisor-process/src/boundary_io.rs diff --git a/AGENTS.md b/AGENTS.md index 540a8d920..487ffecad 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,6 +32,7 @@ These pipelines connect skills into end-to-end workflows. Individual skill files | `crates/openshell-cli/` | CLI binary | User-facing command-line interface | | `crates/openshell-server/` | Gateway server | Control-plane API, sandbox lifecycle, auth boundary | | `crates/openshell-sandbox/` | Sandbox runtime | Container supervision, policy-enforced egress routing | +| `crates/openshell-isolation/` | Isolation backend contract | RFC 0012 `IsolationBackend` trait + types; the supervisor-facing runtime contract for the boundary | | `crates/openshell-policy/` | Policy engine | Filesystem, network, process, and inference constraints | | `crates/openshell-router/` | Privacy router | Privacy-aware LLM routing | | `crates/openshell-bootstrap/` | Gateway metadata | Gateway registration metadata, auth token storage, mTLS bundle storage | diff --git a/Cargo.lock b/Cargo.lock index 528309a97..992258a84 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3896,6 +3896,7 @@ dependencies = [ "bytes", "futures", "openshell-core", + "openshell-isolation", "prost-types", "serde", "serde_json", @@ -3920,6 +3921,7 @@ dependencies = [ "kube-runtime", "miette", "openshell-core", + "openshell-isolation", "openshell-policy", "prost", "prost-types", @@ -3946,6 +3948,7 @@ dependencies = [ "miette", "nix 0.29.0", "openshell-core", + "openshell-isolation", "prost-types", "rustix 1.1.4", "serde", @@ -3974,6 +3977,7 @@ dependencies = [ "nix 0.29.0", "oci-client", "openshell-core", + "openshell-isolation", "openshell-policy", "openshell-vfio", "polling", @@ -4015,6 +4019,15 @@ dependencies = [ "tracing-subscriber", ] +[[package]] +name = "openshell-isolation" +version = "0.0.0" +dependencies = [ + "async-trait", + "openshell-core", + "tokio", +] + [[package]] name = "openshell-ocsf" version = "0.0.0" @@ -4103,11 +4116,13 @@ dependencies = [ name = "openshell-sandbox" version = "0.0.0" dependencies = [ + "async-trait", "clap", "futures", "miette", "nix 0.29.0", "openshell-core", + "openshell-isolation", "openshell-ocsf", "openshell-policy", "openshell-supervisor-middleware", @@ -4280,6 +4295,7 @@ name = "openshell-supervisor-network" version = "0.0.0" dependencies = [ "apollo-parser", + "async-trait", "aws-credential-types", "aws-sigv4", "aws-smithy-runtime-api", @@ -4294,6 +4310,7 @@ dependencies = [ "libc", "miette", "openshell-core", + "openshell-isolation", "openshell-ocsf", "openshell-policy", "openshell-router", @@ -4330,6 +4347,7 @@ name = "openshell-supervisor-process" version = "0.0.0" dependencies = [ "anyhow", + "async-trait", "base64 0.22.1", "capctl", "hex", @@ -4338,6 +4356,7 @@ dependencies = [ "miette", "nix 0.29.0", "openshell-core", + "openshell-isolation", "openshell-ocsf", "openshell-policy", "rand 0.10.2", diff --git a/architecture/sandbox.md b/architecture/sandbox.md index a39f699a5..7368ca905 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -49,6 +49,22 @@ OpenShell uses overlapping controls rather than a single sandbox primitive: The supervisor may enrich baseline filesystem allowances for runtime-required paths, such as proxy support files or GPU device paths when a GPU is present. +## Isolation Backend + +The supervisor does not build the boundary inline. It drives an **isolation +backend** (RFC 0012) through one runtime contract (`ensure_boundary` → `bind` +→ `boundary_ready` → `start_agent`), so the same supervisor code runs wherever +the boundary sits. Today only the in-pod backend exists: it builds the boundary +in the same process that operates it, over the existing primitives (network +namespace, proxy, the pre-exec Landlock/seccomp ceiling, procfs identity). + +The contract covers the full isolation envelope. `ensure_boundary` establishes +the network namespace; `start_agent` establishes filesystem (Landlock) and +syscall (seccomp) at process entry; `identity` resolves the connecting binary +from procfs at runtime. The lifecycle handles are unforgeable type-state +tokens, so the order (and thus "no workload runs before the boundary is +ready") holds by construction rather than by convention. + ## Network and Inference All ordinary agent egress is routed through the sandbox proxy. The proxy diff --git a/crates/openshell-core/src/sandbox_env.rs b/crates/openshell-core/src/sandbox_env.rs index 1549258fa..f943dc1ed 100644 --- a/crates/openshell-core/src/sandbox_env.rs +++ b/crates/openshell-core/src/sandbox_env.rs @@ -17,6 +17,17 @@ pub const ENDPOINT: &str = "OPENSHELL_ENDPOINT"; /// Unique identifier of the sandbox being supervised. pub const SANDBOX_ID: &str = "OPENSHELL_SANDBOX_ID"; +/// Serialized RFC 0012 `TopologyDescriptor` for the boundary this sandbox runs +/// behind. +/// +/// Set by the compute driver that provisioned the topology; driver-required, +/// so it overrides any user-supplied value. This is the in-pod delivery +/// transport for provisioning's descriptor handoff to the supervisor role. +/// The descriptor is supervisor-only bootstrap state: the supervisor reads it +/// once at startup and never exposes it to the agent, exec sessions, or any +/// other process inside the boundary. +pub const TOPOLOGY_DESCRIPTOR: &str = "OPENSHELL_TOPOLOGY_DESCRIPTOR"; + /// Filesystem path to the UNIX socket used for the in-sandbox SSH server. pub const SSH_SOCKET_PATH: &str = "OPENSHELL_SSH_SOCKET_PATH"; diff --git a/crates/openshell-driver-docker/Cargo.toml b/crates/openshell-driver-docker/Cargo.toml index 7e1bc069c..f8c647ddf 100644 --- a/crates/openshell-driver-docker/Cargo.toml +++ b/crates/openshell-driver-docker/Cargo.toml @@ -12,6 +12,7 @@ repository.workspace = true [dependencies] openshell-core = { path = "../openshell-core", default-features = false } +openshell-isolation = { path = "../openshell-isolation" } tokio = { workspace = true } tonic = { workspace = true } diff --git a/crates/openshell-driver-docker/README.md b/crates/openshell-driver-docker/README.md index b2e74231b..67a7a2c3a 100644 --- a/crates/openshell-driver-docker/README.md +++ b/crates/openshell-driver-docker/README.md @@ -136,6 +136,7 @@ overwrites security-critical keys: - `OPENSHELL_SANDBOX` - `OPENSHELL_SSH_SOCKET_PATH` - `OPENSHELL_SANDBOX_COMMAND` +- `OPENSHELL_TOPOLOGY_DESCRIPTOR` (RFC 0012 topology descriptor for the boundary) - TLS path variables when HTTPS is enabled Do not allow sandbox images or templates to override these values. diff --git a/crates/openshell-driver-docker/src/lib.rs b/crates/openshell-driver-docker/src/lib.rs index 502f4ae8f..0f506d2c0 100644 --- a/crates/openshell-driver-docker/src/lib.rs +++ b/crates/openshell-driver-docker/src/lib.rs @@ -2223,6 +2223,12 @@ fn build_environment_for_oci_user( openshell_core::sandbox_env::TELEMETRY_ENABLED.to_string(), openshell_core::telemetry::enabled_env_value().to_string(), ); + // RFC 0012: the driver provisions the topology and delivers its descriptor + // to the supervisor through the driver-controlled environment. + environment.insert( + openshell_core::sandbox_env::TOPOLOGY_DESCRIPTOR.to_string(), + openshell_isolation::contract::TopologyDescriptor::in_pod().to_env_value(), + ); // The root supervisor executes namespace helpers during bootstrap; keep // their search path driver-owned even when the template/spec set PATH. environment.insert("PATH".to_string(), SUPERVISOR_PATH.to_string()); diff --git a/crates/openshell-driver-kubernetes/Cargo.toml b/crates/openshell-driver-kubernetes/Cargo.toml index 2c02f864a..3d64b78cc 100644 --- a/crates/openshell-driver-kubernetes/Cargo.toml +++ b/crates/openshell-driver-kubernetes/Cargo.toml @@ -16,6 +16,7 @@ path = "src/main.rs" [dependencies] openshell-core = { path = "../openshell-core", default-features = false } +openshell-isolation = { path = "../openshell-isolation" } openshell-policy = { path = "../openshell-policy" } tokio = { workspace = true } diff --git a/crates/openshell-driver-kubernetes/src/driver.rs b/crates/openshell-driver-kubernetes/src/driver.rs index 2d947b8a2..9fc0509d3 100644 --- a/crates/openshell-driver-kubernetes/src/driver.rs +++ b/crates/openshell-driver-kubernetes/src/driver.rs @@ -2941,6 +2941,13 @@ fn apply_required_env( openshell_core::sandbox_env::TELEMETRY_ENABLED, openshell_core::telemetry::enabled_env_value(), ); + // RFC 0012: the driver provisions the topology and delivers its descriptor + // to the supervisor through the driver-controlled environment. + upsert_env( + env, + openshell_core::sandbox_env::TOPOLOGY_DESCRIPTOR, + &openshell_isolation::contract::TopologyDescriptor::in_pod().to_env_value(), + ); if !ssh_socket_path.is_empty() { upsert_env( env, diff --git a/crates/openshell-driver-podman/Cargo.toml b/crates/openshell-driver-podman/Cargo.toml index e46d2eed8..c8e92d3b3 100644 --- a/crates/openshell-driver-podman/Cargo.toml +++ b/crates/openshell-driver-podman/Cargo.toml @@ -16,6 +16,7 @@ path = "src/main.rs" [dependencies] openshell-core = { path = "../openshell-core", default-features = false } +openshell-isolation = { path = "../openshell-isolation" } tokio = { workspace = true } tonic = { workspace = true, features = ["transport"] } diff --git a/crates/openshell-driver-podman/README.md b/crates/openshell-driver-podman/README.md index 965a295d1..2e70963c6 100644 --- a/crates/openshell-driver-podman/README.md +++ b/crates/openshell-driver-podman/README.md @@ -274,6 +274,7 @@ via sandbox templates: - `OPENSHELL_SSH_SOCKET_PATH` - `OPENSHELL_CONTAINER_IMAGE` - `OPENSHELL_SANDBOX_COMMAND` +- `OPENSHELL_TOPOLOGY_DESCRIPTOR` (RFC 0012 topology descriptor for the boundary) ## Sandbox Lifecycle diff --git a/crates/openshell-driver-podman/src/container.rs b/crates/openshell-driver-podman/src/container.rs index 90ef0fec2..ed05727ea 100644 --- a/crates/openshell-driver-podman/src/container.rs +++ b/crates/openshell-driver-podman/src/container.rs @@ -462,6 +462,12 @@ fn build_env( openshell_core::sandbox_env::TELEMETRY_ENABLED.into(), openshell_core::telemetry::enabled_env_value().into(), ); + // RFC 0012: the driver provisions the topology and delivers its descriptor + // to the supervisor through the driver-controlled environment. + env.insert( + openshell_core::sandbox_env::TOPOLOGY_DESCRIPTOR.into(), + openshell_isolation::contract::TopologyDescriptor::in_pod().to_env_value(), + ); // 3. TLS client cert paths (when mTLS is enabled). These point to // the container-side mount paths where the cert files are diff --git a/crates/openshell-driver-vm/Cargo.toml b/crates/openshell-driver-vm/Cargo.toml index cef3e67f8..80eb60c89 100644 --- a/crates/openshell-driver-vm/Cargo.toml +++ b/crates/openshell-driver-vm/Cargo.toml @@ -20,6 +20,7 @@ path = "src/main.rs" [dependencies] openshell-core = { path = "../openshell-core", default-features = false } +openshell-isolation = { path = "../openshell-isolation" } openshell-policy = { path = "../openshell-policy" } openshell-vfio = { path = "../openshell-vfio" } diff --git a/crates/openshell-driver-vm/src/driver.rs b/crates/openshell-driver-vm/src/driver.rs index 841457f38..00c21d8eb 100644 --- a/crates/openshell-driver-vm/src/driver.rs +++ b/crates/openshell-driver-vm/src/driver.rs @@ -4125,6 +4125,14 @@ fn build_guest_environment( openshell_core::sandbox_env::LOG_LEVEL.to_string(), openshell_core::driver_utils::sandbox_log_level(sandbox, &config.log_level), ); + // RFC 0012: the driver provisions the topology and delivers its descriptor + // to the supervisor through the driver-controlled environment. The VM + // supervisor is co-located with the agent inside the guest, so it drives + // the in-pod backend. + environment.insert( + openshell_core::sandbox_env::TOPOLOGY_DESCRIPTOR.to_string(), + openshell_isolation::contract::TopologyDescriptor::in_pod().to_env_value(), + ); if config.requires_tls_materials() { environment.insert( openshell_core::sandbox_env::TLS_CA.to_string(), diff --git a/crates/openshell-isolation/Cargo.toml b/crates/openshell-isolation/Cargo.toml new file mode 100644 index 000000000..be703c52c --- /dev/null +++ b/crates/openshell-isolation/Cargo.toml @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "openshell-isolation" +description = "OpenShell Isolation Backend runtime contract (RFC 0012)" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true + +[dependencies] +openshell-core = { path = "../openshell-core", default-features = false } +async-trait = "0.1" +tokio = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true } + +[lints] +workspace = true diff --git a/crates/openshell-isolation/src/contract.rs b/crates/openshell-isolation/src/contract.rs new file mode 100644 index 000000000..a7a59c57e --- /dev/null +++ b/crates/openshell-isolation/src/contract.rs @@ -0,0 +1,675 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Runtime-selectable Isolation Backend contract (RFC 0012). +//! +//! This module is the object-safe, runtime-selectable contract the supervisor +//! role drives. A backend registers an [`IsolationBackendFactory`] under a +//! `backend_id`; the supervisor resolves it from a [`BackendRegistry`] against +//! the admitted backend id and advances the boundary through a fixed chain of +//! boxed states: +//! +//! ```text +//! attach topology + sandbox context -> Bound -> confirm -> Ready +//! -> start_agent -> Running +//! ``` +//! +//! Each transition consumes the prior state by value (`self: Box`), and no +//! state type has a public constructor, so a stage cannot be skipped or +//! replayed. The supervisor holds no `match`/downcast on concrete backends: the +//! registry is the only lookup by `backend_id`, and everything past it is a +//! `Box` / `Arc`. +//! +//! `attach` is atomic from the caller's perspective: it returns `Bound` or fails +//! closed, and it never binds a resource that is already bound to an active +//! boundary. Binary identity travels on every [`MediatedConnection`], resolved +//! by the backend for that exact connection; an unresolved identity denies the +//! connection and never authorizes anything. +//! +//! The in-pod backend (`openshell-sandbox`) and the supervisor drive this +//! contract directly; two mock factories and the conformance tests below +//! exercise it against a second, heterogeneous backend. + +use std::collections::HashMap; +use std::fmt; +use std::net::IpAddr; +use std::path::PathBuf; +use std::sync::Arc; + +use async_trait::async_trait; +use tokio::io::{AsyncRead, AsyncWrite}; + +pub use openshell_core::policy::SandboxPolicy; + +/// The Isolation Backend contract version. The descriptor and the resolved +/// factory must both equal the supervisor-supported version exactly. +pub const CONTRACT_VERSION: u32 = 1; + +/// The well-known backend id of the co-located in-pod backend. +/// +/// Lives in the contract crate because compute drivers create the +/// [`TopologyDescriptor`] naming it; the implementation lives in +/// `openshell-sandbox`. +pub const IN_POD_BACKEND_ID: &str = "in-pod"; + +// ============================================================================ +// Errors +// ============================================================================ + +/// Classified failures at the common contract boundary. +/// +/// Only [`BackendError::is_retryable`] cases may be retried by the supervisor, +/// and a retry must reuse the same backend and boundary; no error downgrades +/// isolation. +#[derive(Debug)] +pub enum BackendError { + /// Descriptor missing, malformed, unsupported, or mismatched against admission. + Descriptor(String), + /// No factory registered for the resolved `backend_id`. + NotRegistered(String), + /// Authenticated attachment rejection (incompatible or already-bound resource). + Denied(String), + /// Boundary temporarily unavailable. The only retryable class. + Unavailable(String), + /// Attachment-phase failure (establishment or mediation bring-up). + Attach(String), + /// Readiness confirmation failed (do not start workload code). + Confirm(String), + /// Process start or exec failure. + Process(String), + /// Abnormal boundary or workload loss, or an operation against an inactive + /// boundary. + Terminated(String), +} + +/// Coarse, machine-readable classification of a [`BackendError`] for supervisor +/// status mapping. The error's variant and message carry the structured context +/// (which operation failed). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BackendErrorKind { + /// Descriptor, version, or backend mismatch. + Invalid, + /// Authenticated attachment rejection. + Denied, + /// Transient inability to serve an operation (the only retryable kind). + Unavailable, + /// Attachment, confirmation, start, or runtime operation failure. + Failed, + /// Abnormal boundary/workload loss, or an operation against an inactive + /// boundary. + Terminated, +} + +impl BackendError { + /// Whether the supervisor may retry, reusing the same backend and boundary. + #[must_use] + pub fn is_retryable(&self) -> bool { + matches!(self, Self::Unavailable(_)) + } + + /// The machine-readable kind for this error. + #[must_use] + pub fn kind(&self) -> BackendErrorKind { + match self { + Self::Descriptor(_) | Self::NotRegistered(_) => BackendErrorKind::Invalid, + Self::Denied(_) => BackendErrorKind::Denied, + Self::Unavailable(_) => BackendErrorKind::Unavailable, + Self::Attach(_) | Self::Confirm(_) | Self::Process(_) => BackendErrorKind::Failed, + Self::Terminated(_) => BackendErrorKind::Terminated, + } + } +} + +impl fmt::Display for BackendError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Descriptor(m) => write!(f, "descriptor error: {m}"), + Self::NotRegistered(m) => write!(f, "backend not registered: {m}"), + Self::Denied(m) => write!(f, "attachment denied: {m}"), + Self::Unavailable(m) => write!(f, "boundary unavailable: {m}"), + Self::Attach(m) => write!(f, "attachment failed: {m}"), + Self::Confirm(m) => write!(f, "confirmation failed: {m}"), + Self::Process(m) => write!(f, "process error: {m}"), + Self::Terminated(m) => write!(f, "boundary terminated: {m}"), + } + } +} + +impl std::error::Error for BackendError {} + +/// Why an identity resolution failed. Resolution failure fails closed: the +/// mediation service denies and audits the connection; it never authorizes. +#[derive(Debug)] +pub enum ResolveError { + /// No process owns the connection (stale or unknown attribution). + NotFound, + /// Resolution attempted but could not produce trustworthy identity. + Failed(String), +} + +impl fmt::Display for ResolveError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::NotFound => write!(f, "connection owner not found"), + Self::Failed(m) => write!(f, "identity resolution failed: {m}"), + } + } +} + +impl std::error::Error for ResolveError {} + +// ============================================================================ +// Descriptor and registry +// ============================================================================ + +/// The common topology descriptor envelope. +/// +/// The compute driver supplies one for every provisioned topology, including +/// resources prepared before sandbox assignment. The opaque payload identifies, +/// or gives the backend enough information to resolve, the exact +/// driver-provisioned resource; its protection is backend-specific. +#[derive(Debug, Clone)] +pub struct TopologyDescriptor { + /// The Isolation Backend contract version this descriptor targets. + pub contract_version: u32, + /// The backend the supervisor must instantiate. + pub backend_id: String, + /// Backend-specific attachment data. + pub payload: Vec, +} + +impl TopologyDescriptor { + /// The descriptor for the co-located in-pod backend: the supervisor + /// process is the provisioned resource, so the payload is empty. + #[must_use] + pub fn in_pod() -> Self { + Self { + contract_version: CONTRACT_VERSION, + backend_id: IN_POD_BACKEND_ID.to_string(), + payload: Vec::new(), + } + } + + /// Serialize for the driver-controlled environment transport + /// (`OPENSHELL_TOPOLOGY_DESCRIPTOR`): `::`. + /// Other transports deliver the same envelope with equivalent integrity. + #[must_use] + pub fn to_env_value(&self) -> String { + format!( + "{}:{}:{}", + self.contract_version, + self.backend_id, + hex_encode(&self.payload) + ) + } + + /// Parse the environment-transport serialization produced by + /// [`Self::to_env_value`]. Missing or malformed input fails closed. + /// + /// # Errors + /// + /// Returns [`BackendError::Descriptor`] for a malformed value: missing + /// fields, a non-numeric version, an empty backend id, or a payload that + /// is not valid hex. + pub fn from_env_value(value: &str) -> Result { + let mut parts = value.splitn(3, ':'); + let (Some(version), Some(backend_id), Some(payload_hex)) = + (parts.next(), parts.next(), parts.next()) + else { + return Err(BackendError::Descriptor( + "topology descriptor must be ::".to_string(), + )); + }; + let contract_version: u32 = version.parse().map_err(|_| { + BackendError::Descriptor(format!( + "topology descriptor version {version:?} is not a number" + )) + })?; + if backend_id.is_empty() { + return Err(BackendError::Descriptor( + "topology descriptor backend id is empty".to_string(), + )); + } + let payload = hex_decode(payload_hex).ok_or_else(|| { + BackendError::Descriptor("topology descriptor payload is not valid hex".to_string()) + })?; + Ok(Self { + contract_version, + backend_id: backend_id.to_string(), + payload, + }) + } +} + +fn hex_encode(bytes: &[u8]) -> String { + use std::fmt::Write as _; + let mut out = String::with_capacity(bytes.len() * 2); + for byte in bytes { + let _ = write!(out, "{byte:02x}"); + } + out +} + +fn hex_decode(hex: &str) -> Option> { + if !hex.is_ascii() || hex.len() % 2 != 0 { + return None; + } + (0..hex.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).ok()) + .collect() +} + +/// A descriptor whose common envelope has passed registry verification. +/// +/// Minted only by [`BackendRegistry::resolve`]; no public constructor, so an +/// unverified descriptor cannot reach a factory. The type does not imply that +/// the opaque payload has been validated: the backend validates the payload and +/// atomically binds it to the sandbox context during `attach`. +pub struct VerifiedTopologyDescriptor { + descriptor: TopologyDescriptor, +} + +impl VerifiedTopologyDescriptor { + /// The verified backend id. + #[must_use] + pub fn backend_id(&self) -> &str { + &self.descriptor.backend_id + } + /// The backend-specific payload (validated by the backend at `attach`). + #[must_use] + pub fn payload(&self) -> &[u8] { + &self.descriptor.payload + } + /// The contract version. + #[must_use] + pub fn contract_version(&self) -> u32 { + self.descriptor.contract_version + } +} + +/// The trusted sandbox context, constructed by trusted common code after the +/// control plane assigns the resource to the admitted sandbox. +/// +/// Carries the create-time policy baseline. Authorized revisions to +/// mediation-evaluated policy continue through the mediation service's +/// policy-update path after `Running`, without a backend lifecycle transition; +/// a revision that requires changing standing enforcement or launch-time +/// controls is rejected. +pub struct SandboxContext { + /// Which sandbox this is. + pub sandbox_id: String, + /// The create-time policy baseline across all four dimensions. + pub policy: SandboxPolicy, + /// The admitted agent workload. + pub agent: AgentSpec, +} + +/// The agent workload to run inside the boundary. +pub use crate::AgentSpec; + +/// Maps `backend_id` to its factory. The only lookup-by-id in the system; the +/// supervisor lifecycle never branches on a concrete backend, and resolution +/// never falls back to another backend. +#[derive(Default)] +pub struct BackendRegistry { + factories: HashMap>, +} + +impl BackendRegistry { + /// An empty registry. + #[must_use] + pub fn new() -> Self { + Self { + factories: HashMap::new(), + } + } + + /// Register a factory. Rejects a duplicate `backend_id` and a factory that + /// does not speak the supervisor-supported contract version exactly. + /// + /// # Errors + /// + /// Returns [`BackendError::Descriptor`] for a duplicate `backend_id` or a + /// contract-version mismatch. + pub fn register( + &mut self, + factory: Arc, + ) -> Result<(), BackendError> { + let id = factory.backend_id().to_string(); + if self.factories.contains_key(&id) { + return Err(BackendError::Descriptor(format!( + "duplicate backend id {id:?}" + ))); + } + if factory.contract_version() != CONTRACT_VERSION { + return Err(BackendError::Descriptor(format!( + "backend {id:?} targets contract version {}, supervisor speaks {CONTRACT_VERSION}", + factory.contract_version() + ))); + } + self.factories.insert(id, factory); + Ok(()) + } + + /// Verify the descriptor's common envelope against the admitted backend id + /// and resolve its factory. Fails closed and never falls back: + /// + /// - the descriptor's contract version must equal [`CONTRACT_VERSION`]; + /// - the descriptor's `backend_id` must equal the admitted id; + /// - a factory must be registered for that id, agreeing on its id; and + /// - the factory's contract version must equal [`CONTRACT_VERSION`] exactly. + /// + /// # Errors + /// + /// Returns [`BackendError::Descriptor`] for a version or admission + /// mismatch, and [`BackendError::NotRegistered`] when no factory is + /// registered for the admitted id. + pub fn resolve( + &self, + descriptor: TopologyDescriptor, + admitted_backend_id: &str, + ) -> Result<(Arc, VerifiedTopologyDescriptor), BackendError> { + if descriptor.contract_version != CONTRACT_VERSION { + return Err(BackendError::Descriptor(format!( + "descriptor contract version {} unsupported (expected {CONTRACT_VERSION})", + descriptor.contract_version + ))); + } + if descriptor.backend_id != admitted_backend_id { + return Err(BackendError::Descriptor(format!( + "descriptor backend {:?} does not match admitted backend {admitted_backend_id:?}", + descriptor.backend_id + ))); + } + let factory = self + .factories + .get(&descriptor.backend_id) + .ok_or_else(|| BackendError::NotRegistered(descriptor.backend_id.clone()))? + .clone(); + // Defense in depth: the resolved factory must agree on its id and version. + if factory.backend_id() != descriptor.backend_id { + return Err(BackendError::Descriptor(format!( + "registry returned backend {:?} for id {:?}", + factory.backend_id(), + descriptor.backend_id + ))); + } + if factory.contract_version() != CONTRACT_VERSION { + return Err(BackendError::Descriptor(format!( + "backend {:?} speaks contract version {}, supervisor requires {CONTRACT_VERSION}", + descriptor.backend_id, + factory.contract_version() + ))); + } + Ok((factory, VerifiedTopologyDescriptor { descriptor })) + } +} + +/// Builds and attaches a concrete backend. Registered once per `backend_id`. +#[async_trait] +pub trait IsolationBackendFactory: Send + Sync { + /// The backend this factory builds. + fn backend_id(&self) -> &'static str; + + /// The Isolation Backend contract version this factory speaks. Matched + /// exactly against [`CONTRACT_VERSION`]; there is no capability negotiation. + fn contract_version(&self) -> u32; + + /// Validate the opaque payload and atomically bind it to the trusted + /// sandbox context: returns `Bound` or fails closed. Never binds a resource + /// that is already bound to an active boundary. + async fn attach( + &self, + descriptor: VerifiedTopologyDescriptor, + sandbox: SandboxContext, + ) -> Result, BackendError>; +} + +// ============================================================================ +// Lifecycle states +// ============================================================================ + +/// Bound: the topology descriptor and trusted sandbox context are bound to the +/// same resource, and the mediation ingress is available. No untrusted workload +/// code is running. +#[async_trait] +pub trait BoundBoundary: Send { + /// The mediation service's backend-neutral source of workload connections. + /// Retained by the supervisor before consuming `Bound`. + fn mediation_ingress(&self) -> Arc; + + /// Confirm standing enforcement. How a backend establishes confidence is + /// private to that backend; confirmation fails closed. + async fn confirm(self: Box) -> Result, BackendError>; +} + +/// Ready: mediation is initialized, standing enforcement is confirmed, and the +/// backend is prepared to ensure the admitted launch-time controls are in force +/// before untrusted execution. Only agent activation is possible from here. +#[async_trait] +pub trait ReadyBoundary: Send { + /// Make the admitted agent runnable behind the boundary and return its + /// handle. `start_agent` is the sole operation that may make the admitted + /// agent runnable, and it fails closed if any `Ready` condition no longer + /// holds. Whether the backend creates the agent process or releases a held, + /// driver-provisioned execution object is backend-specific; every + /// applicable launch-time control is in force before the first untrusted + /// instruction. + async fn start_agent(self: Box) -> Result, BackendError>; +} + +/// Running: the agent is runnable behind the boundary and the returned agent +/// handle owns the complete workload tree. Exec and forwarding are available. +/// All interface accessors return owned `Arc`s so a consumer can retain them +/// past any later state consumption. +pub trait RunningBoundary: Send + Sync { + /// The agent process handle (owns the complete workload tree). + fn agent(&self) -> Arc; + /// The in-boundary exec interface. + fn exec(&self) -> Arc; + /// The loopback port-forward interface. + fn port_forward(&self) -> Arc; +} + +// ============================================================================ +// Process and exec +// ============================================================================ + +/// Placement-neutral terminal status of a boundary process. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BoundaryExitStatus { + /// Exited with a code. + Exited(i32), + /// Killed by a signal. + Signaled(i32), +} + +/// Placement-neutral signal to deliver to a boundary process. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BoundarySignal { + /// Graceful terminate. + Term, + /// Forceful kill. + Kill, + /// Interrupt. + Int, + /// Hangup. + Hup, +} + +/// A process running inside the boundary, owned by its boundary state. `wait` +/// returns one stable status however many times it is called; a host PID, if +/// useful, is diagnostics-only and never the handle. +#[async_trait] +pub trait BoundaryProcess: Send + Sync { + /// Await terminal status (stable across repeated calls). + async fn wait(&self) -> Result; + /// Deliver a signal to the process or its group. + async fn signal(&self, signal: BoundarySignal) -> Result<(), BackendError>; + /// Terminate the process and its complete descendant tree. + async fn terminate(&self) -> Result<(), BackendError>; + /// Optional host PID, for diagnostics only. + fn diagnostic_pid(&self) -> Option { + None + } +} + +/// A boxed async writer into a boundary process's stdin. +pub type BoundaryInput = Box; +/// A boxed async reader from a boundary process's stdout or stderr. +pub type BoundaryOutput = Box; + +/// A PTY attached to an exec session. +#[async_trait] +pub trait BoundaryTerminal: Send + Sync { + /// Resize the terminal. + async fn resize(&self, cols: u16, rows: u16) -> Result<(), BackendError>; +} + +/// An owned exec session: the process handle plus its stdio and optional PTY. +/// Owning the process keeps it alive after `exec` returns. +pub struct ExecSession { + /// The spawned process. + pub process: Arc, + /// Stdin writer, if not a PTY-merged stream. + pub stdin: Option, + /// Stdout reader. + pub stdout: BoundaryOutput, + /// Stderr reader, distinct from stdout for non-PTY exec. + pub stderr: Option, + /// PTY control, present when a terminal was requested. + pub terminal: Option>, +} + +/// What to run inside the boundary via [`BoundaryExec`]. +#[derive(Debug, Clone)] +pub struct ExecSpec { + /// Program to run. + pub program: String, + /// Program arguments. + pub args: Vec, + /// Extra environment over the boundary's base. + pub env: Vec<(String, String)>, + /// Working directory, if any. + pub workdir: Option, + /// Whether to allocate a PTY. + pub pty: bool, +} + +/// In-boundary process entry, consumed by the SSH server and supervisor session. +/// Like `start_agent`, every exec ensures the applicable launch-time controls +/// are in force before the new process executes its first untrusted instruction +/// and preserves the provisioned execution environment. +#[async_trait] +pub trait BoundaryExec: Send + Sync { + /// Spawn `spec` inside the boundary, returning an owned session. + async fn exec(&self, spec: ExecSpec) -> Result; +} + +// ============================================================================ +// Port forward +// ============================================================================ + +/// A loopback-only target inside the boundary, validated at construction. +#[derive(Debug, Clone)] +pub struct LoopbackTarget { + host: IpAddr, + port: u16, +} + +impl LoopbackTarget { + /// Build a loopback target, rejecting any non-loopback host. + /// + /// # Errors + /// + /// Returns [`BackendError::Process`] when `host` is not a loopback address. + pub fn new(host: IpAddr, port: u16) -> Result { + if !host.is_loopback() { + return Err(BackendError::Process(format!( + "port-forward target {host} is not loopback" + ))); + } + Ok(Self { host, port }) + } + /// The loopback host. + #[must_use] + pub fn host(&self) -> IpAddr { + self.host + } + /// The target port. + #[must_use] + pub fn port(&self) -> u16 { + self.port + } +} + +/// A bidirectional byte stream into the boundary. +pub trait DuplexStream: AsyncRead + AsyncWrite + Send + Unpin {} +impl DuplexStream for T {} + +/// An open connection into a boundary loopback target. +pub type BoundaryConn = Box; + +/// Loopback port-forward, consumed by the SSH server and supervisor session. +#[async_trait] +pub trait BoundaryPortForward: Send + Sync { + /// Connect to `target` inside the boundary. + async fn connect(&self, target: LoopbackTarget) -> Result; +} + +// ============================================================================ +// Mediation and binary identity +// ============================================================================ + +/// Executable identity for one accepted connection, resolved by the backend and +/// delivered on [`MediatedConnection`] before the mediation service evaluates +/// policy. +/// +/// A missing digest is `None`, never an empty value; policy that requires an +/// unavailable identity field cannot authorize the connection. How a backend +/// resolves identity is private to that backend; the shape and the fail-closed +/// semantics do not change. +#[derive(Debug, Clone)] +pub struct BinaryIdentity { + /// Absolute path of the connecting binary. + pub binary_path: PathBuf, + /// SHA-256 of the connecting binary, hex-encoded. `None` when unavailable; + /// never an empty string. + pub binary_sha256: Option, + /// Ancestor process binaries, nearest first. + pub ancestors: Vec, + /// Absolute script/interpreter paths drawn from the process cmdlines. + /// Diagnostic context; never authorizes. + pub cmdline_paths: Vec, +} + +/// A workload connection delivered to the mediation service, carrying the +/// identity-resolution result for that exact connection. +/// +/// An `Err` identity denies the connection and is audited; it never authorizes +/// anything. +pub struct MediatedConnection { + /// The workload connection stream. + pub stream: BoundaryConn, + /// Executable identity, resolved by the backend for this connection. + pub binary_identity: Result, +} + +/// A logical per-boundary stream of workload connections, consumed by the +/// mediation service wherever that service runs. +/// +/// It may wrap a dedicated listener or a demultiplexed view over shared +/// transport; how it reaches a co-located proxy, a sidecar, or a shared +/// mediation service is backend-private. The backend authoritatively attributes +/// every returned connection to its active boundary without relying solely on a +/// workload-provided identifier. An `Err` from `accept` means the ingress +/// itself is unusable and fails the boundary closed. +#[async_trait] +pub trait MediationIngress: Send + Sync { + /// Await the next mediated workload connection. + async fn accept(&self) -> Result; +} + +#[cfg(test)] +mod tests; diff --git a/crates/openshell-isolation/src/contract/tests.rs b/crates/openshell-isolation/src/contract/tests.rs new file mode 100644 index 000000000..bc6d14469 --- /dev/null +++ b/crates/openshell-isolation/src/contract/tests.rs @@ -0,0 +1,633 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Conformance harness for the runtime-selectable contract. +//! +//! Two materially different mock factories (`Primary`, `Secondary`) with +//! distinct concrete state structs (each generic over a marker, so each kind +//! monomorphizes to its own types) prove the registry holds heterogeneous +//! factories behind `dyn` with no enum over concrete state, and that one driver +//! runs both unchanged. + +use std::marker::PhantomData; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; + +use super::*; + +// --------------------------------------------------------------------------- +// Marker kinds: two materially different backends. +// --------------------------------------------------------------------------- + +trait MockKind: Send + Sync + 'static { + const BACKEND_ID: &'static str; + /// Whether this backend can produce a binary digest (a heterogeneity axis: + /// one backend resolves a full identity, the other resolves path-only). + const HAS_DIGEST: bool; +} + +struct Primary; +impl MockKind for Primary { + const BACKEND_ID: &'static str = "mock-primary"; + const HAS_DIGEST: bool = true; +} + +struct Secondary; +impl MockKind for Secondary { + const BACKEND_ID: &'static str = "mock-secondary"; + const HAS_DIGEST: bool = false; +} + +// --------------------------------------------------------------------------- +// Runtime interfaces (shared across kinds where behavior is identical). +// --------------------------------------------------------------------------- + +struct MockProcess { + status: BoundaryExitStatus, + alive: AtomicBool, + signals: Mutex>, +} + +impl MockProcess { + fn new() -> Arc { + Arc::new(Self { + status: BoundaryExitStatus::Exited(0), + alive: AtomicBool::new(true), + signals: Mutex::new(Vec::new()), + }) + } +} + +#[async_trait] +impl BoundaryProcess for MockProcess { + async fn wait(&self) -> Result { + // Stable across repeated calls. + Ok(self.status) + } + async fn signal(&self, signal: BoundarySignal) -> Result<(), BackendError> { + self.signals.lock().unwrap().push(signal); + Ok(()) + } + async fn terminate(&self) -> Result<(), BackendError> { + self.alive.store(false, Ordering::SeqCst); + Ok(()) + } + fn diagnostic_pid(&self) -> Option { + Some(4242) + } +} + +/// Mediation ingress: hands the mediation service a connection carrying its +/// per-connection identity-resolution result. +struct MockIngress(PhantomData); + +#[async_trait] +impl MediationIngress for MockIngress { + async fn accept(&self) -> Result { + let (near, _far) = tokio::io::duplex(64); + Ok(MediatedConnection { + stream: Box::new(near), + binary_identity: Ok(BinaryIdentity { + binary_path: PathBuf::from("/usr/bin/agent"), + binary_sha256: K::HAS_DIGEST.then(|| "deadbeef".to_string()), + ancestors: vec![], + cmdline_paths: vec![], + }), + }) + } +} + +/// An ingress whose backend cannot attribute the connection: the connection is +/// still delivered, carrying `Err`, so the mediation service denies and audits +/// it. It never authorizes anything. +struct UnattributedIngress; + +#[async_trait] +impl MediationIngress for UnattributedIngress { + async fn accept(&self) -> Result { + let (near, _far) = tokio::io::duplex(64); + Ok(MediatedConnection { + stream: Box::new(near), + binary_identity: Err(ResolveError::Failed("hash unavailable".to_string())), + }) + } +} + +struct MockExec; + +#[async_trait] +impl BoundaryExec for MockExec { + async fn exec(&self, _spec: ExecSpec) -> Result { + let (_near, far) = tokio::io::duplex(64); + let (out_r, _out_w) = tokio::io::duplex(64); + let (err_r, _err_w) = tokio::io::duplex(64); + Ok(ExecSession { + process: MockProcess::new(), + stdin: Some(Box::new(far)), + stdout: Box::new(out_r), + stderr: Some(Box::new(err_r)), + terminal: None, + }) + } +} + +struct MockPortForward; + +#[async_trait] +impl BoundaryPortForward for MockPortForward { + async fn connect(&self, _target: LoopbackTarget) -> Result { + let (near, far) = tokio::io::duplex(64); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut far = far; + let mut buf = [0u8; 4]; + if far.read_exact(&mut buf).await.is_ok() { + let _ = far.write_all(&buf).await; + } + }); + Ok(Box::new(near)) + } +} + +// --------------------------------------------------------------------------- +// Boxed lifecycle states (distinct concrete struct per kind). +// --------------------------------------------------------------------------- + +struct MockBound { + ingress: Arc>, +} +struct MockReady { + _k: PhantomData, +} +struct MockRunning { + process: Arc, + exec: Arc, + port_forward: Arc, + _k: PhantomData, +} + +#[async_trait] +impl BoundBoundary for MockBound { + fn mediation_ingress(&self) -> Arc { + self.ingress.clone() + } + async fn confirm(self: Box) -> Result, BackendError> { + Ok(Box::new(MockReady:: { _k: PhantomData })) + } +} + +#[async_trait] +impl ReadyBoundary for MockReady { + async fn start_agent(self: Box) -> Result, BackendError> { + Ok(Box::new(MockRunning:: { + process: MockProcess::new(), + exec: Arc::new(MockExec), + port_forward: Arc::new(MockPortForward), + _k: PhantomData, + })) + } +} + +impl RunningBoundary for MockRunning { + fn agent(&self) -> Arc { + self.process.clone() + } + fn exec(&self) -> Arc { + self.exec.clone() + } + fn port_forward(&self) -> Arc { + self.port_forward.clone() + } +} + +/// One factory per boundary resource: `attach` is atomic and never binds a +/// resource that is already bound to an active boundary, so a second attach +/// against the same mock resource is `Denied`. +struct MockFactory { + attached: AtomicBool, + _k: PhantomData, +} + +impl MockFactory { + fn new() -> Self { + Self { + attached: AtomicBool::new(false), + _k: PhantomData, + } + } +} + +#[async_trait] +impl IsolationBackendFactory for MockFactory { + fn backend_id(&self) -> &'static str { + K::BACKEND_ID + } + fn contract_version(&self) -> u32 { + CONTRACT_VERSION + } + async fn attach( + &self, + descriptor: VerifiedTopologyDescriptor, + sandbox: SandboxContext, + ) -> Result, BackendError> { + assert_eq!(descriptor.backend_id(), K::BACKEND_ID); + assert!(!sandbox.sandbox_id.is_empty()); + if self.attached.swap(true, Ordering::SeqCst) { + return Err(BackendError::Denied( + "resource is already bound to an active boundary".to_string(), + )); + } + Ok(Box::new(MockBound:: { + ingress: Arc::new(MockIngress(PhantomData)), + })) + } +} + +/// A factory that speaks the wrong contract version; registration must reject it. +struct WrongVersionFactory; + +#[async_trait] +impl IsolationBackendFactory for WrongVersionFactory { + fn backend_id(&self) -> &'static str { + "mock-wrong-version" + } + fn contract_version(&self) -> u32 { + CONTRACT_VERSION + 1 + } + async fn attach( + &self, + _descriptor: VerifiedTopologyDescriptor, + _sandbox: SandboxContext, + ) -> Result, BackendError> { + unreachable!("must never be resolved") + } +} + +// --------------------------------------------------------------------------- +// Helpers. +// --------------------------------------------------------------------------- + +fn registry() -> BackendRegistry { + let mut reg = BackendRegistry::new(); + reg.register(Arc::new(MockFactory::::new())) + .expect("register primary"); + reg.register(Arc::new(MockFactory::::new())) + .expect("register secondary"); + reg +} + +fn descriptor(backend_id: &str) -> TopologyDescriptor { + TopologyDescriptor { + contract_version: CONTRACT_VERSION, + backend_id: backend_id.to_string(), + payload: vec![], + } +} + +fn sandbox_ctx() -> SandboxContext { + SandboxContext { + sandbox_id: "sb-1".to_string(), + policy: 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(), + }, + agent: AgentSpec { + program: "/bin/true".to_string(), + args: vec![], + workdir: None, + timeout_secs: 0, + interactive: false, + }, + } +} + +/// The backend-independent supervisor sequence. Identical for every backend: +/// this is the proof that adding a backend needs no supervisor lifecycle change. +async fn drive( + reg: &BackendRegistry, + descriptor: TopologyDescriptor, + admitted: &str, +) -> Result, BackendError> { + let (factory, verified) = reg.resolve(descriptor, admitted)?; + let bound = factory.attach(verified, sandbox_ctx()).await?; + // The mediation ingress is retained before consuming `Bound` and stays + // usable across the confirm/start transitions. + let _ingress = bound.mediation_ingress(); + let ready = bound.confirm().await?; + ready.start_agent().await +} + +// --------------------------------------------------------------------------- +// Registry and descriptor. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn registry_selects_correct_backend() { + let reg = registry(); + let (f, _v) = reg + .resolve(descriptor("mock-secondary"), "mock-secondary") + .expect("resolve"); + assert_eq!(f.backend_id(), "mock-secondary"); +} + +#[test] +fn registry_rejects_duplicate_registration() { + let mut reg = BackendRegistry::new(); + reg.register(Arc::new(MockFactory::::new())) + .expect("first"); + let err = reg + .register(Arc::new(MockFactory::::new())) + .expect_err("duplicate must fail"); + assert!(matches!(err, BackendError::Descriptor(_))); +} + +#[test] +fn registry_rejects_wrong_factory_version() { + let mut reg = BackendRegistry::new(); + let err = reg + .register(Arc::new(WrongVersionFactory)) + .expect_err("wrong version must fail"); + assert_eq!(err.kind(), BackendErrorKind::Invalid); +} + +#[test] +fn registry_rejects_unknown_backend() { + let reg = registry(); + let err = reg + .resolve(descriptor("nope"), "nope") + .map(|_| ()) + .expect_err("unknown must fail"); + assert!(matches!(err, BackendError::NotRegistered(_))); +} + +#[test] +fn registry_rejects_descriptor_admission_mismatch_without_fallback() { + let reg = registry(); + // Descriptor names primary, admission says secondary: must fail, and must + // not silently fall back to either backend. + let err = reg + .resolve(descriptor("mock-primary"), "mock-secondary") + .map(|_| ()) + .expect_err("mismatch must fail"); + assert!(matches!(err, BackendError::Descriptor(_))); +} + +#[test] +fn registry_rejects_unsupported_version() { + let reg = registry(); + let mut d = descriptor("mock-primary"); + d.contract_version = CONTRACT_VERSION + 1; + let err = reg + .resolve(d, "mock-primary") + .map(|_| ()) + .expect_err("bad version must fail"); + assert!(matches!(err, BackendError::Descriptor(_))); +} + +// --------------------------------------------------------------------------- +// Lifecycle: one driver, two heterogeneous backends, no consumer change. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn one_driver_runs_both_backends() { + let reg = registry(); + // The exact same driver code runs a backend with distinct concrete state + // structs; the registry holds them behind `dyn`, no enum. + let primary = drive(®, descriptor("mock-primary"), "mock-primary") + .await + .expect("primary lifecycle"); + let secondary = drive(®, descriptor("mock-secondary"), "mock-secondary") + .await + .expect("secondary lifecycle"); + + // Both expose a usable agent process handle past start_agent. + assert_eq!( + primary.agent().wait().await.expect("wait"), + BoundaryExitStatus::Exited(0) + ); + assert_eq!( + secondary.agent().wait().await.expect("wait"), + BoundaryExitStatus::Exited(0) + ); +} + +#[tokio::test] +async fn attach_never_binds_an_already_bound_resource() { + let reg = registry(); + // First attach binds the mock resource. + drive(®, descriptor("mock-primary"), "mock-primary") + .await + .expect("first lifecycle"); + // A second attach against the same active boundary must be denied, not + // silently create a second binding. + let err = drive(®, descriptor("mock-primary"), "mock-primary") + .await + .map(|_| ()) + .expect_err("second attach must fail"); + assert_eq!(err.kind(), BackendErrorKind::Denied); +} + +#[tokio::test] +async fn runtime_interfaces_survive_lifecycle_consumption() { + let reg = registry(); + let (factory, verified) = reg + .resolve(descriptor("mock-primary"), "mock-primary") + .expect("resolve"); + let bound = factory + .attach(verified, sandbox_ctx()) + .await + .expect("attach"); + + // Retain the ingress at Bound, then consume the bound state with confirm. + // The retained Arc must remain usable afterward. + let ingress = bound.mediation_ingress(); + let ready = bound.confirm().await.expect("confirm"); + let _running = ready.start_agent().await.expect("start"); + + let conn = ingress.accept().await.expect("accept after consumption"); + let identity = conn.binary_identity.expect("identity resolves"); + assert_eq!(identity.binary_path, PathBuf::from("/usr/bin/agent")); +} + +// --------------------------------------------------------------------------- +// Process and I/O. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn agent_process_survives_and_wait_is_stable() { + let reg = registry(); + let running = drive(®, descriptor("mock-primary"), "mock-primary") + .await + .expect("lifecycle"); + let agent = running.agent(); + // Survives start_agent returning; wait is stable across repeated calls. + assert_eq!( + agent.wait().await.expect("wait 1"), + BoundaryExitStatus::Exited(0) + ); + assert_eq!( + agent.wait().await.expect("wait 2"), + BoundaryExitStatus::Exited(0) + ); + agent.signal(BoundarySignal::Term).await.expect("signal"); +} + +#[tokio::test] +async fn exec_session_owns_its_process_and_streams() { + let reg = registry(); + let running = drive(®, descriptor("mock-primary"), "mock-primary") + .await + .expect("lifecycle"); + let session = running + .exec() + .exec(ExecSpec { + program: "/bin/sh".to_string(), + args: vec!["-c".to_string(), "true".to_string()], + env: vec![], + workdir: None, + pty: false, + }) + .await + .expect("exec"); + // The exec'd process survives `exec` returning, and stdout/stderr are distinct. + assert!(session.stderr.is_some()); + assert!(session.stdin.is_some()); + assert_eq!( + session.process.wait().await.expect("exec wait"), + BoundaryExitStatus::Exited(0) + ); +} + +#[tokio::test] +async fn port_forward_rejects_non_loopback() { + let target = LoopbackTarget::new("8.8.8.8".parse().unwrap(), 53); + assert!(target.is_err()); + let loopback = LoopbackTarget::new("127.0.0.1".parse().unwrap(), 8080).expect("loopback ok"); + assert_eq!(loopback.port(), 8080); +} + +// --------------------------------------------------------------------------- +// Mediation and binary identity. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn mediated_connection_carries_identity_for_that_connection() { + let reg = registry(); + let (factory, verified) = reg + .resolve(descriptor("mock-primary"), "mock-primary") + .expect("resolve"); + let bound = factory + .attach(verified, sandbox_ctx()) + .await + .expect("attach"); + let conn = bound.mediation_ingress().accept().await.expect("accept"); + let identity = conn.binary_identity.expect("identity resolves"); + assert_eq!(identity.binary_path, PathBuf::from("/usr/bin/agent")); + // A missing digest is `None`, never an empty value. + assert_eq!(identity.binary_sha256.as_deref(), Some("deadbeef")); +} + +#[tokio::test] +async fn missing_digest_is_none_never_empty() { + // The secondary backend resolves path-only identity: the digest is `None`, + // so policy that requires a digest cannot authorize the connection. + let ingress = MockIngress::(PhantomData); + let conn = ingress.accept().await.expect("accept"); + let identity = conn.binary_identity.expect("identity resolves"); + assert!(identity.binary_sha256.is_none()); +} + +#[tokio::test] +async fn unresolved_identity_travels_with_the_connection_and_fails_closed() { + // Attribution failure does not tear down the ingress: the connection is + // delivered carrying `Err`, and the mediation service denies it. + let ingress = UnattributedIngress; + let conn = ingress.accept().await.expect("accept"); + assert!(conn.binary_identity.is_err()); +} + +// --------------------------------------------------------------------------- +// Descriptor environment transport. +// --------------------------------------------------------------------------- + +#[test] +fn descriptor_env_value_round_trips() { + let original = TopologyDescriptor { + contract_version: CONTRACT_VERSION, + backend_id: "mock-primary".to_string(), + payload: vec![0x00, 0x7f, 0xff], + }; + let parsed = TopologyDescriptor::from_env_value(&original.to_env_value()).expect("parse"); + assert_eq!(parsed.contract_version, original.contract_version); + assert_eq!(parsed.backend_id, original.backend_id); + assert_eq!(parsed.payload, original.payload); +} + +#[test] +fn in_pod_descriptor_env_value_round_trips_with_empty_payload() { + let parsed = TopologyDescriptor::from_env_value(&TopologyDescriptor::in_pod().to_env_value()) + .expect("parse"); + assert_eq!(parsed.backend_id, IN_POD_BACKEND_ID); + assert_eq!(parsed.contract_version, CONTRACT_VERSION); + assert!(parsed.payload.is_empty()); +} + +#[test] +fn malformed_descriptor_env_values_fail_closed() { + for bad in [ + "", + "1", + "1:in-pod", + "x:in-pod:", + "1::", + "1:in-pod:zz", + "1:in-pod:abc", + "1:in-pod:aa:bb", + ] { + assert!( + TopologyDescriptor::from_env_value(bad).is_err(), + "{bad:?} must fail closed" + ); + } +} + +// --------------------------------------------------------------------------- +// Errors. +// --------------------------------------------------------------------------- + +#[test] +fn error_kinds_map_and_only_unavailable_retries() { + assert_eq!( + BackendError::Descriptor("x".into()).kind(), + BackendErrorKind::Invalid + ); + assert_eq!( + BackendError::NotRegistered("x".into()).kind(), + BackendErrorKind::Invalid + ); + assert_eq!( + BackendError::Denied("x".into()).kind(), + BackendErrorKind::Denied + ); + assert_eq!( + BackendError::Unavailable("x".into()).kind(), + BackendErrorKind::Unavailable + ); + assert_eq!( + BackendError::Attach("x".into()).kind(), + BackendErrorKind::Failed + ); + assert_eq!( + BackendError::Confirm("x".into()).kind(), + BackendErrorKind::Failed + ); + assert_eq!( + BackendError::Terminated("x".into()).kind(), + BackendErrorKind::Terminated + ); + assert!(BackendError::Unavailable("x".into()).is_retryable()); + assert!(!BackendError::Confirm("x".into()).is_retryable()); +} diff --git a/crates/openshell-isolation/src/lib.rs b/crates/openshell-isolation/src/lib.rs new file mode 100644 index 000000000..b900058c8 --- /dev/null +++ b/crates/openshell-isolation/src/lib.rs @@ -0,0 +1,99 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The `OpenShell` **Isolation Backend** runtime contract (RFC 0012). +//! +//! An isolation backend establishes and enforces an agent's isolation boundary; +//! the supervisor role drives it through one contract. The supervisor-facing +//! contract lives in [`contract`]: an object-safe, runtime-selectable factory +//! plus a fixed chain of boxed lifecycle states the supervisor advances without +//! branching on where the boundary sits. The same calls work whether the +//! boundary lives in the agent's container (the in-pod backend) or further out +//! (a microVM, a node daemon, a separate pod). +//! +//! # The isolation envelope +//! +//! The boundary spans four [dimensions](IsolationDimension): network, +//! filesystem (Landlock), syscall (seccomp), and binary identity. A valid +//! backend establishes standing enforcement before untrusted code runs and +//! ensures the launch-time controls are in force before each process's first +//! untrusted instruction. Binary identity is resolved by the backend for every +//! accepted connection and delivered on +//! [`contract::MediatedConnection`]; network and exec are operated at runtime +//! through [`contract::MediationIngress`], [`contract::BoundaryExec`], and +//! [`contract::BoundaryPortForward`]. +//! +//! # Ordering is a security property +//! +//! The lifecycle states run in order: attach -> Bound -> confirm -> Ready -> +//! start_agent -> Running. Nothing untrusted runs inside the boundary until it +//! is confirmed ready. This is enforced *by construction*: each transition +//! consumes the prior state by value, and no state type has a public +//! constructor, so the supervisor cannot skip a stage or run a workload before +//! [`contract::ReadyBoundary`] exists. +//! +//! [`AgentSpec`] is shared between the workload definition the supervisor +//! submits and the [`contract::SandboxContext`] that `attach` binds to a +//! boundary. + +/// The four dimensions of the isolation envelope (RFC 0012). +/// +/// A valid backend establishes all four before the agent runs. The variants are +/// a machine-readable inventory for an admission layer; the runtime contract in +/// [`contract`] carries all four by shape rather than enumerating them per call. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum IsolationDimension { + /// Network namespace, routing, and proxy mediation. + Network, + /// Filesystem confinement (Landlock), established at the pre-exec ceiling. + Filesystem, + /// Syscall confinement (seccomp), established at the pre-exec ceiling. + Syscall, + /// Binary identity, resolved by the backend per accepted connection and + /// delivered on the mediated connection. + BinaryIdentity, +} + +impl IsolationDimension { + /// Every dimension of the isolation envelope: a machine-readable inventory + /// for an admission layer that needs to iterate the dimensions a backend + /// must satisfy. + pub const ALL: [Self; 4] = [ + Self::Network, + Self::Filesystem, + Self::Syscall, + Self::BinaryIdentity, + ]; +} + +/// The agent workload to run inside the boundary. +/// +/// Carried by [`contract::SandboxContext`] so a backend's `start_agent` takes no +/// spec; the bound boundary already carries what runs inside it. +#[derive(Debug, Clone)] +pub struct AgentSpec { + /// Entrypoint program. + pub program: String, + /// Entrypoint arguments. + pub args: Vec, + /// Working directory for the entrypoint, if any. + pub workdir: Option, + /// Wall-clock timeout for the entrypoint in seconds (0 = no timeout). + pub timeout_secs: u64, + /// Whether the entrypoint runs interactively (inherits the parent pgrp). + pub interactive: bool, +} + +pub mod contract; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn isolation_dimensions_inventory_is_complete() { + assert_eq!(IsolationDimension::ALL.len(), 4); + assert!(IsolationDimension::ALL.contains(&IsolationDimension::Network)); + assert!(IsolationDimension::ALL.contains(&IsolationDimension::BinaryIdentity)); + } +} diff --git a/crates/openshell-sandbox/Cargo.toml b/crates/openshell-sandbox/Cargo.toml index 6a51635b1..3f95830f5 100644 --- a/crates/openshell-sandbox/Cargo.toml +++ b/crates/openshell-sandbox/Cargo.toml @@ -16,6 +16,7 @@ path = "src/main.rs" [dependencies] openshell-core = { path = "../openshell-core", default-features = false } +openshell-isolation = { path = "../openshell-isolation" } openshell-ocsf = { path = "../openshell-ocsf" } openshell-policy = { path = "../openshell-policy" } openshell-supervisor-network = { path = "../openshell-supervisor-network" } @@ -26,6 +27,12 @@ openshell-supervisor-process = { path = "../openshell-supervisor-process" } # Async runtime tokio = { workspace = true } +# Async trait for the IsolationBackend impl +async-trait = "0.1" + +# Stream adapter for the prototype EventSource seam +futures = { workspace = true } + # gRPC (tonic::Status downcast in error mapping) tonic = { workspace = true, features = ["channel", "tls-native-roots"] } prost-types = { workspace = true } diff --git a/crates/openshell-sandbox/src/inpod.rs b/crates/openshell-sandbox/src/inpod.rs new file mode 100644 index 000000000..bcf592c27 --- /dev/null +++ b/crates/openshell-sandbox/src/inpod.rs @@ -0,0 +1,867 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! The in-pod isolation backend (RFC 0012 runtime-selectable contract). +//! +//! This is the co-located placement: the supervisor process hosts the +//! supervisor role, the mediation service, and the backend in the agent's +//! container. It implements the object-safe boxed state chain +//! (`attach -> Bound -> confirm -> Ready -> start_agent -> Running`) over the +//! existing supervisor primitives without changing their behavior: +//! `create_netns_for_proxy` (network), `run_networking` (proxy mediation), +//! the pre-exec ceiling in `spawn_workload` (filesystem/Landlock + +//! syscall/seccomp), and procfs (binary identity). +//! +//! `attach` validates the (empty) in-pod payload and atomically binds the +//! trusted [`SandboxContext`] to the boundary it establishes: the workload +//! network namespace and the mediation service come up inside `attach`, so +//! `Bound` means what the RFC says it means — descriptor and context bound to +//! the same resource, mediation ingress available, no untrusted workload code +//! running. Each transition consumes the prior state by value, so the call +//! order, and thus "no untrusted instruction before the boundary is ready", is +//! enforced by construction. +//! +//! Execution-domain note: the in-pod backend relies on container-runtime +//! inheritance — the supervisor and every child it spawns run in the pod's +//! cgroup with the device set the CRI granted the container — so every workload +//! descendant remains in the compute driver's provisioned execution +//! environment by construction. + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; + +use async_trait::async_trait; + +use openshell_core::activity::ActivitySender; +use openshell_core::denial::DenialEvent; +use openshell_core::policy::NetworkMode; +use openshell_core::provider_credentials::ProviderCredentialState; +use openshell_isolation::contract::{ + BackendError, BoundBoundary, BoundaryExec, BoundaryExitStatus, BoundaryPortForward, + BoundaryProcess, BoundarySignal, CONTRACT_VERSION, ExecSession, ExecSpec, + IsolationBackendFactory, MediatedConnection, MediationIngress, ReadyBoundary, RunningBoundary, + SandboxContext, VerifiedTopologyDescriptor, +}; +use openshell_supervisor_network::identity_source::ProcfsIdentityResolver; +use openshell_supervisor_network::opa::OpaEngine; +use openshell_supervisor_network::run::{Networking, run_networking}; +use openshell_supervisor_process::boundary_io::NetnsPortForward; +use openshell_supervisor_process::process::ProcessEnforcementMode; +use openshell_supervisor_process::run::{AgentSignaler, SpawnedAgent, spawn_workload}; +use tokio::sync::mpsc::UnboundedSender; + +#[cfg(target_os = "linux")] +use openshell_supervisor_process::netns::{NetworkNamespace, create_netns_for_proxy}; + +/// The registered id for the in-pod backend (defined in the contract crate so +/// compute drivers can name it in the descriptors they create). +pub use openshell_isolation::contract::IN_POD_BACKEND_ID; + +// ============================================================================ +// Config and factory +// ============================================================================ + +/// Runtime collaborators the in-pod lifecycle calls need, captured once when the +/// factory is built. Move-once values (the event senders) are held behind a +/// `Mutex>` so the `&self` factory/state methods can take them exactly +/// when the matching transition fires. Policy, workload, and sandbox identity +/// are *not* here; they arrive in the trusted [`SandboxContext`] at `attach`. +pub struct InPodConfig { + pub network_enabled: bool, + pub process_enabled: bool, + pub opa_engine: Option>, + pub retained_proto: Option, + pub entrypoint_pid: Arc, + pub provider_credentials: ProviderCredentialState, + /// Child environment for the agent, resolved at startup. Mutated in place by + /// `attach` if the GCE metadata loopback server fails to come up. + pub provider_env: Mutex>, + /// Process launch-time enforcement level (full privileged setup vs. + /// network-sidecar reduced mode), resolved by the supervisor at startup. + pub process_enforcement_mode: ProcessEnforcementMode, + pub sandbox_name: Option, + pub openshell_endpoint: Option, + pub inference_routes: Option, + pub ssh_socket_path: Option, + /// Workspace watch receiver: handed to `run_networking` so the proxy's + /// policy-local and activity paths observe the workspace the poll loop + /// learns from `GetSandboxConfig`. + pub workspace_rx: tokio::sync::watch::Receiver, + /// Proxy-side denial / activity senders (consumed by `attach`). + pub denial_tx: Mutex>>, + pub activity_tx: Mutex>, + /// Bypass-monitor denial / activity senders (consumed by `start_agent`). + #[cfg(target_os = "linux")] + pub bypass_denial_tx: Mutex>>, + #[cfg(target_os = "linux")] + pub bypass_activity_tx: Mutex>, + /// Output slot: `attach` publishes the proxy's policy-local route context + /// here so the orchestrator's policy poll loop can pick it up without + /// reaching into in-pod-specific state through the `dyn` boundary. + pub policy_local_slot: + Arc>>>, +} + +/// The factory for the in-pod backend. Holds the per-sandbox [`InPodConfig`] and +/// hands it to the boundary on the single `attach`. +pub struct InPodBackendFactory { + config: Mutex>, + /// Whether a prior `attach` consumed the config and then failed during + /// establishment. The one-shot event senders are consumed with it, so the + /// in-pod resource cannot be re-attached; this keeps the error truthful + /// ("attempt failed", not "already bound"). + attach_failed: AtomicBool, +} + +impl InPodBackendFactory { + /// Build the factory from its per-sandbox runtime collaborators. + #[must_use] + pub fn new(config: InPodConfig) -> Self { + Self { + config: Mutex::new(Some(config)), + attach_failed: AtomicBool::new(false), + } + } +} + +#[async_trait] +impl IsolationBackendFactory for InPodBackendFactory { + fn backend_id(&self) -> &'static str { + IN_POD_BACKEND_ID + } + + fn contract_version(&self) -> u32 { + CONTRACT_VERSION + } + + async fn attach( + &self, + descriptor: VerifiedTopologyDescriptor, + sandbox: SandboxContext, + ) -> Result, BackendError> { + // Validate the in-pod payload: the supervisor process *is* the + // resource, so the payload carries nothing. + if !descriptor.payload().is_empty() { + return Err(BackendError::Descriptor( + "in-pod descriptor payload must be empty".to_string(), + )); + } + // `attach` never binds a resource that is already bound to an active + // boundary: the in-pod resource is this process, bindable exactly once. + let config = self + .config + .lock() + .expect("in-pod config lock") + .take() + .ok_or_else(|| { + if self.attach_failed.load(Ordering::SeqCst) { + BackendError::Attach( + "a previous in-pod attach failed during establishment; \ + the in-pod resource cannot be re-attached" + .to_string(), + ) + } else { + BackendError::Denied( + "in-pod resource is already bound to an active boundary".to_string(), + ) + } + })?; + + match establish(config, sandbox).await { + Ok(bound) => Ok(Box::new(bound)), + Err(e) => { + self.attach_failed.store(true, Ordering::SeqCst); + Err(e) + } + } + } +} + +/// Establish the in-pod boundary: standing enforcement (the workload network +/// namespace) and the mediation service (the in-pod proxy), bound atomically to +/// the trusted sandbox context. Consumes the one-shot config; failure fails +/// closed with partial state released by RAII. +async fn establish( + config: InPodConfig, + sandbox: SandboxContext, +) -> Result { + // Establish the network dimension of standing enforcement: create the + // workload's network namespace and install the bypass-detection rules. + // Filesystem and syscall are launch-time controls applied per process; + // binary identity is resolved per accepted connection. + #[cfg(target_os = "linux")] + let netns = if config.network_enabled { + create_netns_for_proxy(&sandbox.policy).map_err(|e| BackendError::Attach(e.to_string()))? + } else { + None + }; + + // Bring up the mediation service (the in-pod proxy) so the mediation + // ingress is available at `Bound`. + let networking = if config.network_enabled { + #[cfg(target_os = "linux")] + let proxy_bind_ip = netns.as_ref().map(NetworkNamespace::host_ip); + #[cfg(not(target_os = "linux"))] + let proxy_bind_ip: Option = None; + + // Take the senders into locals so the Mutex guards drop before the + // await (a guard held across an await would make the future !Send). + let denial_tx = config.denial_tx.lock().expect("denial_tx lock").take(); + let activity_tx = config.activity_tx.lock().expect("activity_tx lock").take(); + let networking = run_networking( + &sandbox.policy, + proxy_bind_ip, + config.opa_engine.as_ref(), + config.retained_proto.as_ref(), + config.entrypoint_pid.clone(), + config.process_enabled, + &config.provider_credentials, + Some(sandbox.sandbox_id.as_str()), + config.sandbox_name.as_deref(), + config.openshell_endpoint.as_deref(), + config.inference_routes.as_deref(), + denial_tx, + activity_tx, + config.workspace_rx.clone(), + ) + .await + .map_err(|e| BackendError::Attach(e.to_string()))?; + Some(networking) + } else { + None + }; + + // Start the GCE metadata loopback server inside the namespace so Go's + // metadata client (which bypasses HTTP_PROXY) can reach it via direct + // TCP. Must come up before start_agent; on failure the GCE env vars are + // stripped so the SDK falls back cleanly. + #[cfg(target_os = "linux")] + if let Some(ns) = netns.as_ref() { + ensure_gce_metadata_server(&config, ns).await; + } + + // Publish the policy-local route context for the orchestrator's poll + // loop (the mediation service's policy-update path: authorized + // revisions apply there without a backend lifecycle transition). + *config + .policy_local_slot + .lock() + .expect("policy_local_slot lock") = networking.as_ref().map(|n| n.policy_local_ctx.clone()); + + // The mediation service's backend-neutral connection source, carrying + // the per-connection identity resolver. The live in-pod proxy still + // accepts directly (see `InPodMediationIngress`); this satisfies the + // contract shape, and a delegated backend supplies a transport-backed + // version. + let mediation_ingress: Arc = Arc::new(InPodMediationIngress { + _identity: ProcfsIdentityResolver { + entrypoint_pid: config.entrypoint_pid.clone(), + }, + }); + + Ok(InPodBound { + config, + sandbox, + #[cfg(target_os = "linux")] + netns, + networking, + mediation_ingress, + }) +} + +// ============================================================================ +// Lifecycle states +// ============================================================================ + +/// Bound: the descriptor and trusted sandbox context are bound to this process's +/// boundary, and the mediation ingress is available. No untrusted workload code +/// is running. +struct InPodBound { + config: InPodConfig, + sandbox: SandboxContext, + #[cfg(target_os = "linux")] + netns: Option, + networking: Option, + mediation_ingress: Arc, +} + +#[async_trait] +impl BoundBoundary for InPodBound { + fn mediation_ingress(&self) -> Arc { + self.mediation_ingress.clone() + } + + async fn confirm(self: Box) -> Result, BackendError> { + // Effective-mediation check (fail closed). In proxy mode the only egress + // is through the proxy bound inside the workload netns; if the namespace + // or the proxy listener is absent, the boundary does not safely gate + // egress, so we must not produce Ready. + // + // Note: this confirms mediation *structure*. The nftables bypass- + // detection ruleset still uses `policy accept` with protocol-specific + // rejects (see `netns/nft_ruleset.rs`); hardening it to a true + // default-deny is tracked as remaining implementation work and is NOT + // certified here. + if self.config.network_enabled + && matches!(self.sandbox.policy.network.mode, NetworkMode::Proxy) + { + #[cfg(target_os = "linux")] + if self.netns.is_none() { + return Err(BackendError::Confirm( + "proxy mode requires a workload network namespace; none established" + .to_string(), + )); + } + let proxy_up = self + .networking + .as_ref() + .and_then(|n| n.proxy.as_ref()) + .is_some(); + if !proxy_up { + return Err(BackendError::Confirm( + "proxy listener is not bound; egress mediation is not in effect".to_string(), + )); + } + } + + Ok(Box::new(InPodReady { + config: self.config, + sandbox: self.sandbox, + #[cfg(target_os = "linux")] + netns: self.netns, + networking: self.networking, + })) + } +} + +/// Ready: standing enforcement and mediation are confirmed. Only agent +/// activation is possible. +struct InPodReady { + config: InPodConfig, + sandbox: SandboxContext, + #[cfg(target_os = "linux")] + netns: Option, + networking: Option, +} + +#[async_trait] +impl ReadyBoundary for InPodReady { + async fn start_agent(self: Box) -> Result, BackendError> { + let this = *self; + let config = this.config; + let sandbox = this.sandbox; + #[cfg(target_os = "linux")] + let netns = this.netns; + let networking = this.networking; + + #[cfg(target_os = "linux")] + let netns_fd = netns.as_ref().and_then(NetworkNamespace::ns_fd); + #[cfg(not(target_os = "linux"))] + let netns_fd: Option = None; + + let port_forward: Arc = Arc::new(NetnsPortForward { netns_fd }); + let exec: Arc = Arc::new(InPodExec); + + // The in-pod backend creates the agent process itself; the launch-time + // controls (Landlock, seccomp, privilege drop) are applied inside + // `spawn_workload`'s pre-exec ceiling, before the first untrusted + // instruction. + let agent: Arc = if config.process_enabled { + let spec = &sandbox.agent; + let ca_file_paths = networking.as_ref().and_then(|n| n.ca_file_paths.clone()); + let provider_env = config + .provider_env + .lock() + .expect("provider_env lock") + .clone(); + + #[cfg(target_os = "linux")] + let bypass_denial_tx = config + .bypass_denial_tx + .lock() + .expect("bypass_denial_tx lock") + .take(); + #[cfg(target_os = "linux")] + let bypass_activity_tx = config + .bypass_activity_tx + .lock() + .expect("bypass_activity_tx lock") + .take(); + + let spawned = spawn_workload( + &spec.program, + &spec.args, + spec.workdir.as_deref(), + spec.timeout_secs, + spec.interactive, + Some(sandbox.sandbox_id.as_str()), + config.openshell_endpoint.as_deref(), + config.ssh_socket_path.clone(), + // In-pod co-locates the SSH socket with the workload; it is not + // shared with a separate network-sidecar container. + false, + &sandbox.policy, + config.process_enforcement_mode, + config.entrypoint_pid.clone(), + // No sidecar control channel awaits the in-pod entrypoint PID. + None, + config.provider_credentials.clone(), + provider_env, + ca_file_paths, + #[cfg(target_os = "linux")] + netns.as_ref(), + #[cfg(target_os = "linux")] + bypass_denial_tx, + #[cfg(target_os = "linux")] + bypass_activity_tx, + ) + .await + .map_err(|e| BackendError::Process(e.to_string()))?; + + Arc::new(InPodAgentProcess::running(spawned)) + } else { + // Network-only (sidecar/legacy) mode: no workload in this pod. The + // boundary is held open until a shutdown signal. This is a legacy + // split mode, not a gated external workload. + Arc::new(InPodAgentProcess::hold_open()) + }; + + Ok(Box::new(InPodRunning { + agent, + exec, + port_forward, + _networking: networking, + #[cfg(target_os = "linux")] + _netns: netns, + })) + } +} + +/// Running: the agent is runnable behind the boundary. Exec, forwarding, wait, +/// and signal are available. The mediation ingress was retained by the +/// supervisor from the `Bound` state. +struct InPodRunning { + agent: Arc, + exec: Arc, + port_forward: Arc, + /// Held to keep the proxy task and network namespace alive for the boundary's + /// life; dropping the running state tears them down (RAII), which is the + /// backend reclaiming backend-private state — the contract defines no public + /// cleanup transition. + _networking: Option, + #[cfg(target_os = "linux")] + _netns: Option, +} + +impl RunningBoundary for InPodRunning { + fn agent(&self) -> Arc { + self.agent.clone() + } + fn exec(&self) -> Arc { + self.exec.clone() + } + fn port_forward(&self) -> Arc { + self.port_forward.clone() + } +} + +// ============================================================================ +// Agent process handle +// ============================================================================ + +/// The agent process running inside the in-pod boundary. `wait` returns a stable +/// terminal status across repeated calls; signals go through the lock-free +/// pid-based [`AgentSignaler`] so they never contend with an in-flight `wait`. +struct InPodAgentProcess { + pid: Option, + signaler: Option, + waiter: tokio::sync::Mutex, +} + +enum AgentWaitState { + Running(SpawnedAgent), + HoldOpen, + Done(BoundaryExitStatus), +} + +impl InPodAgentProcess { + fn running(spawned: SpawnedAgent) -> Self { + Self { + pid: Some(spawned.pid()), + signaler: Some(spawned.signaler()), + waiter: tokio::sync::Mutex::new(AgentWaitState::Running(spawned)), + } + } + + fn hold_open() -> Self { + Self { + pid: None, + signaler: None, + waiter: tokio::sync::Mutex::new(AgentWaitState::HoldOpen), + } + } +} + +#[async_trait] +impl BoundaryProcess for InPodAgentProcess { + async fn wait(&self) -> Result { + // Holding the lock across the wait serializes repeated callers: the first + // performs the wait and caches the status; later callers block on the + // lock, then observe the cached `Done`. Signals never take this lock. + let mut guard = self.waiter.lock().await; + match &mut *guard { + AgentWaitState::Done(status) => Ok(*status), + AgentWaitState::HoldOpen => { + crate::wait_for_shutdown_signal().await; + let status = BoundaryExitStatus::Exited(0); + *guard = AgentWaitState::Done(status); + Ok(status) + } + AgentWaitState::Running(agent) => { + let code = agent + .wait() + .await + .map_err(|e| BackendError::Process(e.to_string()))?; + let status = BoundaryExitStatus::Exited(code); + *guard = AgentWaitState::Done(status); + Ok(status) + } + } + } + + async fn signal(&self, signal: BoundarySignal) -> Result<(), BackendError> { + let Some(signaler) = self.signaler.as_ref() else { + // Network-only hold-open: no workload process to signal. + return Ok(()); + }; + let result = match signal { + BoundarySignal::Term => signaler.term(), + BoundarySignal::Kill => signaler.kill(), + BoundarySignal::Int => signaler.interrupt(), + BoundarySignal::Hup => signaler.hangup(), + }; + result.map_err(|e| BackendError::Process(e.to_string())) + } + + async fn terminate(&self) -> Result<(), BackendError> { + let Some(signaler) = self.signaler.as_ref() else { + return Ok(()); + }; + signaler + .term() + .map_err(|e| BackendError::Process(e.to_string())) + } + + fn diagnostic_pid(&self) -> Option { + self.pid + } +} + +// ============================================================================ +// Exec (SSH adoption pending) +// ============================================================================ + +/// In-pod exec interface. The live SSH server still spawns workload shells +/// directly (inside `spawn_workload`); routing those through an owned +/// [`ExecSession`] (stdio/PTY/wait) is the remaining live-adoption refactor, so +/// this returns an explicit error rather than a half-wired session. +struct InPodExec; + +#[async_trait] +impl BoundaryExec for InPodExec { + async fn exec(&self, _spec: ExecSpec) -> Result { + Err(BackendError::Process( + "in-pod BoundaryExec is not yet the live SSH exec path; SSH execs directly. \ + Wiring ExecSession stdio/PTY through this interface is pending (see POC handoff)." + .to_string(), + )) + } +} + +/// In-pod mediation ingress. The live in-pod proxy still accepts workload +/// connections on its own listener inside the netns and resolves identity +/// inline; routing those through `accept` — so every connection carries this +/// resolver's per-connection [`BinaryIdentity`] result, and a delegated backend +/// can deliver connections over its private transport instead — is the +/// remaining live-adoption refactor, so this returns an explicit error rather +/// than a competing accept loop. +/// +/// [`BinaryIdentity`]: openshell_isolation::contract::BinaryIdentity +struct InPodMediationIngress { + /// The per-connection resolver the adopted accept path will consult. + _identity: ProcfsIdentityResolver, +} + +#[async_trait] +impl MediationIngress for InPodMediationIngress { + async fn accept(&self) -> Result { + Err(BackendError::Process( + "in-pod MediationIngress is not yet the live proxy connection source; \ + the proxy accepts directly (see POC handoff)." + .to_string(), + )) + } +} + +// ============================================================================ +// GCE metadata loopback server +// ============================================================================ + +/// Bring up the GCE metadata loopback server inside the network namespace, +/// stripping the GCE env vars from the agent's environment if it fails so the +/// Go SDK falls back cleanly. +#[cfg(target_os = "linux")] +async fn ensure_gce_metadata_server(config: &InPodConfig, ns: &NetworkNamespace) { + use std::time::Duration; + use tokio::time::timeout; + use tracing::{info, warn}; + + if !config + .provider_credentials + .snapshot() + .child_env + .contains_key("GCE_METADATA_HOST") + { + return; + } + + let ctx = + crate::google_cloud_metadata::MetadataContext::new(config.provider_credentials.clone()); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + match ns + .bind_tcp_in_netns(openshell_core::google_cloud::METADATA_LOOPBACK_ADDR) + .await + { + Ok(listener) => { + tokio::spawn(crate::metadata_server::run(listener, ctx, ready_tx)); + if let Ok(Ok(addr)) = timeout(Duration::from_secs(5), ready_rx).await { + info!(addr = %addr, "GCE metadata loopback server ready"); + } else { + warn!("GCE metadata server failed to become ready, removing metadata env vars"); + strip_gce_env(config); + } + } + Err(e) => { + warn!(error = %e, "GCE metadata server bind failed, Go SDK may not discover credentials"); + strip_gce_env(config); + } + } +} + +/// Remove the GCE metadata env vars from both the agent's child env and the +/// provider credential state. +#[cfg(target_os = "linux")] +fn strip_gce_env(config: &InPodConfig) { + let mut env = config.provider_env.lock().expect("provider_env lock"); + env.remove("GCE_METADATA_HOST"); + env.remove("GCE_METADATA_IP"); + env.remove("METADATA_SERVER_DETECTION"); + drop(env); + config + .provider_credentials + .remove_env_key("GCE_METADATA_HOST"); +} + +#[cfg(test)] +mod tests { + use super::*; + use openshell_core::policy::{ + FilesystemPolicy, LandlockPolicy, NetworkPolicy, ProcessPolicy, SandboxPolicy, + }; + use openshell_isolation::AgentSpec; + use openshell_isolation::contract::{BackendRegistry, TopologyDescriptor}; + + /// A minimal in-pod config with networking and the workload disabled, so the + /// lifecycle runs without root, a network namespace, or a gateway. + fn minimal_config() -> InPodConfig { + InPodConfig { + network_enabled: false, + process_enabled: false, + opa_engine: None, + retained_proto: None, + entrypoint_pid: Arc::new(AtomicU32::new(0)), + provider_credentials: ProviderCredentialState::from_environment( + 0, + HashMap::new(), + HashMap::new(), + HashMap::new(), + ), + provider_env: Mutex::new(HashMap::new()), + process_enforcement_mode: ProcessEnforcementMode::Full, + sandbox_name: None, + openshell_endpoint: None, + inference_routes: None, + ssh_socket_path: None, + workspace_rx: tokio::sync::watch::channel(String::new()).1, + denial_tx: Mutex::new(None), + activity_tx: Mutex::new(None), + #[cfg(target_os = "linux")] + bypass_denial_tx: Mutex::new(None), + #[cfg(target_os = "linux")] + bypass_activity_tx: Mutex::new(None), + policy_local_slot: Arc::new(Mutex::new(None)), + } + } + + fn block_mode_policy() -> SandboxPolicy { + SandboxPolicy { + version: 1, + filesystem: FilesystemPolicy::default(), + network: NetworkPolicy { + mode: NetworkMode::Block, + proxy: None, + }, + landlock: LandlockPolicy::default(), + process: ProcessPolicy::default(), + } + } + + fn descriptor() -> TopologyDescriptor { + TopologyDescriptor { + contract_version: CONTRACT_VERSION, + backend_id: IN_POD_BACKEND_ID.to_string(), + payload: Vec::new(), + } + } + + fn sandbox_context() -> SandboxContext { + SandboxContext { + sandbox_id: "test-sandbox".to_string(), + policy: block_mode_policy(), + agent: AgentSpec { + program: "true".to_string(), + args: vec![], + workdir: None, + timeout_secs: 0, + interactive: false, + }, + } + } + + // ----- Factory and registry ----- + + #[test] + fn factory_speaks_the_contract_version() { + let factory = InPodBackendFactory::new(minimal_config()); + assert_eq!(factory.backend_id(), IN_POD_BACKEND_ID); + assert_eq!(factory.contract_version(), CONTRACT_VERSION); + } + + #[test] + fn registry_selects_in_pod_backend() { + let mut registry = BackendRegistry::new(); + registry + .register(Arc::new(InPodBackendFactory::new(minimal_config()))) + .expect("register"); + let (factory, _verified) = registry + .resolve(descriptor(), IN_POD_BACKEND_ID) + .expect("resolve"); + assert_eq!(factory.backend_id(), IN_POD_BACKEND_ID); + } + + #[test] + fn registry_rejects_duplicate_in_pod() { + let mut registry = BackendRegistry::new(); + registry + .register(Arc::new(InPodBackendFactory::new(minimal_config()))) + .expect("first register"); + assert!( + registry + .register(Arc::new(InPodBackendFactory::new(minimal_config()))) + .is_err() + ); + } + + #[test] + fn registry_rejects_admission_mismatch() { + let mut registry = BackendRegistry::new(); + registry + .register(Arc::new(InPodBackendFactory::new(minimal_config()))) + .expect("register"); + // The descriptor names in-pod, but admission expects a different backend. + assert!( + registry + .resolve(descriptor(), "some-other-backend") + .map(|_| ()) + .is_err() + ); + } + + // ----- Lifecycle (no root / no netns) ----- + + /// Drive the real in-pod chain attach -> Bound -> confirm -> Ready and prove + /// the retained mediation ingress survives the consuming transitions. + #[tokio::test] + async fn lifecycle_reaches_ready_and_retains_ingress() { + let mut registry = BackendRegistry::new(); + registry + .register(Arc::new(InPodBackendFactory::new(minimal_config()))) + .expect("register"); + let (factory, verified) = registry + .resolve(descriptor(), IN_POD_BACKEND_ID) + .expect("resolve"); + + let bound = factory + .attach(verified, sandbox_context()) + .await + .expect("attach"); + let ingress = bound.mediation_ingress(); + let _ready = bound.confirm().await.expect("confirm"); + // The retained ingress remains usable after `Bound` is consumed; the + // in-pod accept adoption is pending, so it reports an explicit error + // rather than a half-wired connection. + assert!(ingress.accept().await.is_err()); + } + + /// `attach` is atomic and never binds a resource that is already bound to an + /// active boundary: the in-pod resource binds exactly once. + #[tokio::test] + async fn second_attach_is_denied() { + let mut registry = BackendRegistry::new(); + registry + .register(Arc::new(InPodBackendFactory::new(minimal_config()))) + .expect("register"); + + let (factory, verified) = registry + .resolve(descriptor(), IN_POD_BACKEND_ID) + .expect("resolve"); + let _bound = factory + .attach(verified, sandbox_context()) + .await + .expect("first attach"); + + let (_factory2, verified2) = registry + .resolve(descriptor(), IN_POD_BACKEND_ID) + .expect("re-resolve"); + let err = factory + .attach(verified2, sandbox_context()) + .await + .map(|_| ()) + .expect_err("second attach must fail"); + assert!(matches!(err, BackendError::Denied(_))); + } + + /// The in-pod payload is empty by construction; a non-empty payload is a + /// descriptor error, validated by the backend at `attach`. + #[tokio::test] + async fn non_empty_payload_is_rejected() { + let mut registry = BackendRegistry::new(); + registry + .register(Arc::new(InPodBackendFactory::new(minimal_config()))) + .expect("register"); + let bad = TopologyDescriptor { + contract_version: CONTRACT_VERSION, + backend_id: IN_POD_BACKEND_ID.to_string(), + payload: vec![1, 2, 3], + }; + let (factory, verified) = registry.resolve(bad, IN_POD_BACKEND_ID).expect("resolve"); + let err = factory + .attach(verified, sandbox_context()) + .await + .map(|_| ()) + .expect_err("payload must be rejected"); + assert!(matches!(err, BackendError::Descriptor(_))); + } +} diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index f9555d9f2..cdb6c26fb 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -9,6 +9,7 @@ mod activity_aggregator; mod denial_aggregator; #[cfg_attr(not(target_os = "linux"), allow(dead_code))] mod google_cloud_metadata; +mod inpod; mod mechanistic_mapper; #[cfg_attr(not(target_os = "linux"), allow(dead_code))] mod metadata_server; @@ -301,23 +302,10 @@ pub async fn run_sandbox( // the entrypoint process's /proc/net/tcp for identity binding. let entrypoint_pid = Arc::new(AtomicU32::new(0)); - // Create the workload's network namespace. It is shared infrastructure: - // the proxy binds to its host-side veth IP, the bypass monitor reads - // /dev/kmsg from inside it, and the workload child / SSH sessions enter - // it via setns(). The RAII handle lives in this frame for the duration - // of the sandbox. - #[cfg(target_os = "linux")] - let netns = if network_enabled && !sidecar_network_enforcement { - openshell_supervisor_process::netns::create_netns_for_proxy(&policy)? - } else { - None - }; - - // The denial channel is owned by the orchestrator: the proxy (in the - // networking leaf) and the bypass monitor (in the process leaf) both - // produce DenialEvents that the denial aggregator (orchestrator-side) - // consumes via the matching receiver. Both leaves are pure producers; - // the orchestrator owns the consumer task spawned below. + // Event channels are owned by the orchestrator: the proxy and the bypass + // monitor produce DenialEvents / activity records that the orchestrator's + // aggregators (spawned below) consume via the matching receiver. Both + // leaves are pure producers; the orchestrator owns the consumers. let (denial_tx, denial_rx, bypass_denial_tx): ( Option>, _, @@ -333,9 +321,7 @@ pub async fn run_sandbox( drop(bypass_denial_tx); // Anonymous activity channel: same orchestrator-owned pattern as the - // denial channel. The proxy and the bypass monitor both emit per-event - // activity records; the orchestrator-side aggregator drains, sanitizes, - // and flushes anonymous summaries to the gateway. + // denial channel. let (activity_tx, activity_rx, bypass_activity_tx) = if sandbox_id.is_some() { let (tx, rx) = tokio::sync::mpsc::channel(openshell_core::activity::ACTIVITY_EVENT_QUEUE_CAPACITY); @@ -352,117 +338,10 @@ pub async fn run_sandbox( // API read the current value so proposals target the correct workspace. let (workspace_tx, workspace_rx) = tokio::sync::watch::channel(String::new()); - let networking = if network_enabled { - #[cfg(target_os = "linux")] - let proxy_bind_ip = netns - .as_ref() - .map(openshell_supervisor_process::netns::NetworkNamespace::host_ip); - #[cfg(not(target_os = "linux"))] - let proxy_bind_ip: Option = None; - - Some( - openshell_supervisor_network::run::run_networking( - &policy, - proxy_bind_ip, - opa_engine.as_ref(), - retained_proto.as_ref(), - entrypoint_pid.clone(), - process_enabled, - &provider_credentials, - sandbox_id.as_deref(), - sandbox_name_for_agg.as_deref(), - openshell_endpoint_for_proxy.as_deref(), - inference_routes.as_deref(), - denial_tx, - activity_tx, - agent_proposals.clone(), - workspace_rx.clone(), - &upstream_proxy_args, - ) - .await?, - ) - } else { - None - }; - - #[cfg(target_os = "linux")] - let sidecar_control_server = if network_enabled && sidecar_network_enforcement { - if !matches!(policy.network.mode, NetworkMode::Proxy) { - return Err(miette::miette!( - "sidecar network enforcement requires proxy network mode" - )); - } - let socket = sidecar_control_socket().ok_or_else(|| { - miette::miette!( - "{} is required for sidecar topology", - openshell_core::sandbox_env::SIDECAR_CONTROL_SOCKET - ) - })?; - let proto = retained_proto.as_ref().ok_or_else(|| { - miette::miette!( - "sidecar topology requires gateway policy data for the process supervisor" - ) - })?; - let ca_paths = networking.as_ref().and_then(|n| n.ca_file_paths.clone()); - Some(sidecar_control::spawn_server( - &socket, - sidecar_control::BootstrapData { - policy_proto: proto.clone(), - provider_env_revision: provider_credentials.snapshot().revision, - provider_child_env: provider_env.clone(), - agent_proposals_enabled: agent_proposals.enabled(), - proxy_ca_cert_path: ca_paths.as_ref().map(|paths| paths.0.clone()), - proxy_ca_bundle_path: ca_paths.as_ref().map(|paths| paths.1.clone()), - }, - sidecar_expected_peer()?, - )?) - } else { - None - }; - #[cfg(not(target_os = "linux"))] - let sidecar_control_server: Option = None; - - let sidecar_control_publisher = sidecar_control_server - .as_ref() - .map(sidecar_control::ServerHandle::publisher); - - #[cfg(target_os = "linux")] - let mut sidecar_control_task = None; - - #[cfg(target_os = "linux")] - if network_enabled - && sidecar_network_enforcement - && let Some(server) = sidecar_control_server - { - let trusted_ssh_socket_path = ssh_socket_path.clone().ok_or_else(|| { - miette::miette!( - "{} is required for sidecar network topology", - openshell_core::sandbox_env::SSH_SOCKET_PATH - ) - })?; - let (entrypoint_rx, connection_task) = server.into_runtime_parts(); - sidecar_control_task = Some(connection_task); - spawn_sidecar_entrypoint_handler( - entrypoint_rx, - entrypoint_pid.clone(), - opa_engine.clone(), - retained_proto.clone(), - openshell_endpoint.clone(), - sandbox_id.clone(), - std::path::PathBuf::from(trusted_ssh_socket_path), - ); - } - - #[cfg(not(target_os = "linux"))] - if network_enabled && sidecar_network_enforcement { - return Err(miette::miette!( - "sidecar network enforcement is only supported on Linux" - )); - } - - // Spawn the denial-aggregator flush task. The aggregator drains denial - // events from the proxy + bypass monitor, batches them, and ships - // summaries to the gateway via `SubmitPolicyAnalysis`. + // Spawn the denial-aggregator flush task. Shared by both topologies: it + // drains denial events from the proxy + bypass monitor, batches them, and + // ships summaries to the gateway via SubmitPolicyAnalysis. Gated on a known + // workspace so proposals are attributed correctly. if let (Some(rx), Some(endpoint)) = (denial_rx, openshell_endpoint_for_proxy.as_deref()) { // SubmitPolicyAnalysis resolves by sandbox *name*, not UUID — fall // back to the ID when the name isn't set. @@ -506,9 +385,7 @@ pub async fn run_sandbox( }); } - // Spawn the activity-aggregator flush task. The aggregator drains - // anonymous activity events from the proxy, sanitizes deny groups, - // and ships periodic summaries to the gateway. + // Spawn the activity-aggregator flush task. Shared by both topologies. if let (Some(rx), Some(endpoint)) = (activity_rx, openshell_endpoint_for_proxy.as_deref()) { let agg_name = sandbox_name_for_agg .clone() @@ -551,118 +428,244 @@ pub async fn run_sandbox( }); } - // Spawn background policy poll task (gRPC mode only). - if !process_uses_sidecar_control - && let (Some(id), Some(endpoint), Some(engine)) = ( - sandbox_id.as_deref(), - openshell_endpoint.as_deref(), - opa_engine.as_ref(), - ) - { - let poll_id = id.to_string(); - let poll_endpoint = endpoint.to_string(); - let poll_engine = engine.clone(); - let poll_ocsf_enabled = ocsf_enabled.clone(); - let poll_pid = entrypoint_pid.clone(); - let poll_provider_credentials = provider_credentials.clone(); - let poll_policy_local = networking.as_ref().map(|n| n.policy_local_ctx.clone()); - let poll_interval_secs: u64 = std::env::var("OPENSHELL_POLICY_POLL_INTERVAL_SECS") - .ok() - .and_then(|v| v.parse().ok()) - .unwrap_or(10); - let poll_ctx = PolicyPollLoopContext { - endpoint: poll_endpoint, - sandbox_id: poll_id, - opa_engine: poll_engine, - loaded_policy_origin, - entrypoint_pid: poll_pid, - interval_secs: poll_interval_secs, - ocsf_enabled: poll_ocsf_enabled, - provider_credentials: poll_provider_credentials, - policy_local_ctx: poll_policy_local, - agent_proposals: agent_proposals.clone(), - middleware_registry_status, - sidecar_control_publisher: sidecar_control_publisher.clone(), - workspace_tx, + // Topology gate. When the network dimension is enforced by a separate + // sidecar container (#2076), run the inline sidecar path: proxy + sidecar + // control server in the network container, process supervisor talking to it + // over the control socket in the workload container. Otherwise drive the + // co-located in-pod isolation backend (RFC 0012), which owns the network + // namespace, proxy, and workload behind the boundary lifecycle. + let exit_code = if sidecar_network_enforcement { + // Legacy in-pod netns is never created under sidecar enforcement; the + // expression evaluates to None here but is kept for structural parity + // with the non-sidecar supervisor path. + #[cfg(target_os = "linux")] + let netns = if network_enabled && !sidecar_network_enforcement { + openshell_supervisor_process::netns::create_netns_for_proxy(&policy)? + } else { + None }; - tokio::spawn(async move { - if let Err(e) = run_policy_poll_loop(poll_ctx).await { - ocsf_emit!( - AppLifecycleBuilder::new(ocsf_ctx()) - .activity(ActivityId::Fail) - .severity(SeverityId::Medium) - .status(StatusId::Failure) - .message(format!("Policy poll loop exited with error: {e}")) - .build() - ); - } - }); - } + let networking = if network_enabled { + #[cfg(target_os = "linux")] + let proxy_bind_ip = netns + .as_ref() + .map(openshell_supervisor_process::netns::NetworkNamespace::host_ip); + #[cfg(not(target_os = "linux"))] + let proxy_bind_ip: Option = None; - // Start GCE metadata loopback server inside the network namespace so - // Go's cloud.google.com/go/compute/metadata (which bypasses HTTP_PROXY) - // can reach it via direct TCP. Must start before the process leaf so SSH - // sessions also see corrected env vars on bind failure. - #[cfg(target_os = "linux")] - if let Some(ns) = netns.as_ref() - && provider_credentials - .snapshot() - .child_env - .contains_key("GCE_METADATA_HOST") - { - let ctx = google_cloud_metadata::MetadataContext::new(provider_credentials.clone()); - let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); - match ns - .bind_tcp_in_netns(openshell_core::google_cloud::METADATA_LOOPBACK_ADDR) - .await + Some( + openshell_supervisor_network::run::run_networking( + &policy, + proxy_bind_ip, + opa_engine.as_ref(), + retained_proto.as_ref(), + entrypoint_pid.clone(), + process_enabled, + &provider_credentials, + sandbox_id.as_deref(), + sandbox_name_for_agg.as_deref(), + openshell_endpoint_for_proxy.as_deref(), + inference_routes.as_deref(), + denial_tx, + activity_tx, + workspace_rx.clone(), + ) + .await?, + ) + } else { + None + }; + + #[cfg(target_os = "linux")] + let sidecar_control_server = if network_enabled && sidecar_network_enforcement { + if !matches!(policy.network.mode, NetworkMode::Proxy) { + return Err(miette::miette!( + "sidecar network enforcement requires proxy network mode" + )); + } + let socket = sidecar_control_socket().ok_or_else(|| { + miette::miette!( + "{} is required for sidecar topology", + openshell_core::sandbox_env::SIDECAR_CONTROL_SOCKET + ) + })?; + let proto = retained_proto.as_ref().ok_or_else(|| { + miette::miette!( + "sidecar topology requires gateway policy data for the process supervisor" + ) + })?; + let ca_paths = networking.as_ref().and_then(|n| n.ca_file_paths.clone()); + Some(sidecar_control::spawn_server( + &socket, + sidecar_control::BootstrapData { + policy_proto: proto.clone(), + provider_env_revision: provider_credentials.snapshot().revision, + provider_child_env: provider_env.clone(), + proxy_ca_cert_path: ca_paths.as_ref().map(|paths| paths.0.clone()), + proxy_ca_bundle_path: ca_paths.as_ref().map(|paths| paths.1.clone()), + }, + sidecar_expected_peer()?, + )?) + } else { + None + }; + #[cfg(not(target_os = "linux"))] + let sidecar_control_server: Option = None; + + let sidecar_control_publisher = sidecar_control_server + .as_ref() + .map(sidecar_control::ServerHandle::publisher); + + #[cfg(target_os = "linux")] + let mut sidecar_control_task = None; + + #[cfg(target_os = "linux")] + if network_enabled + && sidecar_network_enforcement + && let Some(server) = sidecar_control_server { - Ok(listener) => { - tokio::spawn(metadata_server::run(listener, ctx, ready_tx)); - if let Ok(Ok(addr)) = timeout(Duration::from_secs(5), ready_rx).await { - info!(addr = %addr, "GCE metadata loopback server ready"); - } else { - warn!("GCE metadata server failed to become ready, removing metadata env vars"); + let trusted_ssh_socket_path = ssh_socket_path.clone().ok_or_else(|| { + miette::miette!( + "{} is required for sidecar network topology", + openshell_core::sandbox_env::SSH_SOCKET_PATH + ) + })?; + let (entrypoint_rx, connection_task) = server.into_runtime_parts(); + sidecar_control_task = Some(connection_task); + spawn_sidecar_entrypoint_handler( + entrypoint_rx, + entrypoint_pid.clone(), + opa_engine.clone(), + retained_proto.clone(), + openshell_endpoint.clone(), + sandbox_id.clone(), + std::path::PathBuf::from(trusted_ssh_socket_path), + ); + } + + #[cfg(not(target_os = "linux"))] + if network_enabled && sidecar_network_enforcement { + return Err(miette::miette!( + "sidecar network enforcement is only supported on Linux" + )); + } + + // Spawn background policy poll task (gRPC mode only). Skipped for the + // process-only sidecar container, whose policy arrives over the control + // socket rather than a direct gateway poll. + if !process_uses_sidecar_control + && let (Some(id), Some(endpoint), Some(engine)) = ( + sandbox_id.as_deref(), + openshell_endpoint.as_deref(), + opa_engine.as_ref(), + ) + { + let poll_id = id.to_string(); + let poll_endpoint = endpoint.to_string(); + let poll_engine = engine.clone(); + let poll_ocsf_enabled = ocsf_enabled.clone(); + let poll_pid = entrypoint_pid.clone(); + let poll_provider_credentials = provider_credentials.clone(); + let poll_policy_local = networking.as_ref().map(|n| n.policy_local_ctx.clone()); + let poll_interval_secs: u64 = std::env::var("OPENSHELL_POLICY_POLL_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(10); + let poll_ctx = PolicyPollLoopContext { + endpoint: poll_endpoint, + sandbox_id: poll_id, + opa_engine: poll_engine, + loaded_policy_origin, + entrypoint_pid: poll_pid, + interval_secs: poll_interval_secs, + ocsf_enabled: poll_ocsf_enabled, + provider_credentials: poll_provider_credentials, + policy_local_ctx: poll_policy_local, + middleware_registry_status, + sidecar_control_publisher: sidecar_control_publisher.clone(), + workspace_tx, + }; + + tokio::spawn(async move { + if let Err(e) = run_policy_poll_loop(poll_ctx).await { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::Medium) + .status(StatusId::Failure) + .message(format!("Policy poll loop exited with error: {e}")) + .build() + ); + } + }); + } + + // Start GCE metadata loopback server inside the network namespace so + // Go's cloud.google.com/go/compute/metadata (which bypasses HTTP_PROXY) + // can reach it via direct TCP. Must start before the process leaf so SSH + // sessions also see corrected env vars on bind failure. + #[cfg(target_os = "linux")] + if let Some(ns) = netns.as_ref() + && provider_credentials + .snapshot() + .child_env + .contains_key("GCE_METADATA_HOST") + { + let ctx = google_cloud_metadata::MetadataContext::new(provider_credentials.clone()); + let (ready_tx, ready_rx) = tokio::sync::oneshot::channel(); + match ns + .bind_tcp_in_netns(openshell_core::google_cloud::METADATA_LOOPBACK_ADDR) + .await + { + Ok(listener) => { + tokio::spawn(metadata_server::run(listener, ctx, ready_tx)); + if let Ok(Ok(addr)) = timeout(Duration::from_secs(5), ready_rx).await { + info!(addr = %addr, "GCE metadata loopback server ready"); + } else { + warn!( + "GCE metadata server failed to become ready, removing metadata env vars" + ); + provider_env.remove("GCE_METADATA_HOST"); + provider_env.remove("GCE_METADATA_IP"); + provider_env.remove("METADATA_SERVER_DETECTION"); + provider_credentials.remove_env_key("GCE_METADATA_HOST"); + } + } + Err(e) => { + warn!(error = %e, "GCE metadata server bind failed, Go SDK may not discover credentials"); provider_env.remove("GCE_METADATA_HOST"); provider_env.remove("GCE_METADATA_IP"); provider_env.remove("METADATA_SERVER_DETECTION"); provider_credentials.remove_env_key("GCE_METADATA_HOST"); } } - Err(e) => { - warn!(error = %e, "GCE metadata server bind failed, Go SDK may not discover credentials"); - provider_env.remove("GCE_METADATA_HOST"); - provider_env.remove("GCE_METADATA_IP"); - provider_env.remove("METADATA_SERVER_DETECTION"); - provider_credentials.remove_env_key("GCE_METADATA_HOST"); - } } - } - let process_policy = process_policy_for_topology(&policy, sidecar_network_enforcement)?; - let sidecar_bootstrap_ca_file_paths = sidecar_bootstrap.as_ref().and_then(|bootstrap| { - bootstrap - .proxy_ca_cert_path - .clone() - .zip(bootstrap.proxy_ca_bundle_path.clone()) - }); + let process_policy = process_policy_for_topology(&policy, sidecar_network_enforcement)?; + let sidecar_bootstrap_ca_file_paths = sidecar_bootstrap.as_ref().and_then(|bootstrap| { + bootstrap + .proxy_ca_cert_path + .clone() + .zip(bootstrap.proxy_ca_bundle_path.clone()) + }); - let exit_code = if process_enabled { - let ca_file_paths = networking - .as_ref() - .and_then(|n| n.ca_file_paths.clone()) - .or_else(|| { - if sidecar_network_enforcement { - sidecar_bootstrap_ca_file_paths - .clone() - .or_else(sidecar_ca_file_paths) - } else { - None - } - }); + let exit_code = if process_enabled { + let ca_file_paths = networking + .as_ref() + .and_then(|n| n.ca_file_paths.clone()) + .or_else(|| { + if sidecar_network_enforcement { + sidecar_bootstrap_ca_file_paths + .clone() + .or_else(sidecar_ca_file_paths) + } else { + None + } + }); - let entrypoint_started_tx = - if process_uses_sidecar_control && let Some(writer) = process_control_writer.clone() { + let entrypoint_started_tx = if process_uses_sidecar_control + && let Some(writer) = process_control_writer.clone() + { let (tx, rx) = tokio::sync::oneshot::channel(); tokio::spawn(async move { match rx.await { @@ -683,91 +686,269 @@ pub async fn run_sandbox( None }; - let process = openshell_supervisor_process::run::run_process( - program, - args, - workdir.as_deref(), - timeout_secs, - interactive, + // The workload spawn returns an owned handle (RFC 0012); wait on it + // to obtain the exit code, racing the authoritative sidecar control + // channel so a supervisor disconnect terminates the workload. + let mut agent = openshell_supervisor_process::run::spawn_workload( + program, + args, + workdir.as_deref(), + timeout_secs, + interactive, + sandbox_id.as_deref(), + openshell_endpoint.as_deref(), + ssh_socket_path, + sidecar_network_enforcement, + &process_policy, + process_enforcement_mode, + entrypoint_pid, + entrypoint_started_tx, + provider_credentials, + provider_env, + ca_file_paths, + #[cfg(target_os = "linux")] + netns.as_ref(), + #[cfg(target_os = "linux")] + bypass_denial_tx, + #[cfg(target_os = "linux")] + bypass_activity_tx, + ) + .await?; + + if let Some(control_closed) = process_control_closed.as_mut() { + tokio::select! { + result = agent.wait() => result?, + _ = control_closed => { + ocsf_emit!( + AppLifecycleBuilder::new(ocsf_ctx()) + .activity(ActivityId::Fail) + .severity(SeverityId::High) + .status(StatusId::Failure) + .message( + "Authoritative network-sidecar control channel closed; terminating process container" + ) + .build() + ); + return Err(miette::miette!( + "authoritative network-sidecar control channel closed" + )); + } + } + } else { + agent.wait().await? + } + } else { + // Network-only sidecar mode: keep the proxy and its background + // tasks alive (held via the `networking` value) until shutdown. If + // the sole authenticated process-supervisor control connection + // closes, exit non-zero so Kubernetes restarts the network sidecar + // and creates a fresh one-client bootstrap listener for the + // restarted agent. + #[cfg(target_os = "linux")] + if let Some(control_task) = sidecar_control_task { + tokio::select! { + () = wait_for_shutdown_signal() => 0, + result = control_task => { + warn!(?result, "Authoritative sidecar control channel exited; restarting sidecar"); + 1 + } + } + } else { + wait_for_shutdown_signal().await; + 0 + } + #[cfg(not(target_os = "linux"))] + { + wait_for_shutdown_signal().await; + 0 + } + }; + + // Drop networking explicitly so the proxy + bypass monitor RAII handles + // tear down before we return. + drop(networking); + exit_code + } else { + // Drive the isolation boundary through a runtime-selected backend (RFC + // 0012). The supervisor matches the topology descriptor's backend id to + // the admitted backend, resolves its factory from the registry without + // fallback, and advances the boxed state chain: attach -> Bound -> + // confirm -> Ready -> start_agent -> Running. The transitions consume + // each prior state, so "no untrusted instruction before the boundary is + // ready" is enforced by construction. + let policy_local_slot = Arc::new(std::sync::Mutex::new(None)); + + let config = inpod::InPodConfig { + network_enabled, + process_enabled, + opa_engine: opa_engine.clone(), + retained_proto, + entrypoint_pid: entrypoint_pid.clone(), + provider_credentials: provider_credentials.clone(), + provider_env: std::sync::Mutex::new(provider_env), + process_enforcement_mode, + sandbox_name: sandbox_name_for_agg.clone(), + openshell_endpoint: openshell_endpoint_for_proxy.clone(), + inference_routes, + ssh_socket_path, + workspace_rx: workspace_rx.clone(), + denial_tx: std::sync::Mutex::new(denial_tx), + activity_tx: std::sync::Mutex::new(activity_tx), + #[cfg(target_os = "linux")] + bypass_denial_tx: std::sync::Mutex::new(bypass_denial_tx), + #[cfg(target_os = "linux")] + bypass_activity_tx: std::sync::Mutex::new(bypass_activity_tx), + policy_local_slot: policy_local_slot.clone(), + }; + + // The admitted backend for this sandbox. In-pod today; a real deployment + // derives this id (and the descriptor payload) from trusted deployment + // configuration and admission. + let admitted_backend_id = inpod::IN_POD_BACKEND_ID; + let mut registry = openshell_isolation::contract::BackendRegistry::new(); + registry + .register(Arc::new(inpod::InPodBackendFactory::new(config))) + .map_err(|e| miette::miette!("failed to register in-pod backend: {e}"))?; + + // The topology descriptor. A driver-provisioned sandbox receives it + // through the driver-controlled environment (`OPENSHELL_TOPOLOGY_DESCRIPTOR`) + // and missing or invalid input fails closed with no default backend. A + // standalone run (no sandbox id, so no driver and no admission record) is + // its own deployment configuration and uses the in-pod descriptor directly. + let descriptor = match std::env::var(openshell_core::sandbox_env::TOPOLOGY_DESCRIPTOR) { + Ok(value) => openshell_isolation::contract::TopologyDescriptor::from_env_value(&value) + .map_err(|e| miette::miette!("invalid topology descriptor: {e}"))?, + Err(std::env::VarError::NotUnicode(_)) => { + return Err(miette::miette!( + "invalid topology descriptor: value is not valid UTF-8" + )); + } + Err(std::env::VarError::NotPresent) if sandbox_id.is_none() => { + openshell_isolation::contract::TopologyDescriptor::in_pod() + } + Err(std::env::VarError::NotPresent) => { + return Err(miette::miette!( + "no topology descriptor: a driver-provisioned sandbox must receive {} \ + from its compute driver", + openshell_core::sandbox_env::TOPOLOGY_DESCRIPTOR + )); + } + }; + // Verify the admitted backend id and exact contract version, then + // resolve the factory from the registry without fallback. + let (factory, verified) = registry + .resolve(descriptor, admitted_backend_id) + .map_err(|e| miette::miette!("backend resolution failed: {e}"))?; + + // The trusted sandbox context: it carries the create-time policy + // baseline; authorized mediation-evaluated revisions continue through + // the policy poll loop below without a backend lifecycle transition. + let sandbox_context = openshell_isolation::contract::SandboxContext { + sandbox_id: sandbox_id.clone().unwrap_or_default(), + policy, + agent: openshell_isolation::AgentSpec { + program: program.clone(), + args: args.to_vec(), + workdir, + timeout_secs, + interactive, + }, + }; + + // `attach` validates the payload and atomically binds it to the sandbox + // context: it returns `Bound` or fails closed. + let bound = factory + .attach(verified, sandbox_context) + .await + .map_err(|e| miette::miette!("boundary attach failed: {e}"))?; + + // Retain the boundary's mediation ingress before consuming `Bound`. + let _mediation_ingress = bound.mediation_ingress(); + + // Spawn background policy poll task (gRPC mode only). + if let (Some(id), Some(endpoint), Some(engine)) = ( sandbox_id.as_deref(), openshell_endpoint.as_deref(), - ssh_socket_path, - sidecar_network_enforcement, - &process_policy, - resolved_process_identity, - process_enforcement_mode, - entrypoint_pid, - entrypoint_started_tx, - provider_credentials, - provider_env, - ca_file_paths, - agent_proposals.clone(), - #[cfg(target_os = "linux")] - netns.as_ref(), - #[cfg(target_os = "linux")] - bypass_denial_tx, - #[cfg(target_os = "linux")] - bypass_activity_tx, - ); + opa_engine.as_ref(), + ) { + let poll_id = id.to_string(); + let poll_endpoint = endpoint.to_string(); + let poll_engine = engine.clone(); + let poll_ocsf_enabled = ocsf_enabled.clone(); + let poll_pid = entrypoint_pid.clone(); + let poll_provider_credentials = provider_credentials.clone(); + let poll_policy_local = policy_local_slot + .lock() + .expect("policy_local_slot lock") + .clone(); + let poll_interval_secs: u64 = std::env::var("OPENSHELL_POLICY_POLL_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(10); + let poll_ctx = PolicyPollLoopContext { + endpoint: poll_endpoint, + sandbox_id: poll_id, + opa_engine: poll_engine, + loaded_policy_origin, + entrypoint_pid: poll_pid, + interval_secs: poll_interval_secs, + ocsf_enabled: poll_ocsf_enabled, + provider_credentials: poll_provider_credentials, + policy_local_ctx: poll_policy_local, + middleware_registry_status, + sidecar_control_publisher: None, + workspace_tx, + }; - if let Some(control_closed) = process_control_closed.as_mut() { - tokio::select! { - result = process => result?, - _ = control_closed => { + tokio::spawn(async move { + if let Err(e) = run_policy_poll_loop(poll_ctx).await { ocsf_emit!( AppLifecycleBuilder::new(ocsf_ctx()) .activity(ActivityId::Fail) - .severity(SeverityId::High) + .severity(SeverityId::Medium) .status(StatusId::Failure) - .message( - "Authoritative network-sidecar control channel closed; terminating process container" - ) + .message(format!("Policy poll loop exited with error: {e}")) .build() ); - return Err(miette::miette!( - "authoritative network-sidecar control channel closed" - )); } - } - } else { - process.await? + }); } - } else { - // Network-only sidecar mode: keep the proxy and its background - // tasks alive (held via the `networking` value) until shutdown. If the - // sole authenticated process-supervisor control connection closes, - // exit non-zero so Kubernetes restarts the network sidecar and creates - // a fresh one-client bootstrap listener for the restarted agent. - #[cfg(target_os = "linux")] - if let Some(control_task) = sidecar_control_task { - tokio::select! { - () = wait_for_shutdown_signal() => 0, - result = control_task => { - warn!(?result, "Authoritative sidecar control channel exited; restarting sidecar"); - 1 - } - } - } else { - wait_for_shutdown_signal().await; - 0 - } - #[cfg(not(target_os = "linux"))] - { - wait_for_shutdown_signal().await; - 0 - } - }; - // Drop networking explicitly so the proxy + bypass monitor RAII - // handles tear down before we return. - drop(networking); + // Confirm standing enforcement, then make the agent runnable behind the + // boundary. In network-only (sidecar) mode the backend holds the + // boundary open until a shutdown signal instead of running an agent. + let ready = bound + .confirm() + .await + .map_err(|e| miette::miette!("boundary confirm failed: {e}"))?; + let running = ready + .start_agent() + .await + .map_err(|e| miette::miette!("agent start failed: {e}"))?; + + // Wait for the agent (or the held-open boundary) through its process handle. + let exit_status = running + .agent() + .wait() + .await + .map_err(|e| miette::miette!("waiting on agent failed: {e}"))?; + let exit_code = match exit_status { + openshell_isolation::contract::BoundaryExitStatus::Exited(code) => code, + openshell_isolation::contract::BoundaryExitStatus::Signaled(sig) => 128 + sig, + }; + + // Drop the running boundary so the backend reclaims its private state + // (proxy + bypass monitor and the network namespace) before we return. + drop(running); + exit_code + }; Ok(exit_code) } /// Wait for SIGINT or SIGTERM. Used in network-only mode where there is /// no entrypoint child whose lifetime drives the supervisor's exit. -async fn wait_for_shutdown_signal() { +pub(crate) async fn wait_for_shutdown_signal() { #[cfg(unix)] { use tokio::signal::unix::{SignalKind, signal}; @@ -981,7 +1162,13 @@ fn spawn_sidecar_entrypoint_handler( endpoint.clone(), id.clone(), trusted_ssh_socket_path.clone(), - None, + // The sidecar process supervisor connects from its own + // network namespace (no boundary netns to enter). + Arc::new( + openshell_supervisor_process::boundary_io::NetnsPortForward { + netns_fd: None, + }, + ), Some(supervisor_pid), Arc::clone(&terminating), ); diff --git a/crates/openshell-supervisor-network/Cargo.toml b/crates/openshell-supervisor-network/Cargo.toml index 3dead2b8b..9026542b4 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" } +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 000000000..456772ed8 --- /dev/null +++ b/crates/openshell-supervisor-network/src/identity_source.rs @@ -0,0 +1,126 @@ +// 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 exact accepted connection, and 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. The result type lives in the lower +//! `openshell-isolation` crate (network -> isolation -> core, acyclic). +//! +//! Adoption note: the proxy hot path still resolves identity inline through +//! [`BinaryIdentityCache`](crate::identity::BinaryIdentityCache). Routing the +//! live accept path through `MediationIngress` (so every connection carries +//! this resolver's result) is the remaining live-adoption refactor. + +use std::sync::Arc; +use std::sync::atomic::AtomicU32; + +use openshell_isolation::contract::{BinaryIdentity, ResolveError}; + +/// In-pod binary-identity resolver: reads and hashes the connecting binary from +/// procfs. Resolution fails closed; it never fabricates identity fields. +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 a workload connection, keyed by + /// its workload-side TCP peer port. + pub fn resolve_peer_port(&self, peer_port: u16) -> 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(peer_port) + } + #[cfg(not(target_os = "linux"))] + { + let _ = peer_port; + Err(ResolveError::Failed( + "no procfs on this platform; identity resolution unavailable".to_string(), + )) + } + } +} + +#[cfg(target_os = "linux")] +impl ProcfsIdentityResolver { + fn resolve_via_procfs(&self, peer_port: u16) -> 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 (binary_path, owner_pid) = + crate::procfs::resolve_tcp_peer_identity(entrypoint_pid, peer_port) + .map_err(|_| ResolveError::NotFound)?; + + // 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_sha256 = match crate::procfs::file_sha256(&exe) { + Ok(digest) => Some(digest), + Err(_) => { + return Err(ResolveError::Failed( + "could not hash connecting binary; refusing to assert identity".to_string(), + )); + } + }; + + let ancestors = crate::procfs::collect_ancestor_binaries(owner_pid, entrypoint_pid); + let cmdline_paths = crate::procfs::cmdline_absolute_paths(owner_pid); + + Ok(BinaryIdentity { + binary_path, + binary_sha256, + 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_sha256.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_peer_port(12345))); + } + + #[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_peer_port(1))); + } +} diff --git a/crates/openshell-supervisor-network/src/lib.rs b/crates/openshell-supervisor-network/src/lib.rs index f5d0205e3..e81ed02d5 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 575923d4c..e4012af36 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 } hex = "0.4" miette = { workspace = true } 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 000000000..18f2620eb --- /dev/null +++ b/crates/openshell-supervisor-process/src/boundary_io.rs @@ -0,0 +1,90 @@ +// 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, BoundaryConn, BoundaryPortForward, LoopbackTarget, +}; + +/// 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. + pub netns_fd: Option, +} + +#[async_trait] +impl BoundaryPortForward for NetnsPortForward { + async fn connect(&self, target: LoopbackTarget) -> Result { + let addr = format!("{}:{}", target.host(), target.port()); + let stream = crate::ssh::connect_in_netns(&addr, self.netns_fd) + .await + .map_err(|e| BackendError::Process(format!("port-forward connect to {addr}: {e}")))?; + 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 { netns_fd: 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 { netns_fd: None }; + let target = LoopbackTarget::new(Ipv4Addr::LOCALHOST.into(), addr.port()).unwrap(); + assert!(forward_one(&pf, target).await); + } +} diff --git a/crates/openshell-supervisor-process/src/lib.rs b/crates/openshell-supervisor-process/src/lib.rs index ca9323092..a2e5bc89f 100644 --- a/crates/openshell-supervisor-process/src/lib.rs +++ b/crates/openshell-supervisor-process/src/lib.rs @@ -8,6 +8,7 @@ //! and log push. Populated by follow-up commits as modules migrate out of //! `openshell-sandbox`. +pub mod boundary_io; pub mod child_env; pub mod debug_rpc; #[cfg(unix)] diff --git a/crates/openshell-supervisor-process/src/process.rs b/crates/openshell-supervisor-process/src/process.rs index 93ab4787b..db44fda41 100644 --- a/crates/openshell-supervisor-process/src/process.rs +++ b/crates/openshell-supervisor-process/src/process.rs @@ -120,6 +120,9 @@ const SUPERVISOR_ONLY_ENV_VARS: &[&str] = &[ openshell_core::sandbox_env::TLS_CERT, openshell_core::sandbox_env::TLS_KEY, openshell_core::sandbox_env::PROVIDER_SPIFFE_WORKLOAD_API_SOCKET, + // Supervisor-only bootstrap state (RFC 0012): never exposed to the agent, + // exec sessions, or any other process inside the boundary. + openshell_core::sandbox_env::TOPOLOGY_DESCRIPTOR, ]; pub fn is_supervisor_only_env_var(key: &str) -> bool { diff --git a/crates/openshell-supervisor-process/src/run.rs b/crates/openshell-supervisor-process/src/run.rs index a5ff0456c..2ec3b2061 100644 --- a/crates/openshell-supervisor-process/src/run.rs +++ b/crates/openshell-supervisor-process/src/run.rs @@ -42,15 +42,20 @@ fn ocsf_ctx() -> &'static openshell_ocsf::SandboxContext { openshell_ocsf::ctx::ctx() } -/// Spawn the workload entrypoint, wire up SSH and supervisor session, and -/// wait for the entrypoint child to exit. +/// Spawn the workload entrypoint behind the boundary, wire up SSH and the +/// supervisor session, and return an owned [`SpawnedAgent`] handle. +/// +/// The agent keeps running after this returns; the caller drives it through +/// [`SpawnedAgent::wait`]/[`SpawnedAgent::signal`]. This is the placement- +/// sensitive spawn the in-pod backend's `RunningBoundary` owns (RFC 0012): +/// the returned handle, not an exit code, is the process control surface. /// /// # Errors /// -/// Returns an error if SSH server startup fails, if the entrypoint child -/// fails to spawn, or if waiting for the child returns an OS error. +/// Returns an error if SSH server startup fails or if the entrypoint child +/// fails to spawn. #[allow(clippy::too_many_arguments, clippy::implicit_hasher)] -pub async fn run_process( +pub async fn spawn_workload( program: &str, args: &[String], workdir: Option<&str>, @@ -74,7 +79,7 @@ pub async fn run_process( tokio::sync::mpsc::UnboundedSender, >, #[cfg(target_os = "linux")] bypass_activity_tx: Option, -) -> Result { +) -> Result { // Platform drivers with a resolved numeric UID/GID retain the legacy // account-file update. OCI-image identity leaves those environment values // empty, so the image's account files remain unchanged. @@ -116,7 +121,7 @@ pub async fn run_process( // proxy. Spawn it before the entrypoint child so the first packets are // not missed. Best-effort: returns None when dmesg is unavailable. #[cfg(target_os = "linux")] - let _bypass_handle = netns.and_then(|ns| { + let bypass_handle = netns.and_then(|ns| { crate::bypass_monitor::spawn( ns.name().to_string(), entrypoint_pid.clone(), @@ -238,6 +243,13 @@ pub async fn run_process( let (ssh_ready_tx, ssh_ready_rx) = tokio::sync::oneshot::channel(); + // Inject the in-pod loopback port-forward (RFC 0012). The SSH server + // drives it through the `BoundaryPortForward` interface, so a delegated + // backend would supply a different implementation without the SSH + // server changing. + let port_forward: Arc = + Arc::new(crate::boundary_io::NetnsPortForward { netns_fd }); + tokio::spawn(async move { if let Err(err) = crate::ssh::run_ssh_server( listen_path, @@ -252,6 +264,7 @@ pub async fn run_process( resolved_process_identity, enforcement_mode, shared_ssh_socket, + port_forward, ) .await { @@ -308,7 +321,9 @@ pub async fn run_process( endpoint.to_string(), id.to_string(), socket.clone(), - ssh_netns_fd, + Arc::new(crate::boundary_io::NetnsPortForward { + netns_fd: ssh_netns_fd, + }), None, Arc::clone(&supervisor_terminating), ); @@ -316,7 +331,7 @@ pub async fn run_process( } #[cfg(target_os = "linux")] - let mut handle = ProcessHandle::spawn( + let handle = ProcessHandle::spawn( program, args, workdir, @@ -330,7 +345,7 @@ pub async fn run_process( )?; #[cfg(not(target_os = "linux"))] - let mut handle = ProcessHandle::spawn( + let handle = ProcessHandle::spawn( program, args, workdir, @@ -360,49 +375,173 @@ pub async fn run_process( .build() ); - let outcome = - wait_for_process_exit_or_shutdown(&mut handle, timeout_secs, &supervisor_terminating) - .await?; + Ok(SpawnedAgent { + handle, + timeout_secs, + supervisor_terminating, + #[cfg(target_os = "linux")] + _bypass_handle: bypass_handle, + }) +} - let status = match outcome { - ProcessWaitOutcome::Exited(status) => status, - ProcessWaitOutcome::TimedOut => { - ocsf_emit!( - ProcessActivityBuilder::new(ocsf_ctx()) - .activity(ActivityId::Close) - .action(ActionId::Denied) - .disposition(DispositionId::Blocked) - .severity(SeverityId::Critical) - .status(StatusId::Failure) - .message("Process timed out, killing") - .build() - ); - return Ok(124); // Standard timeout exit code +/// An owned, running workload entrypoint plus the background guards whose +/// lifetime is tied to it (the bypass monitor). +/// +/// The in-pod `RunningBoundary` owns this; dropping it kills the child via the +/// handle's `kill_on_drop`. +pub struct SpawnedAgent { + handle: ProcessHandle, + timeout_secs: u64, + supervisor_terminating: Arc, + #[cfg(target_os = "linux")] + _bypass_handle: Option>, +} + +impl SpawnedAgent { + /// The host PID of the entrypoint, for diagnostics only. + #[must_use] + pub fn pid(&self) -> u32 { + self.handle.pid() + } + + /// A lock-free signaling handle derived from the entrypoint's pid. + /// + /// Separated from the waitable [`SpawnedAgent`] so a signal can be delivered + /// while another task holds the agent to await it: the running boundary + /// keeps the agent behind a mutex for `wait`, but signals go through this + /// pid-based handle and never contend for that lock. + #[must_use] + pub fn signaler(&self) -> AgentSignaler { + AgentSignaler { + pid: self.handle.pid(), } - ProcessWaitOutcome::ShutdownSignal { signal, status } => { - info!( - signal, - exit_code = status.code(), - "Entrypoint exited after supervisor shutdown signal" - ); - status - } - }; - supervisor_terminating.store(true, Ordering::Release); + } - ocsf_emit!( - ProcessActivityBuilder::new(ocsf_ctx()) - .activity(ActivityId::Close) - .action(ActionId::Allowed) - .disposition(DispositionId::Allowed) - .severity(SeverityId::Informational) - .status(StatusId::Success) - .exit_code(status.code()) - .message(format!("Process exited with code {}", status.code())) - .build() - ); + /// Wait for the entrypoint to exit, applying the policy timeout. Returns the + /// exit code (128 + signal if signaled, 124 on timeout-kill). + /// + /// # Errors + /// + /// Returns an error if waiting on the child returns an OS error. + pub async fn wait(&mut self) -> Result { + let outcome = wait_for_process_exit_or_shutdown( + &mut self.handle, + self.timeout_secs, + &self.supervisor_terminating, + ) + .await?; - Ok(status.code()) + let status = match outcome { + ProcessWaitOutcome::Exited(status) => status, + ProcessWaitOutcome::TimedOut => { + ocsf_emit!( + ProcessActivityBuilder::new(ocsf_ctx()) + .activity(ActivityId::Close) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(SeverityId::Critical) + .status(StatusId::Failure) + .message("Process timed out, killing") + .build() + ); + return Ok(124); // Standard timeout exit code + } + ProcessWaitOutcome::ShutdownSignal { signal, status } => { + info!( + signal, + exit_code = status.code(), + "Entrypoint exited after supervisor shutdown signal" + ); + status + } + }; + + ocsf_emit!( + ProcessActivityBuilder::new(ocsf_ctx()) + .activity(ActivityId::Close) + .action(ActionId::Allowed) + .disposition(DispositionId::Allowed) + .severity(SeverityId::Informational) + .status(StatusId::Success) + .exit_code(status.code()) + .message(format!("Process exited with code {}", status.code())) + .build() + ); + + Ok(status.code()) + } + + /// Send a signal to the entrypoint process. + /// + /// # Errors + /// + /// Returns an error if the signal cannot be delivered. + #[cfg(unix)] + pub fn signal(&self, sig: nix::sys::signal::Signal) -> Result<()> { + self.handle.signal(sig) + } + + /// Terminate the entrypoint (SIGTERM, then SIGKILL). + /// + /// # Errors + /// + /// Returns an error if the process cannot be killed. + pub fn kill(&mut self) -> Result<()> { + self.handle.kill() + } +} + +/// A lock-free, pid-based signaling handle to a spawned agent. +/// +/// Delivers signals to the entrypoint process without holding the waitable +/// handle's lock, so a signal and an in-flight `wait` never deadlock. +/// Placement-neutral signal mapping (e.g. RFC 0012's `BoundarySignal`) is the +/// caller's job; this handle exposes only the concrete deliveries so `nix` +/// stays in this crate. +#[derive(Clone, Copy)] +pub struct AgentSignaler { + pid: u32, +} + +#[cfg(unix)] +impl AgentSignaler { + fn deliver(self, sig: nix::sys::signal::Signal) -> Result<()> { + use nix::unistd::Pid; + let pid = i32::try_from(self.pid).unwrap_or(i32::MAX); + nix::sys::signal::kill(Pid::from_raw(pid), sig).into_diagnostic() + } + + /// Send `SIGTERM`. + /// + /// # Errors + /// Returns an error if the signal cannot be delivered. + pub fn term(self) -> Result<()> { + self.deliver(nix::sys::signal::Signal::SIGTERM) + } + + /// Send `SIGKILL`. + /// + /// # Errors + /// Returns an error if the signal cannot be delivered. + pub fn kill(self) -> Result<()> { + self.deliver(nix::sys::signal::Signal::SIGKILL) + } + + /// Send `SIGINT`. + /// + /// # Errors + /// Returns an error if the signal cannot be delivered. + pub fn interrupt(self) -> Result<()> { + self.deliver(nix::sys::signal::Signal::SIGINT) + } + + /// Send `SIGHUP`. + /// + /// # Errors + /// Returns an error if the signal cannot be delivered. + pub fn hangup(self) -> Result<()> { + self.deliver(nix::sys::signal::Signal::SIGHUP) + } } enum ProcessWaitOutcome { diff --git a/crates/openshell-supervisor-process/src/ssh.rs b/crates/openshell-supervisor-process/src/ssh.rs index b1250f990..8f345550d 100644 --- a/crates/openshell-supervisor-process/src/ssh.rs +++ b/crates/openshell-supervisor-process/src/ssh.rs @@ -119,6 +119,7 @@ pub async fn run_ssh_server( resolved_identity: ResolvedProcessIdentity, enforcement_mode: ProcessEnforcementMode, shared_socket: bool, + port_forward: Arc, ) -> Result<()> { let (listener, config, ca_paths) = match ssh_server_init( &listen_path, @@ -149,6 +150,7 @@ pub async fn run_ssh_server( let ca_paths = ca_paths.clone(); let provider_credentials = provider_credentials.clone(); let user_environment = user_environment.clone(); + let port_forward = port_forward.clone(); tokio::spawn(async move { if let Err(err) = handle_connection( @@ -163,6 +165,7 @@ pub async fn run_ssh_server( user_environment, resolved_identity, enforcement_mode, + port_forward, ) .await { @@ -192,6 +195,7 @@ async fn handle_connection( user_environment: HashMap, resolved_identity: ResolvedProcessIdentity, enforcement_mode: ProcessEnforcementMode, + port_forward: Arc, ) -> Result<()> { // Access is gated by the Unix-socket filesystem permissions (root-only), // not by an application-level preface. The supervisor bridges the @@ -217,6 +221,7 @@ async fn handle_connection( user_environment, resolved_identity, enforcement_mode, + port_forward, ); russh::server::run_stream(config, stream, handler) .await @@ -247,6 +252,10 @@ struct SshHandler { user_environment: HashMap, resolved_identity: ResolvedProcessIdentity, enforcement_mode: ProcessEnforcementMode, + /// Loopback port-forward, injected by the orchestrator (RFC 0012). In-pod + /// this connects from inside the workload netns; a delegated backend + /// tunnels into its guest. The handler does not know which. + port_forward: Arc, channels: HashMap, } @@ -262,6 +271,7 @@ impl SshHandler { user_environment: HashMap, resolved_identity: ResolvedProcessIdentity, enforcement_mode: ProcessEnforcementMode, + port_forward: Arc, ) -> Self { Self { policy, @@ -273,6 +283,7 @@ impl SshHandler { user_environment, resolved_identity, enforcement_mode, + port_forward, channels: HashMap::new(), } } @@ -360,11 +371,19 @@ impl russh::server::Handler for SshHandler { // SSH protocol port is bounded by u32 but only u16 is meaningful; // saturate as a guard for malformed clients. let port = u16::try_from(port_to_connect).unwrap_or(u16::MAX); - let netns_fd = self.netns_fd; + + // Build the loopback target up front. The host already passed + // `is_loopback_host`, and `LoopbackTarget::new` re-validates the parsed + // address (defense in depth) before the connect. + let Some(target) = loopback_ip(&host) + .and_then(|ip| openshell_isolation::contract::LoopbackTarget::new(ip, port).ok()) + else { + return Ok(false); + }; + let port_forward = self.port_forward.clone(); tokio::spawn(async move { - let addr = format!("{host}:{port}"); - let tcp = match connect_in_netns(&addr, netns_fd).await { + let mut tcp_stream = match port_forward.connect(target).await { Ok(stream) => stream, Err(err) => { ocsf_emit!( @@ -372,7 +391,9 @@ impl russh::server::Handler for SshHandler { .activity(ActivityId::Fail) .severity(SeverityId::Low) .status(StatusId::Failure) - .message(format!("direct-tcpip: failed to connect to {addr}: {err}")) + .message(format!( + "direct-tcpip: failed to connect to {host}:{port}: {err}" + )) .build() ); let _ = channel.close().await; @@ -381,7 +402,6 @@ impl russh::server::Handler for SshHandler { }; let mut channel_stream = channel.into_stream(); - let mut tcp_stream = tcp; let _ = tokio::io::copy_bidirectional(&mut channel_stream, &mut tcp_stream).await; }); @@ -1338,6 +1358,22 @@ fn is_loopback_host(host: &str) -> bool { } } +/// Resolve a (loopback-validated) destination host string to an `IpAddr`, +/// mapping `localhost` to `127.0.0.1`. +/// +/// Returns `None` for anything that does not parse to an IP, so +/// [`LoopbackTarget::new`] never sees a hostname. +fn loopback_ip(host: &str) -> Option { + let host = host + .strip_prefix('[') + .and_then(|h| h.strip_suffix(']')) + .unwrap_or(host); + if host.eq_ignore_ascii_case("localhost") { + return Some(std::net::Ipv4Addr::LOCALHOST.into()); + } + host.parse().ok() +} + #[cfg(test)] #[allow( clippy::doc_markdown, diff --git a/crates/openshell-supervisor-process/src/supervisor_session.rs b/crates/openshell-supervisor-process/src/supervisor_session.rs index 63fcad4c7..35d589cea 100644 --- a/crates/openshell-supervisor-process/src/supervisor_session.rs +++ b/crates/openshell-supervisor-process/src/supervisor_session.rs @@ -11,8 +11,6 @@ //! selection — it has no protocol awareness of the bytes flowing through. use std::net::IpAddr; -#[cfg(target_os = "linux")] -use std::os::fd::RawFd; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; @@ -23,6 +21,7 @@ use openshell_core::proto::{ SupervisorHello, SupervisorMessage, TcpRelayTarget, gateway_message, relay_open, supervisor_message, }; +use openshell_isolation::contract::{BoundaryPortForward, LoopbackTarget}; use openshell_ocsf::{ ActivityId, ConnectionInfo, Endpoint, NetworkActivityBuilder, OcsfEvent, SandboxContext, SeverityId, StatusId, ocsf_emit, @@ -277,7 +276,7 @@ pub fn spawn( endpoint: String, sandbox_id: String, ssh_socket_path: std::path::PathBuf, - netns_fd: Option, + port_forward: Arc, expected_ssh_peer_pid: Option, terminating: Arc, ) -> tokio::task::JoinHandle<()> { @@ -285,7 +284,7 @@ pub fn spawn( endpoint, sandbox_id, ssh_socket_path, - netns_fd, + port_forward, expected_ssh_peer_pid, terminating, )) @@ -295,7 +294,7 @@ async fn run_session_loop( endpoint: String, sandbox_id: String, ssh_socket_path: std::path::PathBuf, - netns_fd: Option, + port_forward: Arc, expected_ssh_peer_pid: Option, terminating: Arc, ) { @@ -309,7 +308,7 @@ async fn run_session_loop( &endpoint, &sandbox_id, &ssh_socket_path, - netns_fd, + port_forward.clone(), expected_ssh_peer_pid, Arc::clone(&terminating), ) @@ -340,7 +339,7 @@ async fn run_single_session( endpoint: &str, sandbox_id: &str, ssh_socket_path: &std::path::Path, - netns_fd: Option, + port_forward: Arc, expected_ssh_peer_pid: Option, terminating: Arc, ) -> Result<(), Box> { @@ -417,7 +416,7 @@ async fn run_single_session( let context = GatewayMessageContext { sandbox_id, ssh_socket_path, - netns_fd, + &port_forward, expected_ssh_peer_pid, channel: &channel, tx: &tx, @@ -445,7 +444,7 @@ async fn run_single_session( struct GatewayMessageContext<'a> { sandbox_id: &'a str, ssh_socket_path: &'a std::path::Path, - netns_fd: Option, + port_forward: &Arc, expected_ssh_peer_pid: Option, channel: &'a grpc_client::AuthedChannel, tx: &'a mpsc::Sender, @@ -464,7 +463,7 @@ fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext< let channel = context.channel.clone(); let ssh_socket_path = context.ssh_socket_path.to_path_buf(); let tx = context.tx.clone(); - let netns_fd = context.netns_fd; + let port_forward = context.port_forward.clone(); let expected_ssh_peer_pid = context.expected_ssh_peer_pid; let terminating = Arc::clone(context.terminating); @@ -476,7 +475,7 @@ fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext< match handle_relay_open( relay_open, &ssh_socket_path, - netns_fd, + port_forward, expected_ssh_peer_pid, channel, tx, @@ -533,7 +532,7 @@ fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext< async fn handle_relay_open( relay_open: RelayOpen, ssh_socket_path: &std::path::Path, - netns_fd: Option, + port_forward: Arc, expected_ssh_peer_pid: Option, channel: grpc_client::AuthedChannel, tx: mpsc::Sender, @@ -543,7 +542,7 @@ async fn handle_relay_open( let target = match open_target( &relay_open, ssh_socket_path, - netns_fd, + &port_forward, expected_ssh_peer_pid, ) .await @@ -688,11 +687,11 @@ async fn send_relay_open_result( async fn open_target( relay_open: &RelayOpen, ssh_socket_path: &std::path::Path, - netns_fd: Option, + port_forward: &Arc, expected_ssh_peer_pid: Option, ) -> Result, Box> { match relay_open.target.as_ref() { - Some(relay_open::Target::Tcp(target)) => open_tcp_target(target, netns_fd).await, + Some(relay_open::Target::Tcp(target)) => open_tcp_target(target, port_forward).await, Some(relay_open::Target::Ssh(_)) | None => { let runtime_path = crate::unix_socket::runtime_path(ssh_socket_path); let stream = tokio::net::UnixStream::connect(runtime_path.as_ref()).await?; @@ -713,53 +712,27 @@ async fn open_target( async fn open_tcp_target( target: &TcpRelayTarget, - netns_fd: Option, + port_forward: &Arc, ) -> Result, Box> { let host = normalize_tcp_target_host(target)?; let port = u16::try_from(target.port).map_err(|_| "tcp target port must fit in u16")?; - let stream = connect_tcp_target(host, port, netns_fd).await?; + // `normalize_tcp_target_host` returns a loopback IP string; parse it and let + // `LoopbackTarget::new` re-validate before connecting. + let ip: IpAddr = host + .parse() + .map_err(|_| "tcp target host must be a loopback IP")?; + let target = LoopbackTarget::new(ip, port) + .map_err(|e| -> Box { e.to_string().into() })?; + // Connect inside the boundary through the injected port-forward interface + // (RFC 0012). In-pod this enters the workload netns, a delegated backend + // tunnels into its guest. + let stream = port_forward + .connect(target) + .await + .map_err(|e| -> Box { e.to_string().into() })?; Ok(Box::new(stream)) } -#[cfg(target_os = "linux")] -async fn connect_tcp_target( - host: String, - port: u16, - netns_fd: Option, -) -> Result> { - if let Some(fd) = netns_fd { - let (tx, rx) = tokio::sync::oneshot::channel(); - std::thread::spawn(move || { - let result = (|| -> std::io::Result { - #[allow(unsafe_code)] - let rc = unsafe { libc::setns(fd, libc::CLONE_NEWNET) }; - if rc != 0 { - return Err(std::io::Error::last_os_error()); - } - std::net::TcpStream::connect((host.as_str(), port)) - })(); - let _ = tx.send(result); - }); - - let stream = rx - .await - .map_err(|_| "netns tcp connect thread panicked")??; - stream.set_nonblocking(true)?; - return Ok(tokio::net::TcpStream::from_std(stream)?); - } - - Ok(tokio::net::TcpStream::connect((host.as_str(), port)).await?) -} - -#[cfg(not(target_os = "linux"))] -async fn connect_tcp_target( - host: String, - port: u16, - _netns_fd: Option, -) -> Result> { - Ok(tokio::net::TcpStream::connect((host.as_str(), port)).await?) -} - #[cfg(test)] fn validate_tcp_target(target: &TcpRelayTarget) -> Result<(), String> { normalize_tcp_target_host(target).map(|_| ()) @@ -1081,7 +1054,12 @@ mod ocsf_event_tests { }); let relay = ssh_relay_open("peer-check"); - let trusted = open_target(&relay, &socket, None, Some(std::process::id())) + // The SSH relay path does not use the port-forward (that is the TCP + // target path); connect from the supervisor's own namespace. + let port_forward: Arc = + Arc::new(crate::boundary_io::NetnsPortForward { netns_fd: None }); + + let trusted = open_target(&relay, &socket, &port_forward, Some(std::process::id())) .await .expect("matching peer PID should be accepted"); drop(trusted); @@ -1089,7 +1067,7 @@ mod ocsf_event_tests { let Err(err) = open_target( &relay, &socket, - None, + &port_forward, Some(std::process::id().saturating_add(1)), ) .await