diff --git a/Cargo.lock b/Cargo.lock index af1dcd23c7..6e15c742ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3837,6 +3837,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "sprout-pair-relay" +version = "0.1.0" +dependencies = [ + "futures-util", + "http-body-util", + "hyper", + "hyper-util", + "parking_lot", + "rand 0.8.5", + "secp256k1", + "serde_json", + "sha2 0.10.9", + "tokio", + "tokio-tungstenite", + "tokio-util", +] + [[package]] name = "sprout-pairing-cli" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 1e112aa7f7..25339baf8d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,6 +20,7 @@ members = [ "crates/sprout-persona", "crates/git-credential-nostr", "crates/git-sign-nostr", + "crates/sprout-pair-relay", ] exclude = ["desktop/src-tauri"] resolver = "2" diff --git a/crates/sprout-pair-relay/Cargo.toml b/crates/sprout-pair-relay/Cargo.toml new file mode 100644 index 0000000000..a4b727895d --- /dev/null +++ b/crates/sprout-pair-relay/Cargo.toml @@ -0,0 +1,36 @@ +[package] +name = "sprout-pair-relay" +description = "Ephemeral sidecar relay for NIP-AB device pairing handshakes" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true + +[lib] +name = "sprout_pair_relay" +path = "src/lib.rs" + +[[bin]] +name = "sprout-pair-relay" +path = "src/main.rs" + +[dependencies] +tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "time", "sync"] } +tokio-tungstenite = { workspace = true } +serde_json = { workspace = true } +futures-util = { workspace = true } +tokio-util = { workspace = true } +parking_lot = "0.12" +hyper = { version = "1", features = ["server", "http1"] } +hyper-util = { version = "0.1", features = ["tokio"] } +http-body-util = "0.1" +secp256k1 = { version = "0.29", features = ["global-context"] } +sha2 = "0.10" + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } +tokio-tungstenite = { workspace = true } +secp256k1 = { version = "0.29", features = ["global-context", "rand-std"] } +sha2 = "0.10" +rand = "0.8" diff --git a/crates/sprout-pair-relay/src/lib.rs b/crates/sprout-pair-relay/src/lib.rs new file mode 100644 index 0000000000..09d44f26cb --- /dev/null +++ b/crates/sprout-pair-relay/src/lib.rs @@ -0,0 +1,1044 @@ +//! Ephemeral sidecar relay for NIP-AB device pairing handshakes. +//! +//! Accepts WebSocket connections, matches incoming kind:24134 events against +//! live `#p`-filtered subscriptions, and forwards matches to the subscriber. +//! No persistence. No auth. No history. +//! +//! # Deployment +//! +//! This binary binds **loopback only** and MUST run behind a reverse proxy +//! (nginx, caddy, etc.) that: +//! - Routes only `/pair` to this sidecar +//! - Enforces HTTP read timeouts (mitigates slowloris at the TCP layer) +//! - Terminates TLS +//! +//! The relay does not enforce path restrictions or pre-upgrade connection +//! limits — those are the reverse proxy's responsibility. +//! +//! # Security Model +//! +//! - **Signature verification** — Schnorr signatures are verified against the +//! NIP-01 event ID hash. Events with invalid signatures are rejected. +//! - **No persistence** — events exist only in-flight between matched pub/sub. +//! - **Bounded resources** — 128 max WS connections, 4 KiB max frame, 120s TTL. +//! - **Session cap** — at most 6 accepted EVENTs per connection. +//! - **Freshness** — `created_at` must be within ±120 s of relay wall-clock. +//! - **Deduplication** — duplicate event IDs are rejected; dedup entries expire after 300 s. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use http_body_util::Full; +use hyper::body::{Bytes, Incoming}; +use hyper::header::{ + HeaderValue, CONNECTION, SEC_WEBSOCKET_ACCEPT, SEC_WEBSOCKET_KEY, SEC_WEBSOCKET_VERSION, + UPGRADE, +}; +use hyper::server::conn::http1; +use hyper::service::service_fn; +use hyper::upgrade::Upgraded; +use hyper::{Method, Request, Response, StatusCode, Version}; +use hyper_util::rt::TokioIo; +use parking_lot::Mutex; +use secp256k1::schnorr::Signature as SchnorrSig; +use secp256k1::{Message as SecpMsg, XOnlyPublicKey}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use tokio::net::TcpListener; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::handshake::derive_accept_key; +use tokio_tungstenite::tungstenite::protocol::{Message, Role, WebSocketConfig}; +use tokio_tungstenite::WebSocketStream; +use tokio_util::sync::CancellationToken; + +// ── Constants ───────────────────────────────────────────────────────────────── + +/// Hard per-connection lifetime. `pub(crate)` for test access. +pub(crate) const CONN_TIMEOUT: Duration = Duration::from_secs(120); + +const MAX_CONNS: u32 = 128; +const CHANNEL_CAP: usize = 4; +const KIND_PAIR: u64 = 24134; +/// Max WebSocket frame/message size. NIP-AB handshake payloads are small +/// (ephemeral pubkeys + encrypted session data), well under 4 KiB. +const MAX_FRAME: usize = 4096; +const RATE_WINDOW: Duration = Duration::from_secs(10); +const RATE_MSG_MAX: u32 = 20; +const RATE_EVENT_MAX: u32 = 10; +const SUB_ID_MAX: usize = 64; + +/// Hard session cap: at most this many attempted EVENTs (post-sig-check) per connection. +const MAX_EVENTS_PER_CONN: u32 = 6; + +/// Per-#p delivery budget: enough for one full pairing from each direction. +const MAX_DELIVERED_PER_P: u32 = 12; + +/// Dedup vec rejects new events when still at capacity after TTL eviction (fail closed). +const DEDUP_CAP: usize = 1024; + +/// Delivered map rejects new #p keys when still at capacity after TTL eviction (fail closed). +const DELIVERED_MAP_CAP: usize = 4096; + +/// TTL for entries in `seen_ids` and `delivered`. Entries older than this are +/// evicted on the next access, keeping both structures bounded over time. +const ENTRY_TTL: Duration = Duration::from_secs(300); + +/// Freshness window in seconds (±). +const FRESHNESS_SECS: i64 = 120; + +// ── Core types ──────────────────────────────────────────────────────────────── + +enum OutMsg { + Text(String), + Pong(Vec), + Close, +} + +struct Sub { + conn_id: u64, + sub_id: String, + p_value: [u8; 32], + writer_tx: mpsc::Sender, +} + +pub struct Relay { + subs: Mutex>, + conn_count: AtomicU32, + next_conn_id: AtomicU64, + /// Global dedup vec — entries expire after ENTRY_TTL; rejects at DEDUP_CAP after eviction. + seen_ids: Mutex>, + /// Per-#p delivery counter — entries expire after ENTRY_TTL; rejects at DELIVERED_MAP_CAP after eviction. + delivered: Mutex>, +} + +impl Default for Relay { + fn default() -> Self { + Self::new() + } +} + +impl Relay { + pub fn new() -> Self { + Self { + subs: Mutex::new(Vec::new()), + conn_count: AtomicU32::new(0), + next_conn_id: AtomicU64::new(0), + seen_ids: Mutex::new(Vec::new()), + delivered: Mutex::new(HashMap::new()), + } + } + + /// Atomically check-and-reserve an event ID. Evicts expired entries first. + /// Returns `Ok(true)` if duplicate (already seen), `Ok(false)` if new + /// (reserved — caller MUST call `unreserve_id` if delivery fails), + /// `Err(())` if at capacity after eviction (fail closed). + fn reserve_id(&self, id: &[u8; 32]) -> Result { + let mut vec = self.seen_ids.lock(); + vec.retain(|(_, ts)| ts.elapsed() < ENTRY_TTL); + if vec.iter().any(|(eid, _)| eid == id) { + return Ok(true); // duplicate + } + if vec.len() >= DEDUP_CAP { + return Err(()); // at capacity after eviction — fail closed + } + // Optimistically reserve the slot. + vec.push((*id, tokio::time::Instant::now())); + Ok(false) + } + + /// Remove a previously reserved ID (called when delivery fails). + fn unreserve_id(&self, id: &[u8; 32]) { + let mut vec = self.seen_ids.lock(); + if let Some(pos) = vec.iter().position(|(eid, _)| eid == id) { + vec.swap_remove(pos); + } + } + + /// Atomically check for exactly one subscriber and deliver. + /// Returns `Ok(true)` if delivered, `Ok(false)` if `try_send` failed, + /// `Err(reason)` if wrong subscriber count or budget exceeded. + fn deliver_single(&self, p_value: &[u8; 32], event: &Value) -> Result { + // Acquire subs lock once for the entire operation. + let subs = self.subs.lock(); + + let matching: Vec<&Sub> = subs.iter().filter(|s| &s.p_value == p_value).collect(); + + match matching.len() { + 0 => return Err("no live subscriber"), + 1 => {} // exactly one — proceed + _ => return Err("ambiguous recipient"), + } + + let sub = matching[0]; + + // Hold delivered lock for the entire check+increment (atomic budget). + let mut delivered = self.delivered.lock(); + + // Evict entries older than ENTRY_TTL before checking capacity. + delivered.retain(|_, (_, ts)| ts.elapsed() < ENTRY_TTL); + + let count = delivered.get(p_value).map(|(c, _)| *c).unwrap_or(0); + if count >= MAX_DELIVERED_PER_P { + return Err("recipient session budget exhausted"); + } + // Fail closed if still at capacity after eviction and key is new. + if delivered.len() >= DELIVERED_MAP_CAP && !delivered.contains_key(p_value) { + return Err("relay at capacity"); + } + + // Build the EVENT message. + let msg = Value::Array(vec![ + Value::String("EVENT".into()), + Value::String(sub.sub_id.clone()), + event.clone(), + ]); + let text = match serde_json::to_string(&msg) { + Ok(s) => s, + Err(_) => return Ok(false), + }; + + // Attempt delivery. + if sub.writer_tx.try_send(OutMsg::Text(text)).is_ok() { + // Increment counter atomically (lock already held), refreshing the timestamp. + let entry = delivered + .entry(*p_value) + .or_insert((0, tokio::time::Instant::now())); + entry.0 += 1; + entry.1 = tokio::time::Instant::now(); + Ok(true) + } else { + Ok(false) + } + } + + fn remove_sub(&self, conn_id: u64) { + self.subs.lock().retain(|s| s.conn_id != conn_id); + } +} + +// ── RAII connection guard ───────────────────────────────────────────────────── + +struct ConnGuard { + relay: Arc, + conn_id: u64, +} + +impl Drop for ConnGuard { + fn drop(&mut self) { + self.relay.remove_sub(self.conn_id); + self.relay.conn_count.fetch_sub(1, Ordering::Relaxed); + eprintln!( + "conn closed conn_id={} active={}", + self.conn_id, + self.relay.conn_count.load(Ordering::Relaxed) + ); + } +} + +// ── Rate limiter ────────────────────────────────────────────────────────────── + +struct RateWindow { + count: u32, + window_start: tokio::time::Instant, +} + +impl RateWindow { + fn new() -> Self { + Self { + count: 0, + window_start: tokio::time::Instant::now(), + } + } + + fn tick(&mut self) -> u32 { + if self.window_start.elapsed() >= RATE_WINDOW { + self.count = 0; + self.window_start = tokio::time::Instant::now(); + } + self.count += 1; + self.count + } +} + +// ── JSON helpers ────────────────────────────────────────────────────────────── + +fn jarr(v: Vec) -> String { + Value::Array(v).to_string() +} + +fn make_ok(id: &str, ok: bool, msg: &str) -> String { + jarr(vec![ + Value::String("OK".into()), + Value::String(id.into()), + Value::Bool(ok), + Value::String(msg.into()), + ]) +} + +fn make_closed(sub_id: &str, msg: &str) -> String { + jarr(vec![ + Value::String("CLOSED".into()), + Value::String(sub_id.into()), + Value::String(msg.into()), + ]) +} + +fn make_eose(sub_id: &str) -> String { + jarr(vec![ + Value::String("EOSE".into()), + Value::String(sub_id.into()), + ]) +} + +fn make_notice(msg: &str) -> String { + jarr(vec![ + Value::String("NOTICE".into()), + Value::String(msg.into()), + ]) +} + +// ── Validation ──────────────────────────────────────────────────────────────── + +fn is_lower_hex(s: &str, len: usize) -> bool { + s.len() == len && s.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) +} + +fn decode_hex32(s: &str) -> Option<[u8; 32]> { + if !is_lower_hex(s, 64) { + return None; + } + let mut out = [0u8; 32]; + for (i, chunk) in s.as_bytes().chunks(2).enumerate() { + let hi = (chunk[0] as char).to_digit(16)?; + let lo = (chunk[1] as char).to_digit(16)?; + out[i] = (hi * 16 + lo) as u8; + } + Some(out) +} + +/// Validate a REQ filter. Returns `Ok(p_value)` or `Err(reason)`. +fn validate_filter(filter: &Value) -> Result<[u8; 32], &'static str> { + let obj = filter.as_object().ok_or("filter must be an object")?; + for key in obj.keys() { + match key.as_str() { + "kinds" | "#p" => {} + _ => return Err("unsupported filter field"), + } + } + if let Some(kinds) = obj.get("kinds") { + let arr = kinds.as_array().ok_or("kinds must be an array")?; + if arr.len() != 1 || arr[0].as_u64() != Some(KIND_PAIR) { + return Err("kinds must be [24134]"); + } + } + let p_arr = obj + .get("#p") + .and_then(|v| v.as_array()) + .ok_or("#p filter required")?; + if p_arr.len() != 1 { + return Err("#p must have exactly one value"); + } + let p_str = p_arr[0].as_str().ok_or("#p value must be a string")?; + decode_hex32(p_str).ok_or("#p value must be 64 lowercase hex chars") +} + +/// Validate NIP-44 content structure (no external crate — manual base64 check). +/// +/// Checks: +/// - Standard base64 alphabet only (A-Z, a-z, 0-9, +, /, =) +/// - Decoded length ≥ 99 bytes (1 version + 32 nonce + 32 min ciphertext + 32 MAC + 2 padding) +/// - First decoded byte is 0x02 (NIP-44 version 2) +fn validate_nip44_content(content: &str) -> Result<(), &'static str> { + if content.is_empty() { + return Err("content must not be empty"); + } + + // Validate base64 alphabet and compute decoded length. + let bytes = content.as_bytes(); + let len = bytes.len(); + + // base64 strings must have length that is a multiple of 4 (with padding). + if !len.is_multiple_of(4) { + return Err("content is not valid base64"); + } + + // Check alphabet and count padding. + let mut pad_count = 0usize; + for (i, &b) in bytes.iter().enumerate() { + match b { + b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'+' | b'/' => { + if pad_count > 0 { + // Non-pad after pad is invalid. + return Err("content is not valid base64"); + } + } + b'=' => { + // Padding only allowed in last two positions. + if i < len - 2 { + return Err("content is not valid base64"); + } + pad_count += 1; + if pad_count > 2 { + return Err("content is not valid base64"); + } + } + _ => return Err("content is not valid base64"), + } + } + + // Decoded byte length = (len / 4) * 3 - pad_count. + let decoded_len = (len / 4) * 3 - pad_count; + if decoded_len < 99 { + return Err("content too short for NIP-44 v2"); + } + + // Decode only the first byte to check the version prefix. + // First base64 char encodes bits 7-2 of byte 0; second encodes bits 1-0 of + // byte 0 (high) and bits 5-2 of byte 1 (low). We only need byte 0. + let b64_val = |c: u8| -> Option { + match c { + b'A'..=b'Z' => Some(c - b'A'), + b'a'..=b'z' => Some(c - b'a' + 26), + b'0'..=b'9' => Some(c - b'0' + 52), + b'+' => Some(62), + b'/' => Some(63), + _ => None, + } + }; + let v0 = b64_val(bytes[0]).ok_or("content is not valid base64")?; + let v1 = b64_val(bytes[1]).ok_or("content is not valid base64")?; + let first_byte = (v0 << 2) | (v1 >> 4); + + if first_byte != 0x02 { + return Err("content is not NIP-44 v2 (expected 0x02 prefix)"); + } + + Ok(()) +} + +/// Validate an EVENT object. Returns `Ok((event_id_str, event_id_bytes, p_value))` or `Err(reason)`. +/// +/// Tightenings applied here: +/// - Exactly 7 top-level keys (tightening #4) +/// - Strict tag shape: exactly `[["p", "<64-hex>"]]` (tightening #3) +/// - NIP-44 content validation (tightening #5) +/// - Freshness window on `created_at` (tightening #6) +fn validate_event(ev: &Value) -> Result<(String, [u8; 32], [u8; 32]), &'static str> { + let obj = ev.as_object().ok_or("event must be an object")?; + + // Tightening #4: reject extra top-level fields. + const ALLOWED_KEYS: &[&str] = &[ + "id", + "pubkey", + "created_at", + "kind", + "tags", + "content", + "sig", + ]; + for key in obj.keys() { + if !ALLOWED_KEYS.contains(&key.as_str()) { + return Err("unknown top-level field"); + } + } + + let id = obj.get("id").and_then(|v| v.as_str()).ok_or("missing id")?; + if !is_lower_hex(id, 64) { + return Err("id must be 64 lowercase hex chars"); + } + let id_bytes = decode_hex32(id).ok_or("id must be 64 lowercase hex chars")?; + + let pubkey = obj + .get("pubkey") + .and_then(|v| v.as_str()) + .ok_or("missing pubkey")?; + if !is_lower_hex(pubkey, 64) { + return Err("pubkey must be 64 lowercase hex chars"); + } + if obj.get("kind").and_then(|v| v.as_u64()) != Some(KIND_PAIR) { + return Err("kind must be 24134"); + } + + // Tightening #6: freshness window. + let created_at = obj + .get("created_at") + .and_then(|v| v.as_i64()) + .ok_or("missing created_at")?; + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0); + if (created_at as i128 - now as i128).unsigned_abs() > FRESHNESS_SECS as u128 { + return Err("created_at outside freshness window"); + } + + let content = obj + .get("content") + .and_then(|v| v.as_str()) + .ok_or("missing content")?; + + // Tightening #5: NIP-44 content validation. + validate_nip44_content(content)?; + + let sig = obj + .get("sig") + .and_then(|v| v.as_str()) + .ok_or("missing sig")?; + if !is_lower_hex(sig, 128) { + return Err("sig must be 128 lowercase hex chars"); + } + + let tags = obj + .get("tags") + .and_then(|v| v.as_array()) + .ok_or("missing tags")?; + + // Tightening #3: exactly one tag, exactly ["p", "<64-hex>"]. + if tags.len() != 1 { + return Err("event must have exactly one p tag"); + } + let tag = tags[0].as_array().ok_or("tag must be an array")?; + if tag.len() != 2 { + return Err("p tag must have exactly 2 elements"); + } + if tag[0].as_str() != Some("p") { + return Err("event must have exactly one p tag"); + } + let p_str = tag[1].as_str().ok_or("p tag value must be a string")?; + let p_bytes = decode_hex32(p_str).ok_or("p tag value must be 64 lowercase hex chars")?; + + Ok((id.to_string(), id_bytes, p_bytes)) +} + +fn safe_event_id(ev: &Value) -> String { + ev.get("id") + .and_then(|v| v.as_str()) + .filter(|s| is_lower_hex(s, 64)) + .unwrap_or("") + .to_string() +} + +/// Verify the Schnorr signature and event ID for a NIP-01 event. +/// +/// Steps: +/// 1. Serialize the commitment array `[0, pubkey, created_at, kind, tags, content]` +/// as compact JSON (no spaces). +/// 2. SHA-256 hash the serialization. +/// 3. Verify the hash matches the claimed `id` field. +/// 4. Verify the Schnorr signature over the hash using the `pubkey` field. +fn verify_event_sig(ev: &Value) -> Result<(), &'static str> { + let obj = ev.as_object().ok_or("event must be an object")?; + + let pubkey_str = obj + .get("pubkey") + .and_then(|v| v.as_str()) + .ok_or("missing pubkey")?; + let created_at = obj + .get("created_at") + .and_then(|v| v.as_i64()) + .ok_or("missing created_at")?; + let kind = obj + .get("kind") + .and_then(|v| v.as_u64()) + .ok_or("missing kind")?; + let tags = obj.get("tags").ok_or("missing tags")?; + let content = obj + .get("content") + .and_then(|v| v.as_str()) + .ok_or("missing content")?; + let id_str = obj.get("id").and_then(|v| v.as_str()).ok_or("missing id")?; + let sig_str = obj + .get("sig") + .and_then(|v| v.as_str()) + .ok_or("missing sig")?; + + // Step 1: build the NIP-01 commitment and hash it. + let commitment = Value::Array(vec![ + Value::Number(0.into()), + Value::String(pubkey_str.to_string()), + Value::Number(created_at.into()), + Value::Number(kind.into()), + tags.clone(), + Value::String(content.to_string()), + ]); + let commitment_json = + serde_json::to_string(&commitment).map_err(|_| "failed to serialize commitment")?; + let hash: [u8; 32] = Sha256::digest(commitment_json.as_bytes()).into(); + + // Step 2: verify hash matches claimed id. + let id_bytes = decode_hex32(id_str).ok_or("id must be 64 lowercase hex chars")?; + if hash != id_bytes { + return Err("invalid: event id mismatch"); + } + + // Step 3: parse pubkey as x-only. + let pubkey_bytes = decode_hex32(pubkey_str).ok_or("pubkey must be 64 lowercase hex chars")?; + let xonly_pk = XOnlyPublicKey::from_slice(&pubkey_bytes).map_err(|_| "invalid: bad pubkey")?; + + // Step 4: parse sig. + let sig_bytes = decode_hex64(sig_str).ok_or("sig must be 128 lowercase hex chars")?; + let schnorr_sig = + SchnorrSig::from_slice(&sig_bytes).map_err(|_| "invalid: bad signature encoding")?; + + // Step 5: verify. + let secp = secp256k1::Secp256k1::verification_only(); + let msg = SecpMsg::from_digest(hash); + secp.verify_schnorr(&schnorr_sig, &msg, &xonly_pk) + .map_err(|_| "invalid: signature verification failed") +} + +/// Decode a 128-char lowercase hex string into 64 bytes. +fn decode_hex64(s: &str) -> Option<[u8; 64]> { + if !is_lower_hex(s, 128) { + return None; + } + let mut out = [0u8; 64]; + for (i, chunk) in s.as_bytes().chunks(2).enumerate() { + let hi = (chunk[0] as char).to_digit(16)?; + let lo = (chunk[1] as char).to_digit(16)?; + out[i] = (hi * 16 + lo) as u8; + } + Some(out) +} + +// ── Writer task ─────────────────────────────────────────────────────────────── + +type WsSink = futures_util::stream::SplitSink>, Message>; + +async fn writer_task(mut sink: WsSink, mut rx: mpsc::Receiver, cancel: CancellationToken) { + loop { + let msg = tokio::select! { + _ = cancel.cancelled() => break, + m = rx.recv() => match m { Some(m) => m, None => break }, + }; + let ws_msg = match msg { + OutMsg::Text(s) => Message::Text(s.into()), + OutMsg::Pong(d) => Message::Pong(d.into()), + OutMsg::Close => Message::Close(None), + }; + let result = tokio::select! { + _ = cancel.cancelled() => break, + r = timeout(Duration::from_secs(5), sink.send(ws_msg)) => r, + }; + match result { + Err(_) => break, // timeout + Ok(Err(_)) => break, // send error + Ok(Ok(())) => {} // success + } + } +} + +// ── Connection handler ──────────────────────────────────────────────────────── + +async fn handle_conn(relay: Arc, conn_id: u64, stream: WebSocketStream>) { + let _guard = ConnGuard { + relay: Arc::clone(&relay), + conn_id, + }; + let (sink, mut source) = stream.split(); + let (tx, rx) = mpsc::channel::(CHANNEL_CAP); + let cancel = CancellationToken::new(); + let writer_handle = tokio::spawn(writer_task(sink, rx, cancel.clone())); + tokio::pin!(writer_handle); + + let mut msg_rate = RateWindow::new(); + let mut event_rate = RateWindow::new(); + let mut sub_id: Option = None; + // Tightening #1: hard session cap — counts all valid+sig-verified EVENT attempts. + let mut events_attempted: u32 = 0; + let deadline = tokio::time::sleep(CONN_TIMEOUT); + tokio::pin!(deadline); + + 'conn: loop { + let frame = tokio::select! { + _ = &mut deadline => break 'conn, + _ = &mut writer_handle => break 'conn, // writer died → close + f = source.next() => match f { Some(f) => f, None => break 'conn }, + }; + + // All inbound frames count toward the message rate limit. + if msg_rate.tick() > RATE_MSG_MAX { + eprintln!("conn_id={} rate-limited (msg)", conn_id); + break 'conn; + } + + let frame = match frame { + Ok(f) => f, + Err(_) => break 'conn, + }; + + match frame { + Message::Binary(_) | Message::Frame(_) => break 'conn, + + Message::Ping(data) => { + if tx.try_send(OutMsg::Pong(data.to_vec())).is_err() { + break 'conn; + } + } + + Message::Pong(_) => {} + + Message::Close(_) => { + let _ = tx.try_send(OutMsg::Close); + break 'conn; + } + + Message::Text(text) => { + let arr: Vec = match serde_json::from_str::(text.as_str()).ok() { + Some(Value::Array(a)) if !a.is_empty() => a, + _ => { + let _ = tx.try_send(OutMsg::Text(make_notice("error: invalid message"))); + continue; + } + }; + + let verb = match arr[0].as_str() { + Some(v) => v.to_string(), + None => { + let _ = tx.try_send(OutMsg::Text(make_notice("error: invalid message"))); + continue; + } + }; + + match verb.as_str() { + "REQ" => { + // Structural validation first (before we have a valid sub_id). + if arr.len() < 3 || !arr[1].is_string() { + let _ = tx.try_send(OutMsg::Text(make_notice("error: invalid REQ"))); + continue; + } + // Now we know arr[1] is a string. + let client_sub_id = match arr[1].as_str() { + Some(s) if s.len() <= SUB_ID_MAX => s.to_string(), + Some(_) => { + let _ = tx.try_send(OutMsg::Text(make_closed( + "", + "error: sub_id too long", + ))); + continue; + } + None => { + continue; + } // arr[1].is_string() checked above + }; + if arr.len() > 3 { + let _ = tx.try_send(OutMsg::Text(make_closed( + &client_sub_id, + "error: multiple filters not supported", + ))); + continue; + } + if !arr[2].is_object() { + let _ = tx.try_send(OutMsg::Text(make_closed( + &client_sub_id, + "error: invalid filter", + ))); + continue; + } + if sub_id.is_some() { + let _ = tx.try_send(OutMsg::Text(make_closed( + &client_sub_id, + "error: already subscribed, send CLOSE first", + ))); + continue; + } + let p_value = match validate_filter(&arr[2]) { + Ok(p) => p, + Err(reason) => { + let _ = tx.try_send(OutMsg::Text(make_closed( + &client_sub_id, + &format!("error: {reason}"), + ))); + continue; + } + }; + // Atomically check #p uniqueness + register under one lock. + // EOSE try_send happens inside the lock to prevent a + // concurrent REQ from racing between check and push. + { + let mut subs = relay.subs.lock(); + if subs.iter().any(|s| s.p_value == p_value) { + let _ = tx.try_send(OutMsg::Text(make_closed( + &client_sub_id, + "error: #p already has a live subscriber", + ))); + continue; + } + // Send EOSE before registering (still under lock). + if tx + .try_send(OutMsg::Text(make_eose(&client_sub_id))) + .is_err() + { + break 'conn; + } + subs.push(Sub { + conn_id, + sub_id: client_sub_id.clone(), + p_value, + writer_tx: tx.clone(), + }); + } + sub_id = Some(client_sub_id); + } + + "EVENT" => { + if arr.len() != 2 { + let _ = tx.try_send(OutMsg::Text(make_notice("error: invalid EVENT"))); + continue; + } + // Rate check before validation (rate check takes priority). + if event_rate.tick() > RATE_EVENT_MAX { + let safe_id = safe_event_id(&arr[1]); + let _ = + tx.try_send(OutMsg::Text(make_ok(&safe_id, false, "rate-limited"))); + continue; + } + // Tightening #1: hard session cap check. + if events_attempted >= MAX_EVENTS_PER_CONN { + let safe_id = safe_event_id(&arr[1]); + let _ = tx.try_send(OutMsg::Text(make_ok( + &safe_id, + false, + "error: session event limit reached", + ))); + continue; + } + if !arr[1].is_object() { + let _ = tx.try_send(OutMsg::Text(make_ok( + "", + false, + "invalid: malformed event", + ))); + continue; + } + match validate_event(&arr[1]) { + Ok((event_id, id_bytes, p_value)) => { + // Tightening #8: Schnorr signature verification BEFORE dedup. + if let Err(reason) = verify_event_sig(&arr[1]) { + let _ = tx + .try_send(OutMsg::Text(make_ok(&event_id, false, reason))); + continue; + } + // Count all valid+sig-verified attempts toward the session cap. + events_attempted += 1; + // Tightening #7: atomic dedup reservation AFTER sig check. + match relay.reserve_id(&id_bytes) { + Ok(true) => { + let _ = tx.try_send(OutMsg::Text(make_ok( + &event_id, + false, + "duplicate: already seen", + ))); + continue; + } + Err(()) => { + let _ = tx.try_send(OutMsg::Text(make_ok( + &event_id, + false, + "relay at capacity", + ))); + continue; + } + Ok(false) => {} // reserved — proceed to delivery + } + // Tightening #2: atomically check subscriber and deliver. + match relay.deliver_single(&p_value, &arr[1]) { + Ok(true) => { + // ID stays reserved (already in dedup vec). + let _ = + tx.try_send(OutMsg::Text(make_ok(&event_id, true, ""))); + } + Ok(false) => { + // Delivery failed — unreserve so ID can be retried. + relay.unreserve_id(&id_bytes); + let _ = tx.try_send(OutMsg::Text(make_ok( + &event_id, + false, + "delivery failed", + ))); + } + Err(reason) => { + // Delivery rejected — unreserve so ID can be retried. + relay.unreserve_id(&id_bytes); + let _ = tx.try_send(OutMsg::Text(make_ok( + &event_id, false, reason, + ))); + } + } + } + Err(reason) => { + let safe_id = safe_event_id(&arr[1]); + let _ = tx.try_send(OutMsg::Text(make_ok( + &safe_id, + false, + &format!("invalid: {reason}"), + ))); + } + } + } + + "CLOSE" => { + if arr.len() != 2 { + let _ = tx.try_send(OutMsg::Text(make_notice("error: invalid CLOSE"))); + continue; + } + match arr[1].as_str() { + Some(sid) => { + if sub_id.as_deref() == Some(sid) { + relay.remove_sub(conn_id); + sub_id = None; + } + // Silently ignore unknown sub_id per NIP-01. + } + None => { + let _ = + tx.try_send(OutMsg::Text(make_notice("error: invalid CLOSE"))); + } + } + } + + _ => { + let _ = + tx.try_send(OutMsg::Text(make_notice("error: unsupported message"))); + } + } + } + } + } + + // Remove the subscription first so its cloned writer_tx is dropped. + // This allows the channel to close when we drop our local tx. + relay.remove_sub(conn_id); + + // Drop the sender so the writer can drain any queued messages (including + // Close frames), then cancel after a brief grace period. + drop(tx); + + // Only await the writer if it hasn't already completed (avoid double-poll panic). + if !writer_handle.is_finished() { + let _ = tokio::time::timeout(Duration::from_millis(100), &mut writer_handle).await; + } + cancel.cancel(); +} + +// ── HTTP upgrade ────────────────────────────────────────────────────────────── + +async fn http_service( + relay: Arc, + mut req: Request, +) -> Result>, hyper::Error> { + let headers = req.headers(); + let key = headers.get(SEC_WEBSOCKET_KEY).cloned(); + let is_ws = req.method() == Method::GET + && req.version() >= Version::HTTP_11 + && headers + .get(CONNECTION) + .and_then(|h| h.to_str().ok()) + .map(|h| { + h.split([' ', ',']) + .any(|p| p.eq_ignore_ascii_case("upgrade")) + }) + .unwrap_or(false) + && headers + .get(UPGRADE) + .and_then(|h| h.to_str().ok()) + .map(|h| h.eq_ignore_ascii_case("websocket")) + .unwrap_or(false) + && headers + .get(SEC_WEBSOCKET_VERSION) + .map(|h| h == "13") + .unwrap_or(false) + && key + .as_ref() + .map(|k| k.len() == 24 && k.as_bytes().iter().all(|&b| b.is_ascii())) + .unwrap_or(false); + + if !is_ws { + let mut r = Response::new(Full::default()); + *r.status_mut() = StatusCode::BAD_REQUEST; + return Ok(r); + } + + // Reserve slot before upgrading. + if relay.conn_count.fetch_add(1, Ordering::Relaxed) >= MAX_CONNS { + relay.conn_count.fetch_sub(1, Ordering::Relaxed); + let mut r = Response::new(Full::default()); + *r.status_mut() = StatusCode::SERVICE_UNAVAILABLE; + return Ok(r); + } + + let conn_id = relay.next_conn_id.fetch_add(1, Ordering::Relaxed); + eprintln!( + "conn opened conn_id={} active={}", + conn_id, + relay.conn_count.load(Ordering::Relaxed) + ); + + let accept = derive_accept_key(key.as_ref().map(|k| k.as_bytes()).unwrap_or(b"")); + let relay_clone = Arc::clone(&relay); + + tokio::spawn(async move { + match hyper::upgrade::on(&mut req).await { + Ok(upgraded) => { + let io = TokioIo::new(upgraded); + let mut ws_config = WebSocketConfig::default(); + ws_config.max_frame_size = Some(MAX_FRAME); + ws_config.max_message_size = Some(MAX_FRAME); + let stream = + WebSocketStream::from_raw_socket(io, Role::Server, Some(ws_config)).await; + handle_conn(relay_clone, conn_id, stream).await; + } + Err(e) => { + eprintln!("upgrade error: {e}"); + relay_clone.conn_count.fetch_sub(1, Ordering::Relaxed); + } + } + }); + + let mut resp = Response::new(Full::default()); + *resp.status_mut() = StatusCode::SWITCHING_PROTOCOLS; + resp.headers_mut() + .insert(CONNECTION, HeaderValue::from_static("Upgrade")); + resp.headers_mut() + .insert(UPGRADE, HeaderValue::from_static("websocket")); + if let Ok(val) = HeaderValue::from_str(&accept) { + resp.headers_mut().insert(SEC_WEBSOCKET_ACCEPT, val); + } + Ok(resp) +} + +// ── Server loop (extracted for testability) ─────────────────────────────────── + +/// Run the relay accept loop on the given listener. +/// Public for integration tests that bind to `:0`. +pub async fn run_server(listener: TcpListener, relay: Arc) { + let addr = listener.local_addr().ok(); + if let Some(a) = addr { + eprintln!("sprout-pair-relay listening on {a}"); + } + loop { + let (tcp, _peer) = match listener.accept().await { + Ok(pair) => pair, + Err(e) => { + eprintln!("accept error: {e}"); + continue; + } + }; + let relay = Arc::clone(&relay); + tokio::spawn(async move { + let io = TokioIo::new(tcp); + let svc = service_fn(move |req| http_service(Arc::clone(&relay), req)); + if let Err(e) = http1::Builder::new() + .serve_connection(io, svc) + .with_upgrades() + .await + { + eprintln!("http error: {e}"); + } + }); + } +} diff --git a/crates/sprout-pair-relay/src/main.rs b/crates/sprout-pair-relay/src/main.rs new file mode 100644 index 0000000000..f64c041edf --- /dev/null +++ b/crates/sprout-pair-relay/src/main.rs @@ -0,0 +1,19 @@ +use std::net::SocketAddr; +use std::sync::Arc; + +use sprout_pair_relay::{run_server, Relay}; +use tokio::net::TcpListener; + +#[tokio::main] +async fn main() { + let addr: SocketAddr = ([127, 0, 0, 1], 5000).into(); + let listener = match TcpListener::bind(addr).await { + Ok(l) => l, + Err(e) => { + eprintln!("fatal: failed to bind {addr}: {e}"); + std::process::exit(1); + } + }; + let relay = Arc::new(Relay::new()); + run_server(listener, relay).await; +} diff --git a/crates/sprout-pair-relay/tests/integration.rs b/crates/sprout-pair-relay/tests/integration.rs new file mode 100644 index 0000000000..ee09093b9a --- /dev/null +++ b/crates/sprout-pair-relay/tests/integration.rs @@ -0,0 +1,1400 @@ +//! Integration tests for sprout-pair-relay. +//! +//! Each test spins up a relay on a random port (`:0`), connects one or more +//! WebSocket clients, and exercises the observable protocol surface. + +use std::sync::Arc; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio::net::TcpListener; +use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; + +use sprout_pair_relay::{run_server, Relay}; + +// ── Crypto imports (for real event signing) ─────────────────────────────────── + +use secp256k1::{Keypair, Secp256k1, SecretKey}; +use sha2::{Digest, Sha256}; + +// ── Constants ───────────────────────────────────────────────────────────────── + +/// A valid 64-char lowercase hex string (all 'a's). +const P_A: &str = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; +/// A different valid 64-char lowercase hex string (all 'b's). +const P_B: &str = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; +/// A valid event id (all 'c's) — used only in tests that are rejected before +/// ID/sig verification (kind check, shape check, etc.). +const EV_ID: &str = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"; +/// A valid pubkey (all 'd's) — used only in pre-sig-check rejection tests. +const PUBKEY: &str = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"; +/// A valid sig (128 'e's) — used only in pre-sig-check rejection tests. +const SIG: &str = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"; + +// ── Test infrastructure ─────────────────────────────────────────────────────── + +type WS = WebSocketStream>; + +/// Start a relay on a random port, return the WebSocket URL. +async fn start_relay() -> String { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let relay = Arc::new(Relay::new()); + tokio::spawn(run_server(listener, relay)); + format!("ws://127.0.0.1:{}", addr.port()) +} + +/// Connect a WebSocket client to the relay. +async fn connect(url: &str) -> WS { + let (ws, _) = connect_async(url).await.unwrap(); + ws +} + +/// Send a JSON value as a text frame. +async fn send(ws: &mut WS, msg: &Value) { + ws.send(Message::Text(msg.to_string().into())) + .await + .unwrap(); +} + +/// Receive the next text frame and parse it as JSON. +/// Panics if no message arrives within 2 seconds. +async fn recv(ws: &mut WS) -> Value { + let msg = tokio::time::timeout(Duration::from_secs(2), ws.next()) + .await + .expect("timed out waiting for message") + .expect("stream ended") + .expect("WebSocket error"); + match msg { + Message::Text(t) => serde_json::from_str(t.as_str()).expect("invalid JSON"), + other => panic!("expected Text frame, got {:?}", other), + } +} + +/// Try to receive the next frame; return None if nothing arrives within 500 ms. +async fn try_recv(ws: &mut WS) -> Option { + tokio::time::timeout(Duration::from_millis(500), ws.next()) + .await + .ok()? + .and_then(|r| r.ok()) +} + +/// Assert the connection is closed (stream ends) within 2 seconds. +async fn assert_closed(ws: &mut WS) { + let result = tokio::time::timeout(Duration::from_secs(2), ws.next()).await; + match result { + Err(_) => panic!("connection did not close within 2 s"), + Ok(None) => {} // clean EOF + Ok(Some(Ok(Message::Close(_)))) => {} // close frame + Ok(Some(Err(_))) => {} // protocol error / reset + Ok(Some(Ok(other))) => panic!("expected close, got {:?}", other), + } +} + +// ── Crypto helpers ──────────────────────────────────────────────────────────── + +/// Generate a random keypair; returns `(SecretKey, pubkey_hex)`. +fn gen_keypair() -> (SecretKey, String) { + let secp = Secp256k1::new(); + let (sk, pk) = secp.generate_keypair(&mut secp256k1::rand::rngs::OsRng); + let xonly = pk.x_only_public_key().0; + let pubkey_hex = xonly + .serialize() + .iter() + .map(|b| format!("{b:02x}")) + .collect(); + (sk, pubkey_hex) +} + +/// Standard base64 encoder (no external dep). +fn base64_encode(data: &[u8]) -> String { + const ALPHA: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + let mut out = String::with_capacity(data.len().div_ceil(3) * 4); + for chunk in data.chunks(3) { + let b0 = chunk[0] as u32; + let b1 = if chunk.len() > 1 { chunk[1] as u32 } else { 0 }; + let b2 = if chunk.len() > 2 { chunk[2] as u32 } else { 0 }; + let triple = (b0 << 16) | (b1 << 8) | b2; + out.push(ALPHA[((triple >> 18) & 0x3F) as usize] as char); + out.push(ALPHA[((triple >> 12) & 0x3F) as usize] as char); + if chunk.len() > 1 { + out.push(ALPHA[((triple >> 6) & 0x3F) as usize] as char); + } else { + out.push('='); + } + if chunk.len() > 2 { + out.push(ALPHA[(triple & 0x3F) as usize] as char); + } else { + out.push('='); + } + } + out +} + +/// Build a minimal valid NIP-44 v2 fake ciphertext and base64-encode it. +/// Layout: 0x02 (version) | 32-byte nonce | 48-byte ciphertext | 32-byte MAC = 113 bytes. +/// 113 bytes ≥ 99 minimum; first decoded byte is 0x02. +fn make_nip44_content() -> String { + let mut blob = vec![0x02u8]; // version + blob.extend_from_slice(&[0xAA; 32]); // nonce + blob.extend_from_slice(&[0xBB; 48]); // ciphertext + blob.extend_from_slice(&[0xCC; 32]); // MAC + // 113 bytes total + base64_encode(&blob) +} + +/// Build a properly signed kind:24134 event targeting `p_hex`. +/// `nonce` is mixed into the content so callers can produce unique IDs from +/// the same keypair without sleeping. +fn make_signed_event(sk: &SecretKey, pubkey_hex: &str, p_hex: &str, nonce: u64) -> Value { + let secp = Secp256k1::new(); + let created_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64; + + // Vary content per nonce so each call produces a unique event ID. + let mut blob = vec![0x02u8]; + blob.extend_from_slice(&nonce.to_le_bytes()); // 8 bytes of nonce + blob.extend_from_slice(&[0xAA; 24]); // pad to 32-byte "nonce" field + blob.extend_from_slice(&[0xBB; 48]); + blob.extend_from_slice(&[0xCC; 32]); + let content = base64_encode(&blob); + + let tags = json!([["p", p_hex]]); + let kind = 24134u64; + + // NIP-01 commitment: [0, pubkey, created_at, kind, tags, content] + let commitment = json!([0, pubkey_hex, created_at, kind, tags, content]); + let commitment_str = serde_json::to_string(&commitment).unwrap(); + let hash: [u8; 32] = Sha256::digest(commitment_str.as_bytes()).into(); + let id_hex: String = hash.iter().map(|b| format!("{b:02x}")).collect(); + + let msg = secp256k1::Message::from_digest(hash); + let keypair = Keypair::from_secret_key(&secp, sk); + let sig = secp.sign_schnorr_no_aux_rand(&msg, &keypair); + let sig_hex: String = sig.serialize().iter().map(|b| format!("{b:02x}")).collect(); + + json!({ + "id": id_hex, + "pubkey": pubkey_hex, + "kind": 24134, + "created_at": created_at, + "content": content, + "sig": sig_hex, + "tags": [["p", p_hex]] + }) +} + +/// Return the current Unix timestamp as i64. +fn now_ts() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_secs() as i64 +} + +/// Send a REQ for the given sub_id and p_hex, then consume the EOSE. +async fn subscribe(ws: &mut WS, sub_id: &str, p_hex: &str) { + send(ws, &json!(["REQ", sub_id, {"#p": [p_hex]}])).await; + let eose = recv(ws).await; + assert_eq!(eose[0], "EOSE", "expected EOSE, got {eose}"); + assert_eq!(eose[1], sub_id); +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +/// 1. No replay: events published before a subscription are not delivered. +/// With tightening #2, publishing with no live subscriber is rejected +/// ("no live subscriber"), so the publisher gets OK false. +#[tokio::test] +async fn test_no_replay() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + // Publish first — no subscriber yet, so relay rejects with "no live subscriber". + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + let ok = recv(&mut pub_ws).await; + assert_eq!(ok[0], "OK"); + assert_eq!(ok[2], false, "expected rejection with no live subscriber"); + assert!( + ok[3].as_str().unwrap_or("").contains("no live subscriber"), + "unexpected message: {}", + ok[3] + ); + + // Subscribe — should only get EOSE, no EVENT. + let mut sub_ws = connect(&url).await; + send(&mut sub_ws, &json!(["REQ", "s1", {"#p": [P_A]}])).await; + let first = recv(&mut sub_ws).await; + assert_eq!(first[0], "EOSE", "expected EOSE, got {first}"); + assert!( + try_recv(&mut sub_ws).await.is_none(), + "received unexpected message after EOSE" + ); +} + +/// 2. Live delivery: events published after subscription are delivered; events +/// for a different p-tag are not. +#[tokio::test] +async fn test_live_delivery() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + let mut sub_ws = connect(&url).await; + subscribe(&mut sub_ws, "s1", P_A).await; + + // Publish matching event from a second connection. + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + let ok = recv(&mut pub_ws).await; + assert_eq!(ok[0], "OK"); + assert!(ok[2].as_bool().unwrap(), "event rejected: {}", ok[3]); + + // Subscriber should receive the event. + let ev_msg = recv(&mut sub_ws).await; + assert_eq!(ev_msg[0], "EVENT"); + assert_eq!(ev_msg[1], "s1"); + + // Publish to a different p-tag — no subscriber for P_B, so OK false. + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_B, 1)]), + ) + .await; + let ok2 = recv(&mut pub_ws).await; + assert_eq!(ok2[0], "OK"); + assert_eq!(ok2[2], false, "expected rejection for unsubscribed p-tag"); + + assert!( + try_recv(&mut sub_ws).await.is_none(), + "subscriber received event for wrong p-tag" + ); +} + +/// 3. Kind rejection: events with kind != 24134 are rejected. +#[tokio::test] +async fn test_kind_rejection() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let bad_event = json!({ + "id": EV_ID, + "pubkey": PUBKEY, + "kind": 1, + "created_at": 1_700_000_000i64, + "content": "hello", + "sig": SIG, + "tags": [["p", P_A]] + }); + send(&mut ws, &json!(["EVENT", bad_event])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); + assert!( + resp[3] + .as_str() + .unwrap_or("") + .contains("kind must be 24134"), + "unexpected message: {}", + resp[3] + ); +} + +/// 4. REQ without #p filter is rejected. +#[tokio::test] +async fn test_no_p_filter() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["REQ", "s1", {}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s1"); + assert!( + resp[2] + .as_str() + .unwrap_or("") + .contains("#p filter required"), + "unexpected message: {}", + resp[2] + ); +} + +/// 5. REQ with multiple #p values is rejected. +#[tokio::test] +async fn test_multi_value_p() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["REQ", "s1", {"#p": [P_A, P_B]}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s1"); + assert!( + resp[2] + .as_str() + .unwrap_or("") + .contains("#p must have exactly one value"), + "unexpected message: {}", + resp[2] + ); +} + +/// 6. REQ with unsupported filter field is rejected. +#[tokio::test] +async fn test_unsupported_filter_field() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send( + &mut ws, + &json!(["REQ", "s1", {"#p": [P_A], "authors": [PUBKEY]}]), + ) + .await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s1"); + assert!( + resp[2] + .as_str() + .unwrap_or("") + .contains("unsupported filter field"), + "unexpected message: {}", + resp[2] + ); +} + +/// 7. Second subscription with a different sub_id is rejected; first still works. +#[tokio::test] +async fn test_second_sub_different_id() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + let mut ws = connect(&url).await; + + subscribe(&mut ws, "s1", P_A).await; + + // Second REQ with a different sub_id. + send(&mut ws, &json!(["REQ", "s2", {"#p": [P_B]}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s2"); + assert!( + resp[2] + .as_str() + .unwrap_or("") + .contains("already subscribed"), + "unexpected message: {}", + resp[2] + ); + + // First subscription still works. + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + let ev_msg = recv(&mut ws).await; + assert_eq!(ev_msg[0], "EVENT"); + assert_eq!(ev_msg[1], "s1"); +} + +/// 8. Second subscription with the same sub_id is rejected; first still works. +#[tokio::test] +async fn test_second_sub_same_id() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + let mut ws = connect(&url).await; + + subscribe(&mut ws, "s1", P_A).await; + + // Same sub_id again. + send(&mut ws, &json!(["REQ", "s1", {"#p": [P_B]}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s1"); + assert!( + resp[2] + .as_str() + .unwrap_or("") + .contains("already subscribed"), + "unexpected message: {}", + resp[2] + ); + + // First subscription still works. + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + let ev_msg = recv(&mut ws).await; + assert_eq!(ev_msg[0], "EVENT"); + assert_eq!(ev_msg[1], "s1"); +} + +/// 9. Connection closes after 120 s (virtual time). +#[tokio::test(start_paused = true)] +async fn test_120s_timeout() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Advance virtual time past the connection timeout. + tokio::time::advance(Duration::from_secs(121)).await; + // Yield to let the relay task run its deadline branch. + tokio::task::yield_now().await; + + assert_closed(&mut ws).await; +} + +/// 10. Backpressure unit test: bounded mpsc channel rejects when full. +#[tokio::test] +async fn test_backpressure_unit() { + let (tx, _rx) = tokio::sync::mpsc::channel::(4); + for i in 0..4 { + tx.try_send(format!("msg {i}")) + .expect("send should succeed"); + } + // Channel is now full; 5th send must fail. + assert!( + tx.try_send("overflow".to_string()).is_err(), + "expected channel to be full" + ); +} + +/// 11. Frame larger than 4096 bytes closes the connection. +#[tokio::test] +async fn test_max_frame_size() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let big = "x".repeat(4097); + // Ignore send errors — the server may close mid-send. + let _ = ws.send(Message::Text(big.into())).await; + + assert_closed(&mut ws).await; +} + +/// 12. EVENT with missing `id` field is rejected. +#[tokio::test] +async fn test_event_shape_missing_id() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let bad = json!({ + "pubkey": PUBKEY, + "kind": 24134, + "created_at": 1_700_000_000i64, + "content": "x", + "sig": SIG, + "tags": [["p", P_A]] + }); + send(&mut ws, &json!(["EVENT", bad])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); +} + +/// 13. EVENT with no `p` tag is rejected. +#[tokio::test] +async fn test_no_p_tag() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let bad = json!({ + "id": EV_ID, + "pubkey": PUBKEY, + "kind": 24134, + "created_at": 1_700_000_000i64, + "content": "x", + "sig": SIG, + "tags": [] + }); + send(&mut ws, &json!(["EVENT", bad])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); +} + +/// 14. EVENT with two `p` tags is rejected. +#[tokio::test] +async fn test_multiple_p_tags() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let bad = json!({ + "id": EV_ID, + "pubkey": PUBKEY, + "kind": 24134, + "created_at": now_ts(), + "content": make_nip44_content(), + "sig": SIG, + "tags": [["p", P_A], ["p", P_B]] + }); + send(&mut ws, &json!(["EVENT", bad])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); + assert!( + resp[3].as_str().unwrap_or("").contains("exactly one p tag"), + "unexpected message: {}", + resp[3] + ); +} + +/// 15. REQ with non-hex #p value is rejected. +#[tokio::test] +async fn test_invalid_hex_in_p_filter() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["REQ", "s1", {"#p": ["not-hex-64-chars"]}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s1"); +} + +/// 16. EVENT with non-hex `id` is rejected. +#[tokio::test] +async fn test_invalid_hex_in_event_id() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let bad = json!({ + "id": "ZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZZ", + "pubkey": PUBKEY, + "kind": 24134, + "created_at": 1_700_000_000i64, + "content": "x", + "sig": SIG, + "tags": [["p", P_A]] + }); + send(&mut ws, &json!(["EVENT", bad])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); +} + +/// 17. Global connection cap: 128 connections succeed; 129th is rejected. +/// After closing one, a new connection succeeds. +#[tokio::test] +async fn test_global_conn_cap() { + let url = start_relay().await; + + let mut conns: Vec = Vec::with_capacity(128); + for _ in 0..128 { + conns.push(connect(&url).await); + } + + // 129th should fail (server returns 503). + let result = connect_async(&url).await; + assert!(result.is_err(), "expected 129th connection to fail"); + + // Close one connection. + let mut dropped = conns.pop().unwrap(); + dropped.close(None).await.unwrap(); + // Give the relay a moment to decrement its counter. + tokio::time::sleep(Duration::from_millis(100)).await; + + // Now a new connection should succeed. + let _new = connect(&url).await; +} + +/// 18. Session event cap: 6 EVENTs are accepted; 7th is rejected with +/// "session event limit reached". (The relay's hard cap is 6 per +/// connection, which is tighter than the per-window rate limit of 10.) +#[tokio::test] +async fn test_event_rate_limit() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + // Subscribe so the relay has exactly one live recipient for P_A. + let mut sub_ws = connect(&url).await; + subscribe(&mut sub_ws, "s1", P_A).await; + + let mut pub_ws = connect(&url).await; + + // First 6 events must be accepted (session cap = 6). + for i in 0..6u64 { + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, i)]), + ) + .await; + let resp = recv(&mut pub_ws).await; + assert_eq!(resp[0], "OK", "event {i}: {resp}"); + assert!( + resp[2].as_bool().unwrap(), + "event {i} rejected: {}", + resp[3] + ); + // Drain the forwarded event from the subscriber so the channel stays open. + let _ = recv(&mut sub_ws).await; + } + + // 7th must be rejected by the session cap. + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 6)]), + ) + .await; + let resp = recv(&mut pub_ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); + assert!( + resp[3] + .as_str() + .unwrap_or("") + .contains("session event limit reached"), + "unexpected message: {}", + resp[3] + ); +} + +/// 19. Message rate limit: 20 messages succeed; 21st closes the connection. +#[tokio::test] +async fn test_message_rate_limit() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Send 20 messages (CLOSE with unknown sub_id — no response generated). + for _ in 0..20 { + send(&mut ws, &json!(["CLOSE", "nonexistent"])).await; + } + // Small delay to let the server process all messages. + tokio::time::sleep(Duration::from_millis(50)).await; + + // 21st message should trigger the rate limit and close the connection. + send(&mut ws, &json!(["CLOSE", "nonexistent"])).await; + assert_closed(&mut ws).await; +} + +/// 20. REQ with multiple filters is rejected. +#[tokio::test] +async fn test_multiple_filters() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["REQ", "s1", {"#p": [P_A]}, {"#p": [P_B]}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s1"); + assert!( + resp[2].as_str().unwrap_or("").contains("multiple filters"), + "unexpected message: {}", + resp[2] + ); +} + +/// 21. Unknown message type receives a NOTICE. +#[tokio::test] +async fn test_unknown_message() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["AUTH", {}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "NOTICE"); + assert!( + resp[1] + .as_str() + .unwrap_or("") + .contains("unsupported message"), + "unexpected message: {}", + resp[1] + ); +} + +/// 22. Sub_id containing JSON special characters does not cause injection. +#[tokio::test] +async fn test_json_injection_sub_id() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let evil_id = r#""],\"evil\":["#; + send(&mut ws, &json!(["REQ", evil_id, {}])).await; + let resp = recv(&mut ws).await; + // Whatever the server responds, it must be valid JSON (recv() already + // parses it). The sub_id in the response must equal the literal string. + assert!(resp.is_array(), "response is not a JSON array: {resp}"); + // The response should be CLOSED with the literal sub_id echoed back safely. + if resp[0] == "CLOSED" { + let echoed = resp[1].as_str().unwrap_or(""); + assert_eq!(echoed, evil_id, "sub_id was not echoed verbatim"); + } +} + +/// 23. CLOSE removes the subscription; subsequent events are not delivered. +#[tokio::test] +async fn test_close_removes_sub() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + let mut ws = connect(&url).await; + + subscribe(&mut ws, "s1", P_A).await; + + send(&mut ws, &json!(["CLOSE", "s1"])).await; + // Give the relay a moment to process the CLOSE. + tokio::time::sleep(Duration::from_millis(50)).await; + + // Publish a matching event — no subscriber, so OK false. + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + let ok = recv(&mut pub_ws).await; + assert_eq!(ok[0], "OK"); + // No subscriber after CLOSE, so event is rejected. + assert_eq!(ok[2], false); + + // Original subscriber must not receive anything. + assert!( + try_recv(&mut ws).await.is_none(), + "received event after CLOSE" + ); +} + +/// 24. CLOSE does not close the WebSocket connection. +#[tokio::test] +async fn test_close_keeps_connection() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + subscribe(&mut ws, "s1", P_A).await; + send(&mut ws, &json!(["CLOSE", "s1"])).await; + + // Connection must still be open — send another message and get a response. + send(&mut ws, &json!(["REQ", "s2", {"#p": [P_B]}])).await; + let eose = recv(&mut ws).await; + assert_eq!(eose[0], "EOSE"); + assert_eq!(eose[1], "s2"); +} + +/// 25. After CLOSE, a new REQ works and receives future events. +#[tokio::test] +async fn test_req_after_close() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + let mut ws = connect(&url).await; + + subscribe(&mut ws, "s1", P_A).await; + send(&mut ws, &json!(["CLOSE", "s1"])).await; + + // Re-subscribe with a new sub_id. + subscribe(&mut ws, "s2", P_A).await; + + // Publish a matching event. + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + let ok = recv(&mut pub_ws).await; + assert_eq!(ok[0], "OK"); + assert!(ok[2].as_bool().unwrap(), "event rejected: {}", ok[3]); + + // New subscription must receive the event. + let ev_msg = recv(&mut ws).await; + assert_eq!(ev_msg[0], "EVENT"); + assert_eq!(ev_msg[1], "s2"); +} + +/// 26. CLOSE with an unknown sub_id is silently ignored. +#[tokio::test] +async fn test_close_unknown_sub_id() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["CLOSE", "nonexistent"])).await; + + // No error response; connection stays open. + assert!( + try_recv(&mut ws).await.is_none(), + "received unexpected response to CLOSE of unknown sub_id" + ); + + // Connection is still usable. + send(&mut ws, &json!(["REQ", "s1", {"#p": [P_A]}])).await; + let eose = recv(&mut ws).await; + assert_eq!(eose[0], "EOSE"); +} + +/// 27. No events delivered after CLOSE (explicit duplicate of test 23). +#[tokio::test] +async fn test_no_events_after_close() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + let mut ws = connect(&url).await; + + subscribe(&mut ws, "sub", P_A).await; + send(&mut ws, &json!(["CLOSE", "sub"])).await; + tokio::time::sleep(Duration::from_millis(50)).await; + + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + recv(&mut pub_ws).await; // consume OK (will be false — no subscriber) + + assert!( + try_recv(&mut ws).await.is_none(), + "received event after CLOSE" + ); +} + +/// 28. Binary WebSocket frame closes the connection. +#[tokio::test] +async fn test_binary_frame() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let _ = ws.send(Message::Binary(b"hello".to_vec().into())).await; + + assert_closed(&mut ws).await; +} + +/// 29. REQ with too few elements receives a NOTICE. +#[tokio::test] +async fn test_malformed_req_too_few() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["REQ"])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "NOTICE"); + assert!( + resp[1].as_str().unwrap_or("").contains("invalid REQ"), + "unexpected message: {}", + resp[1] + ); +} + +/// 30. REQ with a non-string sub_id receives a NOTICE. +#[tokio::test] +async fn test_malformed_req_non_string_sub_id() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["REQ", 123, {}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "NOTICE"); + assert!( + resp[1].as_str().unwrap_or("").contains("invalid REQ"), + "unexpected message: {}", + resp[1] + ); +} + +/// 31. REQ with a non-object filter receives a CLOSED. +#[tokio::test] +async fn test_malformed_req_non_object_filter() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + send(&mut ws, &json!(["REQ", "s1", "bad"])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], "s1"); + assert!( + resp[2].as_str().unwrap_or("").contains("invalid filter"), + "unexpected message: {}", + resp[2] + ); +} + +/// 32. Write timeout / slow reader: subscriber that doesn't read eventually +/// gets disconnected when the server's write buffer fills up. +/// +/// We flood the subscriber's p-tag with events from a second connection. +/// Because the subscriber never reads, the relay's bounded channel fills +/// and subsequent fan-out drops are silent. The subscriber connection +/// itself stays open (fan-out drops don't close). This test verifies the +/// observable behavior: the publisher keeps getting OK responses and the +/// subscriber connection is still alive after the flood. +#[tokio::test] +async fn test_write_timeout() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + // Subscribe but never read. + let mut sub_ws = connect(&url).await; + send(&mut sub_ws, &json!(["REQ", "s1", {"#p": [P_A]}])).await; + // Don't call recv — leave the EOSE unread. + + // Flood from a publisher — stay within the 6-event session cap. + let mut pub_ws = connect(&url).await; + for i in 0..6u64 { + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, i)]), + ) + .await; + let ok = recv(&mut pub_ws).await; + assert_eq!(ok[0], "OK"); + } + + // Publisher connection is still healthy. + send(&mut pub_ws, &json!(["REQ", "check", {"#p": [P_B]}])).await; + let eose = recv(&mut pub_ws).await; + assert_eq!(eose[0], "EOSE"); +} + +/// 33. Ping frame receives a Pong with the same payload. +#[tokio::test] +async fn test_ping_pong() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let payload = b"hello".to_vec(); + ws.send(Message::Ping(payload.clone().into())) + .await + .unwrap(); + + let msg = tokio::time::timeout(Duration::from_secs(2), ws.next()) + .await + .expect("timed out") + .expect("stream ended") + .expect("WebSocket error"); + + match msg { + Message::Pong(data) => assert_eq!(data.as_ref(), payload.as_slice()), + other => panic!("expected Pong, got {:?}", other), + } +} + +/// 34. Client-initiated Close frame receives a Close reply. +#[tokio::test] +async fn test_close_handshake() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + ws.send(Message::Close(None)).await.unwrap(); + + // The stream should end (server sends its own Close and closes). + assert_closed(&mut ws).await; +} + +/// 35. Connection counter does not leak: open/close 5 connections, then open +/// 128 more — all should succeed. +#[tokio::test] +async fn test_conn_counter_no_leak() { + let url = start_relay().await; + + // Open and close 5 connections. + for _ in 0..5 { + let mut ws = connect(&url).await; + ws.close(None).await.unwrap(); + } + // Give the relay time to decrement counters. + tokio::time::sleep(Duration::from_millis(200)).await; + + // Now open 128 connections — all should succeed. + let mut conns: Vec = Vec::with_capacity(128); + for _ in 0..128 { + conns.push(connect(&url).await); + } + assert_eq!(conns.len(), 128); +} + +/// 36. Fan-out drops do not close the subscriber connection. +#[tokio::test] +async fn test_control_msg_backpressure() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + let mut sub_ws = connect(&url).await; + send(&mut sub_ws, &json!(["REQ", "s1", {"#p": [P_A]}])).await; + // Leave EOSE unread to fill the channel quickly. + + // Flood within the 6-event session cap. + let mut pub_ws = connect(&url).await; + for i in 0..6u64 { + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, i)]), + ) + .await; + let _ = recv(&mut pub_ws).await; + } + + // Subscriber connection must still be alive — verify by closing it cleanly. + let close_result = tokio::time::timeout(Duration::from_secs(2), sub_ws.close(None)).await; + assert!( + close_result.is_ok(), + "subscriber connection died after flood" + ); +} + +/// 37. Various malformed inputs do not crash the relay (no panic). +#[tokio::test] +async fn test_no_client_data_in_logs() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // A grab-bag of weird inputs — none should panic the relay. + let inputs: &[Value] = &[ + json!(null), + json!(42), + json!("just a string"), + json!({}), + json!([]), + json!(["UNKNOWN"]), + json!(["EVENT"]), + json!(["EVENT", null]), + json!(["REQ", null]), + json!(["CLOSE"]), + json!(["CLOSE", null]), + ]; + + for input in inputs { + // Ignore send errors (server may close on some inputs). + let _ = ws.send(Message::Text(input.to_string().into())).await; + // Small pause to let the relay process. + tokio::time::sleep(Duration::from_millis(10)).await; + } + + // If we reach here without a panic, the test passes. + // The connection may or may not still be open. +} + +/// 38. EOSE arrives before any EVENT in normal flow. +/// Publishing with no live subscriber is rejected (tightening #2). +#[tokio::test] +async fn test_eose_try_send_failure() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + // Publisher sends an event before the subscriber connects — rejected. + let mut pub_ws = connect(&url).await; + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, 0)]), + ) + .await; + let ok = recv(&mut pub_ws).await; + assert_eq!(ok[0], "OK"); + assert_eq!(ok[2], false, "expected rejection with no live subscriber"); + + // Subscriber connects after the event — should see EOSE first (no EVENT, + // since there is no persistence). + let mut sub_ws = connect(&url).await; + send(&mut sub_ws, &json!(["REQ", "s1", {"#p": [P_A]}])).await; + let first = recv(&mut sub_ws).await; + assert_eq!(first[0], "EOSE", "first message must be EOSE, got {first}"); + + // No further messages (no stored events). + assert!( + try_recv(&mut sub_ws).await.is_none(), + "received unexpected message after EOSE" + ); +} + +/// 39. Ping frames count toward the per-connection message rate limit. +/// We use CLOSE messages (no response generated) to burn through the rate +/// limit without buffered responses interfering with the close assertion. +#[tokio::test] +async fn test_ping_counts_toward_rate_limit() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Send 20 CLOSE messages for a non-existent sub_id. + // Each counts toward the 20-message rate limit but generates no response. + for _ in 0..20 { + send(&mut ws, &json!(["CLOSE", "nope"])).await; + } + + // Small yield to let the server process all 20. + tokio::time::sleep(Duration::from_millis(50)).await; + + // 21st message (a Ping this time) should trigger the rate limit. + let _ = ws.send(Message::Ping(vec![].into())).await; + + assert_closed(&mut ws).await; +} + +/// 40. Fan-out drops do not close the subscriber connection. +#[tokio::test] +async fn test_fan_out_drop_doesnt_close() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + // Subscribe but don't read (leave EOSE buffered). + let mut sub_ws = connect(&url).await; + send(&mut sub_ws, &json!(["REQ", "s1", {"#p": [P_A]}])).await; + + // Flood from a publisher — stay within the 6-event session cap. + let mut pub_ws = connect(&url).await; + for i in 0..6u64 { + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, i)]), + ) + .await; + let _ = recv(&mut pub_ws).await; + } + + // Subscriber connection must still be alive — verify by closing it cleanly. + let result = tokio::time::timeout(Duration::from_secs(2), sub_ws.close(None)).await; + assert!( + result.is_ok(), + "subscriber connection was unexpectedly dead" + ); +} + +/// 41. Reader backpressure: publisher can keep running even when the subscriber +/// never reads. +#[tokio::test] +async fn test_reader_backpressure_closes() { + let url = start_relay().await; + let (sk, pk) = gen_keypair(); + + // Slow subscriber — never reads. + let mut sub_ws = connect(&url).await; + send(&mut sub_ws, &json!(["REQ", "s1", {"#p": [P_A]}])).await; + + // Publisher floods events up to the session cap. + let mut pub_ws = connect(&url).await; + for i in 0..6u64 { + send( + &mut pub_ws, + &json!(["EVENT", make_signed_event(&sk, &pk, P_A, i)]), + ) + .await; + let ok = recv(&mut pub_ws).await; + // Publisher must always get OK (fan-out drops are silent to publisher). + assert_eq!(ok[0], "OK"); + } +} + +/// 42. Connection closes promptly after 120 s (virtual time). +/// Explicit duplicate of test 9 with a slightly different assertion style. +#[tokio::test(start_paused = true)] +async fn test_cancellation_immediate() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + tokio::time::advance(Duration::from_secs(121)).await; + tokio::task::yield_now().await; + + // The connection must be closed — not just slow. + assert_closed(&mut ws).await; +} + +/// 43. Client-initiated graceful close receives a Close reply. +/// Explicit duplicate of test 34 with a different connection state. +#[tokio::test] +async fn test_graceful_close() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Subscribe first, then close gracefully. + subscribe(&mut ws, "s1", P_A).await; + + ws.send(Message::Close(None)).await.unwrap(); + assert_closed(&mut ws).await; +} + +/// 44. Multiple subscribers on the same #p value: event is rejected with +/// "ambiguous recipient" (tightening #2 — exactly one subscriber required). +#[tokio::test] +async fn test_multiple_subscribers_same_p() { + let url = start_relay().await; + + // First subscriber on #p succeeds. + let mut sub1 = connect(&url).await; + subscribe(&mut sub1, "s1", P_A).await; + + // Second subscriber on the SAME #p is rejected at REQ time. + let mut sub2 = connect(&url).await; + send(&mut sub2, &json!(["REQ", "s2", {"#p": [P_A]}])).await; + let resp = recv(&mut sub2).await; + assert_eq!(resp[0], "CLOSED"); + assert!( + resp[2] + .as_str() + .unwrap_or("") + .contains("already has a live subscriber"), + "unexpected message: {}", + resp[2] + ); +} + +/// 45. Uppercase hex in #p filter value is rejected. +#[tokio::test] +async fn test_uppercase_hex_in_p_filter() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Uppercase hex — must be rejected. + let upper_p = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"; + send(&mut ws, &json!(["REQ", "s1", {"#p": [upper_p]}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert!( + resp[2].as_str().unwrap_or("").contains("64 lowercase hex"), + "unexpected: {}", + resp[2] + ); +} + +/// 46. Uppercase hex in event id is rejected. +#[tokio::test] +async fn test_uppercase_hex_in_event_fields() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Build an event with a valid shape but uppercase id — rejected at hex check. + let ev = json!({ + "id": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + "pubkey": PUBKEY, + "kind": 24134, + "created_at": 1_700_000_000i64, + "content": make_nip44_content(), + "sig": SIG, + "tags": [["p", P_A]] + }); + send(&mut ws, &json!(["EVENT", ev])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); + assert!(resp[3].as_str().unwrap_or("").contains("64 lowercase hex")); +} + +/// 47. Overlong sub_id (> 64 bytes) is rejected with CLOSED "". +#[tokio::test] +async fn test_overlong_sub_id() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let long_sub_id = "x".repeat(65); + send(&mut ws, &json!(["REQ", long_sub_id, {"#p": [P_A]}])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "CLOSED"); + assert_eq!(resp[1], ""); // sub_id too long → use "" + assert!(resp[2].as_str().unwrap_or("").contains("sub_id too long")); +} + +/// 48. Negative created_at is rejected by the freshness window check. +/// (Tightening #6: created_at must be within ±120 s of relay wall-clock.) +#[tokio::test] +async fn test_negative_created_at() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // A stale event with created_at = -1 is far outside the ±120 s window. + let ev = json!({ + "id": EV_ID, + "pubkey": PUBKEY, + "kind": 24134, + "created_at": -1i64, + "content": make_nip44_content(), + "sig": SIG, + "tags": [["p", P_A]] + }); + send(&mut ws, &json!(["EVENT", ev])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false, "expected rejection for stale created_at"); + assert!( + resp[3].as_str().unwrap_or("").contains("freshness window"), + "unexpected message: {}", + resp[3] + ); +} + +/// 49. Event with missing sig field is rejected. +#[tokio::test] +async fn test_event_missing_sig() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + let ev = json!({ + "id": EV_ID, + "pubkey": PUBKEY, + "kind": 24134, + "created_at": now_ts(), + "content": make_nip44_content(), + "tags": [["p", P_A]] + }); + send(&mut ws, &json!(["EVENT", ev])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); + assert!(resp[3].as_str().unwrap_or("").contains("missing sig")); +} + +/// 50. EVENT with extra tags (not exactly one `["p", ...]`) is rejected. +/// (Tightening #3: tags must be exactly `[["p", "<64-hex>"]]`.) +#[tokio::test] +async fn test_event_too_many_tags() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Two tags — violates the "exactly one p tag" rule. + let ev = json!({ + "id": EV_ID, + "pubkey": PUBKEY, + "kind": 24134, + "created_at": now_ts(), + "content": make_nip44_content(), + "sig": SIG, + "tags": [["p", P_A], ["x", "extra"]] + }); + send(&mut ws, &json!(["EVENT", ev])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); + assert!( + resp[3].as_str().unwrap_or("").contains("exactly one p tag"), + "unexpected message: {}", + resp[3] + ); +} + +/// 51. EVENT with extra tags (not exactly one `["p", ...]`) is rejected. +/// (Tightening #3: tags must be exactly `[["p", "<64-hex>"]]`.) +/// Variant: two tags where one has a long string value. +#[tokio::test] +async fn test_event_tag_string_too_long() { + let url = start_relay().await; + let mut ws = connect(&url).await; + + // Two tags — violates the "exactly one p tag" rule before any length check. + let long_val = "x".repeat(129); + let ev = json!({ + "id": EV_ID, + "pubkey": PUBKEY, + "kind": 24134, + "created_at": now_ts(), + "content": make_nip44_content(), + "sig": SIG, + "tags": [["p", P_A], ["x", long_val]] + }); + send(&mut ws, &json!(["EVENT", ev])).await; + let resp = recv(&mut ws).await; + assert_eq!(resp[0], "OK"); + assert_eq!(resp[2], false); + // Strict tag check fires first: exactly one tag required. + assert!( + resp[3].as_str().unwrap_or("").contains("exactly one p tag"), + "unexpected message: {}", + resp[3] + ); +}