diff --git a/Cargo.lock b/Cargo.lock index e86ff6209..632f91128 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1046,6 +1046,7 @@ dependencies = [ "base64", "buzz-audit", "buzz-auth", + "buzz-conformance", "buzz-core", "buzz-db", "buzz-media", diff --git a/Cargo.toml b/Cargo.toml index 4b673aca7..1cb20f65d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -109,6 +109,7 @@ schemars = { version = "1", default-features = false } # Internal crates buzz-core = { path = "crates/buzz-core" } +buzz-conformance = { path = "crates/buzz-conformance" } buzz-db = { path = "crates/buzz-db" } buzz-auth = { path = "crates/buzz-auth" } buzz-pubsub = { path = "crates/buzz-pubsub" } diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index 5752b65f4..c9b80d53f 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -17,6 +17,7 @@ path = "src/main.rs" [dependencies] buzz-core = { workspace = true } +buzz-conformance = { workspace = true } buzz-db = { workspace = true } buzz-auth = { workspace = true } buzz-pubsub = { workspace = true } diff --git a/crates/buzz-relay/src/conformance/mod.rs b/crates/buzz-relay/src/conformance/mod.rs new file mode 100644 index 000000000..8910483fc --- /dev/null +++ b/crates/buzz-relay/src/conformance/mod.rs @@ -0,0 +1,228 @@ +//! Runtime conformance harness — the relay's side of the trace seam. +//! +//! This module hosts the [`Tracer`] re-export and the per-request emit +//! helpers that translate the relay's actual decisions into [`TraceStep`]s +//! from `buzz-conformance`. See `crates/buzz-conformance/src/lib.rs` for +//! the schema and `docs/spec/MultiTenantRelay.tla` for what the schema is +//! grounded in. +//! +//! ## Design rules (from `skill-runtime-formal-compliance`) +//! +//! 1. **Project, don't echo.** The trace carries opaque labels (community +//! UUID, channel UUID, blake3-truncated actor) — never the event id, +//! payload, pubkey bytes, signature, or wall-clock timestamps. The +//! only fields that survive are the ones the spec's `Next` and +//! `Inv_NonInterference` reason about. +//! 2. **Don't normalize away violations.** The emitter records +//! `claimed_community` (from the event's `h` tag) SEPARATELY from +//! `resolved_community` (from `TenantContext::community()`). The +//! checker's M2 bite depends on seeing both. +//! 3. **Drop guard is load-bearing.** Every entry to a critical seam must +//! construct an [`EmitGuard`]; if the seam exits without an emit, the +//! guard's `Drop` records [`TraceAction::ImplBug`] which the checker +//! treats as a coverage breach. +//! +//! ## Wire points +//! +//! - **ingest.rs:** AuthCheck at `check_channel_membership` call site; +//! WriteInsert / WriteInsertGlobal / WriteDuplicate at the two +//! `dispatch_persistent_event` sites; SanitizedError at the outer +//! wrapper based on the IngestError variant. +//! - **req.rs / event.rs:** (held back as additive patch for Eva to apply +//! onto Max's req.rs writes — see thread `c882c9b1…`). + +use std::sync::Arc; + +use buzz_core::tenant::TenantContext; +use nostr::PublicKey; +use uuid::Uuid; + +pub use buzz_conformance::{ + AbstractState, ActorLabel, ChannelLabel, CommunityLabel, HostLabel, OpaqueId, SanitizedReason, + TraceAction, TraceStep, Tracer, Verdict, +}; + +mod tracers; +pub use tracers::{JsonlTracer, NoopTracer}; + +/// Build the [`AbstractState`] for a request from its resolved tenant +/// context and authenticated public key. +/// +/// `community` and `host` come straight from `TenantContext` — server- +/// resolved, never client input. `actor` is the lower 16 bytes of +/// `blake3(pubkey_bytes)` as a hex string, opaque and stable across the +/// run. +pub fn state_for_request(tenant: &TenantContext, actor: &PublicKey) -> AbstractState { + AbstractState { + resolved_community: CommunityLabel::from_uuid(*tenant.community().as_uuid()), + bound_host: HostLabel(tenant.host().to_string()), + actor: actor_label(actor), + } +} + +/// Opaque actor label: first 16 hex chars of the pubkey. The pubkey is +/// already a hash from the client's POV (Schnorr X-only) — equality of +/// the prefix is equivalent to equality of the pubkey for tracing +/// purposes, and the relay already prints full pubkey hexes elsewhere, +/// so the prefix discloses nothing the rest of the log doesn't already. +/// Using the pubkey directly also avoids dragging in a hash dep for what +/// is observability code. +fn actor_label(actor: &PublicKey) -> ActorLabel { + let hex = actor.to_hex(); + let n = hex.len().min(16); + ActorLabel(hex[..n].to_string()) +} + +/// Opaque message id label: first 16 hex chars of the event id. Same +/// rationale as actor labels — the id is already a sha256 hash. +pub fn msg_id_label(event_id: &[u8]) -> OpaqueId { + let mut out = String::with_capacity(16); + for b in event_id.iter().take(8) { + use std::fmt::Write; + let _ = write!(&mut out, "{b:02x}"); + } + OpaqueId(out) +} + +/// Map a UUID channel id into a [`ChannelLabel`]. Channels are not secret +/// — they appear in event `h` tags — so this is a direct wrap. +pub fn channel_label(ch: Uuid) -> ChannelLabel { + ChannelLabel(ch) +} + +/// Extract the *client-claimed* community from an event's `h` tag. Used +/// to populate [`TraceAction`]'s `claimed_community` field. The relay +/// does NOT trust this value for resolution — the resolver uses the +/// server-owned channel→community map. Recording it separately is what +/// makes the M2 (claim≠resolved) bite visible to the checker. +/// +/// Returns `None` if there is no `h` tag, or the `h` tag does not parse +/// as a UUID. +pub fn claimed_community_from_event(event: &nostr::Event) -> Option { + for tag in event.tags.iter() { + // The relay's existing convention: `h` tag carries the community + // uuid (or channel uuid, ambiguous — but on the WRITE path the h + // tag's documented use is the community claim). + let raw = tag.as_slice(); + if raw.first().map(|s| s.as_str()) == Some("h") { + if let Some(val) = raw.get(1) { + if let Ok(parsed) = Uuid::parse_str(val) { + return Some(CommunityLabel::from_uuid(parsed)); + } + } + return None; + } + } + None +} + +/// Build a [`TraceStep::new`] with a freshly-computed [`AbstractState`]. +/// Convenience wrapper to keep the call sites in ingest.rs short. +pub fn step(action: TraceAction, state: AbstractState) -> TraceStep { + TraceStep::new(action, state) +} + +/// Record one step on the tracer. Equivalent to `tracer.record(step(...))`. +/// Kept inline so the call sites stay tight and self-documenting. +pub fn emit(tracer: &Arc, action: TraceAction, state: AbstractState) { + tracer.record(TraceStep::new(action, state)); +} + +/// RAII coverage-breach guard. Constructed at the top of any critical +/// seam (currently: `ingest_event`); the guard observes a [`Tracer`] +/// wrapper that counts emits. If the seam exits without any emit +/// reaching the underlying tracer, `Drop` records a synthetic +/// [`TraceAction::ImplBug`] step — the checker treats that as a +/// coverage breach. +/// +/// The guard wraps the original tracer so production code paths never +/// need to "disarm" or pass anything around — they just call +/// `tracer.record(...)` as before, and the wrapper bumps a counter. If +/// at drop time the counter is zero, the guard emits ImplBug onto the +/// underlying tracer. +pub struct EmitGuard { + /// The inner tracer, used both for the production emits during the + /// request AND for the synthetic ImplBug on Drop if the request + /// emitted nothing. + inner: Arc, + state: AbstractState, + counter: Arc, + kind: &'static str, +} + +/// Wrapper tracer that bumps a counter on every record. Returned by +/// [`EmitGuard::counting_tracer`]. +struct CountingTracer { + inner: Arc, + counter: Arc, +} + +impl std::fmt::Debug for CountingTracer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("CountingTracer").finish_non_exhaustive() + } +} + +impl Tracer for CountingTracer { + fn record(&self, step: TraceStep) { + self.counter + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.inner.record(step); + } +} + +impl EmitGuard { + /// Arm a new guard for the given seam name (e.g. + /// `"ingest_exited_without_trace"`). Returns the guard along with a + /// counting wrapper around `tracer` that callers should pass into + /// the request path instead of the original `tracer`. Every emit + /// against the wrapper bumps the guard's counter; if the count is + /// still zero at Drop, the guard records an `ImplBug` on the + /// original tracer. + pub fn arm( + tracer: Arc, + state: AbstractState, + kind: &'static str, + ) -> (Self, Arc) { + let counter = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let counting: Arc = Arc::new(CountingTracer { + inner: tracer.clone(), + counter: counter.clone(), + }); + let guard = Self { + inner: tracer, + state, + counter, + kind, + }; + (guard, counting) + } +} + +impl Drop for EmitGuard { + fn drop(&mut self) { + if self.counter.load(std::sync::atomic::Ordering::Relaxed) == 0 { + let step = TraceStep::new( + TraceAction::ImplBug { + kind: self.kind.to_string(), + }, + self.state.clone(), + ); + self.inner.record(step); + } + } +} + +/// Map an `IngestError` variant onto the closed `SanitizedReason` +/// alphabet (spec line 778, `Inv_SanitizedErrors`). The alphabet is +/// asserted 1:1 with the relay's error variants — if a fourth variant +/// is ever added to `IngestError` this match goes non-exhaustive and +/// CI catches it. +pub fn sanitized_reason_for(err: &crate::handlers::ingest::IngestError) -> SanitizedReason { + use crate::handlers::ingest::IngestError as E; + match err { + E::Rejected(_) => SanitizedReason::Invalid, + E::AuthFailed(_) => SanitizedReason::Restricted, + E::Internal(_) => SanitizedReason::ServerError, + } +} diff --git a/crates/buzz-relay/src/conformance/tracers.rs b/crates/buzz-relay/src/conformance/tracers.rs new file mode 100644 index 000000000..682c1714e --- /dev/null +++ b/crates/buzz-relay/src/conformance/tracers.rs @@ -0,0 +1,73 @@ +//! Concrete [`Tracer`] implementations. Production uses [`NoopTracer`]; +//! conformance tests + the CI replay job use [`JsonlTracer`]. + +use std::fs::{File, OpenOptions}; +use std::io::{BufWriter, Write}; +use std::path::Path; +use std::sync::Mutex; + +use buzz_conformance::{TraceStep, Tracer}; + +/// Zero-cost tracer used in production builds. Records nothing — the +/// emitter call still constructs the action arguments, but the build can +/// have the compiler eliminate them entirely behind a feature flag if +/// the cost ever shows up in benches. +#[derive(Debug, Default, Clone, Copy)] +pub struct NoopTracer; + +impl Tracer for NoopTracer { + fn record(&self, _step: TraceStep) {} +} + +/// JSONL-to-file tracer for tests + the CI replay job. Each `record` call +/// serializes the step as one line of JSON and appends it. The file is +/// opened in append mode so multiple test runs accumulate; consumers are +/// expected to truncate between runs. +/// +/// The internal `Mutex>` serializes writes — concurrent +/// requests producing interleaved JSONL is fine on the read side because +/// the spec doesn't model emission order, only set membership. +pub struct JsonlTracer { + out: Mutex>, +} + +impl JsonlTracer { + /// Open a new JSONL tracer writing to `path`. Truncates any existing + /// file at that path so a fresh test run starts clean. + pub fn create>(path: P) -> std::io::Result { + let file = OpenOptions::new() + .create(true) + .write(true) + .truncate(true) + .open(path)?; + Ok(Self { + out: Mutex::new(BufWriter::new(file)), + }) + } +} + +impl std::fmt::Debug for JsonlTracer { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("JsonlTracer").finish_non_exhaustive() + } +} + +impl Tracer for JsonlTracer { + fn record(&self, step: TraceStep) { + // Acquire-and-write. If the lock is poisoned we accept the panic + // — this is observability code and a poisoned lock means a worse + // bug landed elsewhere. + let mut guard = match self.out.lock() { + Ok(g) => g, + Err(e) => e.into_inner(), + }; + // Best-effort: a write failure here loses one trace step but + // must NOT take down the request path. The Drop guard's + // coverage-breach action is the safety net for systemic loss. + if let Ok(line) = serde_json::to_string(&step) { + let _ = guard.write_all(line.as_bytes()); + let _ = guard.write_all(b"\n"); + let _ = guard.flush(); + } + } +} diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 3633acd50..aa5b3a914 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -245,6 +245,12 @@ pub(crate) async fn dispatch_persistent_event( kind_u32: u32, actor_pubkey_hex: &str, ) -> usize { + // No `crate::conformance` emit here — the spec doesn't have a + // separate fan-out action. Acceptance was already recorded at the + // ingest seam (`crates/buzz-relay/src/handlers/ingest.rs`'s + // WriteInsert/WriteInsertGlobal/WriteDuplicate emit). The fan-out + // surfaces as `ReadMessageRows` observations on the subscriber side + // (read seam in req.rs, emitted by the held-back read-seam diff). let event_id_hex = stored_event.event.id.to_hex(); let topic = match stored_event.channel_id { diff --git a/crates/buzz-relay/src/handlers/ingest.rs b/crates/buzz-relay/src/handlers/ingest.rs index 7e9b95026..2e400cf32 100644 --- a/crates/buzz-relay/src/handlers/ingest.rs +++ b/crates/buzz-relay/src/handlers/ingest.rs @@ -43,6 +43,11 @@ use crate::state::AppState; use super::event::dispatch_persistent_event; +use crate::conformance::{ + self as conf, channel_label, claimed_community_from_event, emit, msg_id_label, + state_for_request, EmitGuard, TraceAction, Verdict, +}; + /// How the HTTP caller authenticated (for [`IngestAuth::Http`]). #[derive(Debug, Clone)] pub enum HttpAuthMethod { @@ -1138,11 +1143,57 @@ fn validate_event_reminder(event: &Event) -> Result<(), &'static str> { /// Shared by WebSocket and HTTP transports. The caller constructs [`IngestAuth`] /// from their transport-specific auth mechanism and maps the result to their /// transport-specific response format. +/// +/// Builds a [`crate::conformance::EmitGuard`] around the actual ingest +/// logic so the trace seam has fail-closed coverage: any exit path that +/// doesn't emit a Write*/SanitizedError action will be caught by the +/// guard's Drop → `ImplBug` → CoverageBreach. The wrapper also maps +/// `IngestError` → SanitizedError in one place, sparing every individual +/// `return Err(...)` from having to emit explicitly. See +/// `crates/buzz-relay/src/conformance/mod.rs` and +/// `docs/spec/MultiTenantRelay.tla`. pub async fn ingest_event( state: &Arc, tenant: &TenantContext, event: Event, auth: IngestAuth, +) -> Result { + let abstract_state = state_for_request(tenant, auth.pubkey()); + let (_guard, tracer) = EmitGuard::arm( + state.tracer.clone(), + abstract_state.clone(), + "ingest_event_exited_without_trace", + ); + + let result = ingest_event_inner(state, &tracer, tenant, event, auth).await; + + // Map terminal error variants onto the closed SanitizedReason + // alphabet (spec line 778). The inner fn's success path emits + // WriteInsert/WriteInsertGlobal/WriteDuplicate explicitly at its + // dispatch points — so on Ok we don't emit here. + if let Err(err) = &result { + let reason = conf::sanitized_reason_for(err); + emit( + &tracer, + TraceAction::SanitizedError { reason }, + abstract_state.clone(), + ); + } + + // _guard drops here. If `tracer` received no records during the + // request (a panic before the first emit, or a future new exit + // path that forgets to emit), Drop records an ImplBug step on + // the underlying tracer — the checker treats that as + // CoverageBreach. + result +} + +async fn ingest_event_inner( + state: &Arc, + tracer: &Arc, + tenant: &TenantContext, + event: Event, + auth: IngestAuth, ) -> Result { let event_id_hex = event.id.to_hex(); let kind_u32 = event_kind_u32(&event); @@ -1348,9 +1399,31 @@ pub async fn ingest_event( || kind_u32 == KIND_NIP29_CREATE_GROUP || auth.has_proxy_scope(); if !skip_membership { - check_channel_membership(tenant, state, ch_id, &pubkey_bytes) - .await - .map_err(IngestError::Rejected)?; + // Spec AuthCheck (line 794): emit the verdict at the actual + // call site. claimed_community comes from the event's h tag + // (recorded separately to bite M2 / M8 — claim or A-host + // driving a B-channel verdict — at the checker). The verdict + // basis is `tenant.community()` server-resolved, confirmed + // at `check_channel_membership`'s `is_member_cached(tenant + // .community(), …)` call (see crates/buzz-relay/src/handlers + // /ingest.rs:424). + let auth_result = check_channel_membership(tenant, state, ch_id, &pubkey_bytes).await; + let claimed = claimed_community_from_event(&event); + let verdict = if auth_result.is_ok() { + Verdict::Allow + } else { + Verdict::Deny + }; + emit( + tracer, + TraceAction::AuthCheck { + channel: channel_label(ch_id), + claimed_community: claimed, + verdict, + }, + state_for_request(tenant, auth.pubkey()), + ); + auth_result.map_err(IngestError::Rejected)?; } } @@ -1815,6 +1888,26 @@ pub async fn ingest_event( } let pubkey_hex = auth.pubkey().to_hex(); + // Spec WriteInsert (line 514) / WriteDuplicate (line 606): emit + // the abstract write action. The persist API returns + // `was_inserted` (true → Insert, false → Duplicate). This branch + // is the reaction path; channel_id is always Some here, so + // WriteInsertGlobal does not apply. + let claimed = claimed_community_from_event(&event); + let action = if was_inserted { + TraceAction::WriteInsert { + msg_id: msg_id_label(event.id.as_bytes()), + channel: channel_label(channel_id.expect("reaction path has channel")), + claimed_community: claimed, + } + } else { + TraceAction::WriteDuplicate { + msg_id: msg_id_label(event.id.as_bytes()), + channel: channel_label(channel_id.expect("reaction path has channel")), + claimed_community: claimed, + } + }; + emit(tracer, action, state_for_request(tenant, auth.pubkey())); dispatch_persistent_event(tenant, state, &stored_event, kind_u32, &pubkey_hex).await; info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline"); @@ -1912,6 +2005,35 @@ pub async fn ingest_event( } let pubkey_hex = auth.pubkey().to_hex(); + // Spec WriteInsert (line 514) / WriteInsertGlobal (line 559) / + // WriteDuplicate (line 606): emit the abstract write at the trailing + // dispatch site. `channel_id.is_some()` distinguishes channel-bearing + // (Insert/Duplicate) from channel-less (InsertGlobal); `was_inserted` + // distinguishes accepted-new (Insert/Global) from no-op-on-conflict + // (Duplicate). The WriteInsertGlobal duplicate case is not modeled + // separately in the spec (channel-less duplicates collapse to the + // same observation shape as channel-less inserts at this seam); + // see docs/spec/MultiTenantRelay.tla lines 559-595. + { + let claimed = claimed_community_from_event(&event); + let action = match (channel_id, was_inserted) { + (Some(ch), true) => TraceAction::WriteInsert { + msg_id: msg_id_label(event.id.as_bytes()), + channel: channel_label(ch), + claimed_community: claimed, + }, + (Some(ch), false) => TraceAction::WriteDuplicate { + msg_id: msg_id_label(event.id.as_bytes()), + channel: channel_label(ch), + claimed_community: claimed, + }, + (None, _) => TraceAction::WriteInsertGlobal { + msg_id: msg_id_label(event.id.as_bytes()), + claimed_community: claimed, + }, + }; + emit(tracer, action, state_for_request(tenant, auth.pubkey())); + } dispatch_persistent_event(tenant, state, &stored_event, kind_u32, &pubkey_hex).await; info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline"); diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 9a7aa50f8..17d56dc2f 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -8,6 +8,11 @@ pub mod api; pub mod audio; /// Relay configuration from environment variables. pub mod config; +/// Runtime conformance harness — abstract trace emission at the +/// ingest/read accept-reject boundary, replayed against +/// `docs/spec/MultiTenantRelay.tla` by the independent `buzz-conformance` +/// checker. +pub mod conformance; /// WebSocket connection lifecycle and state. pub mod connection; /// Relay error types. diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 08c13e843..c9a4559bd 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -278,6 +278,13 @@ pub struct AppState { /// Prevents repeated DB lookups from bursty observer traffic. #[allow(clippy::type_complexity)] pub observer_owner_cache: Arc, Vec), bool>>, + + /// Runtime conformance tracer. Production binds [`crate::conformance::NoopTracer`] + /// (zero cost). Conformance tests bind [`crate::conformance::JsonlTracer`] to + /// record traces for replay against `docs/spec/MultiTenantRelay.tla`. + /// See `crates/buzz-conformance/` and `crate::conformance` for the + /// schema, emitter helpers, and the independent checker. + pub tracer: Arc, } impl AppState { @@ -406,6 +413,11 @@ impl AppState { .time_to_live(std::time::Duration::from_secs(300)) .build(), ), + // Default to NoopTracer: production builds pay zero cost. + // Conformance tests overwrite this with a JsonlTracer after + // construction (see test helpers in + // `crates/buzz-test-client` once those land). + tracer: Arc::new(crate::conformance::NoopTracer), }; ( state,