fix(local-mode): harden sidecar foundations

Signed-off-by: Brother Darryl <146fb160a3266e6165bfa385f6048c975eda9e21cf65da097a0b5ea7952532a5@buzz.block.builderlab.xyz>
This commit is contained in:
Brother Darryl
2026-08-10 16:11:50 -04:00
parent f190c2bdb3
commit 893a262496
4 changed files with 428 additions and 89 deletions

View File

@@ -305,11 +305,110 @@ pub(crate) async fn add_member(
role: crate::channel::MemberRole,
invited_by: Option<&[u8]>,
) -> Result<crate::channel::MemberRecord> {
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<Option<String>> {
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();

View File

@@ -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();
});

View File

@@ -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());
}
}

View File

@@ -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<Mutex<Vec<u8>>>;
pub(crate) struct LocalRelayRuntime {
child: Child,
@@ -104,62 +106,41 @@ pub(crate) fn start(
keys: &Keys,
requested_port: Option<u16>,
) -> Result<LocalRelayRuntime, String> {
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<PathBuf, String
.map_err(|e| format!("resolve app data directory for local relay: {e}"))
}
fn load_or_create_relay_keys(data_dir: &Path) -> Result<Keys, String> {
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<Keys, String> {
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<LocalRelayRuntime, String> {
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<std::process::ChildStderr>,
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<u16>,
) -> 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<u16, String> {
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<PathBuf, String> {
#[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!(