diff --git a/crates/openshell-cli/src/run.rs b/crates/openshell-cli/src/run.rs index eb224685ed..680f877a67 100644 --- a/crates/openshell-cli/src/run.rs +++ b/crates/openshell-cli/src/run.rs @@ -702,6 +702,11 @@ pub async fn sandbox_create( )), await_main_process_attachment, workload_template: template.unwrap_or_default().to_string(), + forward_ports: forward + .as_ref() + .map(|spec| u32::from(spec.port)) + .into_iter() + .collect(), service_exposures: expose .map(|target_port| SandboxServiceExposure { service: String::new(), @@ -1688,6 +1693,33 @@ where println!(" {} {}", "Id:".dimmed(), id); println!(" {} {}", "Name:".dimmed(), name); println!(" {} {}", "Phase:".dimmed(), phase_name(sandbox.phase())); + if let Some(ingress) = sandbox + .status + .as_ref() + .and_then(|status| status.ingress_admission.as_ref()) + { + let ports = if ingress.all_ports { + "all".to_string() + } else { + ingress + .ports + .iter() + .map(u32::to_string) + .collect::>() + .join(", ") + }; + println!( + " {} {}; ports: {ports}", + "Ingress:".dimmed(), + ingress_scope_name(ingress.scope) + ); + } else { + println!( + " {} legacy driver (admission unavailable)", + "Ingress:".dimmed() + ); + } + if let Some(status) = sandbox.status.as_ref() { for condition in &status.conditions { if matches!( @@ -2169,8 +2201,12 @@ pub async fn service_forward_tcp( let (bind_addr, bind_port) = parse_tcp_forward_spec(local, target_port)?; let mut client = grpc_client(server, tls).await?; - let mut replica = fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?; + let (mut replica, scope) = + fetch_ready_sandbox_for_forward(&mut client, name, workspace, target_port).await?; + if scope == openshell_core::proto::IngressScope::SharedHostLoopback { + eprintln!("Target is an admitted shared-host loopback service."); + } let listener = tokio::net::TcpListener::bind((bind_addr.as_str(), bind_port)) .await .into_diagnostic() @@ -2203,7 +2239,7 @@ pub async fn service_forward_tcp( } _ = health_check.tick() => { - replica = fetch_ready_sandbox_for_forward(&mut client, name, workspace).await?; + replica = fetch_ready_sandbox_for_forward(&mut client, name, workspace, target_port).await?.0; } accepted = listener.accept() => { @@ -2274,7 +2310,11 @@ async fn fetch_ready_sandbox_for_forward( client: &mut crate::tls::GrpcClient, name: &str, workspace: &str, -) -> Result> { + target_port: u16, +) -> Result<( + Option, + openshell_core::proto::IngressScope, +)> { let response = match client .get_sandbox(with_owner_hint(GetSandboxRequest { name: name.to_string(), @@ -2307,7 +2347,17 @@ async fn fetch_ready_sandbox_for_forward( )); } - Ok(replica) + let scope = openshell_core::ingress::admit_port( + sandbox + .status + .as_ref() + .and_then(|status| status.ingress_admission.as_ref()), + u32::from(target_port), + false, + ) + .map_err(|error| miette::miette!("{error}"))?; + + Ok((replica, scope)) } #[derive(Debug)] @@ -3012,6 +3062,16 @@ fn configuration_failure_message(sandbox: &Sandbox) -> Option<&str> { .filter(|message| !message.is_empty()) } +fn ingress_scope_name(scope: i32) -> &'static str { + use openshell_core::proto::IngressScope; + match IngressScope::try_from(scope) { + Ok(IngressScope::IsolatedLoopback) => "isolated_loopback", + Ok(IngressScope::SharedHostLoopback) => "shared_host_loopback", + Ok(IngressScope::Unsupported) => "unsupported", + Ok(IngressScope::Unspecified) | Err(_) => "unknown", + } +} + fn sandbox_to_json(sandbox: &Sandbox) -> serde_json::Value { use openshell_core::proto::ConfigurationAdmissionState; @@ -3123,6 +3183,11 @@ fn sandbox_to_json(sandbox: &Sandbox) -> serde_json::Value { "conditions": conditions, "endpoint_statuses": endpoint_statuses, "last_token_grant_failure": last_token_grant_failure, + "ingress_admission": sandbox.status.as_ref().and_then(|status| status.ingress_admission.as_ref()).map(|ingress| serde_json::json!({ + "scope": ingress_scope_name(ingress.scope), + "all_ports": ingress.all_ports, + "ports": ingress.ports, + })), "configuration_admission": admission, "provisioning": provisioning, "created_from_workload_template": created_from_workload_template, @@ -8829,6 +8894,37 @@ mod tests { .error .clear(); assert_eq!(super::configuration_failure_message(&sandbox), None); + assert!(super::sandbox_to_json(&sandbox)["ingress_admission"].is_null()); + for (scope, expected) in [ + ( + openshell_core::proto::IngressScope::IsolatedLoopback as i32, + "isolated_loopback", + ), + ( + openshell_core::proto::IngressScope::SharedHostLoopback as i32, + "shared_host_loopback", + ), + ( + openshell_core::proto::IngressScope::Unsupported as i32, + "unsupported", + ), + (0, "unknown"), + (99, "unknown"), + ] { + sandbox.status.as_mut().unwrap().ingress_admission = + Some(openshell_core::proto::SandboxIngressAdmission { + scope, + ports: vec![8080], + all_ports: false, + }); + let output = super::sandbox_to_json(&sandbox); + assert_eq!(output["ingress_admission"]["scope"], expected); + assert_eq!( + output["ingress_admission"]["ports"], + serde_json::json!([8080]) + ); + assert_eq!(output["ingress_admission"]["all_ports"], false); + } } #[test] diff --git a/crates/openshell-cli/src/ssh.rs b/crates/openshell-cli/src/ssh.rs index 5eb6738ddd..5f4dffabd0 100644 --- a/crates/openshell-cli/src/ssh.rs +++ b/crates/openshell-cli/src/ssh.rs @@ -110,6 +110,7 @@ struct SshSessionConfig { gateway_url: String, token: String, main_terminal: bool, + ingress_admission: Option, replica: Option, } @@ -202,6 +203,10 @@ async fn ssh_session_config( ); Ok(SshSessionConfig { + ingress_admission: sandbox + .status + .as_ref() + .and_then(|status| status.ingress_admission.clone()), proxy_command, sandbox_id: session.sandbox_id.clone(), sandbox_name: name.to_string(), @@ -741,9 +746,18 @@ pub async fn sandbox_forward( tls: &TlsOptions, workspace: &str, ) -> Result<()> { - openshell_core::forward::check_port_available(spec)?; - let session = ssh_session_config(server, name, tls, workspace, None).await?; + let scope = openshell_core::ingress::admit_port( + session.ingress_admission.as_ref(), + u32::from(spec.port), + false, + ) + .map_err(|error| miette::miette!("{error}"))?; + if scope == openshell_core::proto::IngressScope::SharedHostLoopback { + eprintln!("Target is an admitted shared-host loopback service."); + } + + openshell_core::forward::check_port_available(spec)?; let mut command = TokioCommand::from(ssh_forward_command( &session.proxy_command, @@ -2557,6 +2571,7 @@ mod tests { async move { if setup { let session = SshSessionConfig { + ingress_admission: None, proxy_command: String::new(), sandbox_id: String::new(), sandbox_name: String::new(), diff --git a/crates/openshell-core/src/extension_protocol.rs b/crates/openshell-core/src/extension_protocol.rs index d6d93418d8..b2515cab18 100644 --- a/crates/openshell-core/src/extension_protocol.rs +++ b/crates/openshell-core/src/extension_protocol.rs @@ -28,6 +28,8 @@ pub const COMPUTE_SUSPEND_RESUME: &str = "openshell.compute.suspend-resume"; /// Optional stopped-state workload configuration updates retaining the resource and storage. pub const COMPUTE_UPDATE_SANDBOX_CONFIGURATION: &str = "openshell.compute.update-sandbox-configuration"; +/// Immutable per-sandbox forwarding and service exposure admission. +pub const COMPUTE_INGRESS_ADMISSION: &str = "openshell.compute.ingress-admission"; const MAX_IMPLEMENTATION_NAME_BYTES: usize = 128; const MAX_IMPLEMENTATION_VERSION_BYTES: usize = 128; @@ -133,6 +135,7 @@ pub fn gateway_metadata(family: ExtensionFamily) -> PeerMetadata { supported_capabilities.push(COMPUTE_LAUNCH_AUTHENTICATION.to_string()); supported_capabilities.push(COMPUTE_ISOLATION_BACKEND.to_string()); supported_capabilities.push(COMPUTE_UPDATE_SANDBOX_CONFIGURATION.to_string()); + supported_capabilities.push(COMPUTE_INGRESS_ADMISSION.to_string()); } PeerMetadata { protocol_version: Some(ProtocolVersion { diff --git a/crates/openshell-core/src/ingress.rs b/crates/openshell-core/src/ingress.rs new file mode 100644 index 0000000000..b22f4e4205 --- /dev/null +++ b/crates/openshell-core/src/ingress.rs @@ -0,0 +1,61 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Shared validation of compute-admitted ingress; the backend enforces targets. + +use crate::proto::{IngressScope, SandboxIngressAdmission}; +use std::collections::BTreeSet; + +/// Validate a negotiated descriptor without treating unknown evidence as legacy. +pub fn validate(admission: &SandboxIngressAdmission) -> Result { + let scope = + IngressScope::try_from(admission.scope).map_err(|_| "unknown compute ingress scope")?; + if scope == IngressScope::Unspecified { + return Err( + "compute ingress admission is unavailable; recreate the sandbox with an ingress-aware driver", + ); + } + if (admission.all_ports && !admission.ports.is_empty()) + || (scope == IngressScope::Unsupported + && (admission.all_ports || !admission.ports.is_empty())) + || admission.ports.len() > 65535 + { + return Err("invalid compute ingress envelope"); + } + let mut seen = BTreeSet::new(); + for port in &admission.ports { + if !(1..=65535).contains(port) || !seen.insert(port) { + return Err("compute ingress ports must be unique and between 1 and 65535"); + } + } + Ok(scope) +} + +/// Authorize before listeners, service persistence, or any pooled upstream use. +/// Legacy absence is permitted only when the gateway has not negotiated ingress. +pub fn admit_port( + admission: Option<&SandboxIngressAdmission>, + port: u32, + required: bool, +) -> Result { + if !(1..=65535).contains(&port) { + return Err("target port must be between 1 and 65535"); + } + let Some(admission) = admission else { + return if required { + Err("compute ingress admission is missing; recreate the sandbox") + } else { + Ok(IngressScope::Unspecified) + }; + }; + let scope = validate(admission)?; + if scope == IngressScope::Unsupported { + return Err("the compute driver does not support sandbox ingress"); + } + if !admission.all_ports && !admission.ports.contains(&port) { + return Err( + "target port was not admitted by the compute driver; recreate the sandbox with that service port", + ); + } + Ok(scope) +} diff --git a/crates/openshell-core/src/lib.rs b/crates/openshell-core/src/lib.rs index 12d6cd26ff..0f29280c5e 100644 --- a/crates/openshell-core/src/lib.rs +++ b/crates/openshell-core/src/lib.rs @@ -35,6 +35,7 @@ pub mod gpu; pub mod grpc_client; pub mod host_pattern; pub mod image; +pub mod ingress; pub mod isolation_registration; pub mod jwt; pub mod local_api_socket; diff --git a/crates/openshell-driver-docker/src/lib.rs b/crates/openshell-driver-docker/src/lib.rs index e5421431f8..2cb4a7d987 100644 --- a/crates/openshell-driver-docker/src/lib.rs +++ b/crates/openshell-driver-docker/src/lib.rs @@ -1097,7 +1097,7 @@ impl DockerComputeDriver { openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/docker", openshell_core::VERSION, - [], + [openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION.to_string()], )), } } @@ -3357,7 +3357,13 @@ impl ComputeDriver for DockerComputeDriver { CdiGpuDefaultSelector::peek_device_ids, ) .await?; - Ok(Response::new(ValidateSandboxCreateResponse {})) + Ok(Response::new(ValidateSandboxCreateResponse { + ingress_admission: Some(openshell_core::proto::SandboxIngressAdmission { + scope: openshell_core::proto::IngressScope::IsolatedLoopback as i32, + all_ports: true, + ports: Vec::new(), + }), + })) } async fn get_sandbox( diff --git a/crates/openshell-driver-kubernetes/src/driver.rs b/crates/openshell-driver-kubernetes/src/driver.rs index e1a4226187..410f87143e 100644 --- a/crates/openshell-driver-kubernetes/src/driver.rs +++ b/crates/openshell-driver-kubernetes/src/driver.rs @@ -1326,6 +1326,7 @@ impl KubernetesComputeDriver { [ openshell_core::extension_protocol::COMPUTE_UPDATE_SANDBOX_CONFIGURATION .to_string(), + openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION.to_string(), ], )), }) diff --git a/crates/openshell-driver-kubernetes/src/grpc.rs b/crates/openshell-driver-kubernetes/src/grpc.rs index 6ce79654ef..3760ba29dc 100644 --- a/crates/openshell-driver-kubernetes/src/grpc.rs +++ b/crates/openshell-driver-kubernetes/src/grpc.rs @@ -102,7 +102,13 @@ impl ComputeDriver for ComputeDriverService { .sandbox .ok_or_else(|| Status::invalid_argument("sandbox is required"))?; self.driver.validate_sandbox_create(&sandbox).await?; - Ok(Response::new(ValidateSandboxCreateResponse {})) + Ok(Response::new(ValidateSandboxCreateResponse { + ingress_admission: Some(openshell_core::proto::SandboxIngressAdmission { + scope: openshell_core::proto::IngressScope::IsolatedLoopback as i32, + all_ports: true, + ports: Vec::new(), + }), + })) }) .await } diff --git a/crates/openshell-driver-mxc/src/driver.rs b/crates/openshell-driver-mxc/src/driver.rs index 2772f1a69c..e49fdda56f 100644 --- a/crates/openshell-driver-mxc/src/driver.rs +++ b/crates/openshell-driver-mxc/src/driver.rs @@ -480,7 +480,7 @@ impl MxcComputeBackend { openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/mxc", openshell_core::VERSION, - [], + [openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION.to_string()], )), } } diff --git a/crates/openshell-driver-mxc/src/grpc.rs b/crates/openshell-driver-mxc/src/grpc.rs index ca834987cb..071191de71 100644 --- a/crates/openshell-driver-mxc/src/grpc.rs +++ b/crates/openshell-driver-mxc/src/grpc.rs @@ -70,7 +70,13 @@ impl ComputeDriver for ComputeDriverService { .sandbox .ok_or_else(|| Status::invalid_argument("sandbox is required"))?; self.backend.validate_sandbox_create(&sandbox)?; - Ok(Response::new(ValidateSandboxCreateResponse {})) + Ok(Response::new(ValidateSandboxCreateResponse { + ingress_admission: Some(openshell_core::proto::SandboxIngressAdmission { + scope: openshell_core::proto::IngressScope::Unsupported as i32, + all_ports: false, + ports: Vec::new(), + }), + })) } async fn get_sandbox( diff --git a/crates/openshell-driver-podman/src/driver.rs b/crates/openshell-driver-podman/src/driver.rs index 6bea5b7f16..31a9f6ff3e 100644 --- a/crates/openshell-driver-podman/src/driver.rs +++ b/crates/openshell-driver-podman/src/driver.rs @@ -744,7 +744,7 @@ impl PodmanComputeDriver { openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/podman", openshell_core::VERSION, - [], + [openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION.to_string()], )), }) } diff --git a/crates/openshell-driver-podman/src/grpc.rs b/crates/openshell-driver-podman/src/grpc.rs index 5318b3f833..5c9cc885e9 100644 --- a/crates/openshell-driver-podman/src/grpc.rs +++ b/crates/openshell-driver-podman/src/grpc.rs @@ -100,7 +100,13 @@ impl ComputeDriver for ComputeDriverService { .validate_sandbox_create(&sandbox) .await .map_err(Status::from)?; - Ok(Response::new(ValidateSandboxCreateResponse {})) + Ok(Response::new(ValidateSandboxCreateResponse { + ingress_admission: Some(openshell_core::proto::SandboxIngressAdmission { + scope: openshell_core::proto::IngressScope::IsolatedLoopback as i32, + all_ports: true, + ports: Vec::new(), + }), + })) }) .await } diff --git a/crates/openshell-driver-vm/src/driver.rs b/crates/openshell-driver-vm/src/driver.rs index f7946c34c2..653bc54942 100644 --- a/crates/openshell-driver-vm/src/driver.rs +++ b/crates/openshell-driver-vm/src/driver.rs @@ -1539,7 +1539,7 @@ impl VmDriver { openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/vm", openshell_core::VERSION, - [], + [openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION.to_string()], ); extension .required_capabilities @@ -5334,7 +5334,13 @@ impl ComputeDriver for VmDriver { .sandbox .ok_or_else(|| Status::invalid_argument("sandbox is required"))?; self.validate_sandbox(&sandbox)?; - Ok(Response::new(ValidateSandboxCreateResponse {})) + Ok(Response::new(ValidateSandboxCreateResponse { + ingress_admission: Some(openshell_core::proto::SandboxIngressAdmission { + scope: openshell_core::proto::IngressScope::IsolatedLoopback as i32, + all_ports: true, + ports: Vec::new(), + }), + })) } async fn create_sandbox( diff --git a/crates/openshell-gateway-interceptors/src/proto_json.rs b/crates/openshell-gateway-interceptors/src/proto_json.rs index f34c791adc..0763d7bf1d 100644 --- a/crates/openshell-gateway-interceptors/src/proto_json.rs +++ b/crates/openshell-gateway-interceptors/src/proto_json.rs @@ -322,6 +322,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }; let bytes = request.encode_to_vec(); let json = codec diff --git a/crates/openshell-gateway-interceptors/src/runtime.rs b/crates/openshell-gateway-interceptors/src/runtime.rs index 21de44fb9c..c102e106c7 100644 --- a/crates/openshell-gateway-interceptors/src/runtime.rs +++ b/crates/openshell-gateway-interceptors/src/runtime.rs @@ -1094,6 +1094,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }; let bytes = request.encode_to_vec(); diff --git a/crates/openshell-sdk/README.md b/crates/openshell-sdk/README.md index 001b2e9249..da180b8ce2 100644 --- a/crates/openshell-sdk/README.md +++ b/crates/openshell-sdk/README.md @@ -212,3 +212,9 @@ These helpers use the raw client's authentication slot. They do not perform OIDC - Async-only. Tonic is async-native; callers needing a blocking call can wrap with their own runtime. - The curated surface will grow as more RPCs graduate from `raw`. + +## Next minor API migration + +Raw `SandboxStatus` adds optional `ingress_admission`, and raw `CreateSandboxRequest` adds `forward_ports`. Compute-driver `ValidateSandboxCreateResponse` also adds optional `ingress_admission`. Complete generated Rust literals must initialize optional descriptors to `None` and ports to an empty vector, or use `..Default::default()`. + +Upgrade gateway replicas together. If forwarding reports missing admission after an upgrade, recreate the sandbox with the required service ports. diff --git a/crates/openshell-sdk/src/client.rs b/crates/openshell-sdk/src/client.rs index 8241ac49b5..5a195a5d38 100644 --- a/crates/openshell-sdk/src/client.rs +++ b/crates/openshell-sdk/src/client.rs @@ -1481,6 +1481,7 @@ fn create_sandbox_request(spec: SandboxSpec) -> proto::CreateSandboxRequest { command, tty, service_exposures, + forward_ports, restart_policy, } = spec; let template = image.map(|image| proto::SandboxTemplate { @@ -1508,6 +1509,7 @@ fn create_sandbox_request(spec: SandboxSpec) -> proto::CreateSandboxRequest { workspace_scope: Some(proto::workspace_selector("default")), await_main_process_attachment: false, workload_template: String::new(), + forward_ports: forward_ports.into_iter().map(u32::from).collect(), service_exposures: service_exposures .into_iter() .map(|exposure| proto::SandboxServiceExposure { @@ -1533,6 +1535,7 @@ fn create_sandbox_from_template_request( tty, policy, service_exposures, + forward_ports, } = spec; proto::CreateSandboxRequest { request_id: String::new(), @@ -1549,6 +1552,7 @@ fn create_sandbox_from_template_request( workspace_scope: Some(proto::workspace_selector("default")), workload_template: template_name, await_main_process_attachment: false, + forward_ports: forward_ports.into_iter().map(u32::from).collect(), service_exposures: service_exposures .into_iter() .map(|exposure| proto::SandboxServiceExposure { diff --git a/crates/openshell-sdk/src/types.rs b/crates/openshell-sdk/src/types.rs index 571ee42b68..7f6913516e 100644 --- a/crates/openshell-sdk/src/types.rs +++ b/crates/openshell-sdk/src/types.rs @@ -297,6 +297,8 @@ pub struct SandboxSpec { pub tty: bool, /// Loopback HTTP services to expose when the sandbox is created. pub service_exposures: Vec, + /// Target ports to admit before provisioning; does not open listeners. + pub forward_ports: Vec, /// Restart behavior after the canonical main process exits. pub restart_policy: SandboxRestartPolicy, } @@ -348,6 +350,8 @@ pub struct SandboxTemplateCreateSpec { pub tty: bool, /// Loopback HTTP services to expose when the sandbox is created. pub service_exposures: Vec, + /// Target ports to admit before provisioning; does not open listeners. + pub forward_ports: Vec, /// Create-time sandbox policy. The named workload template supplies runtime /// workload fields; policy remains part of the sandbox's governance spec. pub policy: Option, @@ -389,6 +393,8 @@ pub struct SandboxTemplateListOptions { #[derive(Clone, Debug)] #[non_exhaustive] pub struct SandboxRef { + /// Immutable compute-admitted ingress envelope, absent on legacy gateways. + pub ingress_admission: Option, pub id: String, pub name: String, pub workspace: String, @@ -444,6 +450,10 @@ impl SandboxRef { }); let meta = sandbox.metadata.unwrap_or_default(); Self { + ingress_admission: sandbox + .status + .as_ref() + .and_then(|status| status.ingress_admission.clone()), id: meta.id, name: meta.name, workspace: meta.workspace, diff --git a/crates/openshell-sdk/tests/client_mock.rs b/crates/openshell-sdk/tests/client_mock.rs index 1f3f791a08..ab9871e4eb 100644 --- a/crates/openshell-sdk/tests/client_mock.rs +++ b/crates/openshell-sdk/tests/client_mock.rs @@ -1246,6 +1246,7 @@ async fn create_sandbox_passes_spec_through() { image: Some("ghcr.io/foo:bar".to_string()), labels: labels.clone(), gpu: true, + forward_ports: vec![8080], service_exposures: vec![ServiceExposure { service: "web".to_string(), target_port: 8080, @@ -1266,6 +1267,7 @@ async fn create_sandbox_passes_spec_through() { assert_eq!(observed.name, "my-box"); assert_eq!(observed.labels, labels); assert!(observed.annotations.is_empty()); + assert_eq!(observed.forward_ports, vec![8080]); assert_eq!(observed.service_exposures.len(), 1); assert_eq!(observed.service_exposures[0].service, "web"); assert_eq!(observed.service_exposures[0].target_port, 8080); diff --git a/crates/openshell-sdk/tests/sandbox_status_compatibility.rs b/crates/openshell-sdk/tests/sandbox_status_compatibility.rs index 7a080a381f..d7af4d432d 100644 --- a/crates/openshell-sdk/tests/sandbox_status_compatibility.rs +++ b/crates/openshell-sdk/tests/sandbox_status_compatibility.rs @@ -1,14 +1,18 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! Compile-time source-compatibility fixture for the v0.1.5 public status shape. +//! Exhaustive raw status fixture for the documented next-minor API transition. +//! +//! This intentionally advances the v0.1.5 source baseline. Legacy wire payloads +//! still decode; complete raw Rust literals need the new optional field. +//! See the SDK README migration notice. Do not backport into a 0.1.x patch train. use openshell_sdk::raw::proto::SandboxStatus; #[test] -fn baseline_exhaustive_sandbox_status_literal_still_compiles() { - // Intentionally no `..Default::default()`: adding even an optional public - // field breaks existing downstream Rust exhaustive struct literals. +fn next_minor_exhaustive_sandbox_status_literal_compiles() { + // Intentionally no `..Default::default()`: future shape changes must be + // reviewed against this declared minor baseline, never silently accepted. let status = SandboxStatus { agent_pod: String::new(), agent_fd: String::new(), @@ -26,6 +30,7 @@ fn baseline_exhaustive_sandbox_status_literal_still_compiles() { next_restart_time: None, main_process_started_time: None, last_token_grant_failure: None, + ingress_admission: None, }; assert_eq!(status, SandboxStatus::default()); } diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 4ba1311d54..2294d74073 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -691,7 +691,7 @@ struct RemoteComputeDriver { client: RemoteComputeDriverClient, app_endpoints_client: openshell_core::proto::compute::v1::compute_driver_app_endpoints_client::ComputeDriverAppEndpointsClient>, configuration_updates_client: openshell_core::proto::compute::v1::compute_driver_configuration_updates_client::ComputeDriverConfigurationUpdatesClient>, - admission_acknowledgement: Arc>>, + admission_handshake: Arc>>, } type RemoteComputeDriverClient = ComputeDriverClient< @@ -704,7 +704,7 @@ impl RemoteComputeDriver { client: ComputeDriverClient::with_interceptor(channel.clone(), TraceContextInterceptor), app_endpoints_client: openshell_core::proto::compute::v1::compute_driver_app_endpoints_client::ComputeDriverAppEndpointsClient::with_interceptor(channel.clone(), TraceContextInterceptor), configuration_updates_client: openshell_core::proto::compute::v1::compute_driver_configuration_updates_client::ComputeDriverConfigurationUpdatesClient::with_interceptor(channel, TraceContextInterceptor), - admission_acknowledgement: Arc::new(StdMutex::new(None)), + admission_handshake: Arc::new(StdMutex::new(None)), } } @@ -713,8 +713,8 @@ impl RemoteComputeDriver { } async fn verify_admission_policy(&self, external_isolation: bool) -> Result<(), Status> { - let expected = self - .admission_acknowledgement + let (expected, require_ingress) = self + .admission_handshake .lock() .map_err(|_| Status::internal("admission handshake lock poisoned"))? .clone() @@ -735,6 +735,17 @@ impl RemoteComputeDriver { "remote compute driver no longer supports isolation backend registration; restart the gateway after upgrading the driver", )); } + if require_ingress + && !current.extension.as_ref().is_some_and(|metadata| { + metadata.supported_capabilities.iter().any(|capability| { + capability == openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION + }) + }) + { + return Err(Status::failed_precondition( + "remote compute driver no longer supports negotiated ingress admission; restart the gateway after upgrading the driver", + )); + } if current.resource_admission_policy != expected { return Err(Status::failed_precondition( "remote driver admission policy changed; restart gateway after configuring matching policy", @@ -772,10 +783,13 @@ impl ComputeDriver for RemoteComputeDriver { { let mut client = self.client(); let response = client.get_capabilities(request).await?; - self.admission_acknowledgement + self.admission_handshake .lock() .map_err(|_| Status::internal("admission handshake lock poisoned"))? - .get_or_insert_with(|| response.get_ref().resource_admission_policy.clone()); + .get_or_insert_with(|| ( + response.get_ref().resource_admission_policy.clone(), + response.get_ref().extension.as_ref().is_some_and(|metadata| metadata.supported_capabilities.iter().any(|capability| capability == openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION)), + )); Ok(response) } @@ -1487,6 +1501,41 @@ impl ComputeRuntime { } } + pub(crate) fn supports_ingress_admission(&self) -> bool { + self.driver_info + .negotiated_extension + .supported_capabilities + .iter() + .any(|capability| { + capability == openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION + }) + } + + pub(crate) fn admit_ingress_port(&self, sandbox: &Sandbox, port: u32) -> Result<(), Status> { + openshell_core::ingress::admit_port( + sandbox + .status + .as_ref() + .and_then(|status| status.ingress_admission.as_ref()), + port, + self.supports_ingress_admission(), + ) + .map(|_| ()) + .map_err(Status::failed_precondition) + } + + pub(crate) fn project_ingress_admission(&self, sandbox: &mut Sandbox) { + if self.supports_ingress_admission() { + // A retained legacy resource under a modern driver has no evidence. + // Publish explicit unavailable evidence so clients never infer legacy. + sandbox + .status + .get_or_insert_with(Default::default) + .ingress_admission + .get_or_insert_with(Default::default); + } + } + pub async fn validate_sandbox_create(&self, sandbox: &Sandbox) -> Result<(), Status> { self.validate_sandbox_create_with_runtime_inputs( sandbox, @@ -1494,6 +1543,7 @@ impl ComputeRuntime { false, ) .await + .map(|_| ()) } pub(crate) async fn validate_sandbox_create_with_runtime_inputs( @@ -1501,7 +1551,7 @@ impl ComputeRuntime { sandbox: &Sandbox, runtime_inputs: &SandboxCreateRuntimeInputs, await_main_process_attachment: bool, - ) -> Result<(), Status> { + ) -> Result, Status> { self.validate_caller_driver_config( sandbox .spec @@ -1523,7 +1573,8 @@ impl ComputeRuntime { let staged = self.rootfs_tar_staging.peek(&token)?; set_rootfs_tar_path(&mut driver_sandbox, &staged); } - self.driver + let response = self + .driver .call( openshell_otel::rpc::VALIDATE_SANDBOX_CREATE, Some(sandbox.object_id()), @@ -1535,8 +1586,17 @@ impl ComputeRuntime { .await }, ) - .await - .map(|_| ()) + .await? + .into_inner(); + if self.supports_ingress_admission() { + let admission = response.ingress_admission.ok_or_else(|| { + Status::failed_precondition("compute driver omitted negotiated ingress admission") + })?; + openshell_core::ingress::validate(&admission).map_err(Status::failed_precondition)?; + Ok(Some(admission)) + } else { + Ok(None) + } } pub async fn create_sandbox( @@ -1684,6 +1744,26 @@ impl ComputeRuntime { self.isolation_backend_name().into(), ); + if self.supports_ingress_admission() + && sandbox + .status + .as_ref() + .and_then(|status| status.ingress_admission.as_ref()) + .is_none() + { + let admission = self + .validate_sandbox_create_with_runtime_inputs( + &sandbox, + &runtime_inputs, + await_main_process_attachment, + ) + .await?; + sandbox + .status + .get_or_insert_with(Default::default) + .ingress_admission = admission; + } + // Strip the staging token from the public sandbox before anything // persists it: the object store copy is readable by every member of the // workspace, and the token is a bearer credential for the staged @@ -2435,7 +2515,7 @@ impl ComputeRuntime { )?; let mut attempts = 0; - let (previous, starting, launch_authentication) = loop { + let (previous, starting, launch_authentication) = 'start_claim: loop { provisioning_operation::ensure_operation_settled(¤t)?; self.ensure_no_pending_image_update(¤t).await?; let phase = SandboxPhase::try_from(current.phase()).unwrap_or(SandboxPhase::Unknown); @@ -2500,13 +2580,53 @@ impl ComputeRuntime { let authentication = self .encode_persisted_launch_authentication(authority, ¤t) .await?; - let refreshed = self - .store - .get_message::(&sandbox_id) + for _ in 0..START_PHASE_CAS_RETRY_LIMIT { + let refreshed = self + .store + .get_message::(&sandbox_id) + .await + .map_err(|_| Status::unavailable("reload sandbox SSH identity failed"))? + .ok_or_else(|| Status::not_found("sandbox was deleted"))?; + if provisioning_deadline::timed_out(&refreshed) { + return Err(Status::deadline_exceeded( + "provisioning deadline already expired", + )); + } + if sandbox_provisioning_attempt_id(&refreshed).is_some_and(str::is_empty) { + return Err(Status::failed_precondition("provisioning attempt is empty")); + } + provisioning_operation::ensure_operation_settled(&refreshed)?; + if refreshed.phase() != current.phase() + || !provisioning_operation::same_operation(&refreshed, ¤t) + || sandbox_runtime_generation(&refreshed) + != sandbox_runtime_generation(¤t) + { + return Err(Status::aborted("sandbox changed before start claim")); + } + match Box::pin(self.store.update_message_cas::( + &sandbox_id, + sandbox_resource_version(&refreshed), + provisioning_operation::claim_record, + )) .await - .map_err(|_| Status::unavailable("reload sandbox SSH identity failed"))? - .ok_or_else(|| Status::not_found("sandbox was deleted"))?; - break (refreshed.clone(), refreshed, authentication); + { + Ok(claimed) => { + self.sandbox_index.update_from_sandbox(&claimed); + self.sandbox_watch_bus.notify(&sandbox_id); + break 'start_claim (refreshed, claimed, authentication); + } + Err(crate::persistence::PersistenceError::Conflict { .. }) => {} + Err(error) => { + return Err(crate::grpc::persistence_error_to_status( + error, + "claim sandbox start", + )); + } + } + } + return Err(Status::aborted( + "sandbox kept changing before driver dispatch", + )); } let previous = current.clone(); @@ -2540,6 +2660,7 @@ impl ComputeRuntime { self.image_preparation_timeout_seconds, self.provisioning_repair_window_ms, ); + provisioning_operation::claim_record(sandbox); if let Some(metadata) = sandbox.metadata.as_mut() { next_identity_for_update.write(&mut metadata.annotations); write_start_intent( @@ -2620,64 +2741,73 @@ impl ComputeRuntime { lifecycle_guard: SandboxLifecycleGuard, launch_authentication: Vec, ) -> Result { - self.validate_caller_driver_config( - starting - .spec - .as_ref() - .and_then(|spec| spec.template.as_ref()), - )?; - let expected_runtime_identity = sandbox_compute_runtime_identity(&previous); - // One owned operation covers both calls. A NotFound response alone - // does not end ownership while the recovery create can still run. - let recreate = if provisioning_deadline::timed_out(&previous) { - let mut driver_sandbox = self.driver_sandbox_for_launch(&starting)?; - if let Some(spec) = driver_sandbox.spec.as_mut() { - spec.launch_authentication - .clone_from(&launch_authentication); - } - Some(driver_sandbox) - } else { - None - }; - // Starts may need to replace a stopped local workload/supervisor pair. - // The request (including its durable provisioning snapshot) is always - // reconstructed from the gateway's record; never let a driver derive - // provisioning input from mutable runtime inspection output. - let request = - self.driver_start_request(&starting, launch_authentication, expected_runtime_identity)?; + let runtime = self.clone(); + let owned_starting = starting.clone(); + let owned_previous = previous.clone(); let driver = self.driver.clone(); let operation_id = sandbox_id.clone(); - let result = Box::pin(self.await_provisioning_operation(&starting, async move { - let response = driver - .call_owned( - openshell_otel::rpc::START_SANDBOX, - Some(&operation_id), - move |driver| async move { driver.start_sandbox(Request::new(request)).await }, - ) - .await; - match (response, recreate) { - (Err(error), Some(driver_sandbox)) if error.code() == Code::NotFound => { + let result = Box::pin( + self.await_claimed_provisioning_operation(&starting, async move { + runtime.validate_caller_driver_config( + owned_starting + .spec + .as_ref() + .and_then(|spec| spec.template.as_ref()), + )?; + let expected_runtime_identity = sandbox_compute_runtime_identity(&owned_previous); + // One owned operation covers both calls. A NotFound response alone + // does not end ownership while the recovery create can still run. + let recreate = if provisioning_deadline::timed_out(&owned_previous) { + let mut driver_sandbox = runtime.driver_sandbox_for_launch(&owned_starting)?; + if let Some(spec) = driver_sandbox.spec.as_mut() { + spec.launch_authentication + .clone_from(&launch_authentication); + } + Some(driver_sandbox) + } else { + None + }; + // Starts may need to replace a stopped local workload/supervisor pair. + // Reconstruct input from the gateway's durable provisioning snapshot. + let request = runtime.driver_start_request( + &owned_starting, + launch_authentication, + expected_runtime_identity, + )?; + let response = driver .call_owned( - openshell_otel::rpc::CREATE_SANDBOX, + openshell_otel::rpc::START_SANDBOX, Some(&operation_id), move |driver| async move { - use openshell_core::proto::compute::v1::StartSandboxResponse; - let response = driver - .create_sandbox(Request::new(CreateSandboxRequest { - sandbox: Some(driver_sandbox), - })) - .await?; - Ok(tonic::Response::new(StartSandboxResponse { - runtime_identity: response.into_inner().runtime_identity, - })) + driver.start_sandbox(Request::new(request)).await }, ) - .await + .await; + match (response, recreate) { + (Err(error), Some(driver_sandbox)) if error.code() == Code::NotFound => { + driver + .call_owned( + openshell_otel::rpc::CREATE_SANDBOX, + Some(&operation_id), + move |driver| async move { + use openshell_core::proto::compute::v1::StartSandboxResponse; + let response = driver + .create_sandbox(Request::new(CreateSandboxRequest { + sandbox: Some(driver_sandbox), + })) + .await?; + Ok(tonic::Response::new(StartSandboxResponse { + runtime_identity: response.into_inner().runtime_identity, + })) + }, + ) + .await + } + (response, _) => response, } - (response, _) => response, - } - })) + }), + ) .await; match result { @@ -6652,6 +6782,13 @@ impl ComputeRuntime { if existing.updated_at_ms > sweep_started_at_ms { return Ok(()); } + let sandbox = decode_sandbox_record(&existing)?; + if provisioning_deadline::driver_operation_pending(&sandbox) + || (sandbox.phase() == i32::from(SandboxPhase::Starting) + && self.lifecycle_gates.try_lock_for(&snapshot.id).is_none()) + { + return Ok(()); + } existing.resource_version }; @@ -6676,6 +6813,11 @@ impl ComputeRuntime { { return Ok(()); } + let is_starting = + decode_sandbox_record(&existing)?.phase() == i32::from(SandboxPhase::Starting); + if is_starting && self.lifecycle_gates.try_lock_for(&snapshot.id).is_none() { + return Ok(()); + } self.apply_sandbox_update_locked(current, Some(existing)) .await @@ -6706,6 +6848,8 @@ impl ComputeRuntime { if self .pending_driver_operation_blocks_cleanup(&sandbox) .await? + || (sandbox.phase() == i32::from(SandboxPhase::Starting) + && self.lifecycle_gates.try_lock_for(&record.id).is_none()) { return Ok(()); } @@ -6740,6 +6884,12 @@ impl ComputeRuntime { { return Ok(()); } + let sandbox = decode_sandbox_record(¤t_record)?; + if sandbox.phase() == i32::from(SandboxPhase::Starting) + && self.lifecycle_gates.try_lock_for(&sandbox_id).is_none() + { + return Ok(()); + } if let Some(current) = current { return self @@ -6747,7 +6897,6 @@ impl ComputeRuntime { .await; } - let sandbox = decode_sandbox_record(¤t_record)?; let phase = SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); if terminal_stop_exit_code(&sandbox).is_some() { return self @@ -7666,6 +7815,7 @@ fn public_status_from_driver( next_restart_time: None, main_process_started_time: None, last_token_grant_failure: None, + ingress_admission: None, } } @@ -7822,6 +7972,9 @@ fn apply_driver_snapshot( .configuration_admission .clone_from(¤t_status.configuration_admission); status.configuration_activated = current_status.configuration_activated; + status + .ingress_admission + .clone_from(¤t_status.ingress_admission); status.provisioning.clone_from(¤t_status.provisioning); } if old_phase != phase { @@ -8563,7 +8716,7 @@ impl ComputeDriver for NoopTestDriver { Status, > { Ok(tonic::Response::new( - openshell_core::proto::compute::v1::ValidateSandboxCreateResponse {}, + openshell_core::proto::compute::v1::ValidateSandboxCreateResponse::default(), )) } @@ -9229,7 +9382,9 @@ mod tests { .lock() .expect("validated sandbox observation lock poisoned") .push(request.into_inner().sandbox.expect("validation sandbox")); - Ok(tonic::Response::new(ValidateSandboxCreateResponse {})) + Ok(tonic::Response::new( + ValidateSandboxCreateResponse::default(), + )) } async fn get_sandbox( @@ -9771,7 +9926,9 @@ mod tests { &self, _request: Request, ) -> Result, Status> { - Ok(tonic::Response::new(ValidateSandboxCreateResponse {})) + Ok(tonic::Response::new( + ValidateSandboxCreateResponse::default(), + )) } async fn get_sandbox( @@ -13467,22 +13624,61 @@ mod tests { #[tokio::test] async fn retained_starting_transition_retries_driver_operation() { - let driver = ControlledDriver::new(); - let runtime = test_runtime(driver.clone()).await; - let sandbox = sandbox_record( - "sb-retained-start", - "sandbox-retained-start", - SandboxPhase::Starting, - ); - runtime.store.put_message(&sandbox).await.unwrap(); + for preparation_fails in [false, true] { + let driver = ControlledDriver::new(); + let runtime = test_runtime(driver.clone()).await; + let mut sandbox = sandbox_record( + "sb-retained-start", + "sandbox-retained-start", + SandboxPhase::Starting, + ); + if preparation_fails { + sandbox.status.as_mut().unwrap().provisioning = + Some(provisioning_deadline::new_preparation_record( + openshell_core::time::now_ms(), + 300, + runtime.provisioning_repair_window_ms, + )); + sandbox.metadata.as_mut().unwrap().annotations.insert( + crate::auth::sandbox_session::RUNTIME_GENERATION_ANNOTATION.to_string(), + String::new(), + ); + } + runtime.store.put_message(&sandbox).await.unwrap(); + let before = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); - let starting = runtime - .start_sandbox("default", sandbox.object_name()) - .await - .unwrap(); + let result = runtime + .start_sandbox("default", sandbox.object_name()) + .await; + if preparation_fails { + assert_eq!(result.unwrap_err().code(), Code::FailedPrecondition); + assert_eq!(driver.start_calls(), 0); + } else { + assert_eq!(result.unwrap().phase(), SandboxPhase::Starting as i32); + assert_eq!(driver.start_calls(), 1); + } + let stored = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); - assert_eq!(starting.phase(), SandboxPhase::Starting as i32); - assert_eq!(driver.start_calls(), 1); + assert_eq!(stored.phase(), SandboxPhase::Starting as i32); + assert!(!provisioning_deadline::driver_operation_pending(&stored)); + if !preparation_fails { + assert!(sandbox_provisioning_attempt_id(&stored).is_none()); + assert_eq!( + sandbox_resource_version(&stored), + sandbox_resource_version(&before) + 1 + ); + } + } } #[tokio::test] @@ -13701,13 +13897,24 @@ mod tests { SandboxPhase::Error, ] { let driver = ControlledDriver::new(); - let runtime = test_runtime(driver.clone()).await; + let directory = tempfile::tempdir().unwrap(); + let database_url = + format!("sqlite://{}", directory.path().join("gateway.db").display()); + let mut runtime = test_runtime(driver.clone()).await; + runtime.store = Arc::new(Store::connect(&database_url).await.unwrap()); + let pool = sqlx::SqlitePool::connect(&database_url).await.unwrap(); let mut sandbox = sandbox_record("sb-auth-start", "sandbox-auth-start", phase); if phase == SandboxPhase::Error { sandbox.set_phase(SandboxPhase::Ready.into()); apply_main_process_exit(&mut sandbox, "previous-instance", 9); } runtime.store.put_message(&sandbox).await.unwrap(); + sqlx::query("CREATE TABLE start_writes (payload BLOB NOT NULL)") + .execute(&pool) + .await + .unwrap(); + sqlx::query("CREATE TRIGGER record_start AFTER UPDATE ON objects WHEN NEW.id = 'sb-auth-start' BEGIN INSERT INTO start_writes (payload) VALUES (NEW.payload); END") + .execute(&pool).await.unwrap(); let before = runtime .store .get_message::(sandbox.object_id()) @@ -13745,6 +13952,20 @@ mod tests { &stored.metadata.as_ref().unwrap().annotations, ) .unwrap(); + let writes: Vec> = + sqlx::query_scalar("SELECT payload FROM start_writes ORDER BY rowid") + .fetch_all(&pool) + .await + .unwrap(); + let first_start = writes + .iter() + .map(|payload| Sandbox::decode(payload.as_slice()).unwrap()) + .find(|sandbox| sandbox.phase() == i32::from(SandboxPhase::Starting)) + .unwrap(); + assert!( + provisioning_deadline::driver_operation_pending(&first_start), + "the first persisted Starting write must own its driver operation" + ); assert_eq!(stored.phase(), SandboxPhase::Starting as i32); assert_eq!(stored_identity.auth_epoch.get(), 2); assert_ne!( @@ -13768,8 +13989,8 @@ mod tests { ); assert_eq!( sandbox_resource_version(&stored), - sandbox_resource_version(&before) + 3, - "identity/Starting transition, operation claim, and settlement are separate CAS writes" + sandbox_resource_version(&before) + 2, + "identity, Starting, and driver ownership commit together before settlement" ); runtime @@ -14137,15 +14358,26 @@ mod tests { #[tokio::test] async fn stale_container_exit_queued_before_start_cannot_regress_restart() { - for reason in [ - "ContainerExited", - "ContainerStopped", - "ContainerRuntimeRestart", + for (reason, observation) in [ + ("ContainerExited", "watch"), + ("ContainerStopped", "watch"), + ("ContainerRuntimeRestart", "watch"), + ("ContainerExited", "reconcile"), + ("ContainerExited", "legacy-reconcile"), + ("ContainerExited", "prune"), + ("ContainerExited", "prune-missing"), + ("ContainerExited", "peer-reconcile"), ] { let driver = ControlledDriver::new(); driver.block_start(); - let sandbox = - sandbox_record("sb-start-race", "sandbox-start-race", SandboxPhase::Stopped); + // Legacy Starting retries have no durable provisioning claim, but + // still own the local lifecycle gate while their RPC is in flight. + let phase = if matches!(observation, "legacy-reconcile" | "prune" | "prune-missing") { + SandboxPhase::Starting + } else { + SandboxPhase::Stopped + }; + let sandbox = sandbox_record("sb-start-race", "sandbox-start-race", phase); driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new( ready_driver_sandbox(sandbox.object_id(), sandbox.object_name()), ))); @@ -14167,20 +14399,81 @@ mod tests { reason, "container stopped before restart", ))); - let update_runtime = runtime.clone(); - let mut update = - tokio::spawn(async move { update_runtime.apply_sandbox_update(stale_exit).await }); + let mut observer = runtime.clone(); + if observation == "peer-reconcile" { + observer.lifecycle_gates = Arc::new(LifecycleGateRegistry::default()); + observer.mutation_locks = Arc::new(LocalMutationLocks::new()); + let current = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); + assert!(provisioning_deadline::driver_operation_pending(¤t)); + } + if observation == "prune-missing" { + driver.set_get_outcome(ControlledGetOutcome::Missing); + } else if observation != "watch" { + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new(stale_exit.clone()))); + } + let update_runtime = observer.clone(); + let mut update = tokio::spawn(async move { + match observation { + "watch" => update_runtime.apply_sandbox_update(stale_exit).await, + "reconcile" | "legacy-reconcile" | "peer-reconcile" => { + update_runtime + .reconcile_snapshot_sandbox(stale_exit, i64::MAX) + .await + } + "prune" | "prune-missing" => { + let record = update_runtime + .store + .get(Sandbox::object_type(), &stale_exit.id) + .await + .unwrap() + .unwrap(); + update_runtime + .prune_missing_sandbox(record, i64::MAX, 0) + .await + } + _ => unreachable!(), + } + }); - assert!( - tokio::time::timeout(Duration::from_millis(50), &mut update) + if observation == "watch" { + assert!( + tokio::time::timeout(Duration::from_millis(50), &mut update) + .await + .is_err(), + "{reason} watch update must wait for the active start operation" + ); + } else { + tokio::time::timeout(Duration::from_secs(1), &mut update) .await - .is_err(), - "{reason} watch update must wait for the active start operation" - ); + .expect("reconciliation must skip an active start") + .unwrap() + .unwrap(); + assert_eq!( + runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap() + .phase(), + SandboxPhase::Starting as i32, + "{observation} must not apply the old exit during start" + ); + } + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new( + ready_driver_sandbox(sandbox.object_id(), sandbox.object_name()), + ))); driver.release_start(); start.await.unwrap().unwrap(); - update.await.unwrap().unwrap(); + if observation == "watch" { + update.await.unwrap().unwrap(); + } let stored = runtime .store @@ -14191,8 +14484,78 @@ mod tests { assert_eq!( stored.phase(), SandboxPhase::Starting as i32, - "{reason} must not regress the restarted sandbox" + "{observation}: {reason} must not regress the restarted sandbox" ); + if phase == SandboxPhase::Starting { + assert!(sandbox_provisioning_attempt_id(&stored).is_none()); + } + if observation != "watch" { + // These readiness and real-failure controls follow #4983's regression. + register_test_supervisor_session(&observer, sandbox.object_id()); + let ready = ready_driver_sandbox(sandbox.object_id(), sandbox.object_name()); + observer + .reconcile_snapshot_sandbox(ready.clone(), i64::MAX) + .await + .unwrap(); + if stored + .status + .as_ref() + .is_some_and(|status| status.configuration_admission.is_some()) + { + let awaiting_admission = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); + assert_eq!(awaiting_admission.phase(), SandboxPhase::Starting as i32); + runtime + .store + .update_message_cas::(sandbox.object_id(), 0, |sandbox| { + let status = sandbox.status.as_mut().unwrap(); + provisioning_deadline::record_admission_start( + status.provisioning.as_mut().unwrap(), + openshell_core::time::now_ms(), + runtime.provisioning_repair_window_ms, + ) + .unwrap(); + status.configuration_admission.as_mut().unwrap().state = + openshell_core::proto::ConfigurationAdmissionState::Accepted.into(); + }) + .await + .unwrap(); + } + observer + .reconcile_snapshot_sandbox(ready, i64::MAX) + .await + .unwrap(); + let started = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); + assert_eq!(started.phase(), SandboxPhase::Ready as i32); + + let mut failed = ready_driver_sandbox(sandbox.object_id(), sandbox.object_name()); + failed.status = Some(make_driver_status(make_driver_condition( + "ContainerExited", + "replacement container exited", + ))); + driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new(failed.clone()))); + observer + .reconcile_snapshot_sandbox(failed, i64::MAX) + .await + .unwrap(); + let failed = runtime + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); + assert_eq!(failed.phase(), SandboxPhase::Error as i32); + assert_eq!(ready_condition(&failed).unwrap().reason, "ContainerExited"); + } } } @@ -16334,51 +16697,86 @@ mod tests { #[tokio::test] async fn blocked_reconciliation_lookup_does_not_delay_watch_events() { - let driver = ControlledDriver::new(); - driver.block_get(); - let snapshot = ready_driver_sandbox("sb-a", "sandbox-a"); - driver.set_get_outcome(ControlledGetOutcome::Sandbox(Box::new(snapshot.clone()))); - let runtime = test_runtime(driver.clone()).await; - for sandbox in [ - sandbox_record("sb-a", "sandbox-a", SandboxPhase::Provisioning), - sandbox_record("sb-b", "sandbox-b", SandboxPhase::Ready), + for (observation, deleted_id) in [ + ("snapshot", "sb-b"), + ("snapshot", "sb-a"), + ("prune", "sb-a"), ] { - runtime.store.put_message(&sandbox).await.unwrap(); - } + let driver = ControlledDriver::new(); + driver.block_get(); + let snapshot = ready_driver_sandbox("sb-a", "sandbox-a"); + driver.set_get_outcome(if deleted_id == "sb-b" { + ControlledGetOutcome::Sandbox(Box::new(snapshot.clone())) + } else if observation == "snapshot" { + ControlledGetOutcome::Missing + } else { + ControlledGetOutcome::Error("lookup failed after the Deleted watch") + }); + let runtime = test_runtime(driver.clone()).await; + for sandbox in [ + sandbox_record("sb-a", "sandbox-a", SandboxPhase::Provisioning), + sandbox_record("sb-b", "sandbox-b", SandboxPhase::Ready), + ] { + runtime.store.put_message(&sandbox).await.unwrap(); + } - let (shutdown_tx, watch_handle) = start_watch_loop(&runtime, &driver).await; - let mut sandbox_b_rx = runtime.sandbox_watch_bus.subscribe("sb-b"); - let reconcile_runtime = runtime.clone(); - let reconcile = tokio::spawn(async move { - reconcile_runtime - .reconcile_snapshot_sandbox(snapshot, i64::MAX) + let (shutdown_tx, watch_handle) = start_watch_loop(&runtime, &driver).await; + let mut deleted_rx = runtime.sandbox_watch_bus.subscribe(deleted_id); + let reconcile_runtime = runtime.clone(); + let reconcile = tokio::spawn(async move { + if observation == "prune" { + let record = reconcile_runtime + .store + .get(Sandbox::object_type(), "sb-a") + .await + .unwrap() + .unwrap(); + return reconcile_runtime + .prune_missing_sandbox(record, i64::MAX, 0) + .await; + } + reconcile_runtime + .reconcile_snapshot_sandbox(snapshot, i64::MAX) + .await + }); + tokio::time::timeout(Duration::from_secs(1), driver.get_started.notified()) .await - }); - tokio::time::timeout(Duration::from_secs(1), driver.get_started.notified()) - .await - .expect("reconciliation did not reach the driver lookup"); + .expect("reconciliation did not reach the driver lookup"); - driver.send_event(deleted_watch_event("sb-b")); - tokio::time::timeout(Duration::from_secs(1), sandbox_b_rx.recv()) - .await - .expect("watch event was blocked by reconciliation lookup") - .expect("sandbox B watch bus closed before notification"); - assert!( - runtime - .store - .get_message::("sb-b") + driver.send_event(deleted_watch_event(deleted_id)); + tokio::time::timeout(Duration::from_secs(1), deleted_rx.recv()) + .await + .expect("watch event was blocked by reconciliation lookup") + .expect("sandbox watch bus closed before notification"); + assert!( + runtime + .store + .get_message::(deleted_id) + .await + .unwrap() + .is_none() + ); + let cleanup = + tokio::time::timeout(Duration::from_secs(1), driver.delete_started.notified()) + .await; + driver.release_get(); + let reconciled = tokio::time::timeout(Duration::from_secs(1), reconcile) .await .unwrap() - .is_none() - ); - - driver.release_get(); - tokio::time::timeout(Duration::from_secs(1), reconcile) - .await - .unwrap() - .unwrap() - .unwrap(); - stop_watch_loop(shutdown_tx, watch_handle).await; + .unwrap(); + if observation == "prune" { + assert!( + reconciled + .unwrap_err() + .contains("lookup failed after the Deleted watch") + ); + } else { + reconciled.unwrap(); + } + cleanup.expect("the sweep must not suppress deletion-watch driver cleanup"); + assert_eq!(driver.delete_calls(), 1); + stop_watch_loop(shutdown_tx, watch_handle).await; + } } #[tokio::test] @@ -16515,59 +16913,78 @@ mod tests { } #[tokio::test] - async fn apply_sandbox_update_without_status_preserves_existing_status() { - let runtime = test_runtime(Arc::new(TestDriver::default())).await; - let mut sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); - sandbox.status = Some(SandboxStatus { - conditions: vec![SandboxCondition { - r#type: "Ready".to_string(), - status: "True".to_string(), - reason: "DependenciesReady".to_string(), - message: "Pod is Ready".to_string(), - transition_time: None, - }], - current_policy_version: 7, - ..Default::default() - }); - sandbox.set_phase(SandboxPhase::Ready as i32); - runtime.store.put_message(&sandbox).await.unwrap(); + async fn apply_sandbox_update_preserves_gateway_owned_status() { + for observed_status in [ + None, + Some(make_driver_status(DriverCondition { + status: "True".into(), + ..make_driver_condition("DependenciesReady", "Pod is Ready") + })), + ] { + let mut runtime = test_runtime(Arc::new(TestDriver::default())).await; + runtime.driver_info.driver_reports_runtime_readiness = true; + let mut sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + sandbox.status = Some(SandboxStatus { + conditions: vec![SandboxCondition { + r#type: "Ready".to_string(), + status: "True".to_string(), + reason: "DependenciesReady".to_string(), + message: "Pod is Ready".to_string(), + transition_time: None, + }], + current_policy_version: 7, + ..Default::default() + }); + sandbox.set_phase(SandboxPhase::Ready as i32); + let ingress = openshell_core::proto::SandboxIngressAdmission { + scope: openshell_core::proto::IngressScope::SharedHostLoopback as i32, + all_ports: false, + ports: vec![8080], + }; + sandbox.status.as_mut().unwrap().ingress_admission = Some(ingress.clone()); + runtime.store.put_message(&sandbox).await.unwrap(); - runtime - .apply_sandbox_update(DriverSandbox { - id: "sb-1".to_string(), - name: "sandbox-a".to_string(), - namespace: "default".to_string(), - spec: None, - status: None, - workspace: "default".to_string(), - }) - .await - .unwrap(); + runtime + .apply_sandbox_update(DriverSandbox { + id: "sb-1".to_string(), + name: "sandbox-a".to_string(), + namespace: "default".to_string(), + spec: None, + status: observed_status, + workspace: "default".to_string(), + }) + .await + .unwrap(); - let stored = runtime - .store - .get_message::("sb-1") - .await - .unwrap() - .unwrap(); - assert_eq!( - SandboxPhase::try_from(stored.phase()).unwrap(), - SandboxPhase::Ready - ); - assert_eq!(stored.current_policy_version(), 7); - let ready = stored - .status - .as_ref() - .and_then(|status| { - status - .conditions - .iter() - .find(|condition| condition.r#type == "Ready") - }) - .unwrap(); - assert_eq!(ready.status, "True"); - assert_eq!(ready.reason, "DependenciesReady"); - assert_eq!(ready.message, "Pod is Ready"); + let stored = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!( + SandboxPhase::try_from(stored.phase()).unwrap(), + SandboxPhase::Ready + ); + assert_eq!(stored.current_policy_version(), 7); + assert_eq!( + stored.status.as_ref().unwrap().ingress_admission.as_ref(), + Some(&ingress) + ); + let ready = stored + .status + .as_ref() + .and_then(|status| { + status + .conditions + .iter() + .find(|condition| condition.r#type == "Ready") + }) + .unwrap(); + assert_eq!(ready.status, "True"); + assert_eq!(ready.reason, "DependenciesReady"); + assert_eq!(ready.message, "Pod is Ready"); + } } #[tokio::test] @@ -18977,57 +19394,65 @@ endpoint = { kind = "unix", socket_path = "/run/agent-substrate/isolation.sock" #[cfg(unix)] async fn remote_compute_driver_rejects_policy_changes_before_forwarding() { use crate::test_support::{FakeComputeDriver, FakeComputeDriverCall}; - let dir = tempfile::tempdir().unwrap(); - let socket_path = dir.path().join("admission.sock"); - let driver = FakeComputeDriver::new(); - let _server = driver.serve_uds(&socket_path).unwrap(); - let endpoint = connect_remote_compute_driver("external-test", &socket_path) - .await - .unwrap(); - let remote = RemoteComputeDriver::new(endpoint.channel); - remote - .get_capabilities(Request::new(GetCapabilitiesRequest { - gateway: Some(gateway_metadata(ExtensionFamily::Compute)), - })) - .await - .unwrap(); - driver.set_admission_acknowledgement(String::new()); - driver.clear_calls(); - let sandbox = DriverSandbox::default(); - let status = remote - .validate_sandbox_create(Request::new(ValidateSandboxCreateRequest { - sandbox: Some(sandbox.clone()), - })) - .await - .unwrap_err(); - assert_eq!(status.code(), Code::FailedPrecondition); - assert!( + for remove_ingress in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let socket_path = dir.path().join("admission.sock"); + let driver = FakeComputeDriver::new(); + driver.set_ingress_capability(true); + let _server = driver.serve_uds(&socket_path).unwrap(); + let endpoint = connect_remote_compute_driver("external-test", &socket_path) + .await + .unwrap(); + let remote = RemoteComputeDriver::new(endpoint.channel); remote - .create_sandbox(Request::new(CreateSandboxRequest { - sandbox: Some(sandbox) + .get_capabilities(Request::new(GetCapabilitiesRequest { + gateway: Some(gateway_metadata(ExtensionFamily::Compute)), })) .await - .is_err() - ); - assert!( - remote - .start_sandbox(Request::new(StartSandboxRequest::default())) + .unwrap(); + if remove_ingress { + driver.set_ingress_capability(false); + } else { + driver.set_admission_acknowledgement(String::new()); + } + driver.clear_calls(); + let sandbox = DriverSandbox::default(); + let status = remote + .validate_sandbox_create(Request::new(ValidateSandboxCreateRequest { + sandbox: Some(sandbox.clone()), + })) .await - .is_err() - ); - let status = openshell_core::compute::ComputeDriver::update_sandbox_configuration( - &remote, - Request::new( - openshell_core::proto::compute::v1::UpdateSandboxConfigurationRequest::default(), - ), - ) - .await - .unwrap_err(); - assert_eq!(status.code(), Code::FailedPrecondition); - assert_eq!( - driver.calls(), - vec![FakeComputeDriverCall::GetCapabilities; 4] - ); + .unwrap_err(); + assert_eq!(status.code(), Code::FailedPrecondition); + assert!( + remote + .create_sandbox(Request::new(CreateSandboxRequest { + sandbox: Some(sandbox) + })) + .await + .is_err() + ); + assert!( + remote + .start_sandbox(Request::new(StartSandboxRequest::default())) + .await + .is_err() + ); + let status = openshell_core::compute::ComputeDriver::update_sandbox_configuration( + &remote, + Request::new( + openshell_core::proto::compute::v1::UpdateSandboxConfigurationRequest::default( + ), + ), + ) + .await + .unwrap_err(); + assert_eq!(status.code(), Code::FailedPrecondition); + assert_eq!( + driver.calls(), + vec![FakeComputeDriverCall::GetCapabilities; 4] + ); + } } #[tokio::test] diff --git a/crates/openshell-server/src/grpc/mod.rs b/crates/openshell-server/src/grpc/mod.rs index 69208348de..707b0feb74 100644 --- a/crates/openshell-server/src/grpc/mod.rs +++ b/crates/openshell-server/src/grpc/mod.rs @@ -255,7 +255,11 @@ impl openshell_core::proto::sandbox_updates_server::SandboxUpdates for OpenShell &self, request: Request, ) -> Result, Status> { - mutation_replay::run(&self.state, request).await + let mut response = mutation_replay::run(&self.state, request).await?; + if let Some(sandbox) = response.get_mut().sandbox.as_mut() { + self.state.compute.project_ingress_admission(sandbox); + } + Ok(response) } } @@ -336,7 +340,11 @@ impl OpenShell for OpenShellService { &self, request: Request, ) -> Result, Status> { - mutation_replay::run(&self.state, request).await + let mut response = mutation_replay::run(&self.state, request).await?; + if let Some(sandbox) = response.get_mut().sandbox.as_mut() { + self.state.compute.project_ingress_admission(sandbox); + } + Ok(response) } async fn begin_rootfs_tar_staging( @@ -408,14 +416,22 @@ impl OpenShell for OpenShellService { &self, request: Request, ) -> Result, Status> { - mutation_replay::run(&self.state, request).await + let mut response = mutation_replay::run(&self.state, request).await?; + if let Some(sandbox) = response.get_mut().sandbox.as_mut() { + self.state.compute.project_ingress_admission(sandbox); + } + Ok(response) } async fn detach_sandbox_provider( &self, request: Request, ) -> Result, Status> { - mutation_replay::run(&self.state, request).await + let mut response = mutation_replay::run(&self.state, request).await?; + if let Some(sandbox) = response.get_mut().sandbox.as_mut() { + self.state.compute.project_ingress_admission(sandbox); + } + Ok(response) } async fn get_sandbox_provider_status( @@ -443,14 +459,22 @@ impl OpenShell for OpenShellService { &self, request: Request, ) -> Result, Status> { - mutation_replay::run(&self.state, request).await + let mut response = mutation_replay::run(&self.state, request).await?; + if let Some(sandbox) = response.get_mut().sandbox.as_mut() { + self.state.compute.project_ingress_admission(sandbox); + } + Ok(response) } async fn start_sandbox( &self, request: Request, ) -> Result, Status> { - mutation_replay::run(&self.state, request).await + let mut response = mutation_replay::run(&self.state, request).await?; + if let Some(sandbox) = response.get_mut().sandbox.as_mut() { + self.state.compute.project_ingress_admission(sandbox); + } + Ok(response) } async fn suspend_sandbox( diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 35e3122bd4..99c2644ca3 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -79,7 +79,7 @@ use crate::persistence::current_time_ms; const TCP_FORWARD_CHUNK_SIZE: usize = 64 * 1024; const NO_LOGIN_SHELL_ENV: (&str, &str) = ("OPENSHELL_NO_LOGIN_SHELL", "1"); const MAX_TEMPLATES_PER_WORKSPACE: u32 = 1000; -const MAX_CREATE_SERVICE_EXPOSURES: usize = 32; +const MAX_CREATE_INGRESS_PORTS: usize = 32; const SANDBOX_CAS_RETRY_LIMIT: usize = 16; #[cfg(test)] @@ -363,7 +363,7 @@ pub(super) async fn handle_update_sandbox( ) .await?; Ok(Response::new(SandboxResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), service_urls: HashMap::new(), })) } @@ -776,7 +776,7 @@ async fn handle_create_sandbox_inner( state .compute .validate_launch_signer_configured(state.sandbox_session_jwt_authority.is_some())?; - state + let ingress_admission = state .compute .validate_sandbox_create_with_runtime_inputs( &sandbox, @@ -789,6 +789,19 @@ async fn handle_create_sandbox_inner( status })?; + sandbox + .status + .get_or_insert_with(Default::default) + .ingress_admission = ingress_admission; + for port in request.forward_ports.iter().copied().chain( + request + .service_exposures + .iter() + .map(|exposure| exposure.target_port), + ) { + state.compute.admit_ingress_port(&sandbox, port)?; + } + let runtime_identity = crate::auth::sandbox_session::PersistedSandboxIdentity::new() .map_err(|error| Status::internal(error.to_string()))?; if let Some(metadata) = sandbox.metadata.as_mut() { @@ -875,11 +888,18 @@ async fn handle_create_sandbox_inner( "CreateSandbox request completed successfully" ); Ok(Response::new(SandboxResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), service_urls, })) } +/// Project gateway-owned evidence on every outward sandbox value without +/// rewriting retained legacy records or accepting driver-supplied evidence. +fn project_public_sandbox(state: &ServerState, mut sandbox: Sandbox) -> Sandbox { + state.compute.project_ingress_admission(&mut sandbox); + sandbox +} + fn validate_create_sandbox_request_pre_io( request: &CreateSandboxRequest, workload_template_name: &str, @@ -888,9 +908,23 @@ fn validate_create_sandbox_request_pre_io( crate::grpc::validation::validate_labels(&request.labels)?; crate::grpc::validation::validate_caller_annotations(&request.annotations, "annotations")?; - if request.service_exposures.len() > MAX_CREATE_SERVICE_EXPOSURES { + if request.forward_ports.len() > MAX_CREATE_INGRESS_PORTS { + return Err(Status::invalid_argument(format!( + "forward_ports must contain at most {MAX_CREATE_INGRESS_PORTS} entries" + ))); + } + if request + .forward_ports + .iter() + .any(|port| !(1..=65535).contains(port)) + { + return Err(Status::invalid_argument( + "forward_ports must be between 1 and 65535", + )); + } + if request.service_exposures.len() > MAX_CREATE_INGRESS_PORTS { return Err(Status::invalid_argument(format!( - "service_exposures must contain at most {MAX_CREATE_SERVICE_EXPOSURES} entries" + "service_exposures must contain at most {MAX_CREATE_INGRESS_PORTS} entries" ))); } let mut service_names = HashSet::with_capacity(request.service_exposures.len()); @@ -1053,7 +1087,7 @@ pub(super) async fn handle_get_sandbox( None }; let mut response = Response::new(SandboxResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), service_urls: HashMap::new(), }); if let Some(value) = owner.and_then(|owner| owner.parse().ok()) { @@ -1109,11 +1143,14 @@ pub(super) async fn handle_list_sandboxes( label_selector: selector, }, }; - let page = state + let mut page = state .store .list_message_page::(query, after.as_ref(), pagination.page_size()) .await .map_err(|e| Status::internal(format!("list sandboxes failed: {e}")))?; + for sandbox in &mut page.messages { + state.compute.project_ingress_admission(sandbox); + } let next_page_token = pagination.next_object_token(page.next_cursor.as_ref()); Ok(Response::new(ListSandboxesResponse { sandboxes: page.messages, @@ -1641,7 +1678,7 @@ pub(super) async fn handle_attach_sandbox_provider( ); Ok(Response::new(AttachSandboxProviderResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), attached, receipt: Some(receipt), })) @@ -1769,7 +1806,7 @@ pub(super) async fn handle_detach_sandbox_provider( ); Ok(Response::new(DetachSandboxProviderResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), detached, receipt: Some(receipt), })) @@ -1879,7 +1916,7 @@ async fn handle_stop_sandbox_inner( let sandbox = state.compute.stop_sandbox(workspace, name).await?; info!(sandbox_name = %name, "StopSandbox request completed successfully"); Ok(Response::new(SandboxResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), service_urls: HashMap::new(), })) } @@ -1934,7 +1971,7 @@ async fn handle_start_sandbox_inner( .project_endpoint_status(&mut sandbox, remote_authority); info!(sandbox_name = %name, "StartSandbox request completed successfully"); Ok(Response::new(SandboxResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), service_urls: HashMap::new(), })) } @@ -1975,7 +2012,7 @@ async fn handle_suspend_sandbox_inner( let sandbox = state.compute.suspend_sandbox(workspace, name).await?; info!(sandbox_name = %name, "SuspendSandbox request completed successfully"); Ok(Response::new(SandboxResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), service_urls: HashMap::new(), })) } @@ -2031,7 +2068,7 @@ async fn handle_resume_sandbox_inner( .project_endpoint_status(&mut sandbox, remote_authority); info!(sandbox_name = %name, "ResumeSandbox request completed successfully"); Ok(Response::new(SandboxResponse { - sandbox: Some(sandbox), + sandbox: Some(project_public_sandbox(state, sandbox)), service_urls: HashMap::new(), })) } @@ -2229,7 +2266,7 @@ pub(super) async fn handle_watch_sandbox( .send(Ok(SandboxStreamEvent { payload: Some( openshell_core::proto::sandbox_stream_event::Payload::Sandbox( - sandbox.clone(), + project_public_sandbox(&state, sandbox.clone()), ), ), // Status snapshots are re-read, not resumed by cursor. @@ -2509,7 +2546,7 @@ pub(super) async fn handle_watch_sandbox( match state.store.get_message::(&sandbox_id).await { Ok(Some(sandbox)) => { state.sandbox_index.update_from_sandbox(&sandbox); - if tx.send(Ok(SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Sandbox(sandbox.clone())), cursor: String::new() })).await.is_err() { + if tx.send(Ok(SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Sandbox(project_public_sandbox(&state, sandbox.clone()))), cursor: String::new() })).await.is_err() { return; } if stop_on_terminal { @@ -2879,6 +2916,10 @@ pub(super) async fn handle_forward_tcp( ) .await?; + if let relay_open::Target::Tcp(target) = &target { + state.compute.admit_ingress_port(&sandbox, target.port)?; + } + // The main process may finish between minting the SSH token and opening // its transport. Keep the relay reachable until terminal delivery is // finalized so fast commands can attach without a readiness race. @@ -4343,6 +4384,377 @@ mod tests { GpuResourceRequirements, SandboxServiceExposure, ServiceAuthorizationMode, ServiceEndpoint, }; + #[tokio::test] + async fn ingress_admission_rejects_unadmitted_create_ports_before_provisioning() { + use crate::test_support::{FakeComputeDriver, FakeComputeDriverCall}; + use openshell_core::extension_protocol::{ + COMPUTE_INGRESS_ADMISSION, ExtensionFamily, extension_metadata, + }; + use openshell_core::proto::compute::v1::GetCapabilitiesResponse; + use openshell_core::proto::{IngressScope, SandboxIngressAdmission}; + + use futures::StreamExt as _; + use openshell_core::proto::open_shell_server::OpenShell as _; + let directory = tempfile::tempdir().unwrap(); + let key = directory.path().join("private-key"); + std::fs::write(&key, b"test-only stable fingerprint material").unwrap(); + let exact = SandboxIngressAdmission { + scope: IngressScope::SharedHostLoopback as i32, + ports: vec![8080], + all_ports: false, + }; + for (modern, admission, port, accepted) in [ + (false, None, 8081, true), + (true, None, 8080, false), + (true, Some(SandboxIngressAdmission::default()), 8080, false), + ( + true, + Some(SandboxIngressAdmission { + scope: IngressScope::Unsupported as i32, + ..Default::default() + }), + 8080, + false, + ), + (true, Some(exact.clone()), 8081, false), + (true, Some(exact.clone()), 8080, true), + (true, Some(exact.clone()), 0, false), + ( + true, + Some(SandboxIngressAdmission { + ports: vec![8080, 8080], + ..exact.clone() + }), + 8080, + false, + ), + ( + true, + Some(SandboxIngressAdmission { + all_ports: true, + ..exact.clone() + }), + 8080, + false, + ), + ( + true, + Some(SandboxIngressAdmission { + scope: IngressScope::IsolatedLoopback as i32, + all_ports: true, + ports: vec![], + }), + 8081, + true, + ), + ] { + for forward in [false, true] { + let mut state = test_server_state().await; + Arc::get_mut(&mut state).unwrap().config.gateway_jwt = + Some(openshell_core::config::GatewayJwtConfig { + signing_key_path: key.clone(), + public_key_path: directory.path().join("public"), + kid_path: directory.path().join("kid"), + gateway_id: "test".into(), + ttl_secs: None, + }); + let driver = Arc::new( + FakeComputeDriver::new() + .with_ingress_admission(admission.clone()) + .with_capabilities(GetCapabilitiesResponse { + driver_name: "test".into(), + default_image: "test/image".into(), + resource_admission_policy: + openshell_core::resource_admission::DriverAdmissionConfig::default() + .acknowledgement(), + extension: Some(extension_metadata( + ExtensionFamily::Compute, + "openshell/test", + "test", + if modern { + vec![COMPUTE_INGRESS_ADMISSION.to_string()] + } else { + vec![] + }, + )), + ..Default::default() + }), + ); + let compute = crate::compute::ComputeRuntime::from_driver( + "test".into(), + driver.clone(), + None, + state.store.clone(), + crate::sandbox_index::SandboxIndex::new(), + crate::sandbox_watch::SandboxWatchBus::new(), + crate::tracing_bus::TracingLogBus::new(), + Arc::new(crate::supervisor_session::SupervisorSessionRegistry::new()), + ) + .await + .unwrap(); + Arc::get_mut(&mut state).unwrap().compute = compute; + let result = handle_create_sandbox( + &state, + authed_request(CreateSandboxRequest { + name: "ingress".into(), + workspace_scope: Some(openshell_core::proto::workspace_selector("default")), + spec: Some(SandboxSpec { + policy: Some(openshell_policy::restrictive_default_policy()), + ..Default::default() + }), + forward_ports: if forward { vec![port] } else { vec![] }, + service_exposures: if forward { + vec![] + } else { + vec![SandboxServiceExposure { + target_port: port, + ..Default::default() + }] + }, + ..Default::default() + }), + ) + .await; + assert_eq!( + result.is_ok(), + accepted, + "modern={modern}, admission={admission:?}, port={port}" + ); + let provisioned = driver + .calls() + .iter() + .any(|call| matches!(call, FakeComputeDriverCall::CreateSandbox { .. })); + assert_eq!(provisioned, accepted); + let stored = state + .store + .get_message_by_name::("default", "ingress") + .await + .unwrap(); + assert_eq!(stored.is_some(), accepted); + if accepted && modern { + let sandbox = stored.unwrap(); + assert_eq!( + sandbox.status.as_ref().unwrap().ingress_admission, + admission + ); + if !admission.as_ref().unwrap().all_ports { + assert!( + super::super::service::expose_service_endpoint( + &state, + "default", + &sandbox, + "unadmitted", + 8081, + ServiceAuthorizationMode::Strip + ) + .await + .is_err() + ); + assert!( + state + .store + .get_message_by_name::( + "default", + "ingress--unadmitted" + ) + .await + .unwrap() + .is_none() + ); + // A resource created before ingress negotiation has no + // retained evidence. Every current read must expose its + // unavailability without rewriting that legacy record. + let mut legacy = sandbox.clone(); + let metadata = legacy.metadata.as_mut().unwrap(); + metadata.id = "legacy-ingress".into(); + metadata.name = "legacy-ingress".into(); + metadata.resource_version = 0; + legacy.set_phase(SandboxPhase::Ready as i32); + legacy.status.as_mut().unwrap().ingress_admission = None; + state.store.put_message(&legacy).await.unwrap(); + let get = handle_get_sandbox( + &state, + authed_request(GetSandboxRequest { + name: "legacy-ingress".into(), + workspace_scope: Some(openshell_core::proto::workspace_selector( + "default", + )), + }), + ) + .await + .unwrap() + .into_inner() + .sandbox + .unwrap(); + assert!( + openshell_core::ingress::admit_port( + get.status.as_ref().unwrap().ingress_admission.as_ref(), + 8080, + false + ) + .is_err() + ); + let service = crate::grpc::OpenShellService::new(state.clone()); + let start = StartSandboxRequest { + name: "legacy-ingress".into(), + workspace_scope: Some(openshell_core::proto::workspace_selector( + "default", + )), + request_id: uuid::Uuid::new_v4().to_string(), + }; + for replay in [false, true] { + let started = service + .start_sandbox(authed_request(start.clone())) + .await + .unwrap(); + assert_eq!( + started.metadata().contains_key("openshell-replayed"), + replay + ); + let descriptor = started + .into_inner() + .sandbox + .unwrap() + .status + .unwrap() + .ingress_admission; + assert!(descriptor.is_some()); + assert!( + openshell_core::ingress::admit_port( + descriptor.as_ref(), + 8080, + false + ) + .is_err() + ); + } + let mut stream = handle_watch_sandbox( + &state, + authed_request(WatchSandboxRequest { + sandbox: "legacy-ingress".into(), + workspace_scope: Some(openshell_core::proto::workspace_selector( + "default", + )), + follow_status: true, + ..Default::default() + }), + ) + .await + .unwrap() + .into_inner(); + for updated in [false, true] { + if updated { + state.sandbox_watch_bus.notify("legacy-ingress"); + } + let event = tokio::time::timeout( + std::time::Duration::from_secs(1), + stream.next(), + ) + .await + .unwrap() + .unwrap() + .unwrap(); + let Some( + openshell_core::proto::sandbox_stream_event::Payload::Sandbox( + snapshot, + ), + ) = event.payload + else { + panic!("expected status snapshot") + }; + assert!(snapshot.status.unwrap().ingress_admission.is_some()); + } + assert!( + state + .store + .get_message::("legacy-ingress") + .await + .unwrap() + .unwrap() + .status + .unwrap() + .ingress_admission + .is_none() + ); + } + } + } + } + } + + #[tokio::test] + async fn ingress_admission_guards_live_tcp_targets_without_blocking_ssh_transport() { + use openshell_core::proto::open_shell_client::OpenShellClient; + use openshell_core::proto::open_shell_server::OpenShellServer; + use openshell_core::proto::{IngressScope, SandboxIngressAdmission, tcp_forward_frame}; + use tokio_stream::wrappers::TcpListenerStream; + + let state = test_server_state().await; + let mut sandbox = Sandbox { + metadata: Some(ObjectMeta { + id: "ingress-live".into(), + name: "ingress-live".into(), + workspace: "default".into(), + ..Default::default() + }), + ..Default::default() + }; + sandbox.set_phase(SandboxPhase::Stopped as i32); + sandbox.status.as_mut().unwrap().ingress_admission = Some(SandboxIngressAdmission { + scope: IngressScope::SharedHostLoopback as i32, + all_ports: false, + ports: vec![8080], + }); + state.store.put_message(&sandbox).await.unwrap(); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(OpenShellServer::with_interceptor( + crate::OpenShellService::new(state), + crate::test_support::authenticate_as_dev_user, + )) + .serve_with_incoming(TcpListenerStream::new(listener)), + ); + let mut client = OpenShellClient::connect(format!("http://{address}")) + .await + .unwrap(); + for (target, diagnostic) in [ + ( + tcp_forward_init::Target::Tcp(TcpRelayTarget { + host: "127.0.0.1".into(), + port: 8081, + }), + "target port was not admitted", + ), + ( + tcp_forward_init::Target::Tcp(TcpRelayTarget { + host: "127.0.0.1".into(), + port: 8080, + }), + "sandbox is not ready", + ), + ( + tcp_forward_init::Target::Ssh(SshRelayTarget::default()), + "sandbox is not ready", + ), + ] { + let error = client + .forward_tcp(tokio_stream::iter([TcpForwardFrame { + payload: Some(tcp_forward_frame::Payload::Init(TcpForwardInit { + sandbox: "ingress-live".into(), + workspace: "default".into(), + target: Some(target), + ..Default::default() + })), + }])) + .await + .unwrap_err(); + assert!(error.message().contains(diagnostic), "{error}"); + } + server.abort(); + } + #[tokio::test] async fn missing_launch_signer_precedes_driver_validation_and_preserves_staged_archive() { let directory = tempfile::tempdir().unwrap(); @@ -7151,6 +7563,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -7226,6 +7639,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -7264,6 +7678,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -7346,6 +7761,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -7726,6 +8142,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -7789,6 +8206,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -8114,6 +8532,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -8185,6 +8604,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -8236,6 +8656,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -8280,6 +8701,7 @@ mod tests { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -8940,6 +9362,7 @@ mod tests { workload_template: "gpu-kata".to_string(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -9029,6 +9452,7 @@ mod tests { workload_template: "default-image".to_string(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -9083,6 +9507,7 @@ mod tests { workload_template: "default-gpu".to_string(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -9124,6 +9549,7 @@ mod tests { workload_template: "corrupt-template".to_string(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -9166,6 +9592,7 @@ mod tests { workload_template: "gpu-kata".to_string(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -9193,6 +9620,7 @@ mod tests { workload_template: "Invalid_Template_Name".to_string(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -9223,6 +9651,7 @@ mod tests { workload_template: "missing-template".to_string(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await @@ -9253,6 +9682,7 @@ mod tests { workload_template: String::new(), await_main_process_attachment: false, service_exposures: Vec::new(), + forward_ports: Vec::new(), }), ) .await diff --git a/crates/openshell-server/src/grpc/sandbox/image_update_tests.rs b/crates/openshell-server/src/grpc/sandbox/image_update_tests.rs index e83d786ee0..e00ccd1d21 100644 --- a/crates/openshell-server/src/grpc/sandbox/image_update_tests.rs +++ b/crates/openshell-server/src/grpc/sandbox/image_update_tests.rs @@ -135,7 +135,10 @@ async fn state() -> (Arc, Arc, Sandbox) { openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/image-test", "test", - [openshell_core::extension_protocol::COMPUTE_UPDATE_SANDBOX_CONFIGURATION.into()], + [ + openshell_core::extension_protocol::COMPUTE_UPDATE_SANDBOX_CONFIGURATION.into(), + openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION.into(), + ], )), ..Default::default() }), @@ -258,6 +261,20 @@ async fn public_image_update_replays_without_reapplying_or_touching_same_name_re .sandbox .unwrap(); assert_eq!(updated.object_id(), sandbox.object_id()); + assert_eq!( + updated.status.as_ref().unwrap().ingress_admission, + Some(openshell_core::proto::SandboxIngressAdmission::default()) + ); + let stored = state + .store + .get_message::(sandbox.object_id()) + .await + .unwrap() + .unwrap(); + assert!( + stored.status.as_ref().unwrap().ingress_admission.is_none(), + "inspection projection must not invent durable admission evidence" + ); assert_eq!( updated .spec diff --git a/crates/openshell-server/src/grpc/service.rs b/crates/openshell-server/src/grpc/service.rs index d063c315ad..ead82a6496 100644 --- a/crates/openshell-server/src/grpc/service.rs +++ b/crates/openshell-server/src/grpc/service.rs @@ -115,6 +115,7 @@ pub(super) async fn expose_service_endpoint( }) .ok_or_else(|| Status::not_found("sandbox removed"))?; let sandbox = ¤t; + state.compute.admit_ingress_port(sandbox, target_port)?; let sandbox_name = sandbox.object_name(); let now = crate::persistence::current_time_ms(); diff --git a/crates/openshell-server/src/service_routing.rs b/crates/openshell-server/src/service_routing.rs index ece2add1e0..ac4e035849 100644 --- a/crates/openshell-server/src/service_routing.rs +++ b/crates/openshell-server/src/service_routing.rs @@ -777,6 +777,23 @@ async fn proxy_to_endpoint( return Err(err); } + if state + .compute + .admit_ingress_port(&sandbox, endpoint.target_port) + .is_err() + { + let err = ServiceRouteError::endpoint_unavailable(); + emit_service_http_failure( + &state, + &req, + &sandbox_name, + &service_name, + Some(&endpoint), + &err, + ); + return Err(err); + } + let websocket_upgrade = is_websocket_upgrade(&req); let downstream_upgrade = websocket_upgrade.then(|| hyper::upgrade::on(&mut req)); @@ -2332,6 +2349,53 @@ mod tests { ); } + #[tokio::test] + async fn ingress_admission_is_checked_before_a_pooled_upstream_is_taken() { + use openshell_core::proto::{IngressScope, SandboxIngressAdmission}; + let state = crate::grpc::test_support::test_server_state().await; + let endpoint = endpoint(); + let mut sandbox = Sandbox { + metadata: Some(openshell_core::proto::datamodel::v1::ObjectMeta { + id: endpoint.sandbox_id.clone(), + name: endpoint.sandbox.clone(), + workspace: "default".into(), + ..Default::default() + }), + ..Default::default() + }; + sandbox.set_phase(SandboxPhase::Ready as i32); + sandbox.status.as_mut().unwrap().ingress_admission = Some(SandboxIngressAdmission { + scope: IngressScope::SharedHostLoopback as i32, + all_ports: false, + ports: vec![9090], + }); + state.store.put_message(&sandbox).await.unwrap(); + state.store.put_message(&endpoint).await.unwrap(); + let key = upstream_pool_key(endpoint.object_id(), 8080); + let (sender, _peer) = test_upstream().await; + state.service_upstreams.put(&key, sender); + let result = tokio::time::timeout( + Duration::from_millis(250), + proxy_to_endpoint( + state.clone(), + Request::builder().uri("/").body(Body::empty()).unwrap(), + "default", + endpoint.sandbox, + endpoint.name, + ), + ) + .await + .expect("an unadmitted route must reject before pooled I/O"); + assert_eq!( + result.unwrap_err().reason, + ServiceRouteError::endpoint_unavailable().reason + ); + assert!( + state.service_upstreams.take(&key).is_some(), + "admission must precede taking any pooled connection" + ); + } + #[tokio::test] async fn pool_hands_back_a_ready_upstream() { let pool = ServiceUpstreamPool::default(); diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index d8da420871..863502dff8 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -177,12 +177,15 @@ mod tests { // selections absent; storage.v1 remains frozen. Public field 3 retains its // image wire identity, now with explicit presence. Compute configuration // updates use a separate capability/service and add one classified RPC. + // Ingress admission adds a shared status message and scope enum; older + // stored sandboxes decode without an admission envelope. Forward targets + // extend only the public create request and are never stored as requests. const PUBLIC_RPC_SCHEMA_SHA256: &str = - "3b4c2af6cf22e7ed1d84279ff865acb0db7199717bee9bc041caef791334104d"; + "f798ef01e3df216851775f4ada271d7bb623b16017227292d98fdbbc4b0a0413"; const DURABLE_SCHEMA_SHA256: &str = - "cb15f8f79a855064a43fe8c6cd908bf07eca6604e2bf8ab7d2de0ef33e97da4e"; + "a8aa47c33c3fdb54a25c229a1d23600c2ae69daf3d0fa5191b88a8f4a9829dde"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = - "c0c3ea17f105aaa6d843c466fd4240451c2105dce7675384935d4aaecdac8abf"; + "afbb27866df733be657a985ce68702e515f40cb95764817d427a9578f9884897"; // Encoded with the schema that omitted stable_placeholder. Keep these // bytes fixed so the current encoder cannot hide a compatibility change. const LEGACY_STATIC_PROFILE: &str = "0a240a116c65676163792d70726f66696c652d6964120d6c65676163792d737461746963280712510a0d6c65676163792d73746174696312184c6567616379207374617469632063726564656e7469616c2a260a076170695f6b65791a1153594e5448455449435f4150495f4b455920012a06626561726572"; @@ -723,9 +726,9 @@ mod tests { overlap_hash.as_str(), ), ( - (337, 31), - (100, 22), - (86, 22), + (338, 32), + (101, 23), + (87, 23), PUBLIC_RPC_SCHEMA_SHA256, DURABLE_SCHEMA_SHA256, PUBLIC_DURABLE_OVERLAP_SHA256 @@ -783,6 +786,7 @@ mod tests { assert!(status.configuration_admission.is_none()); assert_eq!(status.configuration_activated, None); assert!(status.provisioning.is_none()); + assert!(status.ingress_admission.is_none()); assert!(!crate::policy_store::permits_initial_static_policy_repair( &sandbox )); diff --git a/crates/openshell-server/src/test_support.rs b/crates/openshell-server/src/test_support.rs index b1d5736a93..5ba516c09d 100644 --- a/crates/openshell-server/src/test_support.rs +++ b/crates/openshell-server/src/test_support.rs @@ -143,6 +143,7 @@ pub struct FakeComputeDriver { #[derive(Debug)] struct FakeComputeDriverState { capabilities: GetCapabilitiesResponse, + ingress_admission: Option, sandboxes: HashMap, calls: Vec, traceparents: Vec, @@ -165,10 +166,20 @@ impl FakeComputeDriver { self } + #[must_use] + pub fn with_ingress_admission( + self, + admission: Option, + ) -> Self { + self.with_state(|state| state.ingress_admission = admission); + self + } + #[must_use] pub fn new() -> Self { Self { state: Arc::new(Mutex::new(FakeComputeDriverState { + ingress_admission: None, capabilities: GetCapabilitiesResponse { resource_admission_policy: openshell_core::resource_admission::DriverAdmissionConfig::default() @@ -237,6 +248,24 @@ impl FakeComputeDriver { self } + pub fn set_ingress_capability(&self, supported: bool) { + self.with_state(|state| { + let capabilities = &mut state + .capabilities + .extension + .as_mut() + .unwrap() + .supported_capabilities; + capabilities.retain(|capability| { + capability != openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION + }); + if supported { + capabilities + .push(openshell_core::extension_protocol::COMPUTE_INGRESS_ADMISSION.into()); + } + }); + } + #[must_use] pub fn with_sandbox_image_updates(self) -> Self { self.with_state(|state| { @@ -440,7 +469,9 @@ impl ComputeDriver for FakeComputeDriver { sandbox: request.sandbox, }); }); - Ok(Response::new(ValidateSandboxCreateResponse {})) + Ok(Response::new(ValidateSandboxCreateResponse { + ingress_admission: self.with_state(|state| state.ingress_admission.clone()), + })) } async fn get_sandbox( diff --git a/crates/openshell-tui/src/lib.rs b/crates/openshell-tui/src/lib.rs index 0e807cd452..3a0ddc84e3 100644 --- a/crates/openshell-tui/src/lib.rs +++ b/crates/openshell-tui/src/lib.rs @@ -1495,6 +1495,7 @@ fn spawn_create_sandbox(app: &mut App, tx: mpsc::UnboundedSender) { await_main_process_attachment: false, workload_template: String::new(), service_exposures: Vec::new(), + forward_ports: ports.iter().map(|spec| u32::from(spec.port)).collect(), }; let sandbox_name = @@ -1523,7 +1524,7 @@ fn spawn_create_sandbox(app: &mut App, tx: mpsc::UnboundedSender) { // If ports or command are set, wait for Ready before finishing. if need_ready { let mut attempts = 0; - let _sandbox_id = loop { + let ready_sandbox = loop { attempts += 1; if attempts > 150 { let _ = tx.send(Event::CreateResult(Err( @@ -1544,7 +1545,7 @@ fn spawn_create_sandbox(app: &mut App, tx: mpsc::UnboundedSender) { && let Some(sandbox) = resp.into_inner().sandbox { if sandbox.phase() == SandboxPhase::Ready as i32 { - break sandbox.object_id().to_string(); + break sandbox; } if sandbox.phase() == SandboxPhase::Error as i32 { let _ = tx.send(Event::CreateResult(Err( @@ -1555,6 +1556,25 @@ fn spawn_create_sandbox(app: &mut App, tx: mpsc::UnboundedSender) { } }; + // Inspect admission before creating any SSH sessions or listeners. + for spec in &ports { + if let Err(error) = openshell_core::ingress::admit_port( + ready_sandbox + .status + .as_ref() + .and_then(|status| status.ingress_admission.as_ref()), + u32::from(spec.port), + false, + ) { + let _ = tx.send(Event::ForwardWarnings(vec![format!( + "port {}: {error}", + spec.port + )])); + let _ = tx.send(Event::CreateResult(Ok((sandbox_name, workspace)))); + return; + } + } + // Start port forwards if requested. if !ports.is_empty() { let forward_warnings = start_port_forwards( diff --git a/docs/extensibility/isolation-backends.mdx b/docs/extensibility/isolation-backends.mdx index 8dd5903176..c8743a6555 100644 --- a/docs/extensibility/isolation-backends.mdx +++ b/docs/extensibility/isolation-backends.mdx @@ -39,6 +39,10 @@ accept an operator-selected `IsolationBackendRegistration`, including its Unix, TCP, or vsock endpoint. The driver provisions the resource and delivers the backend's trusted launch metadata to the supervisor. +Compute drivers that advertise `openshell.compute.ingress-admission` return +the supported forwarding scope and allowed ports from `ValidateSandboxCreate`. +The isolation backend must enforce those limits when connecting to a service. + ## Who Owns What | Component | Owns | @@ -50,3 +54,8 @@ backend's trusted launch metadata to the supervisor. For how `openshell-sandbox` enforces the boundary, refer to [Inside the Sandbox Boundary](/about/architecture#inside-the-sandbox-boundary). For driver-specific workload behavior, refer to [Runtimes](/how-it-works/sandboxes/runtimes). + +## Release compatibility + +Upgrade gateway replicas together. For source compatibility changes, see the +[SDK migration notes](https://github.com/NVIDIA/OpenShell/blob/main/crates/openshell-sdk/README.md#next-minor-api-migration). diff --git a/docs/how-it-works/sandboxes/overview.mdx b/docs/how-it-works/sandboxes/overview.mdx index 6365f9e684..0c252504a2 100644 --- a/docs/how-it-works/sandboxes/overview.mdx +++ b/docs/how-it-works/sandboxes/overview.mdx @@ -826,6 +826,12 @@ Press `c` in the Sandboxes panel to create a sandbox. In the optional Command fi ## Port Forwarding +Compute drivers can restrict which service ports you can forward. Declare the +target port at creation with `--forward `. Disallowed ports are rejected +before a local listener opens. If forwarding reports missing admission after an +upgrade, recreate the sandbox with the required port. A shared-host loopback +target is an operator-approved service on the compute host. + Forward a local port to a running sandbox to access services inside it, such as a web server or database: ```shell diff --git a/examples/isolation-backend-v8/src/driver.rs b/examples/isolation-backend-v8/src/driver.rs index 5bbdc8bcef..1be2af3505 100644 --- a/examples/isolation-backend-v8/src/driver.rs +++ b/examples/isolation-backend-v8/src/driver.rs @@ -384,7 +384,7 @@ impl compute_driver_server::ComputeDriver for DriverService { .as_ref() .ok_or_else(|| Status::invalid_argument("sandbox required"))?, )?; - Ok(Response::new(ValidateSandboxCreateResponse {})) + Ok(Response::new(ValidateSandboxCreateResponse::default())) } async fn create_sandbox( &self, diff --git a/proto/compute_driver.proto b/proto/compute_driver.proto index 88d090fc18..9be797e9d4 100644 --- a/proto/compute_driver.proto +++ b/proto/compute_driver.proto @@ -449,7 +449,12 @@ message ValidateSandboxCreateRequest { DriverSandbox sandbox = 1; } -message ValidateSandboxCreateResponse {} +message ValidateSandboxCreateResponse { + // Required when openshell.compute.ingress-admission is negotiated. Complete + // immutable envelope, including an explicit unsupported result. Create/start + // must enforce this envelope from the same retained template/configuration. + openshell.sandbox.v1.SandboxIngressAdmission ingress_admission = 1; +} message GetSandboxRequest { // Stable sandbox ID stored by the gateway. diff --git a/proto/openshell.proto b/proto/openshell.proto index b40ec68263..081cad9a7b 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -1258,6 +1258,9 @@ message SandboxStatus { // failures before a configured endpoint or credential could be selected. // Matching grant success or a policy, provider, or supervisor reset clears it. TokenGrantFailureObservation last_token_grant_failure = 17; + // Gateway-retained immutable compute-driver ingress admission. Absence is + // legacy behavior only when the selected driver has not negotiated ingress. + openshell.sandbox.v1.SandboxIngressAdmission ingress_admission = 19; } // User-facing sandbox condition derived from platform or gateway observations. @@ -1344,6 +1347,9 @@ message CreateSandboxRequest { // registered after the sandbox has been persisted and route only while the // sandbox is ready. repeated SandboxServiceExposure service_exposures = 9; + // Requested local forward targets, checked before provisioning. This does + // not open a listener; clients establish the admitted forwards after create. + repeated uint32 forward_ports = 10; } // Update selected configuration of a stopped sandbox. Omitted fields are unchanged. diff --git a/proto/sandbox.proto b/proto/sandbox.proto index b57c5d4cda..65be37f18d 100644 --- a/proto/sandbox.proto +++ b/proto/sandbox.proto @@ -17,6 +17,23 @@ import "datamodel.proto"; // - Public sandbox resource types live in `openshell.proto`. // - Internal compute-driver sandbox observation types live in `compute_driver.proto`. +// Immutable ingress envelope admitted from the compute driver's resolved inputs. +// Presence with an unspecified scope is invalid; it never means legacy support. +message SandboxIngressAdmission { + IngressScope scope = 1; + // Mutually exclusive with ports. False with an empty list admits no ports. + bool all_ports = 2; + repeated uint32 ports = 3; +} + +enum IngressScope { + INGRESS_SCOPE_UNSPECIFIED = 0; + INGRESS_SCOPE_UNSUPPORTED = 1; + INGRESS_SCOPE_ISOLATED_LOOPBACK = 2; + // The driver has admitted its operator-controlled shared-host opt-in. + INGRESS_SCOPE_SHARED_HOST_LOOPBACK = 3; +} + // Sandbox security policy configuration. message SandboxPolicy { // Policy version. diff --git a/sdk/typescript/src/client.test.ts b/sdk/typescript/src/client.test.ts index 756ff6e44d..cbc4804842 100644 --- a/sdk/typescript/src/client.test.ts +++ b/sdk/typescript/src/client.test.ts @@ -703,6 +703,7 @@ describe('create', () => { it('sends create-time service exposures', async () => { let created: { serviceExposures?: Array<{ service?: string; targetPort?: number; authorizationMode?: number }>; + forwardPorts?: number[]; } = {}; const sandbox = client({ createSandbox: (req) => { @@ -719,6 +720,7 @@ describe('create', () => { const result = await sandbox.create({ image: 'img', + forwardPorts: [8080], serviceExposures: [ { targetPort: 4500 }, { @@ -729,6 +731,7 @@ describe('create', () => { ], }); + expect(created.forwardPorts).toEqual([8080]); expect( created.serviceExposures?.map(({ service, targetPort, authorizationMode }) => ({ service, @@ -2168,6 +2171,18 @@ describe('ssh sessions', () => { }); describe('forward', () => { + it('rejects an unadmitted target before creating a local forward handle', async () => { + const sandbox = client({ + getSandbox: () => ({ + sandbox: { + metadata: { id: 'sb-id', name: 'sb', workspace: 'default' }, + status: { phase: SandboxPhase.READY, ingressAdmission: { scope: 3, allPorts: false, ports: [8080] } }, + }, + }), + }); + await expect(sandbox.forward('sb', { targetPort: 9090 })).rejects.toMatchObject({ code: 'invalid_config' }); + }); + it('binds a local port and relays bytes both ways, minting + revoking a token', async () => { let sshReq: ScopedRequest = {}; let revokedToken: string | undefined; diff --git a/sdk/typescript/src/client.ts b/sdk/typescript/src/client.ts index 18eeb2f698..448c25e5fe 100644 --- a/sdk/typescript/src/client.ts +++ b/sdk/typescript/src/client.ts @@ -38,7 +38,13 @@ import { type TcpForwardFrameSchema, } from './gen/openshell_pb.js'; import type { EffectiveSetting, GetSandboxConfigResponse, SandboxPolicy, SettingValue } from './gen/sandbox_pb.js'; -import { PolicySource, type SandboxPolicySchema, SettingScope, type SettingValueSchema } from './gen/sandbox_pb.js'; +import { + IngressScope, + PolicySource, + type SandboxPolicySchema, + SettingScope, + type SettingValueSchema, +} from './gen/sandbox_pb.js'; import { ProviderClient } from './providers.js'; import { heldUntilReady, OWNER_REPLICA_HEADER, OWNER_REQUEST_HEADER, routedStream } from './routing.js'; import { validateSshResponse } from './ssh-validate.js'; @@ -170,6 +176,8 @@ export interface SandboxSpec { tty?: boolean; /** Loopback HTTP services to expose when the sandbox is created. */ serviceExposures?: ServiceExposure[]; + /** Target ports to admit before provisioning; does not open listeners. */ + forwardPorts?: number[]; /** Restart behavior after the canonical main process exits. */ restartPolicy?: SandboxRestartPolicyName; /** @@ -191,7 +199,7 @@ export interface SandboxSpec { export interface ServiceExposure { /** Service name. Empty or omitted selects the unnamed endpoint. */ service?: string; - /** Loopback TCP port inside the sandbox. */ + /** TCP port in the compute-admitted loopback namespace. */ targetPort: number; /** Handling for an incoming application Authorization header. */ authorizationMode?: ServiceAuthorizationMode; @@ -225,6 +233,8 @@ export interface SandboxFromTemplateSpec { tty?: boolean; /** Loopback HTTP services to expose when the sandbox is created. */ serviceExposures?: ServiceExposure[]; + /** Target ports to admit before provisioning; does not open listeners. */ + forwardPorts?: number[]; /** * Create-time sandbox policy (the safety boundary). The named workload * template supplies runtime workload fields. @@ -232,7 +242,14 @@ export interface SandboxFromTemplateSpec { policy?: MessageInitShape; } +export interface SandboxIngressAdmission { + scope: 'unknown' | 'unsupported' | 'isolated_loopback' | 'shared_host_loopback'; + allPorts: boolean; + ports: number[]; +} + export interface SandboxRef { + ingressAdmission?: SandboxIngressAdmission; id: string; name: string; workspace: string; @@ -470,9 +487,9 @@ export interface WaitDeletedOptions extends WaitOptions { } export interface ForwardOptions extends SandboxWorkspaceOptions { - /** Loopback TCP port inside the sandbox to dial. */ + /** TCP port in the compute-admitted loopback namespace to dial. */ targetPort: number; - /** Target host inside the sandbox (loopback only). Default 127.0.0.1. */ + /** Target host in the admitted loopback namespace. Default 127.0.0.1. */ targetHost?: string; /** Local port to bind. Default 0 (ephemeral). */ localPort?: number; @@ -631,7 +648,22 @@ function sandboxRef(sandbox: Sandbox | undefined, serviceUrls: Record ({ service: exposure.service ?? '', @@ -1271,6 +1304,7 @@ export class SandboxClient { policy: spec.policy, }, workloadTemplate: spec.workloadTemplate, + forwardPorts: [...(spec.forwardPorts ?? [])], serviceExposures: spec.serviceExposures?.map((exposure) => ({ service: exposure.service ?? '', @@ -1746,6 +1780,9 @@ export class SandboxClient { // the `init` (TCP target + token), relay bytes both ways in ~64 KiB chunks, // and revoke the token on close. Process-lifetime only. async forward(name: string, opts: ForwardOptions): Promise { + if (!Number.isInteger(opts.targetPort) || opts.targetPort < 1 || opts.targetPort > 65535) { + throw new SdkError('invalid_config', 'target port must be between 1 and 65535'); + } const targetHost = opts.targetHost ?? '127.0.0.1'; const targetPort = opts.targetPort; const localHost = opts.localHost ?? '127.0.0.1'; @@ -1762,6 +1799,17 @@ export class SandboxClient { if (ref.phase !== 'ready') { throw new SdkError('connect', `sandbox '${name}' is not ready (phase: ${ref.phase})`); } + const ingress = ref.ingressAdmission; + if ( + ingress && + ((ingress.scope !== 'isolated_loopback' && ingress.scope !== 'shared_host_loopback') || + (ingress.allPorts && ingress.ports.length > 0) || + ingress.ports.some((port) => !Number.isInteger(port) || port < 1 || port > 65535) || + new Set(ingress.ports).size !== ingress.ports.length || + (!ingress.allPorts && !ingress.ports.includes(targetPort))) + ) { + throw new SdkError('invalid_config', 'target port was not admitted by the compute driver'); + } sandboxId = ref.id; owner = found.owner; } catch (e) { diff --git a/sdk/typescript/src/index.ts b/sdk/typescript/src/index.ts index 4948c13623..f967f3af66 100644 --- a/sdk/typescript/src/index.ts +++ b/sdk/typescript/src/index.ts @@ -31,6 +31,7 @@ export type { ProviderRef, SandboxConfig, SandboxFromTemplateSpec, + SandboxIngressAdmission, SandboxPhaseName, SandboxPolicy, SandboxRef, diff --git a/skills/openshell-cli/SKILL.md b/skills/openshell-cli/SKILL.md index 1265b8a610..51779827a7 100644 --- a/skills/openshell-cli/SKILL.md +++ b/skills/openshell-cli/SKILL.md @@ -836,6 +836,9 @@ CLI forwards ignore SSH multiplexing and automatic backgrounding settings in the user's SSH config. Only background forwards are tracked by `forward list` and managed by `forward stop`; foreground forwards end when the command exits. +Declare required target ports at creation with `--forward`. For shared-host +loopback targets, choose a different local port if the target already uses it. + Manage or iterate on the sandbox: ```bash diff --git a/tests/testkit/testkit/src/context.rs b/tests/testkit/testkit/src/context.rs index 7f09ad7397..64bf7680fc 100644 --- a/tests/testkit/testkit/src/context.rs +++ b/tests/testkit/testkit/src/context.rs @@ -411,12 +411,14 @@ fn policy_create_request( command, tty, service_exposures: _, + forward_ports, restart_policy, } = spec; proto::CreateSandboxRequest { name: name.unwrap_or_default(), labels, workspace_scope, + forward_ports: forward_ports.into_iter().map(u32::from).collect(), spec: Some(proto::SandboxSpec { environment, template: image.map(|image| proto::SandboxTemplate { @@ -679,13 +681,14 @@ mod tests { let spec = SandboxSpec { image: Some("python:3".to_string()), providers: vec!["provider".to_string()], + forward_ports: vec![8080, 9090], ..SandboxSpec::default() }; let policy = crate::policy::base(); - let spec = policy_create_request(spec, policy.clone(), None) - .spec - .expect("spec"); + let request = policy_create_request(spec, policy.clone(), None); + assert_eq!(request.forward_ports, [8080, 9090]); + let spec = request.spec.expect("spec"); assert_eq!(spec.template.expect("template").image, "python:3"); assert_eq!(spec.providers, ["provider"]);