From ad168092d15c8bc12660884cbb4c7fb0e165ca02 Mon Sep 17 00:00:00 2001 From: Shiju Date: Tue, 3 Nov 2026 17:11:01 +0530 Subject: [PATCH 1/4] feat(policy): admit backend filesystem prerequisites before launch Proxy prerequisites may exist inside the workload namespace rather than on the gateway host. This lets compute drivers declare those paths before launch and keeps create admission, delivered policy and endpoint evidence tied to the same effective policy and provider snapshot. Signed-off-by: Shiju --- .../openshell-core/src/extension_protocol.rs | 3 + .../openshell-core/src/filesystem_baseline.rs | 101 +++++++ crates/openshell-core/src/lib.rs | 1 + crates/openshell-driver-docker/src/lib.rs | 1 + .../openshell-driver-kubernetes/src/driver.rs | 1 + crates/openshell-driver-mxc/src/driver.rs | 1 + crates/openshell-driver-podman/src/driver.rs | 1 + crates/openshell-driver-vm/src/driver.rs | 1 + crates/openshell-server/src/compute/mod.rs | 116 ++++++++ crates/openshell-server/src/grpc/policy.rs | 259 +++++++++++++++++- .../src/grpc/policy/endpoint_status.rs | 88 +++--- crates/openshell-server/src/grpc/sandbox.rs | 1 + crates/openshell-server/src/test_support.rs | 1 + .../openshell-supervisor/src/backend_setup.rs | 21 ++ .../src/legacy_startup_regression_tests.rs | 2 +- crates/openshell-supervisor/src/lib.rs | 190 +++++++------ docs/extensibility/isolation-backends.mdx | 8 + proto/compute_driver.proto | 11 + 18 files changed, 675 insertions(+), 132 deletions(-) create mode 100644 crates/openshell-core/src/filesystem_baseline.rs diff --git a/crates/openshell-core/src/extension_protocol.rs b/crates/openshell-core/src/extension_protocol.rs index d6d93418d8..0052f14689 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"; +/// Driver declares the proxy prerequisites admitted before runtime signing. +pub const COMPUTE_PROXY_FILESYSTEM_BASELINE: &str = "openshell.compute.proxy-filesystem-baseline"; 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_PROXY_FILESYSTEM_BASELINE.to_string()); } PeerMetadata { protocol_version: Some(ProtocolVersion { diff --git a/crates/openshell-core/src/filesystem_baseline.rs b/crates/openshell-core/src/filesystem_baseline.rs new file mode 100644 index 0000000000..befff771b0 --- /dev/null +++ b/crates/openshell-core/src/filesystem_baseline.rs @@ -0,0 +1,101 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Trusted backend prerequisites composed before effective policy admission. + +use std::collections::BTreeSet; + +use crate::proto::SandboxPolicy; +use crate::proto::compute::v1::ProxyFilesystemBaseline; + +/// Proxy runtime paths selected by trusted backend composition. +/// An explicit empty baseline suppresses legacy defaults. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct FilesystemBaseline { + pub read_only: Vec, + pub read_write: Vec, +} + +impl From for FilesystemBaseline { + fn from(value: ProxyFilesystemBaseline) -> Self { + Self { + read_only: value.read_only, + read_write: value.read_write, + } + } +} + +impl From for ProxyFilesystemBaseline { + fn from(value: FilesystemBaseline) -> Self { + Self { + read_only: value.read_only, + read_write: value.read_write, + } + } +} + +impl FilesystemBaseline { + /// Validate bounded, canonical POSIX path syntax without inspecting the + /// gateway host filesystem, which need not be the workload's platform. + pub fn validate(&self) -> Result<(), &'static str> { + if self.read_only.len() + self.read_write.len() > 64 { + return Err("proxy filesystem baseline has too many paths"); + } + let mut seen = BTreeSet::new(); + for path in self.read_only.iter().chain(&self.read_write) { + if path.len() > 4096 + || !path.starts_with('/') + || path.chars().any(char::is_control) + || (path != "/" + && path[1..] + .split('/') + .any(|part| matches!(part, "" | "." | ".."))) + || !seen.insert(path) + { + return Err( + "proxy filesystem baseline requires unique canonical POSIX absolute paths", + ); + } + } + Ok(()) + } + + /// Add prerequisites to a proxy-mode policy without replacing explicit + /// grants. An explicit read-only path stays read-only. `path_exists` is + /// used only by legacy image-local Linux defaults; advertised prerequisites + /// are unconditional and callers pass a predicate that always returns true. + /// Validate the descriptor at the trusted admission boundary before use. + pub fn enrich(&self, policy: &mut SandboxPolicy, path_exists: impl Fn(&str) -> bool) -> bool { + if policy.network_policies.is_empty() + || (self.read_only.is_empty() && self.read_write.is_empty()) + { + return false; + } + let filesystem = policy + .filesystem + .get_or_insert_with(|| crate::proto::FilesystemPolicy { + include_workdir: true, + ..Default::default() + }); + let mut modified = false; + for path in &self.read_only { + if !filesystem.read_only.contains(path) + && !filesystem.read_write.contains(path) + && path_exists(path) + { + filesystem.read_only.push(path.clone()); + modified = true; + } + } + for path in &self.read_write { + if !filesystem.read_only.contains(path) + && !filesystem.read_write.contains(path) + && path_exists(path) + { + filesystem.read_write.push(path.clone()); + modified = true; + } + } + modified + } +} diff --git a/crates/openshell-core/src/lib.rs b/crates/openshell-core/src/lib.rs index 12d6cd26ff..8a3d49ddc8 100644 --- a/crates/openshell-core/src/lib.rs +++ b/crates/openshell-core/src/lib.rs @@ -29,6 +29,7 @@ pub mod error; pub mod extension_protocol; #[cfg(unix)] pub mod external_driver_socket; +pub mod filesystem_baseline; pub mod forward; pub mod google_cloud; pub mod gpu; diff --git a/crates/openshell-driver-docker/src/lib.rs b/crates/openshell-driver-docker/src/lib.rs index e5421431f8..78071dda0d 100644 --- a/crates/openshell-driver-docker/src/lib.rs +++ b/crates/openshell-driver-docker/src/lib.rs @@ -1093,6 +1093,7 @@ impl DockerComputeDriver { }), rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: Some(openshell_core::extension_protocol::extension_metadata( openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/docker", diff --git a/crates/openshell-driver-kubernetes/src/driver.rs b/crates/openshell-driver-kubernetes/src/driver.rs index e1a4226187..9707421323 100644 --- a/crates/openshell-driver-kubernetes/src/driver.rs +++ b/crates/openshell-driver-kubernetes/src/driver.rs @@ -1319,6 +1319,7 @@ impl KubernetesComputeDriver { }), rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: Some(openshell_core::extension_protocol::extension_metadata( openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/kubernetes", diff --git a/crates/openshell-driver-mxc/src/driver.rs b/crates/openshell-driver-mxc/src/driver.rs index 2772f1a69c..9f5492ce35 100644 --- a/crates/openshell-driver-mxc/src/driver.rs +++ b/crates/openshell-driver-mxc/src/driver.rs @@ -476,6 +476,7 @@ impl MxcComputeBackend { resource_capabilities: None, rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: Some(openshell_core::extension_protocol::extension_metadata( openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/mxc", diff --git a/crates/openshell-driver-podman/src/driver.rs b/crates/openshell-driver-podman/src/driver.rs index 6bea5b7f16..5f07d80b40 100644 --- a/crates/openshell-driver-podman/src/driver.rs +++ b/crates/openshell-driver-podman/src/driver.rs @@ -740,6 +740,7 @@ impl PodmanComputeDriver { }), rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: Some(openshell_core::extension_protocol::extension_metadata( openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/podman", diff --git a/crates/openshell-driver-vm/src/driver.rs b/crates/openshell-driver-vm/src/driver.rs index f7946c34c2..eb6aa54fda 100644 --- a/crates/openshell-driver-vm/src/driver.rs +++ b/crates/openshell-driver-vm/src/driver.rs @@ -1574,6 +1574,7 @@ impl VmDriver { .to_string_lossy() .into_owned(), rootfs_tar_max_bytes: self.config.rootfs_tar_max_bytes(), + proxy_filesystem_baseline: None, extension: Some(extension), } } diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 4ba1311d54..4292e4ec99 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -31,6 +31,7 @@ use hyper_util::rt::TokioIo; use openshell_core::extension_protocol::{ ExtensionFamily, NegotiatedExtension, gateway_metadata, negotiate, }; +use openshell_core::filesystem_baseline::FilesystemBaseline; #[cfg(test)] use openshell_core::proto::ServiceAuthorizationMode; use openshell_core::proto::compute::v1::{ @@ -518,6 +519,8 @@ pub struct ComputeDriverInfoSnapshot { pub driver_reports_runtime_readiness: bool, /// Static portable resource request forms from the startup capability snapshot. pub resource_capabilities: Option, + /// Trusted workload proxy prerequisites, present only after negotiation. + pub proxy_filesystem_baseline: Option, /// Directory where rootfs tar files must be staged. pub rootfs_tar_staging_dir: String, /// Maximum rootfs tar file size in bytes. @@ -1024,6 +1027,32 @@ impl ComputeRuntime { capabilities.extension.clone(), ) .map_err(|error| ComputeError::Message(error.to_string()))?; + let declares_baseline = + negotiated_extension + .supported_capabilities + .iter() + .any(|capability| { + capability + == openshell_core::extension_protocol::COMPUTE_PROXY_FILESYSTEM_BASELINE + }); + let proxy_filesystem_baseline = match ( + declares_baseline, + capabilities.proxy_filesystem_baseline.clone(), + ) { + (false, None) => None, + (true, Some(descriptor)) => { + let baseline = FilesystemBaseline::from(descriptor); + baseline + .validate() + .map_err(|error| ComputeError::Message(error.to_string()))?; + Some(baseline) + } + _ => { + return Err(ComputeError::Message( + "compute proxy filesystem baseline capability and descriptor must agree".into(), + )); + } + }; info!( configured_driver = %driver_name, advertised_driver = %capabilities.driver_name, @@ -1038,6 +1067,7 @@ impl ComputeRuntime { supports_sandbox_authentication: capabilities.supports_sandbox_authentication, driver_reports_runtime_readiness: capabilities.driver_reports_runtime_readiness, resource_capabilities: capabilities.resource_capabilities, + proxy_filesystem_baseline, rootfs_tar_staging_dir: capabilities.rootfs_tar_staging_dir, rootfs_tar_max_bytes: capabilities.rootfs_tar_max_bytes, }; @@ -1363,6 +1393,10 @@ impl ComputeRuntime { std::slice::from_ref(&self.driver_info) } + pub(crate) fn proxy_filesystem_baseline(&self) -> Option<&FilesystemBaseline> { + self.driver_info.proxy_filesystem_baseline.as_ref() + } + #[must_use] pub(crate) fn rootfs_tar_staging(&self) -> &rootfs_tar::RootfsTarStagingRegistry { &self.rootfs_tar_staging @@ -8545,6 +8579,7 @@ impl ComputeDriver for NoopTestDriver { resource_capabilities: None, rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: Some(openshell_core::extension_protocol::extension_metadata( ExtensionFamily::Compute, "openshell/noop-test-driver", @@ -8760,6 +8795,7 @@ pub fn new_test_runtime_with_driver( resource_capabilities: None, rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, }, telemetry_compute_driver: TelemetryComputeDriver::custom(), driver_process: None, @@ -9202,6 +9238,7 @@ mod tests { resource_capabilities: None, rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: (!self.omit_protocol_metadata).then(|| { let mut metadata = openshell_core::extension_protocol::extension_metadata( ExtensionFamily::Compute, @@ -9758,6 +9795,7 @@ mod tests { resource_capabilities: None, rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: Some(openshell_core::extension_protocol::extension_metadata( ExtensionFamily::Compute, "openshell/controlled-test-driver", @@ -10059,6 +10097,7 @@ mod tests { resource_capabilities: None, rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, }, telemetry_compute_driver: TelemetryComputeDriver::custom(), driver_process: None, @@ -19184,6 +19223,83 @@ endpoint = { kind = "unix", socket_path = "/run/agent-substrate/isolation.sock" assert!(runtime.image_update_driver.is_none()); } + #[tokio::test] + async fn proxy_filesystem_baseline_handshake_requires_a_valid_matching_descriptor() { + use crate::test_support::FakeComputeDriver; + use openshell_core::extension_protocol::COMPUTE_PROXY_FILESYSTEM_BASELINE; + use openshell_core::proto::compute::v1::ProxyFilesystemBaseline; + + let valid = ProxyFilesystemBaseline { + read_only: vec!["/System".into()], + read_write: vec!["/dev".into()], + }; + for (declared, descriptor, accepted) in [ + (false, None, true), + (true, Some(ProxyFilesystemBaseline::default()), true), + (true, Some(valid.clone()), true), + (true, None, false), + (false, Some(valid), false), + ( + true, + Some(ProxyFilesystemBaseline { + read_only: vec!["/a/../b".into()], + ..Default::default() + }), + false, + ), + ( + true, + Some(ProxyFilesystemBaseline { + read_only: vec!["/same".into()], + read_write: vec!["/same".into()], + }), + false, + ), + ( + true, + Some(ProxyFilesystemBaseline { + read_only: vec!["/a".into(); 65], + ..Default::default() + }), + false, + ), + ] { + let expected = descriptor.clone().map(FilesystemBaseline::from); + let driver = Arc::new(FakeComputeDriver::new().with_capabilities( + GetCapabilitiesResponse { + proxy_filesystem_baseline: descriptor, + extension: Some(openshell_core::extension_protocol::extension_metadata( + ExtensionFamily::Compute, + "openshell/test", + "test", + if declared { + vec![COMPUTE_PROXY_FILESYSTEM_BASELINE.to_string()] + } else { + vec![] + }, + )), + ..Default::default() + }, + )); + let store = Arc::new(Store::connect("sqlite::memory:").await.unwrap()); + let result = ComputeRuntime::from_driver( + "test".into(), + driver, + None, + store, + SandboxIndex::new(), + SandboxWatchBus::new(), + TracingLogBus::new(), + Arc::new(SupervisorSessionRegistry::new()), + ) + .await; + assert_eq!(result.is_ok(), accepted); + if let Ok(runtime) = result { + assert_eq!(runtime.proxy_filesystem_baseline(), expected.as_ref()); + } + } + } + #[tokio::test] async fn negotiated_launch_requirement_rejects_create_before_driver_or_store() { let driver = Arc::new(TestDriver { diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index ee6f07f9b1..07fb3e0e5d 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -34,6 +34,7 @@ use crate::auth::workspace_authz::{ }; use crate::compute::MutationScope; use crate::pagination::Pagination; +#[cfg(test)] use crate::persistence::ObjectType; use crate::persistence::{ DraftChunkRecord, ObjectId, ObjectListQuery, ObjectName, ObjectWorkspace, PolicyRecord, Store, @@ -2000,6 +2001,9 @@ async fn current_effective_policy_from_records( .await .map_err(|e| Status::internal(format!("fetch latest policy failed: {e}")))? }; + // This projection also owns authored-policy proposal/rebase identities. + // Trusted runtime prerequisites are composed at launch/config/credential + // delivery boundaries, not into editable policy history or review hashes. effective_policy_from_captured_inputs( catalog, records, @@ -3024,6 +3028,10 @@ pub async fn build_sandbox_config_snapshot_from_inputs( policy = Some(effective_policy); } + if let Some(effective_policy) = policy.as_mut() { + apply_backend_proxy_baseline(state, effective_policy, policy_source)?; + } + let policy_credential_bindings = policy_static_credential_endpoint_bindings(policy.as_ref())?; extend_credentialed_scopes_from_policy_bindings( &mut provider_policy_context.credentialed_scopes, @@ -3042,6 +3050,7 @@ pub async fn build_sandbox_config_snapshot_from_inputs( if let Err(error) = validate_uninspected_credentialed_endpoints(effective_policy) { configuration_error = bounded_configuration_diagnostic(error.message()); } + validate_backend_policy_size(state, effective_policy)?; policy_hash = deterministic_policy_hash(effective_policy); } @@ -3302,6 +3311,7 @@ pub(super) async fn compute_provider_env_revision_with_catalog( .await } +#[cfg(test)] async fn compute_provider_env_revision_with_catalog_and_policy_bindings( store: &Store, refresh: &dyn CredentialRefresh, @@ -3789,6 +3799,44 @@ pub(super) async fn handle_get_gateway_config( })) } +fn apply_backend_proxy_baseline( + state: &ServerState, + effective_policy: &mut ProtoSandboxPolicy, + source: PolicySource, +) -> Result<(), Status> { + // Compose trusted workload-namespace prerequisites after provider/global + // selection, before compute admission and effective configuration identity. + if let Some(baseline) = state.compute.proxy_filesystem_baseline() + && baseline.enrich(effective_policy, |_| true) + { + *effective_policy = validate_and_canonicalize_policy(effective_policy.clone())?; + validate_policy_safety(effective_policy)?; + } + if state.compute.proxy_filesystem_baseline().is_some() && matches!(source, PolicySource::Global) + { + // Global credential authorities hash the complete effective policy. + // Re-stamp after enrichment so create, config and credential delivery agree. + openshell_core::policy_identity::stamp_global_token_grant_owners(effective_policy); + } + + validate_backend_policy_size(state, effective_policy)?; + Ok(()) +} + +fn validate_backend_policy_size( + state: &ServerState, + policy: &ProtoSandboxPolicy, +) -> Result<(), Status> { + if state.compute.proxy_filesystem_baseline().is_some() + && policy.encoded_len() > super::MAX_POLICY_SIZE + { + return Err(Status::failed_precondition( + "backend proxy filesystem baseline exceeds the effective policy size limit", + )); + } + Ok(()) +} + /// Resolve the effective create-time policy without exporting credential state. /// Provider refresh and secret delivery remain in the supervisor session. pub(super) async fn resolve_sandbox_create_runtime_inputs( @@ -3813,7 +3861,7 @@ pub(super) async fn resolve_sandbox_create_runtime_inputs( provider_names, ) .await?; - let (effective_policy, _) = current_effective_policy_from_records( + let (mut effective_policy, source) = current_effective_policy_from_records( state, &provider_profile_catalog, sandbox, @@ -3821,6 +3869,7 @@ pub(super) async fn resolve_sandbox_create_runtime_inputs( &provider_records, ) .await?; + apply_backend_proxy_baseline(state, &mut effective_policy, source)?; let policy_credential_bindings = policy_static_credential_endpoint_bindings(Some(&effective_policy))?; validate_policy_credential_binding_context( @@ -3904,9 +3953,10 @@ struct ProviderEnvironmentTarget<'a> { provider_env_revision: u64, } -fn provider_environment_target( - inputs: &SandboxConfigInputs, -) -> Result, Status> { +fn provider_environment_target<'a>( + state: &ServerState, + inputs: &'a SandboxConfigInputs, +) -> Result, Status> { let sandbox = &inputs.sandbox; let spec = sandbox .spec @@ -3914,13 +3964,14 @@ fn provider_environment_target( .ok_or_else(|| Status::internal("sandbox has no spec"))?; let provider_profile_catalog = &inputs.catalog; let provider_records = &inputs.provider_records; - let (effective_policy, policy_source) = effective_policy_from_captured_inputs( + let (mut effective_policy, policy_source) = effective_policy_from_captured_inputs( provider_profile_catalog, provider_records, decode_policy_from_global_settings(&inputs.global_settings)?, inputs.latest_policy.as_ref(), sandbox, )?; + apply_backend_proxy_baseline(state, &mut effective_policy, policy_source)?; let policy_credential_bindings = policy_static_credential_endpoint_bindings(Some(&effective_policy))?; validate_policy_credential_binding_context( @@ -3959,7 +4010,7 @@ pub async fn build_provider_environment_snapshot_from_inputs( policy_source, policy_credential_bindings, provider_env_revision, - } = provider_environment_target(inputs)?; + } = provider_environment_target(state, inputs)?; let provider_names = &spec.providers; let provider_profile_catalog = &inputs.catalog; @@ -9663,6 +9714,202 @@ mod tests { assert!(inputs.launch_authentication.is_none()); } + #[tokio::test] + async fn backend_proxy_baseline_is_admitted_before_create_and_config_identity() { + use openshell_core::extension_protocol::{ + COMPUTE_PROXY_FILESYSTEM_BASELINE, ExtensionFamily, extension_metadata, + }; + use openshell_core::proto::compute::v1::{ + GetCapabilitiesResponse, ProxyFilesystemBaseline, + }; + + let mut state = test_server_state().await; + let baseline = ProxyFilesystemBaseline { + read_only: vec!["/platform-runtime-not-on-gateway".into()], + read_write: vec!["/platform-device-not-on-gateway".into()], + }; + let driver = Arc::new( + crate::test_support::FakeComputeDriver::new().with_capabilities( + GetCapabilitiesResponse { + driver_name: "test".into(), + resource_admission_policy: + openshell_core::resource_admission::DriverAdmissionConfig::default() + .acknowledgement(), + proxy_filesystem_baseline: Some(baseline.clone()), + extension: Some(extension_metadata( + ExtensionFamily::Compute, + "openshell/test", + "test", + [COMPUTE_PROXY_FILESYSTEM_BASELINE.to_string()], + )), + ..Default::default() + }, + ), + ); + let compute = crate::compute::ComputeRuntime::from_driver( + "test".into(), + driver, + 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 sandbox = test_sandbox( + "baseline", + "baseline", + mcp_policy_with_versions(&[]), + Vec::new(), + ); + state.store.put_message(&sandbox).await.unwrap(); + let policy = sandbox.spec.as_ref().unwrap().policy.as_ref().unwrap(); + state + .store + .put_policy_revision( + "baseline-v1", + "baseline", + "default", + 1, + &policy.encode_to_vec(), + &deterministic_policy_hash(policy), + ) + .await + .unwrap(); + + for global in [false, true] { + state.store.put_message(&sandbox).await.unwrap(); + if global { + install_test_global_policy(&state).await; + } + let snapshot = build_sandbox_config_snapshot(&state, &sandbox) + .await + .unwrap(); + let provider = build_provider_environment_snapshot(&state, &sandbox, true, true) + .await + .unwrap(); + assert_eq!( + snapshot.provider_env_revision, + provider.provider_env_revision + ); + let create = resolve_sandbox_create_runtime_inputs(&state, &sandbox) + .await + .unwrap(); + let admitted = create.effective_policy.unwrap(); + assert_eq!(snapshot.policy.as_ref(), Some(&admitted)); + assert_eq!(snapshot.policy_hash, deterministic_policy_hash(&admitted)); + let session_id = if global { + "baseline-global" + } else { + "baseline-local" + }; + let (session_tx, _session_rx) = tokio::sync::mpsc::channel(1); + let (shutdown_tx, _shutdown_rx) = tokio::sync::oneshot::channel(); + state.supervisor_sessions.register( + "baseline".into(), + session_id.into(), + session_tx, + shutdown_tx, + ); + reset_endpoint_status_for_supervisor_session(&state, "baseline", session_id) + .await + .unwrap(); + assert!( + state + .supervisor_sessions + .initialize_endpoint_status_authority("baseline", session_id) + ); + for loaded in [false, true] { + if loaded { + handle_report_policy_status( + &state, + with_sandbox( + Request::new(ReportPolicyStatusRequest { + sandbox_id: "baseline".into(), + version: 1, + status: PolicyStatus::Loaded as i32, + ..Default::default() + }), + "baseline", + ), + ) + .await + .unwrap(); + } + let observations = admitted + .network_policies + .values() + .flat_map(|rule| &rule.endpoints) + .map(|endpoint| openshell_core::proto::EndpointObservation { + endpoint_id: openshell_core::endpoint_status::endpoint_id(endpoint), + result: openshell_core::proto::EndpointResult::NoObservedExchange as i32, + failure: None, + }) + .collect(); + handle_report_endpoint_status( + &state, + with_sandbox( + Request::new(openshell_core::proto::ReportEndpointStatusRequest { + sandbox_id: "baseline".into(), + policy_hash: snapshot.policy_hash.clone(), + provider_env_revision: snapshot.provider_env_revision, + observations, + supervisor_session_id: session_id.into(), + report_sequence: if loaded { 2 } else { 1 }, + ..Default::default() + }), + "baseline", + ), + ) + .await + .expect( + "runtime endpoint report matches the admitted config before and after load", + ); + } + let filesystem = admitted.filesystem.unwrap(); + assert!(filesystem.read_only.contains(&baseline.read_only[0])); + assert!(filesystem.read_write.contains(&baseline.read_write[0])); + assert!( + !sandbox + .spec + .as_ref() + .unwrap() + .policy + .as_ref() + .unwrap() + .filesystem + .as_ref() + .unwrap() + .read_only + .contains(&baseline.read_only[0]) + ); + } + let mut oversized = mcp_policy_with_versions(&["2025-11-25"]); + oversized.filesystem.as_mut().unwrap().read_only = (0..128) + .map(|index| format!("/{index:04}{}", "x".repeat(2035))) + .collect(); + let padding = super::super::MAX_POLICY_SIZE - 16 - oversized.encoded_len(); + oversized + .filesystem + .as_mut() + .unwrap() + .read_only + .last_mut() + .unwrap() + .push_str(&"x".repeat(padding)); + assert!(oversized.encoded_len() <= super::super::MAX_POLICY_SIZE); + assert!(validate_and_canonicalize_policy(oversized.clone()).is_ok()); + assert_eq!( + apply_backend_proxy_baseline(&state, &mut oversized, PolicySource::Sandbox) + .unwrap_err() + .code(), + Code::FailedPrecondition + ); + } + #[tokio::test] async fn list_sandbox_policies_traverses_multiple_pages_exactly_once() { let state = test_server_state().await; diff --git a/crates/openshell-server/src/grpc/policy/endpoint_status.rs b/crates/openshell-server/src/grpc/policy/endpoint_status.rs index 6d35131cc5..734910aaa4 100644 --- a/crates/openshell-server/src/grpc/policy/endpoint_status.rs +++ b/crates/openshell-server/src/grpc/policy/endpoint_status.rs @@ -4,13 +4,15 @@ //! Passive network results for configured tool server endpoints. use super::{ - apply_effective_policy_context, canonical_policy_record_identity, - compute_provider_env_revision_with_catalog_and_policy_bindings, - current_effective_policy_for_sandbox, decode_policy_from_global_settings, + apply_backend_proxy_baseline, apply_captured_policy_context, canonical_policy_record_identity, + compute_provider_env_revision_from_records_and_policy_bindings, + current_effective_policy_from_records, decode_policy_from_global_settings, deterministic_policy_hash, load_global_settings, policy_static_credential_endpoint_bindings, + provider_policy_context_from_records, }; use crate::ServerState; use crate::compute::MutationScope; +use crate::grpc::provider::{ProviderEnvironmentRecord, load_provider_environment_records}; use crate::persistence::{ ObjectCursor, ObjectId, ObjectListQuery, ObjectWorkspace, PersistenceError, }; @@ -263,15 +265,28 @@ async fn current_endpoint_context( .provider_profile_sources .snapshot_catalog(state.store.as_ref(), workspace) .await?; - let policy = current_effective_policy_for_sandbox( + let provider_names = sandbox + .spec + .as_ref() + .map(|spec| spec.providers.clone()) + .unwrap_or_default(); + let records = load_provider_environment_records( + state.store.as_ref(), + &state.refresh, + workspace, + &provider_names, + ) + .await?; + let (mut policy, source) = current_effective_policy_from_records( state, &catalog, - workspace, sandbox, sandbox.object_id(), + &records, ) .await?; - derive_endpoint_context(state, sandbox, &catalog, policy).await + apply_backend_proxy_baseline(state, &mut policy, source)?; + derive_endpoint_context(&catalog, &records, policy, source) } async fn active_endpoint_context( @@ -328,16 +343,15 @@ async fn endpoint_context_for_loaded_policy_with_inputs( .as_ref() .map(|spec| spec.providers.clone()) .unwrap_or_default(); - let policy = if let Some(global_policy) = global_policy { - apply_effective_policy_context( - state, - catalog, - workspace, - &provider_names, - global_policy.clone(), - PolicySource::Global, - ) - .await? + let records = load_provider_environment_records( + state.store.as_ref(), + &state.refresh, + workspace, + &provider_names, + ) + .await?; + let (policy, source) = if let Some(global_policy) = global_policy { + (global_policy.clone(), PolicySource::Global) } else { let record = state .store @@ -345,43 +359,33 @@ async fn endpoint_context_for_loaded_policy_with_inputs( .await .map_err(|error| Status::internal(format!("fetch policy revision failed: {error}")))? .ok_or_else(|| Status::not_found("policy revision not found"))?; - let policy = canonical_policy_record_identity(&record)?.0; - apply_effective_policy_context( - state, - catalog, - workspace, - &provider_names, - policy, + ( + canonical_policy_record_identity(&record)?.0, PolicySource::Sandbox, ) - .await? }; - derive_endpoint_context(state, sandbox, catalog, policy).await + let mut policy = apply_captured_policy_context( + provider_policy_context_from_records(catalog, &records), + policy, + source, + )?; + apply_backend_proxy_baseline(state, &mut policy, source)?; + derive_endpoint_context(catalog, &records, policy, source) } -async fn derive_endpoint_context( - state: &ServerState, - sandbox: &Sandbox, +fn derive_endpoint_context( catalog: &EffectiveProviderProfileCatalog, + records: &[ProviderEnvironmentRecord], policy: ProtoSandboxPolicy, + source: PolicySource, ) -> Result { - let workspace = sandbox.object_workspace(); - let provider_names = sandbox - .spec - .as_ref() - .map(|spec| spec.providers.clone()) - .unwrap_or_default(); let bindings = policy_static_credential_endpoint_bindings(Some(&policy))?; - let provider_env_revision = compute_provider_env_revision_with_catalog_and_policy_bindings( - state.store.as_ref(), - &state.refresh, + let provider_env_revision = compute_provider_env_revision_from_records_and_policy_bindings( catalog, - workspace, - &provider_names, + records, &bindings, - ) - .await?; - + matches!(source, PolicySource::Global).then_some(&policy), + )?; Ok(EndpointContext { policy_hash: deterministic_policy_hash(&policy), provider_env_revision, diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 35e3122bd4..c726662a08 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -4367,6 +4367,7 @@ mod tests { default_image: "test/image:latest".to_string(), rootfs_tar_staging_dir: directory.path().to_string_lossy().into_owned(), rootfs_tar_max_bytes: 1024, + proxy_filesystem_baseline: None, resource_admission_policy: admission.acknowledgement(), extension: Some(metadata), ..Default::default() diff --git a/crates/openshell-server/src/test_support.rs b/crates/openshell-server/src/test_support.rs index b1d5736a93..be0baee071 100644 --- a/crates/openshell-server/src/test_support.rs +++ b/crates/openshell-server/src/test_support.rs @@ -182,6 +182,7 @@ impl FakeComputeDriver { resource_capabilities: None, rootfs_tar_staging_dir: String::new(), rootfs_tar_max_bytes: 0, + proxy_filesystem_baseline: None, extension: Some(openshell_core::extension_protocol::extension_metadata( openshell_core::extension_protocol::ExtensionFamily::Compute, "openshell/fake-compute-driver", diff --git a/crates/openshell-supervisor/src/backend_setup.rs b/crates/openshell-supervisor/src/backend_setup.rs index eacec32962..db85da1dc9 100644 --- a/crates/openshell-supervisor/src/backend_setup.rs +++ b/crates/openshell-supervisor/src/backend_setup.rs @@ -11,6 +11,7 @@ use std::sync::{Arc, Mutex}; use miette::Result; use openshell_core::SandboxSessionId; +use openshell_core::filesystem_baseline::FilesystemBaseline; use openshell_core::jwt::{SessionBearerTokenSlot, SupervisorAuthBundle}; use openshell_core::provider_credentials::ProviderCredentialState; use openshell_isolation_interface::AgentSpec; @@ -59,6 +60,13 @@ pub trait BackendSetup: Sync { /// Name chosen by trusted composition and checked against admission before decoding. fn backend_name(&self) -> &str; + /// Trusted proxy prerequisites. `None` retains legacy Linux image defaults; + /// an explicit empty baseline suppresses them. Match the compute driver's + /// advertised baseline so policy admission precedes immutable signing. + fn proxy_filesystem_baseline(&self) -> Option { + None + } + /// Decode native launch data without starting workload code. The returned /// identity must describe the same resource retained by the prepared backend. fn decode( @@ -93,6 +101,7 @@ pub struct SelectedBackend { descriptor: BackendDescriptor, identity: LaunchIdentity, prepared: Box, + proxy_filesystem_baseline: Option, } impl SelectedBackend { @@ -139,10 +148,17 @@ impl SelectedBackend { "supervisor authentication bundle does not match runtime generation" )); } + let proxy_filesystem_baseline = setup.proxy_filesystem_baseline(); + if let Some(baseline) = &proxy_filesystem_baseline { + baseline + .validate() + .map_err(|error| miette::miette!(error))?; + } Ok(Self { descriptor, identity, prepared, + proxy_filesystem_baseline, }) } @@ -151,6 +167,11 @@ impl SelectedBackend { &self.descriptor.backend_name } + /// Baseline captured only after admitted name and launch identity checks. + pub fn proxy_filesystem_baseline(&self) -> Option<&FilesystemBaseline> { + self.proxy_filesystem_baseline.as_ref() + } + /// Carry the selected backend's VM identity constraint into policy handling. pub fn vm_policy_identity(&self) -> Option { self.identity.vm_policy_identity diff --git a/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs b/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs index a28e0cf4bf..08f2083b8b 100644 --- a/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs +++ b/crates/openshell-supervisor/src/legacy_startup_regression_tests.rs @@ -133,7 +133,7 @@ where } "/openshell.v1.OpenShell/GetSandboxConfig" => { let mut policy = openshell_policy::restrictive_default_policy(); - enrich_proto_baseline_paths(&mut policy); + enrich_proto_baseline_paths(&mut policy, None); legacy_unary::( request, GetSandboxConfigResponse { diff --git a/crates/openshell-supervisor/src/lib.rs b/crates/openshell-supervisor/src/lib.rs index 589c87fffe..cb62856efc 100644 --- a/crates/openshell-supervisor/src/lib.rs +++ b/crates/openshell-supervisor/src/lib.rs @@ -43,6 +43,7 @@ use std::time::Duration; use tracing::{Instrument as _, debug, info, warn}; use openshell_core::PolicyValidationFailureMode; +use openshell_core::filesystem_baseline::FilesystemBaseline; use openshell_supervisor_control::RegistrationTraceContext; use openshell_ocsf::{ @@ -1223,6 +1224,7 @@ pub async fn run_sandbox_instance( Some(image_discovery), &RemoteStartupGateway { gateway: gateway.clone(), + proxy_filesystem_baseline: selected_backend.proxy_filesystem_baseline().cloned(), }, ) .instrument(tracing::info_span!(parent: &startup, "supervisor.policy.load")) @@ -2175,80 +2177,32 @@ fn proxy_baseline_paths() -> (Vec, Vec) { ) } -fn enrich_proto_baseline_paths_with( - proto: &mut openshell_core::proto::SandboxPolicy, - ro: &[String], - rw: &[String], - path_exists: F, -) -> bool -where - F: Fn(&str) -> bool, -{ - if ro.is_empty() && rw.is_empty() { - return false; - } - - let fs = proto - .filesystem - .get_or_insert_with(|| openshell_core::proto::FilesystemPolicy { - include_workdir: true, - ..Default::default() - }); - - let mut modified = false; - for path in ro { - if !fs.read_only.iter().any(|p| p == path) && !fs.read_write.iter().any(|p| p == path) { - if !path_exists(path) { - debug!( - path, - "Baseline read-only path does not exist, skipping enrichment" - ); - continue; - } - fs.read_only.push(path.clone()); - modified = true; - } - } - for path in rw { - if fs.read_write.iter().any(|p| p == path) { - continue; - } - if !path_exists(path) { - debug!( - path, - "Baseline read-write path does not exist, skipping enrichment" - ); - continue; - } - if fs.read_only.iter().any(|p| p == path) { - continue; - } - fs.read_write.push(path.clone()); - modified = true; - } - - modified -} - /// Ensure a proto `SandboxPolicy` includes the baseline filesystem paths /// required by proxy-mode sandboxes. Paths are only added if /// missing; user-specified paths are never removed. /// /// Returns `true` if the policy was modified (caller may want to sync back). -fn enrich_proto_baseline_paths(proto: &mut openshell_core::proto::SandboxPolicy) -> bool { +fn enrich_proto_baseline_paths( + proto: &mut openshell_core::proto::SandboxPolicy, + baseline: Option<&FilesystemBaseline>, +) -> bool { if proto.network_policies.is_empty() { return false; } - let (ro, rw) = proxy_baseline_paths(); - - // Baseline paths are system-injected, not user-specified. Skip paths - // that do not exist in this container image to avoid noisy warnings from - // Landlock and, more critically, to prevent a single missing baseline - // path from abandoning the entire Landlock ruleset under best-effort - // mode (see issue #664). - let modified = enrich_proto_baseline_paths_with(proto, &ro, &rw, |path| { - std::path::Path::new(path).exists() - }); + let modified = if let Some(baseline) = baseline { + // Backend-advertised paths belong to the workload namespace, not this + // supervisor's host filesystem. They are unconditional prerequisites. + baseline.enrich(proto, |_| true) + } else { + // Legacy Linux image defaults retain existence filtering: a missing + // path must not cause Landlock to abandon the ruleset (issue #664). + let (read_only, read_write) = proxy_baseline_paths(); + FilesystemBaseline { + read_only, + read_write, + } + .enrich(proto, |path| std::path::Path::new(path).exists()) + }; if modified { ocsf_emit!( @@ -2284,17 +2238,22 @@ fn proto_sync_payload_for_enriched_policy( /// Ensure a `SandboxPolicy` (Rust type) includes the baseline filesystem /// paths required by proxy-mode sandboxes. Used for the /// local-file code path where no proto is available. -fn enrich_sandbox_baseline_paths(policy: &mut SandboxPolicy) { +fn enrich_sandbox_baseline_paths( + policy: &mut SandboxPolicy, + baseline: Option<&FilesystemBaseline>, +) { if !matches!(policy.network.mode, NetworkMode::Proxy) { return; } - let (ro, rw) = proxy_baseline_paths(); + let (ro, rw) = baseline.map_or_else(proxy_baseline_paths, |baseline| { + (baseline.read_only.clone(), baseline.read_write.clone()) + }); let mut modified = false; for path in &ro { let p = std::path::PathBuf::from(path); if !policy.filesystem.read_only.contains(&p) && !policy.filesystem.read_write.contains(&p) { - if !p.exists() { + if baseline.is_none() && !p.exists() { debug!( path, "Baseline read-only path does not exist, skipping enrichment" @@ -2310,7 +2269,7 @@ fn enrich_sandbox_baseline_paths(policy: &mut SandboxPolicy) { if policy.filesystem.read_only.contains(&p) || policy.filesystem.read_write.contains(&p) { continue; } - if !p.exists() { + if baseline.is_none() && !p.exists() { debug!( path, "Baseline read-write path does not exist, skipping enrichment" @@ -2395,7 +2354,7 @@ mod baseline_tests { }, ); - enrich_proto_baseline_paths(&mut policy); + enrich_proto_baseline_paths(&mut policy, None); let filesystem = policy.filesystem.expect("filesystem policy"); assert!(filesystem.read_only.contains(&"/proc".to_string())); @@ -2509,7 +2468,7 @@ mod baseline_tests { ); let original = policy.clone(); - let enriched = enrich_proto_baseline_paths(&mut policy); + let enriched = enrich_proto_baseline_paths(&mut policy, None); assert!(!enriched); assert_eq!(policy, original); @@ -2533,7 +2492,7 @@ mod baseline_tests { process: ProcessPolicy::default(), }; - enrich_sandbox_baseline_paths(&mut policy); + enrich_sandbox_baseline_paths(&mut policy, None); assert!( policy.filesystem.read_only.contains(&PathBuf::from("/tmp")), @@ -2547,6 +2506,57 @@ mod baseline_tests { "baseline enrichment must not promote explicit read_only /tmp to read_write" ); } + + #[test] + fn selected_backend_baseline_replaces_linux_defaults_in_both_policy_formats() { + let baseline = FilesystemBaseline { + read_only: vec!["/platform-runtime-not-on-host".into()], + read_write: vec!["/explicit-read-only".into()], + }; + for baseline in [baseline, FilesystemBaseline::default()] { + baseline.validate().unwrap(); + let mut proto = openshell_policy::restrictive_default_policy(); + proto.filesystem = Some(openshell_core::proto::FilesystemPolicy { + read_only: vec!["/explicit-read-only".into()], + ..Default::default() + }); + proto.network_policies.insert( + "proxy".into(), + openshell_core::proto::NetworkPolicyRule::default(), + ); + enrich_proto_baseline_paths(&mut proto, Some(&baseline)); + let filesystem = proto.filesystem.unwrap(); + let mut expected = vec!["/explicit-read-only".to_string()]; + expected.extend(baseline.read_only.iter().cloned()); + assert_eq!(filesystem.read_only, expected); + assert!(filesystem.read_write.is_empty()); + + let mut local = SandboxPolicy { + version: 1, + filesystem: FilesystemPolicy { + read_only: vec!["/explicit-read-only".into()], + read_write: vec![], + include_workdir: false, + }, + network: NetworkPolicy { + mode: NetworkMode::Proxy, + proxy: None, + }, + landlock: LandlockPolicy::default(), + process: ProcessPolicy::default(), + }; + enrich_sandbox_baseline_paths(&mut local, Some(&baseline)); + assert_eq!( + local.filesystem.read_only, + expected.iter().map(PathBuf::from).collect::>() + ); + assert!(local.filesystem.read_write.is_empty()); + local.network.mode = NetworkMode::Block; + let original = local.filesystem.read_only.clone(); + enrich_sandbox_baseline_paths(&mut local, Some(&baseline)); + assert_eq!(local.filesystem.read_only, original); + } + } } /// Returns `true` if the error is transient and worth retrying. @@ -2624,6 +2634,9 @@ where } trait StartupGateway: Send + Sync { + fn proxy_filesystem_baseline(&self) -> Option { + None + } fn snapshot( &self, sandbox: &str, @@ -2650,6 +2663,7 @@ trait StartupGateway: Send + Sync { struct RemoteStartupGateway { gateway: Option, + proxy_filesystem_baseline: Option, } impl RemoteStartupGateway { @@ -2661,6 +2675,9 @@ impl RemoteStartupGateway { } impl StartupGateway for RemoteStartupGateway { + fn proxy_filesystem_baseline(&self) -> Option { + self.proxy_filesystem_baseline.clone() + } async fn snapshot( &self, sandbox: &str, @@ -2822,6 +2839,7 @@ async fn load_policy( None, &RemoteStartupGateway { gateway: openshell_gateway, + proxy_filesystem_baseline: None, }, ) .await @@ -2853,6 +2871,12 @@ async fn load_policy_with_gateway( Option, )> { use openshell_core::proto::ConfigurationAdmissionState; + let proxy_baseline = gateway.proxy_filesystem_baseline(); + if let Some(baseline) = &proxy_baseline { + baseline + .validate() + .map_err(|error| miette::miette!(error))?; + } // File mode: load OPA engine from rego rules + YAML data (dev override) if let (Some(policy_file), Some(data_file)) = (&policy_rules, &policy_data) { if sandbox_id.is_some() && openshell_gateway.is_some() { @@ -2906,7 +2930,7 @@ async fn load_policy_with_gateway( landlock: config.landlock, process: config.process, }; - enrich_sandbox_baseline_paths(&mut policy); + enrich_sandbox_baseline_paths(&mut policy, proxy_baseline.as_ref()); // File mode has no operator-registered middleware to connect. return Ok(( policy, @@ -2996,7 +3020,7 @@ async fn load_policy_with_gateway( }; // Enrich before syncing so the gateway baseline includes // baseline paths from the start. - enrich_proto_baseline_paths(&mut discovered); + enrich_proto_baseline_paths(&mut discovered, proxy_baseline.as_ref()); strip_proto_provider_policy_entries(&mut discovered); // Sync and re-fetch over a single connection to avoid extra // TLS handshakes. @@ -3057,7 +3081,7 @@ async fn load_policy_with_gateway( // a sandbox-sourced policy is written back. The gateway refuses // every sandbox policy write while a global policy is active, so a // global policy keeps the added paths in this process only. - let enriched = enrich_proto_baseline_paths(&mut proto_policy); + let enriched = enrich_proto_baseline_paths(&mut proto_policy, proxy_baseline.as_ref()); let sync_policy = proto_sync_payload_for_enriched_policy(&proto_policy, enriched) .filter(|_| snapshot.policy_source == openshell_core::proto::PolicySource::Sandbox); if let Some(sync_policy) = sync_policy { @@ -5290,7 +5314,7 @@ network_policies: async fn startup_rejects_vm_identity_until_matching_policy_is_available() { use openshell_core::proto::{ConfigurationAdmissionState, PolicySource}; let mut policy = proto_policy_fixture(); - enrich_proto_baseline_paths(&mut policy); + enrich_proto_baseline_paths(&mut policy, None); policy.process = Some(openshell_core::proto::ProcessPolicy { run_as_user: "10000".into(), run_as_group: "1001".into(), @@ -5433,7 +5457,7 @@ network_policies: async fn startup_pending_gateway_calls_exhaust_their_budgets() { for pending_snapshot in [true, false] { let mut policy = proto_policy_fixture(); - enrich_proto_baseline_paths(&mut policy); + enrich_proto_baseline_paths(&mut policy, None); let (reports, _reported) = tokio::sync::mpsc::unbounded_channel(); let gateway = TestStartupGateway { desired: Arc::new(std::sync::Mutex::new(settings_poll_result( @@ -5558,7 +5582,7 @@ network_policies: async fn startup_waits_for_repair_and_retries_stale_activation_before_returning() { use openshell_core::proto::{ConfigurationAdmissionState, PolicySource}; let mut policy = proto_policy_fixture(); - enrich_proto_baseline_paths(&mut policy); + enrich_proto_baseline_paths(&mut policy, None); let mut rejected = settings_poll_result(Some(policy), 1, PolicySource::Sandbox); rejected.configuration_admitted = false; let (reports, mut reported) = tokio::sync::mpsc::unbounded_channel(); @@ -6007,7 +6031,7 @@ network_policies: // paths, so startup writes an enriched copy back. The repaired policy // already carries those paths and needs no write-back. let mut repaired = proto_tcp_policy_fixture(); - assert!(enrich_proto_baseline_paths(&mut repaired)); + assert!(enrich_proto_baseline_paths(&mut repaired, None)); assert_refused_startup_write_waits_for_repair( settings_poll_result( Some(proto_tcp_policy_fixture()), @@ -6135,7 +6159,7 @@ network_policies: #[tokio::test(start_paused = true)] async fn startup_policy_write_refusal_baseline_repair_before_report_refetches() { let mut repaired = proto_tcp_policy_fixture(); - assert!(enrich_proto_baseline_paths(&mut repaired)); + assert!(enrich_proto_baseline_paths(&mut repaired, None)); Box::pin(assert_startup_repair_before_rejection_refetches( settings_poll_result( Some(proto_tcp_policy_fixture()), @@ -6281,7 +6305,7 @@ network_policies: async fn startup_policy_write_refusal_write_back_outlasts_the_attempt_limit() { // The same limit applies to a refused baseline-path write-back. let mut repaired = proto_tcp_policy_fixture(); - assert!(enrich_proto_baseline_paths(&mut repaired)); + assert!(enrich_proto_baseline_paths(&mut repaired, None)); assert_refused_startup_write_waits_for_repair( settings_poll_result( Some(proto_tcp_policy_fixture()), @@ -6378,7 +6402,7 @@ network_policies: global.global_policy_version = 1; let global_generation = report_generation(&global); let mut enriched = proto_tcp_policy_fixture(); - assert!(enrich_proto_baseline_paths(&mut enriched)); + assert!(enrich_proto_baseline_paths(&mut enriched, None)); let (calls, mut observed) = tokio::sync::mpsc::unbounded_channel(); let gateway = WriteRefusingStartupGateway { desired: Arc::new(std::sync::Mutex::new(global)), diff --git a/docs/extensibility/isolation-backends.mdx b/docs/extensibility/isolation-backends.mdx index 8dd5903176..ccdb2b8dc3 100644 --- a/docs/extensibility/isolation-backends.mdx +++ b/docs/extensibility/isolation-backends.mdx @@ -39,6 +39,14 @@ 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. +A compute driver can advertise `openshell.compute.proxy-filesystem-baseline` +with a `ProxyFilesystemBaseline` descriptor. The gateway composes these trusted +workload-namespace prerequisites into effective proxy-mode policy before compute +admission. The selected supervisor `BackendSetup` supplies the same baseline; +an explicit empty baseline suppresses legacy Linux defaults. The descriptor +and capability must agree, and advertised paths are never filtered against the +gateway host filesystem. Authored read-only grants remain read-only. + ## Who Owns What | Component | Owns | diff --git a/proto/compute_driver.proto b/proto/compute_driver.proto index 88d090fc18..98e2c1cdfe 100644 --- a/proto/compute_driver.proto +++ b/proto/compute_driver.proto @@ -157,6 +157,17 @@ message GetCapabilitiesResponse { // Gateways require an exact policy match before activating the driver. // Empty denotes a legacy driver and requires explicit admission opt-out. string resource_admission_policy = 13; + // Trusted, unconditional workload-namespace prerequisites for proxy mode. + // Requires openshell.compute.proxy-filesystem-baseline. Presence, including + // an empty message, replaces legacy Linux defaults. The gateway includes + // these paths in effective policy before compute admission and signing. + ProxyFilesystemBaseline proxy_filesystem_baseline = 14; +} + +message ProxyFilesystemBaseline { + // POSIX absolute workload paths, never paths checked on the gateway host. + repeated string read_only = 1; + repeated string read_write = 2; } message AuthenticateSandboxRequest { From 489cce69dd799a9b423157d51af978a3b30b8c40 Mon Sep 17 00:00:00 2001 From: Shiju Date: Wed, 4 Nov 2026 12:41:34 +0530 Subject: [PATCH 2/4] docs: trim filesystem backend implementation details Signed-off-by: Shiju --- docs/extensibility/isolation-backends.mdx | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/docs/extensibility/isolation-backends.mdx b/docs/extensibility/isolation-backends.mdx index ccdb2b8dc3..b7fd920c14 100644 --- a/docs/extensibility/isolation-backends.mdx +++ b/docs/extensibility/isolation-backends.mdx @@ -39,13 +39,12 @@ 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. -A compute driver can advertise `openshell.compute.proxy-filesystem-baseline` -with a `ProxyFilesystemBaseline` descriptor. The gateway composes these trusted -workload-namespace prerequisites into effective proxy-mode policy before compute -admission. The selected supervisor `BackendSetup` supplies the same baseline; -an explicit empty baseline suppresses legacy Linux defaults. The descriptor -and capability must agree, and advertised paths are never filtered against the -gateway host filesystem. Authored read-only grants remain read-only. +Drivers advertising `openshell.compute.proxy-filesystem-baseline` must provide +a `ProxyFilesystemBaseline` descriptor and the same baseline in `BackendSetup`. +Use paths in the workload's namespace; they need not exist on the gateway host. +An explicitly empty baseline disables legacy Linux defaults. Required paths +contribute to effective proxy-mode policy without widening authored read-only +grants. ## Who Owns What From a7cc8a031a21cf0cf6b01227b30414b15ed144ee Mon Sep 17 00:00:00 2001 From: Shiju Date: Tue, 3 Nov 2026 19:10:48 +0530 Subject: [PATCH 3/4] fix(compute): coordinate restart claims and reconciliation Signed-off-by: Shiju --- crates/openshell-server/src/compute/mod.rs | 568 ++++++++++++++++----- 1 file changed, 439 insertions(+), 129 deletions(-) diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 4ba1311d54..2433303499 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -2435,7 +2435,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 +2500,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 +2580,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 +2661,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 +6702,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 +6733,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 +6768,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 +6804,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 +6817,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 @@ -13467,22 +13536,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 +13809,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 +13864,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 +13901,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 +14270,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 +14311,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 +14396,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 +16609,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] From 27ef898266e3d9698c9463b9392a51cb0f2bc97f Mon Sep 17 00:00:00 2001 From: Shiju Date: Wed, 11 Nov 2026 13:14:59 +0530 Subject: [PATCH 4/4] fix(server): import object type for log stream checks Signed-off-by: Shiju --- crates/openshell-server/src/grpc/policy.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/openshell-server/src/grpc/policy.rs b/crates/openshell-server/src/grpc/policy.rs index 07fb3e0e5d..86b8e31a77 100644 --- a/crates/openshell-server/src/grpc/policy.rs +++ b/crates/openshell-server/src/grpc/policy.rs @@ -34,10 +34,9 @@ use crate::auth::workspace_authz::{ }; use crate::compute::MutationScope; use crate::pagination::Pagination; -#[cfg(test)] -use crate::persistence::ObjectType; use crate::persistence::{ - DraftChunkRecord, ObjectId, ObjectListQuery, ObjectName, ObjectWorkspace, PolicyRecord, Store, + DraftChunkRecord, ObjectId, ObjectListQuery, ObjectName, ObjectType, ObjectWorkspace, + PolicyRecord, Store, }; use crate::policy_store::{AtomicPolicyRevisionWrite, PolicyStoreExt}; use crate::provider_profile_sources::EffectiveProviderProfileCatalog;