From 636f975093f7cda68596ec3c5ff2bcea90f9c781 Mon Sep 17 00:00:00 2001 From: Matt Jane Date: Thu, 15 Oct 2026 13:08:19 -0700 Subject: [PATCH] fix(supervisor): close only the rotated provider's tunnels on credential rotation A provider environment change always reloaded the sandbox policy into a new generation, and every generation-pinned relay closed. Rotating one provider's credential therefore dropped every open connection in the sandbox, including other providers' WebSockets, and logged it as a denied open "before inspection" with no provider or reason. When the only difference is new values for static credentials with stable placeholders, the supervisor now installs them without a policy reload and announces the rotation. Each credential-bearing relay (CONNECT HTTP relays, TLS-terminated relays, forward-proxy requests and their upgrades) watches for rotations whose binding endpoints match its destination and closes, so no connection keeps working on the old value. Connections to other providers stay open. Every other provider change, including a new revision-scoped value, still reloads the policy. Relay closes are now NetworkActivity Close events with close_reason credential_rotated or policy_changed and the rotated credential key names. Each rotation logs one summary with the keys, the rotated endpoints and the number of connections closed. Neither carries values, placeholders or credential identities, since a stable credential's identity embeds its placeholder handle. Refs: ENG-11222 Signed-off-by: Matt Jane --- .../openshell-core/src/credential_rotation.rs | 353 ++++++++++++++++ crates/openshell-core/src/endpoint_path.rs | 5 + crates/openshell-core/src/host_pattern.rs | 5 + crates/openshell-core/src/lib.rs | 1 + .../src/provider_credentials.rs | 378 ++++++++++++++++++ .../openshell-supervisor-network/src/proxy.rs | 232 ++++++++++- .../src/proxy/relay.rs | 252 +++++++++++- crates/openshell-supervisor/src/lib.rs | 243 ++++++++++- docs/how-it-works/providers/profiles.mdx | 16 + 9 files changed, 1462 insertions(+), 23 deletions(-) create mode 100644 crates/openshell-core/src/credential_rotation.rs 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