diff --git a/architecture/sandbox.md b/architecture/sandbox.md index 647dd8019a..6274f4b2bf 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -121,6 +121,13 @@ partially active or last-known-good static set. Invalid metadata preserves the supplied dynamic snapshot, while a fetch failure preserves the currently active dynamic snapshot. +In the Kubernetes sidecar topology, the provider environment revision remains +an opaque content fingerprint and has no numeric ordering semantics. The +network supervisor assigns a separate, connection-local monotonic generation +to each distinct environment it publishes. The process supervisor applies only +newer generations, which accepts descending fingerprint values while rejecting +duplicate or delayed sidecar messages. + Gateway-managed refresh credentials use an opaque workload handle derived from the sandbox, provider identity, credential key, refresh authorization epoch, and canonical endpoint boundary. The handle remains stable while the gateway diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs index 8328097171..d96f141cc8 100644 --- a/crates/openshell-sandbox/src/lib.rs +++ b/crates/openshell-sandbox/src/lib.rs @@ -363,6 +363,9 @@ pub async fn run_sandbox( .as_ref() .map(|connection| connection.writer.clone()); let process_exit_ack = Arc::new(tokio::sync::Mutex::new(None)); + let initial_provider_env_generation = sidecar_bootstrap + .as_ref() + .map_or(0, |bootstrap| bootstrap.provider_env_generation); let mut process_control_closed = None; if let Some(connection) = process_control_connection { process_control_closed = Some(connection.closed); @@ -371,6 +374,7 @@ pub async fn run_sandbox( provider_credentials.clone(), agent_proposals.clone(), Arc::clone(&process_exit_ack), + initial_provider_env_generation, ); } @@ -569,6 +573,7 @@ pub async fn run_sandbox( sidecar_control::BootstrapData { policy_proto: proto.clone(), provider_env_revision: provider_credentials.snapshot().revision, + provider_env_generation: 0, provider_child_env: provider_env.clone(), agent_proposals_enabled: agent_proposals.enabled(), proxy_ca_cert_path: ca_paths.as_ref().map(|paths| paths.0.clone()), @@ -1072,25 +1077,29 @@ fn spawn_sidecar_control_update_watcher( provider_credentials: ProviderCredentialState, agent_proposals: AgentProposals, exit_ack: MainProcessExitAckWaiter, + mut provider_env_generation: u64, ) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { while let Some(update) = updates.recv().await { match update { sidecar_control::ControlUpdate::ProviderEnv { revision, + generation, provider_child_env, } => { - if revision <= provider_credentials.snapshot().revision { + if generation <= provider_env_generation { continue; } let env_count = provider_credentials .install_child_env_snapshot(revision, provider_child_env); + provider_env_generation = generation; ocsf_emit!( ConfigStateChangeBuilder::new(ocsf_ctx()) .severity(SeverityId::Informational) .status(StatusId::Success) .state(StateId::Enabled, "loaded") .unmapped("provider_env_revision", serde_json::json!(revision)) + .unmapped("provider_env_generation", serde_json::json!(generation)) .message(format!( "Sidecar provider environment refreshed [revision:{revision} env_count:{env_count}]" )) @@ -4350,21 +4359,24 @@ mod tests { } #[tokio::test] - async fn sidecar_control_provider_env_update_installs_newer_revision() { + async fn sidecar_control_provider_env_update_orders_by_generation() { let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); let provider_credentials = ProviderCredentialState::from_child_env_snapshot( - 1, + u64::MAX, std::collections::HashMap::from([("TOKEN".to_string(), "old".to_string())]), ); + let agent_proposals = AgentProposals::new(true); let handle = spawn_sidecar_control_update_watcher( rx, provider_credentials.clone(), - AgentProposals::default(), + agent_proposals.clone(), Arc::new(tokio::sync::Mutex::new(None)), + 10, ); tx.send(sidecar_control::ControlUpdate::ProviderEnv { - revision: 2, + revision: 1, + generation: 11, provider_child_env: std::collections::HashMap::from([( "TOKEN".to_string(), "new".to_string(), @@ -4374,7 +4386,7 @@ mod tests { timeout(Duration::from_secs(1), async { loop { - if provider_credentials.snapshot().revision == 2 { + if provider_credentials.snapshot().revision == 1 { break; } tokio::time::sleep(Duration::from_millis(10)).await; @@ -4383,21 +4395,33 @@ mod tests { .await .unwrap(); let snapshot = provider_credentials.snapshot(); - assert_eq!(snapshot.revision, 2); + assert_eq!(snapshot.revision, 1); assert_eq!( snapshot.child_env.get("TOKEN").map(String::as_str), Some("new") ); tx.send(sidecar_control::ControlUpdate::ProviderEnv { - revision: 1, + revision: 2, + generation: 11, provider_child_env: std::collections::HashMap::from([( "TOKEN".to_string(), - "stale".to_string(), + "duplicate-generation".to_string(), )]), }) .unwrap(); - tokio::time::sleep(Duration::from_millis(20)).await; + tx.send(sidecar_control::ControlUpdate::AgentProposals { + enabled: false, + config_revision: 1, + }) + .unwrap(); + timeout(Duration::from_secs(1), async { + while agent_proposals.enabled() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); assert_eq!( provider_credentials .snapshot() @@ -4406,6 +4430,54 @@ mod tests { .map(String::as_str), Some("new") ); + + tx.send(sidecar_control::ControlUpdate::ProviderEnv { + revision: 2, + generation: 12, + provider_child_env: std::collections::HashMap::from([( + "TOKEN".to_string(), + "newest".to_string(), + )]), + }) + .unwrap(); + timeout(Duration::from_secs(1), async { + loop { + if provider_credentials.snapshot().revision == 2 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + + tx.send(sidecar_control::ControlUpdate::ProviderEnv { + revision: u64::MAX, + generation: 11, + provider_child_env: std::collections::HashMap::from([( + "TOKEN".to_string(), + "stale".to_string(), + )]), + }) + .unwrap(); + tx.send(sidecar_control::ControlUpdate::AgentProposals { + enabled: true, + config_revision: 2, + }) + .unwrap(); + timeout(Duration::from_secs(1), async { + while !agent_proposals.enabled() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + let snapshot = provider_credentials.snapshot(); + assert_eq!(snapshot.revision, 2); + assert_eq!( + snapshot.child_env.get("TOKEN").map(String::as_str), + Some("newest") + ); handle.abort(); } @@ -4420,6 +4492,7 @@ mod tests { provider_credentials, agent_proposals.clone(), Arc::new(tokio::sync::Mutex::new(None)), + 0, ); tx.send(sidecar_control::ControlUpdate::AgentProposals { diff --git a/crates/openshell-sandbox/src/sidecar_control.rs b/crates/openshell-sandbox/src/sidecar_control.rs index 2bd28d5791..658a20132c 100644 --- a/crates/openshell-sandbox/src/sidecar_control.rs +++ b/crates/openshell-sandbox/src/sidecar_control.rs @@ -24,6 +24,7 @@ use tracing::{debug, info, warn}; pub struct BootstrapData { pub policy_proto: openshell_core::proto::SandboxPolicy, pub provider_env_revision: u64, + pub provider_env_generation: u64, pub provider_child_env: HashMap, pub agent_proposals_enabled: bool, pub proxy_ca_cert_path: Option, @@ -49,6 +50,7 @@ pub struct ExpectedPeer { pub enum ControlUpdate { ProviderEnv { revision: u64, + generation: u64, provider_child_env: HashMap, }, Policy { @@ -73,17 +75,22 @@ pub struct Publisher { impl Publisher { pub fn publish_provider_env(&self, revision: u64, provider_child_env: HashMap) { - { - let mut state = self.state.write().expect("sidecar control state poisoned"); - if revision <= state.provider_env_revision { - return; - } - state.provider_env_revision = revision; - state.provider_child_env.clone_from(&provider_child_env); + let mut state = self.state.write().expect("sidecar control state poisoned"); + if revision == state.provider_env_revision { + return; } - + state.provider_env_revision = revision; + state.provider_env_generation = state + .provider_env_generation + .checked_add(1) + .expect("sidecar provider environment generation overflow"); + state.provider_child_env.clone_from(&provider_child_env); + + // Keep generation assignment, bootstrap state, and publication under + // one lock so cloned publishers cannot emit generations out of order. let _ = self.updates.send(WireServerMessage::ProviderEnvUpdated { revision, + generation: state.provider_env_generation, provider_child_env, }); } @@ -177,6 +184,7 @@ enum WireServerMessage { BootstrapResponse { policy_proto: Vec, provider_env_revision: u64, + provider_env_generation: u64, provider_child_env: HashMap, agent_proposals_enabled: bool, proxy_ca_cert_path: Option, @@ -184,6 +192,7 @@ enum WireServerMessage { }, ProviderEnvUpdated { revision: u64, + generation: u64, provider_child_env: HashMap, }, PolicyUpdated { @@ -206,6 +215,7 @@ impl BootstrapData { WireServerMessage::BootstrapResponse { policy_proto: self.policy_proto.encode_to_vec(), provider_env_revision: self.provider_env_revision, + provider_env_generation: self.provider_env_generation, provider_child_env: self.provider_child_env.clone(), agent_proposals_enabled: self.agent_proposals_enabled, proxy_ca_cert_path: self @@ -227,6 +237,7 @@ impl TryFrom for BootstrapData { let WireServerMessage::BootstrapResponse { policy_proto, provider_env_revision, + provider_env_generation, provider_child_env, agent_proposals_enabled, proxy_ca_cert_path, @@ -245,6 +256,7 @@ impl TryFrom for BootstrapData { Ok(Self { policy_proto, provider_env_revision, + provider_env_generation, provider_child_env, agent_proposals_enabled, proxy_ca_cert_path: proxy_ca_cert_path.map(PathBuf::from), @@ -260,9 +272,11 @@ impl TryFrom for ControlUpdate { match message { WireServerMessage::ProviderEnvUpdated { revision, + generation, provider_child_env, } => Ok(Self::ProviderEnv { revision, + generation, provider_child_env, }), WireServerMessage::PolicyUpdated { @@ -675,6 +689,7 @@ mod tests { ..SandboxPolicy::default() }, provider_env_revision: 3, + provider_env_generation: 0, provider_child_env: env.clone(), agent_proposals_enabled: true, proxy_ca_cert_path: Some(PathBuf::from("/tmp/ca.pem")), @@ -688,6 +703,7 @@ mod tests { assert_eq!(received.policy_proto.version, 7); assert_eq!(received.provider_env_revision, 3); + assert_eq!(received.provider_env_generation, 0); assert_eq!(received.provider_child_env, env); assert!(received.agent_proposals_enabled); assert_eq!( @@ -700,6 +716,90 @@ mod tests { ); } + #[tokio::test] + async fn provider_env_updates_use_generation_not_fingerprint_order() { + let dir = tempfile::tempdir().unwrap(); + let socket = dir.path().join("control.sock"); + let server = spawn_server( + &socket, + BootstrapData { + policy_proto: SandboxPolicy::default(), + provider_env_revision: u64::MAX, + provider_env_generation: 7, + provider_child_env: HashMap::from([("TOKEN".to_string(), "first".to_string())]), + agent_proposals_enabled: false, + proxy_ca_cert_path: None, + proxy_ca_bundle_path: None, + }, + current_peer(), + ) + .unwrap(); + let publisher = server.publisher(); + let (_bootstrap, mut connection) = connect_process_client(&socket, Duration::from_secs(1)) + .await + .unwrap(); + + publisher.publish_provider_env( + 1, + HashMap::from([("TOKEN".to_string(), "second".to_string())]), + ); + + let update = tokio::time::timeout(Duration::from_secs(1), connection.updates.recv()) + .await + .unwrap() + .unwrap(); + match update { + ControlUpdate::ProviderEnv { + revision, + generation, + provider_child_env, + } => { + assert_eq!(revision, 1); + assert_eq!(generation, 8); + assert_eq!( + provider_child_env.get("TOKEN").map(String::as_str), + Some("second") + ); + } + other => panic!("unexpected sidecar update: {other:?}"), + } + + publisher.publish_provider_env( + 1, + HashMap::from([("TOKEN".to_string(), "duplicate".to_string())]), + ); + assert!( + tokio::time::timeout(Duration::from_millis(50), connection.updates.recv()) + .await + .is_err(), + "an identical fingerprint must remain a no-op" + ); + + publisher.publish_provider_env( + u64::MAX, + HashMap::from([("TOKEN".to_string(), "third".to_string())]), + ); + let update = tokio::time::timeout(Duration::from_secs(1), connection.updates.recv()) + .await + .unwrap() + .unwrap(); + match update { + ControlUpdate::ProviderEnv { + revision, + generation, + provider_child_env, + } => { + assert_eq!(revision, u64::MAX); + assert_eq!(generation, 9); + assert_eq!( + provider_child_env.get("TOKEN").map(String::as_str), + Some("third") + ); + } + other => panic!("unexpected sidecar update: {other:?}"), + } + } + #[tokio::test] async fn agent_proposals_update_is_delivered_to_process_client() { let dir = tempfile::tempdir().unwrap(); @@ -709,6 +809,7 @@ mod tests { BootstrapData { policy_proto: SandboxPolicy::default(), provider_env_revision: 0, + provider_env_generation: 0, provider_child_env: HashMap::new(), agent_proposals_enabled: false, proxy_ca_cert_path: None, @@ -749,6 +850,7 @@ mod tests { BootstrapData { policy_proto: SandboxPolicy::default(), provider_env_revision: 0, + provider_env_generation: 0, provider_child_env: HashMap::new(), agent_proposals_enabled: false, proxy_ca_cert_path: None, @@ -818,6 +920,7 @@ mod tests { BootstrapData { policy_proto: SandboxPolicy::default(), provider_env_revision: 0, + provider_env_generation: 0, provider_child_env: HashMap::new(), agent_proposals_enabled: false, proxy_ca_cert_path: None, @@ -852,6 +955,7 @@ mod tests { BootstrapData { policy_proto: SandboxPolicy::default(), provider_env_revision: 0, + provider_env_generation: 0, provider_child_env: HashMap::new(), agent_proposals_enabled: false, proxy_ca_cert_path: None, @@ -881,6 +985,7 @@ mod tests { BootstrapData { policy_proto: SandboxPolicy::default(), provider_env_revision: 0, + provider_env_generation: 0, provider_child_env: HashMap::new(), agent_proposals_enabled: false, proxy_ca_cert_path: None, @@ -911,6 +1016,7 @@ mod tests { BootstrapData { policy_proto: SandboxPolicy::default(), provider_env_revision: 0, + provider_env_generation: 0, provider_child_env: HashMap::new(), agent_proposals_enabled: false, proxy_ca_cert_path: None, diff --git a/e2e/rust/tests/credential_gating.rs b/e2e/rust/tests/credential_gating.rs index c65694d625..f4e1c5cee7 100644 --- a/e2e/rust/tests/credential_gating.rs +++ b/e2e/rust/tests/credential_gating.rs @@ -26,6 +26,14 @@ const TEST_HOST: &str = "host.openshell.internal"; const TOKEN_ENV: &str = "E2E_GATING_TOKEN"; const TEST_SECRET: &str = "e2e-gating-secret-value"; const PLACEHOLDER_PREFIX: &str = "openshell:resolve:env:"; +const KEY_PRESENCE_SCRIPT: &str = r#"if [ "${E2E_GATING_TOKEN+x}" != x ]; then + echo TOKEN_ABSENT +else + case "$E2E_GATING_TOKEN" in + openshell:resolve:env:*) echo TOKEN_PLACEHOLDER ;; + *) echo TOKEN_UNSAFE ;; + esac +fi"#; const WEBSOCKET_GUID: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; async fn run_cli(args: &[&str]) -> (bool, String) { @@ -709,6 +717,102 @@ async fn assert_gateway_admission( Ok(()) } +async fn wait_for_provider_key_presence( + sandbox: &SandboxGuard, + expected_present: bool, +) -> Result<(), String> { + let expected = if expected_present { + "TOKEN_PLACEHOLDER" + } else { + "TOKEN_ABSENT" + }; + let mut last_output = String::new(); + for _ in 0..60 { + last_output = sandbox.exec(&["sh", "-lc", KEY_PRESENCE_SCRIPT]).await?; + if last_output.contains(expected) { + return Ok(()); + } + tokio::time::sleep(Duration::from_millis(500)).await; + } + Err(format!( + "provider environment did not converge to {expected}; last key-presence output:\n{last_output}" + )) +} + +async fn assert_endpointless_provider_env_live_update(port: u16) -> Result<(), String> { + let unbound = write_policy( + port, + EndpointMode::RestBody { rewrite: false }, + CredentialSource::ProviderProfile, + )?; + let bound = write_policy( + port, + EndpointMode::RestBody { rewrite: false }, + CredentialSource::PolicyBinding, + )?; + let unbound_path = unbound + .path() + .to_str() + .ok_or_else(|| "unbound policy path is not UTF-8".to_string())?; + let bound_path = bound + .path() + .to_str() + .ok_or_else(|| "bound policy path is not UTF-8".to_string())?; + + let mut sandbox = SandboxGuard::create_keep_with_args( + &[ + "--policy", + unbound_path, + "--provider", + PROVIDER_NAME, + "--no-tty", + ], + &["sh", "-c", "echo PROVIDER_UPDATE_READY && sleep infinity"], + "PROVIDER_UPDATE_READY", + ) + .await?; + + let result = async { + wait_for_provider_key_presence(&sandbox, false).await?; + + let (success, output) = run_cli(&[ + "policy", + "set", + &sandbox.name, + "--policy", + bound_path, + "--wait", + "--timeout", + "120", + ]) + .await; + if !success { + return Err(format!("binding policy update failed:\n{output}")); + } + wait_for_provider_key_presence(&sandbox, true).await?; + + let (success, output) = run_cli(&[ + "policy", + "set", + &sandbox.name, + "--policy", + unbound_path, + "--wait", + "--timeout", + "120", + ]) + .await; + if !success { + return Err(format!("unbinding policy update failed:\n{output}")); + } + wait_for_provider_key_presence(&sandbox, false).await + } + .await; + + sandbox.cleanup().await; + result +} + async fn run_body_sandbox( port: u16, mode: EndpointMode, @@ -835,6 +939,7 @@ async fn credentialed_endpoint_gates_work_end_to_end() { assert_eq!(observations.len(), 3, "observations: {observations:?}"); assert!(!observations[2].saw_placeholder); assert!(!observations[2].saw_secret); + assert_endpointless_provider_env_live_update(server.port).await?; Ok::<(), String>(()) } .await;