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 <jordan.ganoff@docker.com>
This commit is contained in:
Jordan Ganoff
2026-08-23 15:10:34 -07:00
parent 86dc3f21c1
commit ac3cd28864
31 changed files with 3361 additions and 412 deletions
+1
View File
@@ -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 |
Generated
+19
View File
@@ -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",
+16
View File
@@ -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
+11
View File
@@ -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";
@@ -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 }
+1
View File
@@ -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.
@@ -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());
@@ -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 }
@@ -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,
@@ -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"] }
+1
View File
@@ -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
@@ -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
+1
View File
@@ -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" }
+8
View File
@@ -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(),
+22
View File
@@ -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
+675
View File
@@ -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<Self>`), 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<dyn _>` / `Arc<dyn _>`.
//!
//! `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<u8>,
}
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`): `<version>:<backend_id>:<hex payload>`.
/// 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<Self, BackendError> {
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 <version>:<backend_id>:<hex payload>".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<Vec<u8>> {
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<String, Arc<dyn IsolationBackendFactory>>,
}
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<dyn IsolationBackendFactory>,
) -> 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<dyn IsolationBackendFactory>, 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<Box<dyn BoundBoundary>, 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<dyn MediationIngress>;
/// Confirm standing enforcement. How a backend establishes confidence is
/// private to that backend; confirmation fails closed.
async fn confirm(self: Box<Self>) -> Result<Box<dyn ReadyBoundary>, 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<Self>) -> Result<Box<dyn RunningBoundary>, 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<dyn BoundaryProcess>;
/// The in-boundary exec interface.
fn exec(&self) -> Arc<dyn BoundaryExec>;
/// The loopback port-forward interface.
fn port_forward(&self) -> Arc<dyn BoundaryPortForward>;
}
// ============================================================================
// 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<BoundaryExitStatus, BackendError>;
/// 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<u32> {
None
}
}
/// A boxed async writer into a boundary process's stdin.
pub type BoundaryInput = Box<dyn AsyncWrite + Send + Unpin>;
/// A boxed async reader from a boundary process's stdout or stderr.
pub type BoundaryOutput = Box<dyn AsyncRead + Send + Unpin>;
/// 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<dyn BoundaryProcess>,
/// Stdin writer, if not a PTY-merged stream.
pub stdin: Option<BoundaryInput>,
/// Stdout reader.
pub stdout: BoundaryOutput,
/// Stderr reader, distinct from stdout for non-PTY exec.
pub stderr: Option<BoundaryOutput>,
/// PTY control, present when a terminal was requested.
pub terminal: Option<Arc<dyn BoundaryTerminal>>,
}
/// 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<String>,
/// Extra environment over the boundary's base.
pub env: Vec<(String, String)>,
/// Working directory, if any.
pub workdir: Option<String>,
/// 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<ExecSession, BackendError>;
}
// ============================================================================
// 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<Self, BackendError> {
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<T: AsyncRead + AsyncWrite + Send + Unpin> DuplexStream for T {}
/// An open connection into a boundary loopback target.
pub type BoundaryConn = Box<dyn DuplexStream>;
/// 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<BoundaryConn, BackendError>;
}
// ============================================================================
// 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<String>,
/// Ancestor process binaries, nearest first.
pub ancestors: Vec<PathBuf>,
/// Absolute script/interpreter paths drawn from the process cmdlines.
/// Diagnostic context; never authorizes.
pub cmdline_paths: Vec<PathBuf>,
}
/// 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<BinaryIdentity, ResolveError>,
}
/// 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<MediatedConnection, BackendError>;
}
#[cfg(test)]
mod tests;
@@ -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<Vec<BoundarySignal>>,
}
impl MockProcess {
fn new() -> Arc<Self> {
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<BoundaryExitStatus, BackendError> {
// 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<u32> {
Some(4242)
}
}
/// Mediation ingress: hands the mediation service a connection carrying its
/// per-connection identity-resolution result.
struct MockIngress<K>(PhantomData<K>);
#[async_trait]
impl<K: MockKind> MediationIngress for MockIngress<K> {
async fn accept(&self) -> Result<MediatedConnection, BackendError> {
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<MediatedConnection, BackendError> {
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<ExecSession, BackendError> {
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<BoundaryConn, BackendError> {
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<K> {
ingress: Arc<MockIngress<K>>,
}
struct MockReady<K> {
_k: PhantomData<K>,
}
struct MockRunning<K> {
process: Arc<MockProcess>,
exec: Arc<MockExec>,
port_forward: Arc<MockPortForward>,
_k: PhantomData<K>,
}
#[async_trait]
impl<K: MockKind> BoundBoundary for MockBound<K> {
fn mediation_ingress(&self) -> Arc<dyn MediationIngress> {
self.ingress.clone()
}
async fn confirm(self: Box<Self>) -> Result<Box<dyn ReadyBoundary>, BackendError> {
Ok(Box::new(MockReady::<K> { _k: PhantomData }))
}
}
#[async_trait]
impl<K: MockKind> ReadyBoundary for MockReady<K> {
async fn start_agent(self: Box<Self>) -> Result<Box<dyn RunningBoundary>, BackendError> {
Ok(Box::new(MockRunning::<K> {
process: MockProcess::new(),
exec: Arc::new(MockExec),
port_forward: Arc::new(MockPortForward),
_k: PhantomData,
}))
}
}
impl<K: MockKind> RunningBoundary for MockRunning<K> {
fn agent(&self) -> Arc<dyn BoundaryProcess> {
self.process.clone()
}
fn exec(&self) -> Arc<dyn BoundaryExec> {
self.exec.clone()
}
fn port_forward(&self) -> Arc<dyn BoundaryPortForward> {
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<K> {
attached: AtomicBool,
_k: PhantomData<K>,
}
impl<K> MockFactory<K> {
fn new() -> Self {
Self {
attached: AtomicBool::new(false),
_k: PhantomData,
}
}
}
#[async_trait]
impl<K: MockKind> IsolationBackendFactory for MockFactory<K> {
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<Box<dyn BoundBoundary>, 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::<K> {
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<Box<dyn BoundBoundary>, BackendError> {
unreachable!("must never be resolved")
}
}
// ---------------------------------------------------------------------------
// Helpers.
// ---------------------------------------------------------------------------
fn registry() -> BackendRegistry {
let mut reg = BackendRegistry::new();
reg.register(Arc::new(MockFactory::<Primary>::new()))
.expect("register primary");
reg.register(Arc::new(MockFactory::<Secondary>::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<Box<dyn RunningBoundary>, 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::<Primary>::new()))
.expect("first");
let err = reg
.register(Arc::new(MockFactory::<Primary>::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(&reg, descriptor("mock-primary"), "mock-primary")
.await
.expect("primary lifecycle");
let secondary = drive(&reg, 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(&reg, 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(&reg, 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(&reg, 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(&reg, 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::<Secondary>(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());
}
+99
View File
@@ -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<String>,
/// Working directory for the entrypoint, if any.
pub workdir: Option<String>,
/// 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));
}
}
+7
View File
@@ -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 }
+867
View File
@@ -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<Option<_>>` 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<Arc<OpaEngine>>,
pub retained_proto: Option<openshell_core::proto::SandboxPolicy>,
pub entrypoint_pid: Arc<AtomicU32>,
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<HashMap<String, String>>,
/// 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<String>,
pub openshell_endpoint: Option<String>,
pub inference_routes: Option<String>,
pub ssh_socket_path: Option<String>,
/// 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<String>,
/// Proxy-side denial / activity senders (consumed by `attach`).
pub denial_tx: Mutex<Option<UnboundedSender<DenialEvent>>>,
pub activity_tx: Mutex<Option<ActivitySender>>,
/// Bypass-monitor denial / activity senders (consumed by `start_agent`).
#[cfg(target_os = "linux")]
pub bypass_denial_tx: Mutex<Option<UnboundedSender<DenialEvent>>>,
#[cfg(target_os = "linux")]
pub bypass_activity_tx: Mutex<Option<ActivitySender>>,
/// 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<Mutex<Option<Arc<openshell_supervisor_network::policy_local::PolicyLocalContext>>>>,
}
/// 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<Option<InPodConfig>>,
/// 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<Box<dyn BoundBoundary>, 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<InPodBound, BackendError> {
// 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<std::net::IpAddr> = 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<dyn MediationIngress> = 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<NetworkNamespace>,
networking: Option<Networking>,
mediation_ingress: Arc<dyn MediationIngress>,
}
#[async_trait]
impl BoundBoundary for InPodBound {
fn mediation_ingress(&self) -> Arc<dyn MediationIngress> {
self.mediation_ingress.clone()
}
async fn confirm(self: Box<Self>) -> Result<Box<dyn ReadyBoundary>, 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<NetworkNamespace>,
networking: Option<Networking>,
}
#[async_trait]
impl ReadyBoundary for InPodReady {
async fn start_agent(self: Box<Self>) -> Result<Box<dyn RunningBoundary>, 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<std::os::unix::io::RawFd> = None;
let port_forward: Arc<dyn BoundaryPortForward> = Arc::new(NetnsPortForward { netns_fd });
let exec: Arc<dyn BoundaryExec> = 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<dyn BoundaryProcess> = 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<dyn BoundaryProcess>,
exec: Arc<dyn BoundaryExec>,
port_forward: Arc<dyn BoundaryPortForward>,
/// 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<Networking>,
#[cfg(target_os = "linux")]
_netns: Option<NetworkNamespace>,
}
impl RunningBoundary for InPodRunning {
fn agent(&self) -> Arc<dyn BoundaryProcess> {
self.agent.clone()
}
fn exec(&self) -> Arc<dyn BoundaryExec> {
self.exec.clone()
}
fn port_forward(&self) -> Arc<dyn BoundaryPortForward> {
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<u32>,
signaler: Option<AgentSignaler>,
waiter: tokio::sync::Mutex<AgentWaitState>,
}
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<BoundaryExitStatus, BackendError> {
// 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<u32> {
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<ExecSession, BackendError> {
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<MediatedConnection, BackendError> {
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(_)));
}
}
+487 -300
View File
@@ -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<UnboundedSender<DenialEvent>>,
_,
@@ -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<std::net::IpAddr> = 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<sidecar_control::ServerHandle> = 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<std::net::IpAddr> = 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<sidecar_control::ServerHandle> = 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),
);
@@ -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"] }
@@ -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<AtomicU32>,
}
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<BinaryIdentity, ResolveError> {
// 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<BinaryIdentity, ResolveError> {
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/<pid>/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<BinaryIdentity, ResolveError>) -> 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)));
}
}
@@ -9,6 +9,7 @@
//! aggregate them.
pub mod identity;
pub mod identity_source;
pub mod inference_routes;
pub mod l7;
pub mod opa;
@@ -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 }
@@ -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<std::os::unix::io::RawFd>,
}
#[async_trait]
impl BoundaryPortForward for NetnsPortForward {
async fn connect(&self, target: LoopbackTarget) -> Result<BoundaryConn, BackendError> {
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);
}
}
@@ -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)]
@@ -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 {
+188 -49
View File
@@ -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<DenialEvent>,
>,
#[cfg(target_os = "linux")] bypass_activity_tx: Option<ActivitySender>,
) -> Result<i32> {
) -> Result<SpawnedAgent> {
// 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<dyn openshell_isolation::contract::BoundaryPortForward> =
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<AtomicBool>,
#[cfg(target_os = "linux")]
_bypass_handle: Option<tokio::task::JoinHandle<()>>,
}
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<i32> {
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 {
+41 -5
View File
@@ -119,6 +119,7 @@ pub async fn run_ssh_server(
resolved_identity: ResolvedProcessIdentity,
enforcement_mode: ProcessEnforcementMode,
shared_socket: bool,
port_forward: Arc<dyn openshell_isolation::contract::BoundaryPortForward>,
) -> 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<String, String>,
resolved_identity: ResolvedProcessIdentity,
enforcement_mode: ProcessEnforcementMode,
port_forward: Arc<dyn openshell_isolation::contract::BoundaryPortForward>,
) -> 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<String, String>,
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<dyn openshell_isolation::contract::BoundaryPortForward>,
channels: HashMap<ChannelId, ChannelState>,
}
@@ -262,6 +271,7 @@ impl SshHandler {
user_environment: HashMap<String, String>,
resolved_identity: ResolvedProcessIdentity,
enforcement_mode: ProcessEnforcementMode,
port_forward: Arc<dyn openshell_isolation::contract::BoundaryPortForward>,
) -> 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<std::net::IpAddr> {
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,
@@ -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<i32>,
port_forward: Arc<dyn BoundaryPortForward>,
expected_ssh_peer_pid: Option<u32>,
terminating: Arc<AtomicBool>,
) -> 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<i32>,
port_forward: Arc<dyn BoundaryPortForward>,
expected_ssh_peer_pid: Option<u32>,
terminating: Arc<AtomicBool>,
) {
@@ -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<i32>,
port_forward: Arc<dyn BoundaryPortForward>,
expected_ssh_peer_pid: Option<u32>,
terminating: Arc<AtomicBool>,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -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<i32>,
port_forward: &Arc<dyn BoundaryPortForward>,
expected_ssh_peer_pid: Option<u32>,
channel: &'a grpc_client::AuthedChannel,
tx: &'a mpsc::Sender<SupervisorMessage>,
@@ -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<i32>,
port_forward: Arc<dyn BoundaryPortForward>,
expected_ssh_peer_pid: Option<u32>,
channel: grpc_client::AuthedChannel,
tx: mpsc::Sender<SupervisorMessage>,
@@ -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<i32>,
port_forward: &Arc<dyn BoundaryPortForward>,
expected_ssh_peer_pid: Option<u32>,
) -> Result<Box<dyn TargetStream>, Box<dyn std::error::Error + Send + Sync>> {
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<i32>,
port_forward: &Arc<dyn BoundaryPortForward>,
) -> Result<Box<dyn TargetStream>, Box<dyn std::error::Error + Send + Sync>> {
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<dyn std::error::Error + Send + Sync> { 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<dyn std::error::Error + Send + Sync> { e.to_string().into() })?;
Ok(Box::new(stream))
}
#[cfg(target_os = "linux")]
async fn connect_tcp_target(
host: String,
port: u16,
netns_fd: Option<RawFd>,
) -> Result<tokio::net::TcpStream, Box<dyn std::error::Error + Send + Sync>> {
if let Some(fd) = netns_fd {
let (tx, rx) = tokio::sync::oneshot::channel();
std::thread::spawn(move || {
let result = (|| -> std::io::Result<std::net::TcpStream> {
#[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<i32>,
) -> Result<tokio::net::TcpStream, Box<dyn std::error::Error + Send + Sync>> {
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<dyn BoundaryPortForward> =
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