Skip to content
Open
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
2 changes: 2 additions & 0 deletions Justfile
Original file line number Diff line number Diff line change
Expand Up @@ -423,6 +423,8 @@ test-unit:
# never dialed). Keep the Cargo fallback in step.
cargo nextest run -p buzz-relay-mesh
cargo nextest run -p buzz-cli
cargo nextest run -p buzz-ws-client
cargo nextest run -p buzz-pairing-cli
# buzz-sdk builder/validation unit tests: pure event-builder and input
# validation (e.g. the canvas writer-discipline/skew guard and the
# canvas_write_survived predicate), no infra. `--lib` runs all unit
Expand Down
42 changes: 41 additions & 1 deletion crates/buzz-acp/src/relay.rs
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,8 @@ 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::IntoClientRequest;
use tokio_tungstenite::tungstenite::http::{header::USER_AGENT, HeaderValue};
use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream};
use tracing::{debug, info, warn};
use uuid::Uuid;
Expand Down Expand Up @@ -4204,6 +4206,9 @@ where
Err(last_err.unwrap_or(RelayError::ConnectionClosed))
}

/// `User-Agent` sent on every relay WebSocket upgrade.
const RELAY_USER_AGENT: &str = concat!("buzz-acp/", env!("CARGO_PKG_VERSION"));

/// Perform a single WebSocket connect + NIP-42 auth handshake.
///
/// Returns `(ws, buffer)` on success.
Expand All @@ -4216,7 +4221,14 @@ async fn do_connect(
.parse::<url::Url>()
.map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?;

let (ws, _response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(parsed.as_str()))
let mut request = parsed
.as_str()
.into_client_request()
.map_err(|e| RelayError::WebSocket(Box::new(e)))?;
request
.headers_mut()
.insert(USER_AGENT, HeaderValue::from_static(RELAY_USER_AGENT));
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)))?;
Expand Down Expand Up @@ -6375,6 +6387,34 @@ mod tests {
/// WebSocket endpoint, and tungstenite returns `Error::Http` (non-101
/// response) or `Error::Url(UnsupportedUrlScheme)` depending on how far
/// the handshake gets. Either way it must not be retried.
/// The upgrade `do_connect()` sends names buzz-acp and its version.
#[allow(clippy::result_large_err)] // tungstenite's handshake callback type
#[tokio::test]
async fn do_connect_sends_user_agent() {
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};

let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("ws://{}", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut received = None;
let _ws = tokio_tungstenite::accept_hdr_async(
stream,
|request: &Request, response: Response| {
received = request.headers().get(USER_AGENT).cloned();
Ok(response)
},
)
.await
.unwrap();
received
});
// The server hangs up after the upgrade, so the AUTH handshake fails.
let _ = do_connect(&url, &nostr::Keys::generate(), None).await;
assert_eq!(server.await.unwrap().unwrap(), RELAY_USER_AGENT);
assert!(RELAY_USER_AGENT.starts_with("buzz-acp/"));
}

#[tokio::test]
async fn do_connect_wrong_scheme_is_terminal() {
let keys = nostr::Keys::generate();
Expand Down
20 changes: 18 additions & 2 deletions crates/buzz-audio-client/src/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,12 @@ use serde_json::{json, Value};
use std::{collections::BTreeSet, future::Future, time::Duration};
use tokio_tungstenite::{
connect_async_tls_with_config,
tungstenite::{client::IntoClientRequest, protocol::WebSocketConfig, Message},
tungstenite::{
client::IntoClientRequest,
http::{header::USER_AGENT, HeaderValue},
protocol::WebSocketConfig,
Message,
},
Connector, MaybeTlsStream, WebSocketStream,
};

Expand Down Expand Up @@ -96,11 +101,15 @@ async fn control(socket: &mut AudioSocket) -> Result<Value, String> {
}
}

/// `User-Agent` sent on the upgrade unless the request already carries one.
const DEFAULT_USER_AGENT: &str = concat!("buzz-audio-client/", env!("CARGO_PKG_VERSION"));

/// Connect and authenticate using a host-provided asynchronous signer.
///
/// The host validates the destination before calling, and constructs the NIP-42
/// event (including its exact relay tag) from the challenge. `request` is the
/// audio URL, or a request carrying host-supplied upgrade headers. No private key is
/// audio URL, or a request carrying host-supplied upgrade headers; without a
/// `User-Agent` it sends `buzz-audio-client/<version>`. No private key is
/// accepted. Dropping this future cancels connection/admission; the entire
/// operation is bounded to 15 seconds, including signing.
pub async fn connect<F, Fut>(
Expand All @@ -117,6 +126,13 @@ where
let config = WebSocketConfig::default()
.max_message_size(Some(65_536))
.max_frame_size(Some(65_536));
let mut request = request
.into_client_request()
.map_err(|e| format!("audio WS connect failed: {e}"))?;
request
.headers_mut()
.entry(USER_AGENT)
.or_insert(HeaderValue::from_static(DEFAULT_USER_AGENT));
let (mut socket, _) =
connect_async_tls_with_config(request, Some(config), false, connector)
.await
Expand Down
17 changes: 12 additions & 5 deletions crates/buzz-audio-client/tests/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,11 @@ async fn request_headers_reach_the_upgrade() {
let (stream, _) = listener.accept().await.unwrap();
let (seen, header) = oneshot::channel();
let mut socket = accept_hdr_async(stream, |request: &Request, response: Response| {
let _ = seen.send(request.headers().get("authorization").cloned());
let headers = request.headers();
let _ = seen.send((
headers.get("authorization").cloned(),
headers.get("user-agent").cloned(),
));
Ok(response)
})
.await
Expand Down Expand Up @@ -182,8 +186,11 @@ async fn request_headers_reach_the_upgrade() {
})
.await
.unwrap();
assert_eq!(
server.await.unwrap().unwrap().to_str().unwrap(),
"Bearer upgrade-token"
);
let (authorization, user_agent) = server.await.unwrap();
assert_eq!(authorization.unwrap(), "Bearer upgrade-token");
let user_agent = user_agent.unwrap();
assert!(user_agent
.to_str()
.unwrap()
.starts_with("buzz-audio-client/"));
}
24 changes: 22 additions & 2 deletions crates/buzz-cli/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ use crate::error::CliError;

/// NIP-FI assertion header (`docs/nips/NIP-FI.md` §Client-attached transport).
const FEDERATED_IDENTITY_HEADER: &str = "Nostr-Federated-Identity";
/// `User-Agent` sent on every WebSocket upgrade.
const USER_AGENT: &str = concat!("buzz-cli/", env!("CARGO_PKG_VERSION"));

/// Descriptor returned by the relay after a successful upload.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
Expand Down Expand Up @@ -704,10 +706,14 @@ impl BuzzClient {
})
}

/// The NIP-FI assertion as WebSocket upgrade headers (empty without an
/// evidence source).
/// The WebSocket upgrade headers: the CLI's `User-Agent`, and the NIP-FI
/// assertion when there is an evidence source.
async fn upgrade_headers(&self) -> Result<buzz_ws_client::HeaderMap, CliError> {
let mut headers = buzz_ws_client::HeaderMap::new();
headers.insert(
"user-agent",
buzz_ws_client::HeaderValue::from_static(USER_AGENT),
);
if let Some(assertion) = self.evidence().await? {
let mut value =
buzz_ws_client::HeaderValue::from_str(&format!("Bearer {assertion}"))
Expand Down Expand Up @@ -4095,6 +4101,20 @@ mod ws_deadline_tests {

pub(super) type Log = Arc<std::sync::Mutex<Vec<String>>>;

/// Every upgrade names the CLI and its version.
#[tokio::test]
async fn upgrade_headers_carry_the_cli_user_agent() {
let keys = Keys::generate();
let c = BuzzClient::new(
"http://127.0.0.1:1".into(),
Credentials::from_signer(Arc::new(keys.clone()), keys.public_key()),
)
.unwrap();
let headers = c.upgrade_headers().await.unwrap();
assert_eq!(headers.get("user-agent").unwrap(), super::USER_AGENT);
assert!(super::USER_AGENT.starts_with("buzz-cli/"));
}

/// A sign-in signature that comes back after the relay's 5 s deadline
/// reconnects and signs the new challenge, never the old one.
#[tokio::test]
Expand Down
33 changes: 31 additions & 2 deletions crates/buzz-pairing-cli/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,11 @@ use clap::{Parser, Subcommand};
use futures_util::{SinkExt, StreamExt};
use nostr::{Event, EventBuilder, Keys, RelayUrl, SecretKey, ToBech32};
use tokio::time::timeout;
use tokio_tungstenite::tungstenite::{
client::IntoClientRequest,
handshake::client::Request,
http::{header::USER_AGENT, HeaderValue},
};
use tokio_tungstenite::{connect_async, tungstenite::Message};
use zeroize::Zeroizing;

Expand Down Expand Up @@ -133,7 +138,7 @@ async fn cmd_source(relay_url: String, nsec: Option<String>) -> Result<(), CliEr

// Connect to relay and handle NIP-42 auth if required.
// Auth uses the session's ephemeral keys so the relay accepts our events.
let (ws, _) = connect_async(&relay_url).await?;
let (ws, _) = connect_async(relay_request(&relay_url)?).await?;
let (mut write, mut read) = ws.split();
handle_nip42_auth(&mut read, &mut write, &session, &relay_url).await?;

Expand Down Expand Up @@ -235,7 +240,7 @@ async fn cmd_target(relay_override: Option<String>, show_secret: bool) -> Result
let (mut session, offer_event) = PairingSession::new_target(&qr)?;

// Connect to relay and handle NIP-42 auth if required.
let (ws, _) = connect_async(&relay_url).await?;
let (ws, _) = connect_async(relay_request(&relay_url)?).await?;
let (mut write, mut read) = ws.split();
handle_nip42_auth(&mut read, &mut write, &session, &relay_url).await?;

Expand Down Expand Up @@ -629,3 +634,27 @@ fn hex_to_32(s: &str) -> Result<[u8; 32], CliError> {
.try_into()
.map_err(|_| CliError::Other(format!("expected 32 bytes, got wrong length for '{s}'")))
}

/// `User-Agent` sent on the relay WebSocket upgrade.
const RELAY_USER_AGENT: &str = concat!("buzz-pairing-cli/", env!("CARGO_PKG_VERSION"));

/// The relay upgrade request for `url`, naming this tool and its version.
fn relay_request(url: &str) -> Result<Request, tokio_tungstenite::tungstenite::Error> {
let mut request = url.into_client_request()?;
request
.headers_mut()
.insert(USER_AGENT, HeaderValue::from_static(RELAY_USER_AGENT));
Ok(request)
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn relay_request_carries_user_agent() {
let request = relay_request("wss://relay.example.com").unwrap();
assert_eq!(request.headers()[USER_AGENT], RELAY_USER_AGENT);
assert!(RELAY_USER_AGENT.starts_with("buzz-pairing-cli/"));
}
}
75 changes: 69 additions & 6 deletions crates/buzz-ws-client/src/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,8 @@ use nostr::{Event, NostrSigner, Tag};
use serde_json::{json, Value};
use tokio::time::timeout;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::http::HeaderMap;
use tokio_tungstenite::tungstenite::handshake::client::Request;
use tokio_tungstenite::tungstenite::http::{header::USER_AGENT, HeaderMap, HeaderValue};
use tokio_tungstenite::{
client_async, connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream,
};
Expand Down Expand Up @@ -40,6 +41,19 @@ async fn phase<T>(
.map_err(|_| WsClientError::Timeout)?
}

/// `User-Agent` sent on the upgrade unless the caller's headers supply one.
const DEFAULT_USER_AGENT: &str = concat!("buzz-ws-client/", env!("CARGO_PKG_VERSION"));

/// The upgrade request for `url`: the default `User-Agent`, then `headers`.
fn upgrade_request(url: &url::Url, headers: HeaderMap) -> Result<Request, WsClientError> {
let mut request = url.as_str().into_client_request()?;
request
.headers_mut()
.insert(USER_AGENT, HeaderValue::from_static(DEFAULT_USER_AGENT));
request.headers_mut().extend(headers);
Ok(request)
}

/// A NIP-42-capable WebSocket connection to a Nostr relay.
pub struct NostrWsConnection {
ws: WsStream,
Expand Down Expand Up @@ -85,7 +99,8 @@ impl NostrWsConnection {
/// [`Self::connect`], sending `headers` on the WebSocket upgrade request.
///
/// The headers are added after the handshake headers, so supplying a handshake
/// header such as `Sec-WebSocket-Key` replaces it and breaks the upgrade.
/// header such as `Sec-WebSocket-Key` replaces it and breaks the upgrade. A
/// `User-Agent` among them replaces the default `buzz-ws-client/<version>`.
pub async fn connect_with_headers(
url: &str,
headers: HeaderMap,
Expand All @@ -94,8 +109,7 @@ impl NostrWsConnection {
.parse::<url::Url>()
.map_err(|e| WsClientError::Url(e.to_string()))?;

let mut request = parsed.as_str().into_client_request()?;
request.headers_mut().extend(headers);
let request = upgrade_request(&parsed, headers)?;

let (ws, _response) = connect_async(request)
.await
Expand All @@ -117,7 +131,7 @@ impl NostrWsConnection {
let parsed = url
.parse::<url::Url>()
.map_err(|e| WsClientError::Url(e.to_string()))?;
let request = parsed.as_str().into_client_request()?;
let request = upgrade_request(&parsed, HeaderMap::new())?;
let tcp = tokio::net::TcpStream::connect(addr)
.await
.map_err(|e| WsClientError::WebSocket(e.into()))?;
Expand Down Expand Up @@ -426,7 +440,6 @@ mod tests {
#[tokio::test]
async fn connect_with_headers_sends_headers_on_upgrade() {
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};
use tokio_tungstenite::tungstenite::http::HeaderValue;

let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("ws://{}", listener.local_addr().unwrap());
Expand Down Expand Up @@ -460,6 +473,56 @@ mod tests {
);
}

/// The `User-Agent` a relay sees on an upgrade sent with `headers`, through
/// `connect_with_headers` or, with `via_addr`, `connect_to_addr`.
#[allow(clippy::result_large_err)]
async fn upgrade_user_agent(headers: HeaderMap, via_addr: bool) -> Option<HeaderValue> {
use tokio_tungstenite::tungstenite::handshake::server::{Request, Response};

let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap().to_string();
let server = tokio::spawn(async move {
let (stream, _) = listener.accept().await.unwrap();
let mut received = None;
let _ws = tokio_tungstenite::accept_hdr_async(
stream,
|request: &Request, response: Response| {
received = request.headers().get(USER_AGENT).cloned();
Ok(response)
},
)
.await
.unwrap();
received
});
let url = format!("ws://{addr}");
let _conn = if via_addr {
NostrWsConnection::connect_to_addr(&url, &addr).await
} else {
NostrWsConnection::connect_with_headers(&url, headers).await
}
.unwrap();
server.await.unwrap()
}

#[tokio::test]
async fn upgrade_sends_default_user_agent() {
let expected = Some(HeaderValue::from_static(DEFAULT_USER_AGENT));
assert_eq!(upgrade_user_agent(HeaderMap::new(), false).await, expected);
assert_eq!(upgrade_user_agent(HeaderMap::new(), true).await, expected);
assert!(DEFAULT_USER_AGENT.starts_with("buzz-ws-client/"));
}

#[tokio::test]
async fn caller_user_agent_replaces_default() {
let mut headers = HeaderMap::new();
headers.insert(USER_AGENT, HeaderValue::from_static("example/1.0"));
assert_eq!(
upgrade_user_agent(headers, false).await,
Some(HeaderValue::from_static("example/1.0"))
);
}

/// A relay that answers NIP-42 AUTH by closing with `code`, as the relay
/// does on a NIP-FI key mismatch (1008).
async fn close_after_auth(code: u16) -> WsClientError {
Expand Down
Loading
Loading