From 83bd8b1caed7b86d4edf61ec413d82f7110b8cf6 Mon Sep 17 00:00:00 2001 From: Luke Tornquist Date: Mon, 2 Nov 2026 21:04:19 -0500 Subject: [PATCH] fix(loadgen): sign public URLs when dialling connect_addr in plaintext (#8926) --- crates/buzz-test-client/src/loadgen/net.rs | 169 +++++++++++++++++- .../buzz-test-client/src/loadgen/scenario.rs | 6 +- crates/buzz-ws-client/src/connection.rs | 15 +- perf/LOADGEN.md | 13 +- 4 files changed, 182 insertions(+), 21 deletions(-) diff --git a/crates/buzz-test-client/src/loadgen/net.rs b/crates/buzz-test-client/src/loadgen/net.rs index 0c0008674c..2b44a550ff 100644 --- a/crates/buzz-test-client/src/loadgen/net.rs +++ b/crates/buzz-test-client/src/loadgen/net.rs @@ -1,5 +1,6 @@ //! Connection helpers that honor the scenario's `connect_addr` override. +use std::collections::HashSet; use std::net::SocketAddr; use std::time::Duration; @@ -9,12 +10,15 @@ use nostr::{Event, JsonUtil, Keys, Tag}; use super::identity; use super::scenario::{Community, Scenario}; -/// Shared endpoints for one shard: an HTTP client whose community hosts -/// resolve to `connect_addr` when it is set. +/// Shared endpoints for one shard. When `connect_addr` is set, community hosts +/// resolve to it and are dialled in plaintext there, while requests are still +/// signed for the community's own `https://` and `wss://` URLs. #[derive(Clone)] pub struct Net { - pub http: reqwest::Client, + http: reqwest::Client, connect_addr: Option, + /// Community hosts that resolve to `connect_addr`. + plaintext_hosts: HashSet, connect_timeout: Duration, } @@ -23,20 +27,23 @@ impl Net { let mut builder = reqwest::Client::builder() .timeout(Duration::from_secs(60)) .pool_max_idle_per_host(1024); + let mut plaintext_hosts = HashSet::new(); if let Some(addr) = &scenario.target.connect_addr { let resolved: SocketAddr = tokio::net::lookup_host(addr) .await? .next() .ok_or_else(|| anyhow::anyhow!("connect_addr {addr} did not resolve"))?; for community in scenario.communities() { - if let Some(host) = &community.host { - builder = builder.resolve(host, resolved); + if let Some(host) = community.host { + builder = builder.resolve(&host, resolved); + plaintext_hosts.insert(host); } } } Ok(Self { http: builder.build()?, connect_addr: scenario.target.connect_addr.clone(), + plaintext_hosts, connect_timeout: Duration::from_secs_f64(scenario.target.connect_timeout_secs), }) } @@ -64,6 +71,21 @@ impl Net { .map_err(|_| anyhow::anyhow!("connect timeout after {:?}", self.connect_timeout))? } + /// The URL to send a request signed for `url` to: plain `http://` for a + /// community host dialled at `connect_addr`, otherwise `url` itself. + fn transport_url(&self, url: &str) -> String { + match url.strip_prefix("https://") { + Some(rest) + if self + .plaintext_hosts + .contains(rest.split(['/', '?']).next().unwrap_or_default()) => + { + format!("http://{rest}") + } + _ => url.to_string(), + } + } + /// `POST {base}{path}` with a NIP-98 header; returns status and body. pub async fn post( &self, @@ -76,7 +98,7 @@ impl Net { let url = format!("{base}{path}"); let mut request = self .http - .post(&url) + .post(self.transport_url(&url)) .header( "Authorization", identity::nip98_header(keys, "POST", &url, &body), @@ -98,7 +120,7 @@ impl Net { ) -> anyhow::Result<(reqwest::StatusCode, String)> { let response = self .http - .get(url) + .get(self.transport_url(url)) .header( "Authorization", identity::nip98_header(keys, "GET", url, &[]), @@ -197,4 +219,137 @@ hold_secs = 1 assert!(started.elapsed() < Duration::from_secs(2)); held.abort(); } + + /// The tag value named `name` in `event`. + fn tag(event: &Event, name: &str) -> String { + event + .tags + .iter() + .find(|t| t.as_slice()[0] == name) + .map(|t| t.as_slice()[1].clone()) + .unwrap_or_else(|| panic!("no {name} tag")) + } + + #[tokio::test] + // tungstenite's handshake callback returns its large error response by value. + #[allow(clippy::result_large_err)] + async fn connect_addr_dials_plaintext_but_signs_the_public_urls() { + use base64::Engine as _; + use futures_util::{SinkExt, StreamExt}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio_tungstenite::tungstenite::{handshake::server, Message}; + + // A plaintext listener standing in for the relay's in-cluster service, + // while the scenario names the community's public wss:// deployment. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + let scenario = Scenario::parse(&format!( + r#" +run_id = "t" +[target] +ws_url = "wss://public.example" +http_url = "https://public.example" +community_host_template = "lg-{{run}}-{{i}}.example" +connect_addr = "{addr}" +[shape] +communities = 1 +channels_per_community = 1 +[population] +humans = 1 +agents = 0 +channels_per_member = 1 +[[ramp]] +users = 1.0 +hold_secs = 1 +"# + )) + .unwrap(); + let net = Net::new(&scenario).await.unwrap(); + let community = scenario.communities().remove(0); + assert_eq!(community.http_url, "https://lg-t-0.example"); + let keys = Keys::generate(); + + // HTTP: plaintext on the wire, Host names the community, and the + // NIP-98 `u` tag is the https:// URL the relay derives from a wss:// + // relay URL (`{scheme}://{host}{path}`). + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let mut buf = [0u8; 4096]; + while !request.windows(4).any(|w| w == b"\r\n\r\n") { + let n = socket.read(&mut buf).await.unwrap(); + assert!(n > 0, "connection closed before the request headers"); + request.extend_from_slice(&buf[..n]); + } + socket + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\n{}") + .await + .unwrap(); + ( + String::from_utf8(request).expect("plaintext HTTP"), + listener, + ) + }); + let (status, _) = net + .post(&keys, &community.http_url, "/events", b"{}".to_vec(), None) + .await + .unwrap(); + assert!(status.is_success()); + let (request, listener) = server.await.unwrap(); + let header = |name: &str| { + request + .lines() + .find_map(|line| { + let (key, value) = line.split_once(':')?; + key.eq_ignore_ascii_case(name) + .then(|| value.trim().to_string()) + }) + .unwrap_or_else(|| panic!("no {name} header in {request}")) + }; + assert!(request.starts_with("POST /events "), "{request}"); + assert_eq!(header("host"), "lg-t-0.example"); + let token = header("authorization"); + let json = base64::engine::general_purpose::STANDARD + .decode(token.strip_prefix("Nostr ").unwrap()) + .unwrap(); + let nip98 = Event::from_json(json).unwrap(); + assert_eq!(tag(&nip98, "u"), "https://lg-t-0.example/events"); + + // WebSocket: plaintext upgrade with the community's Host, and a NIP-42 + // relay tag of wss://. + let server = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut host = None; + let mut ws = tokio_tungstenite::accept_hdr_async( + socket, + |req: &server::Request, resp: server::Response| { + host = req + .headers() + .get("host") + .map(|h| h.to_str().unwrap().to_string()); + Ok(resp) + }, + ) + .await + .expect("plaintext WebSocket upgrade"); + ws.send(Message::text(r#"["AUTH","challenge"]"#)) + .await + .unwrap(); + let Some(Ok(Message::Text(text))) = ws.next().await else { + panic!("no AUTH reply"); + }; + let frame: serde_json::Value = serde_json::from_str(&text).unwrap(); + let auth = Event::from_json(frame[1].to_string()).unwrap(); + let ok = serde_json::json!(["OK", auth.id.to_hex(), true, ""]); + ws.send(Message::text(ok.to_string())).await.unwrap(); + (host, auth) + }); + net.connect(&community, &keys, None).await.unwrap(); + let (host, auth) = server.await.unwrap(); + assert_eq!(host.as_deref(), Some("lg-t-0.example")); + assert_eq!( + tag(&auth, "relay").trim_end_matches('/'), + "wss://lg-t-0.example" + ); + } } diff --git a/crates/buzz-test-client/src/loadgen/scenario.rs b/crates/buzz-test-client/src/loadgen/scenario.rs index 5aeaa4ffb1..f874d4b9c0 100644 --- a/crates/buzz-test-client/src/loadgen/scenario.rs +++ b/crates/buzz-test-client/src/loadgen/scenario.rs @@ -36,8 +36,10 @@ pub struct Target { /// one community at `ws_url`'s host. #[serde(default)] pub community_host_template: Option, - /// Open TCP connections here (`host:port`) instead of resolving community - /// hosts, e.g. the relay's in-cluster service. Host headers still name the community. + /// Dial community hosts here (`host:port`) in plaintext instead of + /// resolving them, e.g. the relay's in-cluster service. `ws_url` and + /// `http_url` keep the community's public scheme: Host headers and signed + /// NIP-98 and NIP-42 URLs still name the community at that scheme. #[serde(default)] pub connect_addr: Option, /// Deployment-root HTTP URL serving `POST /operator/communities`. diff --git a/crates/buzz-ws-client/src/connection.rs b/crates/buzz-ws-client/src/connection.rs index ade9ba6a9d..d5836aa7f5 100644 --- a/crates/buzz-ws-client/src/connection.rs +++ b/crates/buzz-ws-client/src/connection.rs @@ -8,7 +8,7 @@ use tokio::time::timeout; use tokio_tungstenite::tungstenite::client::IntoClientRequest; use tokio_tungstenite::tungstenite::http::HeaderMap; use tokio_tungstenite::{ - client_async_tls, connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream, + client_async, connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream, }; use tracing::debug; @@ -90,12 +90,13 @@ impl NostrWsConnection { Ok(Self::from_stream(ws, url)) } - /// [`Self::connect`], but opens the TCP connection to `addr` (`host:port`) - /// instead of resolving the URL's host. + /// [`Self::connect`], but opens a plaintext TCP connection to `addr` + /// (`host:port`) instead of resolving the URL's host. /// - /// The URL still supplies the `Host` header, TLS server name and NIP-42 - /// relay tag, so a client inside the relay's cluster can reach a community - /// host through the relay's service address. + /// The URL still supplies the `Host` header and NIP-42 relay tag, even when + /// it is `wss://`, so a client inside the relay's cluster can reach a + /// community through the relay's plaintext service address and still + /// authenticate for the community's public URL. pub async fn connect_to_addr(url: &str, addr: &str) -> Result { let parsed = url .parse::() @@ -104,7 +105,7 @@ impl NostrWsConnection { let tcp = tokio::net::TcpStream::connect(addr) .await .map_err(|e| WsClientError::WebSocket(e.into()))?; - let (ws, _response) = client_async_tls(request, tcp) + let (ws, _response) = client_async(request, MaybeTlsStream::Plain(tcp)) .await .map_err(WsClientError::WebSocket)?; diff --git a/perf/LOADGEN.md b/perf/LOADGEN.md index 86288ac44b..b6413326aa 100644 --- a/perf/LOADGEN.md +++ b/perf/LOADGEN.md @@ -71,11 +71,14 @@ deployment's own configuration. ## Targeting and membership `target.connect_addr` (`host:port`) applies only together with -`target.community_host_template`: it then replaces host resolution for the -community hosts, over both WebSocket and HTTP, while TLS names, `Host` headers -and the NIP-42 relay tag still name the community. Without a host -template, the generator connects to `ws_url` and `http_url` as written and -`connect_addr` has no effect. +`target.community_host_template`: the generator then dials the community +hosts there in plaintext, over both WebSocket and HTTP, such as a relay's +in-cluster service. Set `ws_url` and `http_url` to the community's public +scheme anyway (`wss://` and `https://` for a relay behind TLS): `Host` +headers, NIP-98 `u` tags and the NIP-42 relay tag keep that scheme, which the +relay checks against its own relay URL. Without a host template, the +generator connects to `ws_url` and `http_url` as written and `connect_addr` +has no effect. Seeded identities do not join communities or channels. A relay that requires membership rejects both `seed` and `run`.