mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-08-16 22:36:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

ripple: fail-closed startup, bind policy

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-06-14 18:46:41 +03:00
committed by Tangled
parent 637b817a33
commit a3f729c3cd
13 changed files with 311 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -7709,6 +7709,7 @@ dependencies = [
"async-trait",
"base64 0.22.1",
"redis",
"thiserror 2.0.18",
"tokio-util",
"tracing",
"tranquil-config",
crates/tranquil-cache/Cargo.toml
+1
View File
@@ -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 }
crates/tranquil-cache/src/lib.rs
+16 -30
View File
@@ -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<dyn Cache>, Arc<dyn DistributedRateLimiter>) {
) -> Result<(Arc<dyn Cache>, Arc<dyn DistributedRateLimiter>), 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))
}
crates/tranquil-config/src/lib.rs
+87 -1
View File
@@ -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<String>) {
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<Vec<String>, 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<String>,
/// 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<Vec<String>>,
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>,
crates/tranquil-pds/src/cache/mod.rs
Vendored
+1 -3
View File
@@ -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};
crates/tranquil-pds/src/state.rs
+3 -1
View File
@@ -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();
crates/tranquil-pds/tests/common/mod.rs
+2
View File
@@ -692,6 +692,8 @@ pub async fn spawn_cluster(pool: Option<sqlx::PgPool>, 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
crates/tranquil-ripple/src/config.rs
+90 -1
View File
@@ -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<String>,
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")
);
}
}
crates/tranquil-ripple/src/engine.rs
+14 -2
View File
@@ -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()))?;
crates/tranquil-ripple/src/lib.rs
+1 -1
View File
@@ -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};
crates/tranquil-ripple/tests/bind_policy.rs
+70
View File
@@ -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();
}
crates/tranquil-ripple/tests/two_node_convergence.rs
+8
View File
@@ -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
example.toml
+17 -1
View File
@@ -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.
#