From a3f729c3cd43d2414aa11d67d0d9fa912638c027 Mon Sep 17 00:00:00 2001 From: Lewis Date: Wed, 10 Jun 2026 10:03:59 +0300 Subject: [PATCH] ripple: fail-closed startup, bind policy Lewis: May this revision serve well! --- Cargo.lock | 1 + crates/tranquil-cache/Cargo.toml | 1 + crates/tranquil-cache/src/lib.rs | 46 ++++------ crates/tranquil-config/src/lib.rs | 88 +++++++++++++++++- crates/tranquil-pds/src/cache/mod.rs | 4 +- crates/tranquil-pds/src/state.rs | 4 +- crates/tranquil-pds/tests/common/mod.rs | 2 + crates/tranquil-ripple/src/config.rs | 91 ++++++++++++++++++- crates/tranquil-ripple/src/engine.rs | 16 +++- crates/tranquil-ripple/src/lib.rs | 2 +- crates/tranquil-ripple/tests/bind_policy.rs | 70 ++++++++++++++ .../tests/two_node_convergence.rs | 8 ++ example.toml | 18 +++- 13 files changed, 311 insertions(+), 40 deletions(-) create mode 100644 crates/tranquil-ripple/tests/bind_policy.rs diff --git a/Cargo.lock b/Cargo.lock index d4a02ec..eaa9b51 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7709,6 +7709,7 @@ dependencies = [ "async-trait", "base64 0.22.1", "redis", + "thiserror 2.0.18", "tokio-util", "tracing", "tranquil-config", diff --git a/crates/tranquil-cache/Cargo.toml b/crates/tranquil-cache/Cargo.toml index f6f6580..e06e3b2 100644 --- a/crates/tranquil-cache/Cargo.toml +++ b/crates/tranquil-cache/Cargo.toml @@ -16,5 +16,6 @@ tranquil-ripple = { workspace = true } async-trait = { workspace = true } base64 = { workspace = true } redis = { workspace = true, optional = true } +thiserror = { workspace = true } tokio-util = { workspace = true } tracing = { workspace = true } diff --git a/crates/tranquil-cache/src/lib.rs b/crates/tranquil-cache/src/lib.rs index d0ea4b1..2551cf8 100644 --- a/crates/tranquil-cache/src/lib.rs +++ b/crates/tranquil-cache/src/lib.rs @@ -160,18 +160,17 @@ impl Cache for NoOpCache { } } -pub struct NoOpRateLimiter; - -#[async_trait] -impl DistributedRateLimiter for NoOpRateLimiter { - async fn check_rate_limit(&self, _key: &str, _limit: u32, _window_ms: u64) -> bool { - true - } +#[derive(Debug, thiserror::Error)] +pub enum CacheInitError { + #[error("ripple config: {0}")] + Config(#[from] tranquil_ripple::RippleConfigError), + #[error("ripple start: {0}")] + Start(#[from] tranquil_ripple::RippleStartError), } pub async fn create_cache( shutdown: tokio_util::sync::CancellationToken, -) -> (Arc, Arc) { +) -> Result<(Arc, Arc), CacheInitError> { let cache_cfg = tranquil_config::try_get().map(|c| &c.cache); let backend = cache_cfg.map(|c| c.backend.as_str()).unwrap_or("ripple"); let valkey_url = cache_cfg.and_then(|c| c.valkey_url.as_deref()); @@ -183,7 +182,7 @@ pub async fn create_cache( Ok(cache) => { tracing::info!("using valkey cache at {url}"); let rate_limiter = Arc::new(RedisRateLimiter::new(cache.connection())); - return (Arc::new(cache), rate_limiter); + return Ok((Arc::new(cache), rate_limiter)); } Err(e) => { tracing::warn!("failed to connect to valkey: {e}. falling back to ripple."); @@ -201,26 +200,13 @@ pub async fn create_cache( ); } - match tranquil_ripple::RippleConfig::from_config() { - Ok(config) => { - let peer_count = config.seed_peers.len(); - match tranquil_ripple::RippleEngine::start(config, shutdown).await { - Ok((cache, rate_limiter, _bound_addr)) => { - match peer_count { - 0 => tracing::info!("ripple cache started (single-node)"), - n => tracing::info!("ripple cache started ({n} seed peers)"), - } - (cache, rate_limiter) - } - Err(e) => { - tracing::error!("ripple engine failed to start: {e:#}. running without cache."); - (Arc::new(NoOpCache), Arc::new(NoOpRateLimiter)) - } - } - } - Err(e) => { - tracing::error!("ripple config error: {e:#}. running without cache."); - (Arc::new(NoOpCache), Arc::new(NoOpRateLimiter)) - } + let config = tranquil_ripple::RippleConfig::from_config()?; + let peer_count = config.seed_peers.len(); + let (cache, rate_limiter, _bound_addr) = + tranquil_ripple::RippleEngine::start(config, shutdown).await?; + match peer_count { + 0 => tracing::info!("ripple cache started as a single node"), + n => tracing::info!("ripple cache started with {n} seed peers"), } + Ok((cache, rate_limiter)) } diff --git a/crates/tranquil-config/src/lib.rs b/crates/tranquil-config/src/lib.rs index c7cdd0e..423e584 100644 --- a/crates/tranquil-config/src/lib.rs +++ b/crates/tranquil-config/src/lib.rs @@ -259,6 +259,9 @@ impl TranquilConfig { // -- tls -------------------------------------------------------------- self.server.tls.validate(&mut errors); + // -- cache ------------------------------------------------------------ + self.cache.validate(&mut errors); + // -- SSO providers ---------------------------------------------------- self.validate_sso_provider("sso.github", &self.sso.github, &mut errors); self.validate_sso_provider("sso.google", &self.sso.google, &mut errors); @@ -786,6 +789,31 @@ pub struct CacheConfig { pub ripple: RippleCacheConfig, } +impl CacheConfig { + pub fn validate(&self, errors: &mut Vec) { + let clustered = self + .ripple + .peers + .as_deref() + .unwrap_or(&[]) + .iter() + .any(|p| !p.trim().is_empty()); + let keyed = self + .ripple + .cluster_key + .as_deref() + .is_some_and(|k| !k.trim().is_empty()); + if self.backend == "ripple" && clustered && !keyed && !self.ripple.allow_insecure { + errors.push( + "cache.ripple.peers (RIPPLE_PEERS) is set without cache.ripple.cluster_key \ + (RIPPLE_CLUSTER_KEY); set the cluster key to authenticate peers, or set \ + cache.ripple.allow_insecure (RIPPLE_ALLOW_INSECURE) for a trusted private network" + .to_string(), + ); + } + } +} + #[derive(Debug, Config)] #[config(layer_attr(serde(deny_unknown_fields)))] pub struct PlcConfig { @@ -1444,7 +1472,9 @@ fn split_comma_list(value: &str) -> Result, std::convert::Infallible #[derive(Debug, Config)] #[config(layer_attr(serde(deny_unknown_fields)))] pub struct RippleCacheConfig { - /// Address to bind the Ripple gossip protocol listener. + /// Address to bind the Ripple gossip protocol listener. With the default + /// value and no cluster_key or peers configured, the listener binds + /// loopback instead and runs as a single node. #[config(env = "RIPPLE_BIND", default = "0.0.0.0:0")] pub bind_addr: String, @@ -1463,6 +1493,16 @@ pub struct RippleCacheConfig { /// Maximum cache size in megabytes. #[config(env = "RIPPLE_CACHE_MAX_MB", default = 256)] pub cache_max_mb: usize, + + /// Pre-shared cluster key authenticating ripple peers. Every node in the + /// cluster must set the same value. When unset, peers are unauthenticated. + #[config(env = "RIPPLE_CLUSTER_KEY")] + pub cluster_key: Option, + + /// Allow ripple to bind a non-loopback address without a cluster key. + /// Peers will be unauthenticated. Intended for trusted private networks. + #[config(env = "RIPPLE_ALLOW_INSECURE", default = false)] + pub allow_insecure: bool, } #[derive(Debug, Config)] @@ -1859,6 +1899,52 @@ port = 587 ); } + fn cache_config_for_test( + peers: Option>, + cluster_key: Option<&str>, + allow_insecure: bool, + ) -> CacheConfig { + CacheConfig { + backend: "ripple".to_string(), + valkey_url: None, + ripple: RippleCacheConfig { + bind_addr: "0.0.0.0:0".to_string(), + peers, + machine_id: None, + gossip_interval_ms: 200, + cache_max_mb: 256, + cluster_key: cluster_key.map(str::to_string), + allow_insecure, + }, + } + } + + #[test] + fn cache_validate_rejects_clustered_keyless_ripple() { + let mut errors = Vec::new(); + cache_config_for_test(Some(vec!["10.0.0.7:7000".to_string()]), None, false) + .validate(&mut errors); + assert!( + errors.iter().any(|e| e.contains("RIPPLE_CLUSTER_KEY")), + "expected cluster key error, got {errors:?}" + ); + } + + #[test] + fn cache_validate_accepts_keyed_insecure_or_standalone() { + let mut errors = Vec::new(); + cache_config_for_test( + Some(vec!["10.0.0.7:7000".to_string()]), + Some("nautilus-secret"), + false, + ) + .validate(&mut errors); + cache_config_for_test(Some(vec!["10.0.0.7:7000".to_string()]), None, true) + .validate(&mut errors); + cache_config_for_test(None, None, false).validate(&mut errors); + assert!(errors.is_empty(), "expected no errors, got {errors:?}"); + } + #[derive(Default)] struct EmailOverrides { from_address: Option<&'static str>, diff --git a/crates/tranquil-pds/src/cache/mod.rs b/crates/tranquil-pds/src/cache/mod.rs index c48865b..52ddb33 100644 --- a/crates/tranquil-pds/src/cache/mod.rs +++ b/crates/tranquil-pds/src/cache/mod.rs @@ -1,6 +1,4 @@ -pub use tranquil_cache::{ - Cache, CacheError, DistributedRateLimiter, NoOpCache, NoOpRateLimiter, create_cache, -}; +pub use tranquil_cache::{Cache, CacheError, DistributedRateLimiter, NoOpCache, create_cache}; #[cfg(feature = "valkey")] pub use tranquil_cache::{RedisRateLimiter, ValkeyCache}; diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index 53a7047..1927d5c 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -359,7 +359,9 @@ impl AppState { let rate_limiters = Arc::new(RateLimiters::new()); let repo_write_locks = Arc::new(RepoWriteLocks::new()); let circuit_breakers = Arc::new(CircuitBreakers::new()); - let (cache, distributed_rate_limiter) = create_cache(shutdown.clone()).await; + let (cache, distributed_rate_limiter) = create_cache(shutdown.clone()) + .await + .expect("Failed to initialize cache and distributed rate limiter at startup"); let did_resolver = Arc::new(DidResolver::new()); let cross_pds_oauth = Arc::new(CrossPdsOAuthClient::new(cache.clone())); let sso_config = SsoConfig::init(); diff --git a/crates/tranquil-pds/tests/common/mod.rs b/crates/tranquil-pds/tests/common/mod.rs index 1ca94ff..813e359 100644 --- a/crates/tranquil-pds/tests/common/mod.rs +++ b/crates/tranquil-pds/tests/common/mod.rs @@ -692,6 +692,8 @@ pub async fn spawn_cluster(pool: Option, node_count: usize) -> Vec machine_id: i as u64 + 1, gossip_interval_ms: 100, cache_max_bytes: 64 * 1024 * 1024, + cluster_key: None, + allow_insecure: false, }; let (cache, rate_limiter, addr) = RippleEngine::start(config, shutdown.clone()) .await diff --git a/crates/tranquil-ripple/src/config.rs b/crates/tranquil-ripple/src/config.rs index 001948a..9dd7e5d 100644 --- a/crates/tranquil-ripple/src/config.rs +++ b/crates/tranquil-ripple/src/config.rs @@ -1,4 +1,4 @@ -use std::net::SocketAddr; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; pub(crate) fn fnv1a(data: &[u8]) -> u64 { data.iter().fold(0xcbf29ce484222325u64, |hash, &byte| { @@ -13,6 +13,8 @@ pub struct RippleConfig { pub machine_id: u64, pub gossip_interval_ms: u64, pub cache_max_bytes: usize, + pub cluster_key: Option, + pub allow_insecure: bool, } impl RippleConfig { @@ -53,18 +55,105 @@ impl RippleConfig { .saturating_mul(1024) .saturating_mul(1024); + let cluster_key = ripple + .cluster_key + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(str::to_string); + + let bind_addr = effective_bind_addr( + bind_addr, + cluster_key.is_some(), + ripple.allow_insecure, + !seed_peers.is_empty(), + ); + Ok(Self { bind_addr, seed_peers, machine_id, gossip_interval_ms, cache_max_bytes, + cluster_key, + allow_insecure: ripple.allow_insecure, }) } } +fn effective_bind_addr( + bind_addr: SocketAddr, + has_cluster_key: bool, + allow_insecure: bool, + has_peers: bool, +) -> SocketAddr { + let standalone_default = !has_cluster_key + && !allow_insecure + && !has_peers + && bind_addr.ip().is_unspecified() + && bind_addr.port() == 0; + match standalone_default { + false => bind_addr, + true => { + let loopback: IpAddr = match bind_addr.ip() { + IpAddr::V4(_) => Ipv4Addr::LOCALHOST.into(), + IpAddr::V6(_) => Ipv6Addr::LOCALHOST.into(), + }; + tracing::info!( + "ripple has no cluster key and no peers, binding loopback as a single node" + ); + SocketAddr::new(loopback, 0) + } + } +} + #[derive(Debug, thiserror::Error)] pub enum RippleConfigError { #[error("invalid address: {0}")] InvalidAddr(String), } + +#[cfg(test)] +mod tests { + use super::*; + + fn addr(s: &str) -> SocketAddr { + s.parse().unwrap() + } + + #[test] + fn standalone_default_bind_rewrites_to_loopback() { + assert_eq!( + effective_bind_addr(addr("0.0.0.0:0"), false, false, false), + addr("127.0.0.1:0") + ); + assert_eq!( + effective_bind_addr(addr("[::]:0"), false, false, false), + addr("[::1]:0") + ); + } + + #[test] + fn explicit_or_clustered_binds_are_untouched() { + assert_eq!( + effective_bind_addr(addr("0.0.0.0:7000"), false, false, false), + addr("0.0.0.0:7000") + ); + assert_eq!( + effective_bind_addr(addr("0.0.0.0:0"), true, false, false), + addr("0.0.0.0:0") + ); + assert_eq!( + effective_bind_addr(addr("0.0.0.0:0"), false, true, false), + addr("0.0.0.0:0") + ); + assert_eq!( + effective_bind_addr(addr("0.0.0.0:0"), false, false, true), + addr("0.0.0.0:0") + ); + assert_eq!( + effective_bind_addr(addr("192.0.2.7:0"), false, false, false), + addr("192.0.2.7:0") + ); + } +} diff --git a/crates/tranquil-ripple/src/engine.rs b/crates/tranquil-ripple/src/engine.rs index 334bfe8..0c36af4 100644 --- a/crates/tranquil-ripple/src/engine.rs +++ b/crates/tranquil-ripple/src/engine.rs @@ -5,7 +5,7 @@ use crate::eviction::MemoryBudget; use crate::gossip::{GossipEngine, PeerId}; use crate::metrics; use crate::rate_limiter::RippleRateLimiter; -use crate::transport::Transport; +use crate::transport::{ClusterKey, Transport}; use std::net::SocketAddr; use std::sync::Arc; use tokio_util::sync::CancellationToken; @@ -21,8 +21,20 @@ impl RippleEngine { { let store = Arc::new(ShardedCrdtStore::new(config.machine_id)); + let cluster_key = match config.cluster_key.as_deref() { + Some(secret) => Some(ClusterKey::new(secret)), + None => { + if !config.bind_addr.ip().is_loopback() && !config.allow_insecure { + return Err(RippleStartError::Config(format!( + "ripple is bound to non-loopback {} without RIPPLE_CLUSTER_KEY. Set the cluster key to authenticate peers, or set RIPPLE_ALLOW_INSECURE=true to bind unauthenticated on a trusted network", + config.bind_addr + ))); + } + None + } + }; let (transport, incoming_rx) = - Transport::bind(config.bind_addr, None, shutdown.clone()) + Transport::bind(config.bind_addr, cluster_key, shutdown.clone()) .await .map_err(|e| RippleStartError::Bind(e.to_string()))?; diff --git a/crates/tranquil-ripple/src/lib.rs b/crates/tranquil-ripple/src/lib.rs index 7cfecb8..bd71258 100644 --- a/crates/tranquil-ripple/src/lib.rs +++ b/crates/tranquil-ripple/src/lib.rs @@ -8,5 +8,5 @@ pub mod metrics; pub mod rate_limiter; pub mod transport; -pub use config::RippleConfig; +pub use config::{RippleConfig, RippleConfigError}; pub use engine::{RippleEngine, RippleStartError}; diff --git a/crates/tranquil-ripple/tests/bind_policy.rs b/crates/tranquil-ripple/tests/bind_policy.rs new file mode 100644 index 0000000..8f04f0f --- /dev/null +++ b/crates/tranquil-ripple/tests/bind_policy.rs @@ -0,0 +1,70 @@ +use tokio_util::sync::CancellationToken; +use tranquil_ripple::{RippleConfig, RippleEngine, RippleStartError}; + +fn config(cluster_key: Option<&str>, allow_insecure: bool) -> RippleConfig { + RippleConfig { + bind_addr: "0.0.0.0:0".parse().unwrap(), + seed_peers: Vec::new(), + machine_id: 1, + gossip_interval_ms: 100, + cache_max_bytes: 64 * 1024 * 1024, + cluster_key: cluster_key.map(str::to_string), + allow_insecure, + } +} + +#[tokio::test] +async fn non_loopback_bind_without_key_refuses_to_start() { + let shutdown = CancellationToken::new(); + let result = RippleEngine::start(config(None, false), shutdown.clone()).await; + shutdown.cancel(); + match result { + Err(RippleStartError::Config(msg)) => { + assert!( + msg.contains("RIPPLE_CLUSTER_KEY"), + "unexpected message: {msg}" + ) + } + Err(other) => panic!("expected config error, got {other}"), + Ok(_) => panic!("engine must refuse a keyless non-loopback bind"), + } +} + +#[tokio::test] +async fn non_loopback_bind_with_cluster_key_starts() { + let shutdown = CancellationToken::new(); + let result = + RippleEngine::start(config(Some("nautilus-secret"), false), shutdown.clone()).await; + assert!( + result.is_ok(), + "keyed bind must start: {:?}", + result.err().map(|e| e.to_string()) + ); + shutdown.cancel(); +} + +#[tokio::test] +async fn non_loopback_bind_with_allow_insecure_starts() { + let shutdown = CancellationToken::new(); + let result = RippleEngine::start(config(None, true), shutdown.clone()).await; + assert!( + result.is_ok(), + "allow_insecure bind must start: {:?}", + result.err().map(|e| e.to_string()) + ); + shutdown.cancel(); +} + +#[tokio::test] +async fn loopback_bind_without_key_starts() { + let shutdown = CancellationToken::new(); + let mut cfg = config(None, false); + cfg.bind_addr = "127.0.0.1:0".parse().unwrap(); + let result = RippleEngine::start(cfg, shutdown.clone()).await; + assert!( + result.is_ok(), + "loopback keyless bind must start: {:?}", + result.err().map(|e| e.to_string()) + ); + shutdown.cancel(); +} diff --git a/crates/tranquil-ripple/tests/two_node_convergence.rs b/crates/tranquil-ripple/tests/two_node_convergence.rs index 8410570..97246f3 100644 --- a/crates/tranquil-ripple/tests/two_node_convergence.rs +++ b/crates/tranquil-ripple/tests/two_node_convergence.rs @@ -16,6 +16,8 @@ async fn spawn_pair( machine_id: 1, gossip_interval_ms: 100, cache_max_bytes: 64 * 1024 * 1024, + cluster_key: None, + allow_insecure: false, }; let (cache_a, rl_a, addr_a) = RippleEngine::start(config_a, shutdown.clone()) .await @@ -27,6 +29,8 @@ async fn spawn_pair( machine_id: 2, gossip_interval_ms: 100, cache_max_bytes: 64 * 1024 * 1024, + cluster_key: None, + allow_insecure: false, }; let (cache_b, rl_b, _addr_b) = RippleEngine::start(config_b, shutdown.clone()) .await @@ -634,6 +638,8 @@ async fn two_node_partition_recovery() { machine_id: 100, gossip_interval_ms: 100, cache_max_bytes: 64 * 1024 * 1024, + cluster_key: None, + allow_insecure: false, }; let (cache_a, _rl_a, addr_a) = RippleEngine::start(config_a, shutdown.clone()) .await @@ -660,6 +666,8 @@ async fn two_node_partition_recovery() { machine_id: 200, gossip_interval_ms: 100, cache_max_bytes: 64 * 1024 * 1024, + cluster_key: None, + allow_insecure: false, }; let (cache_b, _rl_b, _addr_b) = RippleEngine::start(config_b, shutdown.clone()) .await diff --git a/example.toml b/example.toml index da135fb..12a5ce0 100644 --- a/example.toml +++ b/example.toml @@ -312,7 +312,9 @@ #valkey_url = [cache.ripple] -# Address to bind the Ripple gossip protocol listener. +# Address to bind the Ripple gossip protocol listener. With the default +# value and no cluster_key or peers configured, the listener binds +# loopback instead and runs as a single node. # # Can also be specified via environment variable `RIPPLE_BIND`. # @@ -343,6 +345,20 @@ # Default value: 256 #cache_max_mb = 256 +# Pre-shared cluster key authenticating ripple peers. Every node in the +# cluster must set the same value. When unset, peers are unauthenticated. +# +# Can also be specified via environment variable `RIPPLE_CLUSTER_KEY`. +#cluster_key = + +# Allow ripple to bind a non-loopback address without a cluster key. +# Peers will be unauthenticated. Intended for trusted private networks. +# +# Can also be specified via environment variable `RIPPLE_ALLOW_INSECURE`. +# +# Default value: false +#allow_insecure = false + [plc] # Base URL of the PLC directory. #