Skip to content

Commit bf5acab

Browse files
tlongwell-blocknpub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf
andauthored
relay: add audit logging disable switch (#2134)
Signed-off-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co> Co-authored-by: npub1jh9wn95s0472h86ahapupaf7m6kx4v9sx2n0atj2hltcfer8k06s5n3pyf <95cae996907d7cab9f5dbf43c0f53edeac6ab0b032a6feae4abfd784e467b3f5@sprout-oss.stage.blox.sqprod.co>
1 parent 2b0f5e9 commit bf5acab

6 files changed

Lines changed: 98 additions & 41 deletions

File tree

‎TESTING.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -270,6 +270,7 @@ out of the box with `just setup` or `just relay`. Common overrides:
270270
| `BUZZ_REQUIRE_AUTH_TOKEN` | `false` | When true, REST requires NIP-98 (no `X-Pubkey` fallback) |
271271
| `BUZZ_REQUIRE_RELAY_MEMBERSHIP` | `false` | When true, only pubkeys in `relay_members` can connect |
272272
| `BUZZ_REQUIRE_MEDIA_GET_AUTH` | `false` | When true, `GET`/`HEAD /media/*` require Blossom kind 24242 `t=get` auth plus relay membership. |
273+
| `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. |
273274
| `BUZZ_AUTO_MIGRATE` | `false` | Opt in with `true`/`1`/`yes`/`on` to run embedded SQLx migrations on relay startup |
274275
| `RELAY_OWNER_PUBKEY` | unset | Bootstrapped as `owner` in `relay_members` at first start |
275276
| `BUZZ_ALLOW_NIP_OA_AUTH` | `false` | Enable NIP-OA owner attestation for membership |

‎crates/buzz-relay/src/api/media.rs‎

Lines changed: 19 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -420,24 +420,25 @@ pub async fn upload_blob(
420420
.increment(1);
421421

422422
// Audit via bounded channel — same pattern as event audit.
423-
let desc = descriptor.clone();
424-
if let Err(e) = state
425-
.audit_tx
426-
.send(NewAuditEntry {
427-
community_id: auth.tenant.community(),
428-
action: AuditAction::MediaUploaded,
429-
actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
430-
object_id: Some(desc.sha256.clone()),
431-
detail: serde_json::json!({
432-
"sha256": desc.sha256,
433-
"size": desc.size,
434-
"mime": desc.mime_type,
435-
}),
436-
})
437-
.await
438-
{
439-
tracing::error!("Media audit channel closed — entry lost: {e}");
440-
metrics::counter!("buzz_audit_send_errors_total").increment(1);
423+
if let Some(audit_tx) = &state.audit_tx {
424+
let desc = descriptor.clone();
425+
if let Err(e) = audit_tx
426+
.send(NewAuditEntry {
427+
community_id: auth.tenant.community(),
428+
action: AuditAction::MediaUploaded,
429+
actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
430+
object_id: Some(desc.sha256.clone()),
431+
detail: serde_json::json!({
432+
"sha256": desc.sha256,
433+
"size": desc.size,
434+
"mime": desc.mime_type,
435+
}),
436+
})
437+
.await
438+
{
439+
tracing::error!("Media audit channel closed — entry lost: {e}");
440+
metrics::counter!("buzz_audit_send_errors_total").increment(1);
441+
}
441442
}
442443

443444
Ok(Json(descriptor))

‎crates/buzz-relay/src/config.rs‎

Lines changed: 46 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -190,6 +190,11 @@ pub struct Config {
190190
/// serving media GET/HEAD. Default off for staged client rollout.
191191
pub require_media_get_auth: bool,
192192

193+
/// Whether tamper-evident event/media audit logging is enabled. Defaults to true.
194+
/// This does not control the separate `moderation_actions` audit trail.
195+
/// Set `BUZZ_AUDIT_ENABLED=false` for deployments that do not require it.
196+
pub audit_enabled: bool,
197+
193198
/// Optional override for ephemeral channel TTL (in seconds).
194199
/// When set, any channel created with a TTL tag will use this value instead
195200
/// of the client-provided one. Useful for testing ephemeral expiry quickly.
@@ -341,9 +346,9 @@ fn parse_push_gateway_delivery_url(raw: &str) -> Result<url::Url, ConfigError> {
341346
Ok(url)
342347
}
343348

344-
fn parse_optional_bool(name: &str) -> Result<bool, ConfigError> {
349+
fn parse_bool(name: &str, default: bool) -> Result<bool, ConfigError> {
345350
match std::env::var(name) {
346-
Err(std::env::VarError::NotPresent) => Ok(false),
351+
Err(std::env::VarError::NotPresent) => Ok(default),
347352
Err(error) => Err(ConfigError::InvalidValue(format!(
348353
"{name} must be valid UTF-8: {error}"
349354
))),
@@ -357,6 +362,10 @@ fn parse_optional_bool(name: &str) -> Result<bool, ConfigError> {
357362
}
358363
}
359364

365+
fn parse_optional_bool(name: &str) -> Result<bool, ConfigError> {
366+
parse_bool(name, false)
367+
}
368+
360369
fn ensure_git_repo_path(
361370
raw: impl Into<std::path::PathBuf>,
362371
) -> Result<std::path::PathBuf, ConfigError> {
@@ -738,6 +747,7 @@ impl Config {
738747
let terms_markdown = read_policy_markdown("BUZZ_TERMS_OF_SERVICE_MARKDOWN")?;
739748
let privacy_markdown = read_policy_markdown("BUZZ_PRIVACY_POLICY_MARKDOWN")?;
740749
let age_attestation_required = parse_optional_bool("BUZZ_AGE_ATTESTATION_REQUIRED")?;
750+
let audit_enabled = parse_bool("BUZZ_AUDIT_ENABLED", true)?;
741751
let join_policy = if terms_markdown.is_none()
742752
&& privacy_markdown.is_none()
743753
&& !age_attestation_required
@@ -849,6 +859,7 @@ impl Config {
849859
media_max_concurrent_uploads_per_pubkey,
850860
media_uploads_per_minute,
851861
require_media_get_auth,
862+
audit_enabled,
852863
ephemeral_ttl_override,
853864
git_repo_path,
854865
git_max_pack_bytes,
@@ -953,6 +964,39 @@ mod tests {
953964
);
954965
}
955966

967+
#[test]
968+
fn audit_logging_defaults_on_and_accepts_explicit_off() {
969+
let _guard = ENV_MUTEX.lock().unwrap();
970+
let previous = std::env::var_os("BUZZ_AUDIT_ENABLED");
971+
std::env::remove_var("BUZZ_AUDIT_ENABLED");
972+
assert!(parse_bool("BUZZ_AUDIT_ENABLED", true).unwrap());
973+
std::env::set_var("BUZZ_AUDIT_ENABLED", "false");
974+
assert!(!parse_bool("BUZZ_AUDIT_ENABLED", true).unwrap());
975+
if let Some(value) = previous {
976+
std::env::set_var("BUZZ_AUDIT_ENABLED", value);
977+
} else {
978+
std::env::remove_var("BUZZ_AUDIT_ENABLED");
979+
}
980+
}
981+
982+
#[test]
983+
fn audit_logging_rejects_invalid_boolean() {
984+
let _guard = ENV_MUTEX.lock().unwrap();
985+
let previous = std::env::var_os("BUZZ_AUDIT_ENABLED");
986+
std::env::set_var("BUZZ_AUDIT_ENABLED", "sometimes");
987+
let result = parse_bool("BUZZ_AUDIT_ENABLED", true);
988+
if let Some(value) = previous {
989+
std::env::set_var("BUZZ_AUDIT_ENABLED", value);
990+
} else {
991+
std::env::remove_var("BUZZ_AUDIT_ENABLED");
992+
}
993+
assert!(matches!(
994+
result,
995+
Err(ConfigError::InvalidValue(ref message))
996+
if message.contains("BUZZ_AUDIT_ENABLED")
997+
));
998+
}
999+
9561000
#[test]
9571001
fn join_policy_age_attestation_rejects_invalid_boolean() {
9581002
let _guard = ENV_MUTEX.lock().unwrap();

‎crates/buzz-relay/src/handlers/event.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -545,6 +545,9 @@ async fn enqueue_event_created_audit(
545545
actor_pubkey_hex: &str,
546546
event_id_hex: &str,
547547
) {
548+
let Some(audit_tx) = &state.audit_tx else {
549+
return;
550+
};
548551
// Audit via bounded channel (capacity 1000). Uses .send().await so entries
549552
// are never silently dropped — backpressure propagates to the event handler
550553
// if the queue is full. This is intentional: the audit advisory lock already
@@ -568,7 +571,7 @@ async fn enqueue_event_created_audit(
568571
"channel_id": stored_event.channel_id,
569572
}),
570573
};
571-
if let Err(e) = state.audit_tx.send(audit_entry).await {
574+
if let Err(e) = audit_tx.send(audit_entry).await {
572575
error!(event_id = %event_id_hex, "Audit channel closed — entry lost: {e}");
573576
metrics::counter!("buzz_audit_send_errors_total").increment(1);
574577
}

‎crates/buzz-relay/src/main.rs‎

Lines changed: 15 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -128,12 +128,14 @@ async fn main() -> anyhow::Result<()> {
128128
health_port = config.health_port,
129129
metrics_port = config.metrics_port,
130130
max_frame_bytes = config.max_frame_bytes,
131+
audit_enabled = config.audit_enabled,
131132
"Config loaded"
132133
);
133134

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

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

331335
let redis_pool = {
332336
let cfg = deadpool_redis::Config::from_url(&config.redis_url);

‎crates/buzz-relay/src/state.rs‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -438,8 +438,8 @@ pub struct AppState {
438438
pub db: Db,
439439
/// Redis pool for readiness health checks.
440440
pub redis_pool: deadpool_redis::Pool,
441-
/// Audit event service.
442-
pub audit: Arc<AuditService>,
441+
/// Audit event service, absent when audit logging is disabled.
442+
pub audit: Option<Arc<AuditService>>,
443443
/// Pub/sub manager for broadcasting events to subscribers.
444444
pub pubsub: Arc<PubSubManager>,
445445
/// Authentication service.
@@ -497,9 +497,8 @@ pub struct AppState {
497497
/// access check so open channels stay zero-cost. Invalidated on a flip.
498498
pub channel_visibility_cache: Arc<moka::sync::Cache<(CommunityId, Uuid), String>>,
499499

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

586-
let audit_arc = Arc::new(audit);
585+
let audit_arc = audit.into().map(Arc::new);
587586
let (audit_tx, mut audit_rx) = mpsc::channel::<buzz_audit::NewAuditEntry>(1000);
588-
let audit_for_worker = Arc::clone(&audit_arc);
587+
let audit_for_worker = audit_arc.clone();
589588
let audit_cancel = CancellationToken::new();
590589
let audit_cancel_worker = audit_cancel.clone();
591590
let audit_worker_handle = tokio::spawn(async move {
591+
let Some(audit_for_worker) = audit_for_worker else {
592+
audit_cancel_worker.cancelled().await;
593+
return;
594+
};
592595
// Normal operation: process entries as they arrive.
593596
loop {
594597
tokio::select! {
@@ -631,6 +634,7 @@ impl AppState {
631634
let nip98_replay: Arc<dyn Nip98ReplayGuard> =
632635
Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
633636
let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
637+
let audit_enabled = audit_arc.is_some();
634638
let state = Self {
635639
config: Arc::new(config),
636640
db,
@@ -678,7 +682,7 @@ impl AppState {
678682
.support_invalidation_closures()
679683
.build(),
680684
),
681-
audit_tx,
685+
audit_tx: audit_enabled.then_some(audit_tx),
682686
media_storage: Arc::new(media_storage),
683687
git_store,
684688
audio_rooms: Arc::new(AudioRoomManager::new()),

0 commit comments

Comments
 (0)