fix(loadgen): sign public URLs when dialling connect_addr in plaintext (#8926)

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<String>,
+    /// Community hosts that resolve to `connect_addr`.
+    plaintext_hosts: HashSet<String>,
     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://<community host>.
+        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"
+        );
+    }
 }

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<String>,
-    /// 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<String>,
     /// Deployment-root HTTP URL serving `POST /operator/communities`.

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<Self, WsClientError> {
         let parsed = url
             .parse::<url::Url>()
@@ -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)?;
 

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`.