diff --git a/crates/openshell-core/src/credential_rotation.rs b/crates/openshell-core/src/credential_rotation.rs new file mode 100644 index 0000000000..3286760af3 --- /dev/null +++ b/crates/openshell-core/src/credential_rotation.rs @@ -0,0 +1,353 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Close tunnels that carry a rotated static provider credential. +//! +//! A value-only rotation replaces a credential without changing policy, so it +//! must not advance the policy generation and close every tunnel in the +//! sandbox. Rotation must still be immediate: a connection that authenticated +//! with the old value, such as a WebSocket that carried it in its handshake, +//! must not keep working. Each credential-bearing relay therefore watches for +//! rotations whose static credential bindings match its destination, and +//! only those relays close. + +use crate::host_pattern::HostPattern; +use std::collections::VecDeque; +use std::sync::{Arc, Mutex, Weak}; +use tokio::sync::watch; + +/// Rotations kept for relays that have not yet observed them. A relay whose +/// captured epoch is older than the retained history closes conservatively. +const MAX_RETAINED_ROTATIONS: usize = 64; + +/// One endpoint a rotated credential is bound to. +#[derive(Debug, Clone)] +pub struct RotatedEndpoint { + host: HostPattern, + port: u16, +} + +impl RotatedEndpoint { + /// Compile a binding endpoint. Invalid patterns were already rejected when + /// the credential was installed, so they are reported rather than skipped. + pub fn new(host: &str, port: u32) -> Result { + let host = HostPattern::new(host)?; + let port = u16::try_from(port).map_err(|_| format!("invalid port {port}"))?; + Ok(Self { host, port }) + } + + pub(crate) fn from_compiled(host: HostPattern, port: u16) -> Self { + Self { host, port } + } + + fn matches(&self, endpoint: &TunnelEndpoint) -> bool { + self.port == endpoint.port && self.host.matches(&endpoint.host) + } +} + +#[derive(Debug)] +struct RotationRecord { + epoch: u64, + keys: Arc<[String]>, + endpoints: Vec, +} + +#[derive(Debug)] +struct TunnelEndpoint { + host: String, + port: u16, +} + +#[derive(Debug, Default)] +struct RotationsInner { + records: VecDeque, + tunnels: Vec>, +} + +/// Rotation history shared by the supervisor, which announces rotations, and +/// the network relays, which close on them. +#[derive(Debug)] +pub struct CredentialRotations { + inner: Mutex, + epoch_tx: watch::Sender, +} + +impl Default for CredentialRotations { + fn default() -> Self { + Self::new() + } +} + +/// What an announcement reached. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RotationAnnouncement { + pub epoch: u64, + /// Credential-bearing tunnels open to a rotated credential's endpoints. + /// Each of them closes. + pub tunnels: usize, + /// Rotated binding endpoints as `host:port`, for diagnosis. Never values, + /// placeholders or credential identities: a stable credential's identity + /// embeds its workload placeholder handle. + pub endpoints: Vec, +} + +/// Why a watched relay must close. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RotatedCredential { + /// Environment keys of the rotated credentials bound to this destination. + /// Empty when the relay fell behind the retained history; it closes anyway. + pub keys: Arc<[String]>, +} + +impl CredentialRotations { + #[must_use] + pub fn new() -> Self { + Self { + inner: Mutex::new(RotationsInner::default()), + epoch_tx: watch::Sender::new(0), + } + } + + /// Start watching rotations for one relay destination. + /// + /// Call this before the relay can send credential material upstream, so a + /// rotation that lands while the relay is starting is not missed. + #[must_use] + pub fn watch(self: &Arc, host: &str, port: u16) -> CredentialRotationWatch { + let endpoint = Arc::new(TunnelEndpoint { + host: host.to_ascii_lowercase(), + port, + }); + let mut inner = self.inner.lock().expect("credential rotations poisoned"); + inner.tunnels.retain(|tunnel| tunnel.strong_count() > 0); + inner.tunnels.push(Arc::downgrade(&endpoint)); + CredentialRotationWatch { + rotations: Arc::clone(self), + endpoint, + captured_epoch: *self.epoch_tx.borrow(), + } + } + + /// Record a rotation and wake the relays bound to its endpoints. + /// + /// Call after the new values are installed, so a relay that reconnects + /// after closing resolves the new value. + pub fn announce( + &self, + keys: Vec, + endpoints: Vec, + ) -> RotationAnnouncement { + let mut inner = self.inner.lock().expect("credential rotations poisoned"); + inner.tunnels.retain(|tunnel| tunnel.strong_count() > 0); + let tunnels = inner + .tunnels + .iter() + .filter_map(Weak::upgrade) + .filter(|tunnel| endpoints.iter().any(|endpoint| endpoint.matches(tunnel))) + .count(); + let mut described = endpoints + .iter() + .map(|endpoint| format!("{}:{}", endpoint.host.source(), endpoint.port)) + .collect::>(); + described.sort(); + described.dedup(); + let epoch = *self.epoch_tx.borrow() + 1; + inner.records.push_back(RotationRecord { + epoch, + keys: keys.into(), + endpoints, + }); + while inner.records.len() > MAX_RETAINED_ROTATIONS { + inner.records.pop_front(); + } + // Publish while holding the lock so a relay that observes the epoch + // always finds its record. + self.epoch_tx.send_replace(epoch); + RotationAnnouncement { + epoch, + tunnels, + endpoints: described, + } + } + + fn rotated_since(&self, epoch: u64, endpoint: &TunnelEndpoint) -> Option { + let inner = self.inner.lock().expect("credential rotations poisoned"); + if inner + .records + .front() + .is_some_and(|oldest| oldest.epoch > epoch + 1) + { + return Some(RotatedCredential { + keys: Arc::from(Vec::new()), + }); + } + let mut keys = Vec::new(); + for record in inner.records.iter().filter(|record| record.epoch > epoch) { + if record + .endpoints + .iter() + .any(|rotated| rotated.matches(endpoint)) + { + keys.extend(record.keys.iter().cloned()); + } + } + if keys.is_empty() { + return None; + } + keys.sort(); + keys.dedup(); + Some(RotatedCredential { keys: keys.into() }) + } +} + +/// One relay's subscription to rotations of credentials bound to its +/// destination. Dropping it stops the relay being counted as open. +#[derive(Debug)] +pub struct CredentialRotationWatch { + rotations: Arc, + endpoint: Arc, + captured_epoch: u64, +} + +impl CredentialRotationWatch { + /// Resolve when a credential bound to this destination rotates. + /// + /// Rotations of credentials bound elsewhere are ignored, so other + /// providers' tunnels stay open. + pub async fn rotated(&self) -> RotatedCredential { + let mut receiver = self.rotations.epoch_tx.subscribe(); + let mut seen = self.captured_epoch; + loop { + let current = *receiver.borrow_and_update(); + if current > seen { + if let Some(rotated) = self.rotations.rotated_since(seen, &self.endpoint) { + return rotated; + } + seen = current; + } + if receiver.changed().await.is_err() { + return std::future::pending().await; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn endpoint(host: &str, port: u32) -> RotatedEndpoint { + RotatedEndpoint::new(host, port).expect("valid endpoint") + } + + async fn rotated_within(watch: &CredentialRotationWatch) -> Option { + tokio::time::timeout(Duration::from_millis(50), watch.rotated()) + .await + .ok() + } + + #[tokio::test] + async fn rotation_wakes_only_relays_bound_to_the_rotated_endpoints() { + let rotations = Arc::new(CredentialRotations::new()); + let slack = rotations.watch("Slack.com", 443); + let linear = rotations.watch("api.linear.app", 443); + + let announced = rotations.announce( + vec!["SLACK_BOT_TOKEN".to_string()], + vec![endpoint("slack.com", 443)], + ); + + assert_eq!(announced.tunnels, 1); + assert_eq!( + rotated_within(&slack).await, + Some(RotatedCredential { + keys: Arc::from(vec!["SLACK_BOT_TOKEN".to_string()]), + }) + ); + assert_eq!(rotated_within(&linear).await, None); + } + + #[tokio::test] + async fn rotation_matches_wildcard_hosts_and_requires_the_port() { + let rotations = Arc::new(CredentialRotations::new()); + let wss = rotations.watch("wss-primary.slack.com", 443); + let other_port = rotations.watch("wss-primary.slack.com", 8443); + + rotations.announce( + vec!["SLACK_APP_TOKEN".to_string()], + vec![endpoint("*.slack.com", 443)], + ); + + assert!(rotated_within(&wss).await.is_some()); + assert_eq!(rotated_within(&other_port).await, None); + } + + #[tokio::test] + async fn a_relay_opened_after_a_rotation_ignores_it() { + let rotations = Arc::new(CredentialRotations::new()); + rotations.announce( + vec!["KEY".to_string()], + vec![endpoint("api.example.com", 443)], + ); + + let reconnected = rotations.watch("api.example.com", 443); + + assert_eq!(rotated_within(&reconnected).await, None); + } + + #[tokio::test] + async fn an_unrelated_rotation_does_not_hide_a_later_matching_one() { + let rotations = Arc::new(CredentialRotations::new()); + let watch = rotations.watch("api.example.com", 443); + rotations.announce( + vec!["OTHER".to_string()], + vec![endpoint("other.example.com", 443)], + ); + assert_eq!(rotated_within(&watch).await, None); + + rotations.announce( + vec!["KEY".to_string()], + vec![endpoint("api.example.com", 443)], + ); + + assert_eq!( + rotated_within(&watch).await.map(|rotated| rotated.keys), + Some(Arc::from(vec!["KEY".to_string()])) + ); + } + + #[tokio::test] + async fn a_relay_behind_the_retained_history_closes() { + let rotations = Arc::new(CredentialRotations::new()); + let watch = rotations.watch("api.example.com", 443); + for _ in 0..=MAX_RETAINED_ROTATIONS { + rotations.announce( + vec!["OTHER".to_string()], + vec![endpoint("other.example.com", 443)], + ); + } + + assert_eq!( + rotated_within(&watch) + .await + .map(|rotated| rotated.keys.len()), + Some(0) + ); + } + + #[test] + fn closed_relays_are_not_counted() { + let rotations = Arc::new(CredentialRotations::new()); + let open = rotations.watch("api.example.com", 443); + drop(rotations.watch("api.example.com", 443)); + + let announced = rotations.announce( + vec!["KEY".to_string()], + vec![endpoint("api.example.com", 443)], + ); + + assert_eq!(announced.tunnels, 1); + drop(open); + } +} diff --git a/crates/openshell-core/src/endpoint_path.rs b/crates/openshell-core/src/endpoint_path.rs index 5f9b37d75b..428d5f3384 100644 --- a/crates/openshell-core/src/endpoint_path.rs +++ b/crates/openshell-core/src/endpoint_path.rs @@ -22,6 +22,11 @@ enum EndpointPathPatternKind { } impl EndpointPathPattern { + /// The pattern this was compiled from. + pub(crate) fn source(&self) -> &str { + &self.source + } + #[must_use] pub fn new(pattern: &str) -> Self { let kind = if pattern.is_empty() || pattern == "**" || pattern == "/**" { diff --git a/crates/openshell-core/src/host_pattern.rs b/crates/openshell-core/src/host_pattern.rs index 5ded1b21c8..9dce529a28 100644 --- a/crates/openshell-core/src/host_pattern.rs +++ b/crates/openshell-core/src/host_pattern.rs @@ -127,6 +127,11 @@ impl HostPattern { Ok(Self { source, labels }) } + /// The lowercase pattern this was compiled from. + pub(crate) fn source(&self) -> &str { + &self.source + } + #[must_use] pub fn matches(&self, host: &str) -> bool { let host = host.to_ascii_lowercase(); diff --git a/crates/openshell-core/src/lib.rs b/crates/openshell-core/src/lib.rs index a8f2e30ac7..936d653bf1 100644 --- a/crates/openshell-core/src/lib.rs +++ b/crates/openshell-core/src/lib.rs @@ -14,6 +14,7 @@ pub mod agent_output; pub mod auth; pub mod config; pub mod container_paths; +pub mod credential_rotation; pub mod denial; pub mod driver_mounts; pub mod driver_utils; diff --git a/crates/openshell-core/src/provider_credentials.rs b/crates/openshell-core/src/provider_credentials.rs index 9c079c73f8..f5eab12ff5 100644 --- a/crates/openshell-core/src/provider_credentials.rs +++ b/crates/openshell-core/src/provider_credentials.rs @@ -3,6 +3,9 @@ //! Runtime provider credential snapshots. +use crate::credential_rotation::{ + CredentialRotationWatch, CredentialRotations, RotatedEndpoint, RotationAnnouncement, +}; use crate::secrets::SecretResolver; use crate::{ endpoint_path::EndpointPathPattern, @@ -82,6 +85,9 @@ struct CompiledStaticCredentialEndpointBinding { #[derive(Debug, Clone)] pub struct ProviderCredentialState { inner: Arc>, + // Shared by every clone, but never copied by `install_prepared`: relays + // watch the live state's history, not a candidate's. + rotations: Arc, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -136,6 +142,7 @@ impl ProviderCredentialState { static_credential_identity_epochs: HashMap::new(), requires_stable_placeholders: false, })), + rotations: Arc::new(CredentialRotations::new()), } } @@ -220,6 +227,7 @@ impl ProviderCredentialState { static_credential_identity_epochs, requires_stable_placeholders: !stable_placeholder_environment_keys.is_empty(), })), + rotations: Arc::new(CredentialRotations::new()), }) } @@ -255,6 +263,7 @@ impl ProviderCredentialState { static_credential_identity_epochs: HashMap::new(), requires_stable_placeholders: false, })), + rotations: Arc::new(CredentialRotations::new()), } } @@ -423,6 +432,116 @@ impl ProviderCredentialState { (resolver, classifier, revision) } + /// Watch for rotations of static credentials bound to a relay destination. + #[must_use] + pub fn rotation_watch(&self, host: &str, port: u16) -> CredentialRotationWatch { + self.rotations.watch(host, port) + } + + /// Announce that the installed values of `keys` replaced earlier values + /// with no other change, closing the relays bound to their endpoints. + /// + /// Call after installing the new values. Keys without an installed static + /// binding cannot be carried by a relay and are ignored. + pub fn announce_rotation(&self, keys: &[String]) -> RotationAnnouncement { + let endpoints = { + let inner = self + .inner + .read() + .expect("provider credential state poisoned"); + keys.iter() + .filter_map(|key| inner.static_credential_bindings.get(key)) + .flat_map(|binding| binding.endpoints.iter()) + .map(|endpoint| { + RotatedEndpoint::from_compiled(endpoint.host.clone(), endpoint.port) + }) + .collect() + }; + self.rotations.announce(keys.to_vec(), endpoints) + } + + /// The credentials a candidate rotates, when replacing the values of + /// stable-placeholder credentials is its only difference from the live + /// state. + /// + /// Equal stable handles imply the same provider, key, endpoints and + /// attachment epoch, and the workload keeps the same placeholder, so such + /// a rotation needs no policy reload: the relays bound to the rotated + /// endpoints close instead. Any other difference returns `None`, + /// including a new value for a credential without a stable placeholder, + /// whose workload placeholder is revision scoped. + #[must_use] + pub fn value_only_rotation(&self, candidate: &Self) -> Option> { + // Release the candidate lock before taking the live lock, including + // when a caller passes another handle to the same state. + let (env, dynamic_credentials, bindings, non_secret_keys, files, stable, resolver) = { + let candidate = candidate + .inner + .read() + .expect("provider credential state poisoned"); + ( + candidate.current.child_env.clone(), + candidate.current.dynamic_credentials.clone(), + candidate.static_credential_bindings.clone(), + candidate.non_secret_environment_keys.clone(), + candidate.files.clone(), + candidate.requires_stable_placeholders, + candidate.current_resolver.clone(), + ) + }; + let live = self + .inner + .read() + .expect("provider credential state poisoned"); + if !live.body_inventory_available + || live.requires_stable_placeholders != stable + || live.current.dynamic_credentials != dynamic_credentials + || live.non_secret_environment_keys != non_secret_keys + || live.files != files + || static_credential_identities(&live.static_credential_bindings) + != static_credential_identities(&bindings) + || static_credential_stable_handles(&live.static_credential_bindings) + != static_credential_stable_handles(&bindings) + || static_credential_endpoints(&live.static_credential_bindings) + != static_credential_endpoints(&bindings) + { + return None; + } + let env = env + .into_iter() + .filter(|(key, _)| !live.suppressed_keys.contains(key)) + .collect::>(); + if env.len() != live.current.child_env.len() { + return None; + } + let resolve = |resolver: Option<&Arc>, value: &str| { + resolver + .and_then(|resolver| resolver.resolve_placeholder(value)) + .map_or_else(|| value.to_string(), str::to_string) + }; + let mut rotated = Vec::new(); + for (key, placeholder) in &env { + let live_placeholder = live.current.child_env.get(key)?; + let unchanged = resolve(live.current_resolver.as_ref(), live_placeholder) + == resolve(resolver.as_ref(), placeholder); + if unchanged { + continue; + } + let stable_handle = bindings + .get(key) + .is_some_and(|binding| !binding.workload_credential_handle.is_empty()); + if !stable_handle || live_placeholder != placeholder { + return None; + } + rotated.push(key.clone()); + } + if rotated.is_empty() { + return None; + } + rotated.sort(); + Some(rotated) + } + #[must_use] pub fn revision(&self) -> u64 { self.inner @@ -1083,6 +1202,28 @@ fn static_credential_identities( .collect() } +fn static_credential_endpoints( + bindings: &HashMap, +) -> HashMap<&str, Vec<(&str, u16, &str)>> { + bindings + .iter() + .map(|(key, binding)| { + let endpoints = binding + .endpoints + .iter() + .map(|endpoint| { + ( + endpoint.host.source(), + endpoint.port, + endpoint.path.source(), + ) + }) + .collect(); + (key.as_str(), endpoints) + }) + .collect() +} + fn static_credential_stable_handles( bindings: &HashMap, ) -> HashMap { @@ -1170,6 +1311,243 @@ mod tests { use super::*; use crate::google_cloud; + /// Raw provider delivery for value-only rotation tests: two stable + /// credentials, one revision-scoped credential and one non-secret value. + #[derive(Clone)] + struct RotationInput { + env: HashMap, + expiries: HashMap, + dynamic: HashMap, + bindings: HashMap, + non_secret: Vec, + stable: Vec, + files: HashMap, + } + + fn rotation_binding( + host: &str, + port: u32, + path: &str, + handle: &str, + ) -> StaticCredentialBinding { + StaticCredentialBinding { + endpoints: vec![StaticCredentialEndpointBinding { + host: host.into(), + port, + path: path.into(), + }], + credential_identity: if handle.is_empty() { + format!("provider-{host}") + } else { + format!("refresh:{handle}") + }, + workload_credential_handle: handle.into(), + } + } + + fn rotation_input() -> RotationInput { + RotationInput { + env: HashMap::from([ + ("A_TOKEN".into(), "a-1".into()), + ("B_TOKEN".into(), "b-1".into()), + ("C_TOKEN".into(), "c-1".into()), + ("REGION".into(), "us-east-1".into()), + ]), + expiries: HashMap::new(), + dynamic: HashMap::new(), + bindings: HashMap::from([ + ( + "A_TOKEN".into(), + rotation_binding("a.example.com", 443, "/v1/**", &"a".repeat(64)), + ), + ( + "B_TOKEN".into(), + rotation_binding("b.example.com", 443, "/**", &"b".repeat(64)), + ), + ( + "C_TOKEN".into(), + rotation_binding("c.example.com", 443, "/**", ""), + ), + ]), + non_secret: vec!["REGION".into()], + stable: vec!["A_TOKEN".into(), "B_TOKEN".into()], + files: HashMap::new(), + } + } + + fn rotation_state(revision: u64, input: RotationInput) -> ProviderCredentialState { + let state = ProviderCredentialState::from_bound_environment_with_stable_placeholders( + revision, + input.env, + input.expiries, + input.dynamic, + input.bindings, + input.non_secret, + input.stable, + ) + .expect("valid rotation input"); + state.set_managed_files(input.files); + state + } + + fn rotation_between(change: impl FnOnce(&mut RotationInput)) -> Option> { + let live = rotation_state(1, rotation_input()); + let mut candidate = rotation_input(); + change(&mut candidate); + live.value_only_rotation(&rotation_state(2, candidate)) + } + + fn set(input: &mut RotationInput, key: &str, value: &str) { + input.env.insert(key.into(), value.into()); + } + + #[test] + fn value_only_rotation_names_the_rotated_stable_credentials() { + assert_eq!( + rotation_between(|input| set(input, "A_TOKEN", "a-2")), + Some(vec!["A_TOKEN".to_string()]) + ); + assert_eq!( + rotation_between(|input| { + set(input, "B_TOKEN", "b-2"); + set(input, "A_TOKEN", "a-2"); + }), + Some(vec!["A_TOKEN".to_string(), "B_TOKEN".to_string()]) + ); + } + + #[test] + fn value_only_rotation_refuses_every_other_difference() { + type Change = fn(&mut RotationInput); + let rotate_a: Change = |input| set(input, "A_TOKEN", "a-2"); + let cases: Vec<(&str, Change)> = vec![ + ("nothing changed", |_| {}), + ("only an expiry changed", |input| { + input.expiries.insert("A_TOKEN".into(), 4_102_444_800_000); + }), + ("endpoint host changed", |input| { + set(input, "A_TOKEN", "a-2"); + input.bindings.get_mut("A_TOKEN").unwrap().endpoints[0].host = + "evil.example.com".into(); + }), + ("endpoint port changed", |input| { + set(input, "A_TOKEN", "a-2"); + input.bindings.get_mut("A_TOKEN").unwrap().endpoints[0].port = 8443; + }), + ("endpoint path changed", |input| { + set(input, "A_TOKEN", "a-2"); + input.bindings.get_mut("A_TOKEN").unwrap().endpoints[0].path = "/**".into(); + }), + ("endpoint added", |input| { + set(input, "A_TOKEN", "a-2"); + input.bindings.get_mut("A_TOKEN").unwrap().endpoints.push( + StaticCredentialEndpointBinding { + host: "other.example.com".into(), + port: 443, + path: "/**".into(), + }, + ); + }), + ("revision-scoped identity changed", |input| { + input + .bindings + .get_mut("C_TOKEN") + .unwrap() + .credential_identity = "provider-other".into(); + }), + ("stable handle changed", |input| { + set(input, "A_TOKEN", "a-2"); + let binding = input.bindings.get_mut("A_TOKEN").unwrap(); + binding.workload_credential_handle = "c".repeat(64); + binding.credential_identity = format!("refresh:{}", "c".repeat(64)); + }), + ("credential added", |input| { + set(input, "A_TOKEN", "a-2"); + set(input, "D_TOKEN", "d-1"); + input.bindings.insert( + "D_TOKEN".into(), + rotation_binding("d.example.com", 443, "/**", &"d".repeat(64)), + ); + input.stable.push("D_TOKEN".into()); + }), + ("credential removed", |input| { + set(input, "A_TOKEN", "a-2"); + input.env.remove("B_TOKEN"); + input.bindings.remove("B_TOKEN"); + input.stable.retain(|key| key != "B_TOKEN"); + }), + ("dynamic credential changed", |input| { + set(input, "A_TOKEN", "a-2"); + input.dynamic.insert( + "grant".into(), + crate::proto::ProviderProfileCredential { + name: "grant".into(), + ..Default::default() + }, + ); + }), + ("non-secret value changed", |input| { + set(input, "A_TOKEN", "a-2"); + set(input, "REGION", "eu-west-1"); + }), + ("managed files changed", |input| { + set(input, "A_TOKEN", "a-2"); + input.files.insert("config.json".into(), "{}".into()); + }), + ("revision-scoped value changed", |input| { + set(input, "C_TOKEN", "c-2"); + }), + ("stable and revision-scoped values changed", |input| { + set(input, "A_TOKEN", "a-2"); + set(input, "C_TOKEN", "c-2"); + }), + ]; + // The base case is a real rotation, so each refusal is meaningful. + assert!(rotation_between(rotate_a).is_some()); + for (name, change) in cases { + assert_eq!(rotation_between(change), None, "{name}"); + } + } + + #[test] + fn value_only_rotation_refuses_a_revoked_live_state() { + let live = rotation_state(1, rotation_input()); + live.revoke_static_provider_environment(1); + let mut candidate = rotation_input(); + set(&mut candidate, "A_TOKEN", "a-2"); + assert_eq!( + live.value_only_rotation(&rotation_state(2, candidate)), + None + ); + } + + #[test] + fn value_only_rotation_ignores_suppressed_keys() { + let live = rotation_state(1, rotation_input()); + live.remove_env_key("B_TOKEN"); + let mut candidate = rotation_input(); + set(&mut candidate, "A_TOKEN", "a-2"); + set(&mut candidate, "B_TOKEN", "b-2"); + assert_eq!( + live.value_only_rotation(&rotation_state(2, candidate)), + Some(vec!["A_TOKEN".to_string()]) + ); + } + + #[test] + fn announced_rotation_describes_endpoints_without_placeholder_handles() { + let live = rotation_state(1, rotation_input()); + let mut candidate = rotation_input(); + set(&mut candidate, "A_TOKEN", "a-2"); + let candidate = rotation_state(2, candidate); + let keys = live.value_only_rotation(&candidate).unwrap(); + live.install_prepared(&candidate); + let announced = live.announce_rotation(&keys); + assert_eq!(announced.endpoints, ["a.example.com:443"]); + let described = format!("{announced:?}"); + assert!(!described.contains(&"a".repeat(64)), "{described}"); + } + fn jwt_alias(placeholder: &str) -> String { // Claims are irrelevant to injection authority: retain the exact issued // identity in the signature while giving clients a parseable JWT envelope. diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 7cf630e420..e28c258fc4 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -6363,6 +6363,13 @@ async fn handle_forward_proxy( return Ok(()); } }; + // Subscribe before credentials are resolved, so a rotation that lands + // while this request is relayed still closes it, including after an + // upgrade that authenticated with the old value. + let credential_rotation = l7_ctx + .provider_credentials + .as_ref() + .map(|credentials| credentials.rotation_watch(&host_lc, port)); // Static credentials are intentionally acquired only after every // asynchronous admission step. Holding an endpoint-scoped resolver across // middleware or token-grant awaits would let a revoked generation reach @@ -6610,7 +6617,7 @@ async fn handle_forward_proxy( let signing_region = forward_upgrade_config .as_ref() .map_or("", |config| config.signing_region.as_str()); - let outcome_result = relay_rewritten_forward_request( + let forward_relay = relay_rewritten_forward_request( method, &upstream_target, rewritten, @@ -6638,8 +6645,19 @@ async fn handle_forward_proxy( }), endpoint_observer: endpoint_observer.as_ref(), }, - ) - .await; + ); + let outcome_result = tokio::select! { + outcome = forward_relay => outcome, + rotated = relay::credential_rotated(credential_rotation.as_ref()) => { + relay::emit_rotated_relay_close(&l7_ctx, &rotated); + if let Some(session) = middleware_session.take() { + session + .end(openshell_core::proto::MiddlewareSessionEndReason::Cancellation) + .await; + } + return Ok(()); + } + }; let outcome_result = match outcome_result { Err(report) => { if let Some(error) = report.downcast_ref::() { @@ -6754,15 +6772,19 @@ async fn handle_forward_proxy( upgrade_options.websocket.permessage_deflate = websocket_permessage_deflate; upgrade_options.middleware_session = middleware_session.take(); upgrade_options.selected_subprotocol = websocket_subprotocol; - crate::l7::relay::handle_upgrade( - client, - &mut upstream, - overflow, - &host_lc, - port, - upgrade_options, - ) - .await?; + tokio::select! { + result = crate::l7::relay::handle_upgrade( + client, + &mut upstream, + overflow, + &host_lc, + port, + upgrade_options, + ) => result?, + rotated = relay::credential_rotated(credential_rotation.as_ref()) => { + relay::emit_rotated_relay_close(&l7_ctx, &rotated); + } + } } } @@ -9494,6 +9516,192 @@ network_policies: scenario.expect("compressed plaintext WebSocket scenario should complete"); } + #[tokio::test] + async fn forward_websocket_closes_when_its_credential_rotates() { + if !cfg!(target_os = "linux") { + eprintln!("skipping: handler identity binding requires /proc (Linux)"); + return; + } + let Some(upstream_ip) = non_loopback_test_ipv4() else { + eprintln!("skipping: no routable non-loopback IPv4 test address"); + return; + }; + + let upstream_listener = TcpListener::bind((upstream_ip, 0)) + .await + .expect("bind upstream listener"); + let upstream_port = upstream_listener.local_addr().unwrap().port(); + let executable = std::env::current_exe().expect("current executable"); + let data = format!( + r#" +network_policies: + allow-upstream: + name: allow-upstream + endpoints: + - host: "{upstream_ip}" + port: {upstream_port} + binaries: + - {{ path: "{executable}" }} +"#, + executable = executable.display(), + ); + let engine = Arc::new( + OpaEngine::from_strings(include_str!("../data/sandbox-policy.rego"), &data) + .expect("load policy"), + ); + let credentials = |revision: u64, value: &str, other: &str| { + use openshell_core::proto::{StaticCredentialBinding, StaticCredentialEndpointBinding}; + let binding = |host: String, port: u32, handle: char| StaticCredentialBinding { + endpoints: vec![StaticCredentialEndpointBinding { + host, + port, + path: "/**".into(), + }], + credential_identity: format!("refresh:{}", handle.to_string().repeat(64)), + workload_credential_handle: handle.to_string().repeat(64), + }; + ProviderCredentialState::from_bound_environment_with_stable_placeholders( + revision, + std::collections::HashMap::from([ + ("WS_TOKEN".to_string(), value.to_string()), + ("OTHER_TOKEN".to_string(), other.to_string()), + ]), + std::collections::HashMap::new(), + std::collections::HashMap::new(), + std::collections::HashMap::from([ + ( + "WS_TOKEN".to_string(), + binding(upstream_ip.to_string(), u32::from(upstream_port), 'a'), + ), + ( + "OTHER_TOKEN".to_string(), + binding("other.example.com".to_string(), 443, 'b'), + ), + ]), + Vec::new(), + vec!["WS_TOKEN".to_string(), "OTHER_TOKEN".to_string()], + ) + .expect("valid bindings") + }; + let live = credentials(1, "first", "other-1"); + let placeholder = live.snapshot().child_env["WS_TOKEN"].clone(); + + let upstream = tokio::spawn(async move { + let (mut socket, _) = upstream_listener.accept().await.unwrap(); + let request = read_http_headers_unbounded(&mut socket).await; + let request = String::from_utf8_lossy(&request).into_owned(); + socket + .write_all( + b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n\r\n", + ) + .await + .unwrap(); + let mut rest = [0u8; 64]; + let closed = socket.read(&mut rest).await.unwrap() == 0; + (request, closed) + }); + + let proxy_listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind proxy listener"); + let proxy_address = proxy_listener.local_addr().unwrap(); + let target = format!("http://{upstream_ip}:{upstream_port}/ws"); + let request = format!( + "GET {target} HTTP/1.1\r\nHost: {upstream_ip}:{upstream_port}\r\nAuthorization: Bearer {placeholder}\r\nConnection: Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n" + ); + let (upgraded_tx, upgraded) = tokio::sync::oneshot::channel(); + let client = tokio::spawn(async move { + let mut socket = TcpStream::connect(proxy_address) + .await + .expect("connect proxy"); + let response = read_http_headers_unbounded(&mut socket).await; + assert!(String::from_utf8_lossy(&response).contains("101 Switching Protocols")); + let _ = upgraded_tx.send(()); + let mut rest = [0u8; 64]; + socket.read(&mut rest).await.unwrap_or(0) + }); + let (proxy_connection, _) = proxy_listener.accept().await.unwrap(); + let socket_addrs = proxy_connection + .peer_addr() + .ok() + .zip(proxy_connection.local_addr().ok()); + let stream: BoundaryDuplexStream = Box::new(proxy_connection); + let mut proxy_connection = tokio::io::BufReader::new(stream); + let handler_credentials = live.clone(); + let handler = tokio::spawn(async move { + Box::pin(handle_forward_proxy( + "GET", + &target, + request.as_bytes(), + request.len(), + &mut proxy_connection, + None, + socket_addrs, + engine, + Arc::new(BinaryIdentityCache::new()), + Arc::new(AtomicU32::new(std::process::id())), + None, + AgentProposals::default(), + Arc::new(None), + Arc::new(None), + Some(handler_credentials), + None, + None, + None, + None, + None, + )) + .await + }); + + let within = std::time::Duration::from_secs(10); + tokio::time::timeout(within, upgraded) + .await + .expect("upgrade established") + .expect("client reports the upgrade"); + + // Rotating another provider's credential leaves the WebSocket open. + let other = credentials(2, "first", "other-2"); + let keys = live + .value_only_rotation(&other) + .expect("only OTHER_TOKEN changed"); + assert_eq!(keys, ["OTHER_TOKEN"]); + live.install_prepared(&other); + assert_eq!(live.announce_rotation(&keys).tunnels, 0); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + assert!( + !handler.is_finished(), + "unrelated rotation keeps the WebSocket open" + ); + + // Rotating its own credential closes it. + let rotated = credentials(3, "second", "other-2"); + let keys = live + .value_only_rotation(&rotated) + .expect("WS_TOKEN changed"); + live.install_prepared(&rotated); + assert_eq!(live.announce_rotation(&keys).tunnels, 1); + + tokio::time::timeout(within, handler) + .await + .expect("rotation closes the upgraded forward connection") + .expect("join handler") + .expect("rotation closure is clean"); + let (seen, upstream_closed) = tokio::time::timeout(within, upstream) + .await + .expect("upstream observes the close") + .expect("join upstream"); + assert!(seen.contains("Authorization: Bearer first\r\n"), "{seen}"); + assert!(upstream_closed, "upstream sees EOF, not further frames"); + assert_eq!( + tokio::time::timeout(within, client) + .await + .expect("client observes the close") + .expect("join client"), + 0 + ); + } + #[tokio::test] async fn malformed_request_lines_are_rejected_before_connect_or_forward_dispatch() { for host in ["api.example.com", "unmatched.example.com"] { diff --git a/crates/openshell-supervisor-network/src/proxy/relay.rs b/crates/openshell-supervisor-network/src/proxy/relay.rs index 54c93586cb..743a6cb3fd 100644 --- a/crates/openshell-supervisor-network/src/proxy/relay.rs +++ b/crates/openshell-supervisor-network/src/proxy/relay.rs @@ -8,9 +8,14 @@ use crate::l7::relay::L7EvalContext; use crate::opa::{NetworkAction, OpaEngine, PolicyGenerationGuard, TunnelPolicyEngine}; use miette::{IntoDiagnostic, Result}; use openshell_core::activity::ActivitySender; +use openshell_core::credential_rotation::{CredentialRotationWatch, RotatedCredential}; use openshell_core::endpoint_status::EndpointObservationSender; use openshell_core::proto::ProviderProfileCredential; use openshell_core::secrets::SecretResolver; +use openshell_ocsf::{ + ActionId, ActivityId, DispositionId, Endpoint, NetworkActivityBuilder, SeverityId, StatusId, + ocsf_emit, +}; use std::collections::HashMap; use std::sync::Arc; use tokio::io::{AsyncRead, AsyncWrite}; @@ -244,6 +249,13 @@ where C: AsyncRead + AsyncWrite + Unpin + Send, U: AsyncRead + AsyncWrite + Unpin + Send, { + // Subscribe before any request can carry credential material, so a + // rotation that lands while the relay starts still closes it. + let rotation = context + .request + .provider_credentials + .as_ref() + .map(|credentials| credentials.rotation_watch(&context.request.host, context.request.port)); match context.policy { PreparedHttpPolicy::Inspect { configs, evaluator } if configs.len() == 1 => { let generation_guard = evaluator.generation_guard().clone(); @@ -259,6 +271,10 @@ where emit_stale_relay_close(context.request, &generation_guard); Ok(()) } + rotated = credential_rotated(rotation.as_ref()) => { + emit_rotated_relay_close(context.request, &rotated); + Ok(()) + } } } PreparedHttpPolicy::Inspect { configs, evaluator } => { @@ -275,6 +291,10 @@ where emit_stale_relay_close(context.request, &generation_guard); Ok(()) } + rotated = credential_rotated(rotation.as_ref()) => { + emit_rotated_relay_close(context.request, &rotated); + Ok(()) + } } } PreparedHttpPolicy::Passthrough { generation_guard } => { @@ -290,6 +310,10 @@ where emit_stale_relay_close(context.request, &generation_guard); Ok(()) } + rotated = credential_rotated(rotation.as_ref()) => { + emit_rotated_relay_close(context.request, &rotated); + Ok(()) + } } } } @@ -346,18 +370,90 @@ where Ok(()) } +/// Resolve when a credential bound to this relay's destination rotates. +/// Relays without provider credentials never close for rotation. +pub(super) async fn credential_rotated( + rotation: Option<&CredentialRotationWatch>, +) -> RotatedCredential { + match rotation { + Some(rotation) => rotation.rotated().await, + None => std::future::pending().await, + } +} + +/// Why an established relay was closed by the supervisor. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum RelayCloseReason { + /// A static credential bound to the destination got a new value. + CredentialRotated, + /// The policy generation advanced: a policy, provider binding, + /// attachment or credential revocation change. + PolicyChanged, +} + +impl RelayCloseReason { + fn as_str(self) -> &'static str { + match self { + Self::CredentialRotated => "credential_rotated", + Self::PolicyChanged => "policy_changed", + } + } +} + fn emit_stale_relay_close(request: &L7EvalContext, guard: &PolicyGenerationGuard) { - emit_l7_tunnel_close_after_policy_change( - &request.host, - request.port, - miette::miette!( - "policy generation is stale [captured_generation:{} current_generation:{}]", + emit_relay_close( + request, + RelayCloseReason::PolicyChanged, + &[], + &format!( + "captured_generation:{} current_generation:{}", guard.captured_generation(), guard.current_generation(), ), ); } +pub(super) fn emit_rotated_relay_close(request: &L7EvalContext, rotated: &RotatedCredential) { + let detail = if rotated.keys.is_empty() { + "rotation history overrun; closed conservatively".to_string() + } else { + format!("keys:{}", rotated.keys.join(",")) + }; + emit_relay_close( + request, + RelayCloseReason::CredentialRotated, + &rotated.keys, + &detail, + ); +} + +/// One Close event per supervisor-closed relay. Keys are environment key +/// names, never values. +fn emit_relay_close( + request: &L7EvalContext, + reason: RelayCloseReason, + credential_keys: &[String], + detail: &str, +) { + let severity = match reason { + RelayCloseReason::CredentialRotated => SeverityId::Low, + RelayCloseReason::PolicyChanged => SeverityId::Medium, + }; + let event = NetworkActivityBuilder::new(openshell_ocsf::ctx::ctx()) + .activity(ActivityId::Close) + .action(ActionId::Denied) + .disposition(DispositionId::Blocked) + .severity(severity) + .status(StatusId::Success) + .dst_endpoint(Endpoint::from_domain(&request.host, request.port)) + .firewall_rule(&request.policy_name, "relay") + .unmapped("close_reason", reason.as_str()) + .unmapped("credential_keys", credential_keys.to_vec()) + .message(format!("Tunnel closed: {} [{detail}]", reason.as_str())) + .build(); + ocsf_emit!(event); +} + #[cfg(test)] mod tests { use super::super::{EgressIntent, EndpointDecision, ProcessIdentityEvidence}; @@ -600,6 +696,152 @@ mod tests { ); } + fn two_provider_credentials( + a_value: &str, + ) -> openshell_core::provider_credentials::ProviderCredentialState { + use openshell_core::proto::{StaticCredentialBinding, StaticCredentialEndpointBinding}; + let binding = |host: &str, handle: char| StaticCredentialBinding { + endpoints: vec![StaticCredentialEndpointBinding { + host: host.into(), + port: 80, + path: "/**".into(), + }], + // Stable credentials carry their handle in the identity, as + // the gateway issues them. + credential_identity: format!("refresh:{}", handle.to_string().repeat(64)), + workload_credential_handle: handle.to_string().repeat(64), + }; + openshell_core::provider_credentials::ProviderCredentialState::from_bound_environment_with_stable_placeholders( + 7, + HashMap::from([ + ("A_TOKEN".to_string(), a_value.to_string()), + ("B_TOKEN".to_string(), "b-value".to_string()), + ]), + HashMap::new(), + HashMap::new(), + HashMap::from([ + ("A_TOKEN".to_string(), binding("a.example.com", 'a')), + ("B_TOKEN".to_string(), binding("b.example.com", 'b')), + ]), + Vec::new(), + vec!["A_TOKEN".to_string(), "B_TOKEN".to_string()], + ) + .expect("valid bindings") + } + + /// A credential-bearing relay to `host`, with the sandbox side, the + /// upstream side and the relay task. + fn spawn_credential_relay( + engine: &Arc, + credentials: &openshell_core::provider_credentials::ProviderCredentialState, + host: &str, + ) -> ( + tokio::io::DuplexStream, + tokio::io::DuplexStream, + tokio::task::JoinHandle>, + ) { + let (caller, mut client) = tokio::io::duplex(4096); + let (mut upstream, server) = tokio::io::duplex(4096); + let engine = Arc::clone(engine); + let mut request = request_context(); + request.host = host.to_string(); + request.provider_credentials = Some(credentials.clone()); + let relay = tokio::spawn(async move { + let decision = decision(engine.current_generation()); + let context = prepare_http_relay(None, &engine, &decision, &request) + .expect("current generation prepares a relay"); + relay_http_stream(&mut client, &mut upstream, context).await + }); + (caller, server, relay) + } + + /// Send one request carrying `placeholder` and return what upstream saw. + async fn exchange( + caller: &mut tokio::io::DuplexStream, + server: &mut tokio::io::DuplexStream, + host: &str, + placeholder: &str, + ) -> String { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let request = format!( + "GET /v1 HTTP/1.1\r\nHost: {host}\r\nAuthorization: Bearer {placeholder}\r\n\r\n" + ); + caller.write_all(request.as_bytes()).await.unwrap(); + let mut received = Vec::new(); + while !received.ends_with(b"\r\n\r\n") { + received.push(server.read_u8().await.unwrap()); + } + let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"; + server.write_all(response).await.unwrap(); + let mut reply = vec![0; response.len()]; + caller.read_exact(&mut reply).await.unwrap(); + String::from_utf8(received).unwrap() + } + + #[tokio::test] + async fn credential_rotation_closes_only_the_rotated_providers_relays() { + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); + let credentials = two_provider_credentials("a-initial"); + let env = credentials.snapshot().child_env.clone(); + let (a_placeholder, b_placeholder) = (&env["A_TOKEN"], &env["B_TOKEN"]); + let within = std::time::Duration::from_secs(5); + + let (mut a_caller, mut a_server, a_relay) = + spawn_credential_relay(&engine, &credentials, "a.example.com"); + let (mut b_caller, mut b_server, b_relay) = + spawn_credential_relay(&engine, &credentials, "b.example.com"); + let seen = tokio::time::timeout( + within, + exchange(&mut a_caller, &mut a_server, "a.example.com", a_placeholder), + ) + .await + .expect("provider A request relayed"); + assert!(seen.contains("Bearer a-initial"), "{seen}"); + let seen = tokio::time::timeout( + within, + exchange(&mut b_caller, &mut b_server, "b.example.com", b_placeholder), + ) + .await + .expect("provider B request relayed"); + assert!(seen.contains("Bearer b-value"), "{seen}"); + + let candidate = two_provider_credentials("a-rotated"); + let rotated = credentials + .value_only_rotation(&candidate) + .expect("only A's value changed"); + assert_eq!(rotated, ["A_TOKEN"]); + credentials.install_prepared(&candidate); + let announced = credentials.announce_rotation(&rotated); + assert_eq!(announced.tunnels, 1); + assert_eq!(announced.endpoints, ["a.example.com:80"]); + + tokio::time::timeout(within, a_relay) + .await + .expect("provider A's relay closes on rotation") + .expect("relay task should not panic") + .expect("rotation closure is clean"); + assert!(!b_relay.is_finished(), "provider B's relay stays open"); + let seen = tokio::time::timeout( + within, + exchange(&mut b_caller, &mut b_server, "b.example.com", b_placeholder), + ) + .await + .expect("provider B's relay still carries requests"); + assert!(seen.contains("Bearer b-value"), "{seen}"); + + // The workload's placeholder is unchanged; a reconnect sends the new value. + let (mut a_caller, mut a_server, _a_relay) = + spawn_credential_relay(&engine, &credentials, "a.example.com"); + let seen = tokio::time::timeout( + within, + exchange(&mut a_caller, &mut a_server, "a.example.com", a_placeholder), + ) + .await + .expect("reconnected provider A request relayed"); + assert!(seen.contains("Bearer a-rotated"), "{seen}"); + b_relay.abort(); + } + #[tokio::test] async fn raw_relay_closes_immediately_when_fail_closed_generation_is_published() { let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); diff --git a/crates/openshell-supervisor/src/lib.rs b/crates/openshell-supervisor/src/lib.rs index 69f6114043..1d8f4c8ea2 100644 --- a/crates/openshell-supervisor/src/lib.rs +++ b/crates/openshell-supervisor/src/lib.rs @@ -2814,6 +2814,36 @@ fn provider_environment_is_installable(reason: ProviderReadinessReason) -> bool ) } +/// One summary per value-only rotation. The relays it closes each emit +/// their own Close event. +fn emit_credential_rotation( + keys: &[String], + announced: &openshell_core::credential_rotation::RotationAnnouncement, +) { + info!( + credential_keys = ?keys, + endpoints = ?announced.endpoints, + tunnels_closed = announced.tunnels, + "Provider credentials rotated; closing tunnels to their endpoints" + ); + ocsf_emit!( + ConfigStateChangeBuilder::new(ocsf_ctx()) + .severity(SeverityId::Informational) + .status(StatusId::Success) + .state(StateId::Enabled, "credential_rotated") + .unmapped("credential_keys", serde_json::json!(keys)) + .unmapped("endpoints", serde_json::json!(announced.endpoints)) + .unmapped("tunnels_closed", serde_json::json!(announced.tunnels)) + .message(format!( + "Provider credentials rotated without a policy reload [keys:{} endpoints:{} tunnels_closed:{}]", + keys.join(","), + announced.endpoints.join(","), + announced.tunnels + )) + .build() + ); +} + fn prepare_provider_environment( provider: &openshell_core::grpc_client::ProviderEnvironmentResult, ) -> Result { @@ -4426,11 +4456,13 @@ async fn run_policy_poll_loop_with_client( let recovering_rejected_policy = reloads_gateway_policy && rejected_policy_generation.is_some() && result.configuration_admitted; - let policy_runtime_changed = (reloads_gateway_policy - && (provider_env_changed - || current_policy_generation - .as_ref() - .is_some_and(PolicyGenerationGuard::is_stale))) + // Everything except new provider credentials that forces a new + // policy generation. A provider change alone may turn out to be a + // value-only rotation, which closes only the relays it affects. + let policy_runtime_changed_by_configuration = (reloads_gateway_policy + && current_policy_generation + .as_ref() + .is_some_and(PolicyGenerationGuard::is_stale)) || recovering_rejected_policy || extension_authentication_changed || gateway_policy_runtime_needs_reconciliation( @@ -4441,6 +4473,8 @@ async fn run_policy_poll_loop_with_client( &result.supervisor_middleware_services, middleware_registry_status, ); + let policy_runtime_changed = policy_runtime_changed_by_configuration + || (reloads_gateway_policy && provider_env_changed); // Recovery already has its own acknowledgement path below. Giving it // precedence here prevents a restored last-known-good policy from // also being acknowledged as an ordinary same-hash revision. @@ -4645,6 +4679,27 @@ async fn run_policy_poll_loop_with_client( None }; + // A value-only rotation installs without a policy reload, so tunnels + // to other providers stay open. + // A local policy override never reloads for provider changes, so a + // value-only rotation there closes its relays the same way. + let rotated_credential_keys = if provider_env_changed + && (!reloads_gateway_policy + || (!policy_runtime_changed_by_configuration && !policy_changed)) + { + prepared_provider + .as_ref() + .and_then(|prepared| ctx.provider_credentials.value_only_rotation(prepared)) + } else { + None + }; + // A rotation replaces only the provider-driven reload; any other + // reason to reload still applies. + let policy_runtime_changed = policy_runtime_changed_by_configuration + || (reloads_gateway_policy + && provider_env_changed + && rotated_credential_keys.is_none()); + if policy_runtime_changed { let pid = ctx.entrypoint_pid.load(Ordering::Acquire); let runtime_result = reload_gateway_configuration_runtime( @@ -4874,7 +4929,9 @@ async fn run_policy_poll_loop_with_client( report_runtime_configuration(&ctx, &result, false, "Effective configuration failed runtime preparation; prior credentials remain installed").await; continue; } - if !reloads_gateway_policy && let Some(prepared) = prepared_provider.as_ref() { + if (!reloads_gateway_policy || rotated_credential_keys.is_some()) + && let Some(prepared) = prepared_provider.as_ref() + { ctx.provider_credentials.install_prepared(prepared); if let Some((identity, expires_at_ms)) = prepared_identity.as_ref() { ctx.provider_readiness.credentials_installed( @@ -4884,6 +4941,12 @@ async fn run_policy_poll_loop_with_client( ); } } + if let Some(keys) = rotated_credential_keys.as_ref() { + // Announce only after installing, so a relay that reconnects + // after closing resolves the new value. + let announced = ctx.provider_credentials.announce_rotation(keys); + emit_credential_rotation(keys, &announced); + } if provider_env_changed || policy_runtime_reconciled { current_provider_env_revision = result.provider_env_revision; } @@ -8070,6 +8133,174 @@ network_policies: let _ = task.await; } + #[tokio::test] + async fn value_only_stable_rotation_closes_its_relays_without_a_policy_reload() { + let engine = Arc::new(OpaEngine::from_proto(&proto_policy_fixture()).unwrap()); + let initial = settings_poll_result( + Some(proto_policy_fixture()), + 1, + openshell_core::proto::PolicySource::Sandbox, + ); + let ctx = policy_poll_test_context( + engine.clone(), + LoadedPolicyOrigin::Gateway { + revision: Some(LoadedPolicyRevision::from_snapshot(&initial)), + has_last_valid_policy: true, + }, + default_middleware_connector(), + ); + let state = ctx.provider_credentials.clone(); + let tracker = ctx.provider_readiness.clone(); + let (policy, polls, mut reports) = scripted_policy_gateway(); + let (requests, mut received) = tokio::sync::mpsc::unbounded_channel(); + polls.send(initial.clone()).unwrap(); + let task = tokio::spawn(run_policy_poll_loop_with_client( + ctx, + ScriptedProviderGateway { policy, requests }, + )); + expect_policy_report(&mut reports, 1).await; + let reference = format!("openshell:resolve:env:s{}_EXTERNAL_TOKEN", "a".repeat(64)); + let resolved = |state: &ProviderCredentialState| { + state + .resolver_for_endpoint("tools.example.com", 443, "/v1/chat") + .and_then(|resolver| resolver.resolve_placeholder(&reference).map(str::to_owned)) + }; + + // The first delivery changes the credential shape, so it reloads. + let mut poll = initial.clone(); + poll.provider_env_revision = 1; + polls.send(poll).unwrap(); + let (_, response) = timeout(Duration::from_secs(5), received.recv()) + .await + .expect("provider refresh requested") + .expect("poll loop active"); + assert!( + response + .send(Ok(stable_provider_environment(1, Some("initial")))) + .is_ok() + ); + timeout(Duration::from_secs(5), async { + while resolved(&state).as_deref() != Some("initial") { + tokio::task::yield_now().await; + } + }) + .await + .expect("initial credentials installed"); + + let generation = engine.current_generation(); + let bound = state.rotation_watch("tools.example.com", 443); + let unrelated = state.rotation_watch("other.example.com", 443); + + let mut poll = initial.clone(); + poll.provider_env_revision = 2; + polls.send(poll).unwrap(); + let (_, response) = timeout(Duration::from_secs(5), received.recv()) + .await + .expect("provider refresh requested") + .expect("poll loop active"); + assert!( + response + .send(Ok(stable_provider_environment(2, Some("rotated")))) + .is_ok() + ); + + let rotated = timeout(Duration::from_secs(5), bound.rotated()) + .await + .expect("relay bound to the rotated credential closes"); + assert_eq!(&*rotated.keys, ["EXTERNAL_TOKEN".to_string()]); + // The workload keeps its placeholder, which now resolves the new value. + assert_eq!(state.snapshot().child_env["EXTERNAL_TOKEN"], reference); + assert_eq!(resolved(&state).as_deref(), Some("rotated")); + // No new policy generation, so other relays stay open. + assert_eq!(engine.current_generation(), generation); + assert!( + timeout(Duration::from_millis(50), unrelated.rotated()) + .await + .is_err() + ); + // The rotation establishes readiness, so the same revision is not + // fetched again. + timeout(Duration::from_secs(5), async { + while !tracker.observation(&state).credentials_installed { + tokio::task::yield_now().await; + } + }) + .await + .expect("rotated credentials reported installed"); + let mut poll = initial.clone(); + poll.provider_env_revision = 2; + polls.send(poll).unwrap(); + assert!( + timeout(Duration::from_millis(200), received.recv()) + .await + .is_err(), + "an installed rotation must not be fetched again" + ); + + task.abort(); + let _ = task.await; + } + + #[tokio::test] + async fn revision_scoped_rotation_still_reloads_policy() { + let engine = Arc::new(OpaEngine::from_proto(&proto_policy_fixture()).unwrap()); + let initial = settings_poll_result( + Some(proto_policy_fixture()), + 1, + openshell_core::proto::PolicySource::Sandbox, + ); + let ctx = policy_poll_test_context( + engine.clone(), + LoadedPolicyOrigin::Gateway { + revision: Some(LoadedPolicyRevision::from_snapshot(&initial)), + has_last_valid_policy: true, + }, + default_middleware_connector(), + ); + let state = ctx.provider_credentials.clone(); + let (policy, polls, mut reports) = scripted_policy_gateway(); + let (requests, mut received) = tokio::sync::mpsc::unbounded_channel(); + polls.send(initial.clone()).unwrap(); + let task = tokio::spawn(run_policy_poll_loop_with_client( + ctx, + ScriptedProviderGateway { policy, requests }, + )); + expect_policy_report(&mut reports, 1).await; + + let mut generation = engine.current_generation(); + for (revision, value) in [(1, "initial"), (2, "rotated")] { + let mut poll = initial.clone(); + poll.provider_env_revision = revision; + polls.send(poll).unwrap(); + let (_, response) = timeout(Duration::from_secs(5), received.recv()) + .await + .expect("provider refresh requested") + .expect("poll loop active"); + assert!( + response + .send(Ok(static_provider_environment(revision, Some(value)))) + .is_ok() + ); + timeout(Duration::from_secs(5), async { + while state.revision() != revision { + tokio::task::yield_now().await; + } + }) + .await + .expect("credentials installed"); + // A revision-scoped placeholder changes with every value, so the + // rotation is not value-only and keeps the generation reload. + assert!( + engine.current_generation() > generation, + "revision {revision}" + ); + generation = engine.current_generation(); + } + + task.abort(); + let _ = task.await; + } + #[tokio::test] async fn provider_poll_preserves_stable_delivery_across_rotation_failure_and_detach() { let engine = Arc::new(OpaEngine::from_proto(&proto_policy_fixture()).unwrap()); diff --git a/docs/how-it-works/providers/profiles.mdx b/docs/how-it-works/providers/profiles.mdx index 7ff7d7c8e6..10ee40d8d4 100644 --- a/docs/how-it-works/providers/profiles.mdx +++ b/docs/how-it-works/providers/profiles.mdx @@ -172,6 +172,22 @@ value. This lets a long-running process use each newly minted access token without restarting. OpenShell does not retain older values behind this stable placeholder. +When a provider update changes only the values of static credentials that use +stable placeholders, the supervisor installs the new values without reloading +the sandbox policy. It closes only the open connections whose destination +matches an endpoint bound to a rotated credential, so a connection that +authenticated with the old value, such as a WebSocket, cannot keep using it. +Connections to other providers stay open. The client reconnects and its next +request carries the new value through the same placeholder. Each closed +connection emits an OCSF network activity Close event with `close_reason` +`credential_rotated` and the rotated `credential_keys`, and the supervisor +logs one summary per rotation with the rotated endpoints and the number of +connections it closed. A gateway-managed refresh is a rotation too, so each +refresh closes the connections to that credential's endpoints. Any +other provider change, including a new value for a credential with a +revision-scoped placeholder, reloads the policy and closes every open +connection with `close_reason` `policy_changed`. + Explicitly configuring refresh again starts a new authorization epoch, even when the provider and credential key are unchanged. Reconfiguration, provider replacement or detachment, refresh deletion, and endpoint-boundary changes