Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions TESTING.md
Original file line number Diff line number Diff line change
Expand Up @@ -270,6 +270,7 @@ out of the box with `just setup` or `just relay`. Common overrides:
| `BUZZ_REQUIRE_AUTH_TOKEN` | `false` | When true, REST requires NIP-98 (no `X-Pubkey` fallback) |
| `BUZZ_REQUIRE_RELAY_MEMBERSHIP` | `false` | When true, only pubkeys in `relay_members` can connect |
| `BUZZ_REQUIRE_MEDIA_GET_AUTH` | `false` | When true, `GET`/`HEAD /media/*` require Blossom kind 24242 `t=get` auth plus relay membership. |
| `BUZZ_AUDIT_ENABLED` | `true` | Tamper-evident event/media audit log. Set `false`/`0`/`off` to skip its DB pool and writes. Does not disable the separate moderation audit trail. |
| `BUZZ_AUTO_MIGRATE` | `false` | Opt in with `true`/`1`/`yes`/`on` to run embedded SQLx migrations on relay startup |
| `RELAY_OWNER_PUBKEY` | unset | Bootstrapped as `owner` in `relay_members` at first start |
| `BUZZ_ALLOW_NIP_OA_AUTH` | `false` | Enable NIP-OA owner attestation for membership |
Expand Down
37 changes: 19 additions & 18 deletions crates/buzz-relay/src/api/media.rs
Original file line number Diff line number Diff line change
Expand Up @@ -420,24 +420,25 @@ pub async fn upload_blob(
.increment(1);

// Audit via bounded channel — same pattern as event audit.
let desc = descriptor.clone();
if let Err(e) = state
.audit_tx
.send(NewAuditEntry {
community_id: auth.tenant.community(),
action: AuditAction::MediaUploaded,
actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
object_id: Some(desc.sha256.clone()),
detail: serde_json::json!({
"sha256": desc.sha256,
"size": desc.size,
"mime": desc.mime_type,
}),
})
.await
{
tracing::error!("Media audit channel closed — entry lost: {e}");
metrics::counter!("buzz_audit_send_errors_total").increment(1);
if let Some(audit_tx) = &state.audit_tx {
let desc = descriptor.clone();
if let Err(e) = audit_tx
.send(NewAuditEntry {
community_id: auth.tenant.community(),
action: AuditAction::MediaUploaded,
actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
object_id: Some(desc.sha256.clone()),
detail: serde_json::json!({
"sha256": desc.sha256,
"size": desc.size,
"mime": desc.mime_type,
}),
})
.await
{
tracing::error!("Media audit channel closed — entry lost: {e}");
metrics::counter!("buzz_audit_send_errors_total").increment(1);
}
}

Ok(Json(descriptor))
Expand Down
48 changes: 46 additions & 2 deletions crates/buzz-relay/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,11 @@ pub struct Config {
/// serving media GET/HEAD. Default off for staged client rollout.
pub require_media_get_auth: bool,

/// Whether tamper-evident event/media audit logging is enabled. Defaults to true.
/// This does not control the separate `moderation_actions` audit trail.
/// Set `BUZZ_AUDIT_ENABLED=false` for deployments that do not require it.
pub audit_enabled: bool,

/// Optional override for ephemeral channel TTL (in seconds).
/// When set, any channel created with a TTL tag will use this value instead
/// of the client-provided one. Useful for testing ephemeral expiry quickly.
Expand Down Expand Up @@ -341,9 +346,9 @@ fn parse_push_gateway_delivery_url(raw: &str) -> Result<url::Url, ConfigError> {
Ok(url)
}

fn parse_optional_bool(name: &str) -> Result<bool, ConfigError> {
fn parse_bool(name: &str, default: bool) -> Result<bool, ConfigError> {
match std::env::var(name) {
Err(std::env::VarError::NotPresent) => Ok(false),
Err(std::env::VarError::NotPresent) => Ok(default),
Err(error) => Err(ConfigError::InvalidValue(format!(
"{name} must be valid UTF-8: {error}"
))),
Expand All @@ -357,6 +362,10 @@ fn parse_optional_bool(name: &str) -> Result<bool, ConfigError> {
}
}

fn parse_optional_bool(name: &str) -> Result<bool, ConfigError> {
parse_bool(name, false)
}

fn ensure_git_repo_path(
raw: impl Into<std::path::PathBuf>,
) -> Result<std::path::PathBuf, ConfigError> {
Expand Down Expand Up @@ -738,6 +747,7 @@ impl Config {
let terms_markdown = read_policy_markdown("BUZZ_TERMS_OF_SERVICE_MARKDOWN")?;
let privacy_markdown = read_policy_markdown("BUZZ_PRIVACY_POLICY_MARKDOWN")?;
let age_attestation_required = parse_optional_bool("BUZZ_AGE_ATTESTATION_REQUIRED")?;
let audit_enabled = parse_bool("BUZZ_AUDIT_ENABLED", true)?;
let join_policy = if terms_markdown.is_none()
&& privacy_markdown.is_none()
&& !age_attestation_required
Expand Down Expand Up @@ -849,6 +859,7 @@ impl Config {
media_max_concurrent_uploads_per_pubkey,
media_uploads_per_minute,
require_media_get_auth,
audit_enabled,
ephemeral_ttl_override,
git_repo_path,
git_max_pack_bytes,
Expand Down Expand Up @@ -953,6 +964,39 @@ mod tests {
);
}

#[test]
fn audit_logging_defaults_on_and_accepts_explicit_off() {
let _guard = ENV_MUTEX.lock().unwrap();
let previous = std::env::var_os("BUZZ_AUDIT_ENABLED");
std::env::remove_var("BUZZ_AUDIT_ENABLED");
assert!(parse_bool("BUZZ_AUDIT_ENABLED", true).unwrap());
std::env::set_var("BUZZ_AUDIT_ENABLED", "false");
assert!(!parse_bool("BUZZ_AUDIT_ENABLED", true).unwrap());
if let Some(value) = previous {
std::env::set_var("BUZZ_AUDIT_ENABLED", value);
} else {
std::env::remove_var("BUZZ_AUDIT_ENABLED");
}
}

#[test]
fn audit_logging_rejects_invalid_boolean() {
let _guard = ENV_MUTEX.lock().unwrap();
let previous = std::env::var_os("BUZZ_AUDIT_ENABLED");
std::env::set_var("BUZZ_AUDIT_ENABLED", "sometimes");
let result = parse_bool("BUZZ_AUDIT_ENABLED", true);
if let Some(value) = previous {
std::env::set_var("BUZZ_AUDIT_ENABLED", value);
} else {
std::env::remove_var("BUZZ_AUDIT_ENABLED");
}
assert!(matches!(
result,
Err(ConfigError::InvalidValue(ref message))
if message.contains("BUZZ_AUDIT_ENABLED")
));
}

#[test]
fn join_policy_age_attestation_rejects_invalid_boolean() {
let _guard = ENV_MUTEX.lock().unwrap();
Expand Down
5 changes: 4 additions & 1 deletion crates/buzz-relay/src/handlers/event.rs
Original file line number Diff line number Diff line change
Expand Up @@ -545,6 +545,9 @@ async fn enqueue_event_created_audit(
actor_pubkey_hex: &str,
event_id_hex: &str,
) {
let Some(audit_tx) = &state.audit_tx else {
return;
};
// Audit via bounded channel (capacity 1000). Uses .send().await so entries
// are never silently dropped — backpressure propagates to the event handler
// if the queue is full. This is intentional: the audit advisory lock already
Expand All @@ -568,7 +571,7 @@ async fn enqueue_event_created_audit(
"channel_id": stored_event.channel_id,
}),
};
if let Err(e) = state.audit_tx.send(audit_entry).await {
if let Err(e) = audit_tx.send(audit_entry).await {
error!(event_id = %event_id_hex, "Audit channel closed — entry lost: {e}");
metrics::counter!("buzz_audit_send_errors_total").increment(1);
}
Expand Down
26 changes: 15 additions & 11 deletions crates/buzz-relay/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -128,12 +128,14 @@ async fn main() -> anyhow::Result<()> {
health_port = config.health_port,
metrics_port = config.metrics_port,
max_frame_bytes = config.max_frame_bytes,
audit_enabled = config.audit_enabled,
"Config loaded"
);

let usage_interval_secs = usage_metrics_interval_secs();
let usage_idle_timeout_secs = usage_metrics_idle_timeout_secs(usage_interval_secs);
relay_metrics::install(config.metrics_port, usage_idle_timeout_secs);
metrics::gauge!("buzz_audit_enabled").set(if config.audit_enabled { 1.0 } else { 0.0 });
info!(
port = config.metrics_port,
idle_timeout_secs = usage_idle_timeout_secs,
Expand Down Expand Up @@ -316,17 +318,19 @@ async fn main() -> anyhow::Result<()> {
Err(e) => error!("Failed to backfill d_tags: {e}"),
}

let audit_pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(5)
.min_connections(1)
.connect(&config.database_url)
.await
.map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?;
let audit = AuditService::new(audit_pool);
// Audit schema is provisioned by the sqlx migrations at startup (the
// `audit_log` DDL is part of the migrated schema), so there is no runtime
// schema-ensure step.
info!("Audit service ready");
let audit = if config.audit_enabled {
let audit_pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(5)
.min_connections(1)
.connect(&config.database_url)
.await
.map_err(|e| anyhow::anyhow!("Audit DB connection failed: {e}"))?;
info!("Audit service ready");
Some(AuditService::new(audit_pool))
} else {
info!("Audit logging disabled by BUZZ_AUDIT_ENABLED");
None
};

let redis_pool = {
let cfg = deadpool_redis::Config::from_url(&config.redis_url);
Expand Down
22 changes: 13 additions & 9 deletions crates/buzz-relay/src/state.rs
Original file line number Diff line number Diff line change
Expand Up @@ -438,8 +438,8 @@ pub struct AppState {
pub db: Db,
/// Redis pool for readiness health checks.
pub redis_pool: deadpool_redis::Pool,
/// Audit event service.
pub audit: Arc<AuditService>,
/// Audit event service, absent when audit logging is disabled.
pub audit: Option<Arc<AuditService>>,
/// Pub/sub manager for broadcasting events to subscribers.
pub pubsub: Arc<PubSubManager>,
/// Authentication service.
Expand Down Expand Up @@ -497,9 +497,8 @@ pub struct AppState {
/// access check so open channels stay zero-cost. Invalidated on a flip.
pub channel_visibility_cache: Arc<moka::sync::Cache<(CommunityId, Uuid), String>>,

/// Bounded channel for audit logging — backpressure instead of unbounded spawns.
/// Uses .send().await (blocks caller if full) because audit entries must not be lost.
pub audit_tx: mpsc::Sender<buzz_audit::NewAuditEntry>,
/// Bounded channel for audit logging, absent when audit logging is disabled.
pub audit_tx: Option<mpsc::Sender<buzz_audit::NewAuditEntry>>,
/// Media storage client (S3/MinIO).
pub media_storage: Arc<MediaStorage>,
/// Git object-store backend (content-addressed packs/manifests plus
Expand Down Expand Up @@ -571,7 +570,7 @@ impl AppState {
config: Config,
db: Db,
redis_pool: deadpool_redis::Pool,
audit: AuditService,
audit: impl Into<Option<AuditService>>,
pubsub: Arc<PubSubManager>,
auth: AuthService,
search: SearchService,
Expand All @@ -583,12 +582,16 @@ impl AppState {
let max_concurrent_handlers = config.max_concurrent_handlers;
let search_arc = Arc::new(search);

let audit_arc = Arc::new(audit);
let audit_arc = audit.into().map(Arc::new);
let (audit_tx, mut audit_rx) = mpsc::channel::<buzz_audit::NewAuditEntry>(1000);
let audit_for_worker = Arc::clone(&audit_arc);
let audit_for_worker = audit_arc.clone();
let audit_cancel = CancellationToken::new();
let audit_cancel_worker = audit_cancel.clone();
let audit_worker_handle = tokio::spawn(async move {
let Some(audit_for_worker) = audit_for_worker else {
audit_cancel_worker.cancelled().await;
return;
};
// Normal operation: process entries as they arrive.
loop {
tokio::select! {
Expand Down Expand Up @@ -631,6 +634,7 @@ impl AppState {
let nip98_replay: Arc<dyn Nip98ReplayGuard> =
Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
let audit_enabled = audit_arc.is_some();
let state = Self {
config: Arc::new(config),
db,
Expand Down Expand Up @@ -678,7 +682,7 @@ impl AppState {
.support_invalidation_closures()
.build(),
),
audit_tx,
audit_tx: audit_enabled.then_some(audit_tx),
media_storage: Arc::new(media_storage),
git_store,
audio_rooms: Arc::new(AudioRoomManager::new()),
Expand Down
Loading