From 893a2624964354284daec5fa61033b3f6ea0093b Mon Sep 17 00:00:00 2001 From: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> Date: Mon, 10 Aug 2026 16:11:50 -0400 Subject: [PATCH] fix(local-mode): harden sidecar foundations Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz> --- crates/buzz-db/src/sqlite.rs | 203 ++++++++++++++++++++- crates/buzz-relay/src/main.rs | 28 ++- crates/buzz-relay/src/metrics.rs | 34 +++- desktop/src-tauri/src/local_relay.rs | 252 ++++++++++++++++++--------- 4 files changed, 428 insertions(+), 89 deletions(-) diff --git a/crates/buzz-db/src/sqlite.rs b/crates/buzz-db/src/sqlite.rs index 2c2e6f6358..516b3b712c 100644 --- a/crates/buzz-db/src/sqlite.rs +++ b/crates/buzz-db/src/sqlite.rs @@ -305,11 +305,110 @@ pub(crate) async fn add_member( role: crate::channel::MemberRole, invited_by: Option<&[u8]>, ) -> Result { - get_channel(pool, community, channel_id).await?; + if pubkey.len() != 32 { + return Err(crate::DbError::InvalidData(format!( + "pubkey must be 32 bytes, got {}", + pubkey.len() + ))); + } + + let mut tx = pool.begin().await?; + let channel = sqlx::query( + "SELECT * FROM channels WHERE community_id = ?1 AND id = ?2 AND deleted_at IS NULL", + ) + .bind(community.as_uuid().to_string()) + .bind(channel_id.to_string()) + .fetch_optional(&mut *tx) + .await? + .ok_or(crate::DbError::ChannelNotFound(channel_id))?; + let channel = channel_record(channel)?; + + let effective_role = if channel.visibility == "private" { + let inviter = invited_by.ok_or_else(|| { + crate::DbError::AccessDenied("private channel requires an invite".to_string()) + })?; + let creator_bootstrap = inviter == pubkey && inviter == channel.created_by.as_slice(); + if !creator_bootstrap { + let inviter_role = active_member_role(&mut tx, channel_id, inviter) + .await? + .ok_or_else(|| { + crate::DbError::AccessDenied("inviter is not an active member".to_string()) + })?; + if role.is_elevated() && !is_elevated_role(&inviter_role) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may grant elevated roles".to_string(), + )); + } + } + role + } else if role.is_elevated() { + let granter_role = match invited_by { + Some(inviter) => active_member_role(&mut tx, channel_id, inviter).await?, + None => None, + }; + if !granter_role.is_some_and(|role| is_elevated_role(&role)) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may grant elevated roles".to_string(), + )); + } + role + } else { + role + }; + + if let Some(current_role) = active_member_role(&mut tx, channel_id, pubkey).await? { + if current_role != effective_role.as_str() { + let actor_role = match invited_by { + Some(inviter) => active_member_role(&mut tx, channel_id, inviter).await?, + None => None, + }; + if !actor_role.is_some_and(|role| is_elevated_role(&role)) { + return Err(crate::DbError::AccessDenied( + "only owners/admins may change an active member's role".to_string(), + )); + } + if current_role == "owner" && effective_role != crate::channel::MemberRole::Owner { + let owner_count: i64 = sqlx::query_scalar( + "SELECT count(*) FROM channel_members WHERE channel_id = ?1 AND role = 'owner' AND removed_at IS NULL", + ) + .bind(channel_id.to_string()) + .fetch_one(&mut *tx) + .await?; + if owner_count <= 1 { + return Err(crate::DbError::AccessDenied( + "cannot demote the last owner — transfer ownership first".to_string(), + )); + } + } + } + } + sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(channel_id, pubkey) DO UPDATE SET role = excluded.role, invited_by = excluded.invited_by, removed_at = NULL") - .bind(channel_id.to_string()).bind(pubkey).bind(role.as_str()).bind(invited_by).execute(pool).await?; - member_record(sqlx::query("SELECT channel_id, pubkey, role, joined_at, invited_by, removed_at FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2") - .bind(channel_id.to_string()).bind(pubkey).fetch_one(pool).await?) + .bind(channel_id.to_string()).bind(pubkey).bind(effective_role.as_str()).bind(invited_by).execute(&mut *tx).await?; + let row = sqlx::query("SELECT channel_id, pubkey, role, joined_at, invited_by, removed_at FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2") + .bind(channel_id.to_string()).bind(pubkey).fetch_one(&mut *tx).await?; + let record = member_record(row)?; + tx.commit().await?; + Ok(record) +} + +async fn active_member_role( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + channel_id: Uuid, + pubkey: &[u8], +) -> Result> { + sqlx::query_scalar( + "SELECT role FROM channel_members WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL", + ) + .bind(channel_id.to_string()) + .bind(pubkey) + .fetch_optional(&mut **tx) + .await + .map_err(Into::into) +} + +fn is_elevated_role(role: &str) -> bool { + matches!(role, "owner" | "admin") } pub(crate) async fn is_member( @@ -566,6 +665,102 @@ mod tests { std::fs::remove_file(path).unwrap(); } + #[tokio::test] + async fn sqlite_membership_preserves_public_contract_invariants() { + let pool = connect(":memory:").await.unwrap(); + let community = ensure_configured_community(&pool, "local.buzz") + .await + .unwrap(); + let owner = Keys::generate().public_key().to_bytes(); + let member = Keys::generate().public_key().to_bytes(); + let other = Keys::generate().public_key().to_bytes(); + let channel_id = Uuid::new_v4(); + create_channel_with_id( + &pool, + community.id, + channel_id, + "private", + crate::channel::ChannelType::Stream, + crate::channel::ChannelVisibility::Private, + None, + &owner, + None, + ) + .await + .unwrap(); + + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &[0; 31], + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await, + Err(crate::DbError::InvalidData(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + None, + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await + .unwrap(); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &other, + crate::channel::MemberRole::Admin, + Some(&member), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &member, + crate::channel::MemberRole::Guest, + Some(&member), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + assert!(matches!( + add_member( + &pool, + community.id, + channel_id, + &owner, + crate::channel::MemberRole::Member, + Some(&owner), + ) + .await, + Err(crate::DbError::AccessDenied(_)) + )); + } + #[tokio::test] async fn rebind_prefers_owned_then_oldest_community() { let pool = connect(":memory:").await.unwrap(); diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index f972caaff4..a6adbbf083 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1,4 +1,5 @@ use std::collections::{HashMap, HashSet}; +use std::net::{IpAddr, Ipv4Addr}; use std::sync::atomic::Ordering; use std::sync::Arc; @@ -18,7 +19,7 @@ use buzz_db::{Db, DbConfig}; use buzz_pubsub::{rate_limiter::AdmissionRateLimiter, InProcessNip98ReplayGuard, PubSubManager}; use buzz_search::SearchService; -use buzz_relay::config::{Config, MAX_DRAIN_JITTER_MS}; +use buzz_relay::config::{Config, RelayProfile, MAX_DRAIN_JITTER_MS}; use buzz_relay::metrics as relay_metrics; use buzz_relay::router::{build_health_router, build_router}; use buzz_relay::state::{AppBackends, AppState}; @@ -1217,7 +1218,7 @@ async fn run_single_node(config: Config, tracer_init: telemetry::TracerInit) -> )); } - relay_metrics::install(config.metrics_port, usage_metrics_idle_timeout_secs(60)); + relay_metrics::install_loopback(config.metrics_port, usage_metrics_idle_timeout_secs(60)); let db_path = std::env::var("BUZZ_LOCAL_DB").unwrap_or_else(|_| "buzz-local.sqlite".to_string()); let media_root = @@ -1340,6 +1341,24 @@ async fn run_single_node(config: Config, tracer_init: telemetry::TracerInit) -> /// roughly the 5s grace plus the ack wait. const GRACEFUL_DRAIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); +fn health_listener_ip(profile: RelayProfile) -> IpAddr { + match profile { + RelayProfile::SingleNode => IpAddr::V4(Ipv4Addr::LOCALHOST), + RelayProfile::Production => IpAddr::V4(Ipv4Addr::UNSPECIFIED), + } +} + +#[cfg(test)] +mod listener_tests { + use super::{health_listener_ip, RelayProfile}; + + #[test] + fn single_node_health_listener_is_loopback_only() { + assert!(health_listener_ip(RelayProfile::SingleNode).is_loopback()); + assert!(health_listener_ip(RelayProfile::Production).is_unspecified()); + } +} + async fn serve( router: axum::Router, health_router: axum::Router, @@ -1347,10 +1366,11 @@ async fn serve( ) -> anyhow::Result<()> { let config = &state.config; - let health_listener = tokio::net::TcpListener::bind(("0.0.0.0", config.health_port)) + let health_host = health_listener_ip(config.profile); + let health_listener = tokio::net::TcpListener::bind((health_host, config.health_port)) .await .map_err(|e| anyhow::anyhow!("Failed to bind health port {}: {e}", config.health_port))?; - info!(port = config.health_port, "Health probe listener started"); + info!(host = %health_host, port = config.health_port, "Health probe listener started"); tokio::spawn(async move { axum::serve(health_listener, health_router).await.ok(); }); diff --git a/crates/buzz-relay/src/metrics.rs b/crates/buzz-relay/src/metrics.rs index 16e521a44e..21d1ff43c4 100644 --- a/crates/buzz-relay/src/metrics.rs +++ b/crates/buzz-relay/src/metrics.rs @@ -14,6 +14,7 @@ //! recorded by [`track_metrics`] middleware on the app router. Buzz-specific //! metrics are recorded inline at their call sites. +use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::time::{Duration, Instant}; use axum::{ @@ -56,6 +57,17 @@ const GIT_PACK_BUCKETS: [f64; 9] = [0.0, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 1 /// Integer-count buckets for fan-out recipient histograms. const FANOUT_BUCKETS: [f64; 9] = [0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0, 1000.0]; +fn prometheus_listener(port: u16, loopback_only: bool) -> SocketAddr { + SocketAddr::new( + IpAddr::V4(if loopback_only { + Ipv4Addr::LOCALHOST + } else { + Ipv4Addr::UNSPECIFIED + }), + port, + ) +} + /// Install the global metrics recorder and spawn the Prometheus HTTP exporter. /// /// `build()` returns the recorder + exporter future and internally spawns @@ -64,8 +76,17 @@ const FANOUT_BUCKETS: [f64; 9] = [0.0, 1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 500.0, /// Must be called from within a Tokio runtime. /// Panics if a recorder is already installed or the port is in use. pub fn install(port: u16, gauge_idle_timeout_secs: u64) { + install_on(prometheus_listener(port, false), gauge_idle_timeout_secs); +} + +/// Install the metrics exporter on loopback for the single-node profile. +pub fn install_loopback(port: u16, gauge_idle_timeout_secs: u64) { + install_on(prometheus_listener(port, true), gauge_idle_timeout_secs); +} + +fn install_on(listener: SocketAddr, gauge_idle_timeout_secs: u64) { let (recorder, exporter) = PrometheusBuilder::new() - .with_http_listener(([0, 0, 0, 0], port)) + .with_http_listener(listener) // Remove gauge series that the relay intentionally stops emitting. .idle_timeout( MetricKindMask::GAUGE, @@ -205,3 +226,14 @@ pub async fn track_metrics(req: Request, next: Next) -> Response { response } + +#[cfg(test)] +mod tests { + use super::prometheus_listener; + + #[test] + fn single_node_metrics_listener_is_loopback_only() { + assert!(prometheus_listener(9102, true).ip().is_loopback()); + assert!(prometheus_listener(9102, false).ip().is_unspecified()); + } +} diff --git a/desktop/src-tauri/src/local_relay.rs b/desktop/src-tauri/src/local_relay.rs index 6fcf4f66c1..4813631af5 100644 --- a/desktop/src-tauri/src/local_relay.rs +++ b/desktop/src-tauri/src/local_relay.rs @@ -5,12 +5,11 @@ //! supervises the bundled `buzz-relay` binary. use std::{ - fs::OpenOptions, - io::Write, + io::Read, net::{IpAddr, Ipv4Addr, SocketAddr, TcpListener}, path::{Path, PathBuf}, process::{Child, Command, Stdio}, - sync::Mutex, + sync::{Arc, Mutex}, time::{Duration, Instant}, }; @@ -19,6 +18,9 @@ use tauri::{AppHandle, Manager}; const STARTUP_TIMEOUT: Duration = Duration::from_secs(15); const POLL_INTERVAL: Duration = Duration::from_millis(100); +const STDERR_TAIL_BYTES: usize = 8 * 1024; + +type StderrTail = Arc>>; pub(crate) struct LocalRelayRuntime { child: Child, @@ -104,62 +106,41 @@ pub(crate) fn start( keys: &Keys, requested_port: Option, ) -> Result { - let relay_port = requested_port.unwrap_or(free_loopback_port()?); let owner_pubkey = keys.public_key().to_hex(); let data_dir = local_data_dir(app, &owner_pubkey)?; let relay_keys = load_or_create_relay_keys(&data_dir)?; - let config = LocalRelayConfig { - relay_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), relay_port), - health_port: free_loopback_port()?, - metrics_port: free_loopback_port()?, - data_dir, - owner_pubkey, - relay_private_key: relay_keys.secret_key().to_secret_hex(), - }; - let url = config.url(); - let media_dir = config.data_dir.join("media"); + let media_dir = data_dir.join("media"); std::fs::create_dir_all(&media_dir) .map_err(|e| format!("create local relay media dir: {e}"))?; - let binary = relay_binary()?; - let mut command = Command::new(&binary); - command - .envs(config.environment()) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()); - let mut child = command - .spawn() - .map_err(|e| format!("start bundled buzz-relay at {}: {e}", binary.display()))?; - let info_url = format!("http://{}/info", config.relay_addr); - let deadline = Instant::now() + STARTUP_TIMEOUT; - while Instant::now() < deadline { - if child - .try_wait() - .map_err(|e| format!("inspect local relay: {e}"))? - .is_some() - { - return Err("bundled buzz-relay exited before becoming ready".to_string()); + // The relay owns its TCP listeners, so the desktop cannot safely retain a + // reservation across `spawn`. Retry a complete, distinct three-port set + // when an automatically-selected port loses that race; explicit user ports + // fail with the child diagnostic instead of silently moving the endpoint. + let attempts = if requested_port.is_some() { 1 } else { 3 }; + let mut last_error = None; + for _ in 0..attempts { + let (relay_port, health_port, metrics_port) = + allocate_distinct_loopback_ports(requested_port)?; + let config = LocalRelayConfig { + relay_addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), relay_port), + health_port, + metrics_port, + data_dir: data_dir.clone(), + owner_pubkey: owner_pubkey.clone(), + relay_private_key: relay_keys.secret_key().to_secret_hex(), + }; + + match spawn_and_wait(&binary, config) { + Ok(runtime) => return Ok(runtime), + Err(error) if requested_port.is_none() && is_bind_conflict(&error) => { + last_error = Some(error); + } + Err(error) => return Err(error), } - if reqwest::blocking::Client::new() - .get(&info_url) - .header("Accept", "application/nostr+json") - .send() - .is_ok_and(|response| response.status().is_success()) - { - return Ok(LocalRelayRuntime { - child, - url, - owner_pubkey: config.owner_pubkey, - }); - } - std::thread::sleep(POLL_INTERVAL); } - - let _ = child.kill(); - let _ = child.wait(); - Err("bundled buzz-relay did not become ready within 15 seconds".to_string()) + Err(last_error.expect("at least one local relay startup attempt")) } pub(crate) fn ensure_started( @@ -237,40 +218,132 @@ fn local_data_dir(app: &AppHandle, owner_pubkey: &str) -> Result Result { - let path = data_dir.join("relay-service.key"); - match std::fs::read_to_string(&path) { - Ok(secret) => Keys::parse(secret.trim()) - .map_err(|error| format!("parse local relay service key: {error}")), - Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - std::fs::create_dir_all(data_dir) - .map_err(|error| format!("create local relay data dir: {error}"))?; - let keys = Keys::generate(); - let mut file = OpenOptions::new() - .write(true) - .create_new(true) - .open(&path) - .map_err(|error| format!("create local relay service key: {error}"))?; - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - file.set_permissions(std::fs::Permissions::from_mode(0o600)) - .map_err(|error| format!("secure local relay service key: {error}"))?; - } - file.write_all(keys.secret_key().to_secret_hex().as_bytes()) - .map_err(|error| format!("write local relay service key: {error}"))?; - Ok(keys) - } - Err(error) => Err(format!("read local relay service key: {error}")), - } -} - fn sqlite_url(path: &Path) -> String { // sqlx accepts an absolute sqlite URL with three slashes. Path display is // intentional: app-data paths come from the OS, not user relay input. format!("sqlite://{}", path.display()) } +/// Load the relay's service identity from its identity-scoped nest, or create +/// it once with the same atomic, owner-only file semantics used for private +/// desktop identities. This key signs relay-originated metadata; it is never +/// the human/owner key supplied by `apply_workspace`. +fn load_or_create_relay_keys(data_dir: &Path) -> Result { + let path = data_dir.join("relay-service.key"); + if path.exists() { + let value = std::fs::read_to_string(&path) + .map_err(|e| format!("read local relay service key: {e}"))?; + return Keys::parse(value.trim()) + .map_err(|e| format!("parse local relay service key: {e}")); + } + + std::fs::create_dir_all(data_dir).map_err(|e| format!("create local relay data dir: {e}"))?; + let keys = Keys::generate(); + crate::app_state::save_key_file(&path, &keys) + .map_err(|e| format!("persist local relay service key: {e}"))?; + Ok(keys) +} + +fn is_bind_conflict(error: &str) -> bool { + error.contains("Address already in use") || error.contains("address already in use") +} + +fn spawn_and_wait(binary: &Path, config: LocalRelayConfig) -> Result { + let url = config.url(); + let info_url = format!("http://{}/info", config.relay_addr); + let stderr = Arc::new(Mutex::new(Vec::new())); + let mut command = Command::new(binary); + command + .envs(config.environment()) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()); + let mut child = command + .spawn() + .map_err(|e| format!("start bundled buzz-relay at {}: {e}", binary.display()))?; + let stderr_reader = capture_stderr(child.stderr.take(), stderr.clone()); + + let client = reqwest::blocking::Client::new(); + let deadline = Instant::now() + STARTUP_TIMEOUT; + while Instant::now() < deadline { + if let Some(status) = child + .try_wait() + .map_err(|e| format!("inspect local relay: {e}"))? + { + let _ = stderr_reader.join(); + return Err(format!( + "bundled buzz-relay exited before becoming ready ({status}){}", + stderr_context(&stderr) + )); + } + if client + .get(&info_url) + .header("Accept", "application/nostr+json") + .send() + .is_ok_and(|response| response.status().is_success()) + { + return Ok(LocalRelayRuntime { + child, + url, + owner_pubkey: config.owner_pubkey, + }); + } + std::thread::sleep(POLL_INTERVAL); + } + + let _ = child.kill(); + let _ = child.wait(); + let _ = stderr_reader.join(); + Err(format!( + "bundled buzz-relay did not become ready within 15 seconds{}", + stderr_context(&stderr) + )) +} + +fn capture_stderr( + stderr: Option, + tail: StderrTail, +) -> std::thread::JoinHandle<()> { + std::thread::spawn(move || { + if let Some(mut stderr) = stderr { + let mut buffer = [0; 1024]; + while let Ok(read) = stderr.read(&mut buffer) { + if read == 0 { + break; + } + let mut tail = tail.lock().expect("stderr tail mutex poisoned"); + tail.extend_from_slice(&buffer[..read]); + if tail.len() > STDERR_TAIL_BYTES { + let excess = tail.len() - STDERR_TAIL_BYTES; + tail.drain(..excess); + } + } + } + }) +} + +fn stderr_context(stderr: &StderrTail) -> String { + let tail = stderr.lock().expect("stderr tail mutex poisoned"); + let text = String::from_utf8_lossy(&tail).trim().to_string(); + (!text.is_empty()) + .then(|| format!("; stderr: {text}")) + .unwrap_or_default() +} + +fn allocate_distinct_loopback_ports( + requested_port: Option, +) -> Result<(u16, u16, u16), String> { + let relay_port = requested_port.unwrap_or(free_loopback_port()?); + let mut ports = vec![relay_port]; + while ports.len() < 3 { + let port = free_loopback_port()?; + if !ports.contains(&port) { + ports.push(port); + } + } + Ok((ports[0], ports[1], ports[2])) +} + fn free_loopback_port() -> Result { let listener = TcpListener::bind("127.0.0.1:0").map_err(|e| format!("pick local relay port: {e}"))?; @@ -304,16 +377,35 @@ fn relay_binary() -> Result { #[cfg(test)] mod tests { use super::{ - is_local_relay_url, load_or_create_relay_keys, local_relay_url, sqlite_url, - LocalRelayConfig, LOCAL_RELAY_SENTINEL, + allocate_distinct_loopback_ports, is_local_relay_url, load_or_create_relay_keys, + local_relay_url, sqlite_url, stderr_context, LocalRelayConfig, StderrTail, + LOCAL_RELAY_SENTINEL, }; use std::{ collections::HashMap, net::{IpAddr, Ipv4Addr, SocketAddr}, path::{Path, PathBuf}, process::{Command, Stdio}, + sync::{Arc, Mutex}, }; + #[test] + fn selected_ports_are_distinct_in_one_startup_attempt() { + let (relay, health, metrics) = allocate_distinct_loopback_ports(None).expect("ports"); + assert_ne!(relay, health); + assert_ne!(relay, metrics); + assert_ne!(health, metrics); + } + + #[test] + fn startup_error_includes_bounded_stderr_tail() { + let tail: StderrTail = Arc::new(Mutex::new(b"bind: address already in use\n".to_vec())); + assert_eq!( + stderr_context(&tail), + "; stderr: bind: address already in use" + ); + } + #[test] fn sqlite_url_is_absolute() { assert_eq!(