From 48d45cf7f40536e440345de70b57d1f2f57c45f5 Mon Sep 17 00:00:00 2001 From: npub1x3mmseqygyar04742djuepgk0t2d2t4chzm9sl0hl4vlc2m9whvqza9e5y <3477b86404413a37d7d55365cc85167ad4d52eb8b8b6587df7fd59fc2b6575d8@buzz.block.builderlab.xyz> Date: Mon, 3 Aug 2026 09:48:51 -0500 Subject: [PATCH] feat(relay): report client app, platform, and version for live connections The relay could not tell which application, platform, or version its connections came from. `buzz_ws_connections_active` carries no labels and `buzz_ws_connections_total` carries only `community`, so questions like "how many macOS desktop users are on 0.4?" or "is anyone still on the old CLI?" were unanswerable, and no client sent identifying handshake metadata for the relay to read. Rust clients now announce themselves with an advisory RFC 8941 `Buzz-Client` header, and the relay turns it into labeled metrics: Buzz-Client: v=1, app=buzz-desktop, platform=macos, app-version="0.5.2" buzz_client_connections_total{app,platform,app_version} buzz_client_connections_active{app,platform,app_version} buzz_client_header_parse_failures_total{reason} Senders are wired for desktop, buzz-ws-client (CLI and test client), and the buzz-acp harness. The vocabularies and serializer live in buzz-core so all of them share one definition. The live gauge is deliberately a new series rather than labels on `buzz_ws_connections_active`: that gauge is the HPA scaling input (`deploy/charts/buzz/values.yaml` `websocketMetricName`, an unlabeled `AverageValue` target), and labeling it would shard the series the autoscaler reads. `deploy/` is untouched and the chart's HPA unit tests still pass. Retired label sets are dropped by the recorder's existing gauge idle timeout. Complements #2596, which adds the mobile sender and its own relay-side parser. This change is based on main rather than on that branch, and the parser ignores unknown dictionary keys, so it already accepts mobile's richer header (`app-build`, `os-version`, `os-api`) unchanged. Widening the allowlists to desktop/CLI/ACP and macOS/Windows/Linux, adding the live gauge, and bucketing unidentified connections as `unknown` so the totals reconcile are the gaps this closes. Trust and cardinality: the header arrives before NIP-42 AUTH and is forgeable, so it is advisory only and never touches authentication, authorization, tenant selection, or admission. `app` and `platform` resolve to `&'static str` from closed allowlists and `app_version` is narrowed to bounded MAJOR.MINOR, so a forged header cannot introduce an unbounded label value. Clients send the header only over TLS or to loopback, so it is not exposed to on-path observers on cleartext connections. Signed-off-by: npub1x3mmseqygyar04742djuepgk0t2d2t4chzm9sl0hl4vlc2m9whvqza9e5y <3477b86404413a37d7d55365cc85167ad4d52eb8b8b6587df7fd59fc2b6575d8@buzz.block.builderlab.xyz> Co-authored-by: Atish Patel Signed-off-by: Atish Patel --- Cargo.lock | 1 + crates/buzz-acp/src/relay.rs | 25 +- crates/buzz-core/src/client_identity.rs | 445 +++++++++++++ crates/buzz-core/src/lib.rs | 2 + crates/buzz-relay/src/client_info.rs | 737 ++++++++++++++++++++++ crates/buzz-relay/src/connection.rs | 26 +- crates/buzz-relay/src/lib.rs | 2 + crates/buzz-relay/src/router.rs | 6 +- crates/buzz-ws-client/Cargo.toml | 1 + crates/buzz-ws-client/src/connection.rs | 49 +- desktop/src-tauri/src/native_websocket.rs | 26 +- 11 files changed, 1314 insertions(+), 6 deletions(-) create mode 100644 crates/buzz-core/src/client_identity.rs create mode 100644 crates/buzz-relay/src/client_info.rs diff --git a/Cargo.lock b/Cargo.lock index eaea5b35a8..adba73b665 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1444,6 +1444,7 @@ dependencies = [ name = "buzz-ws-client" version = "0.1.0" dependencies = [ + "buzz-core", "futures-util", "nostr 0.44.7", "serde_json", diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 17a818867d..c87e25b228 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -116,6 +116,9 @@ const GATED_OBSERVER_QUEUE_CAP: usize = 256; use std::time::Instant; +use buzz_core::client_identity::{ + client_header_value_for_host, may_identify_to, ClientApp, CLIENT_HEADER, +}; use buzz_core::kind::{ KIND_AGENT_OBSERVER_FRAME, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_TYPING_INDICATOR, @@ -125,6 +128,7 @@ use nostr::{Event, EventBuilder, Keys, Kind, RelayUrl, Tag}; use serde_json::{json, Value}; use tokio::sync::mpsc; use tokio::time::timeout; +use tokio_tungstenite::tungstenite::client::ClientRequestBuilder; use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; use tracing::{debug, info, warn}; use uuid::Uuid; @@ -3862,7 +3866,26 @@ async fn do_connect( .parse::() .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?; - let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(parsed.as_str())) + // Advisory `Buzz-Client` identity so the relay can attribute this + // connection to the harness and its version. Best-effort: an unshipped + // platform or a destination we must not identify to simply connects + // without the header. + let request = { + let uri = parsed + .as_str() + .parse() + .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?; + let builder = ClientRequestBuilder::new(uri); + match may_identify_to(parsed.as_str()) + .then(|| client_header_value_for_host(ClientApp::Acp, env!("CARGO_PKG_VERSION"))) + .flatten() + { + Some(value) => builder.with_header(CLIENT_HEADER, value), + None => builder, + } + }; + + let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(request)) .await .map_err(|_| RelayError::ConnectionClosed)? // timeout → treat as connection failure .map_err(|e| RelayError::WebSocket(Box::new(e)))?; diff --git a/crates/buzz-core/src/client_identity.rs b/crates/buzz-core/src/client_identity.rs new file mode 100644 index 0000000000..f185d20a32 --- /dev/null +++ b/crates/buzz-core/src/client_identity.rs @@ -0,0 +1,445 @@ +//! The `Buzz-Client` advisory identity header, shared by every Rust client. +//! +//! Buzz-owned servers cannot otherwise tell which application, platform, or +//! version a connection came from, which makes support triage, rollout +//! monitoring, and deprecation decisions guesswork. Each client sends a +//! [RFC 8941](https://www.rfc-editor.org/rfc/rfc8941) structured dictionary: +//! +//! ```text +//! Buzz-Client: v=1, app=buzz-desktop, platform=macos, app-version="0.5.2" +//! ``` +//! +//! This module owns only the *sending* half: the canonical header name, the +//! app/platform vocabularies, and the serializer. The relay parses the header +//! with its own lenient reader (`buzz-relay`'s `client_info`) and treats it as +//! advisory — it never participates in authentication, authorization, or +//! tenant selection. +//! +//! # Privacy +//! +//! The field set is deliberately minimal: application, platform, and a +//! user-visible version. No device model, install identifier, timezone, +//! network type, or raw bundle ID. The header is sent only to the relay a +//! client is already configured to talk to, never to third-party origins, and +//! is never embedded in signed events. + +use std::fmt::Write as _; + +/// Canonical header name. Lowercase so it can be used directly as an HTTP/2 +/// field name and compared against `http::HeaderName` without reallocating. +pub const CLIENT_HEADER: &str = "buzz-client"; + +/// Structured-dictionary format version. Bump only on a breaking field change; +/// the relay rejects any other value rather than guessing. +pub const CLIENT_HEADER_FORMAT_VERSION: i64 = 1; + +/// The application sending a request, as an RFC 8941 token. +/// +/// A closed vocabulary is the cardinality guard: these values become +/// Prometheus label values on the relay, so they must never derive from +/// free-form input. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ClientApp { + /// The Tauri desktop application. + Desktop, + /// The Flutter mobile application. + Mobile, + /// The `buzz` command-line interface. + Cli, + /// The agent harness that drives ACP runtimes. + Acp, +} + +impl ClientApp { + /// The RFC 8941 token for this application. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Desktop => "buzz-desktop", + Self::Mobile => "buzz-mobile", + Self::Cli => "buzz-cli", + Self::Acp => "buzz-acp", + } + } + + /// Every known application token, for the relay's parse allowlist. + #[must_use] + pub const fn all() -> &'static [&'static str] { + &["buzz-desktop", "buzz-mobile", "buzz-cli", "buzz-acp"] + } +} + +/// The operating-system family a client is running on, as an RFC 8941 token. +/// +/// Deliberately coarse — the family only, never a kernel or build string. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ClientPlatform { + /// Apple macOS. + MacOs, + /// Microsoft Windows. + Windows, + /// Linux, including the AppImage and `.deb` desktop builds. + Linux, + /// Apple iOS and iPadOS. + Ios, + /// Android. + Android, +} + +impl ClientPlatform { + /// The RFC 8941 token for this platform. + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::MacOs => "macos", + Self::Windows => "windows", + Self::Linux => "linux", + Self::Ios => "ios", + Self::Android => "android", + } + } + + /// Every known platform token, for the relay's parse allowlist. + #[must_use] + pub const fn all() -> &'static [&'static str] { + &["macos", "windows", "linux", "ios", "android"] + } + + /// The platform this binary was compiled for, or `None` on a target Buzz + /// does not ship. + /// + /// Returning `None` rather than an `"other"` token keeps the vocabulary + /// closed: an unshipped target simply sends no header and is counted as + /// unidentified, instead of inventing a label value. + #[must_use] + pub fn current() -> Option { + // `std::env::consts::OS` is fixed per compile target. Not a `const fn` + // because `str` cannot be matched in const context. + match std::env::consts::OS { + "macos" => Some(Self::MacOs), + "windows" => Some(Self::Windows), + "linux" => Some(Self::Linux), + "ios" => Some(Self::Ios), + "android" => Some(Self::Android), + _ => None, + } + } +} + +/// Maximum serialized header length. A Buzz-generated value is far shorter; +/// this only bounds a pathological `app_version` before it reaches the wire. +const MAX_HEADER_LEN: usize = 256; + +/// Longest accepted `app-version`, matching the relay's own bound. +const MAX_APP_VERSION_LEN: usize = 32; + +/// Build the `Buzz-Client` header value for this client. +/// +/// Returns `None` when the platform is not one Buzz ships, or when +/// `app_version` is empty or not a plausible version string. A missing header +/// is a supported state on the relay, so refusing to send is always safe — +/// callers must never substitute a placeholder. +/// +/// # Examples +/// +/// ``` +/// use buzz_core::client_identity::{client_header_value, ClientApp, ClientPlatform}; +/// +/// let value = client_header_value(ClientApp::Desktop, ClientPlatform::MacOs, "0.5.2").unwrap(); +/// assert_eq!(value, r#"v=1, app=buzz-desktop, platform=macos, app-version="0.5.2""#); +/// ``` +#[must_use] +pub fn client_header_value( + app: ClientApp, + platform: ClientPlatform, + app_version: &str, +) -> Option { + // Only horizontal whitespace is trimmed. Trimming CR/LF would silently + // sanitize a header-injection attempt into an accepted value; those are + // rejected by `is_serializable_app_version` instead. + let app_version = app_version.trim_matches([' ', '\t']); + if !is_serializable_app_version(app_version) { + return None; + } + + let mut value = String::with_capacity(64); + // Field order matches the wire example and the relay's tests; RFC 8941 + // dictionaries are order-independent, so this is presentation only. + let _ = write!( + value, + "v={CLIENT_HEADER_FORMAT_VERSION}, app={}, platform={}, app-version=\"{app_version}\"", + app.as_str(), + platform.as_str(), + ); + + // Unreachable for a validated version, but never emit an oversized header. + if value.len() > MAX_HEADER_LEN { + return None; + } + Some(value) +} + +/// Build the header for the current compile target, or `None` on an unshipped +/// platform. +/// +/// This is the entry point clients should use; it removes the chance of a +/// caller hardcoding the wrong platform for a build. +#[must_use] +pub fn client_header_value_for_host(app: ClientApp, app_version: &str) -> Option { + client_header_value(app, ClientPlatform::current()?, app_version) +} + +/// Whether `url` is a destination this client may identify itself to. +/// +/// The header names the application, platform, and version of the software a +/// user is running, so it is sent only to the relay the client is already +/// configured to talk to. A relay URL is operator-supplied and can point +/// anywhere, so `ws://`/`wss://` alone is not sufficient: a plaintext `ws://` +/// origin would leak the header to any on-path observer. Loopback is exempted +/// so local development still exercises the same code path. +/// +/// # Examples +/// +/// ``` +/// use buzz_core::client_identity::may_identify_to; +/// +/// assert!(may_identify_to("wss://buzz.example.com/")); +/// assert!(may_identify_to("ws://127.0.0.1:8080/")); +/// assert!(!may_identify_to("ws://buzz.example.com/")); +/// ``` +#[must_use] +pub fn may_identify_to(url: &str) -> bool { + let Some((scheme, rest)) = url.split_once("://") else { + return false; + }; + match scheme.to_ascii_lowercase().as_str() { + // TLS: the header is only visible to the relay itself. + "wss" | "https" => true, + // Cleartext is acceptable only when it cannot leave the machine. + "ws" | "http" => is_loopback_authority(rest), + _ => false, + } +} + +/// Whether an authority (`host[:port]`, possibly followed by a path) is +/// loopback. +fn is_loopback_authority(rest: &str) -> bool { + // Trim the path/query/fragment, then any `userinfo@` prefix. + let authority = rest.split(['/', '?', '#']).next().unwrap_or(""); + let authority = authority + .rsplit_once('@') + .map_or(authority, |(_userinfo, host)| host); + let host = match authority.strip_prefix('[') { + // IPv6 literal: `[::1]:port`. + Some(inner) => inner.split(']').next().unwrap_or(""), + None => authority.split(':').next().unwrap_or(""), + }; + if host.eq_ignore_ascii_case("localhost") { + return true; + } + // Parse as an address rather than prefix-matching "127.": a name like + // `127.0.0.1.example.com` shares the prefix but is a remote host. + host.parse::() + .is_ok_and(|ip| ip.is_loopback()) +} + +/// Whether `app_version` is safe to place inside an RFC 8941 quoted string. +/// +/// RFC 8941 quoted strings admit only printable ASCII, and `"`/`\` would need +/// escaping. Rather than escape, reject: a Buzz version is always +/// dot-separated alphanumerics with optional `-`/`+` pre-release and build +/// metadata, so anything else means the caller passed something unexpected. +fn is_serializable_app_version(app_version: &str) -> bool { + if app_version.is_empty() || app_version.len() > MAX_APP_VERSION_LEN { + return false; + } + // Must start with a digit so a placeholder like "unknown" or "dev" is + // refused rather than published as a version label. + if !app_version.starts_with(|c: char| c.is_ascii_digit()) { + return false; + } + app_version + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-' | b'+')) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn serializes_the_documented_wire_format() { + let value = + client_header_value(ClientApp::Desktop, ClientPlatform::MacOs, "0.5.2").unwrap(); + assert_eq!( + value, + r#"v=1, app=buzz-desktop, platform=macos, app-version="0.5.2""# + ); + } + + #[test] + fn app_and_platform_tokens_are_stable() { + // These strings are Prometheus label values and a cross-repo wire + // contract with the mobile client; changing one is a breaking change. + assert_eq!(ClientApp::Desktop.as_str(), "buzz-desktop"); + assert_eq!(ClientApp::Mobile.as_str(), "buzz-mobile"); + assert_eq!(ClientApp::Cli.as_str(), "buzz-cli"); + assert_eq!(ClientApp::Acp.as_str(), "buzz-acp"); + assert_eq!(ClientPlatform::MacOs.as_str(), "macos"); + assert_eq!(ClientPlatform::Windows.as_str(), "windows"); + assert_eq!(ClientPlatform::Linux.as_str(), "linux"); + assert_eq!(ClientPlatform::Ios.as_str(), "ios"); + assert_eq!(ClientPlatform::Android.as_str(), "android"); + } + + #[test] + fn allowlists_cover_every_enum_variant() { + // The relay allowlists are written as string slices; keep them in sync + // with the enums so a new variant cannot be silently unparseable. + for app in [ + ClientApp::Desktop, + ClientApp::Mobile, + ClientApp::Cli, + ClientApp::Acp, + ] { + assert!(ClientApp::all().contains(&app.as_str()), "{app:?}"); + } + assert_eq!(ClientApp::all().len(), 4); + for platform in [ + ClientPlatform::MacOs, + ClientPlatform::Windows, + ClientPlatform::Linux, + ClientPlatform::Ios, + ClientPlatform::Android, + ] { + assert!( + ClientPlatform::all().contains(&platform.as_str()), + "{platform:?}" + ); + } + assert_eq!(ClientPlatform::all().len(), 5); + } + + #[test] + fn rejects_versions_that_are_not_versions() { + for bad in ["", " ", "unknown", "dev", "v1.2.3", "nightly"] { + assert!( + client_header_value(ClientApp::Cli, ClientPlatform::Linux, bad).is_none(), + "expected {bad:?} to be refused" + ); + } + } + + #[test] + fn rejects_versions_that_would_break_the_quoted_string() { + // A quote or backslash would need RFC 8941 escaping; a newline would + // allow header injection. All must be refused, never escaped. + for bad in [ + "0.5\"2", + "0.5\\2", + "0.5.2\r\nX-Evil: 1", + "0.5.2\n", + "0.5.2 extra", + "0.5.2\u{7f}", + "0.5.2é", + ] { + assert!( + client_header_value(ClientApp::Desktop, ClientPlatform::MacOs, bad).is_none(), + "expected {bad:?} to be refused" + ); + } + } + + #[test] + fn rejects_an_absurdly_long_version() { + let long = format!("0.{}", "9".repeat(MAX_APP_VERSION_LEN)); + assert!(long.len() > MAX_APP_VERSION_LEN); + assert!(client_header_value(ClientApp::Cli, ClientPlatform::Linux, &long).is_none()); + } + + #[test] + fn accepts_prerelease_and_build_metadata() { + let value = client_header_value(ClientApp::Cli, ClientPlatform::Linux, "1.2.3-rc.1+build9") + .unwrap(); + assert!(value.ends_with(r#"app-version="1.2.3-rc.1+build9""#)); + } + + #[test] + fn trims_surrounding_spaces_and_tabs_before_validating() { + let value = client_header_value(ClientApp::Desktop, ClientPlatform::Windows, " \t0.5.2\t ") + .unwrap(); + assert_eq!( + value, + r#"v=1, app=buzz-desktop, platform=windows, app-version="0.5.2""# + ); + } + + #[test] + fn identifies_only_to_tls_or_loopback_origins() { + for allowed in [ + "wss://buzz.example.com/", + "wss://buzz.example.com:8443/relay", + "WSS://BUZZ.EXAMPLE.COM/", + "https://buzz.example.com/", + // Cleartext loopback cannot leave the machine. + "ws://localhost:8080/", + "ws://127.0.0.1:8080/", + "ws://127.3.2.1/", + "ws://[::1]:8080/", + ] { + assert!(may_identify_to(allowed), "expected {allowed:?} allowed"); + } + } + + #[test] + fn refuses_to_identify_over_cleartext_to_a_remote_host() { + for refused in [ + // The header would be readable by any on-path observer. + "ws://buzz.example.com/", + "http://buzz.example.com/", + "ws://192.168.1.5:8080/", + "ws://10.0.0.1/", + // `127.0.0.1.example.com` is a remote host, not loopback. + "ws://127.0.0.1.example.com/", + // Loopback in userinfo must not fool the host check. + "ws://localhost@evil.example.com/", + // Non-WebSocket and malformed destinations. + "file:///etc/passwd", + "buzz.example.com", + "", + ] { + assert!(!may_identify_to(refused), "expected {refused:?} refused"); + } + } + + #[test] + fn host_helper_agrees_with_the_explicit_platform_on_shipped_targets() { + // Tests only run on targets Buzz ships, so `current()` is `Some` here. + let platform = ClientPlatform::current().expect("test host is a shipped platform"); + assert_eq!( + client_header_value_for_host(ClientApp::Cli, "0.1.0"), + client_header_value(ClientApp::Cli, platform, "0.1.0") + ); + } + + #[test] + fn generated_headers_stay_within_the_length_bound() { + for app in [ + ClientApp::Desktop, + ClientApp::Mobile, + ClientApp::Cli, + ClientApp::Acp, + ] { + for platform in [ + ClientPlatform::MacOs, + ClientPlatform::Windows, + ClientPlatform::Linux, + ClientPlatform::Ios, + ClientPlatform::Android, + ] { + let value = client_header_value(app, platform, "10.20.30-rc.1").unwrap(); + assert!(value.len() <= MAX_HEADER_LEN, "{value}"); + } + } + } +} diff --git a/crates/buzz-core/src/lib.rs b/crates/buzz-core/src/lib.rs index 7424915c83..45e5d72800 100644 --- a/crates/buzz-core/src/lib.rs +++ b/crates/buzz-core/src/lib.rs @@ -9,6 +9,8 @@ pub mod agent_turn_metric; /// Channel and membership enums shared across crates. pub mod channel; +/// The `Buzz-Client` advisory identity header sent by every Rust client. +pub mod client_identity; /// NIP-AE Agent Engrams — slug grammar, conversation key, d-tag derivation, /// body parse/serialize, envelope build/validate, head selection. pub mod engram; diff --git a/crates/buzz-relay/src/client_info.rs b/crates/buzz-relay/src/client_info.rs new file mode 100644 index 0000000000..3007e5d5e0 --- /dev/null +++ b/crates/buzz-relay/src/client_info.rs @@ -0,0 +1,737 @@ +//! Advisory parsing of the `Buzz-Client` identity header. +//! +//! Clients announce themselves with an [RFC 8941](https://www.rfc-editor.org/rfc/rfc8941) +//! structured dictionary so the relay can report which application, platform, +//! and version its live connections come from: +//! +//! ```text +//! Buzz-Client: v=1, app=buzz-desktop, platform=macos, app-version="0.5.2" +//! ``` +//! +//! # This input is never trusted +//! +//! The header is unauthenticated: it arrives before NIP-42 AUTH and anyone can +//! forge it. It is therefore **advisory only** — it never participates in +//! authentication, authorization, tenant selection, rate limiting, or any +//! other decision. It feeds metrics and logs, nothing else. A missing header +//! is a normal, supported state; a malformed one is counted and discarded. No +//! input on this path can ever cause a connection to be rejected. +//! +//! # Cardinality is bounded by construction +//! +//! Label values are the guard against a forged header exploding Prometheus +//! series. `app` and `platform` are resolved to `&'static str` from closed +//! allowlists — an unrecognized token yields no label, never a passthrough of +//! attacker-controlled bytes. `app_version` is narrowed to `MAJOR.MINOR` with +//! both components bounded in length, so the worst case is +//! `apps × platforms × plausible versions`. +//! +//! # Why a hand-written parser +//! +//! Only the tiny dictionary subset above is accepted, so a dedicated strict +//! reader is easier to audit than a general RFC 8941 implementation and costs +//! the relay no new dependency. Unknown keys are ignored rather than rejected, +//! which is what lets clients add fields (mobile already sends `app-build`, +//! `os-version`, and `os-api`) without a coordinated relay deploy. + +use axum::http::HeaderMap; +use buzz_core::client_identity::{ + ClientApp, ClientPlatform, CLIENT_HEADER, CLIENT_HEADER_FORMAT_VERSION, +}; + +/// Label value for connections with no usable `Buzz-Client` header. +/// +/// Emitting an explicit bucket rather than omitting the series is what lets +/// `sum(buzz_client_connections_total)` reconcile with +/// `sum(buzz_ws_connections_total)`, and puts "unidentified" on a dashboard as +/// a visible line instead of a silent gap. +pub const UNKNOWN_LABEL: &str = "unknown"; + +/// Longest `Buzz-Client` header the relay will parse, in bytes. +/// +/// A Buzz-generated value is well under 128 bytes. This bounds parser work on +/// a hostile input before any allocation. +const MAX_HEADER_LEN: usize = 512; + +/// Longest accepted `MAJOR` or `MINOR` component of `app-version`. +/// +/// Five digits admits every plausible version while capping the label +/// alphabet at 10^5 values per component. +const MAX_VERSION_COMPONENT_LEN: usize = 5; + +/// Longest raw value retained for logging. +/// +/// Log fields are not Prometheus labels, so they need no allowlist — only a +/// length bound so a hostile header cannot bloat a log line. +const MAX_LOGGED_VALUE_LEN: usize = 32; + +/// Why a present `Buzz-Client` header could not be used. +/// +/// Rendered as a bounded `reason` label on +/// `buzz_client_header_parse_failures_total`; every variant is a fixed string. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum ParseFailure { + /// Not valid ASCII, or longer than [`MAX_HEADER_LEN`]. + Unreadable, + /// Not a well-formed dictionary of the accepted subset. + Malformed, + /// `v` was absent or not [`CLIENT_HEADER_FORMAT_VERSION`]. + UnsupportedVersion, + /// `app` was absent or outside the allowlist. + UnknownApp, + /// `platform` was absent or outside the allowlist. + UnknownPlatform, + /// `app-version` was absent or not a bounded `MAJOR.MINOR[...]`. + BadAppVersion, +} + +impl ParseFailure { + /// The fixed `reason` label for this failure. + const fn as_str(self) -> &'static str { + match self { + Self::Unreadable => "unreadable", + Self::Malformed => "malformed", + Self::UnsupportedVersion => "unsupported_version", + Self::UnknownApp => "unknown_app", + Self::UnknownPlatform => "unknown_platform", + Self::BadAppVersion => "bad_app_version", + } + } +} + +/// A validated, allowlisted client identity. +/// +/// Construction guarantees every field is safe to use as a metric label or log +/// field. There is no way to build one holding unvalidated input. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ClientInfo { + /// Allowlisted application token. + app: &'static str, + /// Allowlisted platform token. + platform: &'static str, + /// `MAJOR.MINOR`, safe as a metric label. + app_version: String, + /// Exact version as sent, for logs only — never a label. + app_version_detail: String, +} + +impl ClientInfo { + /// Parse the `Buzz-Client` header, recording a failure metric when a + /// header is present but unusable. + /// + /// Returns `None` both when no header was sent (silently — that is the + /// common case for older clients and the web app) and when parsing failed + /// (counted). Callers treat `None` as [`UNKNOWN_LABEL`]. + #[must_use] + pub fn from_headers(headers: &HeaderMap) -> Option { + // No header at all is normal and deliberately not counted as a + // failure: it would otherwise fire continuously for every client that + // predates this feature. + let raw = headers.get(CLIENT_HEADER)?; + + match Self::parse_bytes(raw.as_bytes()) { + Ok(info) => Some(info), + Err(failure) => { + metrics::counter!( + "buzz_client_header_parse_failures_total", + "reason" => failure.as_str() + ) + .increment(1); + None + } + } + } + + /// Parse a raw header value. + fn parse_bytes(raw: &[u8]) -> Result { + if raw.len() > MAX_HEADER_LEN { + return Err(ParseFailure::Unreadable); + } + let raw = std::str::from_utf8(raw).map_err(|_| ParseFailure::Unreadable)?; + if !raw.is_ascii() { + return Err(ParseFailure::Unreadable); + } + Self::parse(raw) + } + + /// Parse an ASCII header value of the accepted dictionary subset. + fn parse(raw: &str) -> Result { + let mut format_version: Option = None; + let mut app: Option<&str> = None; + let mut platform: Option<&str> = None; + let mut app_version: Option<&str> = None; + + for member in split_members(raw)? { + let (key, value) = split_member(member)?; + // Unknown keys are ignored on purpose: it lets clients add fields + // (mobile's `app-build` / `os-version` / `os-api`) without this + // relay having to know about them first. + match key { + "v" => format_version = Some(value.as_integer().ok_or(ParseFailure::Malformed)?), + "app" => app = Some(value.as_token().ok_or(ParseFailure::Malformed)?), + "platform" => platform = Some(value.as_token().ok_or(ParseFailure::Malformed)?), + "app-version" => { + app_version = Some(value.as_string().ok_or(ParseFailure::Malformed)?); + } + _ => {} + } + } + + if format_version != Some(CLIENT_HEADER_FORMAT_VERSION) { + return Err(ParseFailure::UnsupportedVersion); + } + + // Resolve to the allowlist's own `&'static str`, so the label value can + // never be attacker-controlled bytes even if they compared equal. + let app = app + .and_then(|candidate| ClientApp::all().iter().copied().find(|a| *a == candidate)) + .ok_or(ParseFailure::UnknownApp)?; + let platform = platform + .and_then(|candidate| { + ClientPlatform::all() + .iter() + .copied() + .find(|p| *p == candidate) + }) + .ok_or(ParseFailure::UnknownPlatform)?; + + let app_version_raw = app_version.ok_or(ParseFailure::BadAppVersion)?; + let app_version = major_minor(app_version_raw).ok_or(ParseFailure::BadAppVersion)?; + + Ok(Self { + app, + platform, + app_version, + app_version_detail: truncate_for_log(app_version_raw), + }) + } + + /// Allowlisted application token. + #[must_use] + pub const fn app(&self) -> &'static str { + self.app + } + + /// Allowlisted platform token. + #[must_use] + pub const fn platform(&self) -> &'static str { + self.platform + } + + /// `MAJOR.MINOR` version, safe as a metric label. + #[must_use] + pub fn app_version(&self) -> &str { + &self.app_version + } + + /// Exact version as sent, for logs only. + #[must_use] + pub fn app_version_detail(&self) -> &str { + &self.app_version_detail + } +} + +/// The three label values for a connection, using [`UNKNOWN_LABEL`] when the +/// client did not identify itself. +fn labels(info: Option<&ClientInfo>) -> (&str, &str, &str) { + match info { + Some(info) => (info.app, info.platform, info.app_version.as_str()), + None => (UNKNOWN_LABEL, UNKNOWN_LABEL, UNKNOWN_LABEL), + } +} + +/// Count a newly established connection by client identity. +pub fn record_connection(info: Option<&ClientInfo>) { + let (app, platform, app_version) = labels(info); + metrics::counter!( + "buzz_client_connections_total", + "app" => app.to_owned(), + "platform" => platform.to_owned(), + "app_version" => app_version.to_owned() + ) + .increment(1); +} + +/// Add a live connection to the per-client active gauge. +/// +/// This is a **separate** series from `buzz_ws_connections_active`, which is +/// consumed by the HPA as an unlabeled `AverageValue` target. Labeling that +/// gauge would shard the series the autoscaler reads; this parallel gauge +/// gives the same breakdown without touching scaling behaviour. +pub fn increment_active(info: Option<&ClientInfo>) { + let (app, platform, app_version) = labels(info); + metrics::gauge!( + "buzz_client_connections_active", + "app" => app.to_owned(), + "platform" => platform.to_owned(), + "app_version" => app_version.to_owned() + ) + .increment(1.0); +} + +/// Remove a closed connection from the per-client active gauge. +/// +/// Must be paired with exactly one [`increment_active`] call, or the gauge +/// drifts. Retired label sets are dropped by the recorder's configured gauge +/// idle timeout rather than going stale. +pub fn decrement_active(info: Option<&ClientInfo>) { + let (app, platform, app_version) = labels(info); + metrics::gauge!( + "buzz_client_connections_active", + "app" => app.to_owned(), + "platform" => platform.to_owned(), + "app_version" => app_version.to_owned() + ) + .decrement(1.0); +} + +/// A dictionary member's value, still in its wire form. +enum Value<'a> { + /// An unquoted token, e.g. `buzz-desktop`. + Token(&'a str), + /// The contents of a quoted string, e.g. `0.5.2` from `"0.5.2"`. + Quoted(&'a str), +} + +impl<'a> Value<'a> { + /// The value as an RFC 8941 token, or `None` if it was quoted or not a + /// valid token. + fn as_token(&self) -> Option<&'a str> { + match self { + Self::Token(token) if is_token(token) => Some(token), + _ => None, + } + } + + /// The value as a quoted string's contents, or `None` if it was a token. + fn as_string(&self) -> Option<&'a str> { + match self { + Self::Quoted(text) => Some(text), + Self::Token(_) => None, + } + } + + /// The value as an integer, or `None` if it was quoted or not an integer. + fn as_integer(&self) -> Option { + match self { + Self::Token(token) => token.parse().ok(), + Self::Quoted(_) => None, + } + } +} + +/// Split a dictionary into members on commas outside quoted strings. +/// +/// Splitting naively on `,` would let `app-version="1,2"` smuggle a member +/// boundary, so quotes are tracked. +fn split_members(raw: &str) -> Result, ParseFailure> { + let mut members = Vec::new(); + let mut in_quotes = false; + let mut escaped = false; + let mut start = 0; + + for (idx, ch) in raw.char_indices() { + if escaped { + escaped = false; + continue; + } + match ch { + '\\' if in_quotes => escaped = true, + '"' => in_quotes = !in_quotes, + // Control characters, including the CR/LF of a header-injection + // attempt, are never valid. Tab is exempt: it is legal whitespace + // between members and is stripped by `split_member`. + c if c.is_ascii_control() && c != '\t' => return Err(ParseFailure::Malformed), + ',' if !in_quotes => { + members.push(&raw[start..idx]); + start = idx + 1; + } + _ => {} + } + } + + if in_quotes || escaped { + return Err(ParseFailure::Malformed); + } + members.push(&raw[start..]); + Ok(members) +} + +/// Split one member into its key and value. +/// +/// RFC 8941 parameters (`;k=v`) are discarded: nothing in this vocabulary uses +/// them, and ignoring them keeps an unknown-parameter-bearing member usable +/// instead of failing the whole header. +fn split_member(member: &str) -> Result<(&str, Value<'_>), ParseFailure> { + let member = member.trim_matches([' ', '\t']); + let (key, raw_value) = member.split_once('=').ok_or(ParseFailure::Malformed)?; + let key = key.trim_matches([' ', '\t']); + if !is_token(key) { + return Err(ParseFailure::Malformed); + } + + let raw_value = raw_value.trim_matches([' ', '\t']); + if let Some(rest) = raw_value.strip_prefix('"') { + // Parameters may follow the closing quote; the string ends at it. + let end = rest.find('"').ok_or(ParseFailure::Malformed)?; + let text = &rest[..end]; + if text.contains('\\') { + // Buzz never emits escapes, so refuse rather than implement + // unescaping that would only ever process hostile input. + return Err(ParseFailure::Malformed); + } + Ok((key, Value::Quoted(text))) + } else { + let value = raw_value + .split(';') + .next() + .unwrap_or("") + .trim_matches([' ', '\t']); + if value.is_empty() { + return Err(ParseFailure::Malformed); + } + Ok((key, Value::Token(value))) + } +} + +/// Whether `candidate` is a plausible RFC 8941 key or token. +/// +/// Intentionally narrower than the RFC: this vocabulary only uses lowercase +/// alphanumerics with `-`, `_`, `.`, and `*`. +fn is_token(candidate: &str) -> bool { + !candidate.is_empty() + && candidate + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'*')) +} + +/// Narrow a version to `MAJOR.MINOR`, or `None` if it is not one. +/// +/// Trailing components and pre-release/build metadata are discarded so patch +/// releases do not each create a new time series. The *whole* string is still +/// validated first: the discarded tail is retained for logging, so leaving it +/// unchecked would let a hostile client put arbitrary text in a log field even +/// though the metric label stayed bounded. +fn major_minor(version: &str) -> Option { + if !is_plausible_version(version) { + return None; + } + let mut parts = version.split('.'); + let major = numeric_component(parts.next()?)?; + let minor = numeric_component(parts.next()?)?; + Some(format!("{major}.{minor}")) +} + +/// Whether the full version string looks like a version Buzz produced. +/// +/// Matches the sender-side rule in `buzz_core::client_identity`: dot-separated +/// alphanumerics with optional `-`/`+` pre-release and build metadata. +fn is_plausible_version(version: &str) -> bool { + !version.is_empty() + && version.len() <= MAX_LOGGED_VALUE_LEN + && version + .bytes() + .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-' | b'+')) +} + +/// A single bounded, all-digit version component. +/// +/// Digits only: anything else (`x`, `1-rc`, whitespace) means this is not a +/// version Buzz produced, and guessing would widen the label alphabet. +fn numeric_component(component: &str) -> Option<&str> { + if component.is_empty() + || component.len() > MAX_VERSION_COMPONENT_LEN + || !component.bytes().all(|b| b.is_ascii_digit()) + { + return None; + } + Some(component) +} + +/// Bound a raw value's length for inclusion in a log line. +fn truncate_for_log(value: &str) -> String { + value.chars().take(MAX_LOGGED_VALUE_LEN).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core::client_identity::client_header_value; + + fn parse(raw: &str) -> Result { + ClientInfo::parse(raw) + } + + #[test] + fn parses_a_desktop_header() { + let info = parse(r#"v=1, app=buzz-desktop, platform=macos, app-version="0.5.2""#).unwrap(); + assert_eq!(info.app(), "buzz-desktop"); + assert_eq!(info.platform(), "macos"); + assert_eq!(info.app_version(), "0.5"); + assert_eq!(info.app_version_detail(), "0.5.2"); + } + + #[test] + fn round_trips_every_header_the_shared_builder_emits() { + // The builder in buzz-core and this parser are a wire contract across + // crates; assert they agree rather than restating the format. + for app in [ + ClientApp::Desktop, + ClientApp::Mobile, + ClientApp::Cli, + ClientApp::Acp, + ] { + for platform in [ + ClientPlatform::MacOs, + ClientPlatform::Windows, + ClientPlatform::Linux, + ClientPlatform::Ios, + ClientPlatform::Android, + ] { + let raw = client_header_value(app, platform, "1.2.3").expect("builder emits"); + let info = parse(&raw).unwrap_or_else(|e| panic!("{raw:?} rejected as {e:?}")); + assert_eq!(info.app(), app.as_str()); + assert_eq!(info.platform(), platform.as_str()); + assert_eq!(info.app_version(), "1.2"); + } + } + } + + #[test] + fn accepts_the_mobile_header_with_its_extra_fields() { + // Mobile (block/buzz#2596) sends app-build, os-version, and os-api. + // Unknown keys must be ignored, not rejected, or that client would be + // counted as unidentified. + let info = parse( + r#"v=1, app=buzz-mobile, platform=android, app-version="0.4.5", app-build="6", os-version="15", os-api=35"#, + ) + .unwrap(); + assert_eq!(info.app(), "buzz-mobile"); + assert_eq!(info.platform(), "android"); + assert_eq!(info.app_version(), "0.4"); + } + + #[test] + fn ignores_keys_it_has_never_heard_of() { + let info = parse(r#"v=1, app=buzz-cli, platform=linux, app-version="2.0.0", future-key=7"#) + .unwrap(); + assert_eq!(info.app_version(), "2.0"); + } + + #[test] + fn is_order_independent() { + let info = + parse(r#"app-version="3.4.5", platform=windows, app=buzz-desktop, v=1"#).unwrap(); + assert_eq!(info.app(), "buzz-desktop"); + assert_eq!(info.platform(), "windows"); + assert_eq!(info.app_version(), "3.4"); + } + + #[test] + fn narrows_versions_to_major_minor() { + for (raw, expected) in [ + ("1.2", "1.2"), + ("1.2.3", "1.2"), + ("1.2.3.4", "1.2"), + ("0.0.1", "0.0"), + ("10.20.30", "10.20"), + ] { + let header = format!(r#"v=1, app=buzz-cli, platform=linux, app-version="{raw}""#); + assert_eq!(parse(&header).unwrap().app_version(), expected, "{raw}"); + } + } + + #[test] + fn rejects_versions_that_would_widen_the_label_alphabet() { + for bad in [ + "1", // no minor + "1.x", // non-numeric minor + "1.2-rc.1", // pre-release in the minor slot + "v1.2", // non-numeric major + "1.123456", // minor over the component bound + "123456.1", // major over the component bound + "", // empty + ".1", // empty major + "1.", // empty minor + "1. 2", // whitespace + "\u{7f}1.2", // control character + ] { + let header = format!(r#"v=1, app=buzz-cli, platform=linux, app-version="{bad}""#); + assert!(parse(&header).is_err(), "expected {bad:?} to be refused"); + } + } + + #[test] + fn rejects_apps_and_platforms_outside_the_allowlist() { + assert_eq!( + parse(r#"v=1, app=evil-client, platform=macos, app-version="1.0.0""#), + Err(ParseFailure::UnknownApp) + ); + assert_eq!( + parse(r#"v=1, app=buzz-desktop, platform=freebsd, app-version="1.0.0""#), + Err(ParseFailure::UnknownPlatform) + ); + // A high-cardinality forgery attempt must be refused, not passed + // through as a label value. + assert_eq!( + parse(r#"v=1, app=buzz-desktop-9e1f, platform=macos, app-version="1.0.0""#), + Err(ParseFailure::UnknownApp) + ); + } + + #[test] + fn rejects_an_unsupported_format_version() { + for bad in ["v=2", "v=0", "v=99"] { + let header = format!(r#"{bad}, app=buzz-desktop, platform=macos, app-version="1.0.0""#); + assert_eq!( + parse(&header), + Err(ParseFailure::UnsupportedVersion), + "{bad}" + ); + } + } + + #[test] + fn requires_every_mandatory_field() { + assert_eq!( + parse(r#"app=buzz-desktop, platform=macos, app-version="1.0.0""#), + Err(ParseFailure::UnsupportedVersion) + ); + assert_eq!( + parse(r#"v=1, platform=macos, app-version="1.0.0""#), + Err(ParseFailure::UnknownApp) + ); + assert_eq!( + parse(r#"v=1, app=buzz-desktop, app-version="1.0.0""#), + Err(ParseFailure::UnknownPlatform) + ); + assert_eq!( + parse(r#"v=1, app=buzz-desktop, platform=macos"#), + Err(ParseFailure::BadAppVersion) + ); + } + + #[test] + fn rejects_structurally_broken_input() { + for bad in [ + "", // empty + "garbage", // no `=` + r#"v=1, app=buzz-desktop, platform=macos, app-version="#, // empty value + r#"v=1, app=buzz-desktop, platform=macos, app-version="1.0"#, // unterminated quote + r#"v=1, app=buzz-desktop, platform=macos, ="1.0""#, // empty key + r#"v="1", app=buzz-desktop, platform=macos, app-version="1.0""#, // quoted v + r#"v=1, app="buzz-desktop", platform=macos, app-version="1.0""#, // quoted app + r#"v=1, app=buzz-desktop, platform=macos, app-version=1.0"#, // unquoted version + ] { + assert!(parse(bad).is_err(), "expected {bad:?} to be refused"); + } + } + + #[test] + fn a_comma_inside_a_quoted_string_cannot_forge_a_member() { + // If members were split naively on `,`, this would parse as a valid + // header plus a smuggled `app=buzz-cli` member. + let raw = r#"v=1, app=buzz-desktop, platform=macos, app-version="1.0.0, app=buzz-cli""#; + assert_eq!(parse(raw), Err(ParseFailure::BadAppVersion)); + } + + #[test] + fn rejects_control_characters_and_header_injection() { + for bad in [ + "v=1, app=buzz-desktop, platform=macos, app-version=\"1.0.0\"\r\nX-Evil: 1", + "v=1, app=buzz-desktop,\nplatform=macos, app-version=\"1.0.0\"", + "v=1, app=buzz-desktop, platform=macos, app-version=\"1.0\u{0}0\"", + ] { + assert_eq!(parse(bad), Err(ParseFailure::Malformed), "{bad:?}"); + } + } + + #[test] + fn rejects_backslash_escapes_rather_than_unescaping_them() { + let raw = r#"v=1, app=buzz-desktop, platform=macos, app-version="1.\"0""#; + assert_eq!(parse(raw), Err(ParseFailure::Malformed)); + } + + #[test] + fn rejects_non_ascii_and_oversized_headers() { + assert_eq!( + ClientInfo::parse_bytes( + "v=1, app=buzz-desktop, platform=macos, app-version=\"1.0é\"".as_bytes() + ), + Err(ParseFailure::Unreadable) + ); + let padding = "x".repeat(MAX_HEADER_LEN); + let oversized = + format!(r#"v=1, app=buzz-desktop, platform=macos, app-version="1.0.0", pad={padding}"#); + assert!(oversized.len() > MAX_HEADER_LEN); + assert_eq!( + ClientInfo::parse_bytes(oversized.as_bytes()), + Err(ParseFailure::Unreadable) + ); + } + + #[test] + fn tolerates_whitespace_variations_around_delimiters() { + for raw in [ + r#"v=1,app=buzz-desktop,platform=macos,app-version="1.0.0""#, + "v=1, app=buzz-desktop,\tplatform=macos, app-version=\"1.0.0\"", + r#" v=1, app = buzz-desktop, platform =macos, app-version= "1.0.0" "#, + ] { + let info = parse(raw).unwrap_or_else(|e| panic!("{raw:?} rejected as {e:?}")); + assert_eq!(info.app(), "buzz-desktop"); + assert_eq!(info.app_version(), "1.0"); + } + } + + #[test] + fn a_missing_header_is_not_a_parse_failure() { + assert!(ClientInfo::from_headers(&HeaderMap::new()).is_none()); + } + + #[test] + fn reads_the_header_from_a_header_map() { + let mut headers = HeaderMap::new(); + headers.insert( + CLIENT_HEADER, + r#"v=1, app=buzz-cli, platform=linux, app-version="9.9.9""# + .parse() + .expect("valid header value"), + ); + let info = ClientInfo::from_headers(&headers).expect("parses"); + assert_eq!(info.app(), "buzz-cli"); + assert_eq!(info.app_version(), "9.9"); + } + + #[test] + fn unidentified_connections_get_the_unknown_bucket() { + assert_eq!(labels(None), (UNKNOWN_LABEL, UNKNOWN_LABEL, UNKNOWN_LABEL)); + let info = parse(r#"v=1, app=buzz-cli, platform=linux, app-version="1.0.0""#).unwrap(); + assert_eq!(labels(Some(&info)), ("buzz-cli", "linux", "1.0")); + } + + #[test] + fn failure_reasons_are_a_small_fixed_set() { + // These become a Prometheus label; they must stay bounded and stable. + for (failure, expected) in [ + (ParseFailure::Unreadable, "unreadable"), + (ParseFailure::Malformed, "malformed"), + (ParseFailure::UnsupportedVersion, "unsupported_version"), + (ParseFailure::UnknownApp, "unknown_app"), + (ParseFailure::UnknownPlatform, "unknown_platform"), + (ParseFailure::BadAppVersion, "bad_app_version"), + ] { + assert_eq!(failure.as_str(), expected); + } + } + + #[test] + fn logged_version_detail_is_length_bounded() { + // `app-version` is bounded by the builder, but the parser must not + // rely on a hostile client honouring that. + let long = "1.2".to_owned() + &".9".repeat(64); + assert_eq!( + truncate_for_log(&long).chars().count(), + MAX_LOGGED_VALUE_LEN + ); + } +} diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index c37421e7e8..cbe2600b9d 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -18,6 +18,7 @@ use buzz_auth::{generate_challenge, AuthContext, LimitType}; use buzz_core::tenant::TenantContext; use nostr::Filter; +use crate::client_info::ClientInfo; use crate::handlers; use crate::protocol::{ClientMessage, RelayMessage}; use crate::state::{ @@ -123,11 +124,16 @@ impl ConnectionState { /// /// Acquires a connection semaphore permit, sends the NIP-42 AUTH challenge, /// then drives the send, heartbeat, and receive loops until the connection closes. +/// +/// `client` is the advisory `Buzz-Client` identity parsed from the upgrade +/// request, or `None` when the client did not identify itself. It is used only +/// for metrics and logs — never for any access decision. pub async fn handle_connection( socket: WebSocket, state: Arc, addr: SocketAddr, tenant: TenantContext, + client: Option, ) { let conn_id = Uuid::new_v4(); let cancel = CancellationToken::new(); @@ -142,7 +148,9 @@ pub async fn handle_connection( community_id, control, move || async move { check_state.db.is_community_active(community_id).await }, - move |control| handle_active_connection(socket, run_state, addr, tenant, conn_id, control), + move |control| { + handle_active_connection(socket, run_state, addr, tenant, conn_id, control, client) + }, ) .await; } @@ -154,6 +162,7 @@ async fn handle_active_connection( tenant: TenantContext, conn_id: Uuid, control: CommunityConnectionControl, + client: Option, ) { let cancel = control.cancellation_token(); let disconnect_reason = control.disconnect_reason(); @@ -195,12 +204,20 @@ async fn handle_active_connection( grace_limit: state.config.slow_client_grace_limit, }); - info!(conn_id = %conn_id, addr = %addr, "WebSocket connection established"); + info!( + conn_id = %conn_id, + addr = %addr, + client.app = client.as_ref().map(ClientInfo::app), + client.platform = client.as_ref().map(ClientInfo::platform), + client.app_version = client.as_ref().map(ClientInfo::app_version_detail), + "WebSocket connection established" + ); metrics::counter!( "buzz_ws_connections_total", "community" => conn.tenant.host().to_owned() ) .increment(1); + crate::client_info::record_connection(client.as_ref()); let challenge_msg = RelayMessage::auth_challenge(&challenge); if tx @@ -215,6 +232,10 @@ async fn handle_active_connection( // Gauge incremented AFTER challenge send succeeds — early disconnects // don't leak. Decremented in the cleanup path below. metrics::gauge!("buzz_ws_connections_active").increment(1.0); + // Per-client breakdown of the same lifecycle. Kept as a separate series + // because `buzz_ws_connections_active` is the unlabeled gauge the HPA + // scales on; labeling it would shard the autoscaler's input. + crate::client_info::increment_active(client.as_ref()); // Register after challenge succeeds — avoids leaked entries on early disconnect. state.conn_manager.register( @@ -306,6 +327,7 @@ async fn handle_active_connection( } } metrics::gauge!("buzz_ws_connections_active").decrement(1.0); + crate::client_info::decrement_active(client.as_ref()); info!(conn_id = %conn_id, addr = %addr, "WebSocket connection closed"); drop(permit); diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 314adad92e..0c363a8649 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -8,6 +8,8 @@ mod admission; pub mod api; /// WebSocket audio relay for huddle voice channels. pub mod audio; +/// Advisory `Buzz-Client` identity header parsing and per-client metrics. +pub mod client_info; /// Relay configuration from environment variables. pub mod config; /// Runtime conformance harness — abstract trace emission at the diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 82ad9938a2..d4e4b7b897 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -323,8 +323,12 @@ async fn nip11_or_ws_handler( if state.shutting_down.load(Ordering::Relaxed) { return (StatusCode::SERVICE_UNAVAILABLE, "relay restarting").into_response(); } + // Advisory only: parsed before the upgrade purely so metrics and + // logs can attribute the connection to an app/platform/version. + // Never consulted for auth, tenant selection, or admission. + let client = crate::client_info::ClientInfo::from_headers(&headers); limit_relay_websocket(ws, max_frame_bytes) - .on_upgrade(move |socket| handle_connection(socket, state, addr, tenant)) + .on_upgrade(move |socket| handle_connection(socket, state, addr, tenant, client)) .into_response() } Err(_) => { diff --git a/crates/buzz-ws-client/Cargo.toml b/crates/buzz-ws-client/Cargo.toml index 5cec925677..6a858408c7 100644 --- a/crates/buzz-ws-client/Cargo.toml +++ b/crates/buzz-ws-client/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true repository.workspace = true [dependencies] +buzz-core = { workspace = true } nostr = { workspace = true } tokio = { workspace = true } tokio-tungstenite = { workspace = true } diff --git a/crates/buzz-ws-client/src/connection.rs b/crates/buzz-ws-client/src/connection.rs index bec5b56bb4..6a0f160f5b 100644 --- a/crates/buzz-ws-client/src/connection.rs +++ b/crates/buzz-ws-client/src/connection.rs @@ -1,10 +1,14 @@ use std::collections::VecDeque; use std::time::Duration; +use buzz_core::client_identity::{ + client_header_value_for_host, may_identify_to, ClientApp, CLIENT_HEADER, +}; use futures_util::{SinkExt, StreamExt}; use nostr::{Event, Keys, Tag}; use serde_json::{json, Value}; use tokio::time::timeout; +use tokio_tungstenite::tungstenite::client::ClientRequestBuilder; use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; use tracing::debug; @@ -13,6 +17,32 @@ use crate::message::{build_auth_event, parse_relay_message, OkResponse, RelayMes type WsStream = WebSocketStream>; +/// Build the handshake request for `url`, attaching the advisory `Buzz-Client` +/// header when the destination permits it. +/// +/// Attaching the header is never allowed to fail a connection: an unshipped +/// platform, an unusable version, or a destination outside +/// [`may_identify_to`] simply yields a request without the header. +fn client_request( + url: &str, + app: ClientApp, + app_version: &str, +) -> Result { + let uri = url + .parse() + .map_err(|e: tokio_tungstenite::tungstenite::http::uri::InvalidUri| { + WsClientError::Url(e.to_string()) + })?; + let builder = ClientRequestBuilder::new(uri); + if !may_identify_to(url) { + return Ok(builder); + } + match client_header_value_for_host(app, app_version) { + Some(value) => Ok(builder.with_header(CLIENT_HEADER, value)), + None => Ok(builder), + } +} + /// Seconds to wait for the relay to send the NIP-42 AUTH challenge after connecting. pub const AUTH_CHALLENGE_TIMEOUT_SECS: u64 = 20; @@ -46,11 +76,28 @@ impl NostrWsConnection { /// Connects to the relay at `url` without performing authentication. pub async fn connect(url: &str) -> Result { + Self::connect_as(url, ClientApp::Cli, env!("CARGO_PKG_VERSION")).await + } + + /// Connects to the relay at `url`, identifying as `app` version + /// `app_version` in the advisory `Buzz-Client` header. + /// + /// The header lets the relay report which client versions and platforms + /// its live connections come from. It is best-effort: if it cannot be + /// built (unshipped platform, unusable version) or the destination is not + /// one this client may identify itself to, the connection proceeds without + /// it and the relay counts it as unidentified. + pub async fn connect_as( + url: &str, + app: ClientApp, + app_version: &str, + ) -> Result { let parsed = url .parse::() .map_err(|e| WsClientError::Url(e.to_string()))?; - let (ws, _response) = connect_async(parsed.as_str()) + let request = client_request(parsed.as_str(), app, app_version)?; + let (ws, _response) = connect_async(request) .await .map_err(WsClientError::WebSocket)?; diff --git a/desktop/src-tauri/src/native_websocket.rs b/desktop/src-tauri/src/native_websocket.rs index 128f2df79d..3b94f4015c 100644 --- a/desktop/src-tauri/src/native_websocket.rs +++ b/desktop/src-tauri/src/native_websocket.rs @@ -1,11 +1,15 @@ use std::{collections::HashMap, sync::Arc, time::Duration}; +use buzz_core_pkg::client_identity::{ + client_header_value_for_host, may_identify_to, ClientApp, CLIENT_HEADER, +}; use futures_util::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use tauri::{ipc::Channel, plugin::TauriPlugin, Manager, Runtime}; use tokio::sync::{mpsc, oneshot, Mutex}; use tokio_tungstenite::{ connect_async, + tungstenite::client::ClientRequestBuilder, tungstenite::protocol::{frame::coding::CloseCode, CloseFrame, Message}, }; use tokio_util::sync::CancellationToken; @@ -15,6 +19,25 @@ const WRITE_TIMEOUT: Duration = Duration::from_secs(10); const SHUTDOWN_TIMEOUT: Duration = Duration::from_millis(250); const SEND_QUEUE_CAPACITY: usize = 64; +/// Build the handshake request for `url`, attaching the advisory `Buzz-Client` +/// header when the destination permits it. +/// +/// The header tells the relay which app, platform, and version a live +/// connection belongs to. Attaching it never fails a connection: a destination +/// outside `may_identify_to` or an unusable version just omits it, and the +/// relay counts the connection as unidentified. +fn client_request(url: &str) -> Result { + let uri = url.parse().map_err(|error| format!("{error}"))?; + let builder = ClientRequestBuilder::new(uri); + let Some(value) = may_identify_to(url) + .then(|| client_header_value_for_host(ClientApp::Desktop, env!("CARGO_PKG_VERSION"))) + .flatten() + else { + return Ok(builder); + }; + Ok(builder.with_header(CLIENT_HEADER, value)) +} + pub(crate) fn install_crypto_provider() { // Dependencies enable both rustls providers; choose one before TLS setup. let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); @@ -127,9 +150,10 @@ async fn open_connection( on_message: Channel, ) -> Result { let connect_cancel = manager.connect_cancel.lock().await.clone(); + let request = client_request(url)?; let (socket, _) = tokio::select! { _ = connect_cancel.cancelled() => return Err("WebSocket connection cancelled".to_string()), - result = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(url)) => result + result = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(request)) => result .map_err(|_| "WebSocket connection timed out".to_string())? .map_err(|error| error.to_string())?, };