diff --git a/crates/openshell-supervisor-middleware/src/lib.rs b/crates/openshell-supervisor-middleware/src/lib.rs index 63e71c869b..c1bfc6da7e 100644 --- a/crates/openshell-supervisor-middleware/src/lib.rs +++ b/crates/openshell-supervisor-middleware/src/lib.rs @@ -364,7 +364,7 @@ impl OnError { } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq)] pub struct ChainEntry { pub name: String, pub implementation: String, diff --git a/crates/openshell-supervisor-network/src/l7/relay.rs b/crates/openshell-supervisor-network/src/l7/relay.rs index 61c342bc17..75f5da882d 100644 --- a/crates/openshell-supervisor-network/src/l7/relay.rs +++ b/crates/openshell-supervisor-network/src/l7/relay.rs @@ -3629,6 +3629,11 @@ where let mut response_selection = None; let req = if let Some(engine) = middleware_engine { let input = middleware_network_input(ctx); + // Give a revalidating guard the chance to re-pin through an + // unrelated reload before comparing raw generation numbers. + if generation_guard.is_stale() { + return Ok(()); + } let (chain, generation) = engine.query_middleware_chain_with_generation(&input)?; if generation != generation_guard.captured_generation() { return Ok(()); diff --git a/crates/openshell-supervisor-network/src/opa.rs b/crates/openshell-supervisor-network/src/opa.rs index 257bd3ce2e..50b8dc8b16 100644 --- a/crates/openshell-supervisor-network/src/opa.rs +++ b/crates/openshell-supervisor-network/src/opa.rs @@ -33,6 +33,9 @@ use tokio::sync::watch; use tracing::info; mod raw_schema; +mod revalidation; + +pub use revalidation::{AuthorizedConnection, ConnectionPolicyContract, PolicyRevalidator}; /// Baked-in rego rules for OPA policy evaluation. /// These rules define the network access decision logic and static config @@ -85,7 +88,7 @@ pub enum NetworkAction { } /// Endpoint identity and metadata captured with one policy generation. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq)] pub struct MatchedEndpoint { pub policy_name: String, pub endpoint_index: usize, @@ -115,6 +118,7 @@ pub struct EgressAuthorization { } /// Input for a network access policy evaluation. +#[derive(Debug, Clone)] pub struct NetworkInput { pub host: String, pub port: u16, @@ -183,6 +187,12 @@ pub struct OpaEngine { websocket_assembly_budget: crate::l7::websocket::WebSocketAssemblyBudget, generation_tx: watch::Sender, fail_closed_reason: RwLock>, + /// Generation published by the most recent fail-closed quarantine, or 0. + /// Written under the engine lock. + last_fail_closed_generation: AtomicU64, + /// Incremented whenever the middleware runner's registry is replaced. + /// Written under the engine lock. + middleware_registry_epoch: AtomicU64, } #[cfg(test)] @@ -204,24 +214,68 @@ pub(crate) fn test_opa_query_count() -> u64 { } /// Generation guard captured when an HTTP tunnel or request path starts. +/// +/// A plain guard is stale as soon as the policy generation moves. A guard +/// with a [`PolicyRevalidator`] (see [`Self::with_revalidator`]) instead +/// re-checks the connection it protects and re-pins to the new generation +/// when the policy change does not affect that connection. +/// +/// Clones share one pin, so every clone held by a relay observes a re-pin +/// performed through any of them. #[derive(Clone, Debug)] pub struct PolicyGenerationGuard { - captured_generation: u64, + captured_generation: Arc, current_generation: Arc, generation_rx: watch::Receiver, + revalidator: Option>, } impl PolicyGenerationGuard { + fn pinned( + generation: u64, + current_generation: &Arc, + generation_tx: &watch::Sender, + ) -> Self { + Self { + captured_generation: Arc::new(AtomicU64::new(generation)), + current_generation: Arc::clone(current_generation), + generation_rx: generation_tx.subscribe(), + revalidator: None, + } + } + + /// Let this guard follow policy changes that leave its connection's + /// contract unchanged. `None` keeps the strict close-on-change behavior. + #[must_use] + pub fn with_revalidator(mut self, revalidator: Option>) -> Self { + self.revalidator = revalidator; + self + } + pub fn captured_generation(&self) -> u64 { - self.captured_generation + self.captured_generation.load(Ordering::Acquire) } pub fn current_generation(&self) -> u64 { self.current_generation.load(Ordering::Acquire) } + /// Whether the policy that authorized this connection no longer applies. + /// + /// With a revalidator, a generation change triggers one re-evaluation of + /// the connection (a short, synchronous OPA query under the engine lock). + /// Callers must not hold the OPA engine lock. pub fn is_stale(&self) -> bool { - self.current_generation() != self.captured_generation + let Some(revalidator) = &self.revalidator else { + return self.current_generation() != self.captured_generation(); + }; + if revalidator.is_invalidated() { + return true; + } + if self.current_generation() == self.captured_generation() { + return false; + } + !revalidator.follow(&self.captured_generation, &self.current_generation) } pub fn ensure_current(&self) -> Result<()> { @@ -235,10 +289,12 @@ impl PolicyGenerationGuard { Ok(()) } - /// Wait until the policy generation changes. + /// Wait until the policy that authorized this connection no longer applies. /// /// Relay boundaries use this to close even an idle or raw stream as soon - /// as a new generation (including fail-closed quarantine) is published. + /// as a relevant generation (always including fail-closed quarantine) is + /// published. A revalidating guard re-pins through unrelated changes and + /// keeps waiting. pub async fn wait_until_stale(&self) { let mut receiver = self.generation_rx.clone(); while !self.is_stale() { @@ -259,6 +315,18 @@ pub struct TunnelPolicyEngine { } impl TunnelPolicyEngine { + /// Let this tunnel survive policy changes that leave its connection's + /// contract unchanged. + /// + /// The cloned evaluator is kept as-is after a re-pin. That is sound + /// because the contract includes every endpoint per-request rules read; + /// see [`ConnectionPolicyContract`]. + #[must_use] + pub fn with_revalidator(mut self, revalidator: Option>) -> Self { + self.generation_guard = self.generation_guard.with_revalidator(revalidator); + self + } + pub fn captured_generation(&self) -> u64 { self.generation_guard.captured_generation() } @@ -317,9 +385,16 @@ impl OpaEngine { websocket_assembly_budget: crate::l7::websocket::WebSocketAssemblyBudget::default(), generation_tx, fail_closed_reason: RwLock::new(None), + last_fail_closed_generation: AtomicU64::new(0), + middleware_registry_epoch: AtomicU64::new(0), } } + /// Whether a fail-closed quarantine was published after `generation`. + pub(crate) fn fail_closed_after(&self, generation: u64) -> bool { + self.last_fail_closed_generation.load(Ordering::Acquire) > generation + } + /// Whether network authorization requires a workload binary identity. pub const fn binary_identity_required(&self) -> bool { self.binary_identity_required @@ -607,12 +682,69 @@ impl OpaEngine { #[cfg(test)] record_test_opa_query(); - let input_json = network_input_json(input); + let mut engine = self + .engine + .lock() + .map_err(|_| miette::miette!("OPA engine lock poisoned"))?; + self.authorize_egress_locked(&mut engine, input) + } + + /// Re-evaluate the contract of an established connection. + /// + /// Authorization and middleware selection run under one engine lock, so + /// the contract and the returned generation describe the same policy. + pub fn connection_policy_contract( + &self, + input: &NetworkInput, + ) -> Result<(ConnectionPolicyContract, u64)> { + let mut engine = self + .engine + .lock() + .map_err(|_| miette::miette!("OPA engine lock poisoned"))?; + let authorization = self.authorize_egress_locked(&mut engine, input)?; + let middleware_chain = if matches!(authorization.action, NetworkAction::Allow { .. }) { + query_middleware_chain_locked(&mut engine, input)? + } else { + Vec::new() + }; + let contract = ConnectionPolicyContract::new( + &AuthorizedConnection { + action: &authorization.action, + endpoint_configs: &authorization.endpoint_configs, + matched_endpoints: &authorization.matched_endpoints, + exact_declared_endpoint_host: authorization.exact_declared_endpoint_host, + }, + middleware_chain, + self.middleware_registry_epoch.load(Ordering::Acquire), + ); + Ok((contract, authorization.generation)) + } + /// Middleware chain and registry epoch for `input`, if `expected_generation` + /// is still current. Used to complete a connection's baseline contract. + pub(crate) fn middleware_selection_at( + &self, + input: &NetworkInput, + expected_generation: u64, + ) -> Result, u64)>> { let mut engine = self .engine .lock() .map_err(|_| miette::miette!("OPA engine lock poisoned"))?; + if self.current_generation() != expected_generation { + return Ok(None); + } + let chain = query_middleware_chain_locked(&mut engine, input)?; + let epoch = self.middleware_registry_epoch.load(Ordering::Acquire); + Ok(Some((chain, epoch))) + } + + fn authorize_egress_locked( + &self, + engine: &mut regorus::Engine, + input: &NetworkInput, + ) -> Result { + let input_json = network_input_json(input); let generation = self.current_generation(); let fail_closed_reason = self @@ -630,7 +762,7 @@ impl OpaEngine { }); } - set_regorus_input(&mut engine, input_json)?; + set_regorus_input(engine, input_json)?; let result = engine .eval_rule("data.openshell.sandbox.egress_authorization".into()) @@ -810,6 +942,7 @@ impl OpaEngine { *engine = new_engine; if let Some(new_runner) = new_runner { *runner = new_runner; + self.middleware_registry_epoch.fetch_add(1, Ordering::AcqRel); } *fail_closed_reason = None; self.generation_guard(generation) @@ -821,7 +954,9 @@ impl OpaEngine { /// The existing compiled engine remains available for an explicit /// `retain_last_valid` posture or a later valid reload, but all new network /// decisions deny with `reason` while the quarantine is active. Advancing - /// the generation invalidates and wakes every pinned relay. + /// the generation invalidates and wakes every pinned relay, including + /// revalidating relays: a quarantine is never re-pinned through, even if + /// the previous policy is restored before a relay observes it. pub fn enter_fail_closed(&self, reason: impl Into) -> Result { let _engine = self .engine @@ -832,7 +967,9 @@ impl OpaEngine { .write() .map_err(|_| miette::miette!("OPA fail-closed state lock poisoned"))? = Some(reason.into()); - Ok(self.advance_generation()) + let generation = self.advance_generation(); + self.last_fail_closed_generation.store(generation, Ordering::Release); + Ok(generation) } pub fn fail_closed_reason(&self) -> Option { @@ -904,6 +1041,7 @@ impl OpaEngine { .write() .map_err(|_| miette::miette!("middleware runner lock poisoned"))?; *runner = runner.with_replacement_registry(registry); + self.middleware_registry_epoch.fetch_add(1, Ordering::AcqRel); self.advance_generation(); Ok(()) } @@ -934,11 +1072,11 @@ impl OpaEngine { "policy changed before HTTP relay started [expected_generation:{expected_generation} current_generation:{generation}]" )); } - Ok(PolicyGenerationGuard { - captured_generation: generation, - current_generation: Arc::clone(&self.generation), - generation_rx: self.generation_tx.subscribe(), - }) + Ok(PolicyGenerationGuard::pinned( + generation, + &self.generation, + &self.generation_tx, + )) } /// Query static sandbox configuration from the OPA data module. @@ -1130,11 +1268,11 @@ impl OpaEngine { } Ok(TunnelPolicyEngine { engine: Mutex::new(engine.clone()), - generation_guard: PolicyGenerationGuard { - captured_generation: generation, - current_generation: Arc::clone(&self.generation), - generation_rx: self.generation_tx.subscribe(), - }, + generation_guard: PolicyGenerationGuard::pinned( + generation, + &self.generation, + &self.generation_tx, + ), middleware_runner: self.middleware_runner()?, websocket_assembly_budget: self.websocket_assembly_budget(), }) @@ -12591,4 +12729,288 @@ network_policies: cmdline_paths: vec![], } } + + // --- Connection revalidation across policy generations --- + + fn claude_input(host: &str) -> NetworkInput { + NetworkInput { + host: host.into(), + port: 443, + binary_path: PathBuf::from("/usr/local/bin/claude"), + binary_sha256: String::new(), + ancestors: vec![], + cmdline_paths: vec![], + } + } + + /// Pin a revalidating guard the way the proxy does after authorization. + fn revalidating_guard(engine: &Arc, input: &NetworkInput) -> PolicyGenerationGuard { + let authorization = engine.authorize_egress(input).unwrap(); + assert!( + matches!(authorization.action, NetworkAction::Allow { .. }), + "fixture must authorize {}", + input.host + ); + let revalidator = PolicyRevalidator::capture( + engine, + input.clone(), + &AuthorizedConnection { + action: &authorization.action, + endpoint_configs: &authorization.endpoint_configs, + matched_endpoints: &authorization.matched_endpoints, + exact_declared_endpoint_host: authorization.exact_declared_endpoint_host, + }, + authorization.generation, + None, + ); + assert!(revalidator.is_some(), "an allow must capture a contract"); + engine + .generation_guard(authorization.generation) + .unwrap() + .with_revalidator(revalidator) + } + + fn proto_endpoint(host: &str) -> NetworkEndpoint { + NetworkEndpoint { + host: host.to_string(), + port: 443, + ..Default::default() + } + } + + fn rest_endpoint(host: &str, path: &str) -> NetworkEndpoint { + NetworkEndpoint { + host: host.to_string(), + port: 443, + protocol: "rest".to_string(), + enforcement: openshell_core::proto::NetworkEnforcementMode::Enforce as i32, + rules: vec![L7Rule { + allow: Some(L7Allow { + method: "GET".to_string(), + path: path.to_string(), + ..Default::default() + }), + }], + ..Default::default() + } + } + + fn policy_mut<'a>(proto: &'a mut ProtoSandboxPolicy, name: &str) -> &'a mut NetworkPolicyRule { + proto.network_policies.get_mut(name).expect("fixture policy") + } + + #[test] + fn revalidating_guard_follows_reload_that_adds_an_unrelated_endpoint() { + let mut proto = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let input = claude_input("api.anthropic.com"); + let guard = revalidating_guard(&engine, &input); + let strict = engine.generation_guard(guard.captured_generation()).unwrap(); + + policy_mut(&mut proto, "gitlab") + .endpoints + .push(proto_endpoint("registry.gitlab.com")); + engine.reload_from_proto(&proto).unwrap(); + + assert!(strict.is_stale(), "a plain guard still closes on any change"); + assert!(!guard.is_stale(), "an unrelated endpoint must not close the connection"); + assert_eq!(guard.captured_generation(), engine.current_generation()); + assert!(guard.ensure_current().is_ok()); + } + + #[test] + fn revalidating_guard_ignores_endpoint_index_shifts_in_its_own_policy() { + let mut proto = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let guard = revalidating_guard(&engine, &claude_input("api.anthropic.com")); + + policy_mut(&mut proto, "claude_code") + .endpoints + .insert(0, proto_endpoint("console.anthropic.com")); + engine.reload_from_proto(&proto).unwrap(); + + assert!(!guard.is_stale()); + } + + #[test] + fn revalidating_guard_closes_when_its_endpoint_changes_and_stays_closed() { + let original = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&original).unwrap()); + let guard = revalidating_guard(&engine, &claude_input("api.anthropic.com")); + + let mut narrowed = original.clone(); + policy_mut(&mut narrowed, "claude_code").endpoints[0] = + rest_endpoint("api.anthropic.com", "/v1/messages"); + engine.reload_from_proto(&narrowed).unwrap(); + assert!(guard.is_stale(), "an L7 rule on this endpoint changes the contract"); + + engine.reload_from_proto(&original).unwrap(); + assert!( + guard.is_stale(), + "restoring the policy must not revive a connection already found stale" + ); + } + + #[test] + fn revalidating_guard_closes_when_l7_rules_of_its_endpoint_change() { + let mut proto = test_proto(); + policy_mut(&mut proto, "claude_code").endpoints[0] = + rest_endpoint("api.anthropic.com", "/v1/messages"); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let input = claude_input("api.anthropic.com"); + let guard = revalidating_guard(&engine, &input); + + // Every reload restamps policy_hash on every endpoint; that alone + // must not count as a change. + policy_mut(&mut proto, "gitlab") + .endpoints + .push(proto_endpoint("registry.gitlab.com")); + engine.reload_from_proto(&proto).unwrap(); + assert!(!guard.is_stale(), "policy_hash restamping is not a contract change"); + + policy_mut(&mut proto, "claude_code").endpoints[0] = + rest_endpoint("api.anthropic.com", "/v1/**"); + engine.reload_from_proto(&proto).unwrap(); + assert!(guard.is_stale(), "per-request rules for this endpoint changed"); + } + + #[test] + fn revalidating_guard_closes_when_its_binary_loses_access() { + let mut proto = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let guard = revalidating_guard(&engine, &claude_input("api.anthropic.com")); + + policy_mut(&mut proto, "claude_code").binaries = vec![NetworkBinary { + path: "/usr/bin/other".to_string(), + }]; + engine.reload_from_proto(&proto).unwrap(); + + assert!(guard.is_stale()); + } + + #[test] + fn revalidating_guard_closes_when_a_new_policy_matches_its_connection() { + let mut proto = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let guard = revalidating_guard(&engine, &claude_input("api.anthropic.com")); + + // A second policy admitting the same binary and host changes the set + // of endpoints per-request rules consult, even if it only widens. + proto.network_policies.insert( + "another".to_string(), + NetworkPolicyRule { + name: "another".to_string(), + endpoints: vec![proto_endpoint("api.anthropic.com")], + binaries: vec![NetworkBinary { + path: "/usr/local/bin/claude".to_string(), + }], + }, + ); + engine.reload_from_proto(&proto).unwrap(); + + assert!(guard.is_stale()); + } + + #[test] + fn revalidating_guard_never_follows_a_fail_closed_quarantine() { + let proto = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let guard = revalidating_guard(&engine, &claude_input("api.anthropic.com")); + + engine.enter_fail_closed("candidate rejected").unwrap(); + engine.exit_fail_closed().unwrap(); + + assert!( + guard.is_stale(), + "quarantine is a hard stop even when the previous policy returns" + ); + } + + #[test] + fn revalidating_guard_clones_share_one_pin() { + let mut proto = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let guard = revalidating_guard(&engine, &claude_input("api.anthropic.com")); + let clone = guard.clone(); + + policy_mut(&mut proto, "gitlab") + .endpoints + .push(proto_endpoint("registry.gitlab.com")); + engine.reload_from_proto(&proto).unwrap(); + + assert!(!guard.is_stale()); + assert_eq!(clone.captured_generation(), engine.current_generation()); + } + + #[tokio::test] + async fn revalidating_guard_waits_through_unrelated_reloads() { + let mut proto = test_proto(); + let engine = Arc::new(OpaEngine::from_proto(&proto).unwrap()); + let guard = revalidating_guard(&engine, &claude_input("api.anthropic.com")); + let waiter = tokio::spawn({ + let guard = guard.clone(); + async move { guard.wait_until_stale().await } + }); + tokio::task::yield_now().await; + + policy_mut(&mut proto, "gitlab") + .endpoints + .push(proto_endpoint("registry.gitlab.com")); + engine.reload_from_proto(&proto).unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert!(!waiter.is_finished(), "an unrelated reload must not wake the relay closed"); + + policy_mut(&mut proto, "claude_code").binaries.clear(); + engine.reload_from_proto(&proto).unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(1), waiter) + .await + .expect("a relevant reload must end the wait") + .expect("waiter must not panic"); + } + + #[test] + fn connection_contract_excludes_policy_hash_metadata() { + let mut proto = test_proto(); + policy_mut(&mut proto, "claude_code").endpoints[0] = + rest_endpoint("api.anthropic.com", "/v1/messages"); + let before = OpaEngine::from_proto(&proto).unwrap(); + policy_mut(&mut proto, "gitlab") + .endpoints + .push(proto_endpoint("registry.gitlab.com")); + let after = OpaEngine::from_proto(&proto).unwrap(); + let input = claude_input("api.anthropic.com"); + + let before_configs = before.authorize_egress(&input).unwrap().endpoint_configs; + let after_configs = after.authorize_egress(&input).unwrap().endpoint_configs; + assert_ne!( + before_configs, after_configs, + "fixture must exercise the per-reload policy_hash stamp" + ); + assert_eq!( + before.connection_policy_contract(&input).unwrap().0, + after.connection_policy_contract(&input).unwrap().0 + ); + } + + #[test] + fn denied_connection_captures_no_revalidator() { + let engine = Arc::new(OpaEngine::from_proto(&test_proto()).unwrap()); + let input = claude_input("not-allowed.example.com"); + let authorization = engine.authorize_egress(&input).unwrap(); + assert!(matches!(authorization.action, NetworkAction::Deny { .. })); + + let revalidator = PolicyRevalidator::capture( + &engine, + input, + &AuthorizedConnection { + action: &authorization.action, + endpoint_configs: &authorization.endpoint_configs, + matched_endpoints: &authorization.matched_endpoints, + exact_declared_endpoint_host: authorization.exact_declared_endpoint_host, + }, + authorization.generation, + None, + ); + assert!(revalidator.is_none()); + } } diff --git a/crates/openshell-supervisor-network/src/opa/revalidation.rs b/crates/openshell-supervisor-network/src/opa/revalidation.rs new file mode 100644 index 0000000000..bc5ed86ae6 --- /dev/null +++ b/crates/openshell-supervisor-network/src/opa/revalidation.rs @@ -0,0 +1,302 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Re-check established connections when the policy generation advances. +//! +//! Every policy reload, middleware registry replacement, and fail-closed +//! transition advances one sandbox-wide generation. Relays pin the generation +//! that authorized them. Without a revalidator, a pinned guard becomes stale as +//! soon as the generation moves, so approving an endpoint for one host closes +//! every open connection in the sandbox. +//! +//! A [`PolicyRevalidator`] lets a pinned guard follow the generation instead. +//! When the generation moves, the guard re-evaluates the connection's original +//! policy input against the live engine. If the resulting +//! [`ConnectionPolicyContract`] equals the contract that authorized the +//! connection, the guard re-pins to the new generation and the relay keeps +//! running. Any difference, an evaluation error, or an intervening fail-closed +//! quarantine makes the guard stale permanently, and the relay closes exactly +//! as it did before. +//! +//! # What the contract must cover +//! +//! The contract must contain every policy input a relay consults after setup, +//! otherwise a connection could outlive a change that should have narrowed it: +//! +//! - The L4 decision and matched policy name. +//! - `endpoint_configs`, in order. The first entry selects the TLS mode and +//! `allowed_ips`; the list feeds L7 route selection. +//! - `matched_endpoints`: the complete endpoint objects of every policy whose +//! binaries admit the caller and whose endpoints match the host and port. +//! Per-request Rego rules (`allow_request`, `deny_request`, +//! `request_deny_reason`, `allowed_token_grant_owners`, and the protocol +//! helpers they call) read only these endpoints, so an unchanged list means +//! every per-request decision is unchanged. This is why a tunnel's cloned +//! evaluator remains valid after it re-pins. +//! - `exact_declared_endpoint_host`, which selects SSRF handling. +//! - The middleware chain selected for the host and, when it is non-empty, the +//! middleware registry epoch, because tunnels hold a clone of the runner. +//! +//! Endpoint objects carry `policy_hash`, the hash of the complete canonical +//! policy. It changes on every reload and only binds endpoint-status +//! observations to an inventory, so it is excluded from the comparison. A +//! re-pinned MCP tunnel therefore keeps its original hash, and its later +//! observations are rejected by the new inventory until the client reconnects. +//! That affects status reporting only, never authorization. +//! +//! # Provider credentials +//! +//! Relays resolve provider credentials from the live provider state for every +//! request and WebSocket message, under their own revision guard. What a later +//! request cannot undo is an upstream session that an earlier request already +//! authenticated, such as an upgraded WebSocket. A relay that can inject +//! credentials therefore pins the provider credential installation as well, +//! and closes when it changes, exactly as it did before revalidation. Raw TCP +//! and TLS passthrough relays never inject credentials and do not pin it. +//! The pin covers the whole installation, not just this endpoint's providers; +//! narrowing it is possible but needs per-endpoint credential fingerprints. + +use super::{MatchedEndpoint, NetworkAction, NetworkInput, OpaEngine}; +use openshell_core::provider_credentials::ProviderCredentialState; +use openshell_supervisor_middleware::ChainEntry; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Weak}; +use tracing::debug; + +/// Endpoint keys that are policy metadata rather than policy semantics. +/// +/// `policy_hash` hashes the complete canonical policy, so it differs after +/// every reload even when this endpoint did not change. +const NON_SEMANTIC_ENDPOINT_KEYS: &[&str] = &["policy_hash"]; + +/// Upper bound on back-to-back reloads a single staleness check will follow. +/// +/// Each round re-evaluates against the live engine. If reloads keep landing +/// while a check runs, the guard gives up and reports stale, which is the +/// pre-revalidation behavior. +const MAX_REVALIDATION_ROUNDS: usize = 4; + +/// Policy facts that decide how an established connection is relayed. +/// +/// Two contracts compare equal only when a connection authorized under one +/// would be authorized and relayed identically under the other. See the +/// module documentation for why each field is present. +#[derive(Debug, Clone, PartialEq)] +pub struct ConnectionPolicyContract { + action: NetworkAction, + endpoint_configs: Vec, + /// `(policy_name, endpoint)` pairs. The endpoint index is deliberately + /// omitted: inserting an unrelated endpoint earlier in the same policy + /// shifts indexes without changing what this connection may do. + matched_endpoints: Vec<(String, regorus::Value)>, + exact_declared_endpoint_host: bool, + middleware_chain: Vec, + /// Present only when the chain is non-empty; a registry swap cannot affect + /// a connection that runs no middleware. + middleware_registry_epoch: Option, +} + +/// The authorization facts an adapter captured for one connection. +pub struct AuthorizedConnection<'a> { + pub action: &'a NetworkAction, + pub endpoint_configs: &'a [regorus::Value], + pub matched_endpoints: &'a [MatchedEndpoint], + pub exact_declared_endpoint_host: bool, +} + +impl ConnectionPolicyContract { + pub(super) fn new( + authorized: &AuthorizedConnection<'_>, + middleware_chain: Vec, + middleware_registry_epoch: u64, + ) -> Self { + let middleware_registry_epoch = + (!middleware_chain.is_empty()).then_some(middleware_registry_epoch); + Self { + action: authorized.action.clone(), + endpoint_configs: authorized + .endpoint_configs + .iter() + .map(semantic_endpoint) + .collect(), + matched_endpoints: authorized + .matched_endpoints + .iter() + .map(|endpoint| { + ( + endpoint.policy_name.clone(), + semantic_endpoint(&endpoint.endpoint), + ) + }) + .collect(), + exact_declared_endpoint_host: authorized.exact_declared_endpoint_host, + middleware_chain, + middleware_registry_epoch, + } + } +} + +/// Copy an endpoint object without keys that change on every reload. +fn semantic_endpoint(value: ®orus::Value) -> regorus::Value { + let mut value = value.clone(); + let has_metadata = matches!( + &value, + regorus::Value::Object(fields) + if NON_SEMANTIC_ENDPOINT_KEYS + .iter() + .any(|key| fields.contains_key(®orus::Value::from(*key))) + ); + if has_metadata && let Ok(fields) = value.as_object_mut() { + for key in NON_SEMANTIC_ENDPOINT_KEYS { + fields.remove(®orus::Value::from(*key)); + } + } + value +} + +/// Re-evaluates one connection's policy contract after a generation change. +/// +/// Shared by every clone of the guards pinned for that connection. Once a +/// re-evaluation finds a different contract the revalidator is invalidated for +/// good, so restoring the earlier policy cannot revive a connection that was +/// already observed under a policy that excluded it. +#[derive(Debug)] +pub struct PolicyRevalidator { + engine: Weak, + input: NetworkInput, + baseline: ConnectionPolicyContract, + credentials: Option, + invalidated: AtomicBool, +} + +/// The provider credential installation a relay may have injected from. +#[derive(Debug)] +struct CredentialPin { + state: ProviderCredentialState, + revision: u64, + installation_id: String, +} + +impl CredentialPin { + fn capture(state: &ProviderCredentialState) -> Self { + let snapshot = state.snapshot(); + Self { + state: state.clone(), + revision: snapshot.revision, + installation_id: snapshot.installation_id.clone(), + } + } + + fn is_current(&self) -> bool { + let snapshot = self.state.snapshot(); + snapshot.revision == self.revision && snapshot.installation_id == self.installation_id + } +} + +impl PolicyRevalidator { + /// Capture the contract that authorized a connection at + /// `authorized_generation`. + /// + /// Pass the relay's provider credential state when the relay can inject + /// credentials; the connection then also closes when that installation + /// changes. See the module documentation. + /// + /// Returns `None` for denials, and when the engine has already moved past + /// `authorized_generation` (the caller's pin fails in that case anyway). A + /// guard without a revalidator keeps the strict close-on-change behavior. + pub fn capture( + engine: &Arc, + input: NetworkInput, + authorized: &AuthorizedConnection<'_>, + authorized_generation: u64, + credentials: Option<&ProviderCredentialState>, + ) -> Option> { + if !matches!(authorized.action, NetworkAction::Allow { .. }) { + return None; + } + let (middleware_chain, middleware_registry_epoch) = engine + .middleware_selection_at(&input, authorized_generation) + .ok() + .flatten()?; + Some(Arc::new(Self { + engine: Arc::downgrade(engine), + baseline: ConnectionPolicyContract::new( + authorized, + middleware_chain, + middleware_registry_epoch, + ), + input, + credentials: credentials.map(CredentialPin::capture), + invalidated: AtomicBool::new(false), + })) + } + + pub(super) fn is_invalidated(&self) -> bool { + self.invalidated.load(Ordering::Acquire) + } + + /// Re-evaluate against the live engine. + /// + /// Returns the generation at which the unchanged contract was evaluated, + /// or `None` (permanently) when the contract no longer holds. + fn revalidate(&self, captured_generation: u64) -> Option { + if self.is_invalidated() { + return None; + } + let outcome = self.evaluate(captured_generation); + if outcome.is_none() { + self.invalidated.store(true, Ordering::Release); + } + outcome + } + + fn evaluate(&self, captured_generation: u64) -> Option { + if !self + .credentials + .as_ref() + .is_none_or(CredentialPin::is_current) + { + return None; + } + let engine = self.engine.upgrade()?; + // A quarantine is a hard stop even if a later generation restores the + // previous policy before this connection noticed. Evaluation errors + // also fail closed. + if engine.fail_closed_after(captured_generation) { + return None; + } + let (contract, generation) = engine.connection_policy_contract(&self.input).ok()?; + (contract == self.baseline).then_some(generation) + } + + /// Advance a shared pin while the contract still holds. + /// + /// Returns `true` once `pin` equals the live generation. + pub(super) fn follow(&self, pin: &AtomicU64, live: &AtomicU64) -> bool { + for _ in 0..MAX_REVALIDATION_ROUNDS { + let captured = pin.load(Ordering::Acquire); + if captured == live.load(Ordering::Acquire) { + return !self.is_invalidated(); + } + let Some(generation) = self.revalidate(captured) else { + debug!( + host = %self.input.host, + port = self.input.port, + captured_generation = captured, + "Policy change affects established connection; closing" + ); + return false; + }; + // Clones of a guard share the pin; never move it backwards. + let pinned = pin.fetch_max(generation, Ordering::AcqRel).max(generation); + debug!( + host = %self.input.host, + port = self.input.port, + previous_generation = captured, + pinned_generation = pinned, + "Policy change does not affect established connection; re-pinned" + ); + } + pin.load(Ordering::Acquire) == live.load(Ordering::Acquire) && !self.is_invalidated() + } +} diff --git a/crates/openshell-supervisor-network/src/proxy.rs b/crates/openshell-supervisor-network/src/proxy.rs index 96c0b8f50a..4a5b6e2a4f 100644 --- a/crates/openshell-supervisor-network/src/proxy.rs +++ b/crates/openshell-supervisor-network/src/proxy.rs @@ -1288,9 +1288,7 @@ async fn handle_transparent_tcp_connection( // generation before correlating endpoint identity or constructing a // connector. This prevents combining an old DNS answer with a newer // policy decision (or vice versa). - let Ok(generation_guard) = - relay::pin_policy_generation(&opa_engine, decision.policy_generation) - else { + let Ok(generation_guard) = relay::pin_policy_generation(&opa_engine, &decision, None) else { emit_transparent_mapping_denial( workload_addr, original, @@ -2929,21 +2927,17 @@ async fn handle_mediated_connection( return Ok(()); } - let connect_generation_guard = - match relay::pin_policy_generation(&opa_engine, decision.policy_generation) { - Ok(guard) => guard, - Err(error) => { - reject_stale_connect_policy( - &mut client, - &host_lc, - port, - activity_tx.as_ref(), - error, - ) + // Setup only: the relay chosen below pins its own guard, and pins provider + // credentials if it can inject them. + let connect_pin = relay::pin_policy_generation(&opa_engine, &decision, None); + let connect_generation_guard = match connect_pin { + Ok(guard) => guard, + Err(error) => { + reject_stale_connect_policy(&mut client, &host_lc, port, activity_tx.as_ref(), error) .await?; - return Ok(()); - } - }; + return Ok(()); + } + }; // Resolve the route's TLS treatment up front. `query_tls_mode` reads only // the policy decision + host/port (no peeked bytes), so it is valid before @@ -5763,10 +5757,9 @@ async fn handle_forward_proxy( "Forward proxy L4 policy decision" ); let sandbox_entrypoint_pid = entrypoint_pid.load(Ordering::Acquire); - let forward_generation_guard = match relay::pin_policy_generation( - &opa_engine, - decision.policy_generation, - ) { + let forward_credentials = provider_credentials.as_ref(); + let forward_pin = relay::pin_policy_generation(&opa_engine, &decision, forward_credentials); + let forward_generation_guard = match forward_pin { Ok(guard) => guard, Err(e) => { warn!( @@ -5909,7 +5902,13 @@ async fn handle_forward_proxy( .await?; return Ok(()); } - let tunnel_engine = match relay::pin_l7_evaluator(&opa_engine, route.l7_policy_generation) { + let tunnel_pin = relay::pin_l7_evaluator( + &opa_engine, + &decision, + route.l7_policy_generation, + l7_ctx.provider_credentials.as_ref(), + ); + let tunnel_engine = match tunnel_pin { Ok(engine) => engine, Err(e) => { warn!( diff --git a/crates/openshell-supervisor-network/src/proxy/egress.rs b/crates/openshell-supervisor-network/src/proxy/egress.rs index 77b89a9b7e..78b4976f47 100644 --- a/crates/openshell-supervisor-network/src/proxy/egress.rs +++ b/crates/openshell-supervisor-network/src/proxy/egress.rs @@ -170,6 +170,25 @@ pub(super) struct EgressDecision { pub(super) cmdline_paths: Vec, } +impl EgressDecision { + /// The policy input this decision was evaluated with. + /// + /// Every adapter builds its OPA input from the intent destination and the + /// identity fields stored here (an absent binary is the empty path used by + /// endpoint-only mode). The binary digest is omitted because Rego never + /// reads it; integrity is enforced in Rust before authorization. + pub(super) fn policy_input(&self) -> crate::opa::NetworkInput { + crate::opa::NetworkInput { + host: self.intent.destination.host.clone(), + port: self.intent.destination.port, + binary_path: self.binary.clone().unwrap_or_default(), + binary_sha256: String::new(), + ancestors: self.ancestors.clone(), + cmdline_paths: self.cmdline_paths.clone(), + } + } +} + pub(super) fn connect_span(intent: &EgressIntent) -> tracing::Span { tracing::debug_span!( "supervisor.egress.connect", diff --git a/crates/openshell-supervisor-network/src/proxy/relay.rs b/crates/openshell-supervisor-network/src/proxy/relay.rs index 495c22241b..30f30c333a 100644 --- a/crates/openshell-supervisor-network/src/proxy/relay.rs +++ b/crates/openshell-supervisor-network/src/proxy/relay.rs @@ -5,10 +5,14 @@ use super::{EgressDecision, L7RouteSnapshot, emit_l7_tunnel_close_after_policy_change}; use crate::l7::relay::L7EvalContext; -use crate::opa::{NetworkAction, OpaEngine, PolicyGenerationGuard, TunnelPolicyEngine}; +use crate::opa::{ + AuthorizedConnection, NetworkAction, OpaEngine, PolicyGenerationGuard, PolicyRevalidator, + TunnelPolicyEngine, +}; use miette::{IntoDiagnostic, Result}; use openshell_core::activity::ActivitySender; use openshell_core::endpoint_status::EndpointObservationSender; +use openshell_core::provider_credentials::ProviderCredentialState; use openshell_core::secrets::SecretResolver; use std::sync::Arc; use tokio::io::{AsyncRead, AsyncWrite}; @@ -45,7 +49,7 @@ pub(super) struct RelaySignals { /// Build the request-processing context shared by CONNECT and forward HTTP. pub(super) fn http_context( decision: &EgressDecision, - provider_credentials: Option, + provider_credentials: Option, secret_resolver: Option>, policy_local: Option<&crate::policy_local::PolicyLocalContext>, agent_proposals: openshell_core::proposals::AgentProposals, @@ -104,20 +108,60 @@ pub(super) fn http_context( } } -/// Pin a generation for a relay or the forward HTTP single-request path. +/// Capture the contract that authorized `decision` so its relay can re-pin +/// through policy changes that do not affect it. +/// +/// Returns `None` for denials and when the policy already moved on; the guard +/// then closes on any change, which was the only behavior before +/// revalidation. Pass `credentials` for relays that can inject provider +/// credentials so a credential change still closes them. See +/// [`crate::opa::PolicyRevalidator`]. +fn connection_revalidator( + opa_engine: &Arc, + decision: &EgressDecision, + credentials: Option<&ProviderCredentialState>, +) -> Option> { + PolicyRevalidator::capture( + opa_engine, + decision.policy_input(), + &AuthorizedConnection { + action: &decision.action, + endpoint_configs: &decision.endpoint.policy_configs, + matched_endpoints: &decision.endpoint.matched_endpoints, + exact_declared_endpoint_host: decision.endpoint.exact_declared_host, + }, + decision.policy_generation, + credentials, + ) +} + +/// Pin the generation that authorized `decision` for a relay, CONNECT setup, +/// or the forward HTTP single-request path. +/// +/// The pin fails if the policy changed after authorization. Once pinned, the +/// guard follows later changes that leave the connection's contract intact. +/// `credentials` is the provider state the guarded path may inject from, or +/// `None` when it never injects credentials. pub(super) fn pin_policy_generation( - opa_engine: &OpaEngine, - expected_generation: u64, + opa_engine: &Arc, + decision: &EgressDecision, + credentials: Option<&ProviderCredentialState>, ) -> Result { - opa_engine.generation_guard(expected_generation) + let guard = opa_engine.generation_guard(decision.policy_generation)?; + let revalidator = connection_revalidator(opa_engine, decision, credentials); + Ok(guard.with_revalidator(revalidator)) } /// Clone an L7 evaluator for a relay or the forward HTTP single-request path. pub(super) fn pin_l7_evaluator( - opa_engine: &OpaEngine, + opa_engine: &Arc, + decision: &EgressDecision, expected_generation: u64, + credentials: Option<&ProviderCredentialState>, ) -> Result { - opa_engine.clone_engine_for_tunnel(expected_generation) + let evaluator = opa_engine.clone_engine_for_tunnel(expected_generation)?; + let revalidator = connection_revalidator(opa_engine, decision, credentials); + Ok(evaluator.with_revalidator(revalidator)) } pub(super) fn validate_route_generation( @@ -144,7 +188,7 @@ pub(super) fn validate_route_generation( /// attempting to write an HTTP response into it. pub(super) fn prepare_http_relay<'a>( route: Option<&L7RouteSnapshot>, - opa_engine: &'a OpaEngine, + opa_engine: &'a Arc, decision: &EgressDecision, request: &'a L7EvalContext, ) -> Option> { @@ -157,8 +201,11 @@ pub(super) fn prepare_http_relay<'a>( return None; } + // HTTP relays can inject provider credentials, so they also pin them. + let credentials = request.provider_credentials.as_ref(); let policy = if let Some(route) = route.filter(|route| !route.configs.is_empty()) { - let evaluator = match pin_l7_evaluator(opa_engine, decision.policy_generation) { + let generation = decision.policy_generation; + let evaluator = match pin_l7_evaluator(opa_engine, decision, generation, credentials) { Ok(evaluator) => evaluator, Err(error) => { emit_l7_tunnel_close_after_policy_change( @@ -179,7 +226,7 @@ pub(super) fn prepare_http_relay<'a>( evaluator: Box::new(evaluator), } } else { - let generation_guard = match pin_policy_generation(opa_engine, decision.policy_generation) { + let generation_guard = match pin_policy_generation(opa_engine, decision, credentials) { Ok(guard) => guard, Err(error) => { emit_l7_tunnel_close_after_policy_change( @@ -196,16 +243,18 @@ pub(super) fn prepare_http_relay<'a>( Some(RelayContext { request, policy, - middleware_engine: opa_engine, + middleware_engine: opa_engine.as_ref(), }) } -/// Pin the generation used by a raw relay so policy activation or quarantine -/// closes streams that otherwise have no request boundary at which to notice -/// a stale decision. +/// Pin the generation used by a raw relay so a policy change that affects the +/// connection, or any quarantine, closes streams that otherwise have no +/// request boundary at which to notice a stale decision. +/// +/// Raw relays never inject provider credentials, so they do not pin them. pub(super) fn prepare_raw_relay( route: Option<&L7RouteSnapshot>, - opa_engine: &OpaEngine, + opa_engine: &Arc, decision: &EgressDecision, ) -> Option { if let Err(error) = validate_route_generation(route, decision.policy_generation) { @@ -217,7 +266,7 @@ pub(super) fn prepare_raw_relay( return None; } - match pin_policy_generation(opa_engine, decision.policy_generation) { + match pin_policy_generation(opa_engine, decision, None) { Ok(guard) => Some(guard), Err(error) => { emit_l7_tunnel_close_after_policy_change( @@ -412,7 +461,7 @@ mod tests { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let (mut upstream, mut server) = tokio::io::duplex(1024); - let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); let decision = decision(engine.current_generation()); let request = request_context(); let context = prepare_http_relay(None, &engine, &decision, &request).unwrap(); @@ -501,7 +550,7 @@ mod tests { #[test] fn relay_without_route_pins_l4_decision_generation() { - let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); let decision = decision(engine.current_generation()); let request = request_context(); @@ -519,7 +568,7 @@ mod tests { #[test] fn empty_hydrated_route_cannot_replace_stale_l4_generation() { - let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); let decision = decision(u64::MAX); let route = L7RouteSnapshot { configs: vec![], @@ -535,7 +584,7 @@ mod tests { #[test] fn inspected_route_cannot_replace_stale_l4_generation() { - let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); let decision = decision(u64::MAX); let route = L7RouteSnapshot { configs: vec![super::super::L7ConfigSnapshot { @@ -573,7 +622,7 @@ mod tests { #[test] fn raw_route_cannot_replace_stale_l4_generation() { - let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); let decision = decision(u64::MAX); let route = L7RouteSnapshot { configs: vec![], @@ -588,7 +637,7 @@ mod tests { #[test] fn stale_generation_fails_before_relay_context_is_created() { - let engine = OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, EMPTY_POLICY_DATA).unwrap()); let decision = decision(engine.current_generation()); let request = request_context(); engine.reload(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); @@ -624,4 +673,137 @@ mod tests { .expect("relay task should not panic") .expect("stale relay closure should be clean"); } + + const CURL_EXAMPLE_POLICY: &str = r" +network_policies: + web: + name: web + endpoints: + - { host: example.com, port: 80 } + binaries: + - { path: /usr/bin/curl } +"; + + const CURL_EXAMPLE_AND_OTHER_POLICY: &str = r" +network_policies: + web: + name: web + endpoints: + - { host: example.com, port: 80 } + binaries: + - { path: /usr/bin/curl } + other: + name: other + endpoints: + - { host: other.example.net, port: 443 } + binaries: + - { path: /usr/bin/curl } +"; + + /// Authorize `example.com:80` for curl the way the CONNECT adapter does. + fn authorized_decision(engine: &OpaEngine) -> EgressDecision { + let intent = EgressIntent::connect("example.com".to_string(), 80); + let mut decision = EgressDecision { + intent, + action: NetworkAction::Deny { + reason: String::new(), + }, + policy_generation: 0, + identity: ProcessIdentityEvidence::Available, + endpoint: EndpointDecision::default(), + binary: Some(std::path::PathBuf::from("/usr/bin/curl")), + binary_pid: None, + ancestors: vec![], + cmdline_paths: vec![], + }; + let authorization = engine.authorize_egress(&decision.policy_input()).unwrap(); + decision.action = authorization.action.clone(); + decision.policy_generation = authorization.generation; + decision.endpoint = EndpointDecision::from_authorization(&authorization); + assert!(matches!(decision.action, NetworkAction::Allow { .. })); + decision + } + + #[tokio::test] + async fn raw_relay_survives_reload_that_does_not_affect_it_and_closes_on_one_that_does() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, CURL_EXAMPLE_POLICY).unwrap()); + let decision = authorized_decision(&engine); + let guard = prepare_raw_relay(None, &engine, &decision).expect("current decision pins"); + let request = request_context(); + let (mut client_peer, mut proxy_client) = tokio::io::duplex(64); + let (mut upstream_peer, mut proxy_upstream) = tokio::io::duplex(64); + let relay = tokio::spawn(async move { + relay_tcp(&mut proxy_client, &mut proxy_upstream, &guard, &request).await + }); + + // Approve an endpoint for a different host. + engine + .reload(POLICY_REGO, CURL_EXAMPLE_AND_OTHER_POLICY) + .unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert!(!relay.is_finished(), "unrelated approval must not close the relay"); + + client_peer.write_all(b"ping").await.unwrap(); + let mut received = [0u8; 4]; + tokio::time::timeout( + std::time::Duration::from_secs(1), + upstream_peer.read_exact(&mut received), + ) + .await + .expect("bytes must keep flowing after the reload") + .unwrap(); + assert_eq!(&received, b"ping"); + + // Revoke the endpoint this connection uses. + engine.reload(POLICY_REGO, EMPTY_POLICY_DATA).unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(1), relay) + .await + .expect("revoking this endpoint must close the relay") + .expect("relay task should not panic") + .expect("stale relay closure should be clean"); + } + + #[test] + fn http_relay_survives_unrelated_reload_but_not_a_credential_change() { + use std::collections::HashMap; + + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, CURL_EXAMPLE_POLICY).unwrap()); + let decision = authorized_decision(&engine); + let credentials = ProviderCredentialState::from_child_env_snapshot(1, HashMap::new()); + let mut request = request_context(); + request.provider_credentials = Some(credentials.clone()); + let context = prepare_http_relay(None, &engine, &decision, &request).unwrap(); + let PreparedHttpPolicy::Passthrough { generation_guard } = context.policy else { + panic!("route-less relay should use a generation guard"); + }; + + engine + .reload(POLICY_REGO, CURL_EXAMPLE_AND_OTHER_POLICY) + .unwrap(); + assert!(!generation_guard.is_stale(), "unchanged credentials and contract"); + + // Provider credentials change together with a runtime reload. + credentials.install_child_env_snapshot(2, HashMap::new()); + engine.reload(POLICY_REGO, CURL_EXAMPLE_POLICY).unwrap(); + assert!( + generation_guard.is_stale(), + "an upstream session may hold the previous credential" + ); + } + + #[test] + fn reload_between_authorization_and_pin_still_refuses_the_relay() { + let engine = Arc::new(OpaEngine::from_strings(POLICY_REGO, CURL_EXAMPLE_POLICY).unwrap()); + let decision = authorized_decision(&engine); + engine + .reload(POLICY_REGO, CURL_EXAMPLE_AND_OTHER_POLICY) + .unwrap(); + + assert!( + prepare_raw_relay(None, &engine, &decision).is_none(), + "revalidation applies to established relays, not to setup" + ); + } } diff --git a/docs/how-it-works/policies/manage-policies.mdx b/docs/how-it-works/policies/manage-policies.mdx index c20af7b67b..52ef30e26f 100644 --- a/docs/how-it-works/policies/manage-policies.mdx +++ b/docs/how-it-works/policies/manage-policies.mdx @@ -188,17 +188,31 @@ running sandbox. The startup sections do not: | Change | Effect on a running sandbox | Required action | |---|---|---| -| Network rules | The sandbox loads the new rules. Connections opened under the previous rules close. | Apply the change, then retry the request. | +| Network rules | The sandbox loads the new rules. Open connections whose rules changed close; other connections stay open. | Apply the change, then retry the request. | | Middleware in `network_middlewares` | You can add, remove, or reconfigure built-in middleware and external services already registered with the gateway. | Apply the change with `openshell policy set`. | | External middleware registration | Policy changes cannot register a service or change its gateway connection settings. | Update the gateway configuration and restart the gateway. | | Added filesystem paths | The saved policy can change, but the running workload keeps its existing filesystem permissions. | Recreate the sandbox. | | Removed filesystem paths, or changed `include_workdir`, Landlock, or process settings | OpenShell rejects the change after the workload starts. | Recreate the sandbox. | -When new network rules take effect, OpenShell closes connections that were -opened under the previous rules, including HTTP keep-alive connections, tunnels, -WebSocket connections, and long-lived streams. Clients must reconnect, and their +When new network rules take effect, OpenShell re-checks every open connection, +including HTTP keep-alive connections, tunnels, WebSocket connections, and +long-lived streams, against the new rules. A connection stays open only if the +new rules treat it exactly as the old ones did: the same allow decision, the +same matching policies and endpoints (including their L7 rules), and the same +middleware. Otherwise OpenShell closes it, the client must reconnect, and its next requests are checked against the new rules. +For example, approving an endpoint for one host leaves connections to other +hosts open, while changing the rules for a host closes that host's +connections. Some changes always close connections: + +- A fail-closed quarantine after an invalid policy, even if the previous policy + is restored before the connection is checked. +- A change to the middleware services registered with the gateway, for + connections that run middleware. +- A change to the sandbox's provider credentials, for HTTP, WebSocket, and + other inspected connections that can carry injected credentials. + OpenShell validates every change twice before it takes effect: ```mermaid @@ -207,7 +221,7 @@ flowchart TD B -->|Invalid| C["Change rejected
Current policy stays active"] B -->|Valid| D["Gateway saves a new revision"] D --> E{"Sandbox validates
and loads the revision"} - E -->|Loads| F["New rules active
Old connections close"] + E -->|Loads| F["New rules active
Affected connections close"] E -->|Fails| G["Failure mode applies
Block traffic or keep the last valid policy"] ``` diff --git a/docs/how-it-works/policies/network-rules.mdx b/docs/how-it-works/policies/network-rules.mdx index d7dec792a6..131efcecae 100644 --- a/docs/how-it-works/policies/network-rules.mdx +++ b/docs/how-it-works/policies/network-rules.mdx @@ -452,9 +452,12 @@ Set `allow_uninspected_credentials: true` to relay binary frames anyway. Set placeholders that must be resolved. This explicitly authorizes WebSocket text rewriting for ordinary static credentials, including credentials with HTTP placement metadata. -When new network rules take effect, OpenShell closes open WebSocket connections, -so the client must reconnect. The client might receive close code `1012`, or the -connection might close without a close frame. +When new network rules change how a WebSocket connection is treated, OpenShell +closes it, so the client must reconnect. The client might receive close code +`1012`, or the connection might close without a close frame. Connections the +change does not affect stay open; see +[How changes take effect](/how-it-works/policies/manage-policies#how-changes-take-effect) +for which changes close a connection. ### Allow GraphQL Operations